Method registerCallbackOnNewExpression

Adds a list of matching consumers in a new entry to the appropriate wildcard
or exact expression table.

@param topicExpression
@param isWildcarded
@param callback
@param matchingConsumers
@return
11445
public boolean registerCallbackOnNewExpression(
    ConnectionImpl connection,
    String topicExpression,
    boolean isWildcarded,
    ConsumerSetChangeCallback callback,
    ArrayList matchingConsumers)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.entry(
          tc,
          "registerCallbackOnNewExpression",
          new Object[] { connection,
                         topicExpression,
                         new Boolean(isWildcarded),
                         callback,
                         matchingConsumers});

    boolean areConsumers = false;

    // Add the new callback to the index, so we can find it when we deregister
    addCallbackToConnectionIndex(connection, topicExpression, isWildcarded, callback);

    // Set up the callback list for this registration
    ArrayList callbackList = new ArrayList();
    WrappedConsumerSetChangeCallback wcb = new WrappedConsumerSetChangeCallback(callback);
    callbackList.add(wcb);

    // Register the monitor in the appropriate table
    RegisteredCallbacks regMonitor =
      new RegisteredCallbacks(callbackList, matchingConsumers);

    if(isWildcarded)
      _registeredWildcardConsumerMonitors.put(topicExpression, regMonitor);
    else
      _registeredExactConsumerMonitors.put(topicExpression, regMonitor);

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "registerCallbackOnNewExpression", new Boolean(areConsumers));
    return areConsumers;
  }
508423
@Check(CheckType.FAST)
	public void checkForbiddenCalls(XAbstractFeatureCall expression) {
		if (this.featureCallValidator.isDisallowedCall(expression)) {
			error(
					MessageFormat.format(
							Messages.SARLValidator_36,
							expression.getFeature().getIdentifier()),
					expression,
					null,
					ValidationMessageAcceptor.INSIGNIFICANT_INDEX,
					FORBIDDEN_REFERENCE);
		}
	}
122494
public TopicConsumer createTopicListener(final String topic, final ConsumerCallback<Message> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new TopicConsumer(connectionFactory, jmsConfig, topic, messageCallback);
    }
624435
public void dispatchConsumerSetChangeCallbackEvent(ConsumerSetChangeCallback consumerSetChangeCallback,boolean isEmpty)
   {
      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "dispatchConsumerSetChangeCallbackEvent",
            new Object[]{consumerSetChangeCallback, isEmpty});

      //Create a new ConsumerSetChangeCallbackThread and dispatch it.
      final ConsumerSetChangeCallbackThread thread = new ConsumerSetChangeCallbackThread(consumerSetChangeCallback,isEmpty);
      dispatchThread(thread);

      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "dispatchConsumerSetChangeCallbackEvent");
   }
773987
@Override
    public void registerStoppableAsynchConsumerCallback(
                                                        StoppableAsynchConsumerCallback callback,
                                                        int maxActiveMessages,
                                                        long messageLockExpiry,
                                                        int maxBatchSize,
                                                        Reliability unrecoverableReliability,
                                                        boolean inLine,
                                                        OrderingContext extendedMessageOrderingContext,
                                                        int maxSequentialFailures,
                                                        long hiddenMessageDelay)
                    throws SISessionUnavailableException, SISessionDroppedException, SIIncorrectCallException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "registerStoppableAsynchConsumerCallback",
                        new Object[] { callback,
                                      maxActiveMessages,
                                      messageLockExpiry,
                                      maxBatchSize,
                                      unrecoverableReliability,
                                      inLine,
                                      extendedMessageOrderingContext,
                                      maxSequentialFailures,
                                      hiddenMessageDelay });

        //register the AsynchConsumer with the LCP
        _localConsumerPoint.registerStoppableAsynchConsumer(
                                                            callback,
                                                            maxActiveMessages,
                                                            messageLockExpiry,
                                                            maxBatchSize,
                                                            unrecoverableReliability,
                                                            inLine,
                                                            (OrderingContextImpl) extendedMessageOrderingContext,
                                                            null, // No callback busy lock is used for non-mediation consumers.
                                                            maxSequentialFailures,
                                                            hiddenMessageDelay);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "registerStoppableAsynchConsumerCallback");
    }
624747
public void setConsumerDispatcher(ConsumerDispatcher consumerDispatcher)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "setConsumerDispatcher", consumerDispatcher);

        this.consumerDispatcher = consumerDispatcher;

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "setConsumerDispatcher");
    }
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");
    }
1299180
public ServerListenerBuilder addStoppingCallbacks(Iterable<Consumer<? super Server>> consumers) {
        requireNonNull(consumers, "consumers");
        for (Consumer<? super Server> consumer : consumers) {
            serverStoppingCallbacks.add(requireNonNull(consumer, "consumer"));
        }
        return this;
    }
965173
@Override
    public void registerAsynchConsumerCallback(
                                               AsynchConsumerCallback callback,
                                               int maxActiveMessages,
                                               long messageLockExpiry,
                                               int maxBatchSize,
                                               OrderingContext context)
                    throws SISessionUnavailableException, SISessionDroppedException,
                    SIErrorException,
                    SIIncorrectCallException
    {
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.entry(CoreSPIConsumerSession.tc, "registerAsynchConsumerCallback",
                        new Object[] { this,
                                      callback,
                                      new Integer(maxActiveMessages),
                                      new Long(messageLockExpiry),
                                      new Integer(maxBatchSize),
                                      context });
        registerAsynchConsumerCallback(callback,
                                       maxActiveMessages,
                                       messageLockExpiry,
                                       maxBatchSize,
                                       null,
                                       false,
                                       context);

        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.exit(CoreSPIConsumerSession.tc, "registerAsynchConsumerCallback");
    }
1304599
boolean hasNoAlternativeWildcardRegistration() {
        return parent == null || PathElement.WILDCARD_VALUE.equals(valueString) || !parent.getChildNames().contains(PathElement.WILDCARD_VALUE);
    }
1023264
protected List<ConsumerHolder> filterConsumersForClass(Class<? extends Consumer> consumerClass) {
        List<ConsumerHolder> consumerHolderSubList = new LinkedList<ConsumerHolder>();
        for (ConsumerHolder consumerHolder : consumerHolders) {
            if (consumerClass.isAssignableFrom(consumerHolder.getConsumer().getClass())) {
                consumerHolderSubList.add(consumerHolder);
            }
        }
        return consumerHolderSubList;
    }
172007
@SafeVarargs
    public final ServerListenerBuilder addStoppingCallbacks(Consumer<? super Server>... consumers) {
        return addStoppingCallbacks(Arrays.asList(consumers));
    }
808242
public void setConsumerDispatcherStateFilter(ConsumerDispatcherState consumerDispatcherState)
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "setConsumerDispatcherStateFilter", consumerDispatcherState);

    this._consumerDispatcherState = consumerDispatcherState;
    this._subscriptionId = null;
    this._destination = null;
    
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "setConsumerDispatcherStateFilter");
  }
954867
@Override
  public Description matchMethod(MethodTree methodTree, VisitorState state) {
    SuggestedFix.Builder fix = null;
    if (isInjectedConstructor(methodTree, state)) {
      for (AnnotationTree annotationTree : methodTree.getModifiers().getAnnotations()) {
        if (OPTIONAL_INJECTION_MATCHER.matches(annotationTree, state)) {
          // Replace the annotation with "@Inject"
          if (fix == null) {
            fix = SuggestedFix.builder();
          }
          fix = fix.replace(annotationTree, "@Inject");
        } else if (BINDING_ANNOTATION_MATCHER.matches(annotationTree, state)) {
          // Remove the binding annotation
          if (fix == null) {
            fix = SuggestedFix.builder();
          }
          fix = fix.delete(annotationTree);
        }
      }
    }
    if (fix == null) {
      return Description.NO_MATCH;
    } else {
      return describeMatch(methodTree, fix.build());
    }
  }
639082
protected void attachBifurcatedConsumer(BifurcatedConsumerSessionImpl consumer)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "attachBifurcatedConsumer", consumer);

        // Create a bifurcated list if required
        if (_bifurcatedConsumers == null)
        {
            synchronized (this)
            {
                if (_bifurcatedConsumers == null)
                    _bifurcatedConsumers = new LinkedList<BifurcatedConsumerSessionImpl>();
            }
        }

        synchronized (_bifurcatedConsumers)
        {
            _bifurcatedConsumers.add(consumer);
        }

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

        try
        {
            _subscriptionIndex.remove(cd);

            // Remove all entries for the consumerDispatcher from the matchspace

            SelectionCriteria[] selCriteria = cd.getConsumerDispatcherState().getSelectionCriteriaList();

            if (selCriteria != null)
            {
                for (int i = 0; i < selCriteria.length; i++)
                {
                    // Remove the CD from the MatchSpace
                    _messageProcessor
                                    .getMessageProcessorMatching()
                                    .removeConsumerDispatcherMatchTarget(cd, selCriteria[i]);
                }
            }

            // 166833.2
            // Calls to the proxy code only need to occur on a non durable delete
            // subscription, or a createDurableSubscription rolled back.
            // It won't be called in the case that a durableSubscription.
            // eventCommittedRemove as the proxy message will have already been sent.

            //716917
            //isNonDurable was passed insted of callProxyCode
            //Due to which always false was being passed when we had durable subscriber, hence
            //the unsubscribe events was not being sent to neighbouring ME's when durable subscriber was unsubscribed
            //Its an Fix to 166833.2
//    sanjay liberty change
//      if (callProxyCode)
//        _messageProcessor.getProxyHandler().unsubscribeEvent(
//          cd.getConsumerDispatcherState(),
//          null,
//          callProxyCode);
//
//      // 175488
            if (isNonDurable)
            {
                // Need to attempt cleanup of the subscription.  If it cant be completed
                // now, it will be attempted by the asynch-deletion thread at a later
                // time.

                try
                {
                    //remove the CD from ME wide hash map maintained for non durable shared consumers in Destination Manager.
                    // All non-durable shared consumer's subscriberID do not start with _NON_DURABLE_NON_SHARED
                    if (!(cd.getConsumerDispatcherState().getSubscriberID().contains("_NON_DURABLE_NON_SHARED")))
                        _destinationManager.getNondurableSharedSubscriptions().remove(cd.getConsumerDispatcherState().getSubscriberID());
                    // Find the subscription in the topicspace itemstream
                    final SubscriptionItemStream subscription =
                                    (
                                    SubscriptionItemStream) _pubsubMessageItemStream
                                                    .findFirstMatchingReferenceStream(
                                                    new SubscriptionFilter(cd));

                    // Remove any items on the subscription reference stream, and remove the stream itself
                    if (subscription != null)
                    {
                        //Mark the subscription as requiring deletion
                        subscription.markAsToBeDeleted();

                        //Drive the deletion of the subscription now.  It may not be possible to
                        //complete it now if there are indoubt messages on the subscription, in
                        //which case the commit/backout of those messages will drive the delete.
                        subscription.deleteIfPossible(true);
                    }
                } catch (MessageStoreException e)
                {
                    FFDCFilter.processException(
                                                e,
                                                "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.dereferenceSubscriptionConsumerDispatcher",
                                                "1:1835: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:1844:1.35.2.4",
                                              e,
                                              cd.getConsumerDispatcherState().getSubscriberID() });

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

                    throw new SIResourceException(
                                    nls.getFormattedMessage(
                                                            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                                                            new Object[] {
                                                                          "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                                                          "1:1856:1.35.2.4",
                                                                          e,
                                                                          cd.getConsumerDispatcherState().getSubscriberID() },
                                                            null),
                                    e);
                }
            } // end of 175488
        } catch (SIException e)
        {
            // Exceptions should not be thrown by proxy handler code
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.dereferenceSubscriptionConsumerDispatcher",
                                        "1:1869: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:1878:1.35.2.4",
                                      e,
                                      cd.getConsumerDispatcherState().getSubscriberID() });

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

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

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "dereferenceSubscriptionConsumerDispatcher");
    }
1035622
@Override
    public void deregisterConsumerSetMonitor(
                                             ConsumerSetChangeCallback callback)
                    throws SINotPossibleInCurrentConfigurationException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "deregisterConsumerSetMonitor",
                        new Object[] { callback });

        _messageProcessor.
                        getMessageProcessorMatching().
                        deregisterConsumerSetMonitor(this, callback);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "deregisterConsumerSetMonitor");
    }