spring - Fire and forget with reactor

Question

I have a method like below in my Spring boot app.

public Flux<Data> search(SearchRequest request) {
  Flux<Data> result = searchService.search(request);//this returns Flux<Data>
  Mono<List<Data>> listOfData = result.collectList();
//  doThisAsync() // here I want to pass this list and run some processing on it
// the processing should happen async and the search method should return immediately.
  return result;
}

//this method uses the complete List<Data> returned by above method
public void doThisAsync(List<Data> data) {
  //do some processing here
}

Currently, I'm using @Async annotated service class with doThisAsync, but don't know how to pass the List<Data>, because I don't want to call block. All I have is Mono<List<Data>>.

My main problem is how to process this Mono separately and the search method should return the Flux<Data>.

Answer 1

1, If your fire-and-forget is already async returning Mono/Flux

public Flux<Data> search(SearchRequest request)
{
    return searchService.search(request)
                        .collectList()
                        .doOnNext(data -> doThisAsync(data).subscribe())  // add error logging here or inside doThisAsync
                        .flatMapMany(Flux::fromIterable);
}

public Mono<Void> doThisAsync(List<Data> data) {
    //do some async/non-blocking processing here like calling WebClient
}

2, If your fire-and-forget does blocking I/O

public Flux<Data> search(SearchRequest request)
{
    return searchService.search(request)
                        .collectList()
                        .doOnNext(data -> Mono.fromRunnable(() -> doThisAsync(data))
                                              .subscribeOn(Schedulers.elastic())  // delegate to proper thread to not block main flow
                                              .subscribe())  // add error logging here or inside doThisAsync
                        .flatMapMany(Flux::fromIterable);
}

public void doThisAsync(List<Data> data) {
    //do some blocking I/O on calling thread
}

Note that in both of the above cases you lose backpressure support. If the doAsyncThis slows down for some reason, then the data producer won't care and keep producing items. This is a natural consequence of the fire-and-foget mechanism.