Ensuring method execution in Spring Webflux

Question

override fun uploadFileAndNotifyCadmium(
        validatedPages: List<KycFile>,
        documentId: UUID,
    ): Mono<Unit> {

        validatedPages.forEach { it ->
            s3Service.uploadToToxicBucket(
                documentId, it.fileName })
                .then()
        }

        return cadmiumClient.notifyCadmium(documentId) // API call to other microservice
    }

I have this method where I am trying to call a method (s3Service.uploadToToxicBucket())which returns a Mono of Unit while iterating a list. I want to ensure that this method is called for each of the list elements. Therefore I have added then() to the resulting Mono. Also I want the method cadmiumClient.notifyCadmium() to be executed only after all the files have been uploaded. Is this the correct way of doing this or can I use some other operator. cadmiumClient.notifyCadmium() returns a Mono of Unit as well.

Also I am calling uploadFileAndNotifyCadmium().then() when I am calling uploadFileAndNotifyCadmium().

I am using then() because the execution is lazy, and I don't want to call subscribe() to make sure of the execution of these methods. Also the method uploadFileAndNotifyCadmium() is called in the controller therefore it is auto subscribed by the Spring Webflux. My understanding regarding the usage of then() maybe wrong.

Approach 2: I have also thought of doing this:

override fun uploadFileAndNotifyCadmium(
        validatedPages: List<KycFile>,
        documentId: UUID,
    ): Mono<Unit> {

        val list = mutableListOf<Mono<Unit>>()

        validatedPages.forEach { it ->
            val result =s3Service.uploadToToxicBucket(
                documentId, it.fileName })

        list.add(result)

        }

        //Zip all the elements in list together into some variable zippedResult

        return zippedResult.then(cadmiumClient.notifyCadmium(documentId)) // API call to other microservice
    }

For this approach I can't find any operator to zip all the elements together when the size of list is not known before.

your understanding is wrong, whomever initiating the call, is the one subscribing. So the calling client will under the hood call subscribe and start the chain, if you have a list you should make it a flux and for each emitted object from the flux you can do what it is you want to do. Its very hard when you have just given us a piece of the code. There's no idea to help you since we dont know any return types from any of the functions you have posted. Are they returning Mono, strings, fluxes, lists etc etc — Toerktumlare
@ThomasAndolf I have updated the question with the return types. Also I understand that calling client subscribes the method, in my case this method is being called from the controller therefore it is subscribed by Webflux. I am not sure if I am using then() correctly and approach 2 seems a better way to do it, but I don't know how to combine multiple monos when the size of the list is not known beforehand. — Raman Preet Singh

Toerktumlare Toerktumlare · Accepted Answer · 2020-03-08T19:57:14

Im trying to understand what it is you want to do, because your explanation is quite unclear.

I don't code in Kotlin, but if i get it right, what you want to do is:

Take in a list of n number of objects
perform some type of side-effect on each object in the list
collect some form of result of all the side effects
return this result

this is how i would of done it in plain reactive java

    return Flux.fromIterable(List.of("one", "two", "three"))
            .flatMap( // do some side effect )
            .collect(Collectors.toList());

Convert the list to a Flux<T>
Perform a side effect in flatMap for each emitted item
Collect all the results as a list
return the Mono<List<T>>

Ensuring method execution in Spring Webflux

1 Answers