stream.py 文件源码

def put_message(self, message):
        """
        :param message: protobuf generated validator_pb2.Message
        """
        if not self._ready_event.is_set():
            return

        with self._condition:
            self._condition.wait_for(
                lambda: self._event_loop is not None
                and self._send_queue is not None
            )

        asyncio.run_coroutine_threadsafe(
            self._put_message(message),
            self._event_loop)