Here is example handler. My code shows error like "handle_text_stream is not awaited."
async def handle_text_stream(reader, participant_identity):
info = reader.info
print(
f'Text stream received from {participant_identity}\n'
f' Topic: {info.topic}\n'
f' Timestamp: {info.timestamp}\n'
f' ID: {info.id}\n'
f' Size: {info.size}' # Optional, only available if the stream was sent with `send_text`
)
# Option 1: Process the stream incrementally using an async for loop.
async for chunk in reader:
print(f"Next chunk: {chunk}")
# Option 2: Get the entire text after the stream completes.
text = await reader.read_all()
print(f"Received text: {text}")
room.register_text_stream_handler(
"my-topic",
handle_text_stream
)
As the coding in Room.py, the handler is called without async context.
def _handle_stream_header(
self, header: proto_room.DataStream.Header, participant_identity: str
):
stream_type = header.WhichOneof("content_header")
if stream_type == "text_header":
text_stream_handler = self._text_stream_handlers.get(header.topic)
if text_stream_handler is None:
logging.info(
"ignoring text stream with topic '%s', no callback attached",
header.topic,
)
return
text_reader = TextStreamReader(header)
self._text_stream_readers[header.stream_id] = text_reader
text_stream_handler(text_reader, participant_identity)
elif stream_type == "byte_header":
byte_stream_handler = self._byte_stream_handlers.get(header.topic)
if byte_stream_handler is None:
logging.info(
"ignoring byte stream with topic '%s', no callback attached",
header.topic,
)
return
byte_reader = ByteStreamReader(header)
self._byte_stream_readers[header.stream_id] = byte_reader
byte_stream_handler(byte_reader, participant_identity)
else:
logging.warning("received unknown header type, %s", stream_type)
Here is example handler. My code shows error like "handle_text_stream is not awaited."
As the coding in
Room.py, the handler is called without async context.