How to turn a Mono into a truly asynchronous (not reactive!) method call?-CodePudding

I have a method

@Service
public class MyService {
    public Mono<Integer> processData() {
        ... // very long reactive operation
    }
}

In the normal program flow, I call this method asynchronously via a Kafka event.

For testing purposes I need to expose the method as a web service, but the method should be exposed as asynchronous: returning only HTTP code 200 OK ("request accepted") and continuing the data processing in the background.

Is it OK (= doesn't it have any unwanted side effects) just to call Mono#subscribe() and return from the controller method?

@RestController
@RequiredArgsConstructor
public class MyController {
    private final MyService service;

    @GetMapping
    public void processData() {
        service.processData()
            .subscribeOn(Schedulers.boundedElastic())
            .subscribe();
    }
}

Or is it better to do it like this (here I am confused by the warning from IntelliJ, maybe the same as https://youtrack.jetbrains.com/issue/IDEA-276018 ?):

public Mono<Void> processData() {
    service.processData()
        .subscribeOn(Schedulers.boundedElastic())
        .subscribe(); // IntelliJ complains "Inappropriate 'subscribe' call" but I think it's a false alarm in my case(?)
    return Mono.empty();
}

Or some other solution?

CodePudding user response：

Is it OK (= doesn't it have any unwanted side effects) just to call Mono#subscribe() and return from the controller method?

There are side effects, but you may be ok living with them:

It truly is fire and forget - which means while you'll never be notified about a success (which most people realise), you'll also never be notified about a failure (which far fewer people realise.)
If the process hangs for some reason, that publisher will never complete, and you'll have no way of knowing. Since you're subscribing on the bounded elastic threadpool, it'll also tie up one of those limited threads indefinitely too.

The first point you might be fine with, or you might want to put some error logging further down that reactive chain as a side-effect somehow so you at least have an internal notification if something goes wrong.

For the second point - I'd recommend putting a (generous) timeout on your method call so it at least gets cancelled if it hasn't completed in a set time, and is no longer hanging around consuming resources. If you're running an asynchronous task, then this isn't a massive issue as it'll just consume a bit of memory. If you're wrapping a blocking call on the elastic scheduler then this is worse however, as you're tying up a thread in that threadpool indefinitely.

I'd also question why you need to use the bounded elastic scheduler at all here - it's used for wrapping blocking calls, which doesn't seem to be the foundation of this use case. (To be clear, if your service is blocking then you should absolutely wrap it on the elastic scheduler - but if not then there's no reason to do so.)

Finally, this example:

public Mono<Void> processData() {
    service.processData()
        .subscribeOn(Schedulers.boundedElastic())
        .subscribe();
    return Mono.empty();
}

...is a brilliant example of what not to do, as you're creating a kind of "imposter reactive method" - someone may very reasonably subscribe to that returned publisher thinking it will complete when the underlying publisher completes, which obviously isn't what's happening here. Using a void return type and thus not returning anything is the correct thing to do in this scenario.

CodePudding user response：

Your option with the following code is actually ok:

@GetMapping
public void processData() {
    service.processData()
        .subscribeOn(Schedulers.boundedElastic())
        .subscribe();
}

This is actually what you do in a @Scheduled method which simply returns nothing and you explicitly subscribe to the Mono or Flux so that elements are emitted.