Feature request: Multi.createFrom().generator() with async generator

[bboyz269]

I'm using generator get data from DB in chunk (quarkus-hibernate-reactive-panache)

fun data() : Multi<List<ENTITY>> = Multi.createFrom().generator(
    { ENTITY.findAll().page(Page.ofSize(10)) }, // Create a pagable query for all data upon subscription
    { pagableQuery, emitter ->
        emitter.emit(pagableQuery.list()) // retrieve data of current page (problem 1)
        if (!pagableQuery.hasNext()) emitter.complete() // check if there's no next page then complete (problem 2)
        return pagableQuery.nextPage() // jump to next page
    }
)

Problems:

1. pagableQuery.list() return Uni<List> instead of List<ENTITY>, db access up on request
2. pagableQuery.hasNext() return Uni<Boolean> instead of Boolean, db access up on request

Wouldn't it make sense that generator is able to accept / resolve values from Uni source?

Something like

interface GeneratorEmitterFromUni<T> {
    fun emitFromUni(itemSource: Uni<T>)
    fun completeFromUni(completionSource: Uni<Boolean>)
}

// Sorry code not in Java

[jponge]

I'm not sure why you want to wrap into a Multi here (and also it's more a Panache question than a Mutiny question).

According to https://quarkus.io/guides/hibernate-reactive-panache#paging . pageCount() gives you the number of pages, so you could build a Multi by repeating calls to a Uni supplier, as in the following non-Panache example:

Uni<List<String>> batch = Uni.createFrom().item(List.of("a", "b", "c"));

Multi.createBy().repeating().uni(() -> batch).atMost(4) // replace the value in atMost by the number of pages
        .subscribe().with(System.out::println);

Hope that helps!

[jponge]

...and if you want to flatten the batches into a stream of elements rather than a stream of lists:

Multi.createBy().repeating().uni(() -> batch).atMost(4)
                .onItem().transformToIterable(list -> list)
                .subscribe().with(System.out::println);

[bboyz269]

Thank you for your answer!

I'm not sure why you want to wrap into a Multi here

It allows me to process chunk of records in parallel, smaller transactions. Something likes

multiOfList
    .onItem().transformToUni {records -> processInNewTransaction(records) }
    .merge(5) // parallel 5
    .collectFailures() 

Anyway, I just want to make use of Panache apis elegantly, consider ditching it now.