3
\$\begingroup\$

The original purpose of this script was saving data to a local file, which is still the primary purpose.

Since I may only obtain one copy of a data stream (not this particular Coinbase one; it's just a free example), I'd like to add a secondary purpose using websockets to mirror the data for a Python web-server, but absolutely without interfering with the primary purpose above. Here is what I have so far:

import asyncio
import websockets
import json
import aiofiles

from websockets.exceptions import ConnectionClosedOK

subscribers = set()
sub_limit = 3


async def mirror_ticks():
    async with websockets.connect("wss://ws-feed.pro.coinbase.com") as ws:
        await ws.send(json.dumps({
            "type": "subscribe",
            "product_ids": ["BTC-USD", "ETH-USD"],
            "channels": ["ticker"]
        }))
        while True:
            data = await ws.recv()
            # Critical primary purpose: Save data to local file
            async with aiofiles.open('filename', mode='a') as file:
                await file.write(data + '\n')
            # Secondary purpose: Mirror data to Python web-server
            for sub in subscribers:
                try:
                    await sub.send(data)
                except:
                    pass


async def subscribe_ws(ws, path):
    if len(subscribers) >= sub_limit:
        ws.send("All client slots are currently in use")
        print(f"Slots full, turned away: {ws.request_headers.get('User-Agent', '')}")
    subscribers.add(ws)
    while True:
        try:
            await ws.recv()
        except ConnectionClosedOK:
            print(f"Client disconnected: {ws.request_headers.get('User-Agent', '')}")
            subscribers.discard(ws)
            break


start_server = websockets.serve(subscribe_ws, "localhost", 8765)
asyncio.get_event_loop().run_until_complete(asyncio.gather(start_server, mirror_ticks()))
# asyncio.get_event_loop().run_forever()

I've added some precautions by limiting the number of subscribers to prevent overloading the secondary task from gobbling up resources, and I've tried to make sure any disconnects are handled so that it doesn't interfere with the primary purpose.

  1. Any edge-cases or unexpected consequences to beware of here? Eg. could a misbehaving subscriber somehow still interfere with the primary purpose?
  2. The last line is commented-out because it doesn't seem necessary, can I remove it? Where would it be necessary?
\$\endgroup\$

1 Answer 1

1
\$\begingroup\$

I'm not familiar with how websocket works, but from your code I think I can infer it. So here a few comments:

It seems your subscribers list depends on currently opened connections. That is bad practise. The first reason being, HTTP sessions have a timeout. Hence, after a while (I think the default is a couple of minutes), you will lose connection to your subscribers. Keeping the connection open, you are also wasting resources, as you don't really need a connection until you are about to send something.

Also, it seems you are only removing a subscriber from the list if the connection closes correctly. However, what happens if the connection is closed unexpectedly? Instead, you should have caught ConnectionClosed exception.

To solve both these issues, you could have your subscribers be web servers. To subscribe, they would send the server a message with a url to which send the data (and the close the connection), and the server would send them the data when available. Then, the subscribers could send another message to be unsubscribed.

Now, regarding the second question, both lines of code would have the same behaviour as mirror_ticks has an infinite loop. Hence, no matter if you wait for it to finish, it never will

Edit

You are also catching all exceptions:

try:
    await sub.send(data)
except:
    pass

That is an awful practice, as that would catch all errors, like a SyntaxError. If you want to catch all websocket exceptions, you should catch WebSocketException instead.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.