OkCoinStreamingMarketDataService implements StreamingMarketDataService { @Override public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) { String channel = String.format( "ok_sub_spot_%s_%s_depth", currencyPair.base.toString().toLowerCase(), currencyPair.counter.toString().toLowerCase()); if (args.length > 0) { if (args[0] instanceof FuturesContract) { FuturesContract contract = (FuturesContract) args[0]; channel = String.format( "ok_sub_future%s_%s_depth_%s", currencyPair.counter.toString().toLowerCase(), currencyPair.base.toString().toLowerCase(), contract.getName()); if (args.length > 1) { channel = channel + "_" + args[1]; } } else { channel = channel + "_" + args[1]; } } final String key = channel; return service .subscribeChannel(channel) .map( s -> { OkCoinOrderbook okCoinOrderbook; if (!orderbooks.containsKey(key)) { OkCoinDepth okCoinDepth = mapper.treeToValue(s.get("data"), OkCoinDepth.class); okCoinOrderbook = new OkCoinOrderbook(okCoinDepth); orderbooks.put(key, okCoinOrderbook); } else { okCoinOrderbook = orderbooks.get(key); if (s.get("data").has("asks")) { if (s.get("data").get("asks").size() > 0) { BigDecimal[][] askLevels = mapper.treeToValue(s.get("data").get("asks"), BigDecimal[][].class); okCoinOrderbook.updateLevels(askLevels, Order.OrderType.ASK); } } if (s.get("data").has("bids")) { if (s.get("data").get("bids").size() > 0) { BigDecimal[][] bidLevels = mapper.treeToValue(s.get("data").get("bids"), BigDecimal[][].class); okCoinOrderbook.updateLevels(bidLevels, Order.OrderType.BID); } } } return OkCoinAdapters.adaptOrderBook( okCoinOrderbook.toOkCoinDepth(s.get("data").get("timestamp").asLong()), currencyPair); }); } OkCoinStreamingMarketDataService(OkCoinStreamingService service); @Override Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args); @Override Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args); Observable<FutureTicker> getFutureTicker(
      CurrencyPair currencyPair, FuturesContract contract); @Override Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args);  }
@Test public void testGetOrderBook() throws Exception { ObjectMapper objectMapper = new ObjectMapper(); JsonNode jsonNode = objectMapper.readTree( ClassLoader.getSystemClassLoader().getResourceAsStream("order-book.json")); when(okCoinStreamingService.subscribeChannel(any())).thenReturn(Observable.just(jsonNode)); Date timestamp = new Date(1484602135246L); List<LimitOrder> bids = new ArrayList<>(); bids.add( new LimitOrder( Order.OrderType.BID, new BigDecimal("0.922"), CurrencyPair.BTC_USD, null, timestamp, new BigDecimal("819.9"))); bids.add( new LimitOrder( Order.OrderType.BID, new BigDecimal("0.085"), CurrencyPair.BTC_USD, null, timestamp, new BigDecimal("818.63"))); List<LimitOrder> asks = new ArrayList<>(); asks.add( new LimitOrder( Order.OrderType.ASK, new BigDecimal("0.035"), CurrencyPair.BTC_USD, null, timestamp, new BigDecimal("821.6"))); asks.add( new LimitOrder( Order.OrderType.ASK, new BigDecimal("5.18"), CurrencyPair.BTC_USD, null, timestamp, new BigDecimal("821.65"))); asks.add( new LimitOrder( Order.OrderType.ASK, new BigDecimal("2.89"), CurrencyPair.BTC_USD, null, timestamp, new BigDecimal("821.7"))); OrderBook expected = new OrderBook(timestamp, asks, bids); TestObserver<OrderBook> test = marketDataService.getOrderBook(CurrencyPair.BTC_USD).test(); test.assertResult(expected); }