diff --git a/tractor/_streaming.py b/tractor/_streaming.py index 14067c6..68aff2a 100644 --- a/tractor/_streaming.py +++ b/tractor/_streaming.py @@ -9,7 +9,7 @@ from dataclasses import dataclass from typing import ( Any, Iterator, Optional, Callable, AsyncGenerator, Dict, - AsyncIterator, Awaitable + AsyncIterator ) import warnings @@ -263,7 +263,7 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel): self, # use memory channel size by default self._rx_chan._state.max_buffer_size, # type: ignore - receive_afunc=self.receive, + receive_afunc=self.receive, ) # NOTE: we override the original stream instance's receive @@ -276,6 +276,8 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel): # https://github.com/python/mypy/issues/708 async with self._broadcaster.subscribe() as bstream: + assert bstream.key != self._broadcaster.key + assert bstream._recv == self._broadcaster._recv yield bstream