How to properly implement an async pub/sub client?

[F483]

I have the following solution but it does not work very well. It does not seem to disconnect and is very slow on joining.

#!/usr/bin/env python


import time
import threading
from socketIO_client import SocketIO


class TestSubscribe():

    def connect(self):
        baseurl = "http://v1.livenet.bitcoin.chromanode.net"
        self._socketIO = SocketIO(baseurl, 80)
        self._socketIO.on('new-block', self.on_newblock)
        self._socketIO.emit('subscribe', 'new-block')
        self._socketIO_thread = threading.Thread(target=self._socketIO.wait)
        self._socketIO_thread.start()

    def on_newblock(self, blockid, blockheight):
        print "new-block:", blockid, blockheight
        # add to input queue for further processing

    def disconnect(self):
        # XXX works ... badly
        print "disconnect before:", self._socketIO.connected
        self._socketIO.disconnect()
        self._socketIO_thread.join() # takes quite some time
        print "disconnect after:", self._socketIO.connected # still connected?
        self._socketIO = None
        self._socketIO_thread = None


testsubscribe = TestSubscribe()
testsubscribe.connect()
time.sleep(10)
testsubscribe.disconnect()

[invisibleroads]

Hi Fabian, please tell me if the above fix resolves your issue.

[F483]

While it still displays connected as true afterword, it does terminate much faster.

Thank you very much!

[invisibleroads]

That's good to know, I'll look into it later when I have more time.

[F483]

The fix looks a little to simple. Does that not just set wait to always be 1 and break the basic functionality? I'm not familiar with socketIO-client code base, will have to check if my other stuff still works.

[invisibleroads]

The fix only affects the websocket transport.

Previously, the websocket transport would block on recv() until ping_timeout.

Now, wait() blocks on recv() for one second and then loops more often, making the loop more responsive; there shouldn't be any changes in functionality.

[F483]

Good to know, thank you very much!

[tlc]

Blocks for one second? Is this still true? What is the granularity of wait(timeout) ?

I tried to follow the code but it looked like the recv() block was controlled by a connection setting value passed from the other side via engineIO_session. Why isn't the timeout from wait() passed down?