KucoinWebsocketClient does not handle batch (un)subscribe correctly

[Dhaneor]

It seems like KucoinWebsocketClient can only handle single topics but not batch topics (string with comma-separated topics). According to the API documentation, batch requests are possible, for instance for tickers.

But the code on the subscribe/unsubscribe methods does not account for that. Everything is considered to be single topic and appended to the topics list. This causes no issue with subscriptions, but it does when trying to unsubscribe for multiple topics at once, because then the comma-separated string is only found in the topics list if it is the same string that was used to subscribe.

Here is the relevant code:

async def subscribe(self, topic):
    """Subscribe to a channel
    :param topic: required
    :type topic: str
    :returns: None
    """

    req_msg = {
        'type': 'subscribe',
        'topic': topic,
        'response': True
    }
    self._conn.topics.append(topic)
    await self._conn.send_message(req_msg)

async def unsubscribe(self, topic):
    """Unsubscribe from a topic

    :param topic: required
    :type topic: str
    :returns: None
    """

    req_msg = {
        'type': 'unsubscribe',
        'topic': topic,
        'response': True
    }
    self._conn.topics.remove(topic)
    await self._conn.send_message(req_msg)

It lacks a check if the topic contains commas, which would require a split before appending to or removing from the list.

[Dhaneor]

I've changed the code to this, and now it works for batch requests:

async def subscribe(self, topic):
    """Subscribe to a channel
    :param topic: required
    :type topic: str
    :returns: None
    """

    req_msg = {
        'type': 'subscribe',
        'topic': topic,
        'response': True
    }
    
    if "," in topic:
        prefix, topics = topic.split(":")
        single_topics = topics.split(",")
        
        for topic in single_topics:
            self._conn.topics.append(f"{prefix}:{topic}")
    else:
        self._conn.topics.append(topic)
                   
    await self._conn.send_message(req_msg)

async def unsubscribe(self, topic):
    """Unsubscribe from a topic

    :param topic: required
    :type topic: str
    :returns: None
    """

    req_msg = {
        'type': 'unsubscribe',
        'topic': topic,
        'response': True
    }
    
    if "," in topic:
        prefix, topics = topic.split(":")
        single_topics = topics.split(",")
        
        for topic in single_topics:
            self._conn.topics.remove(f"{prefix}:{topic}")
    else:
        self._conn.topics.remove(topic)
    
    await self._conn.send_message(req_msg)

I've never done a pull request, but I will try if I find the time.

[progressivehed]

Dear Dhaneor,

Hello. This is Hed from KuCoin API team.

Thanks for your nice and comprehensive feedback and sharing above.

The unsubscribe process should also be possible to be applied, like the way subscription happens for batch topics.

Here is an example:

//Spot Unsubscribe Topic
{
"id": "1545910840805", //The id should be an unique value
"type": "unsubscribe",
"topic": "/market/ticker:BTC-USDT,ETH-USDT", //Topic needs to be unsubscribed. Some topics support to divisional unsubscribe the informations of multiple trading pairs through ",".
"privateChannel": false,
"response": true //Whether the server needs to return the receipt information of this subscription or not. Set as false by default.
}

Here is the doc link: https://www.kucoin.com/docs/websocket/basic-info/unsubscribe/introduction

Please let me know if this is the same solution you have applied or not.

[Dhaneor]

Hi Hed,

Thank you for your answer. The link that you provided describes what I'm doing. But the issue is that, the code does not handle edge cases.

For instance, when the caller gives "/market/ticker:BTC-USDT,ETH-USDT" as topic for the subscription and later on sends "/market/ticker:BTC-USDT" to unsubscribe(), the topic won't be found in self._conn.topics. This results in an exception, which crashes the client.

All of this is probably not a problem for most users. But when the application changes topics dynamically and uses batch requests to limit the amount of API requests, this situation may occur quite often. I think, it should be possible to unsubscribe form single topics that originally were part of a batch request.

Locally, I've now changed the code as shown in my first comment, and it works as expected now without exceptions/crashing. If there are a lot of topics, another problem may come from that change - exceeding the rate limit when recovering topics after a reconnect (which probably should be another GH issue).

[progressivehed]

I will report all of above feedback to dev team.

I think this is something that DEV team should consider when updating SDKs, to avoid them happening in future.

Thanks a lot for your nice feedback.

Please share more feedback if you faced new issues in this regard again.

[Dhaneor]

Thanks a lot for considering this and for asking for more feedback! I've now forked the repository.

I will contact you again after implementing some changes, which IMO will further improve the WS code.