async for message in websocket:
        await consumer(websocket, message)
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)
434824
def websocket_handler(request):
    """"""

    logger.debug('New websocket connection.')
    websocket = web.WebSocketResponse()
    yield from websocket.prepare(request)
    uuid = uuid4()
    request.app['dispatcher'].subscribe(uuid, websocket)

    while True:
        # Consume input buffer
        try:
            msg = yield from websocket.receive()
        except RuntimeError as e:
            logger.debug('Websocket exception: %s', str(e))
            break
        except CancelledError:
            logger.debug('Websocket closed')
            break
        if msg.type == aiohttp.WSMsgType.CLOSED:
            logger.debug('Websocket closed')
            break
        elif msg.type == aiohttp.WSMsgType.ERROR:
            logger.debug('Websocket exception: %s', websocket.exception())
            break

    request.app['dispatcher'].unsubscribe(uuid)
    return websocket
88996
def _process_msg(self, sid, subject, reply, data):
        
        payload_size = len(data)
        self.stats['in_msgs'] += 1
        self.stats['in_bytes'] += payload_size

        msg = Msg(subject=subject.decode(), reply=reply.decode(), data=data)

        # Don't process the message if the subscription has been removed
        sub = self._subs.get(sid)
        if sub is None:
            raise tornado.gen.Return()

        # Check if it is an old style request.
        if sub.future is not None:
            sub.future.set_result(msg)

            # Discard subscription since done
            self._subs.pop(sid, None)
            self._remove_subscription(sub)
            raise tornado.gen.Return()

        # Let subscription wait_for_msgs coroutine process the messages,
        # but in case sending to the subscription task would block,
        # then consider it to be an slow consumer and drop the message.
        try:
            sub.pending_size += payload_size
            if sub.pending_size >= sub.pending_bytes_limit:
                # Substract again the bytes since throwing away
                # the message so would not be pending data.
                sub.pending_size -= payload_size

                if self._error_cb is not None:
                    yield self._error_cb(ErrSlowConsumer())
                raise tornado.gen.Return()
            sub.pending_queue.put_nowait(msg)
        except tornado.queues.QueueFull:
            if self._error_cb is not None:
                yield self._error_cb(ErrSlowConsumer())
1183712
def async_session_handler(self, signal: str) -> None:
        
        if signal == 'data':
            self.async_event_handler(self.websocket.data)
        elif signal == 'state':
            if self.async_connection_status_callback:
                self.async_connection_status_callback(
                    self.websocket.state == 'running')
1099872
async def incoming(self):
        
        msg = await self._queue.get()
        self._queue.task_done()
        return msg
869826
async def rpc_message(self, request, message):
        ''''''
        await self.pubsub.publish(self.channel, message)
        return 'OK'
285475
@Override
  public void onMessage(WebSocket webSocket, String message) {
    JsonObject json = new JsonParser().parse(message).getAsJsonObject();
    if (json.has(ERROR)) {
      String error = json.get(ERROR).getAsString();
      callback.onError(new RuntimeException(error));

    } else if (json.has(WARNINGS)) {
      String warning = json.get(WARNINGS).getAsString();
      callback.onWarning(new RuntimeException(warning));
    } else if (json.has(BINARY_STREAMS)) {
      String contentType = json.get(BINARY_STREAMS)
          .getAsJsonArray()
          .get(0)
          .getAsJsonObject()
          .get(CONTENT_TYPE)
          .getAsString();
      callback.onContentType(contentType);
    } else if (json.has(WORDS)) {
      callback.onTimings(GSON.fromJson(message, Timings.class));
    } else if (json.has(MARKS)) {
      callback.onMarks(GSON.fromJson(message, Marks.class));
    }
  }
310042
async def handler(self, request):
        

        def task_done(future):
            self.tasks.remove(future)
            exception = future.exception()
            if exception:
                log.warning(
                    'While processing message: {0}\nDetails: {1}'.format(
                        exception,
                        traceback.format_exc())
                )

        client = web.WebSocketResponse()
        client_id = id(client)

        # upgrade to Websockets
        await client.prepare(request)

        log.info('Opening Websocket {0}'.format(id(client)))
        log.debug('Tasks: {0}'.format(self.tasks))
        log.debug('Clients: {0}'.format(self.clients))

        try:
            log.debug('Sending root info to {0}'.format(client_id))
            await client.send_json({
                '$': {'type': CONTROL_MESSAGE, 'monitor': True},
                'root': self.call_and_serialize(lambda: self.root),
                'type': self.call_and_serialize(lambda: type(self.root))
            })
            log.debug('Root info sent to {0}'.format(client_id))
        except Exception:
            log.exception('While sending root info to {0}'.format(client_id))

        try:
            self.clients[client] = self.send_worker(client)
            # Async receive client data until websocket is closed
            async for msg in client:
                task = self.loop.create_task(self.process(msg))
                task.add_done_callback(task_done)
                self.tasks += [task]
        except Exception:
            log.exception('While reading from socket:')
        finally:
            log.info('Closing WebSocket {0}'.format(id(client)))
            await client.close()
            del self.clients[client]

        return client
693661
function (ws, message) {
  let clientMessage;

  try {
    clientMessage = JSON.parse(message);
    /* jshint ignore:start */
    if (_.get(clientMessage, 'type') === 'message') {
      ws.send(JSON.stringify({
        ok: true,
        reply_to: _.get(clientMessage, 'id'),
        ts: Date.now(),
        message: _.get(clientMessage, 'text', message),
        type: _.get(clientMessage, 'type', 'message'),
        user: _.get(clientMessage, 'user', 'U1234567'),
        channel: _.get(clientMessage, 'channel', 'C1234567'),
      }));
    }
    /* jshint ignore:end */
  } catch (err) {
    logger.info('Invalid socket data ', message);
  }
}
446111
function worker(message) {
    console.log(`Processing "${message.message.data}"...`);

    setTimeout(() => {
      console.log(`Finished procesing "${message.message.data}".`);
      isProcessed = true;
    }, 30000);
  }
949121
def send_persisted_messages(self, websocket):
        
        for channel in self._subscription.channels:
            message = self._connection.get(channel)
            if message:
                websocket.send(message)
1360068
async function(serviceProvider, value, emit) {
		// Each message is an object with key corresponding to 'emits' property in metadata
		emit([{msg: 'Hello world!'}]);
		await timeout(230);
		// Component can emit multiple messages at once
		emit([{msg: '¡Hola mundo!'}, {msg: 'Ahoj světe!'}]);
		await timeout(150);
		emit([{msg: 'Hallo Welt!'}, {msg: 'Bonjour monde!'}]);
		await timeout(200);
	}
300594
def trigger_async(channels, event_name, data, params = {})
      post_async('/events', trigger_params(channels, event_name, data, params))
    end
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)
  }
}
1250907
function postMessage(message, target, force) {
            var consumer;
            var parsed;

            if (!this.disposed) {
                try {
                    if (message) {
                        if (this.ready || force) {
                            // Post the message
                            consumer = target || this.receiver;
                            parsed = _prepareMessage.call(this, message);
                            consumer.postMessage(parsed);
                            return true;
                        }
                        else if (this.maxConcurrency >= this.messageQueue.length) {
                            // Need to delay/queue messages till target is ready
                            this.messageQueue.push(message);
                            return true;
                        }
                        else {
                            return false;
                        }
                    }
                }
                catch(ex) {
                    /* istanbul ignore next  */
                    PostMessageUtilities.log("Error while trying to post the message", "ERROR", "PostMessageChannel");
                    return false;
                }
            }
        }
492228
void deliverAsyncEvent(final EntryEvent<K,V> _event) {
    if (asyncListenerByType.get(_event.getEventType()).isEmpty()) {
      return;
    }
    List<Listener<K,V>> _listeners =
      new ArrayList<Listener<K, V>>(asyncListenerByType.get(_event.getEventType()));
    if (_listeners.isEmpty()) {
      return;
    }
    K key = _event.getKey();
    synchronized (getLockObject(key)) {
      Queue<EntryEvent<K,V>> q = keyQueue.get(key);
      if (q != null) {
        q.add(_event);
        return;
      }
      q = new LinkedList<EntryEvent<K, V>>();
      keyQueue.put(key, q);
    }
    runAllListenersInParallel(_event, _listeners);
  }
206341
def consume(self, timeout=None, loop=None):
        
        if self._consumer_fn is None:
            raise ValueError('Consumer function is not defined yet')

        logger.info('Start consuming the stream')

        @asyncio.coroutine
        def worker(conn_url):
            extra_headers = {
                'Connection': 'upgrade',
                'Upgrade': 'websocket',
                'Sec-Websocket-Version': 13,
            }

            ws = yield from websockets.connect(
                conn_url, extra_headers=extra_headers)

            if ws is None:
                raise RuntimeError("Couldn't connect to the '%s'" % conn_url)

            try:
                while True:
                    message = yield from ws.recv()
                    yield from self._consumer_fn(message)
            finally:
                yield from ws.close()

        if loop is None:
            loop = asyncio.new_event_loop()

        asyncio.set_event_loop(loop)
        try:
            task = worker(conn_url=self._conn_url)
            if timeout:
                logger.info('Running task with timeout %s sec', timeout)
                loop.run_until_complete(
                    asyncio.wait_for(task, timeout=timeout))
            else:
                loop.run_until_complete(task)
        except asyncio.TimeoutError:
            logger.info('Timeout is reached. Closing the loop')
            loop.close()
        except KeyboardInterrupt:
            logger.info('Closing the loop')
            loop.close()