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

Minor style changes and warning on unexpected msg

size_in_shm_token
Guillermo Rodriguez 2022-08-27 09:12:02 -03:00
parent 4facd161a9
commit 0c323fdc0b
No known key found for this signature in database
GPG Key ID: EC3AB66D5D83B392
1 changed files with 18 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
piker/data/_web_bs.py
View File

@ -179,8 +179,8 @@ async def open_jsonrpc_session(
trio.open_nursery() as n, trio.open_nursery() as n,
open_autorecon_ws(url) as ws open_autorecon_ws(url) as ws
): ):
_rpc_id: Iterable = count(start_id) rpc_id: Iterable = count(start_id)
_rpc_results: dict[int, dict] = {} rpc_results: dict[int, dict] = {}
async def json_rpc(method: str, params: dict) -> dict: async def json_rpc(method: str, params: dict) -> dict:
''' '''
@ -189,31 +189,31 @@ async def open_jsonrpc_session(
''' '''
msg = { msg = {
'jsonrpc': '2.0', 'jsonrpc': '2.0',
'id': next(_rpc_id), 'id': next(rpc_id),
'method': method, 'method': method,
'params': params 'params': params
} }
_id = msg['id'] _id = msg['id']
_rpc_results[_id] = { rpc_results[_id] = {
'result': None, 'result': None,
'event': trio.Event() 'event': trio.Event()
} }
await ws.send_msg(msg) await ws.send_msg(msg)
await _rpc_results[_id]['event'].wait() await rpc_results[_id]['event'].wait()
ret = _rpc_results[_id]['result'] ret = rpc_results[_id]['result']
del _rpc_results[_id] del rpc_results[_id]
if ret.error is not None: if ret.error is not None:
raise Exception(json.dumps(ret.error, indent=4)) raise Exception(json.dumps(ret.error, indent=4))
return ret return ret
async def _recv_task(): async def recv_task():
''' '''
receives every ws message and stores it in its corresponding result receives every ws message and stores it in its corresponding result
field, then sets the event to wakeup original sender tasks. field, then sets the event to wakeup original sender tasks.
@ -221,17 +221,18 @@ async def open_jsonrpc_session(
async for msg in ws: async for msg in ws:
msg = dtype(**msg) msg = dtype(**msg)
if msg.id not in _rpc_results: if msg.id not in rpc_results:
# in case this message wasn't beign accounted for store it log.warning(f'Wasn\'t expecting ws msg: {json.dumps(msg, indent=4)}')
_rpc_results[msg.id] = {
'result': None,
'event': trio.Event()
}
_rpc_results[msg.id]['result'] = msg res = rpc_results.setdefault(
_rpc_results[msg.id]['event'].set() msg.id,
{'result': None, 'event': trio.Event()}
)
res['result'] = msg
res['event'].set()
n.start_soon(_recv_task) n.start_soon(recv_task)
yield json_rpc yield json_rpc
n.cancel_scope.cancel() n.cancel_scope.cancel()