HitbtcStreamingMarketDataService implements StreamingMarketDataService { @Override public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) { String pair = currencyPair.base.toString() + currencyPair.counter.toString(); String channelName = getChannelName("orderbook", pair); final ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper(); Observable<JsonNode> jsonNodeObservable = service.subscribeChannel(channelName); return jsonNodeObservable .map(s -> mapper.readValue(s.toString(), HitbtcWebSocketOrderBookTransaction.class)) .map( s -> { HitbtcWebSocketOrderBook hitbtcOrderBook = s.toHitbtcOrderBook(orderbooks.getOrDefault(currencyPair, null)); orderbooks.put(currencyPair, hitbtcOrderBook); return HitbtcAdapters.adaptOrderBook( hitbtcOrderBook.toHitbtcOrderBook(), currencyPair); }); } 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 testOrderbookCommon() throws Exception { String orderBook = IOUtils.toString( getClass().getResource("/example/notificationSnapshotOrderBook.json"), "UTF8"); when(streamingService.subscribeChannel(eq("orderbook-BTCEUR"))) .thenReturn(Observable.just(objectMapper.readTree(orderBook))); List<LimitOrder> bids = new ArrayList<>(); bids.add( new LimitOrder( Order.OrderType.BID, new BigDecimal("0.500"), CurrencyPair.BTC_EUR, null, null, new BigDecimal("0.054558"))); bids.add( new LimitOrder( Order.OrderType.BID, new BigDecimal("0.076"), CurrencyPair.BTC_EUR, null, null, new BigDecimal("0.054557"))); bids.add( new LimitOrder( Order.OrderType.BID, new BigDecimal("7.725"), CurrencyPair.BTC_EUR, null, null, new BigDecimal("0.054524"))); List<LimitOrder> asks = new ArrayList<>(); asks.add( new LimitOrder( Order.OrderType.ASK, new BigDecimal("0.245"), CurrencyPair.BTC_EUR, null, null, new BigDecimal("0.054588"))); asks.add( new LimitOrder( Order.OrderType.ASK, new BigDecimal("2.784"), CurrencyPair.BTC_EUR, null, null, new BigDecimal("0.054591"))); TestObserver<OrderBook> test = marketDataService.getOrderBook(CurrencyPair.BTC_EUR).test(); test.assertValue( orderBook1 -> { assertThat(orderBook1.getAsks()).as("Asks").isEqualTo(asks); assertThat(orderBook1.getBids()).as("Bids").isEqualTo(bids); return true; }); }