binance: wrap streamer async-gen in aclosing()

pikers · Mar 13, 2023 · 5a07542 · 5a07542
1 parent 889e920
commit 5a07542
Showing 1 changed file with 8 additions and 7 deletions.
diff --git a/piker/brokers/binance.py b/piker/brokers/binance.py
@@ -26,6 +26,7 @@
 )
 import time
 
+from async_generator import aclosing
 import trio
 from trio_typing import TaskStatus
 import pendulum
@@ -529,14 +530,14 @@ async def subscribe(ws: wsproto.WSConnection):
                 # XXX: do we need to ack the unsub?
                 # await ws.recv_msg()
 
-        async with open_autorecon_ws(
-            'wss://stream.binance.com/ws',
-            fixture=subscribe,
-        ) as ws:
-
+        async with (
+            open_autorecon_ws(
+                'wss://stream.binance.com/ws',
+                fixture=subscribe,
+            ) as ws,
+            aclosing(stream_messages(ws)) as msg_gen,
+        ):
             # pull a first quote and deliver
-            msg_gen = stream_messages(ws)
-
             typ, quote = await msg_gen.__anext__()
 
             while typ != 'trade':