Send a subscribe message for the given doctype trough the given websocket, but only if it is in a ready state. If not, retry a few milliseconds later.
11761
function subscribeWhenReady(doctype, socket) {
  if (socket.readyState === WEBSOCKET_STATE.OPEN) {
    try {
      socket.send(
        JSON.stringify({
          method: 'SUBSCRIBE',
          payload: {
            type: doctype
          }
        })
      )
    } catch (error) {
      // eslint-disable-next-line no-console
      console.warn(`Cannot subscribe to doctype ${doctype}: ${error.message}`)
      throw error
    }
  } else {
    setTimeout(() => {
      subscribeWhenReady(doctype, socket)
    }, 10)
  }
}
1123935
def start_stream(self, stream_type, **stream_parameters):
        """"""
        if stream_type:
            self.subscribe(stream_type, **stream_parameters)
        else:
            self.handle_error("You need to set a stream type")
14406
function _onReady() {
            if (!this.disposed && !this.ready) {
                this.ready = true;

                // Handshake was successful, Channel is ready for messages
                // Set the counter back to original value for dealing with iframe reloads
                this.handshakeAttempts = this.handshakeAttemptsOrig;

                // Process queued messages if any
                if (this.messageQueue && this.messageQueue.length) {
                    PostMessageUtilities.delay(function() {
                        var message;
                        var parsed;

                        if (!this.disposed && this.ready) {
                            while (this.messageQueue && this.messageQueue.length) {
                                message = this.messageQueue.shift();
                                try {
                                    parsed = _prepareMessage.call(this, message);
                                    this.receiver.postMessage(parsed);
                                }
                                catch(ex) {
                                    /* istanbul ignore next  */
                                    PostMessageUtilities.log("Error while trying to post the message from queue", "ERROR", "PostMessageChannel");
                                }
                            }

                            // Invoke the callback for ready
                            this.onready();
                        }
                    }.bind(this));
                }
                else {
                    // Invoke the callback for ready
                    this.onready();
                }
            }
        }
853024
def subscribe(self, handler, topic=None, options=None):
        
        def proxy_handler(*args, **kwargs):
            return self._callbacks_runner.put(partial(handler, *args, **kwargs))
        return self._async_session.subscribe(proxy_handler, topic=topic, options=options)
335137
def send(self, msg):
        
        with self._pub_lock:
            self.publish.send_string(msg)
        return self
260894
private boolean subscribe(String path, EventType watchType,
      long transactionId)
          throws TransactionIdTooOldException, InterruptedException {
    ServerHandler.Client server = connectionManager.getServer();
    NamespaceEvent event =  new NamespaceEvent(path, watchType.getByteValue());
    
    if (LOG.isDebugEnabled()) {
      LOG.debug(listeningPort + ": subscribe: Trying to subscribe for " +
          NotifierUtils.asString(event) + " ... from txId " + transactionId);
    }
    
    for (int retries = 0; retries < 3; retries ++) {
      try {
        server.subscribe(connectionManager.getId(), event, transactionId);
        if (LOG.isDebugEnabled()) {
          LOG.debug(listeningPort + ": subscribe: successful");
        }
        return true;
      } catch (TransactionIdTooOldException e) {
        LOG.warn(listeningPort + ": Failed to subscribe [1]", e);
        throw e;
      } catch (InvalidClientIdException e) {
        LOG.warn(listeningPort + ": Failed to subscribe [2]", e);
      } catch (TException e) {
        LOG.warn(listeningPort + ": Failed to subscribe [3]", e);
        Thread.sleep(1000);
      }
    }
    
    return false;
  }
1332508
function subscribe(name, id, callback, timeout, failswitch) {
        if (typeof name !== s_string) {
            throw new TypeError('topic name must be a string.');
        }
        if (id === undefined || id === null) {
            if (typeof callback === s_function) {
                id = callback;
            }
            else {
                throw new TypeError('callback must be a function.');
            }
        }
        if (!_topics[name]) {
            _topics[name] = Object.create(null);
        }
        if (typeof id === s_function) {
            timeout = callback;
            callback = id;
            autoId().then(function(id) {
                subscribe(name, id, callback, timeout, failswitch);
            });
            return;
        }
        if (typeof callback !== s_function) {
            throw new TypeError('callback must be a function.');
        }
        if (Future.isPromise(id)) {
            id.then(function(id) {
                subscribe(name, id, callback, timeout, failswitch);
            });
            return;
        }
        // Default subscribe timeout is 5 minutes.
        if (timeout === undefined) { timeout = 300000; }
        var topic = getTopic(name, id);
        if (topic === null) {
            var cb = function() {
                _invoke(self, name, [id, topic.handler, cb, {
                    idempotent: true,
                    failswitch: failswitch,
                    timeout: timeout
                }], false);
            };
            topic = {
                handler: function(result) {
                    var topic = getTopic(name, id);
                    if (topic) {
                        if (result !== null) {
                            var callbacks = topic.callbacks;
                            for (var i = 0, n = callbacks.length; i < n; ++i) {
                                try {
                                    callbacks[i](result);
                                }
                                catch (e) {}
                            }
                        }
                        if (getTopic(name, id) !== null) { cb(); }
                    }
                },
                callbacks: [callback]
            };
            _topics[name][id] = topic;
            cb();
        }
        else if (topic.callbacks.indexOf(callback) < 0) {
            topic.callbacks.push(callback);
        }
    }
671196
async def send(bot: NoneBot, ctx: Context_T,
               message: Message_T, *,
               ensure_private: bool = False,
               ignore_failure: bool = True,
               **kwargs) -> Any:
    """"""
    try:
        if ensure_private:
            ctx = ctx.copy()
            ctx['message_type'] = 'private'
        return await bot.send(ctx, message, **kwargs)
    except CQHttpError:
        if not ignore_failure:
            raise
        return None
44673
def subscribe(self,
                  event_type: Type[TSubscribeEvent],
                  handler: Callable[[TSubscribeEvent], None]) -> Subscription:
        
        if event_type not in self._handler:
            self._handler[event_type] = []

        casted_handler = cast(Callable[[BaseEvent], Any], handler)

        self._handler[event_type].append(casted_handler)

        return Subscription(lambda: self._handler[event_type].remove(casted_handler))
1322844
def subscribe(self, topic, qos):
        """"""
        if self.sock == NC.INVALID_SOCKET:
            return NC.ERR_NO_CONN
        
        self.logger.info("SUBSCRIBE: %s", topic)
        return self.send_subscribe(False, [(utf8encode(topic), qos)])
457645
function subscribeBrowser (client, subscriptions, backlog, channel, callback, done_callback) {
    if ( addToBacklog(client, backlog, subscribeBrowser, [client, subscriptions, backlog, channel, callback, done_callback]) ) return;
    if (subscriptions[channel]===undefined) {
        subscriptions[channel] = [callback];
        client.subscribe(channel, {qos: 0, onSuccess: function() {
            // If there is a done_callback defined, execute it
            if (done_callback!==undefined) done_callback();
        }});
    } else {
        subscriptions[channel].push(callback);
        // If there is a done_callback defined, execute it
        if (done_callback!==undefined) done_callback();
    }
}
214389
def subscribe(self, event, handler):
        
        self._handlers.sync_handlers[event].append(handler)
67373
def _execute(self, sender, event_args):
        '''
        
        '''
        with self._lock:
            while not self._messages.empty():
                msg, args, kwargs = self._messages.get(False)
                for subscriber in self._subscribers[msg]:
                    try:
                        subscriber(*args, **kwargs)
                    except weakref.ReferenceError:
                        # Reference to handler is lost and it is OK to silence it
                        pass
1238803
@Override
	public Promise subscribe(String channel) {
		if (dispatcher != null) {
			try {
				dispatcher.subscribe(channel);
			} catch (Exception cause) {
				return Promise.reject(cause);
			}
		}
		return Promise.resolve();
	}
507815
async def receiver_handler(websocket):
    runner.addifnew(websocket)

    # The commented only works with python3-websockets 4.x
    '''
    
    '''

    while True:
        message = await websocket.recv()
        await consumer(websocket, message)
333496
function sendClientConsoleMessage(ws, message) {
  var msg = JSON.stringify({
    custom: {
      console: message
    }
  });
  ws.write(msg);
}
226576
def on_message(self, message):
        
        log.debug('Got websocket message %s', message)
        if message == 'websocket client ready':
            if self.connected:
                # TBD: Add ability to run commands in this branch
                log.debug('Websocket already connected, returning')
                return

            self.connected = True

            while True:
                try:
                    event = yield self.application.event_listener.get_event(self)
                    self.write_message(
                        salt.utils.json.dumps(event, _json_module=_json))
                except Exception as err:
                    log.info('Error! Ending server side websocket connection. Reason = %s', err)
                    break

            self.close()
        else:
            # TBD: Add logic to run salt commands here
            pass