def _async_route(self, msg, in_stream=None):
        """
        Arrange for `msg` to be forwarded towards its destination. If its
        destination is the local context, then arrange for it to be dispatched
        using the local handlers.

        This is a lower overhead version of :meth:`route` that may only be
        called from the :class:`Broker` thread.

        :param Stream in_stream:
            If not :data:`None`, the stream the message arrived on. Used for
            performing source route verification, to ensure sensitive messages
            such as ``CALL_FUNCTION`` arrive only from trusted contexts.
        """
        _vv and IOLOG.debug('%r._async_route(%r, %r)', self, msg, in_stream)

        if len(msg.data) > self.max_message_size:
            self._maybe_send_dead(msg, self.too_large_msg % (
                self.max_message_size,
            ))
            return

        # Perform source verification.
        if in_stream:
            parent = self._stream_by_id.get(mitogen.parent_id)
            expect = self._stream_by_id.get(msg.auth_id, parent)
            if in_stream != expect:
                LOG.error('%r: bad auth_id: got %r via %r, not %r: %r',
                          self, msg.auth_id, in_stream, expect, msg)
                return

            if msg.src_id != msg.auth_id:
                expect = self._stream_by_id.get(msg.src_id, parent)
                if in_stream != expect:
                    LOG.error('%r: bad src_id: got %r via %r, not %r: %r',
                              self, msg.src_id, in_stream, expect, msg)
                    return

            if in_stream.auth_id is not None:
                msg.auth_id = in_stream.auth_id

            # Maintain a set of IDs the source ever communicated with.
            in_stream.egress_ids.add(msg.dst_id)

        if msg.dst_id == mitogen.context_id:
            return self._invoke(msg, in_stream)

        out_stream = self._stream_by_id.get(msg.dst_id)
        if out_stream is None:
            out_stream = self._stream_by_id.get(mitogen.parent_id)

        if out_stream is None:
            self._maybe_send_dead(msg, self.no_route_msg,
                                  msg.dst_id, mitogen.context_id)
            return

        if in_stream and self.unidirectional and not \
                (in_stream.is_privileged or out_stream.is_privileged):
            self._maybe_send_dead(msg, self.unidirectional_msg,
                in_stream.remote_id, out_stream.remote_id)
            return

        out_stream._send(msg)
def _async_route ( self , msg , in_stream = None ) : _vv and IOLOG . debug ( '%r._async_route(%r, %r)' , self , msg , in_stream ) if len ( msg . data ) > self . max_message_size : self . _maybe_send_dead ( msg , self . too_large_msg % ( self . max_message_size , ) ) return if in_stream : parent = self . _stream_by_id . get ( mitogen . parent_id ) expect = self . _stream_by_id . get ( msg . auth_id , parent ) if in_stream != expect : LOG . error ( '%r: bad auth_id: got %r via %r, not %r: %r' , self , msg . auth_id , in_stream , expect , msg ) return if msg . src_id != msg . auth_id : expect = self . _stream_by_id . get ( msg . src_id , parent ) if in_stream != expect : LOG . error ( '%r: bad src_id: got %r via %r, not %r: %r' , self , msg . src_id , in_stream , expect , msg ) return if in_stream . auth_id is not None : msg . auth_id = in_stream . auth_id in_stream . egress_ids . add ( msg . dst_id ) if msg . dst_id == mitogen . context_id : return self . _invoke ( msg , in_stream ) out_stream = self . _stream_by_id . get ( msg . dst_id ) if out_stream is None : out_stream = self . _stream_by_id . get ( mitogen . parent_id ) if out_stream is None : self . _maybe_send_dead ( msg , self . no_route_msg , msg . dst_id , mitogen . context_id ) return if in_stream and self . unidirectional and not ( in_stream . is_privileged or out_stream . is_privileged ) : self . _maybe_send_dead ( msg , self . unidirectional_msg , in_stream . remote_id , out_stream . remote_id ) return out_stream . _send ( msg )
Arrange for `msg` to be forwarded towards its destination. If its
        destination is the local context, then arrange for it to be dispatched
        using the local handlers.

        This is a lower overhead version of :meth:`route` that may only be
        called from the :class:`Broker` thread.

        :param Stream in_stream:
            If not :data:`None`, the stream the message arrived on. Used for
            performing source route verification, to ensure sensitive messages
            such as ``CALL_FUNCTION`` arrive only from trusted contexts.