@Test public void testCatchUpToLatestAndSubscribeToNewBlockFlowable() throws Exception { List<AppBlock> expected = Arrays.asList( createBlock(0), createBlock(1), createBlock(2), createBlock(3), createBlock(4), createBlock(5), createBlock(6)); List<AppBlock> appBlocks = Arrays.asList( expected.get(2), expected.get(0), expected.get(1), expected.get(2), expected.get(4), expected.get(3), expected.get(4), expected.get(4), expected.get(5), expected.get(6)); OngoingStubbing<AppBlock> stubbing = when(CITAjService.send(any(Request.class), eq(AppBlock.class))); for (AppBlock appBlock : appBlocks) { stubbing = stubbing.thenReturn(appBlock); } AppFilter appFilter = objectMapper.readValue( "{\n" + " \"id\":1,\n" + " \"jsonrpc\": \"2.0\",\n" + " \"result\": \"0x1\"\n" + "}", AppFilter.class); AppLog appLog = objectMapper.readValue( "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":[" + "\"0x31c2342b1e0b8ffda1507fbffddf213c4b3c1e819ff6a84b943faabb0ebf2403\"" + "]}", AppLog.class); AppUninstallFilter appUninstallFilter = objectMapper.readValue( "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":true}", AppUninstallFilter.class); when(CITAjService.send(any(Request.class), eq(AppFilter.class))) .thenReturn(appFilter); when(CITAjService.send(any(Request.class), eq(AppLog.class))) .thenReturn(appLog); when(CITAjService.send(any(Request.class), eq(AppUninstallFilter.class))) .thenReturn(appUninstallFilter); Flowable<AppBlock> flowable = citaj .catchUpToLatestAndSubscribeToNewBlocksFlowable( new DefaultBlockParameterNumber(BigInteger.ZERO), false); CountDownLatch transactionLatch = new CountDownLatch(expected.size()); CountDownLatch completedLatch = new CountDownLatch(1); List<AppBlock> results = new ArrayList<>(expected.size()); Disposable subscription = flowable.subscribe( result -> { results.add(result); transactionLatch.countDown(); }, throwable -> fail(throwable.getMessage()), () -> completedLatch.countDown()); transactionLatch.await(1250, TimeUnit.MILLISECONDS); assertThat(results, equalTo(expected)); subscription.dispose(); completedLatch.await(1, TimeUnit.SECONDS); assertTrue(subscription.isDisposed()); }
public Flowable<AppBlock> catchUpToLatestAndSubscribeToNewBlocksFlowable( DefaultBlockParameter startBlock, boolean fullTransactionObjects, long pollingInterval) { return catchUpToLatestBlockFlowable( startBlock, fullTransactionObjects, blockFlowable(fullTransactionObjects, pollingInterval)); }
JsonRpc2_0Rx { public Flowable<AppBlock> catchUpToLatestAndSubscribeToNewBlocksFlowable( DefaultBlockParameter startBlock, boolean fullTransactionObjects, long pollingInterval) { return catchUpToLatestBlockFlowable( startBlock, fullTransactionObjects, blockFlowable(fullTransactionObjects, pollingInterval)); } }
JsonRpc2_0Rx { public Flowable<AppBlock> catchUpToLatestAndSubscribeToNewBlocksFlowable( DefaultBlockParameter startBlock, boolean fullTransactionObjects, long pollingInterval) { return catchUpToLatestBlockFlowable( startBlock, fullTransactionObjects, blockFlowable(fullTransactionObjects, pollingInterval)); } JsonRpc2_0Rx(CITAj citaj, ScheduledExecutorService scheduledExecutorService); }
JsonRpc2_0Rx { public Flowable<AppBlock> catchUpToLatestAndSubscribeToNewBlocksFlowable( DefaultBlockParameter startBlock, boolean fullTransactionObjects, long pollingInterval) { return catchUpToLatestBlockFlowable( startBlock, fullTransactionObjects, blockFlowable(fullTransactionObjects, pollingInterval)); } JsonRpc2_0Rx(CITAj citaj, ScheduledExecutorService scheduledExecutorService); Flowable<String> appBlockHashFlowable(long pollingInterval); Flowable<String> appPendingTransactionHashFlowable(long pollingInterval); Flowable<Log> appLogFlowable(AppFilter appFilter, long pollingInterval); Flowable<Transaction> transactionFlowable(long pollingInterval); Flowable<Transaction> pendingTransactionFlowable(long pollingInterval); Flowable<AppBlock> blockFlowable(
            boolean fullTransactionObjects, long pollingInterval); Flowable<AppBlock> replayBlocksFlowable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock,
            boolean fullTransactionObjects); Flowable<AppBlock> replayBlocksFlowable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock,
            boolean fullTransactionObjects, boolean ascending); Flowable<Transaction> replayTransactionsFlowable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock); Flowable<AppBlock> catchUpToLatestBlockFlowable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects,
            Flowable<AppBlock> onCompleteFlowable); Flowable<AppBlock> catchUpToLatestBlockFlowable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects); Flowable<Transaction> catchUpToLatestTransactionFlowable(
            DefaultBlockParameter startBlock); Flowable<AppBlock> catchUpToLatestAndSubscribeToNewBlocksFlowable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects,
            long pollingInterval); Flowable<Transaction> catchUpToLatestAndSubscribeToNewTransactionsFlowable(
            DefaultBlockParameter startBlock, long pollingInterval); }
JsonRpc2_0Rx { public Flowable<AppBlock> catchUpToLatestAndSubscribeToNewBlocksFlowable( DefaultBlockParameter startBlock, boolean fullTransactionObjects, long pollingInterval) { return catchUpToLatestBlockFlowable( startBlock, fullTransactionObjects, blockFlowable(fullTransactionObjects, pollingInterval)); } JsonRpc2_0Rx(CITAj citaj, ScheduledExecutorService scheduledExecutorService); Flowable<String> appBlockHashFlowable(long pollingInterval); Flowable<String> appPendingTransactionHashFlowable(long pollingInterval); Flowable<Log> appLogFlowable(AppFilter appFilter, long pollingInterval); Flowable<Transaction> transactionFlowable(long pollingInterval); Flowable<Transaction> pendingTransactionFlowable(long pollingInterval); Flowable<AppBlock> blockFlowable(
            boolean fullTransactionObjects, long pollingInterval); Flowable<AppBlock> replayBlocksFlowable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock,
            boolean fullTransactionObjects); Flowable<AppBlock> replayBlocksFlowable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock,
            boolean fullTransactionObjects, boolean ascending); Flowable<Transaction> replayTransactionsFlowable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock); Flowable<AppBlock> catchUpToLatestBlockFlowable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects,
            Flowable<AppBlock> onCompleteFlowable); Flowable<AppBlock> catchUpToLatestBlockFlowable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects); Flowable<Transaction> catchUpToLatestTransactionFlowable(
            DefaultBlockParameter startBlock); Flowable<AppBlock> catchUpToLatestAndSubscribeToNewBlocksFlowable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects,
            long pollingInterval); Flowable<Transaction> catchUpToLatestAndSubscribeToNewTransactionsFlowable(
            DefaultBlockParameter startBlock, long pollingInterval);  }