Purges all messages from a message channel. Prerequisite is that channel is
of type {@link QueueChannel}.

@param channel
425226
private void purgeChannel(MessageChannel channel) {
        if (channel instanceof QueueChannel) {
            List<Message<?>> messages = ((QueueChannel)channel).purge(messageSelector);
            
            if (log.isDebugEnabled()) {
                log.debug("Purged channel " + ((QueueChannel)channel).getComponentName() + " - removed " + messages.size() + " messages");
            }
        }
    }
1346677
public void init(SocketChannel channel)
  {
    _channel = channel;
    _readBuffer.clear();
    _readBuffer.flip();
    
    _needsFlush = false;
  }
341324
def flush(self, queue_name):
        
        for name in (queue_name, dq_name(queue_name), xq_name(queue_name)):
            self.channel.queue_purge(name)
80354
private void removeChannel(String channelName) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "removeChannel", channelName);
        // Neither of the thrown exceptions are permanent failures: 
        // they usually indicate that we're the victim of a race.
        // If the CFW is also tearing down the chain at the same time 
        // (for example, the SSL feature was removed), then this could
        // fail.
        ChannelFramework cfw = CommsClientServiceFacade.getChannelFramewrok();

        try {
            if (cfw.getChannel(channelName) != null)
                cfw.removeChannel(channelName);
        } catch (ChannelException e) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                SibTr.debug(this, tc, "Error removing channel " + channelName, e);
            }
        } catch (ChainException e) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                SibTr.debug(this, tc, "Error removing channel " + channelName, e);
            }
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "removeChannel");
        }
    }
847629
function purgeADQueue (channel, connectionName, options, messages) {
  const name = options.uniqueName || options.name;
  return new Promise(function (resolve, reject) {
    const messageCount = messages.messages.length;
    if (messageCount > 0) {
      log.info(`Purge operation for queue '${options.name}' on '${connectionName}' is waiting for resolution on ${messageCount} messages`);
      messages.once('empty', function () {
        channel.purgeQueue(name)
          .then(
            result => resolve(result.messageCount),
            reject
          );
      });
    } else {
      channel.purgeQueue(name)
        .then(
          result => resolve(result.messageCount),
          reject
        );
    }
  });
}
238098
private void closeChannel() {
        if (this.channel != null) {
            if (this.channel.isConnected()) {
                try {
                    this.channel.disconnect();
                } catch (IOException e) {
                    logger.warn(e.getMessage(), e);
                }
            }

            if (this.channel.isOpen()) {
                try {
                    this.channel.close();
                } catch (IOException e) {
                    logger.warn(e.getMessage(), e);
                }
            }
        }
    }
286231
protected static void checkChannels(final Iterable<String> channels) {
        if (channels == null) {
            throw new IllegalArgumentException("channels must not be null");
        }
        for (final String channel : channels) {
            if (channel == null || "".equals(channel)) {
                throw new IllegalArgumentException("channels' members must not be null: " + channels);
            }
        }
    }
214651
public void deleteChannel(FileIOChannel.ID channel) throws IOException {
		if (channel != null) {
			if (channel.getPathFile().exists() && !channel.getPathFile().delete()) {
				LOG.warn("IOManager failed to delete temporary file {}", channel.getPath());
			}
		}
	}
870729
public void free()
    {
        if (m_messageMap != null)
        {
            for (BaseMessageQueue messageQueue : m_messageMap.values())
            {
                if (messageQueue != null)
                {       // Don't worry about removing these, since you are removing them all.
                    messageQueue.setMessageManager(null);   // Make sure it doesn't try to remove itself again
                    messageQueue.free();
                }                
            }
        }
        m_messageMap = null;
        super.free();
    }
274916
public void removeMessageListener(Class<? extends Message> messageClass,
                                      MessageListener<Message> messageListener) {
        removeMessageListener(messageListener.getChannelIdentifier(), messageClass, messageListener);
    }
1046747
@Override
	public Promise subscribe(String channel) {
		if (client != null) {
			try {
				String postfix = '.' + nodeID;
				String queueName;
				if (channel.endsWith(postfix)) {

					// Create queue
					queueName = channel;
					this.channel.queueDeclareNoWait(queueName, true, exclusive, true, queueProperties);

				} else {

					// Create queue and exchange
					queueName = channel + '.' + nodeID;
					this.channel.queueDeclareNoWait(queueName, true, exclusive, true, queueProperties);
					this.channel.exchangeDeclareNoWait(channel, exchangeType, true, false, internal,
							exchangeProperties);
					this.channel.queueBind(queueName, channel, "");
				}
				this.channel.basicConsume(queueName, true, new Consumer() {

					// --- MESSAGE RECEIVED ---

					@Override
					public final void handleDelivery(String consumerTag, Envelope envelope,
							AMQP.BasicProperties properties, byte[] body) throws IOException {

						// We are running in the shared executor's pool,
						// do not create new task.
						processReceivedMessage(channel, body);
					}

					// --- CONNECTION LOST ---

					@Override
					public final void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
						synchronized (factory) {
							if (client != null) {
								logger.info("AMQP pub-sub connection aborted.");
								if (started.get()) {
									reconnect();
								}
							}
						}
					}

					// --- UNUSED METHODS ---

					@Override
					public final void handleConsumeOk(String consumerTag) {
						if (debugHeartbeats) {
							logger.info("ConsumeOk packet received (consumerTag: " + consumerTag + ").");
						}
					}

					@Override
					public final void handleCancelOk(String consumerTag) {
						if (debugHeartbeats) {
							logger.info("CancelOk packet received (consumerTag: " + consumerTag + ").");
						}
					}

					@Override
					public final void handleCancel(String consumerTag) throws IOException {
						if (debugHeartbeats) {
							logger.info("Cancel packet received (consumerTag: " + consumerTag + ").");
						}
					}

					@Override
					public final void handleRecoverOk(String consumerTag) {
						if (debugHeartbeats) {
							logger.info("RecoverOk packet received (consumerTag: " + consumerTag + ").");
						}
					}

				});

			} catch (Exception cause) {
				return Promise.reject(cause);
			}
		}
		return Promise.resolve();
	}
333290
protected void removeMessage(LMEMessage message)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "removeMessage", new Object[] {new Integer(hashCode()), message, this });

    // If this was the message we entered the callback with we need
    // to move the start point back a bit
    if(message == callbackEntryMsg)
      callbackEntryMsg = message.previous;

    // If this is our current message we need to move the cursor
    // back a bit
    if(message == currentMsg)
      currentMsg = message.previous;

    // If this message was the next to expire we move the expiry cursor on to
    // the next eligible message. We leave any alarm registered so that it pops,
    // when this happens it realises the alarm is not valid for the next expiring
    // message and re-registers itself for the correct time. This gives us the
    // best performance when no locks are expiring as we're not continually
    // registering and de-registering alarms for every message
    if(message == nextMsgToExpire)
    {
      do
      {
        nextMsgToExpire = nextMsgToExpire.next;
      }
      while((nextMsgToExpire != null) && (nextMsgToExpire.expiryTime == 0));
    }

    if(message == nextMsgReferenceToExpire)
    {
      do
      {
        nextMsgReferenceToExpire = nextMsgReferenceToExpire.next;
      }
      while((nextMsgReferenceToExpire != null) && (nextMsgReferenceToExpire.expiryTime == 0));
    }

    // Unlink this message from the list
    if(message.previous != null)
      message.previous.next = message.next;
    else
      firstMsg = message.next;

    if(message.next != null)
      message.next.previous = message.previous;
    else
      lastMsg = message.previous;

    // Pool the element if there's space
    if(pooledCount < poolSize)
    {
      message.message = null;
      message.next = pooledMsg;
      message.previous = null;
      message.owner = null;
      message.jsMessage = null;
      message.expiryTime = 0;
      message.expiryMsgReferenceTime = 0;
      pooledMsg = message;
      pooledCount++;
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "removeMessage", this);
  }
470859
private void fireOnDeleteQueue(ChannelEvent e) {
        List<EventListener> listeners = changes.getListenerList(AMQP);
        for (EventListener listener : listeners) {
            ChannelListener amqpListener = (ChannelListener)listener;
            amqpListener.onDeleteQueue(e);
        }
    }
942479
public void setChannels(java.util.Collection<Channel> channels) {
        if (channels == null) {
            this.channels = null;
            return;
        }

        this.channels = new java.util.ArrayList<Channel>(channels);
    }
1219118
private void stopChannel(Channel channel) throws ChannelException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "stopChannel");
        }
        String channelName = channel.getName();
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "channelName=" + channelName);
        }
        ChannelContainer channelContainer = channelRunningMap.get(channelName);
        try {
            channel.stop(0);
        } catch (ChannelException ce) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                Tr.debug(tc, "Channel " + channel + " threw ChannelException " + ce.getMessage());
            throw ce;
        } catch (Throwable e) {
            FFDCFilter.processException(e, getClass().getName() + ".stopChannel", "1338", this, new Object[] { channel });
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                Tr.debug(tc, "Channel " + channel + " threw non-ChannelException " + e.getMessage());
            throw new ChannelException(e);
        }
        // Update the state.
        channelContainer.setState(RuntimeState.INITIALIZED);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "stopChannel");
        }
    }
633854
public final void writeAndRemoveAll(ChannelHandlerContext ctx) {
        decrementReadableBytes(readableBytes);
        Throwable pending = null;
        ByteBuf previousBuf = null;
        for (;;) {
            Object entry = bufAndListenerPairs.poll();
            try {
                if (entry == null) {
                    if (previousBuf != null) {
                        ctx.write(previousBuf, ctx.voidPromise());
                    }
                    break;
                }

                if (entry instanceof ByteBuf) {
                    if (previousBuf != null) {
                        ctx.write(previousBuf, ctx.voidPromise());
                    }
                    previousBuf = (ByteBuf) entry;
                } else if (entry instanceof ChannelPromise) {
                    ctx.write(previousBuf, (ChannelPromise) entry);
                    previousBuf = null;
                } else {
                    ctx.write(previousBuf).addListener((ChannelFutureListener) entry);
                    previousBuf = null;
                }
            } catch (Throwable t) {
                if (pending == null) {
                    pending = t;
                } else {
                    logger.info("Throwable being suppressed because Throwable {} is already pending", pending, t);
                }
            }
        }
        if (pending != null) {
            throw new IllegalStateException(pending);
        }
    }
577489
@Handler(channels = EncryptedChannel.class)
    public void onPurge(Purge event, IOSubchannel encryptedChannel) {
        @SuppressWarnings("unchecked")
        final Optional<PlainChannel> plainChannel
            = (Optional<PlainChannel>) LinkedIOSubchannel
                .downstreamChannel(this, encryptedChannel);
        if (plainChannel.isPresent()) {
            plainChannel.get().purge();
        }
    }