def send_message(self, msg, connection_id=None):
        """
        :param msg: protobuf validator_pb2.Message
        """
        zmq_identity = None
        if connection_id is not None and self._connections is not None:
            if connection_id in self._connections:
                connection_info = self._connections.get(connection_id)
                if connection_info.connection_type == \
                        ConnectionType.ZMQ_IDENTITY:
                    zmq_identity = connection_info.connection
            else:
                LOGGER.debug("Can't send to %s, not in self._connections",
                             connection_id)

        self._ready.wait()

        if zmq_identity is None:
            message_bundle = [msg.SerializeToString()]
        else:
            message_bundle = [bytes(zmq_identity),
                              msg.SerializeToString()]

        try:
            asyncio.run_coroutine_threadsafe(
                self._send_message_frame(message_bundle),
                self._event_loop)
        except RuntimeError:
            # run_coroutine_threadsafe will throw a RuntimeError if
            # the eventloop is closed. This occurs on shutdown.
            pass
def send_message ( self , msg , connection_id = None ) : zmq_identity = None if connection_id is not None and self . _connections is not None : if connection_id in self . _connections : connection_info = self . _connections . get ( connection_id ) if connection_info . connection_type == ConnectionType . ZMQ_IDENTITY : zmq_identity = connection_info . connection else : LOGGER . debug ( "Can't send to %s, not in self._connections" , connection_id ) self . _ready . wait ( ) if zmq_identity is None : message_bundle = [ msg . SerializeToString ( ) ] else : message_bundle = [ bytes ( zmq_identity ) , msg . SerializeToString ( ) ] try : asyncio . run_coroutine_threadsafe ( self . _send_message_frame ( message_bundle ) , self . _event_loop ) except RuntimeError : pass
:param msg: protobuf validator_pb2.Message