Why am i unable to print the elements in a Flux payload/ save it to database without returning?

Question

So my endpoint is like this :-

public Mono<Void> postData( Flux<SomeModelClass> dataFlux ){
...
repository.saveAll(dataFlux);
...
}

Now since this is not saving the 'dataFlux' to my database , i had to change the return type and method definition as:

public Flux<SomeModelClass> postData( Flux<SomeModelClass> dataFlux ){
...
return repository.saveAll(dataFlux);
...
}

But i still want to be able to save the data without having to return anything.

I tried to create subscription explicitly but that too doesn't help save the data to db:

public Mono<Void> postData( Flux<SomeModelClass> dataFlux ){

dataFlux.subscribe();

repository.saveAll(dataFlux);
}

I tried reading the elements of the payload by doing:

public Mono<Void> postData( Flux<SomeModelClass> dataFlux ){

dataFlux.subscribe().doOnNext(e -> System.out.println(e));

repository.saveAll(dataFlux);
}

But this to does not print the elements to the logs.

Answer 1

From the documentation:

Nothing happens until you subscribe

Reactive stream has to been subscribed but fortunately the framework does it for you though for this you have to return with a reference to the given stream. These examples are not correct but I guess you didnt return with the result of the saveAll but something else (Mono.empty()). This wont work as nothing subscribes to the saveAll pipeline.

The FW will only subscribe to the returning stream reference. The rule of thumb of webflux is no subscribe to any stream if you dont know what you do (i would say do not use until you learn how it works).

So if you want to save the data, log it and return with Void Mono, you can do something like this:

@PostMapping
public Mono<Void> postData(@RequestBody Flux<SomeModelClass> dataFlux) {
    return repository
            .saveAll(dataFlux)
            .doOnNext(n -> logger.info(n.toString()))
            .then();
}

All the operations are in the same pipeline, so all will run when the FW subscribes to it. First it saves the data then log every elements one by one and the then() call will return with a Mono<Void> after the upstream (Flux) completes.