Create multi from "while(hasNext)" loop

[jtama]

I am using elastic search async client which is able to return CompletableFuture. The request to elastic are paginated and I want to create an enpoint that returns a Multi that complete when I have reached the end of elastic pages.

Whith sync clilent I can do :

return Multi.createFrom().emitter(emitter -> {
            var hasNext = true;
            List<String> searchAfterList = new ArrayList<>();
            while (hasNext){
                var result = getPaginateItems(dataSourceId, sort, searchAfterList);
                hasNext = !result.isEmpty();
                if (hasNext) {
                    emitter.emit(result);
                    searchAfterList.clear();
                    searchAfterList.addAll(result.getSearchAfterList());
                } else {
                    emitter.complete();
                }
            }
        });

But this is blocking. How can I achieve the same with getPaginateItems returning CompletableFuture ?

[jponge]

The emitter doesn't have to be a in loop, and it can outlive the lifespan of the lambda / callback.

I'd suggest you use one of the CompletableFuture method to be notified when it completes, then push elements using emitter, then call getPaginateItems again.

[jtama]

So just to sure, if I have Uni instead of CompletableFuture, what you suggest is somthing like :

    public Multi<ItemsSearchResult> getAllItems(UUID dataSourceId, SortDto sort) {
        return Multi.createFrom().emitter(emitter -> {
            String searchAfter = "";
            getPaginatedItems(dataSourceId, sort, searchAfter)
                .subscribe()
                .with(result -> processResult(result, emitter, dataSourceId, sort));
        });
    }

    private void processResult(ItemsSearchResult result, MultiEmitter emitter, UUID dataSourceId, SortDto sort) {
        if (!result.isEmpty()) {
            emitter.emit(result);
            getPaginatedItems(dataSourceId, sort, result.getSearchAfter()).subscribe()
                .with(follow -> processResult(follow, emitter, dataSourceId, sort));
        } else {
           emitter.complete();
        }
    }

Recursive async io code, it's what I end up with and want to be clear I am not heading the wrong direction...

[jponge]

I'm not familiar with your code base, but that sounds like an approach that should work, and makes sure you don't block an event-loop AFAIK.

[jtama]

yep we already run into blocking... Solved by doing :

 Uni.createFrom()
                .voidItem()
                .emitOn(Infrastructure.getDefaultWorkerPool())
                .onItem().transform(ignored -> blockerCode());