MemoryObjectStream print message after receive another message

[andrea-mucci]

I will intent to explain what happen, i'm quite sure that this behaviour depend mostly of my ignorance with anyio, anyway I will appreciate some help

I have a service class, a broker class and a consumer class:

the Service class call the Broker class and pass the send_event_stream

class Service:
    # ... dome additional code above
     async def receive_events(self, receive_event_stream: MemoryObjectReceiveStream[dict[str, Any]]):
           async with receive_event_stream:
               async for item in receive_event_stream:
                   print('received', item)

    def start():
         send_event_stream, receive_event_stream = create_memory_object_stream[dict[str, Any]](max_buffer_size=1)
        async with create_task_group() as tg:
            if events_processes:
                consumer = Broker(self._settings.broker)
                tg.start_soon(consumer.start, list(self._events_routes.keys()), send_event_stream, tg)
                tg.start_soon(self.receive_events, receive_event_stream)

the Broker class, receive the send_event and pass to the consumer class

class Broker:
    #.,.... some code above
    async def start(self, topics: list[str],
              send_event_stream: MemoryObjectSendStream[dict[str, Any]],
              tg: TaskGroup) -> None:

        consumer = Consumer(self._settings, topics)
        tg.start_soon(consumer.start, send_event_stream)

finally the Consumer class
this class open a connection with a broker and start a while loop waiting some message

class Consumer:
   # additional code above 
   async def start(self,
                    send_event: MemoryObjectSendStream[dict[str, Any]]):
        broker.subscribe(.....)

        try:
            while True:
                msg = broker.poll(timeout=1.0)
                if msg is None:
                    continue
                elif msg.error():
                    raise BrokerConsumerException(msg.error())
                else:
                    await send_event.send({
                        'topic': msg.topic(),
                        'content': msg.value().decode('utf-8')
                    })
                    print("message received")
        except KeyboardInterrupt:
            sys.stderr.write("Aborted By User")
        finally:
            broker.close()

what happen:

i start the service with run(), i send from the broker a message and i get "message received" but no the message sent by the MemoryObjectStream, i send another message from the broker and i get the print of the previous message

seems that the MessageStream flush the content a step forward or maybe i do something wrong

could someone help me to understand what happen?

cheers

[graingert]

Can you create a complete runable example, and show the actual output and your expected output?