BitstampStreamingMarketDataService implements StreamingMarketDataService { @Override public Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args) { String channelName = "live_trades" + getChannelPostfix(currencyPair); return service .subscribeChannel(channelName, BitstampStreamingService.EVENT_TRADE) .map( s -> { ObjectMapper mapper = StreamingObjectMapperHelper.getObjectMapper(); BitstampWebSocketTransaction transactions = mapper.readValue(s, BitstampWebSocketTransaction.class); return BitstampAdapters.adaptTrade(transactions, currencyPair, 1000); }); } BitstampStreamingMarketDataService(PusherStreamingService service); Observable<OrderBook> getDifferentialOrderBook(CurrencyPair currencyPair, Object... args); @Override Observable<OrderBook> getOrderBook(CurrencyPair currencyPair, Object... args); @Override Observable<Ticker> getTicker(CurrencyPair currencyPair, Object... args); @Override Observable<Trade> getTrades(CurrencyPair currencyPair, Object... args);  }
@Test public void testGetTrades() throws Exception { String trade = new String( Files.readAllBytes(Paths.get(ClassLoader.getSystemResource("trade.json").toURI()))); when(streamingService.subscribeChannel(eq("live_trades"), eq("trade"))) .thenReturn(Observable.just(trade)); Trade expected = new Trade.Builder() .type(Order.OrderType.ASK) .originalAmount(new BigDecimal("34.390000000000001")) .currencyPair(CurrencyPair.BTC_USD) .price(new BigDecimal("914.38999999999999")) .timestamp(new Date(1484858423000L)) .id("177827396") .build(); TestObserver<Trade> test = marketDataService.getTrades(CurrencyPair.BTC_USD).test(); validateTrades(expected, test); }