jc211
/
tractor
forked from goodboy/tractor
1
0
Fork 0
Code Pull Requests Activity

Add some bcaster ref sanity asserts around subscriptions

tokio_backup
Tyler Goodlet 2021-08-31 13:06:17 -04:00
parent c82ca67263
commit 727d666cb4
1 changed files with 4 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tractor/_streaming.py
View File

@ -9,7 +9,7 @@ from dataclasses import dataclass
from typing import ( from typing import (
Any, Iterator, Optional, Callable, Any, Iterator, Optional, Callable,
AsyncGenerator, Dict, AsyncGenerator, Dict,
AsyncIterator, Awaitable AsyncIterator
) )
import warnings import warnings
@ -263,7 +263,7 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel):
self, self,
# use memory channel size by default # use memory channel size by default
self._rx_chan._state.max_buffer_size, # type: ignore 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 # 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 # https://github.com/python/mypy/issues/708
async with self._broadcaster.subscribe() as bstream: async with self._broadcaster.subscribe() as bstream:
assert bstream.key != self._broadcaster.key
assert bstream._recv == self._broadcaster._recv
yield bstream yield bstream