Receiving streaming data after implementing asyncio websockets as a class?

Question

My question is closely related to the following question on Stackoverflow and the documentation here. I am defining a websockets-connection as a class. Next, I create a new class where I call the earlier defined websocket-class as self.ws and tell which data to send to the websocket with self.request. My problem is that the current script only runs once, whereas my desired output is continuous data.

The second link shows that I can retrieve continuous / streaming data using

asyncio.get_event_loop().run_until_complete(call_api(json.dumps(msg)))

I include all of the above code in my code (call_api is defined differently due to the desire to write it as a class). Below is my code:

import sys, json
import asyncio
from websockets import connect

class EchoWebsocket:
   def __init__(self, URL, CLIENT_ID=None, CLIENT_SECRET=None):
      self.url = URL
      self.client_id = CLIENT_ID
      self.client_secret = CLIENT_SECRET

   async def __aenter__(self):
      self._conn = connect(self.url)
      self.websocket = await self._conn.__aenter__()
      return self

   async def __aexit__(self, *args, **kwargs):
      await self._conn.__aexit__(*args, **kwargs)

   async def send(self, message):
      await self.websocket.send(message)

   async def receive(self):
      return await self.websocket.recv()

class DERIBIT:
   def __init__(self):
      self.ws = EchoWebsocket(URL='wss://test.deribit.com/ws/api/v2')
      self.loop = asyncio.get_event_loop()
      self.request = \
                   {"jsonrpc": "2.0",
                    "method": "public/subscribe",
                    "id": 42,
                    "params": {
                        "channels": ["deribit_price_index.btc_usd"]}
                   }

   def get_ticks(self):
      return self.loop.run_until_complete(self.__async__get_ticks())

   async def __async__get_ticks(self):
      async with self.ws as echo:
         await echo.send(json.dumps(self.request))
         response = await echo.receive()
         print(response)


if __name__ == "__main__":
   deribit = DERIBIT()
   deribit.get_ticks()

This script gives the following output:

{"jsonrpc": "2.0", "method": "public/subscribe", "id": 42, "params": {"channels": ["deribit_price_index.btc_usd"]}}

whereas I would like to see

Please advice.

Have you tried adding a `while True` around the whole `async with` block in `__async__get_ticks`? You could also add a loop around `run_until_complete` in `get_ticks`, which is what the linked answer does, but adding the while in the async function is better style because it enables you to use `asyncio.run` and completely rid of the sync `get_ticks` method. — user4815162342, Apr 16 '20 at 21:39
hi, I tried that by placing `while True` after `async with self.ws as echo`, but I am getting the following error: raise self.connection_closed_exc() websockets.exceptions.ConnectionClosedOK: code = 1000 (OK), reason = over_limit — HJA24, Apr 17 '20 at 05:46
yes, it works before the connection gets closed. I think each time I print the output the connection closes and gets reconnected again, resulting in an overload. How to prevent this? — HJA24, Apr 17 '20 at 11:45

score 1 · Answer 1 · answered Apr 19 '20 at 12:30

I only worked with Tornado's websockets but they work pretty well and Tornado has many helpers for dealing with async code:

import json
import tornado
from tornado.ioloop import PeriodicCallback
from tornado.websocket import websocket_connect


class EchoWebsocket:

    def __init__(self, url, client_id=None, client_secret=None):
        self.url = url
        self.client_id = client_id
        self.client_secret = client_secret
        self.websocket = None

    async def connect(self):
        if not self.websocket:
            self.websocket = await websocket_connect(self.url)

    async def close(self):
        await self.websocket.close()
        self.websocket = None

    async def read(self):
        return await self.websocket.read_message()

    async def write(self, message):
        await self.websocket.write_message(message)


class DERIBIT:

    def __init__(self):
        self.ws = EchoWebsocket(url='wss://test.deribit.com/ws/api/v2')
        self.request = {
            "jsonrpc": "2.0",
            "method": "public/subscribe",
            "id": 42,
            "params": {
                "channels": ["deribit_price_index.btc_usd"]}
        }
        self.callback = PeriodicCallback(self.get_ticks, 1000)
        self.callback.start()

    async def get_ticks(self):
        if not self.ws.websocket:
            await self.ws.connect()
        await self.ws.write(json.dumps(self.request))
        response = await self.ws.read()
        print(response)


if __name__ == "__main__":
    deribit = DERIBIT()
    tornado.ioloop.IOLoop.current().start()

Output:

{"jsonrpc":"2.0","id":42,"result":["deribit_price_index.btc_usd"],"usIn":1587298852138977,"usOut":1587298852139023,"usDiff":46,"testnet":true}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587298851526,"price":7173.46,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587298852533,"price":7173.53,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","id":42,"result":["deribit_price_index.btc_usd"],"usIn":1587298852932540,"usOut":1587298852932580,"usDiff":40,"testnet":true}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587298852533,"price":7173.53,"index_name":"btc_usd"}}}

The example above could be simplified a lot if you integrate the websocket into the DERIBIT class rather than create a separate class for it.

score 1 · Answer 2 · answered Apr 23 '20 at 15:12

the problem is in the function

first loop.run_until_complete run until the future is complete doc run_until_complete
that mean your function receive will run only one response. run_until_complete is not a callback function!.

so in your case the main:
deribit.get_ticks() -> run the future instance __async__get_ticks
so __async__get_ticks is task: let's see what the task do:
1.open ws connection:
2.send request
3.wait the response of the ws
4. print(response)
here the task is done that why you see only one line

   async def __async__get_ticks(self):
      async with self.ws as echo:
         await echo.send(json.dumps(self.request))
         response = await echo.receive()
         print(response)

after explanation: the solution will be simple: need to wrap the line response with while

async def __async__get_ticks(self):
      async with self.ws as echo:
         await echo.send(json.dumps(self.request))
         while True:
                response = await echo.receive()
                print(response)

output

{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654476817,"price":7540.54,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654477824,"price":7540.52,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654478831,"price":7540.15,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654479838,"price":7539.83,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654480845,"price":7539.2,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654481852,"price":7538.96,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654482859,"price":7538.9,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654483866,"price":7538.89,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654484873,"price":7538.47,"index_name":"btc_usd"}}}
{"jsonrpc":"2.0","method":"subscription","params":{"channel":"deribit_price_index.btc_usd","data":{"timestamp":1587654485880,"price":7537.15,"index_name":"btc_usd"}}}

Receiving streaming data after implementing asyncio websockets as a class?

2 Answers2

Linked