Reactor Mono - execute parallel tasks

Question

I am new to Reactor framework and trying to utilize it in one of our existing implementations. LocationProfileService and InventoryService both return a Mono and are to executed in parallel and have no dependency on each other (from the MainService). Within LocationProfileService - there are 4 queries issued and the last 2 queries have a dependency on the first query.

What is a better way to write this? I see the calls getting executed sequentially, while some of them should be executed in parallel. What is the right way to do it?

public class LocationProfileService {
        static final Cache<String, String> customerIdCache //define Cache

        @Override
        public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {
            //These 2 are not interdependent and can be executed immediately
            Mono<String> customerAccountMono = getCustomerArNumber(customerId,location) LocationNumber).subscribeOn(Schedulers.parallel()).switchIfEmpty(Mono.error(new CustomerNotFoundException(location, customerId))).log();
            Mono<LocationProfile> locationProfileMono = Mono.fromFuture(//location query).subscribeOn(Schedulers.parallel()).log();

    //Should block be called, or is there a better way to do ?
            String custAccount = customerAccountMono.block(); // This is needed to execute and the value from this is needed for the next 2 calls

            Mono<Customer> customerMono = Mono.fromFuture(//query uses custAccount from earlier step).subscribeOn(Schedulers.parallel()).log();
            Mono<Result<LocationPricing>> locationPricingMono = Mono.fromFuture(//query uses custAccount from earlier step).subscribeOn(Schedulers.parallel()).log();

            return Mono.zip(locationProfileMono,customerMono,locationPricingMono).flatMap(tuple -> {
                LocationProfileInfo locationProfileInfo = new LocationProfileInfo();
                //populate values from tuple
                return Mono.just(locationProfileInfo);
            });


        }

        private Mono<String> getCustomerAccount(String conversationId, String customerId, String location) {
            return CacheMono.lookup((Map)customerIdCache.asMap(),customerId).onCacheMissResume(Mono.fromFuture(//query).subscribeOn(Schedulers.parallel()).map(x -> x.getAccountNumber()));
        }

}


public class InventoryService {

    @Override
    public Mono<InventoryInfo> getInventoryInfo(String inventoryId) {
        Mono<Inventory> inventoryMono = Mono.fromFuture(//inventory query).subscribeOn(Schedulers.parallel()).log();
        Mono<List<InventorySale>> isMono = Mono.fromFuture(//inventory sale query).subscribeOn(Schedulers.parallel()).log();

        return Mono.zip(inventoryMono,isMono).flatMap(tuple -> {
            InventoryInfo inventoryInfo = new InventoryInfo();
            //populate value from tuple

            return Mono.just(inventoryInfo);

        });
    }

}

public class MainService {

        @Autowired
        LocationProfileService locationProfileService;
        @Autowired
        InventoryService inventoryService

        public void mainService(String customerId, String location, String inventoryId) {
            Mono<LocationProfileInfo> locationProfileMono = locationProfileService.getProfileInfoByLocationAndCustomer(....);
            Mono<InventoryInfo> inventoryMono = inventoryService.getInventoryInfo(....);

            //is using block fine or is there a better way to do?
            Mono.zip(locationProfileMono,inventoryMono).subscribeOn(Schedulers.parallel()).block();
        }

}

Answer 1

You don't need to block in order to get the pass that parameter your code is very close to the solution. I wrote the code using the class names that you provided. Just replace all the Mono.just(....) with the call to the correct service.

    public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {
    Mono<String> customerAccountMono = Mono.just("customerAccount");
    Mono<LocationProfile> locationProfileMono = Mono.just(new LocationProfile());

    return Mono.zip(customerAccountMono, locationProfileMono)
            .flatMap(tuple -> {
                Mono<Customer> customerMono = Mono.just(new Customer(tuple.getT1()));
                Mono<Result<LocationPricing>> result = Mono.just(new Result<LocationPricing>());
                Mono<LocationProfile> locationProfile = Mono.just(tuple.getT2());
                return Mono.zip(customerMono, result, locationProfile);
            })
            .map(LocationProfileInfo::new)
    ;
}

public static class LocationProfileInfo {
    public LocationProfileInfo(Tuple3<Customer, Result<LocationPricing>, LocationProfile> tuple){
        //do wathever
    }
}


public static class LocationProfile {}

private static class Customer {
    public Customer(String cutomerAccount) {
    }
}

private static class Result<T> {}

private static class LocationPricing {}

Pleas remember that the first zip is not necessary. I re write it to mach your solution. But I would solve the problem a little bit differently. It would be clearer.

public Mono<LocationProfileInfo> getProfileInfoByLocationAndCustomer(String customerId, String location) {
return Mono.just("customerAccount") //call the service                                                
        .flatMap(customerAccount -> {                                                                 
            //declare the call to get the customer                                                    
            Mono<Customer> customerMono = Mono.just(new Customer(customerAccount));                   

            //declare the call to get the location pricing                                            
            Mono<Result<LocationPricing>> result = Mono.just(new Result<LocationPricing>());          

            //declare the call to get the location profile                                            
            Mono<LocationProfile> locationProfileMono = Mono.just(new LocationProfile());             

            //in the zip call all the services actually are executed                                  
            return Mono.zip(customerMono, result, locationProfileMono);                               
        })                                                                                            
        .map(LocationProfileInfo::new)                                                                
;                                                                                                     

}