ReconnectingWebsocket is not reconnecting

[MarcelBeining]

Hi there, I have the issue that the websocket seems to be closed after approx 60 seconds. I subscribed to a 3 tickers and my callback function informs me when it was not called for at least 15 sec. The first minute everything works fine, but then I do not receive messages anymore from the websocket. When I then run ksm.unsubscribe for my tickers, I get:

  File "C:/Users/xxxxxxxxxx", line 307, in kunsubscribe
    await self.ksm.unsubscribe('/market/ticker:ETH-BTC')

  File "C:\ProgramData\Anaconda3\lib\site-packages\kucoin\asyncio\websockets.py", line 239, in unsubscribe
    await self._conn.send_message(req_msg)

  File "C:\ProgramData\Anaconda3\lib\site-packages\kucoin\asyncio\websockets.py", line 133, in send_message
    await self._socket.send(json.dumps(msg))

  File "C:\ProgramData\Anaconda3\lib\site-packages\websockets\protocol.py", line 361, in send
    yield from self.ensure_open()

  File "C:\ProgramData\Anaconda3\lib\site-packages\websockets\protocol.py", line 501, in ensure_open
    self.close_code, self.close_reason) from self.transfer_data_exc

ConnectionClosed: WebSocket connection is closed: code = 1006 (connection closed abnormally [internal]), no reason

This error does not appear when I unsubscribe from the websocket before the 1 minute has passed. So altogether it seems the ReconnectingWebsocket class you created does not reconnect in this case as the Websocket connection is down according to the error. Any idea what is wrong?

[MarcelBeining]

According to https://docs.kucoin.com/#ping, it is crucial to send a ping message with an interval < 60 seconds, otherwise the connection gets closed by the server, but in asyncio/websockets.py you only send pings when there is already an error, which is too late, I guess. I did a very rude workaround by adding these lines after https://github.com/sammchardy/python-kucoin/blob/develop/kucoin/asyncio/websockets.py#L52:

                counter = 0
                while keep_waiting:
                    counter += 1
                    if counter > 5:
                        await self.send_ping()
                        counter = 0

This seems to fix the issue, however I guess you might want to implement it differently ;-)

[nielskool]

I will look into it tonight. For me it works fine. It has been running for a week now without problems.

[MarcelBeining]

tried it on linux and windows, on both machines I get a disconnect after 1 minute if not implementing my workaround

[MarcelBeining]

PS: Guess this is a better solution, considering the recommended 10 sec ping:

ts = time.time()
while keep_waiting:
     if time.time() - ts > 10:
            await self.send_ping()
            ts = time.time()

[MarcelBeining]

Seems it was a bug on my side, sorry for taking your time

[CopyPasteJedi]

websocket seems to be closed after approx 60 seconds

Have same problem. Just started testing and after 60 sec stopped to feed data and seeing only:" sleeping to keep loop open". Where was the problem?

[MarcelBeining]

actually some issues also still persist on my side, I am not sure if it is fault of kucoin or whats the problem. @nielskool : One issue I found with your implementation, I think: What happens if the websocket gets so many updates (because it was subscribed to many topics) such that the recv timeout never happens and thus your send_ping method is never called? if you do not send a ping, kucoin closes connection after 1 minute and then it is too late for sending the ping

[MarcelBeining]

Also it would be good if your class would have an attribute showing when the last pong message was received. right now all messages that do not contain "data" are filtered by your class, so it is not possible to check if pongs are received or not

[nielskool]

@MarcelBeining My code fails randomly sometimes after 6 hours and sometimes after 2 days. i have about 99 subscriptions.

For now I use the following code in case it lost the connection:

async def main():
    global loop, nr_no_events, fifo_buffer

    # callback function that receives messages from the socket
    async def handle_evt(msg):
        global nr_no_events
        global fifo_buffer
        ....
        ....
        nr_no_events = 0

    client = Client(api_key, api_secret, api_passphrase, sandbox=False)
    ksm = await KucoinSocketManager.create(loop, client, handle_evt)

    await ksm.subscribe('/market/match:'+sys.argv[1])

    while True:
        print(nr_no_events, "sleeping to keep loop open")
        nr_no_events = nr_no_events+1
        if(nr_no_events>2):
            try:
                try:
                    await ksm.unsubscribe('/market/match:'+sys.argv[1])
                except:
                    pass #possibly nothing to unsubscribe
                #resubscribe
                client = Client(api_key, api_secret, api_passphrase, sandbox=False)
                ksm = await KucoinSocketManager.create(loop, client, handle_evt)
                await ksm.subscribe('/market/match:'+sys.argv[1])
                nr_no_events = 0
            except:
                pass
        await asyncio.sleep(20)

Small note: I am not the owner of this repository.

[MarcelBeining]

@nielskool thanks for your example. I have a similar implementation but using time of no updates + a subscription where I know there should be an update every 2 sec (market/snapshot) to check if something's wrong

I see, I meant @sammchardy ^^

[CopyPasteJedi]

@nielskool Thx for solution.

@MarcelBeining it's good that working, but it's wast of subscripions per minute, if You want to monitor more markets than max 100. Like some prefered ones and rest with rotation every few seconds...You have only 20 subs/min left. Unless you know diffrent easy ways to monitor 200 or 300 markets?

[nielskool]

@Full4me I monitor about ~250 markets and send them to a database file. for now this is split over 3 instances which are managed by a manager. I think it is also possible to do this in a single file.

[CopyPasteJedi]

@nielskool Thx 4 answer. So I can make just few instances on one api key from same IP? What with that 100 max subs limit? It's 100 for every instance? My idea was to make this in threads and manage needed values with lists or dictionaries.

[nielskool]

@Full4me jep you can. I can subscribe to 100 markets per instance at the moment with the same API and ip

[MarcelBeining]

As explained in pull request https://github.com/sammchardy/python-kucoin/pull/59, there is indeed a problem with sending the ping. @sammchardy Please approve the PR. @nielskool The reason you might not have been affected by the issue is that ReconnectingWebsocketis automatically reconnecting when the websocket is closed after 1 min, which however means some update messages might become skipped.

Note: I had added logging into the send_ping callback and indeed it was never called before my change, instead there was always a ConnectionClosed error (catched in line https://github.com/sammchardy/python-kucoin/blob/develop/kucoin/asyncio/websockets.py#L67) after 1min resulting in reconnecting websocket

[nielskool]

@MarcelBeining You are indeed right, the send_ping is never called.