LgoStreamingAccountService implements StreamingAccountService { @Override public Observable<Balance> getBalanceChanges(Currency currency, Object... args) { ensureSubscription(); return subscription.map(u -> u.getWallet().get(currency)); } LgoStreamingAccountService(LgoStreamingService lgoStreamingService); @Override Observable<Balance> getBalanceChanges(Currency currency, Object... args); Observable<Wallet> getWallet();  }
@Test public void it_connects_one_time_and_filters_by_currency() throws IOException { JsonNode snapshot = TestUtils.getJsonContent("/account/balance-snapshot.json"); JsonNode update = TestUtils.getJsonContent("/account/balance-update.json"); LgoStreamingService streamingService = mock(LgoStreamingService.class); LgoStreamingAccountService service = new LgoStreamingAccountService(streamingService); when(streamingService.subscribeChannel(anyString())) .thenReturn(Observable.just(snapshot, update)); TestObserver<Balance> btcChanges = service.getBalanceChanges(Currency.BTC).test(); TestObserver<Balance> usdChanges = service.getBalanceChanges(Currency.USD).test(); verify(streamingService, times(1)).subscribeChannel("balance"); btcChanges.assertValueAt( 1, new Balance( Currency.BTC, new BigDecimal("2299.01329566"), new BigDecimal("2295.01329566"), new BigDecimal("4.00000000"))); usdChanges.assertValueAt( 1, new Balance( Currency.USD, new BigDecimal("453616.3125"), new BigDecimal("453616.3125"), new BigDecimal("0.0000"))); }