Enables a service to subscribe to a channel
 @param channel [String]
 @param service [Service]
972630
def register_channel(channel, service)
      _channel = channel.downcase
      unless @channels[_channel]
        @channels[_channel] = service
        GameOverseer::Console.log("ChannelManager> mapped '#{_channel}' to '#{service.class}'.")
      else
        raise "Could not map channel '#{_channel}' because '#{@channels[data[_channel]].class}' is already using it."
      end
    end
146734
def _handle_subscribed(self, *args,  chanId=None, channel=None, **kwargs):
        
        log.debug("_handle_subscribed: %s - %s - %s", chanId, channel, kwargs)
        if chanId in self.channels:
            raise AlreadyRegisteredError()

        self._heartbeats[chanId] = time.time()

        try:
            channel_key = ('raw_'+channel
                           if kwargs['prec'].startswith('R') and channel == 'book'
                           else channel)
        except KeyError:
            channel_key = channel

        try:
            self.channels[chanId] = self._data_handlers[channel_key]
        except KeyError:
            raise UnknownChannelError()

        # prep kwargs to be used as secondary value in dict key
        try:
            kwargs.pop('event')
        except KeyError:
            pass

        try:
            kwargs.pop('len')
        except KeyError:
            pass

        try:
            kwargs.pop('chanId')
        except KeyError:
            pass

        self.channel_labels[chanId] = (channel_key, kwargs)
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();
	}
934797
def startSubscription(self, reqId, subscriber, contract=None):
        
        self._reqId2Contract[reqId] = contract
        self.reqId2Subscriber[reqId] = subscriber
299627
def add_channel(channel: EFBChannel):
    
    global master, slaves
    if isinstance(channel, EFBChannel):
        if channel.channel_type == ChannelType.Slave:
            slaves[channel.channel_id] = channel
        else:
            master = channel
    else:
        raise TypeError("Channel instance is expected")
1274401
def on_open(self, websocket):
        ''''''
        self.pubsub.add_client(ChatClient(websocket, self.channel))
532471
def create_channel(self, channel):
        

        data = {'channels': channel}
        return super(ApiInterfaceRequest, self).post('api/v3/channel/', data)
333792
function(channel, subscription, context, once) {
      if (!channels[channel]) channels[channel] = [];
      channels[channel].push({fn: subscription, context: context || this, once: once});
    }
451496
public function subscribe($channel /*, ... */)
    {
        $this->writeRequest(self::SUBSCRIBE, func_get_args());
        $this->statusFlags |= self::STATUS_SUBSCRIBED;
    }
552757
def _bind_channels(self, events, channels):
        
        for channel_name in channels:
            if channel_name in self.channels:
                channel = self.pusher.subscribe(channel_name)
                if isinstance(events, list):
                    for event in events:
                        channel.bind(event, channels[channel_name])
                else:
                    channel.bind(events, channels[channel_name])
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();
	}
1253420
def bind_to_channel(self):
        
        if self.channel.code_name != self.user.prv_exchange:
            channel = self._connect_mq()
            channel.exchange_bind(source=self.channel.code_name, destination=self.user.prv_exchange)
884930
def auto_subscribe(self, app_manager):
        """"""
        for handler in app_manager.handler_classes:
            app_manager.subscribe(handler.channel(), self)
1067800
def subscribe(self, topic, channel):
        """"""
        self.send(nsq.subscribe(topic, channel))
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"])
951937
function (event) {
    this.debug("Requesting subscription for " + event);
    sendReceive.call(this, {'do': 'subscribe', params: [event]}, function (result) {
        if (result.success) {
            this.debug("Succesfully subscribed to %s", event);
        } else {
            this.debug("Subscription request for %s failed", event);
        }
    });
    this.subscribedEvents.push(event);
}
880843
def subscribe_parameter(param, &callback)
      sub = ParameterSubscriber.new(param, callback)
      @manager.add_parameter_subscriber(sub)
      sub
    end