JsonRpc2_0Rx { public Observable<PlatonBlock> replayBlocksObservable( DefaultBlockParameter startBlock, DefaultBlockParameter endBlock, boolean fullTransactionObjects) { return replayBlocksObservable(startBlock, endBlock, fullTransactionObjects, true); } 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 testReplayBlocksObservable() throws Exception { List<PlatonBlock> ethBlocks = Arrays.asList(createBlock(0), createBlock(1), createBlock(2)); OngoingStubbing<PlatonBlock> stubbing = when(web3jService.send(any(Request.class), eq(PlatonBlock.class))); for (PlatonBlock ethBlock : ethBlocks) { stubbing = stubbing.thenReturn(ethBlock); } Observable<PlatonBlock> observable = web3j.replayBlocksObservable( new DefaultBlockParameterNumber(BigInteger.ZERO), new DefaultBlockParameterNumber(BigInteger.valueOf(2)), false); final CountDownLatch transactionLatch = new CountDownLatch(ethBlocks.size()); final CountDownLatch completedLatch = new CountDownLatch(1); final List<PlatonBlock> results = new ArrayList<PlatonBlock>(ethBlocks.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(1, TimeUnit.SECONDS); assertThat(results, equalTo(ethBlocks)); subscription.unsubscribe(); completedLatch.await(1, TimeUnit.SECONDS); assertTrue(subscription.isUnsubscribed()); }
@Test public void testReplayBlocksDescendingObservable() throws Exception { List<PlatonBlock> ethBlocks = Arrays.asList(createBlock(2), createBlock(1), createBlock(0)); OngoingStubbing<PlatonBlock> stubbing = when(web3jService.send(any(Request.class), eq(PlatonBlock.class))); for (PlatonBlock ethBlock : ethBlocks) { stubbing = stubbing.thenReturn(ethBlock); } Observable<PlatonBlock> observable = web3j.replayBlocksObservable( new DefaultBlockParameterNumber(BigInteger.ZERO), new DefaultBlockParameterNumber(BigInteger.valueOf(2)), false, false); final CountDownLatch transactionLatch = new CountDownLatch(ethBlocks.size()); final CountDownLatch completedLatch = new CountDownLatch(1); final List<PlatonBlock> results = new ArrayList<PlatonBlock>(ethBlocks.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(1, TimeUnit.SECONDS); assertThat(results, equalTo(ethBlocks)); subscription.unsubscribe(); completedLatch.await(1, TimeUnit.SECONDS); assertTrue(subscription.isUnsubscribed()); }