diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index aaca7b41..f95c2977 100644 --- a/piker/data/_web_bs.py +++ b/piker/data/_web_bs.py @@ -360,7 +360,7 @@ async def open_autorecon_ws( ''' JSONRPC response-request style machinery for transparent multiplexing -of msgs over a NoBsWs. +of msgs over a `NoBsWs`. ''' @@ -377,6 +377,16 @@ async def open_jsonrpc_session( url: str, start_id: int = 0, response_type: type = JSONRPCResult, + msg_recv_timeout: float = float('inf'), + # ^NOTE, since only `deribit` is using this jsonrpc stuff atm + # and options mkts are generally "slow moving".. + # + # FURTHER if we break the underlying ws connection then since we + # don't pass a `fixture` to the task that manages `NoBsWs`, i.e. + # `_reconnect_forever()`, the jsonrpc "transport pipe" get's + # broken and never restored with wtv init sequence is required to + # re-establish a working req-resp session. + # request_type: Optional[type] = None, # request_hook: Optional[Callable] = None, # error_hook: Optional[Callable] = None, @@ -388,12 +398,18 @@ async def open_jsonrpc_session( async with ( trio.open_nursery() as n, - open_autorecon_ws(url) as ws + open_autorecon_ws( + url=url, + msg_recv_timeout=msg_recv_timeout, + ) as ws ): rpc_id: Iterable[int] = count(start_id) rpc_results: dict[int, dict] = {} - async def json_rpc(method: str, params: dict) -> dict: + async def json_rpc( + method: str, + params: dict, + ) -> dict: ''' perform a json rpc call and wait for the result, raise exception in case of error field present on response