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

Symbol subs must be cid specific

kivy_mainline_and_py3.8
Tyler Goodlet 2018-12-23 20:48:06 -05:00
parent 11222e1176
commit eb8c9e1a99
1 changed files with 6 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
piker/brokers/data.py
View File

@ -159,17 +159,16 @@ async def fan_out_to_chans(
): ):
chan_payloads = {} chan_payloads = {}
for quote in quotes: for quote in quotes:
# set symbol quotes for each subscriber packet = {quote['symbol']: quote}
# for chan, cid in symbols2chans.get(quote['key'], set()): for chan, cid in symbols2chans.get(quote['key'], set()):
for chan, cid in symbols2chans[quote['key']]:
chan_payloads.setdefault( chan_payloads.setdefault(
chan, (chan, cid),
{'yield': {}, 'cid': cid} {'yield': {}, 'cid': cid}
)['yield'].update({quote['symbol']: quote}) )['yield'].update(packet)
# deliver to each subscriber (fan out) # deliver to each subscriber (fan out)
if chan_payloads: if chan_payloads:
for chan, payload in chan_payloads.items(): for (chan, cid), payload in chan_payloads.items():
try: try:
await chan.send(payload) await chan.send(payload)
except ( except (
@ -269,7 +268,7 @@ def modify_quote_stream(broker, feed_type, symbols, chan, cid):
chanset = symbols2chans.get(ticker) chanset = symbols2chans.get(ticker)
# XXX: cid will be different on unsub call # XXX: cid will be different on unsub call
for item in chanset.copy(): for item in chanset.copy():
if chan in item: if (chan, cid) == item:
chanset.discard(item) chanset.discard(item)
if not chanset: if not chanset: