public Flowable<Transaction> replayTransactionsFlowable(
            DefaultBlockParameter startBlock, DefaultBlockParameter endBlock) {
        return replayBlocksFlowable(startBlock, endBlock, true)
                .flatMapIterable(JsonRpc2_0Rx::toTransactions);
    }
@Test
    public void testReplayTransactionsFlowable() throws Exception {

        List<EthBlock> ethBlocks =
                Arrays.asList(
                        createBlockWithTransactions(
                                0,
                                Arrays.asList(
                                        createTransaction("0x1234"),
                                        createTransaction("0x1235"),
                                        createTransaction("0x1236"))),
                        createBlockWithTransactions(
                                1,
                                Arrays.asList(
                                        createTransaction("0x2234"),
                                        createTransaction("0x2235"),
                                        createTransaction("0x2236"))),
                        createBlockWithTransactions(
                                2,
                                Arrays.asList(
                                        createTransaction("0x3234"), createTransaction("0x3235"))));

        OngoingStubbing<EthBlock> stubbing =
                when(web3jService.send(any(Request.class), eq(EthBlock.class)));
        for (EthBlock ethBlock : ethBlocks) {
            stubbing = stubbing.thenReturn(ethBlock);
        }

        List<Transaction> expectedTransactions =
                ethBlocks.stream()
                        .flatMap(it -> it.getBlock().getTransactions().stream())
                        .map(it -> (Transaction) it.get())
                        .collect(Collectors.toList());

        Flowable<Transaction> flowable =
                web3j.replayPastTransactionsFlowable(
                        new DefaultBlockParameterNumber(BigInteger.ZERO),
                        new DefaultBlockParameterNumber(BigInteger.valueOf(2)));

        CountDownLatch transactionLatch = new CountDownLatch(expectedTransactions.size());
        CountDownLatch completedLatch = new CountDownLatch(1);

        List<Transaction> results = new ArrayList<>(expectedTransactions.size());
        Disposable subscription =
                flowable.subscribe(
                        result -> {
                            results.add(result);
                            transactionLatch.countDown();
                        },
                        throwable -> fail(throwable.getMessage()),
                        () -> completedLatch.countDown());

        transactionLatch.await(1, TimeUnit.SECONDS);
        assertEquals(results, (expectedTransactions));

        subscription.dispose();

        completedLatch.await(1, TimeUnit.SECONDS);
        assertTrue(subscription.isDisposed());
    }