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

Drop the `Router.clients: set`, `.subscribers` is enough

offline_dark_clearing
Tyler Goodlet 2022-10-05 13:08:53 -04:00
parent 525f805cdb
commit 2fbfe583dd
1 changed files with 15 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
piker/clearing/_ems.py
View File

@ -248,8 +248,6 @@ async def clear_dark_triggers(
await brokerd_orders_stream.send(brokerd_msg) await brokerd_orders_stream.send(brokerd_msg)
# book._msgflows[oid].maps.insert(0, live_req)
case _: case _:
raise ValueError(f'Invalid dark book entry: {cmd}') raise ValueError(f'Invalid dark book entry: {cmd}')
@ -333,9 +331,6 @@ class Router(Struct):
# broker to book map # broker to book map
books: dict[str, _DarkBook] = {} books: dict[str, _DarkBook] = {}
# order id to client stream map
clients: set[tractor.MsgStream] = set()
# sets of clients mapped from subscription keys # sets of clients mapped from subscription keys
subscribers: defaultdict[ subscribers: defaultdict[
str, # sub key, default fqsn str, # sub key, default fqsn
@ -577,7 +572,6 @@ class Router(Struct):
trio.BrokenResourceError, trio.BrokenResourceError,
): ):
to_remove.add(client_stream) to_remove.add(client_stream)
self.clients.remove(client_stream)
log.warning( log.warning(
f'client for {client_stream} was already closed?') f'client for {client_stream} was already closed?')
@ -1324,8 +1318,6 @@ async def _emsd_main(
# register the client side before starting the # register the client side before starting the
# brokerd-side relay task to ensure the client is # brokerd-side relay task to ensure the client is
# delivered all exisiting open orders on startup. # delivered all exisiting open orders on startup.
_router.clients.add(client_stream)
# TODO: instead of by fqsn we need a subscription # TODO: instead of by fqsn we need a subscription
# system/schema here to limit what each new client is # system/schema here to limit what each new client is
# allowed to see in terms of broadcasted order flow # allowed to see in terms of broadcasted order flow
@ -1345,29 +1337,20 @@ async def _emsd_main(
) )
finally: finally:
# try to remove client from "registry" # try to remove client from subscription registry
try:
_router.clients.remove(client_stream)
except KeyError:
log.warning(
f'Stream {client_stream._ctx.chan.uid}'
' was already dropped?'
)
_router.subscribers[fqsn].remove(client_stream) _router.subscribers[fqsn].remove(client_stream)
dialogs = _router.dialogs
for oid, client_streams in dialogs.items():
if client_stream in client_streams:
client_streams.remove(client_stream)
# TODO: for order dialogs left "alive" in for oid, client_streams in _router.dialogs.items():
# the ems this is where we should allow some client_streams.discard(client_stream)
# system to take over management. Likely we
# want to allow the user to choose what kind # TODO: for order dialogs left "alive" in
# of policy to use (eg. cancel all orders # the ems this is where we should allow some
# from client, run some algo, etc.) # system to take over management. Likely we
if not client_streams: # want to allow the user to choose what kind
log.warning( # of policy to use (eg. cancel all orders
f'Order dialog is being unmonitored:\n' # from client, run some algo, etc.)
f'{oid} ->\n{client_stream._ctx.chan.uid}' if not client_streams:
) log.warning(
f'Order dialog is not being monitored:\n'
f'{oid} ->\n{client_stream._ctx.chan.uid}'
)