Bind a consumer's routing keys to its queue, and set up a subscription to
 receive messages sent to the queue.
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
160028
function () {
			// subscribe to routing
			this.router = UIComponent.getRouterFor(this);
			this.router.attachRoutePatternMatched(this.onRouteMatched, this);

			// subscribe to app events
			this._component = Component.getOwnerComponentFor(this.getView());
			this._component.getEventBus().subscribe("app", "selectEntity", this.onSelectEntity, this);

			// set the group to the default used in the view
			this._sCurrentGroup = this._oDefaultSettings.groupProperty;

			// subscribe to nav container events
			this.getView().addEventDelegate({
				onBeforeFirstShow: jQuery.proxy(this.onBeforeFirstShow, this)
			});
		}
364238
def send(self, message, *args, **kwargs):
        
        routing_keys = kwargs.get('routing_key') or self.routing_key
        routing_keys = [routing_keys] if isinstance(routing_keys, basestring) else routing_keys
        correlation_id = kwargs.get('correlation_id', None)
        reply_to = kwargs.get('reply_to', None)
        declare=[self.exchange] + kwargs.get('declare', [])
        conn = self.get_connection()
        with connections[conn].acquire(block=True) as connection:
            self.exchange.maybe_bind(connection)
            #reply_to.maybe_bind(connection)
            #reply_to.declare(True)
            with producers[connection].acquire(block=True) as producer:
                for routing_key in routing_keys:
                    LOGGER.info('Send message %s to exchange %s with routing_key %s reply_to %s correlation_id %s',
                                message, self.exchange.name, routing_key, reply_to, correlation_id)
                    producer.publish(
                        message,
                        exchange=self.exchange,
                        declare=declare,
                        serializer=self.settings['serializer'],
                        routing_key=routing_key,
                        correlation_id=correlation_id,
                        retry=self.settings['retry'],
                        delivery_mode=self.settings['delivery_mode'],
                        reply_to=reply_to,
                        retry_policy=self.settings['retry_policy'])
94128
@Override
	public Promise subscribe(String channel) {
		Promise promise = new Promise();
		if (client != null) {
			try {
				client.subscribe(new Subscription[] { new Subscription(channel, qos) });
				subscriptions.put(channel, promise);
			} catch (Exception cause) {
				promise.complete(cause);
			}
		} else {
			promise.complete(new Throwable("Not connected!"));
		}
		return promise;
	}
1068012
def subscribe(self, path, callback, *args):
        
        request = Subscribe(self._get_next_id(), path, delta=False)
        request.set_callback(self._q.put)
        # If self is in args, then make weak version of it
        saved_args = []
        for arg in args:
            if arg is self:
                saved_args.append(weakref.proxy(self))
            else:
                saved_args.append(arg)
        self._subscriptions[request.id] = (callback, saved_args)
        future = self._dispatch_request(request)
        return future
1027755
def subscribe(self, subscription_channel, **observer_params):
        
        return self.get_channel(subscription_channel, **observer_params).ensure_started().observer
119007
private void send(T value)
  {
    for (SubscriberNode sub : _subscribers) {
      sub.next(value);
    }
    
    long seq = _sequence++;
    int size = _consumers.size();
    
    if (size > 0) {
      int index = (int) (seq % size);
      
      _consumers.get(index).next(value);
    }
  }
1078282
def handle_subscribe(self, request):
        # type: (Subscribe) -> CallbackResponses
        """"""
        ret = self._tree.handle_subscribe(request, request.path[1:])
        self._subscription_keys[request.generate_key()] = request
        return ret
836269
def subscribe(self, socket_id, channel):
        """"""
        con = self._get_connection(socket_id)
        self.subscriptions.setdefault(channel, set()).add(socket_id)
        con.subscriptions.add(channel)
1162599
def start
      queue = ::Queue.new

      @thread = Thread.start do
        @listener_redis.subscribe(@channel, @my_channel) do |on|
          on.subscribe do |channel|
            queue.push(:subscribed) if channel == @channel
          end

          on.message do |channel, message|
            handle_message(channel, message)
          end
        end
      end

      queue.pop
    end
1149342
def get_channel(self, subscription_channel, **observer_params):
        """"""
        keys = subscription_channel.get_routing_keys()
        # watch keys are unique sets of keys that we will attempt to extract from inbound items
        self.watch_keys.add(frozenset({k_v[0] for key in keys for k_v in key}))
        self.create_route(subscription_channel, keys)
        subscription_channel._configure(self, self.connect_api, observer_params)
        return subscription_channel
474475
public void subscribe(Object identifier) {
        final long id = Thread.currentThread().getId();
        queues.put(id, new ConcurrentLinkedQueue<T>());
        if (identifier != null) {
            if (subscriberIdentifiers == null) {
                subscriberIdentifiers = new ConcurrentHashMap<>();
            }
            subscriberIdentifiers.put(identifier, id);
        }
    }
123621
public static synchronized void subscribe(String key, RpcConfigListener listener) {
        List<RpcConfigListener> listeners = CFG_LISTENER.get(key);
        if (listeners == null) {
            listeners = new ArrayList<RpcConfigListener>();
            CFG_LISTENER.put(key, listeners);
        }
        listeners.add(listener);
    }
419568
def subscribe(self, topic, callback, qos):
        """"""
        if topic in self.topics:
            return

        def _message_callback(mqttc, userdata, msg):
            """Callback added to callback list for received message."""
            callback(msg.topic, msg.payload.decode('utf-8'), msg.qos)

        self._mqttc.subscribe(topic, qos)
        self._mqttc.message_callback_add(topic, _message_callback)
        self.topics[topic] = callback
1196257
def mqtt_subscribe(self, topics, packet_id):
        

        # Build and send SUBSCRIBE message
        subscribe = SubscribePacket.build(topics, packet_id)
        yield from self._send_packet(subscribe)

        # Wait for SUBACK is received
        waiter = futures.Future(loop=self._loop)
        self._subscriptions_waiter[subscribe.variable_header.packet_id] = waiter
        return_codes = yield from waiter

        del self._subscriptions_waiter[subscribe.variable_header.packet_id]
        return return_codes
793954
def subscribe(
            self,
            subject="",
            queue="",
            cb=None,
            future=None,
            max_msgs=0,
            is_async=False,
            pending_msgs_limit=DEFAULT_SUB_PENDING_MSGS_LIMIT,
            pending_bytes_limit=DEFAULT_SUB_PENDING_BYTES_LIMIT,
    ):
        
        if self.is_closed:
            raise ErrConnectionClosed

        if self.is_draining:
            raise ErrConnectionDraining

        self._ssid += 1
        sid = self._ssid
        sub = Subscription(
            subject=subject,
            queue=queue,
            cb=cb,
            future=future,
            max_msgs=max_msgs,
            is_async=is_async,
            sid=sid,
        )
        self._subs[sid] = sub

        if cb is not None:
            sub.pending_msgs_limit = pending_msgs_limit
            sub.pending_bytes_limit = pending_bytes_limit
            sub.pending_queue = tornado.queues.Queue(
                maxsize=pending_msgs_limit)

            @tornado.gen.coroutine
            def wait_for_msgs():
                while True:
                    sub = wait_for_msgs.sub
                    err_cb = wait_for_msgs.err_cb

                    try:
                        sub = wait_for_msgs.sub
                        if sub.closed:
                            break

                        msg = yield sub.pending_queue.get()
                        if msg is None:
                            break

                        sub.received += 1
                        sub.pending_size -= len(msg.data)
                        if sub.max_msgs > 0 and sub.received >= sub.max_msgs:
                            # If we have hit the max for delivered msgs, remove sub.
                            self._subs.pop(sub.sid, None)
                            self._remove_subscription(sub)

                        # Invoke depending of type of handler.
                        if sub.is_async:
                            # NOTE: Deprecate this usage in a next release,
                            # the handler implementation ought to decide
                            # the concurrency level at which the messages
                            # should be processed.
                            self._loop.spawn_callback(sub.cb, msg)
                        else:
                            yield sub.cb(msg)
                    except Exception as e:
                        # All errors from calling an async subscriber
                        # handler are async errors.
                        if err_cb is not None:
                            yield err_cb(e)

            # Bind the subscription and error cb if present
            wait_for_msgs.sub = sub
            wait_for_msgs.err_cb = self._error_cb
            self._loop.spawn_callback(wait_for_msgs)

        elif future is not None:
            # Used to handle the single response from a request
            # based on auto unsubscribe.
            sub.future = future

        # Send SUB command...
        sub_cmd = b''.join([
            SUB_OP, _SPC_,
            sub.subject.encode(), _SPC_,
            sub.queue.encode(), _SPC_, ("%d" % sid).encode(), _CRLF_
        ])
        yield self.send_command(sub_cmd)
        yield self._flush_pending()
        raise tornado.gen.Return(sid)
360179
def accept_subscription(&blk)
      pres = connection.presence_stanza('to'=>jid.bare, 'type' => 'subscribed')
      connection.send_stanza pres, &blk
    end