* This method binds a queue to an exchange. Until a queue is bound it will not receive any messages. In a classic messaging model, store-and-forward
queues are bound to a direct exchange and subscription queues are bound to a topic exchange.
197409
public AmqpChannel bindQueue(String queue, String exchange, String routingKey, boolean noWait, AmqpArguments arguments) {
        Object[] args = {0, queue, exchange, routingKey, noWait, arguments};
        WrappedByteBuffer bodyArg = null;
        HashMap<String, Object> headersArg = null;

        String methodName = "bindQueue";
        String methodId = "50" + "20";
        AmqpMethod amqpMethod = MethodLookup.LookupMethod(methodId);
        Object[] methodArguments = {this, amqpMethod, this.id, args, bodyArg, headersArg};

        boolean hasnowait = false;
        for (int i = 0; i < amqpMethod.allParameters.size(); i++) {
            String argname = amqpMethod.allParameters.get(i).name;
            if (argname == "noWait") {
                hasnowait = true;
                break;
            }
        }

        if (client.getReadyState() == ReadyState.OPEN) {
            asyncClient.enqueueAction(methodName, "channelWrite", methodArguments, null, null);
        }

        if (hasnowait && noWait) {
            asyncClient.enqueueAction("nowait", null, null, null, null);
        }
        return this;
    }
744033
def bind_queue(queue, routing_keys)
      unbind_redundant_bindings(queue, routing_keys)

      # Ensure all the desired bindings are present
      routing_keys.each do |routing_key|
        logger.debug "creating binding #{queue.name} <--> #{routing_key}"
        queue.bind(exchange, routing_key: routing_key)
      end
    end
358198
def unbind(self, arguments=None):
        
        if self.deleted:
            raise Deleted("Queue {} was already unbound from exchange {}".format(self.queue.name, self.exchange.name))

        self.sender.send_QueueUnbind(self.queue.name, self.exchange.name, self.routing_key, arguments or {})
        yield from self.synchroniser.wait(spec.QueueUnbindOK)
        self.deleted = True
        self.reader.ready()
884854
@Override
	public Promise subscribe(String channel) {
		try {

			// Create publisher
			createOrGetPublisher(channel);

			// Create subscriber
			synchronized (subscribers) {
				if (!subscribers.containsKey(channel)) {
					TopicSubscriber subscriber = session.createSubscriber(session.createTopic(channel));
					subscribers.put(channel, subscriber);
					subscriber.setMessageListener((message) -> {
						try {
							BytesMessage msg = (BytesMessage) message;
							byte[] bytes = new byte[(int) msg.getBodyLength()];
							msg.readBytes(bytes);
							received(channel, bytes);
						} catch (Exception cause) {
							logger.error("Unable to deserialize message!", cause);
						}
					});
				}
			}
		} catch (Exception cause) {
			return Promise.reject(cause);
		}
		return Promise.resolve();
	}
763123
public QueueConsumer createQueueTextMessageListener(final String topic, final ConsumerCallback<String> messageCallback)
    {
        Preconditions.checkState(connectionFactory != null, "connection factory was never injected!");
        return new QueueConsumer(connectionFactory, jmsConfig, topic, new TextMessageConsumerCallback(messageCallback));
    }
53469
def _bind(self):
        

        credentials = pika.PlainCredentials(self.user, self.password)
        params = pika.ConnectionParameters(credentials=credentials,
                                           host=self.server,
                                           virtual_host=self.vhost,
                                           port=self.port)

        self.connection = pika.BlockingConnection(params)
        self.channel = self.connection.channel()

        # NOTE : PIKA version uses 'exchange_type' instead of 'type'

        self.channel.exchange_declare(exchange=self.topic_exchange,
                                      exchange_type="topic")
306168
def bind(exchange, opts = {})
      if exchange.respond_to?(:add_route)

        # we can do the binding ourselves
        exchange.add_route opts.fetch(:routing_key, @name), self
      else

        # we need the channel to look up the exchange
        @channel.xchg_bind self, opts.fetch(:routing_key, @name), exchange
      end

      self
    end
1140218
private void recoverRelatedExchanges(Set<String> recoveredExchanges, List<Binding> queueBindings)
    throws Exception {
    if (config.isExchangeRecoveryEnabled() && queueBindings != null)
      synchronized (queueBindings) {
        for (Binding queueBinding : queueBindings) {
          String exchangeName = queueBinding.source;
          if (recoveredExchanges.add(exchangeName)) {
            ResourceDeclaration exchangeDeclaration = connectionHandler.exchangeDeclarations.get(exchangeName);
            if (exchangeDeclaration != null)
              recoverExchange(exchangeName, exchangeDeclaration);
            recoverExchangeBindings(connectionHandler.exchangeBindings.get(exchangeName));
          }
        }
      }
  }
359427
def setup_queue(consumer)
      logger.info "setting up queue: #{consumer.get_queue_name}"

      queue = @broker.queue(consumer.get_queue_name, consumer.get_arguments)
      @broker.bind_queue(queue, consumer.routing_keys)

      queue.subscribe(consumer_tag: unique_consumer_tag, manual_ack: true) do |*args|
        delivery_info, properties, payload = Hutch::Adapter.decode_message(*args)
        handle_message(consumer, delivery_info, properties, payload)
      end
    end
1348839
@Override
	public void send(Queue queue, Message message) throws JMSException
    {
        send((Destination)queue,message,defaultDeliveryMode,defaultPriority,defaultTimeToLive);
    }
223994
public AmqpChannel unbindQueue(String queue, String exchange, String routingKey, AmqpArguments arguments) {
        Object[] args = {0, queue, exchange, routingKey, arguments};
        WrappedByteBuffer bodyArg = null;
        HashMap<String, Object> headersArg = null;

        String methodName = "unbindQueue";
        String methodId = "50" + "50";
        AmqpMethod amqpMethod = MethodLookup.LookupMethod(methodId);
        Object[] methodArguments = {this, amqpMethod, this.id, args, bodyArg, headersArg};
        asyncClient.enqueueAction(methodName, "channelWrite", methodArguments, null, null);
        return this;
    }
495501
def on_input_queue_declare(self, queue):
        
        self.in_channel.queue_bind(callback=None,
                                   exchange='input_exc',
                                   queue=self.INPUT_QUEUE_NAME,
                                   routing_key="#")
61587
def send(self, message):
        
        dest = message.headers.get('destination')
        if not dest:
            raise ValueError(
                "Cannot send frame with no destination: %s" % message)

        message.cmd = 'message'

        message.headers.setdefault('message-id', str(uuid.uuid4()))

        bad_subscribers = set()
        for subscriber in self._topics[dest]:
            try:
                subscriber.send_frame(message)
            except:
                self.log.exception(
                    "Error delivering message to subscriber %s; client will be disconnected." % subscriber)
                # We queue for deletion so we are not modifying the topics dict
                # while iterating over it.
                bad_subscribers.add(subscriber)

        for subscriber in bad_subscribers:
            self.disconnect(subscriber)
418599
@Bean
    public Queue dmfReceiverQueue() {
        return new Queue(amqpProperties.getReceiverQueue(), true, false, false,
                amqpDeadletterProperties.getDeadLetterExchangeArgs(amqpProperties.getDeadLetterExchange()));
    }
684895
public boolean commit(KafkaMessage msg) {
        KafkaConsumer<String, byte[]> consumer = _getConsumer(msg.topic());
        synchronized (consumer) {
            Set<String> subscription = consumer.subscription();
            if (subscription == null || !subscription.contains(msg.topic())) {
                // this consumer has not subscribed to the topic
                return false;
            } else {
                try {
                    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                    offsets.put(new TopicPartition(msg.topic(), msg.partition()),
                            new OffsetAndMetadata(msg.offset() + 1));
                    consumer.commitSync(offsets);
                } catch (WakeupException e) {
                } catch (Exception e) {
                    throw new KafkaException(e);
                }
                return true;
            }
        }
    }
879730
public function queueUnbind($queue = '', $exchange, $routingKey = '', $arguments = [])
    {
        return $this->getClient()->queueUnbind($this->getChannelId(), $queue, $exchange, $routingKey, $arguments);
    }
261545
def send(self, message):
        
        self.logger.debug('Send message: {}'.format(message))

        if self.state == 'pending' or self.state == 'connection_pending':
            self.logger.info('Connection not established. Add message to queue.')
            self.message_queue.append(message)
            return
        elif self.state == 'unsubscribed' or self.state == 'rejected':
            self.logger.warning('Not subscribed! Message discarded.')
            return

        data = {
            'command': 'message',
            'identifier': self._identifier_string(),
            'data': message.raw_message()
        }

        self.connection.send(data)