pikers
/
piker
3
0
Fork 0
Code Issues 2 Pull Requests 5 Packages Projects Releases Wiki Activity

Fix context attr lookup..

misc_backend_fixes
Tyler Goodlet 2021-12-20 14:16:13 -05:00
parent 7e9cbd7d9e
commit 54712827ee
1 changed files with 12 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
piker/data/_sampling.py
View File

@ -135,6 +135,7 @@ async def increment_ohlc_buffer(
async def iter_ohlc_periods( async def iter_ohlc_periods(
ctx: tractor.Context, ctx: tractor.Context,
delay_s: int, delay_s: int,
) -> None: ) -> None:
""" """
Subscribe to OHLC sampling "step" events: when the time Subscribe to OHLC sampling "step" events: when the time
@ -270,18 +271,20 @@ async def sample_and_broadcast(
trio.ClosedResourceError, trio.ClosedResourceError,
trio.EndOfChannel, trio.EndOfChannel,
): ):
# XXX: do we need to deregister here ctx = getattr(stream, '_ctx', None)
# if it's done in the fee bus code? if ctx:
# so far seems like no since this should all
# be single-threaded.
log.warning( log.warning(
f'{stream._ctx.chan.uid} dropped ' f'{ctx.chan.uid} dropped '
'`brokerd`-quotes-feed connection' '`brokerd`-quotes-feed connection'
) )
if tick_throttle: if tick_throttle:
assert stream.closed() assert stream.closed()
# await stream.aclose()
# XXX: do we need to deregister here
# if it's done in the fee bus code?
# so far seems like no since this should all
# be single-threaded. Doing it anyway though
# since there seems to be some kinda race..
subs.remove((stream, tick_throttle)) subs.remove((stream, tick_throttle))