Handling more websocket connections

Question

I have the following basic code, which connects to a websocket server and receives some data:

import websocket, json, time

def process_message(ws, msg):
    message = json.loads(msg)
    print(message)

def on_error(ws, error):
    print('Error', e)

def on_close(ws):
    print('Closing')

def on_open(ws):
    def run(*args):
        Subs = []
       
        tradeStr=  """{"method": "SUBSCRIBE", "params":%s, "id": 1}"""%(json.dumps(Subs))
        ws.send(tradeStr)

    thread.start_new_thread(run, ())

def Connect():
    websocket.enableTrace(False)
    ws = websocket.WebSocketApp("wss://myurl", on_message = process_message, on_error = on_error, on_close = on_close)
    ws.on_open = on_open
    ws.run_forever()

Connect()

Now, I would like to create more connections to different servers and receive data concurrently in the same script. I tried the following:

def run(url):

    def process_message(ws, msg):
        message = json.loads(msg)
        print(message)

    def on_error(ws, error):
        print('Error', e)

    def on_close(ws):
        print('Closing')

    def on_open(ws):
        def run(*args):
            Subs = []
           
            tradeStr=  """{"method": "SUBSCRIBE", "params":%s, "id": 1}"""%(json.dumps(Subs))
            ws.send(tradeStr)

        thread.start_new_thread(run, ())

    def Connect():
        websocket.enableTrace(False)
        ws = websocket.WebSocketApp(url, on_message = process_message, on_error = on_error, on_close = on_close)
        ws.on_open = on_open
        ws.run_forever()

    Connect()

threading.Thread(target=run, kwargs={'url': 'url1'}).start()
threading.Thread(target=run, kwargs={'url': 'url2'}).start()
threading.Thread(target=run, kwargs={'url': 'url3'}).start()

Now, this code works, but I'm connecting to different URLs and I'm streaming data from all of them, but it seemed to me an "hacky" solution. I also don't know if what I'm doing could be bad practice or not. Each connection will send around 600/700 small JSON dictionaries, and I need to update every record to the db.

So my question is: is this implementation ok? Since it works with threads, can it create problems in the long run? Should I do another library such as Tornado?

Answer 1

Welcome to code review community. Few initial thoughts, when writing python code; following PEP-8 style guide makes your code more maintainable. Which would amount to (but not limited to):

1. Functions and variables named in lower_snake_case.
2. Avoiding whitespaces in function parameters/arguments.

---

Moving on, you can put your websocket handling code in its own submodule. For eg.

class WebSocket:
    def __init__(self, url):
        self._url = url

    def close(self, ws):
        # handle closed connection callback
        pass

    def connect(self):
        self.ws = websocket.WebSocketApp(self._url, on_close=self.close)
        self.ws.on_open = self.open  # for eg.
        self.ws.run_forever()
    .
    .
    .

If you want it to support threading, you can extend this class with threading.Thread sub-classing (you may also choose to subclass multiprocessing.Process later):

class ThreadedWebSocket(WebSocket, threading.Thread):
    def run(self):
        # This simply calls `self.connect()`

this way, if you want to initialise threads:

url1_thread = ThreadedWebSocket(url=url1)
url1_thread.start()
url1_thread.join()

The subclassing is mostly a matter of preference.

---

In the on_open callback, you are dumping a dict to json and then inserting it to a string template. The following achieves the same result, and is more intuitive (imo):

def on_open(ws):
    def run(*args):
        subs = []
        trade =  {
            "method": "SUBSCRIBE",
            "params": subs,
            "id": 1
        }
        ws.send(json.dumps(trade))

---

Coming to your question about library alternates, and performance, network I/O is generally a blocking call. This was one of the major reason I suggested going with the websockets package. The other being; we've been using websockets at our organisation for over an year in production. And it has been working really impressively for us, handling a throughput of few GB data everyday with a single low-tier machine.

Answer 2

This is the same approach you used, just rewritten as a oop approach per pacmaninbw with a class socketConn inheriting from websocket.WebSocketApp

Because the on_message, on_error, and on_close functions you define are so short/simple you can set them directly in the overloading init with lambda functions instead of writing out each one separately and passing them as parameters

New connections are formed with ws = SocketConn('url'). You also should be looping over an list or set of urls if you're connecting to multiple (which you probably are doing in your actual code but worth noting just in case)

import websocket, json, threading, _thread

class SocketConn(websocket.WebSocketApp):
    def __init__(self, url):
        #WebSocketApp sets on_open=None, pass socketConn.on_open to init
        super().__init__(url, on_open=self.on_open)

        self.on_message = lambda ws, msg: print(json.loads(msg))
        self.on_error = lambda ws, e: print('Error', e)
        self.on_close = lambda ws: print('Closing')

        self.run_forever()

    def on_open(self, ws):
        def run(*args):
            Subs = []
            
            tradeStr={"method":"SUBSCRIBE", "params":Subs, "id":1}
            ws.send(json.dumps(tradeStr))

        _thread.start_new_thread(run, ())



if __name__ == '__main__':
    websocket.enableTrace(False)

    urls = {'url1', 'url2', 'url3'}


    for url in urls:
        # args=(url,) stops the url from being unpacked as chars
        threading.Thread(target=SocketConn, args=(url,)).start()