LgoStreamingTradeService implements StreamingTradeService { @Override public Observable<UserTrade> getUserTrades(CurrencyPair currencyPair, Object... args) { return getRawBatchOrderEvents(currencyPair) .filter(lgoOrderEvent -> "match".equals(lgoOrderEvent.getType())) .map( matchEvent -> LgoAdapter.adaptUserTrade(currencyPair, (LgoMatchOrderEvent) matchEvent)); } LgoStreamingTradeService(
      LgoStreamingService streamingService,
      LgoKeyService keyService,
      LgoSignatureService signatureService,
      SynchronizedValueFactory<Long> nonceFactory); @Override Observable<Order> getOrderChanges(CurrencyPair currencyPair, Object... args); Observable<OpenOrders> getOpenOrders(CurrencyPair currencyPair); Observable<Collection<Order>> getOrderBatchChanges(CurrencyPair currencyPair); @Override Observable<UserTrade> getUserTrades(CurrencyPair currencyPair, Object... args); Observable<LgoOrderEvent> getRawAllOrderEvents(Collection<CurrencyPair> currencyPairs); Observable<LgoOrderEvent> getRawReceivedOrderEvents(); Observable<LgoOrderEvent> getRawBatchOrderEvents(CurrencyPair currencyPair); String placeMarketOrder(MarketOrder marketOrder); String placeLimitOrder(LimitOrder limitOrder); boolean cancelOrder(String orderId);  }
@Test public void it_handles_trades() throws IOException, ParseException { JsonNode snapshot = TestUtils.getJsonContent("/trade/user-trades-snapshot.json"); JsonNode update = TestUtils.getJsonContent("/trade/user-trades-update.json"); when(streamingService.subscribeChannel(anyString())) .thenReturn(Observable.just(snapshot, update)); Observable<UserTrade> userTrades = service.getUserTrades(CurrencyPair.BTC_USD); verify(streamingService).subscribeChannel("user-BTC-USD"); Date date = dateFormat.parse("2019-08-06T10:00:05.658Z"); ArrayList<UserTrade> trades = Lists.newArrayList(userTrades.blockingIterable()); assertThat(trades).hasSize(1); assertThat(trades.get(0)) .usingRecursiveComparison() .isEqualTo( new UserTrade.Builder() .type(Order.OrderType.ASK) .originalAmount(new BigDecimal("0.50000000")) .currencyPair(CurrencyPair.BTC_USD) .price(new BigDecimal("955.3000")) .timestamp(date) .id("4441691") .orderId("156508560418400001") .feeAmount(new BigDecimal("0.2388")) .feeCurrency(Currency.USD) .build()); }