Streaming best practices

[m1dnightc0der]

Hi, I have just started to use the streaming api, and hitting a few issues, wondered if there were any best practices.

I am subscribing to multiple prices, if I do this with one thread instantiating the exchange and subscribing to multiple pairs, the books all get confused and error. If I do this with multiple threads, instantiate the exchange as a singleton then use it across all threads, it kinda works but it drops book from time to time and has to automatically reset, quite frequently (ever few hours or so).

Say I have 12 pairs I want to subscribe to, I create 12 threads (one per pair, that use the same exchange).

So I create the StreamingExchange once, and add it to a map

info.bitrich.xchangestream.core.StreamingExchange exchange = StreamingExchangeFactory.INSTANCE.createExchange(spec);
  streamingExchangesByMarket.put(coinTraderMarket, exchange);

Then once the StreamingExchange instance is created, I create 12 threads, one per pair the singleton StreamingExchange instance out of the map.

            XchangeUtil.getStreamingExchangeForMarket(market).getStreamingMarketDataService().getTrades((CurrencyPair) pair, params).retryWhen(
                    throwableObservable -> throwableObservable.zipWith(Observable.range(1, 3), (throwable, integer) -> integer)
                        .flatMap(integer -> Observable.timer(integer, TimeUnit.SECONDS)))

                .subscribe(
                    trade -> handleTrade(trade, lastTradeTimes, lastTradeIds, logLags, coinTraderExchange, market, tradeFactory, failedTradeCounts,
                        context, true));

            XchangeUtil.getStreamingExchangeForMarket(market).getStreamingMarketDataService().getOrderBook((CurrencyPair) pair, params)
                .subscribe(orderBook -> handleOrderBook(orderBook, coinTraderExchange, market, bookFactory, failedBookCounts, context));