How to wrap blocking IO operation for Webflux applicaiton

Question

I have a Spring Boot/Webflux application. I need an async REST endpoint that does:

1. Generate random job id.
2. Call one service via HTTP.
3. Call another service via HTTP.
4. Combine responses from the services and write the result into a file.
5. Return the job id to the client.

The endpoint should be async. That's why a client shouldn't wait the result of steps: 2,3,4. The client should receive a job id instantly.

Right now I have following implementation:

@Override
  public Mono<String> saveData() {
    String jobId = UUID.randomUUID().toString();
    Mono<ResponseFromService1> response1 = service1.getData();
    Mono<ResponseFromService2> response2 = service2.getData();
    return fileService.saveData(response1, response2)
        .map(filePath -> log.info("File has been stored at {}", filePath))
        .map(jobId);

service1 and service2 are implemented by using reactive WebClient. Implementation of fileService.saveData looks like:

  public Mono<Path> saveDataInFile(Mono<ResponseFromService1> response1,Mono<ResponseFromService2> response2) {
return Mono.fromCallable(() ->
    Mono.zip(response1, response2)
        .map(tuple -> blockingIOsaveMethod(tuple.getT1(), tuple.getT2()))
).publishOn(Schedulers.elastic())
    .flatMap(mono -> mono);
  }

The problem is that this endpoint is not asynchronous. The client of the endpoint gets the job id after the file with data was saved. How should I update saveDataInFile and saveData to return the job id instantly?

Answer 1

The client should receive a job id instantly.

This seems to show Mono<String> isn't the right return type for saveData(), as you don't want to wait for any asynchronous operation to finish apparently:

@Override
public String saveData() {
  String jobId = UUID.randomUUID().toString();
  // ...
  return jobId;
}

The ... looks like a "fire-and-forget" operation, you could perhaps manually subscribe to the Mono:

@Override
public String saveData() {
  String jobId = UUID.randomUUID().toString();
  fileService.saveData(...).subscribe(...); // look at the different overloads of #subscribe(...)
  return jobId;
}

Make sure everything is well logged, so that you don't lose track of what happens after the HTTP response is returned.