pikers
/
piker
3
0
Fork 0
Code Issues 4 Pull Requests 6 Packages Projects Releases Wiki Activity

Pop subscriber streams on connection errors

backup_asyncify_input_modes
Tyler Goodlet 2021-06-25 00:53:32 -04:00
parent 998775dfd9
commit 2465c8fc78
1 changed files with 11 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
piker/data/_sampling.py
View File

@ -248,12 +248,15 @@ async def sample_and_broadcast(
# so far seems like no since this should all # so far seems like no since this should all
# be single-threaded. # be single-threaded.
log.error(f'{stream._ctx.chan.uid} dropped connection') log.error(f'{stream._ctx.chan.uid} dropped connection')
subs.remove((stream, tick_throttle))
async def uniform_rate_send( async def uniform_rate_send(
rate: float, rate: float,
quote_stream: trio.abc.ReceiveChannel, quote_stream: trio.abc.ReceiveChannel,
stream: tractor.MsgStream, stream: tractor.MsgStream,
) -> None: ) -> None:
sleep_period = 1/rate - 0.000616 sleep_period = 1/rate - 0.000616
@ -289,8 +292,14 @@ async def uniform_rate_send(
# TODO: now if only we could sync this to the display # TODO: now if only we could sync this to the display
# rate timing exactly lul # rate timing exactly lul
try:
await stream.send({first_quote['symbol']: first_quote}) await stream.send({first_quote['symbol']: first_quote})
break break
except trio.ClosedResourceError:
# if the feed consumer goes down then drop
# out of this rate limiter
log.warning(f'{stream} closed')
return
end = time.time() end = time.time()
diff = end - start diff = end - start