Merge pull request #473 from pikers/binance_ws_ep_update

`binance`: use built-in `anext()` add note about new ws ep URL, fix agen streaming within `NoBsWs` usage

piker/brokers/binance.py

@@ -26,6 +26,7 @@ from typing import (
 )
 import time
 
+from trio_util import trio_async_generator
 import trio
 from trio_typing import TaskStatus
 import pendulum
@@ -317,7 +318,10 @@ class AggTrade(Struct):
     M: bool  # Ignore
 
 
-async def stream_messages(ws: NoBsWs) -> AsyncGenerator[NoBsWs, dict]:
+@trio_async_generator
+async def stream_messages(
+    ws: NoBsWs,
+) -> AsyncGenerator[NoBsWs, dict]:
 
     timeouts = 0
     while True:
@@ -529,19 +533,23 @@ async def stream_quotes(
                 # XXX: do we need to ack the unsub?
                 # await ws.recv_msg()
 
-        async with open_autorecon_ws(
+        async with (
+            open_autorecon_ws(
+                # XXX: see api docs which show diff addr?
+                # https://developers.binance.com/docs/binance-trading-api/websocket_api#general-api-information
+                # 'wss://ws-api.binance.com:443/ws-api/v3',
                 'wss://stream.binance.com/ws',
                 fixture=subscribe,
-        ) as ws:
+            ) as ws,
+
+            # avoid stream-gen closure from breaking trio..
+            stream_messages(ws) as msg_gen,
+        ):
+            typ, quote = await anext(msg_gen)
 
             # pull a first quote and deliver
-            msg_gen = stream_messages(ws)
-
-            typ, quote = await msg_gen.__anext__()
-
             while typ != 'trade':
-                # TODO: use ``anext()`` when it lands in 3.10!
+                typ, quote = await anext(msg_gen)
-                typ, quote = await msg_gen.__anext__()
 
             task_status.started((init_msgs,  quote))
 

piker/brokers/ib/_util.py

@@ -24,6 +24,10 @@ import subprocess
 
 import tractor
 
+from piker.log import get_logger
+
+log = get_logger(__name__)
+
 
 _reset_tech: Literal[
     'vnc',
@@ -134,6 +138,7 @@ def i3ipc_xdotool_manual_click_hack() -> None:
         # 'IB',  # gw running in i3 (newer version?)
     ]
 
+    try:
         for name in win_names:
             results = t.find_titled(name)
             print(f'results for {name}: {results}')
@@ -162,8 +167,8 @@ def i3ipc_xdotool_manual_click_hack() -> None:
                         'xdotool',
                         'windowactivate', '--sync', win_id,
 
-                    # move mouse to bottom left of window (where there should
+                        # move mouse to bottom left of window (where
-                    # be nothing to click).
+                        # there should be nothing to click).
                         'mousemove_relative', '--sync', str(w-4), str(h-4),
 
                         # NOTE: we may need to stick a `--retry 3` in here..
@@ -177,11 +182,10 @@ def i3ipc_xdotool_manual_click_hack() -> None:
                     )
 
     # re-activate and focus original window
-    try:
         subprocess.call([
             'xdotool',
             'windowactivate', '--sync', str(orig_win_id),
             'click', '--window', str(orig_win_id), '1',
         ])
     except subprocess.TimeoutExpired:
-        log.exception(f'xdotool timed out?')
+        log.exception('xdotool timed out?')

piker/brokers/kraken/feed.py

@@ -27,11 +27,11 @@ from typing import (
 )
 import time
 
-from async_generator import aclosing
 from fuzzywuzzy import process as fuzzy
 import numpy as np
 import pendulum
 from trio_typing import TaskStatus
+from trio_util import trio_async_generator
 import tractor
 import trio
 
@@ -122,6 +122,7 @@ async def stream_messages(
                 yield msg
 
 
+@trio_async_generator
 async def process_data_feed_msgs(
     ws: NoBsWs,
 ):
@@ -378,7 +379,12 @@ async def stream_quotes(
                 'wss://ws.kraken.com/',
                 fixture=subscribe,
             ) as ws,
-            aclosing(process_data_feed_msgs(ws)) as msg_gen,
+
+            # avoid stream-gen closure from breaking trio..
+            # NOTE: not sure this actually works XD particularly
+            # if we call `ws._connect()` manally in the streaming
+            # async gen..
+            process_data_feed_msgs(ws) as msg_gen,
         ):
             # pull a first quote and deliver
             typ, ohlc_last = await anext(msg_gen)

setup.py

@@ -53,6 +53,7 @@ setup(
         # async
         'trio',
         'trio-websocket',
+        'trio-util',
         'async_generator',
 
         # from github currently (see requirements.txt)