LgoStreamingTradeService implements StreamingTradeService { public Observable<LgoOrderEvent> getRawAllOrderEvents(Collection<CurrencyPair> currencyPairs) { Observable<LgoOrderEvent> ackObservable = getRawReceivedOrderEvents(); return currencyPairs.stream() .map(this::getRawBatchOrderEvents) .reduce(Observable::mergeWith) .map(ackObservable::mergeWith) .orElse(ackObservable); } 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_all_order_events() throws IOException, ParseException { JsonNode snapshotOrders = TestUtils.getJsonContent("/trade/all-orders-snapshot.json"); JsonNode snapshotAFR = TestUtils.getJsonContent("/trade/all-afr-snapshot.json"); JsonNode update1 = TestUtils.getJsonContent("/trade/all-afr-update.json"); JsonNode update2 = TestUtils.getJsonContent("/trade/all-orders-update.json"); when(streamingService.subscribeChannel("afr")) .thenReturn(Observable.just(snapshotAFR, update1)); when(streamingService.subscribeChannel("user-BTC-USD")) .thenReturn(Observable.just(snapshotOrders, update2)); Observable<LgoOrderEvent> events = service.getRawAllOrderEvents(Collections.singletonList(CurrencyPair.BTC_USD)); verify(streamingService).subscribeChannel("afr"); verify(streamingService).subscribeChannel("user-BTC-USD"); Date date1 = dateFormat.parse("2019-07-25T07:16:21.600Z"); Date date2 = dateFormat.parse("2019-07-25T07:16:22.959Z"); ArrayList<LgoOrderEvent> lgoOrderEvents = Lists.newArrayList(events.blockingIterable()); assertThat(lgoOrderEvents.get(0)) .isEqualToComparingFieldByField( new LgoReceivedOrderEvent("156403898160000001", "0", "received", date1)); assertThat(lgoOrderEvents.get(1)) .isEqualToComparingFieldByField( new LgoPendingOrderEvent( 6393996L, "pending", "156403898160000001", date1, "L", new BigDecimal("7000.0000"), Order.OrderType.ASK, new BigDecimal("3.00000000"))); assertThat(lgoOrderEvents.get(2)) .isEqualToComparingFieldByField( new LgoOpenOrderEvent(6393996L, "open", "156403898160000001", date2)); }