Dispatches a thread which will call the stoppableConsumerSessionStopped method on consumerSessionProxy.

@param consumerSessionProxy
501471
public void dispatchStoppableConsumerSessionStopped(ConsumerSessionProxy consumerSessionProxy)
   {
      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "dispatchStoppableConsumerSessionStopped", consumerSessionProxy);

      //Create a new StoppableAsynchConsumerCallbackThread and dispatch it.
      final StoppableAsynchConsumerCallbackThread thread = new StoppableAsynchConsumerCallbackThread(consumerSessionProxy);
      dispatchThread(thread);

      if(TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "dispatchStoppableConsumerSessionStopped");
   }
1140648
@Override
    public void runIsolatedAsynch(boolean deliverImmediately) throws SIIncorrectCallException, SISessionUnavailableException, SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "runIsolatedAsynch", new Object[] { this, Boolean.valueOf(deliverImmediately) });

        synchronized (_asynchConsumerBusyLock)
        {
            // Lock the consumer session while we check that it is in a valid
            // state
            this.lock();
            try
            {
                // Only valid if the consumer session is still open
                checkNotClosed();

                //if there is no callback registered, throw an exception
                if (!_asynchConsumerRegistered)
                {
                    SIIncorrectCallException e =
                                    new SIIncorrectCallException(
                                                    nls.getFormattedMessage(
                                                                            "ASYNCH_CONSUMER_ERROR_CWSIP0175",
                                                                            new Object[] { _consumerDispatcher.getDestination().getName(),
                                                                                          _messageProcessor.getMessagingEngineName() },
                                                                            null));

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

                    throw e;
                }
                // we can't do an isolated run if the LCP is not in a stopped state
                // so throw an exception
                if (!_stopped)
                {
                    SIIncorrectCallException e =
                                    new SIIncorrectCallException(
                                                    nls.getFormattedMessage(
                                                                            "ASYNCH_CONSUMER_RUN_ERROR_CWSIP0176",
                                                                            new Object[] { _consumerDispatcher.getDestination().getName(),
                                                                                          _consumerDispatcher.getMessageProcessor().getMessagingEngineName() },
                                                                            null));

                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                        SibTr.exit(tc, "runIsolatedAsynch", e);
                    throw e;
                }

                // If the consumer has been stopped because the destination is not
                // allowing consumers to get messages then we simply return.
                if (_stoppedForReceiveAllowed)
                {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                        SibTr.exit(tc, "runIsolatedAsynch", "Receive not allowed");

                    return;
                }
            } // synchronized (this)
            finally
            {
                this.unlock();
            }
        } // synchronized (asynchConsumer)

        //if we get this far then if deliverImmediately is set then this
        //implies that the callback should be inline
        if (deliverImmediately)
        {
            //run the asynch inline
            try
            {
                runAsynchConsumer(true);
            } catch (Throwable e)
            {
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.JSLocalConsumerPoint.runIsolatedAsynch",
                                            "1:3182:1.22.5.1",
                                            this);

                SibTr.exception(tc, e);

                try
                {
                    // Since the asynchConsumer has experienced an error of some kind, the best form
                    // of cleanup is to close down the session. This ensures any listeners get notified
                    // and can retry at some point later.
                    _consumerSession.close();
                    // don't notify asynchconsumer as we are been called inline
                } catch (Exception ee)
                {
                    // No FFDC code needed
                    SibTr.exception(tc, ee);
                }

                if (e instanceof ThreadDeath)
                {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                        SibTr.exit(tc, "runIsolatedAsynch", e);
                    throw (ThreadDeath) e;
                }

                SISessionDroppedException sessionDroppedException = new SISessionDroppedException(
                                nls.getFormattedMessage("CONSUMER_CLOSED_ERROR_CWSIP0177"
                                                        , new Object[] { _consumerDispatcher.getDestination().getName(),
                                                                        _messageProcessor.getMessagingEngineName() }
                                                        , null), e);

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

                //inline call so throwing exception rather than notifying
                throw sessionDroppedException;
            }
        }
        else
        {
            //start up a new thread (from the MP's thread pool)
            //to deliver the message asynchronously
            try
            {
                _messageProcessor.startNewThread(new AsynchThread(this, true));
            } catch (InterruptedException e)
            {
                // No FFDC code needed

                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, "runIsolatedAsynch", e);
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "runIsolatedAsynch");
    }
976822
private void setAsyncListener(MessageListener listener) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setAsyncListener", listener);

        consumer = new Consumer(listener, session, sessionAckMode);
        try {
            // ConsumerSession must be stopped to register the callback
            coreConsumerSession.stop();
            int batchSize = 1;
            boolean deliverImmediately = false;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(this, tc, "Registering async callback for consumer" + consumer);
            // the register call will replace any existing callback, so no need for
            // separate deregister call.
            coreConsumerSession.registerAsynchConsumerCallback(consumer,
                                                               0, // maxActiveMessages - no limit
                                                               0, // messageLockExpiry - no expiry
                                                               batchSize,
                                                               session.getOrderingContext()
                            );

            // Deal with the case where the listener is set after the connection
            // is started. Note that this will handle the restarting of the core
            // consumer session if we paused it earlier in this block.
            if (session.getState() == STARTED)
                coreConsumerSession.start(deliverImmediately);
        } catch (SIIncorrectCallException sie) {
            // No FFDC code needed
            // d222942 review. Default message ok.
            // d238447 FFDC Review. FFDC here, see comment in next catch for more detail.
            throw (JMSException) JmsErrorUtils.newThrowable(JMSException.class,
                                                            "EXCEPTION_RECEIVED_CWSIA0085",
                                                            new Object[] { sie, "JmsMsgConsumerImpl.setAsyncListener" },
                                                            sie,
                                                            "JmsMsgConsumerImpl.setAsyncListener#1",
                                                            this,
                                                            tc);
        } catch (SIException sie) {
            // No FFDC code needed
            // The exceptions thrown by stop, registerAsynchConsumerCallback and start are mostly
            // the "connection lost" family. registerAsynchConsumerCallback can also throw an
            // SIIncorrectCallException if the coreConsumerSession hasn't been stopped, but that
            // shouldn't be possible in the code above. Default message ok.
            // d238447 FFDC Review. SIIncorrectCall separated into its own catch block above,
            //   no FFDCs for this group.
            throw (JMSException) JmsErrorUtils.newThrowable(JMSException.class,
                                                            "EXCEPTION_RECEIVED_CWSIA0085",
                                                            new Object[] { sie, "JmsMsgConsumerImpl.setAsyncListener" },
                                                            sie,
                                                            null, // null probeId = no FFDC
                                                            this,
                                                            tc);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "setAsyncListener");
    }
1078382
def stop_consumer(self):
        
        try:
            LOGGER.info('Shutting down the consumer')
            self.consumer.shutdown()
        except AttributeError:
            LOGGER.debug('Consumer does not have a shutdown method')
220643
private void checkReceiveState()
                    throws SIIncorrectCallException, SISessionUnavailableException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "checkReceiveState");

        // Only valid if the consumer session is still open
        checkNotClosed();

        //if there is an AsynchConsumerCallback registered, throw an exception
        if (_asynchConsumerRegistered)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "checkReceiveState", "asynchConsumerRegistered == true ");

            throw new SIIncorrectCallException(
                            nls.getFormattedMessage(
                                                    "RECEIVE_USAGE_ERROR_CWSIP0171",
                                                    new Object[] { _consumerDispatcher.getDestination().getName(),
                                                                  _messageProcessor.getMessagingEngineName() },
                                                    null));
        }
        // If we already have the waiting bit set it implies another
        // thread is sat waiting in a receive for this consumer (we release
        // the consumer lock while waiting so this is possible).
        else if (_waiting)
        {
            SIIncorrectCallException e =
                            new SIIncorrectCallException(
                                            nls.getFormattedMessage(
                                                                    "RECEIVE_USAGE_ERROR_CWSIP0178",
                                                                    new Object[] { _consumerDispatcher.getDestination().getName(),
                                                                                  _messageProcessor.getMessagingEngineName() },
                                                                    null));

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "checkReceiveState", "receive already in progress");

            throw e;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "checkReceiveState");
    }
503224
public void closeConsumers() throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "closeConsumers");

    //Tell any destinations that target this destination to close their consumers
    if (aliasesThatTargetThisDest != null)
    {
      synchronized(aliasesThatTargetThisDest)
      {
        Iterator i = aliasesThatTargetThisDest.iterator();
        while (i.hasNext())
        {
          AbstractAliasDestinationHandler abstractAliasDestinationHandler = 
            (AbstractAliasDestinationHandler) i.next();
          abstractAliasDestinationHandler.closeConsumers();
        }
      }
    }
    
    // Close all consumers
    if (isPubSub())
    {
      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();

        //Look for subscriptions made through this destination (as opposed to
        //through an alias)
        if (cd.getConsumerDispatcherState().getTopicSpaceUuid().equals(getUuid()))
        {
          //Close the consumers of the subscription  
          cd.closeAllConsumersForDelete(this);
        }
      }

      // Remove the ProxyHandlers attached to this destination
      // sanjay liberty change
      //messageProcessor.getProxyHandler().topicSpaceDeletedEvent(this);
      
      //Delete any durable subscriptions if they are made through an alias
      //in-case the underlying topicspace is not being deleted
      if (isAlias())
      {
        itr = getSubscriptionIndex().iterator(filter);  
        while(itr.hasNext())  
        {  
          ControllableSubscription subscription = (ControllableSubscription)itr.next(); 
          ConsumerDispatcher cd = (ConsumerDispatcher)subscription.getOutputHandler();
          ConsumerDispatcherState subState = cd.getConsumerDispatcherState();

          //Look for subscriptions made through this alias
          if ((subState.getTopicSpaceUuid().equals(getUuid())) &&
              (cd.isDurable()))
          {
            try {
              deleteDurableSubscription(subState.getSubscriberID(), subState.getDurableHome());
            } catch (SIException e) 
            {
              FFDCFilter.processException(
                  e,
                  "com.ibm.ws.sib.processor.impl.AbstractDestinationHandler.closeConsumers",
                  "1:347:1.57",
                  this);

              SibTr.exception(tc, e);
            }
          }
        } 
        itr.finished();
        
      }

    }
    else
    {
      LocalizationPoint ptoPMessageItemStream = getQueuePoint(messageProcessor.getMessagingEngineUuid());
      if (ptoPMessageItemStream != null)
      {
        ConsumerDispatcher consumerManager =
          (ConsumerDispatcher)ptoPMessageItemStream.getConsumerManager();
        consumerManager.closeAllConsumersForDelete(this);
      }
    }
    

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

    return;
  }
728023
synchronized void add(ConsumerSessionImpl consumer)
  {
    consumer.setId(_consumerCount);
    
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "add", new Long(consumer.getIdInternal()) );

    _consumers.put(new Long(_consumerCount), consumer);

    _consumerCount++;

    if (tc.isEntryEnabled())
      SibTr.exit(tc, "add");
  }
315806
def on_consumer_cancelled(self, _frame):
        
        self.logger.debug('Consumer has been cancelled')
        if self.is_shutting_down:
            self.channel.close()
        else:
            self.set_state(self.STATE_CONNECTED)
594777
public ConsumerSession getConsumerSession()
  throws SISessionUnavailableException, SISessionDroppedException,
         SIIncorrectCallException
  {
    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
      SibTr.entry(CoreSPILockedMessageEnumeration.tc, "getConsumerSession",
        new Integer(hashCode()));

    checkValidState("getConsumerSession");

    localConsumerPoint.checkNotClosed();

    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
      SibTr.exit(CoreSPILockedMessageEnumeration.tc, "getConsumerSession", consumerSession);

    return consumerSession;
  }
556437
synchronized ConsumerSessionImpl get(long id)
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "get", new Long(id));

    ConsumerSessionImpl consumer  = null ;
    
    if( _messageProcessor.isStarted() )
    {
      consumer = (ConsumerSessionImpl) _consumers.get(new Long(id));
    }

    if (tc.isEntryEnabled())
      SibTr.exit(tc, "get", consumer);

    return consumer;
  }
576978
@Override
	public void unregisterConsumer(LocalMessageConsumer consumer)
    {
        super.unregisterConsumer(consumer);
        if (!consumer.isDurable())
        {
            log.debug("Removing non-durable subscription "+consumer.getSubscriberId());
            subscriptionsLock.writeLock().lock();
            try
    		{
            	LocalTopicSubscription subscription = subscriptionMap.remove(consumer.getSubscriberId());
            	if (subscription != null)
            		removeSubscription(subscription);
    		}
            finally
            {
            	subscriptionsLock.writeLock().unlock();
            }
        }
    }
250126
public void deregisterConsumerSetMonitor(ConnectionImpl connection,
                                           ConsumerSetChangeCallback callback)
  throws SINotPossibleInCurrentConfigurationException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "deregisterConsumerSetMonitor", new Object[] { connection, callback });

    // Deregister under a lock on the targets table
    synchronized(_targets)
    {
      _consumerMonitoring.deregisterConsumerSetMonitor(connection, callback);
    }

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

  }
1087777
void removeConsumer(JmsMsgConsumerImpl consumer) throws JMSException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "removeConsumer", consumer);

        // if DUPS_OK_ACKNOWLEDGE is in use, commit any messages delivered in the
        // last incomplete batch (if the session is closing it will already have
        // done this and the count of uncommitted messages will be zero)
        if ((acknowledgeMode == Session.DUPS_OK_ACKNOWLEDGE) && (uncommittedReceiveCount > 0)) {
            commitTransaction();
        }
        syncConsumers.remove(consumer);
        asyncConsumers.remove(consumer);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "removeConsumer");
    }
754811
@Override
    public ConsumerSession createConsumerSessionForDurableSubscription(String subscriptionName,
                                                                       boolean enableReadAhead,
                                                                       Reliability unrecoverableReliability,
                                                                       boolean bifurcatable)

                    throws SIConnectionUnavailableException, SIConnectionDroppedException,
                    SIErrorException,
                    SIDurableSubscriptionNotFoundException, SIDurableSubscriptionMismatchException,
                    SIDestinationLockedException, SIIncorrectCallException, SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConnection.tc.isEntryEnabled())
            SibTr.entry(
                        CoreSPIConnection.tc,
                        "createConsumerSessionForDurableSubscription",
                        new Object[] {
                                      this,
                                      subscriptionName,
                                      Boolean.valueOf(enableReadAhead),
                                      unrecoverableReliability,
                                      Boolean.valueOf(bifurcatable) });

        // Do not need to check that the destination information is correct.
        // As we are using the subscription data straight from the named subscription

        // Find the ConsumerDispatcher for this subscription
        // Note that it must be homed on this ME
        HashMap durableSubs = _destinationManager.getDurableSubscriptionsTable();

        ConsumerDispatcher cd = null;
        synchronized (durableSubs)
        {
            //Look up the consumer dispatcher for this subId in the system durable subs list
            cd =
                            (ConsumerDispatcher) durableSubs.get(subscriptionName);

            // Check that the durable subscription exists
            if (cd == null)
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription");

                throw new SIDurableSubscriptionNotFoundException(
                                nls.getFormattedMessage(
                                                        "SUBSCRIPTION_DOESNT_EXIST_ERROR_CWSIP0072",
                                                        new Object[] { subscriptionName,
                                                                      _messageProcessor.getMessagingEngineName() },
                                                        null));
            }
        }

        //Get the destinationHandler from the ConsumerDispatcher
        //Note that we know it is PubSub  
        DestinationHandler destination = cd.getDestination();
        SIDestinationAddress destinationAddress = SIMPUtils.createJsDestinationAddress(
                                                                                       destination.getName(),
                                                                                       _messageProcessor.getMessagingEngineUuid());

        //Get the state object representing the subscription from the ConsumerDispatcher
        ConsumerDispatcherState subState = cd.getConsumerDispatcherState();

        ConsumerSessionImpl consumer = null;

        // Synchronize on the close object, we don't want the connection closing
        // while we try to add the subscription.
        synchronized (this)
        {
            // See if this connection has been closed
            checkNotClosed();

            //create the consumer session for this subscription
            try
            {
                consumer =
                                new ConsumerSessionImpl(destination,
                                                destinationAddress,
                                                subState,
                                                this,
                                                enableReadAhead,
                                                false,
                                                unrecoverableReliability,
                                                bifurcatable,
                                                true,
                                                false);
            } catch (SINotPossibleInCurrentConfigurationException e)
            {
                // No FFDC code needed
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(e);
            } catch (SISessionUnavailableException e)
            {
                // FFDC
                FFDCFilter
                                .processException(
                                                  e,
                                                  "com.ibm.ws.sib.processor.impl.ConnectionImpl.createConsumerSessionForDurableSubscription",
                                                  "1:2129:1.347.1.25",
                                                  this);

                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                          "1:1809:1.285",
                                          e });

                // This should never be thrown
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                                                      "1:2146:1.347.1.25",
                                                                      e },
                                                        null),
                                e);
            } catch (SITemporaryDestinationNotFoundException e)
            {
                // FFDC
                FFDCFilter
                                .processException(
                                                  e,
                                                  "com.ibm.ws.sib.processor.impl.ConnectionImpl.createConsumerSessionForDurableSubscription",
                                                  "1:2159:1.347.1.25",
                                                  this);

                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                          "1:2165:1.347.1.25",
                                          e });

                // This should never be thrown
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                                                      "1:2176:1.347.1.25",
                                                                      e },
                                                        null),
                                e);
            } catch (SINonDurableSubscriptionMismatchException e)
            {
                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                          "1:1953:1.347.1.25",
                                          e });

                // This should never be thrown
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "createConsumerSessionForDurableSubscription", "SIErrorException");
                throw new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.impl.ConnectionImpl",
                                                                      "1:1964:1.347.1.25",
                                                                      e },
                                                        null),
                                e);
            }

            synchronized (_consumers)
            {
                //store a reference
                _consumers.add(consumer);
            }

            _messageProcessor.addConsumer(consumer);
        }

        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConnection.tc.isEntryEnabled())
            SibTr.exit(CoreSPIConnection.tc, "createConsumerSessionForDurableSubscription", consumer);
        return consumer;
    }
191990
protected void deactivateConsumers(List<ConsumerHolder> consumerHolders) {
        synchronized (activationMonitor) {
            for (ConsumerHolder consumerHolder : consumerHolders) {
                consumerHolder.deactivate();
            }
        }
    }
801937
void sessionError(final SibRaMessagingEngineConnection connection,
            final ConsumerSession session, final Throwable throwable) {

        /*
         * Do nothing - sessions are only used synchronously so the caller will
         * be notified
         */

        final String methodName = "sessionError";
        if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled()) {
            SibTr.entry(this, TRACE, methodName, new Object[] { connection,
                    session, throwable });
            SibTr.exit(this, TRACE, methodName);
        }

    }
491396
private void checkNotClosed() throws SISessionUnavailableException 
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "checkNotClosed");

    // Check that the consumer session isn't closed
    _consumerSession.checkNotClosed();
    
    // Now check that this consumer hasn't closed.
    synchronized (this)
    {
      if(_closed)
      {
        SISessionUnavailableException e =
          new SISessionUnavailableException(
            nls.getFormattedMessage(
              "CONSUMER_CLOSED_ERROR_CWSIP0177",
              new Object[] { _localConsumerPoint.getConsumerManager().getDestination().getName(),
                             _localConsumerPoint.getConsumerManager().getMessageProcessor().getMessagingEngineName()},                
              null));

        if (tc.isEntryEnabled())
          SibTr.exit(tc, "checkNotClosed", "consumer closed");

        throw e;
      }
    }
      
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "checkNotClosed");
  }