Combine streams question

[lz-44]

I started using binance-derivatives-trading-usds-futures module, and wondering if is there any possibility to combine streams there?
As a work around, I can see I can create WebsocketMarketStreamsApi by passing there a connection made from configuration manually, but it seems a bit dirty, so maybe there are some cleaner ways?
Thanks!

[theo-s68]

Hello @lz-44 ,

You can subscribe to multiple stream following this example:

private DerivativesTradingUsdsFuturesWebSocketStreams api;

    public DerivativesTradingUsdsFuturesWebSocketStreams getApi() {
        if (api == null) {
            WebSocketClientConfiguration clientConfiguration =
                    DerivativesTradingUsdsFuturesWebSocketStreamsUtil.getClientConfiguration();
            api = new DerivativesTradingUsdsFuturesWebSocketStreams(clientConfiguration);
        }
        return api;
    }
    
    public void streamsExample() throws ApiException, InterruptedException {
        KlineCandlestickStreamsRequest klineCandlestickStreamsRequest =
                new KlineCandlestickStreamsRequest();
        klineCandlestickStreamsRequest.symbol("btcusdt");
        klineCandlestickStreamsRequest.interval("1m");
        StreamBlockingQueueWrapper<KlineCandlestickStreamsResponse> kline =
                getApi().klineCandlestickStreams(klineCandlestickStreamsRequest);

        AggregateTradeStreamsRequest aggregateTradeStreamsRequest = new AggregateTradeStreamsRequest();
        aggregateTradeStreamsRequest.setSymbol("btcusdt");
        StreamBlockingQueueWrapper<AggregateTradeStreamsResponse> trade = getApi().aggregateTradeStreams(aggregateTradeStreamsRequest);

        while (true) {
            System.out.println(kline.take());
            System.out.println(trade.take());
        }
    }

Let us know if that answers the question, or could you give us an example of what you need to do ?

Thank you.

[lz-44]

Hi @theo-s68 ,
thank you for your answer.
Please, correct me if I am wrong in my following thoughts.
My goal is the following: let's say, I want to subscribe to 350 symbols for real-time orderbook, agg trade and other updates. I don't want to make a connection per stream, because it would be too many connections; and also I don't want let's say 350 orderbook real-time symbol streams in one connection, because most probably this connection and the underneath queues (or whatever) would be overloaded and not flexible in terms of failures handling, etc..

Before moving to the current modular connector I was using binance-futures-connector-java (which now is marked as deprecated). There I managed to subscribe for example to all orderbook / agg trade symbols by combining 50 symbols/streams per one connection. For streams combining I used the "UMWebsocketClientImpl.combineStreams" functionality which as I understood implemented the stream combination functionality mentioned in docs (https://developers.binance.com/docs/derivatives/usds-margined-futures/websocket-market-streams).

My question is how can I achieve the similar approach in the new java modular connector?

From my understanding, in your code example, you created 2 separate requests, which would result in two separate queues (so 350 queues for 350 symbols). And moreover, I am not sure how the connections are managed in that case.

Thanks in advance for clarification

[theo-s68]

Hello @lz-44 ,

Thanks for the clarification.

Indeed, the SDK doesn't support combined stream out of the box, but it does support connection pool for websocket:

    public DerivativesTradingUsdsFuturesWebSocketStreams getApi() {
        if (api == null) {
            WebSocketClientConfiguration clientConfiguration =
                    DerivativesTradingUsdsFuturesWebSocketStreamsUtil.getClientConfiguration();
            clientConfiguration.setUsePool(true); // use connection pool
            clientConfiguration.setPoolSize(5); // use a pool of 5 connections
            api = new DerivativesTradingUsdsFuturesWebSocketStreams(clientConfiguration);
        }
        return api;
    }

And the SDK will take care of distributing the subscriptions across the connections in the pool using round-robin.
So indeed you will get 350 queues for 350 symbols, but using only 5 connections in this example.

Please give it a try and let us know if you encounter any issues with having too many queues. We can then evaluate and find a better solution.

Thank you.

[lz-44]

@theo-s68
Round-robin....hm....
So what I am thinking of is how this new approach may impact the existing code for my case (and hence for many other migration cases).
So my current approach (that is using binance-futures-connector-java) is the following:
I want to subscribe to, say, 350 orderbook symbols.
I split 350 into 7 batches / 50 symbols each.
Each batch is then formed into 1 combined stream which afterwards downstream to Flux (I use Reactor) and corresponds to one connection. So, in other words, 350 symbols get split in 7 connection with 50 symbols in each.
This approach has few benefits, one of them is when some connection is closed or failure happened, it is going to be re-connected. So, 50 symbols of 350 is going to be reconnected, the rest 300 would continue working normally. Or, I often got experienced with silent close of connection. In that case I relied on timeout. If no signal in stream for 5 sec, then I reconnected this specific connection.

If I apply my existing code to the new connector, what I would have? - I would still have 50 symbols per batch, but each of this symbol in one batch would be spread across different connections (this is what round robin would take care of), right?
From my understanding that would cause this approach is not working as expected and would require the redesign.

Having said that, from my understanding, would be still nice to have either combining stream functionality, or more control on connection configuration, so that I control what streams are included in each connection.

[w-hpcai-afk]

Hello @lz-44 ,

You can subscribe to multiple stream following this example:

private DerivativesTradingUsdsFuturesWebSocketStreams api;

public DerivativesTradingUsdsFuturesWebSocketStreams getApi() {
    if (api == null) {
        WebSocketClientConfiguration clientConfiguration =
                DerivativesTradingUsdsFuturesWebSocketStreamsUtil.getClientConfiguration();
        api = new DerivativesTradingUsdsFuturesWebSocketStreams(clientConfiguration);
    }
    return api;
}

public void streamsExample() throws ApiException, InterruptedException {
    KlineCandlestickStreamsRequest klineCandlestickStreamsRequest =
            new KlineCandlestickStreamsRequest();
    klineCandlestickStreamsRequest.symbol("btcusdt");
    klineCandlestickStreamsRequest.interval("1m");
    StreamBlockingQueueWrapper<KlineCandlestickStreamsResponse> kline =
            getApi().klineCandlestickStreams(klineCandlestickStreamsRequest);

    AggregateTradeStreamsRequest aggregateTradeStreamsRequest = new AggregateTradeStreamsRequest();
    aggregateTradeStreamsRequest.setSymbol("btcusdt");
    StreamBlockingQueueWrapper<AggregateTradeStreamsResponse> trade = getApi().aggregateTradeStreams(aggregateTradeStreamsRequest);

    while (true) {
        System.out.println(kline.take());
        System.out.println(trade.take());
    }
}

Let us know if that answers the question, or could you give us an example of what you need to do ?

Thank you.

Hi @theo-s68,
I have a similar question. If my understanding is correct, for the last part, the kline and trade are two blocking queues, and trade has to wait untile kline.take() is finished? Let's say if the kline is blocked for 10 seconds without any data, then trade should also wait for 10 seconds? Unless we use one separate thread for one symbol/market data type pair. But for the combined stream approach, one symbol/market data type won' block the others.

Thank you!