JsonRpc2_0Rx { public Observable<NeoGetBlock> catchUpToLatestAndSubscribeToNewBlocksObservable( BlockParameter startBlock, boolean fullTransactionObjects, long pollingInterval) { return catchUpToLatestBlockObservable( startBlock, fullTransactionObjects, blockObservable(fullTransactionObjects, pollingInterval)); } 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 testCatchUpToLatestAndSubscribeToNewBlockObservable() throws Exception { List<NeoGetBlock> expected = Arrays.asList( createBlock(0), createBlock(1), createBlock(2), createBlock(3), createBlock(4), createBlock(5), createBlock(6) ); OngoingStubbing<NeoGetBlock> stubbingNeoGetBlock = when(neow3jService.send(any(Request.class), eq(NeoGetBlock.class))); for (int i = 0; i < 7; i++) { stubbingNeoGetBlock = stubbingNeoGetBlock.thenReturn(expected.get(i)); } OngoingStubbing<NeoBlockCount> stubbingNeoBlockCount = when(neow3jService.send(any(Request.class), eq(NeoBlockCount.class))); NeoBlockCount neoBlockCount = new NeoBlockCount(); BigInteger currentBlock = BigInteger.valueOf(4); neoBlockCount.setResult(currentBlock); stubbingNeoBlockCount.thenReturn(neoBlockCount); Observable<NeoGetBlock> observable = neow3j .catchUpToLatestAndSubscribeToNewBlocksObservable( new BlockParameterIndex(BigInteger.ZERO), false); CountDownLatch transactionLatch = new CountDownLatch(expected.size()); CountDownLatch completedLatch = new CountDownLatch(1); List<NeoGetBlock> results = new ArrayList<>(expected.size()); Disposable disposable = observable.subscribe( result -> { results.add(result); transactionLatch.countDown(); System.out.println("TransactionLatch countDown"); System.out.println(result.getBlock()); }, throwable -> fail(throwable.getMessage()), () -> { System.out.println("Completed"); completedLatch.countDown(); }); for (int i = 4; i < 7; i++) { Thread.sleep(2000); BigInteger added = neoBlockCount.getBlockIndex().add(BigInteger.ONE); neoBlockCount.setResult(added); stubbingNeoBlockCount = stubbingNeoBlockCount.thenReturn(neoBlockCount); } completedLatch.await(15250, TimeUnit.MILLISECONDS); assertThat(results.size(), equalTo(expected.size())); assertThat(results, equalTo(expected)); disposable.dispose(); assertTrue(disposable.isDisposed()); assertThat(transactionLatch.getCount(), is(0L)); }
@Test @Ignore("Ignored due to a missing feature. " + "A feature to buffer blocks that come out of order should be implemented on neow3j lib.") public void testCatchUpToLatestAndSubscribeToNewBlockObservable_NotContinuousBlocks() throws Exception { List<NeoGetBlock> expected = Arrays.asList( createBlock(0), createBlock(1), createBlock(2), createBlock(3), createBlock(4), createBlock(5), createBlock(6) ); NeoGetBlock block7 = createBlock(7); List<NeoGetBlock> neoGetBlocks = Arrays.asList( expected.get(0), expected.get(1), expected.get(2), expected.get(3), expected.get(4), expected.get(6), block7 ); OngoingStubbing<NeoGetBlock> stubbingNeoGetBlock = when(neow3jService.send(any(Request.class), eq(NeoGetBlock.class))); for (int i = 0; i < 7; i++) { stubbingNeoGetBlock = stubbingNeoGetBlock.thenReturn(neoGetBlocks.get(i)); } OngoingStubbing<NeoBlockCount> stubbingNeoBlockCount = when(neow3jService.send(any(Request.class), eq(NeoBlockCount.class))); NeoBlockCount neoBlockCount = new NeoBlockCount(); BigInteger currentBlock = BigInteger.valueOf(4); neoBlockCount.setResult(currentBlock); stubbingNeoBlockCount.thenReturn(neoBlockCount); Observable<NeoGetBlock> observable = neow3j .catchUpToLatestAndSubscribeToNewBlocksObservable( new BlockParameterIndex(BigInteger.ZERO), false); CountDownLatch transactionLatch = new CountDownLatch(expected.size()); CountDownLatch completedLatch = new CountDownLatch(1); List<NeoGetBlock> results = new ArrayList<>(expected.size()); Disposable disposable = observable.subscribe( result -> { results.add(result); transactionLatch.countDown(); System.out.println("TransactionLatch countDown"); System.out.println(result.getBlock()); }, throwable -> fail(throwable.getMessage()), () -> { System.out.println("Completed"); completedLatch.countDown(); }); for (int i = 4; i < 7; i++) { Thread.sleep(2000); BigInteger added = BigInteger.valueOf(neoGetBlocks.get(i).getBlock().getIndex()); neoBlockCount.setResult(added); stubbingNeoBlockCount = stubbingNeoBlockCount.thenReturn(neoBlockCount); } transactionLatch.await(15250, TimeUnit.MILLISECONDS); assertThat(results.size(), equalTo(expected.size())); assertThat(results, new ReflectionEquals(expected)); disposable.dispose(); completedLatch.await(1, TimeUnit.SECONDS); assertTrue(disposable.isDisposed()); }