/*
(non-Javadoc)

@see com.ibm.ws.sib.processor.impl.interfaces.DestinationHandler#createSubscriptionConsumerDispatcherAndAttachCP(com.ibm.ws.sib.processor.impl.interfaces.LocalConsumerPoint,
com.ibm.ws.sib.processor.impl.ConsumerDispatcherState)
166507
@Override
    //Thsi is dummy method. Liberty does not use this class.
    public ConsumerKey createSubscriptionConsumerDispatcherAndAttachCP(LocalConsumerPoint consumerPoint, ConsumerDispatcherState subState) throws SIDiscriminatorSyntaxException, SISelectorSyntaxException, SIResourceException, SINonDurableSubscriptionMismatchException, SINotPossibleInCurrentConfigurationException, SIDestinationLockedException, SISessionDroppedException {
        return null;
    }
1238030
public ChannelFuture requestSubpartition(
			final ResultPartitionID partitionId,
			final int subpartitionIndex,
			final RemoteInputChannel inputChannel,
			int delayMs) throws IOException {

		checkNotClosed();

		LOG.debug("Requesting subpartition {} of partition {} with {} ms delay.",
				subpartitionIndex, partitionId, delayMs);

		clientHandler.addInputChannel(inputChannel);

		final PartitionRequest request = new PartitionRequest(
				partitionId, subpartitionIndex, inputChannel.getInputChannelId(), inputChannel.getInitialCredit());

		final ChannelFutureListener listener = new ChannelFutureListener() {
			@Override
			public void operationComplete(ChannelFuture future) throws Exception {
				if (!future.isSuccess()) {
					clientHandler.removeInputChannel(inputChannel);
					SocketAddress remoteAddr = future.channel().remoteAddress();
					inputChannel.onError(
							new LocalTransportException(
								String.format("Sending the partition request to '%s' failed.", remoteAddr),
								future.channel().localAddress(), future.cause()
							));
				}
			}
		};

		if (delayMs == 0) {
			ChannelFuture f = tcpChannel.writeAndFlush(request);
			f.addListener(listener);
			return f;
		} else {
			final ChannelFuture[] f = new ChannelFuture[1];
			tcpChannel.eventLoop().schedule(new Runnable() {
				@Override
				public void run() {
					f[0] = tcpChannel.writeAndFlush(request);
					f[0].addListener(listener);
				}
			}, delayMs, TimeUnit.MILLISECONDS);

			return f[0];
		}
	}
704150
protected static ControlCreateDurable createDurableCreateDurable(
    MessageProcessor MP,
    ConsumerDispatcherState subState,
    long reqID,
    SIBUuid8 dme) 
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createDurableCreateDurable", new Object[] {MP, subState, new Long(reqID), dme});

    ControlCreateDurable msg = null;
      
    try
    {
      // Create and initialize the message
      msg = MessageProcessor.getControlMessageFactory().createNewControlCreateDurable();
      initializeControlMessage(MP.getMessagingEngineUuid(), msg, dme);
      
      // Parameterize for CreateStream
      msg.setRequestID(reqID);
      msg.setDurableSubName(subState.getSubscriberID());
      
      SelectionCriteria criteria = subState.getSelectionCriteria();
      //check for null values for MFP - discriminator can be null
      //the discriminator
      
      if(criteria==null || criteria.getDiscriminator()==null)
      {
        msg.setDurableDiscriminator(null); 
      } 
      else
      {
        msg.setDurableDiscriminator(criteria.getDiscriminator()); 
      } 
      //the selector
      if(criteria==null || criteria.getSelectorString()==null)
      {
        msg.setDurableSelector(null); 
      } 
      else
      {
        msg.setDurableSelector(subState.getSelectionCriteria().getSelectorString());
      } 
      //the selector domain
      if(criteria==null || criteria.getSelectorDomain()==null)
      {
        msg.setDurableSelectorDomain(SelectorDomain.SIMESSAGE.toInt());
      }  
      else
      {
        msg.setDurableSelectorDomain(criteria.getSelectorDomain().toInt());
      }
      
      // Check the selectorProperties Map to see if we need to convey any additional properties associated
      // with the selector. At present (26/03/08) there is only one additional property
      // which is itself a map (of name spaces). The name space map is used in the XPath10 selector domain
      // to map URLs to prefixes. The use of a selectorProperties map keeps the Core SPI generic but
      // when conveying information over JMF we need a simpler structure and so will need to
      // break out individual properties for transportation.
      if(criteria==null)
      {
        msg.setDurableSelectorNamespaceMap(null);
      }  
      else
      {
        // See if these criteria have any selector properties. They might if they are MPSelectionCriteria
        if(criteria instanceof MPSelectionCriteria)
        {
          MPSelectionCriteria mpCriteria = (MPSelectionCriteria)criteria;
          Map<String, Object> selectorProperties = mpCriteria.getSelectorProperties();
          
          if(selectorProperties != null)
          {
            Map<String, String> selectorNamespaceMap = 
              (Map<String, String>)selectorProperties.get("namespacePrefixMappings");

            if(selectorNamespaceMap != null)
              msg.setDurableSelectorNamespaceMap(selectorNamespaceMap);
            else
              msg.setDurableSelectorNamespaceMap(null);
          }
          else
          {
            msg.setDurableSelectorNamespaceMap(null);
          }
        } 
        else
        {
          msg.setDurableSelectorNamespaceMap(null);
        } // eof instanceof MPSelectionCriteria
      } // eof null criteria
      
      //defect 259036
      msg.setCloned(subState.isCloned());
      msg.setNoLocal(subState.isNoLocal());

      msg.setSecurityUserid(subState.getUser());

      // Set the flag that signals whether this is
      // the privileged SIBServerSubject.
      msg.setSecurityUseridSentBySystem(subState.isSIBServerSubject());          
    }
    catch (Exception e)
    {
      FFDCFilter.processException(e,
        "com.ibm.ws.sib.processor.impl.DurableInputHandler.createDurableCreateDurable",
        "1:495:1.52.1.1",
        DurableInputHandler.class);
      SibTr.exception(tc, e);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createDurableCreateDurable", msg);
      
    return msg;
  }
535463
@Override
    public long newReadyConsumer(JSConsumerKey consumerKey, boolean bSelector)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "newReadyConsumer",
                        new Object[] { consumerKey, Boolean.valueOf(bSelector) });
        // If this is a forward scanning consumer just remember we have one
        if (consumerKey.getForwardScanning())
        {
            //move the consumerPoint in to the ready list
            readyFwdScanningCPs.put((SimpleEntry) consumerKey);
        }
        else
        {
            // If the consumer has a selector...
            if (bSelector)
            {
                //increment the specific consumer count and the version number
                specificReadyConsumerCount++;

                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, "sepcificReadyConsumerCount:" + specificReadyConsumerCount);
            }
            else
            {
                //move the consumerPoint in to the ready list
                nonSpecificReadyCPs.put((SimpleEntry) consumerKey);
            }
        }

        // Increment the ready version
        ++readyConsumerVersion;

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "newReadyConsumer", Long.valueOf(readyConsumerVersion));
        return readyConsumerVersion;
    }
773622
@Override
    public IPortletEntity createPortletEntity(
            IPortletDefinitionId portletDefinitionId, String layoutNodeId, int userId) {
        if (layoutNodeId.startsWith(TransientUserLayoutManagerWrapper.SUBSCRIBE_PREFIX)) {
            final String transientLayoutNodeId = layoutNodeId;
            layoutNodeId = this.getPersistentLayoutNodeId(portletDefinitionId);

            final IPortletEntity portletEntity =
                    this.delegatePortletEntityDao.createPortletEntity(
                            portletDefinitionId, layoutNodeId, userId);
            return new TransientPortletEntity(portletEntity, transientLayoutNodeId);
        }

        return this.delegatePortletEntityDao.createPortletEntity(
                portletDefinitionId, layoutNodeId, userId);
    }
1084696
public SubscriptionDefinition createSubscriptionDefinition()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createSubscriptionDefinition");

    SubscriptionDefinition subDef = new SubscriptionDefinitionImpl();

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createSubscriptionDefinition", subDef);

    return subDef;
  }
795979
public void setSubscriptionIDFilter(String subscriptionId)
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "setSubscriptionIDFilter", subscriptionId);

    this._subscriptionId = subscriptionId;
    this._destination = null;
    this._consumerDispatcherState = null;
    
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "setSubscriptionIDFilter");
  }
216239
@Override
	public void addBufferConsumer(BufferConsumer bufferConsumer, int subpartitionIndex) throws IOException {
		checkNotNull(bufferConsumer);

		ResultSubpartition subpartition;
		try {
			checkInProduceState();
			subpartition = subpartitions[subpartitionIndex];
		}
		catch (Exception ex) {
			bufferConsumer.close();
			throw ex;
		}

		if (subpartition.add(bufferConsumer)) {
			notifyPipelinedConsumers();
		}
	}
853707
public ConsumerDispatcher getDurableSubscriptionConsumerDispatcher(ConsumerDispatcherState subState)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "getDurableSubscriptionConsumerDispatcher", subState);

        ConsumerDispatcher consumerDispatcher = null;

        // NOTE: there are two unsynchronized calls to this method from SIMPSubscriptionAssert
        // The more common call from createDurableSubscription already holds the lock
        // here in which case the extra synchronized is minor overhead.

        synchronized (_consumerDispatchersDurable)
        {
            consumerDispatcher =
                            (ConsumerDispatcher) _consumerDispatchersDurable.get(
                                            subState.getSubscriberID());
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(
                       tc,
                       "getDurableSubscriptionConsumerDispatcher",
                       consumerDispatcher);

        return consumerDispatcher;
    }
979468
protected SIDestinationAddress getConsumerSIDestinationAddress() throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "getConsumerSIDestinationAddress");

        if (consumerDestinationAddress == null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(this, tc, "No cached value - create a new one.");

            // This variable should be initialised already since it is set up in the static
            // init for the class, which must have been run by the time we get to access it.

            // Establish whether this producer should be scoped to a local queue point only.
            boolean localOnly = isLocalOnly();

            // The consumer details are those that we use as the 'big' destination information.
            consumerDestinationAddress = JmsMessageImpl.destAddressFactory.createSIDestinationAddress(getConsumerDestName(), localOnly, getBusName());
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "getConsumerSIDestinationAddress", consumerDestinationAddress);
        return consumerDestinationAddress;
    }
376925
protected static ControlCreateStream createDurableCreateStream(
    MessageProcessor MP,
    ConsumerDispatcherState subState,
    long reqID,
    SIBUuid8 dme) 
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createDurableCreateStream", new Object[] {MP, subState, new Long(reqID), dme});
    
    ControlCreateStream msg = null;
         
    try
    {

      // Create and initialize the message
      msg = MessageProcessor.getControlMessageFactory().createNewControlCreateStream();
      initializeControlMessage(MP.getMessagingEngineUuid(), msg, dme);
      
      // Parameterize for CreateStream
      msg.setRequestID(reqID);
      msg.setDurableSubName(subState.getSubscriberID());
      msg.setGuaranteedTargetDestinationDefinitionUUID(subState.getTopicSpaceUuid());
      
      SelectionCriteria criteria = subState.getSelectionCriteria();
      //check for null values for MFP - defect 251989
      //the discriminator
      if(criteria==null || criteria.getDiscriminator()==null)
      {
        msg.setDurableDiscriminator(null); 
      } 
      else
      {
        msg.setDurableDiscriminator(criteria.getDiscriminator()); 
      } 
      //the selector
      if(criteria==null || criteria.getSelectorString()==null)
      {
        msg.setDurableSelector(null); 
      } 
      else
      {
        msg.setDurableSelector(subState.getSelectionCriteria().getSelectorString());
      } 
      //the selector domain
      if(criteria==null || criteria.getSelectorDomain()==null)
      {
        msg.setDurableSelectorDomain(SelectorDomain.SIMESSAGE.toInt());
      }  
      else
      {
        msg.setDurableSelectorDomain(criteria.getSelectorDomain().toInt());
      }
      
      //defect 259036
      msg.setCloned(subState.isCloned());
      msg.setNoLocal(subState.isNoLocal());
            
      msg.setSecurityUserid(subState.getUser());
      // Set the flag that signals whether this is
      // the privileged SIBServerSubject.
      msg.setSecurityUseridSentBySystem(subState.isSIBServerSubject());
    }
    catch (Exception e)
    {
      FFDCFilter.processException(e,
        "com.ibm.ws.sib.processor.impl.DurableInputHandler.createDurableCreateStream",
        "1:372:1.52.1.1",
        DurableInputHandler.class);
      SibTr.exception(tc, e);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createDurableCreateStream", msg);
      
    return msg;
  }
227550
@Override
    public void removeConsumerPointMatchTarget(DispatchableKey consumerPointData)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "removeConsumerPointMatchTarget", consumerPointData);

        // Remove the consumer point from the matchspace
        // This used to use the ptoPRealization but this is no longer the case.

        messageProcessor
                        .getMessageProcessorMatching()
                        .removeConsumerPointMatchTarget(
                                                        consumerPointData);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "removeConsumerPointMatchTarget");
    }
360257
public void dereferenceSubscriptionConsumerDispatcher(
                                                          ConsumerDispatcher cd,
                                                          boolean isNonDurable,
                                                          boolean callProxyCode)
                    throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "dereferenceSubscriptionConsumerDispatcher",
                        new Object[] {
                                      cd,
                                      Boolean.valueOf(isNonDurable),
                                      Boolean.valueOf(callProxyCode) });

        _pubSubRealization.dereferenceSubscriptionConsumerDispatcher(cd,
                                                                     isNonDurable,
                                                                     callProxyCode);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "dereferenceSubscriptionConsumerDispatcher");
    }
587002
public void deleteLocalDurableSubscription(
                                               String subscriptionId)
                    throws
                    SIResourceException,
                    SIMPDestinationLockedException,
                    SIDurableSubscriptionNotFoundException
    {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "deleteLocalDurableSubscription",
                        new Object[] { subscriptionId });

        // Lock the table of durable subscriptions to prevent others accessing the subscription 
        // We must perform the following work inside the sync block, in the following
        // order to protect against any chance of a duplicate subscription:
        // - Find the ConsumerDispatcher from the map
        // - Find the matching subscription item stream
        // - Persist the toBeDeleted flag on the item stream, so that it can never pop
        //   back into existence in the future.
        // - Remove the ConsumerDispatcher from both the map and subscription index,
        //   allowing a new subscription to be created with the same ID
        // We can then release the lock before attempting to cleanup the messages on the
        // existing subscription.
        DurableSubscriptionItemStream rstream = null;
        synchronized (_consumerDispatchersDurable)
        {
            // Get the consumer dispatcher from the table of durable subscriptions
            ConsumerDispatcher consumerDispatcher =
                            (ConsumerDispatcher) _consumerDispatchersDurable.get(subscriptionId);

            if (consumerDispatcher == null
                || !consumerDispatcher.getConsumerDispatcherState().isReady())
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(
                               tc,
                               "deleteLocalDurableSubscription",
                               "Sub not found " + subscriptionId);
                throw new SIDurableSubscriptionNotFoundException(
                                nls.getFormattedMessage(
                                                        "SUBSCRIPTION_DOESNT_EXIST_ERROR_CWSIP0146",
                                                        new Object[] {
                                                                      subscriptionId,
                                                                      _messageProcessor.getMessagingEngineName() },
                                                        null));
            }

            // If consumers are attached to this consumerDispatcher then reject the delete.
            if (consumerDispatcher.hasConsumersAttached())
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(
                               tc,
                               "deleteLocalDurableSubscription",
                               " SIDestinationLockedException" + subscriptionId);

                throw new SIMPDestinationLockedException(// defect 177277
                nls.getFormattedMessage(
                                        "SUBSCRIPTION_IN_USE_ERROR_CWSIP0152",
                                        new Object[] {
                                                      subscriptionId,
                                                      _messageProcessor.getMessagingEngineName() },
                                        null),
                                SIMPDestinationLockedException.CONSUMERS_ATTACHED);
            }

            // Now find the matching item stream, using the consumer dispatcher
            final SubscriptionStateFilter subStateFilter =
                            new SubscriptionStateFilter();
            subStateFilter.setConsumerDispatcherStateFilter(consumerDispatcher.getConsumerDispatcherState());

            try
            {
                // Find the subscription in the topicspace itemstream
                rstream = (DurableSubscriptionItemStream) _pubsubMessageItemStream.
                                findFirstMatchingReferenceStream(subStateFilter);

                // Check there are no in-doubt receives on this stream.
                // Note this does not prevent us deleting a stream with in-doubt sends, which
                // is correct as producers sending to a subscription shouldn't prevent the
                // subscriber from deleting its subscription.
                if (rstream != null)
                {
                    if (rstream.getStatistics().getRemovingItemCount() != 0)
                    {
                        // There are uncommitted gets on this stream.           // 177786
                        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                            SibTr.exit(
                                       tc,
                                       "deleteLocalDurableSubscription",
                                       "SIDestinationLockedException"
                                                       + rstream.getStatistics().getUnavailableItemCount()
                                                       + ":"
                                                       + subscriptionId);

                        throw new SIMPDestinationLockedException(
                                        nls.getFormattedMessage("SUBSCRIPTION_IN_USE_ERROR_CWSIP0153",
                                                                new Object[] { subscriptionId, _messageProcessor.getMessagingEngineName() },
                                                                null),
                                        SIMPDestinationLockedException.UNCOMMITTED_MESSAGES);
                    }
                }

            } catch (MessageStoreException e)
            {
                // MessageStoreException shouldn't occur so FFDC.
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.deleteLocalDurableSubscription",
                                            "1:1472:1.35.2.4",
                                            this);

                SibTr.exception(tc, e);
                SibTr.error(
                            tc,
                            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                          "1:1481:1.35.2.4",
                                          e,
                                          subscriptionId });

                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "deleteLocalDurableSubscription", e);

                throw new SIResourceException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                                                      "1:1493:1.35.2.4",
                                                                      e,
                                                                      subscriptionId },
                                                        null),
                                e);
            }

            // Mark the subscription to be deleted, and persist the flag
            boolean persistingToBeDeletedFlag = false;
            try
            {
                if (rstream != null)
                {
                    rstream.markAsToBeDeleted();
                    if (rstream.isInStore())
                    {
                        // If we fail while performing the persistent update,
                        // we need to rollback the toBeDeleted mark.
                        persistingToBeDeletedFlag = true;
                        rstream.requestUpdate(
                                        _messageProcessor.getTXManager().createAutoCommitTransaction());
                        persistingToBeDeletedFlag = false;
                    }
                }
            } catch (MessageStoreException e)
            {
                // MessageStoreException shouldn't occur so FFDC.
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.deleteLocalDurableSubscription",
                                            "1:1523:1.35.2.4",
                                            this);

                SibTr.exception(tc, e);
                SibTr.error(
                            tc,
                            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                          "1:1532:1.35.2.4",
                                          e,
                                          subscriptionId });

                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "deleteLocalDurableSubscription", e);

                throw new SIResourceException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                                                      "1:1544:1.35.2.4",
                                                                      e,
                                                                      subscriptionId },
                                                        null),
                                e);
            } finally
            {
                if (persistingToBeDeletedFlag)
                {
                    // We failed while persisting the toBeDeleted flag.
                    // We cannot remove the subscription from the in-memory tables, as otherwise
                    // we could have a duplicate subscription on restart if the user created a new one.
                    // As such we unmark the deletion before throwing the exception, allowing the user
                    // to continue to use this subscription (and attempt the deletion again later).
                    rstream.clearToBeDeleted();
                }
            }

            // Now we can safely remove this subscription from our map, allowing the
            // user to create a new subscription with the same ID.
            // Remove consumerDispatcher from durable subscriptions table
            _consumerDispatchersDurable.remove(subscriptionId);

            // Delete consumerDispatcher ( implicit remove from matchspace) 
            try
            {
                // Delete from matchspace and send proxy message
                consumerDispatcher.deleteConsumerDispatcher(
                                true);
            } catch (SIException e)
            {
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.deleteLocalDurableSubscription",
                                            "1:1580:1.35.2.4",
                                            this);

                SibTr.exception(tc, e);

                // Could not delete consumer dispatcher
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "deleteLocalDurableSubscription", e);
            }

        } // Drop synchronization on durable subscription map

        // If all was well, attempt cleanup of the subscription.
        if (rstream != null)
        {
            // Synchronization and exception handling all dealt with in SubscriptionItemStream
            rstream.deleteIfPossible(true);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "deleteLocalDurableSubscription", subscriptionId);

    }
1288570
@Override
    public boolean registerConsumerSetMonitor(
                                              SIDestinationAddress destinationAddress,
                                              String discriminatorExpression,
                                              ConsumerSetChangeCallback callback)
                    throws SIResourceException,
                    SINotPossibleInCurrentConfigurationException,
                    SIConnectionUnavailableException,
                    SIConnectionDroppedException,
                    SIIncorrectCallException,
                    SICommandInvocationFailedException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "registerConsumerSetMonitor",
                        new Object[] { destinationAddress,
                                      discriminatorExpression,
                                      callback });

        // Check that a listener has been specified
        if (callback == null)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_CONSUMERSETCHANGECALLBACK_CWSIP0667", null, null));
        } else if (destinationAddress == null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_DESTINATIONADDRESS_CWSIP0668", null, null));

        } else if (discriminatorExpression == null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_DISCRIMINATOREXPRESSION_CWSIP0669", null, null));

        }
        boolean areConsumers = false;
        try
        {
            DestinationHandler topicSpace =
                            _destinationManager.getDestination((JsDestinationAddress) destinationAddress, false);

            areConsumers =
                            _messageProcessor.
                                            getMessageProcessorMatching().
                                            registerConsumerSetMonitor(topicSpace,
                                                                       discriminatorExpression,
                                                                       this,
                                                                       callback);
        } catch (Exception e)
        {
            // FFDC
            FFDCFilter
                            .processException(
                                              e,
                                              "com.ibm.ws.sib.processor.impl.ConnectionImpl.registerConsumerSetMonitor",
                                              "1:7378:1.347.1.25",
                                              this);

            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.exception(tc, e);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIErrorException");
            //SIErrorException is RuntimeExcetion hence catching this exception in StaticCATConnection
            throw new SIErrorException(e);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "registerConsumerSetMonitor", Boolean.valueOf(areConsumers));
        return areConsumers;
    }
1033207
public void removeConsumerPointMatchTarget(DispatchableKey consumerPointData)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "removeConsumerPointMatchTarget", consumerPointData);

    // Remove the consumer point from the matchspace
    try
    {
      removeTarget(consumerPointData);
    }
    catch (MatchingException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.matching.MessageProcessorMatching.removeConsumerPointMatchTarget",
        "1:840:1.117.1.11",
        this);

      SibTr.exception(tc, e);

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "removeConsumerPointMatchTarget", "SICoreException");

      SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
      new Object[] {
        "com.ibm.ws.sib.processor.matching.MessageProcessorMatching",
        "1:851:1.117.1.11",
        e });

      throw new SIErrorException(
        nls.getFormattedMessage(
          "INTERNAL_MESSAGING_ERROR_CWSIP0002",
          new Object[] {
            "com.ibm.ws.sib.processor.matching.MessageProcessorMatching",
            "1:859:1.117.1.11",
            e },
          null),
        e);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "removeConsumerPointMatchTarget");
  }
1042349
public void notifyReceiveAllowed(DestinationHandler destinationHandler)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "notifyReceiveAllowed", new Object[] {destinationHandler});
      
    if(isPubSub())
    {
      // Notify consumers on this localization 
      SubscriptionTypeFilter filter = new SubscriptionTypeFilter();  
      filter.LOCAL = Boolean.TRUE;  
      SIMPIterator itr = getSubscriptionIndex().iterator(filter);  
      while(itr.hasNext())  
      {  
        ControllableSubscription subscription = (ControllableSubscription)itr.next(); 
        ConsumerDispatcher cd = (ConsumerDispatcher)subscription.getOutputHandler();
        if(cd != null) 
          cd.notifyReceiveAllowed(destinationHandler); 
      } 
      itr.finished(); 
    }
    else
    {
      //tell the local consumer dispatcher that this destinations receiveAllowed has changed
      ConsumerDispatcher cm = (ConsumerDispatcher)getLocalPtoPConsumerManager();
      if (cm != null)
      {
        cm.notifyReceiveAllowed(destinationHandler);
      }
    }

    //tell the any RME remote consumer dispatchers that this destinations receiveAllowed has changed
    notifyReceiveAllowedRCD(destinationHandler);
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "notifyReceiveAllowed");
  }