Creates an event queue (thread safe) with the correct binding.

@param channel Name of event handled by queue.
507954
private synchronized void createChannel(String channel) {
        if (!channelExists(channel)) {
            Queue queue = new Queue(channel, true, false, true);
            admin.declareQueue(queue);
            Binding binding = new Binding(channel, DestinationType.QUEUE, exchange.getName(), channel + ".#", null);
            admin.declareBinding(binding);
        }
    }
59692
protected void setupQueue() throws IOException {
		if (queueName != null) {
			channel.queueDeclare(queueName, false, false, false, null);
		}
	}
1346677
public void init(SocketChannel channel)
  {
    _channel = channel;
    _readBuffer.clear();
    _readBuffer.flip();
    
    _needsFlush = false;
  }
787523
public T open(RequestChannel channel)
    {
        return clientManager.createClient(channel, clientType, clientName, eventHandlers);
    }
94556
@Override
	public QueueReceiver createReceiver(Queue queue, String messageSelector) throws JMSException
    {
    	externalAccessLock.readLock().lock();
    	try
		{
	        checkNotClosed();
	        LocalQueueReceiver receiver = new LocalQueueReceiver(engine,this,queue,messageSelector,idProvider.createID());
	        registerConsumer(receiver);
	        receiver.initDestination();
	        return receiver;
		}
    	finally
    	{
    		externalAccessLock.readLock().unlock();
    	}
    }
932681
private void writeEvent(final ChannelHandlerContext ctx, final Object message) {
        if (isFinishedSent)
            return;
        
        if (message instanceof FinishedMessage)
            isFinishedSent = true;

        Channels.write(ctx, Channels.future(null), message);
    }
1224623
public void queue(final AsyncEvent<K> _event) {
    final K key = _event.getKey();
    synchronized (getLockObject(key)) {
      Queue<AsyncEvent<K>> q = keyQueue.get(key);
      if (q != null) {
        q.add(_event);
        return;
      }
      q = new LinkedList<AsyncEvent<K>>();
      keyQueue.put(key, q);
    }
    Runnable r = new Runnable() {
      @Override
      public void run() {
        runMoreOrStop(_event);
      }
    };
    executor.execute(r);
  }
703879
private void fireOnPurgeQueue(ChannelEvent e) {
        List<EventListener> listeners = changes.getListenerList(AMQP);
        for (EventListener listener : listeners) {
            ChannelListener amqpListener = (ChannelListener)listener;
            amqpListener.onPurgeQueue(e);
        }
    }
151249
@Override
    @Trivial
    public void start() throws ChannelException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
            Tr.event(tc, "Start channel: " + this);
        }
        stop0Called = false;
    }
341018
@Process(actionType = CreateConfig.class)
    public void createConfig(final CreateConfig action, final Dispatcher.Channel channel) {
        create(action.getAddressTemplate(), action.getInstanceName(), action.getNewModel(),
                getModelsFor(action.getAddressTemplate()), channel);
    }
1212435
protected void enqueueEvent(Object event, EventHandler handler) {
    eventsToDispatch.get().offer(new EventWithHandler(event, handler));
  }
944645
def _on_channel_open(self, channel):
        
        channel.add_on_close_callback(self._on_channel_close)
        channel.add_on_cancel_callback(self._on_cancel)

        channel.basic_qos(callback=self._on_qosok, **config.conf["qos"])
1126982
public final Promise subscribe(String channel) {
		return new Promise(client.subscribe(channel.getBytes(StandardCharsets.UTF_8)));
	}
628299
@Override
    public void handleEvent(Event event) {
        String topic = event.getTopic();
        if (topic.equalsIgnoreCase(ChannelFramework.EVENT_STOPCHAIN.getName())) {
            String chainName = event.getProperty(ChannelFramework.EVENT_CHAINNAME, String.class);
            if (null != chainName) {
                stopChain(chainName, event);
            } else {
                String channelName = event.getProperty(ChannelFramework.EVENT_CHANNELNAME, String.class);
                if (null != channelName) {
                    stopChannel(channelName, event);
                }
            }
        } else if (topic.equalsIgnoreCase(EVENT_CHECK_MISSING.getName())) {
            ChannelUtils.checkMissingConfig();
        }
    }
1089755
@Process(actionType = AddBufferPool.class)
    public void add(final AddBufferPool action, final Dispatcher.Channel channel) {
        String name = action.getBufferPool().get(NAME).asString();
        operationDelegate.onCreateResource(RESOURCE_ADDRESS, name, action.getBufferPool(), new RefreshCallback(channel));
    }
1081120
protected Channel createChannel() throws IOException {
        LOGGER.debug("Creating channel");
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        LOGGER.debug("Created channel");
        return channel;
    }
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();
	}