/*
Helper method. Called when all the consumers have been detached.
545145
private void cardinalityChangeConsumersDetached() throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(tc, "cardinalityChangeConsumersDetached");

    // Tell the AIH to start the flush.
    // this call will acquire a lock on AIH such that it is guaranteed that all calls to
    // RDC.put() for messages on this stream will have returned before rejectAll() returns
    // Therefore we can safely go through itemStream and remove the messages that have
    // already been put and reject them.
    _aih.rejectAll(); // reject everything and start flushing the stream

    // then delete all the messages, since know that put() must have completed
    removeAllMessagesItemStream();

    synchronized (consumerPoints)
    {
      // set cardinalityChange to false. Note that any consumers that try to attach right after this will not
      // be able to use the old stream since already initiated the flush
      _cardinalityChange = false;
      // tell the AIH to change its cardinality
      _aih.changeReceiveExclusive(_cardinalityOne);

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

  }
1212303
void removeConsumer(ConsumerSessionImpl consumer) {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "removeConsumer", consumer);

        _consumerList.remove(consumer);

        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "removeConsumer");
    }
555359
public synchronized boolean allEvaluatorsAreClosed() {
    synchronized (this.evaluators) {
      for (final EvaluatorManager eval : this.evaluators.values()) {
        if (!eval.isClosed()) {
          return false;
        }
      }
    }
    return true;
  }
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");
  }
879003
protected void onRemoved() {
        List<TransformerChangeListener> listeners = getAllListeners();
        for (TransformerChangeListener listener : listeners) {
            listener.onOutputChanged(this, new LinkedList<MutableInputColumn<?>>());
            listener.onRemove(this);
        }
    }
191990
protected void deactivateConsumers(List<ConsumerHolder> consumerHolders) {
        synchronized (activationMonitor) {
            for (ConsumerHolder consumerHolder : consumerHolders) {
                consumerHolder.deactivate();
            }
        }
    }
1094256
public void resumeConsumer(int suspendFlag)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "resumeConsumer", this);
    
    ArrayList<AORequestedTick> satisfiedTicks = null;

    synchronized(parent) // AOStream
    {
      try
      {
        // Take the lock
        this.lock();
        try
        {    
          if (_consumerSuspended)
          {
            // clear the bit provided in the _suspendFlags
            _suspendFlags &= ~suspendFlag;
      
            if (_suspendFlags == 0) // No flags set so resume the consumer
            {
              _consumerSuspended = false;
      
              // If the consumer is still active (started) we need
              // to kickstart the consumer back into life to check for more
              // messages
              if(!closed)
                satisfiedTicks = processQueuedMsgs(null);
            }
          }
        }
        finally
        {
          this.unlock();
        }   
      }
      catch(SINotPossibleInCurrentConfigurationException e)
      {
        // No FFDC code needed
        notifyException(e);
      }
      
      if (satisfiedTicks!=null)
      {
        // inform parent about satisfied ticks - outside lock
        int length = satisfiedTicks.size();
        for (int i = 0; i < length; i++)
        {
          AORequestedTick aotick = (AORequestedTick) satisfiedTicks.get(i);
          parent.satisfiedRequest(aotick.tick, aotick.getMessage());
        }
      }
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "resumeConsumer");
  }
838014
@Override
	protected void onClosed() {
		assert promise == null;
		if (pool == client.poolKeepAlive) {
			AddressLinkedList addresses = client.addresses.get(remoteAddress);
			addresses.removeNode(this);
			if (addresses.isEmpty()) {
				client.addresses.remove(remoteAddress);
			}
		}

		// pool will be null if socket was closed by the value just before connection.send() invocation
		// (eg. if connection was in open(null) or taken(null) states)
		pool.removeNode(this);
		pool = null;

		client.onConnectionClosed();
		if (response != null) {
			response.recycle();
			response = null;
		}
	}
1142653
def _cleanup(self):
        '''
        
        '''
        self._pending_consumers = None
        self._consumer_cb = None
        self._get_cb = None
        self._recover_cb = None
        self._cancel_cb = None
        self._return_listener = None
        super(BasicClass, self)._cleanup()
1237047
def stop(self):
        
        while self._consumers:
            consumer = self._consumers.pop()
            consumer.stop()
        self._stopped.set()
        self._connection.close()
154317
public void reclaimAll ()
    {
        // make sure any bound sources are released
        for (PooledSource pooled : _sources) {
            if (pooled.holder != null) {
                pooled.holder.stop();
                pooled.holder.reclaim();
                pooled.holder = null;
            }
        }
    }
34400
void consumerSessionStopped()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "consumerSessionStopped");      

    try
    {
      if (asynchConsumerCallback instanceof StoppableAsynchConsumerCallback)
      {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "Entering asynchConsumerCallback.consumerSessionStopped",
                       asynchConsumerCallback);      

        ((StoppableAsynchConsumerCallback)asynchConsumerCallback).consumerSessionStopped();

        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "Exiting asynchConsumerCallback.consumerSessionStopped");      
      }
    }
    catch (Throwable e)
    {
      //Catch any exceptions thrown by consumeMessages.
      FFDCFilter.processException(
       e,
       "com.ibm.ws.sib.processor.impl.AsynchConsumer.consumerSessionStopped",
       "1:264:1.37",
       this);

      //Trace the exception but otherwise swallow it since it was not
      //a failure in MP code or code upon which the MP is critically
      //dependent.
      if (e instanceof Exception)
        SibTr.exception(tc, (Exception)e);
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Exception occurred in consumerSessionStopped " + e);
         
      // We're not allowed to swallow this exception - let it work its way back to
      // the threadpool
      if(e instanceof ThreadDeath)
      {     
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
          SibTr.exit(tc, "consumerSessionStopped", e); 
        throw (ThreadDeath)e;
      }
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "consumerSessionStopped");
  }
1362136
private void closeRemainingConsumers()
    {
        List<AbstractMessageConsumer> consumersToClose = new ArrayList<>(consumersMap.size());
        synchronized (consumersMap)
        {
            consumersToClose.addAll(consumersMap.values());
        }
        for (int n = 0 ; n < consumersToClose.size() ; n++)
        {
            MessageConsumer consumer = consumersToClose.get(n);
            log.debug("Auto-closing unclosed consumer : "+consumer);
            try
            {
                consumer.close();
            }
            catch (Exception e)
            {
                log.error("Could not close consumer "+consumer,e);
            }
        }
    }
679889
def on_consumer_cancelled(self, method_frame):
        
        _logger.info('Consumer was cancelled remotely, shutting down: %r', method_frame)
        if self._channel:
            self._channel.close()
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");
    }
1047748
def _flush(self):
        
        for consumer in self.consumers:
            if not getattr(consumer, "closed", False):
                consumer.flush()
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)