LgoStreamingMarketDataService implements StreamingMarketDataService { @Override public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) { return tradeSubscriptions .computeIfAbsent(currencyPair, this::createTradeSubscription) .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_trades() throws IOException, ParseException { LgoStreamingService streamingService = mock(LgoStreamingService.class); LgoStreamingMarketDataService service = new LgoStreamingMarketDataService(streamingService); JsonNode message = TestUtils.getJsonContent("/marketdata/trades-update.json"); when(streamingService.subscribeChannel(anyString())).thenReturn(Observable.just(message)); Observable<Trade> trades = service.getTrades(CurrencyPair.BTC_USD); verify(streamingService).subscribeChannel("trades-BTC-USD"); assertThat(trades.blockingFirst()) .isEqualToComparingFieldByField( new Trade.Builder() .type(Order.OrderType.ASK) .originalAmount(new BigDecimal("4.36920000")) .currencyPair(CurrencyPair.BTC_USD) .price(new BigDecimal("428.5000")) .timestamp(dateFormat.parse("2019-07-19T12:25:01.596Z")) .id("3128770") .build()); assertThat(trades.blockingLast()) .isEqualToComparingFieldByField( new Trade.Builder() .type(Order.OrderType.BID) .originalAmount(new BigDecimal("1.85390000")) .currencyPair(CurrencyPair.BTC_USD) .price(new BigDecimal("420.3000")) .timestamp(dateFormat.parse("2019-07-19T12:25:05.860Z")) .id("3128771") .build()); }