pikers
/
piker
4
1
Fork 0
Code Issues 10 Pull Requests 14 Packages Projects Releases Wiki Activity

Convert `iter_ohlc_periods()` to a `@tractor.context`

mkts_backup
Tyler Goodlet 2022-03-07 07:25:01 -05:00
parent 3dc87e0426
commit 00d7bb089f
1 changed files with 19 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
piker/data/_sampling.py
View File

@ -133,18 +133,20 @@ async def increment_ohlc_buffer(
# a given sample period.
subs = sampler.subscribers.get(delay_s, ())
for ctx in subs:
for stream in subs:
try:
await ctx.send_yield({'index': shm._last.value})
await stream.send({'index': shm._last.value})
except (
trio.BrokenResourceError,
trio.ClosedResourceError
):
log.error(f'{ctx.chan.uid} dropped connection')
subs.remove(ctx)
log.error(
f'{stream._ctx.chan.uid} dropped connection'
)
subs.remove(stream)
@tractor.stream
@tractor.context
async def iter_ohlc_periods(
ctx: tractor.Context,
delay_s: int,
@ -158,17 +160,19 @@ async def iter_ohlc_periods(
'''
# add our subscription
subs = sampler.subscribers.setdefault(delay_s, [])
subs.append(ctx)
await ctx.started()
async with ctx.open_stream() as stream:
subs.append(stream)
try:
# stream and block until cancelled
await trio.sleep_forever()
finally:
try:
subs.remove(ctx)
subs.remove(stream)
except ValueError:
log.error(
f'iOHLC step stream was already dropped for {ctx.chan.uid}?'
f'iOHLC step stream was already dropped {ctx.chan.uid}?'
)