HitbtcStreamingMarketDataService implements StreamingMarketDataService { @Override public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) { String pair = currencyPair.base.toString() + currencyPair.counter.toString(); String channelName = getChannelName("trades", pair); final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper(); return service .subscribeChannel(channelName) .map(s -> mapper.readValue(s.toString(), HitbtcWebSocketTradesTransaction.class)) .map(HitbtcWebSocketTradesTransaction::getParams) .filter(Objects::nonNull) .map(HitbtcWebSocketTradeParams::getData) .filter(Objects::nonNull) .map(Arrays::asList) .flatMapIterable( s -> { Trades adaptedTrades = HitbtcAdapters.adaptTrades(s, currencyPair); return adaptedTrades.getTrades(); }); } HitbtcStreamingMarketDataService(HitbtcStreamingService service); @Override Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args); @Override Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args); @Override Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args);  }
@Test public void testGetTrades() throws Exception { String trades = IOUtils.toString( getClass().getResource("/example/notificationSnapshotTrades.json"), "UTF8"); when(streamingService.subscribeChannel(eq("trades-BTCUSD"))) .thenReturn(Observable.just(objectMapper.readTree(trades))); Trade expected1 = new Trade.Builder() .type(Order.OrderType.BID) .originalAmount(new BigDecimal("0.057")) .currencyPair(CurrencyPair.BTC_USD) .price(new BigDecimal("0.054656")) .timestamp(new Date(1508430822821L)) .id("54469456") .build(); Trade expected2 = new Trade.Builder() .type(Order.OrderType.BID) .originalAmount(new BigDecimal("0.092")) .currencyPair(CurrencyPair.BTC_USD) .price(new BigDecimal("0.054656")) .timestamp(new Date(1508430828754L)) .id("54469497") .build(); Trade expected3 = new Trade.Builder() .type(Order.OrderType.BID) .originalAmount(new BigDecimal("0.002")) .currencyPair(CurrencyPair.BTC_USD) .price(new BigDecimal("0.054669")) .timestamp(new Date(1508430853288L)) .id("54469697") .build(); TestObserver<Trade> test = marketDataService.getTrades(CurrencyPair.BTC_USD).test(); test.assertValues(expected1, expected2, expected3); validateTrade(0, test, expected1); validateTrade(1, test, expected2); validateTrade(2, test, expected3); }