@Test
public void testCatchUpToLatestAndSubscribeToNewBlockObservable() throws Exception {

    List<EthBlock> expected = Arrays.asList(
            createBlock(0), createBlock(1), createBlock(2),
            createBlock(3), createBlock(4), createBlock(5),
            createBlock(6));
    List<EthBlock> ethBlocks = Arrays.asList(
            expected.get(2),  // greatest block
            expected.get(0), expected.get(1), expected.get(2),
            expected.get(4), // greatest block
            expected.get(3), expected.get(4),
            expected.get(4),  // greatest block
            expected.get(5),  // initial response from ethGetFilterLogs call
            expected.get(6)); // subsequent block from new block observable
    OngoingStubbing<EthBlock> stubbing =
            when(web3jService.send(any(Request.class), eq(EthBlock.class)));
    for (EthBlock ethBlock : ethBlocks) {
        stubbing = stubbing.thenReturn(ethBlock);
    }
    EthFilter ethFilter = objectMapper.readValue(
            "{\n"
                    + "  \"id\":1,\n"
                    + "  \"jsonrpc\": \"2.0\",\n"
                    + "  \"result\": \"0x1\"\n"
                    + "}", EthFilter.class);
    EthLog ethLog = objectMapper.readValue(
            "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":["
                    + "\"0x31c2342b1e0b8ffda1507fbffddf213c4b3c1e819ff6a84b943faabb0ebf2403\""
                    + "]}",
            EthLog.class);
    EthUninstallFilter ethUninstallFilter = objectMapper.readValue(
            "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":true}", EthUninstallFilter.class);
    when(web3jService.send(any(Request.class), eq(EthFilter.class)))
            .thenReturn(ethFilter);
    when(web3jService.send(any(Request.class), eq(EthLog.class)))
            .thenReturn(ethLog);
    when(web3jService.send(any(Request.class), eq(EthUninstallFilter.class)))
            .thenReturn(ethUninstallFilter);
    Observable<EthBlock> observable = web3j.catchUpToLatestAndSubscribeToNewBlocksObservable(
            new DefaultBlockParameterNumber(BigInteger.ZERO),
            false);
    final CountDownLatch transactionLatch = new CountDownLatch(expected.size());
    final CountDownLatch completedLatch = new CountDownLatch(1);
    final List<EthBlock> results = new ArrayList<EthBlock>(expected.size());
    Subscription subscription = observable.subscribe(
            new Action1<EthBlock>() {
                @Override
                public void call(EthBlock 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());
}