Completed event for ethLogFlowable

[Svanar]

Hello, Recently I had opportunity to work with web3j and i really enjoy that, however I have question about processing old data. Use case I am trying to implement looks like:

1. Get some block number that already happened
2. Process all transactions within that block according to given filter
3. Notify about finish processing

Everything seems to work fine apart from point 3. When using ethLogFlowable I am not able to determinate whether processing is finished or not. I am working only with historic data, so I know that there is finite number of events, but I don't get mentioned completed event.

While I tried to use replayPastTransactionsFlowable it got job done, BUT it produces a lot more events that eth log with filter.

So what to sum up I would like to ask about:

1. Is this some bug that completed events for block that already exists is not received ? What I would expect would be that this event is present once upper bound block is created
2. In case above is not possible - is there something else that I can use to take advantage of both filter (like ethLog) and completed events (like replayPastTransactionsFlowable) ?

I have attached code example that I used to test. Thank you for any replies !

class CompleteEvent {

    public static void main(String[] args) {
        BigInteger block = BigInteger.valueOf(10549000L);
        DefaultBlockParameterNumber blockNumber = new DefaultBlockParameterNumber(block);

        String apiKey = "MY_API_KEY";
        String gateway = "https://rinkeby.infura.io/v3/" + apiKey;
        Web3j web3j = Web3j.build(new HttpService(gateway));


        replayPastTransactionsTest(blockNumber, web3j);
        ethLogFlowableTest(blockNumber, web3j);
    }

    private static void replayPastTransactionsTest(DefaultBlockParameterNumber blockNumber, Web3j web3j) {
        System.out.println("Starting 'replayPastTransactionsFlowable'");
        Flowable<Transaction> transactionFlowable = web3j.replayPastTransactionsFlowable(blockNumber, blockNumber);
        transactionFlowable
                .subscribe(
                        trx -> System.out.println("Processing transaction " + trx.getHash() + "..."),
                        Throwable::printStackTrace,
                        () -> System.out.println("Completed event for 'replayPastTransactionsFlowable'")
                );
        System.out.println("replayPastTransactionsFlowable - done");
    }

    private static void ethLogFlowableTest(DefaultBlockParameterNumber blockNumber, Web3j web3j) {
        System.out.println("Starting 'ethLogFlowable'");

        Event eventTopic = new Event("Transfer",
                Arrays.asList(new TypeReference<Address>(true) {},
                        new TypeReference<Address>(true) {},
                        new TypeReference<Uint256>(true) {}));

        EthFilter filter = new EthFilter(blockNumber, blockNumber, new LinkedList<>())
                .addSingleTopic(EventEncoder.encode(eventTopic));

        Flowable<Log> logFlowable = web3j.ethLogFlowable(filter);
        logFlowable
                .subscribe(
                        log -> System.out.println("Processing log " + log.getTransactionHash() + "..."),
                        Throwable::printStackTrace,
                        () -> System.out.println("Completed event for 'ethLogFlowable'")
                );
        System.out.println("ethLogFlowable - done");
    }

}

And output looks like:

Starting 'replayPastTransactionsFlowable'
replayPastTransactionsFlowable - done
Starting 'ethLogFlowable'
Processing transaction 0xdd4ff1ceaca3f140b969586b6f44072cd1e2c9a7b43887e92591d64c29f60558...
...
Processing transaction 0x34b95ec71ab23b0ae40bc0ab65a1f3fa2729210c3918d0fe6b83196c6ebee408...
Completed event for 'replayPastTransactionsFlowable'

Processing log 0xd76e09b3872ebd6407ae16d645e16d41e95f8f30ad257360ec9f8a5c8a3b4552...
...
Processing log 0xeea7a51162f07c927e90df349fa771f11341711e3f8b71d48b5bce0b265436fe...

ethLogFlowable - done

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.

[baasanbat]

so was it a bug or feature? I am still interested in detail on that. Because I can't release the thread on subscription.

[netbizltd]

Did anybody figure this out?

I am currently running into the same problem where I have an outer web3j.replayPastAndFutureBlocksFlowable and for each block I have an inner contract.transferEventFlowable. The inner flowable - for a single block - never completes even though the block is in the past.

[kirzhanov]

Did anybody figure this out?

That's still actual question..

[13520001752]

I used transferEventFlowable to subscribe historical token transactions, also need the complement event to finish the subscription.

[13520001752]

Did anybody figure this out?

I am currently running into the same problem where I have an outer web3j.replayPastAndFutureBlocksFlowable and for each block I have an inner contract.transferEventFlowable. The inner flowable - for a single block - never completes even though the block is in the past.

@netbizltd Have you god any solution on this problem.