Enable stream backpressure by default, add `MsgStream.ctx: Context`
parent
55ec4d2a8b
commit
414a7ecc4c
|
@ -69,7 +69,7 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel):
|
||||||
'''
|
'''
|
||||||
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,
|
||||||
|
|
||||||
|
@ -82,6 +82,9 @@ class ReceiveMsgStream(trio.abc.ReceiveChannel):
|
||||||
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()
|
||||||
|
@ -380,7 +383,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