JsonRpc2_0Rx { public Observable<NeoGetBlock> replayBlocksObservable( BlockParameter startBlock, BlockParameter endBlock, boolean fullTransactionObjects) { return replayBlocksObservable(startBlock, endBlock, fullTransactionObjects, true); } JsonRpc2_0Rx(Neow3j neow3j, ScheduledExecutorService scheduledExecutorService); Observable<BigInteger> neoBlockObservable(long pollingInterval); Observable<NeoGetBlock> replayBlocksObservable(
            BlockParameter startBlock, BlockParameter endBlock,
            boolean fullTransactionObjects); Observable<NeoGetBlock> replayBlocksObservable(
            BlockParameter startBlock, BlockParameter endBlock,
            boolean fullTransactionObjects, boolean ascending); Observable<NeoGetBlock> catchUpToLatestBlockObservable(
            BlockParameter startBlock, boolean fullTransactionObjects,
            Observable<NeoGetBlock> onCompleteObservable); Observable<NeoGetBlock> catchUpToLatestBlockObservable(
            BlockParameter startBlock, boolean fullTransactionObjects); Observable<Transaction> catchUpToLatestTransactionObservable(
            BlockParameter startBlock); Observable<NeoGetBlock> catchUpToLatestAndSubscribeToNewBlocksObservable(
            BlockParameter startBlock, boolean fullTransactionObjects,
            long pollingInterval); Observable<NeoGetBlock> blockObservable(boolean fullTransactionObjects, long pollingInterval);  }
@Test public void testReplayBlocksObservable() throws Exception { List<NeoGetBlock> neoGetBlocks = Arrays .asList(createBlock(0), createBlock(1), createBlock(2)); OngoingStubbing<NeoGetBlock> stubbing = when(neow3jService.send(any(Request.class), eq(NeoGetBlock.class))); for (NeoGetBlock neoGetBlock : neoGetBlocks) { stubbing = stubbing.thenReturn(neoGetBlock); } Observable<NeoGetBlock> observable = neow3j.replayBlocksObservable( new BlockParameterIndex(BigInteger.ZERO), new BlockParameterIndex(BigInteger.valueOf(2)), false); CountDownLatch transactionLatch = new CountDownLatch(neoGetBlocks.size()); CountDownLatch completedLatch = new CountDownLatch(1); List<NeoGetBlock> results = new ArrayList<>(neoGetBlocks.size()); Disposable disposable = observable.subscribe( result -> { results.add(result); transactionLatch.countDown(); }, throwable -> fail(throwable.getMessage()), () -> completedLatch.countDown()); completedLatch.await(5, TimeUnit.SECONDS); assertThat(results, equalTo(neoGetBlocks)); disposable.dispose(); assertTrue(disposable.isDisposed()); assertThat(transactionLatch.getCount(), is(0L)); }
@Test public void testReplayBlocksDescendingObservable() throws Exception { List<NeoGetBlock> neoGetBlocks = Arrays .asList(createBlock(2), createBlock(1), createBlock(0)); OngoingStubbing<NeoGetBlock> stubbing = when(neow3jService.send(any(Request.class), eq(NeoGetBlock.class))); for (NeoGetBlock neoGetBlock : neoGetBlocks) { stubbing = stubbing.thenReturn(neoGetBlock); } Observable<NeoGetBlock> observable = neow3j.replayBlocksObservable( new BlockParameterIndex(BigInteger.ZERO), new BlockParameterIndex(BigInteger.valueOf(2)), false, false); CountDownLatch transactionLatch = new CountDownLatch(neoGetBlocks.size()); CountDownLatch completedLatch = new CountDownLatch(1); List<NeoGetBlock> results = new ArrayList<>(neoGetBlocks.size()); Disposable disposable = observable.subscribe( result -> { results.add(result); transactionLatch.countDown(); }, throwable -> fail(throwable.getMessage()), () -> completedLatch.countDown()); completedLatch.await(5, TimeUnit.SECONDS); assertThat(results, equalTo(neoGetBlocks)); disposable.dispose(); assertTrue(disposable.isDisposed()); assertThat(transactionLatch.getCount(), is(0L)); }