Add support for combined ws streams

[tiagosiebler]

There's two types of combined streams to support, each with a diff use case. The first one is easier than the second:

• static
  • topics are known before the stream is opened
  • these topics wont ever change for this stream after it is opened
  • if disconnected, this can just respawn using the URL of the stream
• dynamic
  • exact topics are unknown before the stream opens
  • we need the ability to subscribe to new topics after the stream opens
    • a partial implementation is here.
    • I'm not sure how to determine the wskey. We could just dynamically generate one, and increment if the max limit is hit and a new one is made (see below). This is essentially just a primary key used to store state for a ws connection.
  • we need the ability to unsubscribe from some topics after the stream opens.
    • a partial implementation is here.
    • Same problem as before, not sure yet how to determine the wskey used to store a list of topics for that connection (or how to find the correct wskey using the topic passed into unsubscribe()). Could just be a query in wsStore to find any/all wsKeys that have that topic.
  • we will need to persist which topics are currently subscribed for this stream
    • if the stream disconnects, we can reconnect and we know which topics to re-subscribe to.
    • already partially implemented in this.wsStore, see the above partial implementation.
    • needs to be refactored similar to FTX connector, because topics are objects (not just strings).
    • if easy to do, would be nice to keep it consistent with how the ftx connector does it (as it also tracks topics using objects used to sub to them).

Stretch goals for dynamic combined streams

• we may need to track how many topics are currently active on a stream (because there is a max limit of 200 per stream).
  • if that cap is reached,
    • we may need to see if we have other combined streams
    • maybe using a wskey prefix (get all wskeys that start with `combined)
    • or some dictionary to track this state
  • if another combined stream is available AND max limit is not reached, request subscribe via that combined stream
  • if not, spawn new combined stream (with new wskey) and send topic sub into the newly spawned stream.

[MrAlvaroPS]

Cheers!!

I'm not sure how to create a combined stream with this API.

I think I just made a workaround that makes it, but maybe you have it previously considered and can throw me a bone here.

I'm trying to reach all 5m klines available in one stream. I've done it before directly with Binance API but I like the way you parse the response, so I made this:

.

Is there a better way to do this?

[tiagosiebler]

This seems like a way to do it. For the wskey, I would recommend to include all dynamic props...maybe something like symbols.sort().join().replace('USDT') isn't too long of a string? Reasoning is if you want to have multiple calls to this with diff symbols, you don't want any risk for wskeys to get mixed up. They should be unique, or when a reconnect happens you'll get the wrong websocket url being reconnected.

maybe streamName (in getWsKeyWithContext) would also be good as something suggestive like kline_multiple. Might need some testing to see how the reconnect logic behaves if your internet connection (and thus the stream) drops.

[MrAlvaroPS]

Cheers!

Done that and so far so good. I'll keep an eye on it and tell you if any errors comes up.

Thanks again for your work!!

[tiagosiebler]

Glad to hear! Would you mind sharing the snippet you're currently using? Are reconnects working OK? Would be great to add it to the lib.

[MrAlvaroPS]

Yeah, already tried the reconnects and it went all fine.

We were just talking in Telegram. I can send you anything else you need.

The snippet to connect all the cryptos at the same time:

`/**

• Get exchange info to retrieve assets
  */
  function getAllAssets() {
  futuresClient
  .getExchangeInfo()
  .then((x) => {
  openWebsockets(x);
  })
  .catch((e) => console.log(e));
  }`

`/**

• Connect all WS
  */
  function openWebsockets(exchangeInfoData) {
  exchangeInfoData.symbols.forEach((c, index) => {
  if (
  c.contractType === config.exchangeAssetInfo.contractType && // Looking for just perpetuals
  c.quoteAsset === config.exchangeAssetInfo.quoteAsset // Looking for just 'USDT' base asset
  ) {
  allWs[index] = wsClient.subscribeKlines(c.symbol, "3m", "usdm"); //I like to have it this way if I want to disconnect from all.
  }
  });
  }`