forked from goodboy/tractor
Enable stream backpressure by default, add `MsgStream.ctx: Context`
parent
3f2e33a120
commit
f7a1f3832f
tractor
|
@ -70,7 +70,7 @@ class MsgStream(trio.abc.Channel):
|
||||||
'''
|
'''
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
ctx: 'Context', # typing: ignore # noqa
|
ctx: Context, # typing: ignore # noqa
|
||||||
rx_chan: trio.MemoryReceiveChannel,
|
rx_chan: trio.MemoryReceiveChannel,
|
||||||
_broadcaster: Optional[BroadcastReceiver] = None,
|
_broadcaster: Optional[BroadcastReceiver] = None,
|
||||||
|
|
||||||
|
@ -83,6 +83,9 @@ class MsgStream(trio.abc.Channel):
|
||||||
self._eoc: bool = False
|
self._eoc: bool = False
|
||||||
self._closed: bool = False
|
self._closed: bool = False
|
||||||
|
|
||||||
|
def ctx(self) -> Context:
|
||||||
|
return self._ctx
|
||||||
|
|
||||||
# delegate directly to underlying mem channel
|
# delegate directly to underlying mem channel
|
||||||
def receive_nowait(self):
|
def receive_nowait(self):
|
||||||
msg = self._rx_chan.receive_nowait()
|
msg = self._rx_chan.receive_nowait()
|
||||||
|
@ -390,7 +393,7 @@ class Context:
|
||||||
# only set on the callee side
|
# only set on the callee side
|
||||||
_scope_nursery: Optional[trio.Nursery] = None
|
_scope_nursery: Optional[trio.Nursery] = None
|
||||||
|
|
||||||
_backpressure: bool = False
|
_backpressure: bool = True
|
||||||
|
|
||||||
async def send_yield(self, data: Any) -> None:
|
async def send_yield(self, data: Any) -> None:
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue