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

Drop references to deprecated `tractor.msg.pub`

dark_vlm
Tyler Goodlet 2022-01-29 12:44:45 -05:00
parent 6a0fba1eb3
commit 95b31cbc0f
3 changed files with 7 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
piker/brokers/data.py
View File

@ -98,7 +98,7 @@ class BrokerFeed:
) )
@tractor.msg.pub(tasks=['stock', 'option']) @tractor.trionics.msgpub(tasks=['stock', 'option'])
async def stream_poll_requests( async def stream_poll_requests(
get_topics: Callable, get_topics: Callable,
get_quotes: Coroutine, get_quotes: Coroutine,
@ -293,7 +293,7 @@ async def start_quote_stream(
await stream_poll_requests( await stream_poll_requests(
# ``msg.pub`` required kwargs # ``trionics.msgpub`` required kwargs
task_name=feed_type, task_name=feed_type,
ctx=ctx, ctx=ctx,
topics=symbols, topics=symbols,

2
piker/brokers/kraken.py
View File

@ -553,8 +553,6 @@ async def stream_quotes(
quote = ohlc quote = ohlc
topic = quote['symbol'].lower() topic = quote['symbol'].lower()
# XXX: format required by ``tractor.msg.pub``
# requires a ``Dict[topic: str, quote: dict]``
await send_chan.send({topic: quote}) await send_chan.send({topic: quote})

11
piker/data/_sampling.py
View File

@ -176,12 +176,11 @@ async def sample_and_broadcast(
# TODO: ``numba`` this! # TODO: ``numba`` this!
for sym, quote in quotes.items(): for sym, quote in quotes.items():
# TODO: in theory you can send the IPC msg *before* # TODO: in theory you can send the IPC msg *before* writing
# writing to the sharedmem array to decrease latency, # to the sharedmem array to decrease latency, however, that
# however, that will require `tractor.msg.pub` support # will require at least some way to prevent task switching
# here or at least some way to prevent task switching # at the yield such that the array write isn't delayed while
# at the yield such that the array write isn't delayed # another consumer is serviced..
# while another consumer is serviced..
# start writing the shm buffer with appropriate # start writing the shm buffer with appropriate
# trade data # trade data