Project reactor: repeat service call and collect response results until condition met

Question

Greetings venerable SO community...

I come to everyone today in need of a reactive solution for a problem I've not yet encountered. I would still consider myself a reactive neophyte at this point, and although I've been reading over "Hands On Reactive Programming in Spring" I've still got the boss looking over me and tapping his foot if you know what I mean, so I'm again hoping that some benevolent and brilliant soul might point me in the right direction for an implementation I'm presently tasked with.

Here's the situation: I have to call an external service providing a list of transactions(containing transaction data) for a given date range. In addition to the date range being passed in the request to the external service, there's also number of transactions per page, we'll call "N", and page number, we'll call "P."

So say for a certain date range there is a total of 23 transactions.

Then:

• For N=10 and P=1 they will return 10 transactions, from 1-10
• For N=10 and P=2 we’d get transactions 11-20.
• For N=10 and P=3 we’d get transactions 21-23.
• For N=10 and P=4 or higher no transactions would be returned.
• For N=7 and P=3 we’d get transactions 15-21.
• For N=30 and P=1 we’d get all 23 transactions and know that we have all of the transactions because the number returned is less than number requested.

(Please note, we have no idea how many transaction records in total will ultimately be returned.)

So what I'm after is some sort of reactive construct that can potentially repeatedly call the remote transaction list service, check if the total number returned is less than the total number requested, and if not, call again while saving up/accumulating all the records in another list, and then once we reach the end, send this other list of all the transactions back in a response to the service that invoked this code.

Does this make sense, or no?

I've been searching SO, and so far I found this article that seems somewhat applicable and promising using the "repeat" functionality, but I'm really not sure if this is the right path or not.

I also found this article which also seems like it might be relevant.

Again, and as always, any and all help/guidance here is immensely appreciated!

Answer 1

Greetings once again SO community.

I was finally able to come up with a solution for this problem thanks to help from a more experienced co-worker.

public Flux<TransactionListData> getTransactionList(String cardId, TransactionLookupCriteria lookupCriteria) {
    return cardService.getByCardId(cardId)
        .map(Card::getAccountId)
        .flatMap(accountService::getAccount)
        .flatMap(account -> fetchTransactions(account, lookupCriteria))
        .flatMapMany(Flux::fromIterable)
        .map(this::negateTransactionAmountIfNecessary)
        .map(this::setTransactionApprovalFlag);
}

private Mono<List<TransactionListData>> fetchTransactions(Account account,
                                                          TransactionLookupCriteria lookupCriteria) {
    final PageNumber pageNumber = new PageNumber();
    return Flux.defer(() -> getTransactions(account, lookupCriteria, pageNumber))
        .repeatWhen(transactions -> transactions.takeWhile(transactionCount -> transactionCount == TRANSACTIONS_REQUESTED_PER_PAGE))
        .collectList();
}

private Flux<TransactionListData> getTransactions(Account account, TransactionLookupCriteria lookupCriteria, PageNumber pageNumber) {
    return Mono.just(createGetTransactionListServiceRequestData(account, lookupCriteria, pageNumber.get()))
        .flatMap(cartaClient::getTransactionList)
        .map(GetTransactionListServiceResponse::getServiceResponseData)
        .switchIfEmpty(Mono.defer(() -> Mono.error(new ServiceException("Transaction response data empty"))))
        .doOnNext(x -> pageNumber.increment())
        .flatMapIterable(GetTransactionListServiceResponseData::getTransactionList);
}

private class PageNumber {

    int page = 1;

    void increment() {
        page++;
    }

    public int get() {
        return page;
    }
}