@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());
}