LgoStreamingMarketDataService implements StreamingMarketDataService { @Override public Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args) { return level2Subscriptions .computeIfAbsent(currencyPair, this::createLevel2Subscription) .getSubscription(); } LgoStreamingMarketDataService(LgoStreamingService lgoStreamingService); @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 it_reads_level2() throws IOException { LgoStreamingService streamingService = mock(LgoStreamingService.class); LgoStreamingMarketDataService service = new LgoStreamingMarketDataService(streamingService); JsonNode snapshot = TestUtils.getJsonContent("/marketdata/level2-snapshot.json"); JsonNode update = TestUtils.getJsonContent("/marketdata/level2-update.json"); when(streamingService.subscribeChannel(anyString())) .thenReturn(Observable.just(snapshot, update)); Observable<OrderBook> orderBook = service.getOrderBook(CurrencyPair.BTC_USD); verify(streamingService).subscribeChannel("level2-BTC-USD"); OrderBook firstBook = orderBook.blockingFirst(); assertThat(firstBook.getAsks()) .usingElementComparatorIgnoringFields("id", "userReference") .contains(sellOrder("1111.1000", "9.39370000"), sellOrder("1115.9000", "0.88420000")); assertThat(firstBook.getBids()) .usingElementComparatorIgnoringFields("id", "userReference") .contains(buyOrder("1089.1000", "0.10000000")); OrderBook lastBook = orderBook.blockingLast(); assertThat(lastBook.getBids()).isEmpty(); assertThat(lastBook.getAsks()) .usingElementComparatorIgnoringFields("id", "userReference") .contains( sellOrder("1111.1000", "0.10000000"), sellOrder("1115.9000", "0.88420000"), sellOrder("1116.9000", "0.20000000")); }