How to stop main thread to complete all Mono calls?

Question

I'm making multiple mono calls to DB.And result of all Mono response is needed to compute final result which is written after declared Mono logic.

if (SomeObject.getAccountLevelActiveList() != null) {

                SomeObject.getAccountLevelActiveList().parallelStream().forEach(account -> {
                    Mono<SubLine> subLineMono= SubLineService
                            .getLineLevelCustProfile(preNbsLineLevelConverter.getSubLine(account ));
                

                    subLineMono.subscribe(subLine-> PollObject.getSubList()
                            .put(accountLevelMtn.getMtn(), Optional.ofNullable(subLine)));

                });

            }

But my main logic is getting executed before mono result stored to the PollObject. so i'm getting null in the PollObject. So i want to stop my main thread until Mono results stored into the PollObject.

You generally should be chaining your "main logic" off of the reactive pipeline. — chrylis -cautiouslyoptimistic-, Sep 25 '20 at 16:55
@chrylis-cautiouslyoptimistic- yes, i have main logic after if block execution by passing 'PollObject'. — John, Sep 25 '20 at 16:58
I understand, and I'm saying to move that logic _into_ the reactive pipeline. — chrylis -cautiouslyoptimistic-, Sep 25 '20 at 17:00
Could you please elaborate on your comment.@chrylis-cautiouslyoptimistic- — John, Sep 25 '20 at 17:02

score 1 · Accepted Answer · answered Sep 28 '20 at 09:13

If you want to stop main thread then you can use blocking instead of subscribing, but you must first convert the List into Flux and then flatMap-it using provided Mono. The logic you have in the subscribe method can be moved into a side effect operator doOnNext either of the Mono or the wrapping Flux:

Flux.fromIterable(SomeObject.getAccountLevelActiveList())
    .flatMap(account ->
        SubLineService.getLineLevelCustProfile(
            preNbsLineLevelConverter.getSubLine( account ))
    ).doOnNext(subLine ->
        PollObject.getSubList().put(accountLevelMtn.getMtn(),
            Optional.ofNullable(subLine))
    ).blockLast();
    // the following code will be executed first when all monos are completed

If your code following the if is not required to run in the main thread, it would be better stay reactive as the @chrylis-cautiouslyoptimistic already suggested. Use reduce operator to put all the result together producing mono that is completed when all the provided monos are completed:

Flux.fromIterable(SomeObject.getAccountLevelActiveList())
    .flatMap(account ->
        SubLineService.getLineLevelCustProfile(
            preNbsLineLevelConverter.getSubLine( account ))
    ).reduce(PollObject.getSubList(), (subList, subLine) ->
        sublist.put(accountLevelMtn.getMtn(), Optional.ofNullable(subLine))
    ).map(subList -> {
        // the code here will be executed first when all monos are completed
    })
    // ... other operators if necessary
    // eventually subscribing or returning the mono for further processing 
    .subscribe();

How to stop main thread to complete all Mono calls?

1 Answers1