5
\$\begingroup\$

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?

\$\endgroup\$
6
  • \$\begingroup\$ Welcome to code review. You might want to consider an object oriented approach, but that is for others to discuss in their answers. \$\endgroup\$
    – pacmaninbw
    Sep 14 '20 at 17:51
  • \$\begingroup\$ Thank you! I was considering this too. I'd like to hear some points of view before though! \$\endgroup\$
    – JayK23
    Sep 14 '20 at 18:10
  • \$\begingroup\$ Is this the websocket package in use? pypi.org/project/websocket_client \$\endgroup\$
    – hjpotter92
    Sep 17 '20 at 17:11
  • \$\begingroup\$ Yes, It's that package \$\endgroup\$
    – JayK23
    Sep 17 '20 at 17:13
  • \$\begingroup\$ open to use alternative libraries? check the why/why not for another package: websockets here: github.com/aaugustin/websockets#why-should-i-use-websockets \$\endgroup\$
    – hjpotter92
    Sep 17 '20 at 18:26
3
+100
\$\begingroup\$

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.

\$\endgroup\$
1
  • \$\begingroup\$ Thank you a lot. This was very helpful. Yes, since i'll have multiple connections and a lot of data coming in, moving to websockets might be the best choice \$\endgroup\$
    – JayK23
    Sep 19 '20 at 13:28
3
\$\begingroup\$

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()
\$\endgroup\$
1
  • 1
    \$\begingroup\$ Thank you a lot! This was very helpful. Making it OOP oriented is something i didn't consider. Since it's not my field it was really helpful \$\endgroup\$
    – JayK23
    Sep 19 '20 at 13:26

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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