/* (non-Javadoc)
@see com.ibm.ws.sib.processor.impl.interfaces.DestinationHandler#closeConsumers()
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;
  }
282332
public void close() {
		synchronized (receivers) {
			Iterator i = receivers.iterator();
			while (i.hasNext()) {
				ReceiverQueue r = (ReceiverQueue) i.next();
				r.onTransmissionClose();
			}
			closed = true;
			receivers.clear();
		}
	}
515315
public boolean close(int closedReason, SIBUuid8 qpoint)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "close", 
          new Object[] { Integer.valueOf(closedReason), 
                         qpoint});

    /*
     * Indicate that the consumer is closed, due to the local destination
     * being deleted/receiveExclusive/unreachable
     */
    this.closedReason = closedReason;
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "close", true);
    return true;
  }
156703
@Override
    public void close() throws SIResourceException, SINotPossibleInCurrentConfigurationException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "close", this);

        // On closing, we need to run the async consumer a final time to process
        // any messages it has attached.
        //
        // We need to take the async consumer lock so we don't step on a currently
        // running consumer (so that running asyncs can complete with all their
        // resources available - i.e. without object closed exceptions - as per the
        // JMS 1.1 spec).
        //
        // Taking this lock also prevents other invocations of close which take
        // place before this invocation has completed from proceeding (as per the
        // JMS 1.1 spec).
        //
        // We also taken the consumer session lock for most of the method.  In
        // theory, we could release this lock while running the async consumer,
        // but this makes things a little more complicated and there seems little
        // point when we're closing down anyway.
        //
        // The async consumer could concievably spin forever and prevent the close.
        // This is an accepted fact.

        // We need to get any asynch thread to stop looping round the queue to
        // allow us to have a chance of closing this consumer. We have to do this outside
        // of the asynch lock as that will be held by the other thread at this time.
        _interruptConsumer = true;

        synchronized (_asynchConsumerBusyLock)
        {
            if (_closed || _closing)
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "close", "Already Closed");
                return;
            }

            _lockRequested = true;

            // Lock the KeyGroup if we have one before we lock this consumer
            if (_keyGroup != null)
                _keyGroup.lock();

            try
            {
                this.lock();
                try
                {
                    _lockRequested = false;

                    _consumerKey.stop();

                    // We don't want new messages to be attached if we're closing.
                    unsetReady();

                    // We need to release the LCP lock while we process attached messages
                    // but first we mark the consumer as closing to stop other calls succeeding
                    _closing = true;
                } finally
                {
                    this.unlock();
                }
            } finally
            {
                if (_keyGroup != null)
                    _keyGroup.unlock();
            }

            // Process any attached messages already on the async consumer.
            if (_asynchConsumerRegistered)
            {
                // If we reach here and an async consumer is running, close() must
                // have been called from the async consumer callback as we have been
                // able to acquire the asynchConsumer lock.
                //
                // Therefore, do not erroneously recursively invoke the callback.
                if (!_asynchConsumer.isAsynchConsumerRunning())
                {
                    try
                    {
                        processAttachedMsgs();
                    } catch (SISessionDroppedException e)
                    {
                        //No FFDC code needed
                        //essentially the session is already closed but we'll try to carry
                        //cleaning up
                    }
                }

                // If we are a member of an ordering group then we need to leave it
                // before we close
                if (_keyGroup != null)
                {
                    _consumerKey.leaveKeyGroup();
                    _keyGroup = null;
                }
            }

            //Now take the lock back and unlock any locked messages and kick out
            // any waiting receiver
            this.lock();
            try
            {
                synchronized (_allLockedMessages)
                {
                    try
                    {
                        // Unlock any remaining locked messages
                        _allLockedMessages.unlockAll();
                    } catch (SIMPMessageNotLockedException e)
                    {
                        // No FFDC code needed
                        // This exception has occurred beause someone has deleted the
                        // message(s). Ignore this exception as it is unlocked anyway
                    } catch (SISessionDroppedException e)
                    {
                        //No FFDC code needed
                        //essentially the session is already closed but we'll try to carry
                        //cleaning up
                    }

                    // Mark this consumer session (point) as closed
                    _closed = true;
                }

                // If we have a synchronous receive we need to wake it up
                if (_waiting)
                {
                    _waiter.signal();
                }
            } // synchronized(this)
            finally
            {
                this.unlock();
            }

            _interruptConsumer = false;

            // If we've managed to interupt another thread that's currently checking for messages
            // to give to a consumer (runAsynchConsumer) then we need to wait until they realise
            // that we've interupted them before we return. Otherwise we'll have a dangling thread
            // that's technically 'working', even though the consumer is closed
            if (_runningAsynchConsumer && !_asynchConsumer.isAsynchConsumerRunning())
            {
                _runAsynchConsumerInterupted = true;
                try
                {
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                        SibTr.debug(this, tc, "Waiting for runAsynchConsumer to complete");

                    _asynchConsumerBusyLock.wait();
                } catch (InterruptedException e)
                {
                    // No FFDC code needed
                }
            }
        } // synchronized(asynch)

        // Now we no longer hold a lock, unlock any messages we happen
        // to have hidden for this consumer (SIB0115)
        unlockAllHiddenMessages();

        //detach from the CD - we CANNOT hold any consumer locks at this point!
        _consumerKey.detach();

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "close");
    }
32894
protected void enableConsumers(List<ConsumerHolder> consumerHolders) throws IOException {
        checkPreconditions(consumerHolders);
        try {
            for (ConsumerHolder consumerHolder : consumerHolders) {
                consumerHolder.enable();
            }
        } catch (IOException e) {
            LOGGER.error("Failed to enable consumers - disabling already enabled consumers");
            disableConsumers(consumerHolders);
            throw e;
        }
    }
1330923
public void closeRemoteConsumer(String key)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "closeRemoteConsumer", key);

    AnycastInputHandler aih = _anycastInputHandlers.get(key);

    if (aih != null)
    {
      ConsumerDispatcher rcd = aih.getRCD();

      if (rcd != null) 
      {
        rcd.closeAllConsumersForDelete(_baseDestinationHandler);
      }
    } 
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "closeRemoteConsumer");

    return;
  }
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;
    }
570953
public void close() 
  throws SIResourceException, SIConnectionLostException,
         SIErrorException
  {
    if (CoreSPIBifurcatedConsumerSession.tc.isEntryEnabled()) 
      SibTr.entry(CoreSPIBifurcatedConsumerSession.tc, "close", this);
    
    boolean closedNow =_close();
    
    // If this close operation actually closed the consumer, then
    // remove the consumer from the list stored in the connection
    // Also remove the Consumer from the list store in the Consumer
    // Session.
    if (closedNow)
    {
      _connection.removeBifurcatedConsumerSession(this);
    }
    
    if (CoreSPIBifurcatedConsumerSession.tc.isEntryEnabled()) 
      SibTr.exit(CoreSPIBifurcatedConsumerSession.tc, "close");
  }
1220328
public java.util.List<Consumer> getConsumers() {
        if (consumers == null) {
            consumers = new com.amazonaws.internal.SdkInternalList<Consumer>();
        }
        return consumers;
    }
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");

  }
844486
public void setConsumer(org.apache.kafka.clients.consumer.KafkaConsumer<Object, Object> consumer) {
        this.consumer = consumer;
    }
333773
void _close()
                    throws SIResourceException, SIConnectionLostException,
                    SIErrorException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "_close");

        try
        {
            //close the LCP
            _localConsumerPoint.close();
        } catch (SINotPossibleInCurrentConfigurationException e)
        {
            // No FFDC code needed

            //Do nothing ... probably means that the Destination is being deleted
            //and so the LCP will be closed anyway
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(tc, "_close", e);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "_close");
    }
199227
public void registerConsumer( LocalMessageConsumer consumer )
    {
    	consumersLock.writeLock().lock();
    	try
    	{
    		localConsumers.add(consumer);
    	}
    	finally
    	{
    		consumersLock.writeLock().unlock();
    	}
    }
738851
public void registerCloseables(final Collection<Closeable> registry, Closeable... closeables) {

        registry.addAll(Arrays.asList(closeables));

        addListener(resource -> {
            for (Closeable closeable : closeables) {
                if (closeable == RedisChannelHandler.this) {
                    continue;
                }

                try {
                    if (closeable instanceof AsyncCloseable) {
                        ((AsyncCloseable) closeable).closeAsync();
                    } else {
                        closeable.close();
                    }
                } catch (IOException e) {
                    if (debugEnabled) {
                        logger.debug(e.toString(), e);
                    }
                }
            }

            registry.removeAll(Arrays.asList(closeables));
        });
    }
1242547
def close
      @consumers.each { |consumer| consumer.close } if @consumers
      @consumers = []

      @sessions.each { |session| session.close } if @sessions
      @session=[]

      @jms_connection.close if @jms_connection
    end
515882
@SuppressWarnings("WeakerAccess")
    @MainThread
    public void removeObservers(@NonNull final LifecycleOwner owner) {
        assertMainThread("removeObservers");
        for (Map.Entry<Observer<T>, ObserverWrapper> entry : mObservers) {
            if (entry.getValue().isAttachedTo(owner)) {
                removeObserver(entry.getKey());
            }
        }
    }
545015
public void removeConsumer(MonitoredConsumer mc)
  {
    if (tc.isEntryEnabled()) 
      SibTr.entry(tc, 
                  "removeConsumer", 
                  new Object[]{mc});
      
    boolean selector = mc.isSelector();
    boolean isWildcarded = mc.isWildcarded();
    String topic = mc.getTopic();
    ArrayList consumerList = 
      _subscriptionRegistrar.getConsumerListForExpression(topic, selector, isWildcarded);
        
    //TODO: handle empty return
        
    if (tc.isDebugEnabled())
      SibTr.debug(tc, "Found existing entry with list: " + consumerList +
                      ", of size: " + consumerList.size());
        
    // Remove the consumer      
    boolean removed = consumerList.remove(mc);
    if(!removed)
    {
      // Couldn't find the consumer. This means we have an internal 
      // inconsistency in our tables, so throw an SIErrorException
    
      if (tc.isEntryEnabled()) 
        SibTr.exit(tc, "removeConsumer", "SIErrorException");
      
      SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
        new Object[] {
          "com.ibm.ws.sib.processor.matching.ConsumerMonitoring",
          "1:248:1.7"
          });  
        
      throw new SIErrorException(
        nls.getFormattedMessage(
          "INTERNAL_MESSAGING_ERROR_CWSIP0001",
          new Object[] {
            "com.ibm.ws.sib.processor.matching.ConsumerMonitoring",
            "1:256:1.7"
          },
          null));      
    }          
                 
    // Only remove the row in the table from the subscription registrar
    // if this was the last consumer in the list
    if(consumerList.isEmpty())
    {
      _subscriptionRegistrar.removeConsumerListForExpression(topic,selector,isWildcarded);                  
    }
        
    // 1. Need to remove sub references from consumer monitor tables
    // 2. Drive callbacks where appropriate
    ArrayList exactMonitorList = (ArrayList)mc.getMatchingExactMonitorList();
    ArrayList wildcardMonitorList = (ArrayList)mc.getMatchingWildcardMonitorList();
        
    _consumerMonitorRegistrar.
      removeConsumerFromRegisteredMonitors(mc, exactMonitorList, wildcardMonitorList);
          
    if (tc.isEntryEnabled()) SibTr.exit(tc, "removeConsumer"); 
  }