Given a message, route it either to the incoming queue, or to the
        future associated with its correlation_id.
9750
async def route_msg(self, msg):
        
        if msg.correlation_id in self._futures:
            self._set_reply(msg.correlation_id, msg)
        else:
            await self._push_incoming(msg)
25636
async def _on_message(self, update):
        """"""

        name = update['service']
        message_obj = update['payload']

        if name not in self.services:
            return

        with self._state_lock:
            self.services[name].post_message(message_obj['level'], message_obj['message'])
242289
def inplace_delete_message_queue(
        chain_state: ChainState,
        state_change: Union[ReceiveDelivered, ReceiveProcessed],
        queueid: QueueIdentifier,
) -> None:
    
    queue = chain_state.queueids_to_queues.get(queueid)
    if not queue:
        return

    inplace_delete_message(
        message_queue=queue,
        state_change=state_change,
    )

    if len(queue) == 0:
        del chain_state.queueids_to_queues[queueid]
    else:
        chain_state.queueids_to_queues[queueid] = queue
411755
def _on_trace(self, sequence, topic, message):
        

        try:
            conn_key = self._find_connection(topic)
            conn_id = self.conns.get_connection_id(conn_key)
        except ArgumentError:
            self._logger.warn("Dropping trace message that does not correspond with a known connection, topic=%s", topic)
            return

        try:
            tracing = messages.TracingNotification.verify(message)
            self._trigger_callback('on_trace', conn_id, tracing['trace'])
        except Exception:
            self._logger.exception("Error processing trace conn_id=%d", conn_id)
1184593
def process(self, msg, kwargs):
        
        kwargs['extra'] = {
            'correlation_id': self.parent.correlation_id,
            'parent': self.parent.name
        }
        return msg, kwargs
1142939
def _get_path(queue_name=None, include_messages=None, message_id=None):
    '''
    
    '''
    if queue_name and include_messages and message_id:
        return '/{0}/messages/{1}'.format(_str(queue_name), message_id)
    if queue_name and include_messages:
        return '/{0}/messages'.format(_str(queue_name))
    elif queue_name:
        return '/{0}'.format(_str(queue_name))
    else:
        return '/'
327330
def process_notifications(self, messages):
        """"""
        for object_id, object_size, metadata_size in messages:
            if object_size > 0 and object_id in self._waiting_dict:
                linked_list = self._waiting_dict[object_id]
                self._complete_future(linked_list)
1268152
def _got_response(self, msg):
        '''
            
        '''

        #logger.debug("MSG: %r" % msg)

        resp_id = msg.get('id', None)

        if resp_id is None:
            # subscription traffic comes with method set, but no req id.
            method = msg.get('method', None)
            if not method:
                logger.error("Incoming server message had no ID nor method in it", msg)
                return

            # not obvious, but result is on params, not result, for subscriptions
            result = msg.get('params', None)

            logger.debug("Traffic on subscription: %s" % method)

            subs = self.subscriptions.get(method)
            for q in subs:
                self.loop.create_task(q.put(result))

            return

        assert 'method' not in msg
        result = msg.get('result')

        # fetch and forget about the request
        inf = self.inflight.pop(resp_id) 
        if not inf:
            logger.error("Incoming server message had unknown ID in it: %s" % resp_id)
            return

        # it's a future which is done now
        req, rv = inf

        if 'error' in msg:
            err = msg['error']
            
            logger.info("Error response: '%s'" % err)
            rv.set_exception(ElectrumErrorResponse(err, req))

        else:
            rv.set_result(result)
726625
def _receive_message(self):
        
        while True:
            try:
                if self._socket.getsockopt(zmq.TYPE) == zmq.ROUTER:
                    zmq_identity, msg_bytes = \
                        yield from self._socket.recv_multipart()
                    if msg_bytes == b'':
                        # send ACK for connection probes
                        LOGGER.debug("ROUTER PROBE FROM %s", zmq_identity)
                        self._socket.send_multipart(
                            [bytes(zmq_identity), msg_bytes])
                    else:
                        self._received_from_identity(zmq_identity)
                        self._dispatcher_queue.put_nowait(
                            (zmq_identity, msg_bytes))
                else:
                    msg_bytes = yield from self._socket.recv()
                    self._last_message_time = time.time()
                    self._dispatcher_queue.put_nowait((None, msg_bytes))
                self._get_queue_size_gauge(self.connection).set_value(
                    self._dispatcher_queue.qsize())

            except CancelledError:  # pylint: disable=try-except-raise
                # The concurrent.futures.CancelledError is caught by asyncio
                # when the Task associated with the coroutine is cancelled.
                # The raise is required to stop this component.
                raise
            except Exception as e:  # pylint: disable=broad-except
                LOGGER.exception("Received a message on address %s that "
                                 "caused an error: %s", self._address, e)
1055269
def consume(self, message):
        
        if 'jsonrpc' not in message or message['jsonrpc'] != JSONRPC_VERSION:
            log.warn("Unknown message type %s", message)
            return

        if 'id' not in message:
            log.debug("Handling notification from client %s", message)
            self._handle_notification(message['method'], message.get('params'))
        elif 'method' not in message:
            log.debug("Handling response from client %s", message)
            self._handle_response(message['id'], message.get('result'), message.get('error'))
        else:
            try:
                log.debug("Handling request from client %s", message)
                self._handle_request(message['id'], message['method'], message.get('params'))
            except JsonRpcException as e:
                log.exception("Failed to handle request %s", message['id'])
                self._consumer({
                    'jsonrpc': JSONRPC_VERSION,
                    'id': message['id'],
                    'error': e.to_dict()
                })
            except Exception:  # pylint: disable=broad-except
                log.exception("Failed to handle request %s", message['id'])
                self._consumer({
                    'jsonrpc': JSONRPC_VERSION,
                    'id': message['id'],
                    'error': JsonRpcInternalError.of(sys.exc_info()).to_dict()
                })
888577
def stringReceived(self, response):
        
        correlation_id = response[0:4]
        try:
            d = self._pending.pop(correlation_id)
        except KeyError:
            self._log.warn((
                "Response has unknown correlation ID {correlation_id!r}."
                " Dropping connection to {peer}."
            ), correlation_id=correlation_id, peer=self.transport.getPeer())
            self.transport.loseConnection()
        else:
            d.callback(response)
645968
function(params, cb) {
    var correlationId = _cbt.put(offsetResponse(cb));
    var msg = envelope(off.encode()
                          .correlation(correlationId)
                          .client(_options.clientId)
                          .replica()
                          .topic(params.topic)
                          .partition(params.partition)
                          .timestamp()
                          .maxOffsets()
                          .end());
    sendMsg(msg, correlationId, cb);
  }
1103349
def dispatch(self, message: Message) -> Iterator[Any]:
        
        if "text" in message:
            text = message["text"] or ""
        elif "message" in message:
            text = message["message"].get("text", "")
        else:
            text = ""

        msg_subtype = message.get("subtype")

        for subtype, matchs in itertools.chain(
            self._routes[message["channel"]].items(), self._routes["*"].items()
        ):
            if msg_subtype == subtype or subtype is None:
                for match, endpoints in matchs.items():
                    if match.search(text):
                        yield from endpoints
40904
function(id, msg) {
        var q = queues[id];
        // execute failure callback
        if (q.onFailure) {
            var sc=q.scope || q.win;
            q.onFailure.call(sc, _returnData(q, msg));
        }
    }
657704
async def probe_message(self, _message, context):
        
        client_id = context.user_data

        await self.probe(client_id)
1043866
def message_received(self, message):
        """"""
        # If the message identifer is outstanding, then someone is
        # waiting for the respone so we save it here
        identifier = message.identifier or 'type_' + str(message.type)
        if identifier in self._outstanding:
            outstanding = OutstandingMessage(
                self._outstanding[identifier].semaphore, message)
            self._outstanding[identifier] = outstanding
            self._outstanding[identifier].semaphore.release()
        else:
            asyncio.ensure_future(self._dispatch(message), loop=self.loop)
980209
public static String getCorrelationId(Message message) {
        String correlationId = (String) message.get(CORRELATION_ID_KEY);
        if(null == correlationId) {
            correlationId = readCorrelationId(message);
        }
        if(null == correlationId) {
            correlationId = readCorrelationIdSoap(message);
        }
        return correlationId;
    }