How to continuously listen to websocket streams?

[yagggi]

Below is the code copied from example with 2 lines removed

import asyncio
import os
import logging

from binance_sdk_derivatives_trading_usds_futures.derivatives_trading_usds_futures import (
    DerivativesTradingUsdsFutures,
    DERIVATIVES_TRADING_USDS_FUTURES_WS_STREAMS_PROD_URL,
    ConfigurationWebSocketStreams,
)


# Configure logging
logging.basicConfig(level=logging.INFO)

# Create configuration for the WebSocket Streams
configuration_ws_streams = ConfigurationWebSocketStreams(
    stream_url=os.getenv(
        "STREAM_URL", DERIVATIVES_TRADING_USDS_FUTURES_WS_STREAMS_PROD_URL
    )
)

# Initialize DerivativesTradingUsdsFutures client
client = DerivativesTradingUsdsFutures(config_ws_streams=configuration_ws_streams)


async def individual_symbol_book_ticker_streams():
    connection = None
    try:
        connection = await client.websocket_streams.create_connection()

        stream = await connection.individual_symbol_book_ticker_streams(
            symbol="btcusdt",
        )
        stream.on("message", lambda data: print(f"{data}"))

    except Exception as e:
        logging.error(f"individual_symbol_book_ticker_streams() error: {e}")
    finally:
        if connection:
            await connection.close_connection(close_session=True)

Then here's the result:

In [2]: asyncio.run(individual_symbol_book_ticker_streams())
INFO:root:Connecting to wss://fstream.binance.com/stream with proxy None
INFO:root:Establishing Websocket connection with id 00dad199-8c19-43a9-bc94-3d11fd080f92 to: wss://fstream.binance.com/stream
INFO:root:Subscribing to streams: ['btcusdt@bookTicker']
INFO:root:WebSocket 00dad199-8c19-43a9-bc94-3d11fd080f92 closed.

In [3]:

What's the recommanded way to keep connection open and continuously receive messages from websocket ?

[yagggi]

Currently my workaround is yield control with asyncio.sleep:

async def individual_symbol_book_ticker_streams():
    connection = None
    try:
        connection = await client.websocket_streams.create_connection()

        stream = await connection.individual_symbol_book_ticker_streams(
            symbol="btcusdt",
        )
        stream.on("message", lambda data: print(f"{data}"))
        while True:
            await asyncio.sleep(0)
    except Exception as e:
        logging.error(f"individual_symbol_book_ticker_streams() error: {e}")
    finally:
        if connection:
            await connection.close_connection(close_session=True)

I'm just wondering if there's more explicit way to achieve this.

[M-F-B-B]

put this as last line in your code (not in the function),or if you use main function as last line in:

await asyncio.Event().wait()

Also your finally code will close the connection regardless...

here is your complete code:

import asyncio
import os
import logging

from binance_sdk_derivatives_trading_usds_futures.derivatives_trading_usds_futures import (
    DerivativesTradingUsdsFutures,
    DERIVATIVES_TRADING_USDS_FUTURES_WS_STREAMS_PROD_URL,
    ConfigurationWebSocketStreams,
)


# Configure logging
logging.basicConfig(level=logging.INFO)

# Create configuration for the WebSocket Streams
configuration_ws_streams = ConfigurationWebSocketStreams(
    stream_url=os.getenv(
        "STREAM_URL", DERIVATIVES_TRADING_USDS_FUTURES_WS_STREAMS_PROD_URL
    )
)

# Initialize DerivativesTradingUsdsFutures client
client = DerivativesTradingUsdsFutures(config_ws_streams=configuration_ws_streams)


async def individual_symbol_book_ticker_streams():
    connection = None
    try:
        connection = await client.websocket_streams.create_connection()

        stream = await connection.individual_symbol_book_ticker_streams(
            symbol="btcusdt",
        )
        stream.on("message", lambda data: print(f"{data}"))

    except Exception as e:
        logging.error(f"individual_symbol_book_ticker_streams() error: {e}")
    # finally:
    #     if connection:
    #         await connection.close_connection(close_session=True)

# asyncio.run(individual_symbol_book_ticker_streams())

async def main():
    await individual_symbol_book_ticker_streams()
    await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

[yagggi]

@M-F-B-B Hi, I think wait for an void event is just as same as while loop, thanks for your reply, but I'm looking for an explicit way such as await stream.run() or something else, I'm also ok if there's none.

Closing connection in finnaly section is recommended since you don't need it anymore

[EntomoSci]

A solution for me was waiting an asyncio.Event to be set with process interruption using the following snippet:

import asyncio
import signal

# Setup gracefull shutdown.
shutdown_event = asyncio.Event()

def handle_shutdown(signum, frame):
    logging.info("Handling shutdown")
    shutdown_event.set()

for s in (signal.SIGINT, signal.SIGTERM):
    signal.signal(s, handle_shutdown)

Then I use it where we want to stop to allow the callback to listen continuously, like:

try:
    # Here is your connection's logon and subscription code
    await shutdown_event.wait()  # <- Solution
except Exception as e:
    # Here is your error handling code
finally:
    # Here is your connection cleansing code