JsonRpc2_0Rx { public Observable<PlatonBlock> catchUpToLatestAndSubscribeToNewBlocksObservable( DefaultBlockParameter startBlock, boolean fullTransactionObjects, long pollingInterval) { return catchUpToLatestBlockObservable( startBlock, fullTransactionObjects, blockObservable(fullTransactionObjects, pollingInterval)); } JsonRpc2_0Rx(Web3j web3j, ScheduledExecutorService scheduledExecutorService); Observable<String> ethBlockHashObservable(final long pollingInterval); Observable<String> ethPendingTransactionHashObservable(final long pollingInterval); Observable<Log> ethLogObservable(
            final PlatonFilter ethFilter,
            final long pollingInterval); Observable<Transaction> transactionObservable(final long pollingInterval); Observable<Transaction> pendingTransactionObservable(final long pollingInterval); Observable<PlatonBlock> blockObservable(
            final boolean fullTransactionObjects, final long pollingInterval); Observable<PlatonBlock> replayBlocksObservable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock,
            boolean fullTransactionObjects); Observable<PlatonBlock> replayBlocksObservable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock,
            boolean fullTransactionObjects, boolean ascending); Observable<Transaction> replayTransactionsObservable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock); Observable<PlatonBlock> catchUpToLatestBlockObservable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects,
            Observable<PlatonBlock> onCompleteObservable); Observable<PlatonBlock> catchUpToLatestBlockObservable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects); Observable<Transaction> catchUpToLatestTransactionObservable(
            DefaultBlockParameter startBlock); Observable<PlatonBlock> catchUpToLatestAndSubscribeToNewBlocksObservable(
            DefaultBlockParameter startBlock, boolean fullTransactionObjects,
            long pollingInterval); Observable<Transaction> catchUpToLatestAndSubscribeToNewTransactionsObservable(
            DefaultBlockParameter startBlock, long pollingInterval);  }
@Test public void testCatchUpToLatestAndSubscribeToNewBlockObservable() throws Exception { List<PlatonBlock> expected = Arrays.asList( createBlock(0), createBlock(1), createBlock(2), createBlock(3), createBlock(4), createBlock(5), createBlock(6)); List<PlatonBlock> ethBlocks = 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<PlatonBlock> stubbing = when(web3jService.send(any(Request.class), eq(PlatonBlock.class))); for (PlatonBlock ethBlock : ethBlocks) { stubbing = stubbing.thenReturn(ethBlock); } PlatonFilter ethFilter = objectMapper.readValue( "{\n" + " \"id\":1,\n" + " \"jsonrpc\": \"2.0\",\n" + " \"result\": \"0x1\"\n" + "}", PlatonFilter.class); PlatonLog ethLog = objectMapper.readValue( "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":[" + "\"0x31c2342b1e0b8ffda1507fbffddf213c4b3c1e819ff6a84b943faabb0ebf2403\"" + "]}", PlatonLog.class); PlatonUninstallFilter ethUninstallFilter = objectMapper.readValue( "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":true}", PlatonUninstallFilter.class); when(web3jService.send(any(Request.class), eq(PlatonFilter.class))) .thenReturn(ethFilter); when(web3jService.send(any(Request.class), eq(PlatonLog.class))) .thenReturn(ethLog); when(web3jService.send(any(Request.class), eq(PlatonUninstallFilter.class))) .thenReturn(ethUninstallFilter); Observable<PlatonBlock> observable = web3j.catchUpToLatestAndSubscribeToNewBlocksObservable( new DefaultBlockParameterNumber(BigInteger.ZERO), false); final CountDownLatch transactionLatch = new CountDownLatch(expected.size()); final CountDownLatch completedLatch = new CountDownLatch(1); final List<PlatonBlock> results = new ArrayList<PlatonBlock>(expected.size()); Subscription subscription = observable.subscribe( new Action1<PlatonBlock>() { @Override public void call(PlatonBlock result) { results.add(result); transactionLatch.countDown(); } }, new Action1<Throwable>() { @Override public void call(Throwable throwable) { fail(throwable.getMessage()); } }, new Action0() { @Override public void call() { completedLatch.countDown(); } }); transactionLatch.await(1250, TimeUnit.MILLISECONDS); assertThat(results, equalTo(expected)); subscription.unsubscribe(); completedLatch.await(1, TimeUnit.SECONDS); assertTrue(subscription.isUnsubscribed()); }