Skip to content

Commit

Permalink
Add example to illustrate how to achieve concurrency with blocking we…
Browse files Browse the repository at this point in the history
…bsocket subscriptions by using a simple thread an wrapping the problematic async call. (#218)
  • Loading branch information
sirEven authored Aug 14, 2024
1 parent a124005 commit 64a9e63
Showing 1 changed file with 56 additions and 0 deletions.
56 changes: 56 additions & 0 deletions v4-client-py-v2/examples/websoket_concurrency_example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
import asyncio
import threading
from typing import Any, Dict

from dydx_v4_client.indexer.candles_resolution import CandlesResolution
from dydx_v4_client.indexer.socket.websocket import IndexerSocket
from dydx_v4_client.network import TESTNET

ETH_USD = "ETH-USD"
RESOLUTION = CandlesResolution.ONE_MINUTE


class LiveCandleRepresentation:
def __init__(self):
self._ws = IndexerSocket(
TESTNET.websocket_indexer,
on_message=self.handle_message,
)
self._count = 1
self.representation: Dict[str, Any] = {}

def wrap_async_func(self) -> None:
# NOTE: ._ws.connect() is a blocking async function call
asyncio.run(self._ws.connect())

def start_websocket_connection(self) -> None:
t = threading.Thread(target=self.wrap_async_func)
t.start()

def handle_message(self, ws: IndexerSocket, message: dict):
if message["type"] == "connected":
ws.candles.subscribe(ETH_USD, RESOLUTION)

if message["type"] == "channel_batch_data":
if candle_dict := message["contents"][0]:
self.representation = candle_dict
print(f"Received {RESOLUTION.value}-candle update #{self._count}.\n")
self._count += 1


async def some_candle_query(live_candle: LiveCandleRepresentation):
while True:
if candle := live_candle.representation:
print(f"Query current candle: {candle}\n")
await asyncio.sleep(20) # Query every 20 seconds


async def test():
live_candle = LiveCandleRepresentation()
live_candle.start_websocket_connection()

tasks = [asyncio.create_task(some_candle_query(live_candle))]
await asyncio.gather(*tasks)


asyncio.run(test())

0 comments on commit 64a9e63

Please sign in to comment.