@Test public void it_handles_open_orders_updates() throws IOException, ParseException { JsonNode snapshot = TestUtils.getJsonContent("/trade/orders-snapshot.json"); JsonNode update = TestUtils.getJsonContent("/trade/orders-update.json"); when(streamingService.subscribeChannel(anyString())) .thenReturn(Observable.just(snapshot, update)); Observable<OpenOrders> openOrders = service.getOpenOrders(CurrencyPair.BTC_USD); verify(streamingService).subscribeChannel("user-BTC-USD"); Date date1 = dateFormat.parse("2019-07-23T15:36:14.304Z"); Date date2 = dateFormat.parse("2019-07-18T12:24:21.891Z"); Date date3 = dateFormat.parse("2019-07-24T08:37:19.116Z"); LimitOrder order1 = new LimitOrder( Order.OrderType.ASK, BigDecimal.ONE, BigDecimal.ZERO, CurrencyPair.BTC_USD, "156389617430400001", date1, new BigDecimal(6000)); order1.setOrderStatus(Order.OrderStatus.NEW); LimitOrder order2 = new LimitOrder( Order.OrderType.ASK, BigDecimal.ONE, BigDecimal.ZERO, CurrencyPair.BTC_USD, "156345266189100001", date2, new BigDecimal(6000)); order2.setOrderStatus(Order.OrderStatus.NEW); LimitOrder order3 = new LimitOrder( Order.OrderType.BID, new BigDecimal(2), BigDecimal.ZERO, CurrencyPair.BTC_USD, "156395743911600001", date3, new BigDecimal(8000)); order3.setOrderStatus(Order.OrderStatus.NEW); assertThat(openOrders.blockingFirst()) .usingRecursiveComparison() .isEqualTo(new OpenOrders(Arrays.asList(order2, order1))); assertThat(openOrders.blockingLast()) .usingRecursiveComparison() .isEqualTo(new OpenOrders(Arrays.asList(order3, order1))); }
public Observable<OpenOrders> getOpenOrders(CurrencyPair currencyPair) { return getOrderUpdates(currencyPair) .map( u -> u.getAllOpenOrders().values().stream() .filter(order -> order instanceof LimitOrder) .map(order -> (LimitOrder) order) .collect(Collectors.toList())) .map(OpenOrders::new); }
LgoStreamingTradeService implements StreamingTradeService { public Observable<OpenOrders> getOpenOrders(CurrencyPair currencyPair) { return getOrderUpdates(currencyPair) .map( u -> u.getAllOpenOrders().values().stream() .filter(order -> order instanceof LimitOrder) .map(order -> (LimitOrder) order) .collect(Collectors.toList())) .map(OpenOrders::new); } }
LgoStreamingTradeService implements StreamingTradeService { public Observable<OpenOrders> getOpenOrders(CurrencyPair currencyPair) { return getOrderUpdates(currencyPair) .map( u -> u.getAllOpenOrders().values().stream() .filter(order -> order instanceof LimitOrder) .map(order -> (LimitOrder) order) .collect(Collectors.toList())) .map(OpenOrders::new); } LgoStreamingTradeService(
      LgoStreamingService streamingService,
      LgoKeyService keyService,
      LgoSignatureService signatureService,
      SynchronizedValueFactory<Long> nonceFactory); }
LgoStreamingTradeService implements StreamingTradeService { public Observable<OpenOrders> getOpenOrders(CurrencyPair currencyPair) { return getOrderUpdates(currencyPair) .map( u -> u.getAllOpenOrders().values().stream() .filter(order -> order instanceof LimitOrder) .map(order -> (LimitOrder) order) .collect(Collectors.toList())) .map(OpenOrders::new); } 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); }
LgoStreamingTradeService implements StreamingTradeService { public Observable<OpenOrders> getOpenOrders(CurrencyPair currencyPair) { return getOrderUpdates(currencyPair) .map( u -> u.getAllOpenOrders().values().stream() .filter(order -> order instanceof LimitOrder) .map(order -> (LimitOrder) order) .collect(Collectors.toList())) .map(OpenOrders::new); } 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);  }