diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index 256b35af..2889bddf 100644 --- a/piker/data/_web_bs.py +++ b/piker/data/_web_bs.py @@ -359,8 +359,8 @@ async def open_autorecon_ws( ''' -JSONRPC response-request style machinery for transparent multiplexing of msgs -over a NoBsWs. +JSONRPC response-request style machinery for transparent multiplexing +of msgs over a `NoBsWs`. ''' @@ -377,19 +377,35 @@ async def open_jsonrpc_session( url: str, start_id: int = 0, response_type: type = JSONRPCResult, - request_type: Optional[type] = None, - request_hook: Optional[Callable] = None, - error_hook: Optional[Callable] = None, + 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, ) -> Callable[[str, dict], dict]: 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 = 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