LgoStreamingTradeService implements StreamingTradeService { public Observable<LgoOrderEvent> getRawBatchOrderEvents(CurrencyPair currencyPair) { return getOrderUpdates(currencyPair) .map(LgoGroupedUserUpdate::getEvents) .flatMap(Observable::fromIterable); } 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_orders_events() throws IOException, ParseException { JsonNode snapshot = TestUtils.getJsonContent("/trade/order-events-snapshot.json"); JsonNode update = TestUtils.getJsonContent("/trade/order-events-update.json"); when(streamingService.subscribeChannel(anyString())) .thenReturn(Observable.just(snapshot, update)); Observable<LgoOrderEvent> events = service.getRawBatchOrderEvents(CurrencyPair.BTC_USD); verify(streamingService).subscribeChannel("user-BTC-USD"); Date date1 = dateFormat.parse("2019-07-24T08:37:19.116Z"); Date date2 = dateFormat.parse("2019-07-24T08:37:19.849Z"); Date date3 = dateFormat.parse("2019-07-24T08:37:19.922Z"); ArrayList<LgoOrderEvent> lgoOrderEvents = Lists.newArrayList(events.blockingIterable()); assertThat(lgoOrderEvents.get(0)) .isEqualToComparingFieldByField( new LgoPendingOrderEvent( 6317543L, "pending", "156395743911600001", date1, "L", new BigDecimal("8000.0000"), Order.OrderType.BID, new BigDecimal("2.00000000"))); assertThat(lgoOrderEvents.get(1)) .isEqualToComparingFieldByField( new LgoOpenOrderEvent(6317543L, "open", "156395743911600001", date2)); assertThat(lgoOrderEvents.get(2)) .isEqualToComparingFieldByField( new LgoPendingOrderEvent( 6317543L, "pending", "156395743912700001", date2, "L", new BigDecimal("8000.0001"), Order.OrderType.BID, new BigDecimal("2.00000000"))); assertThat(lgoOrderEvents.get(3)) .isEqualToComparingFieldByField( new LgoInvalidOrderEvent( 6317543L, "invalid", "156395743912700001", date2, "InvalidAmount")); assertThat(lgoOrderEvents.get(4)) .isEqualToComparingFieldByField( new LgoDoneOrderEvent(6317543L, "done", "156345266189100001", date3, "canceled", null)); }