how to pass attributes across sequence of reactor calls returing Flux

Question

I am moving code from imperative to reactive java.

In imperative programming we hold important information in DTO/Context and pass along with request, is there a way to achieve same in reactor paradigm using spring-webflux / project-reactor ?

In other words:: how can I maintain required values from more than sequence of non blocking calls.

For instance, Let's say I have Employee and Department mongo collection and used reactive mongo repository to fetch result from database.

Employee :

• empId

• status: ACTIVE | INACTIVE

• empName

• empAge

• managerId

• deptIds

Department:

• deptId

• deptName

Sequence of calls

Flux<Employee> empFlux =  empRepository.findEmployees("ACTIVE"); // (1)
Flux<Department> deptFlux=  deptRepository.findDepartments(deptIds); // (2)


webclient (empId, managerId, deptId, deptName); // (3)

I googled it's seems I can use flatmap to sequnce these flows, but with each transition i will lost information in previous call.

Like in above code:

(1) : get list of ACTIVE employees

(2) : with flatmap by id we get list of eligible departments but lost information about employees.

(3) I need to make reactive rest call to webclient with information about employee , dept.

My question is how can I retain this information across sequence of non-blocking flux and Monos.

Answer 1

You could also use the zipwith operator like this:

Flux<WebClientReturn> infoFlux = empRepository.findEmployees("ACTIVE")
       .zipWith(deptRepository.findDepartments(deptIds), (employee, dept) -> {
          webclient (employee.getempId(), employee.getmanagerId(), dept.getdeptId(), dept.getdeptName());
        })

I think this is the most descriptive and easiest readable way to do it.

Answer 2

For this kind of situations Tuples are useful:

empRepository.findEmployees("ACTIVE")
        .flatMap(emp -> deptRepository.findDepartments(emp.getDeptIds()).map(dept -> Tuples.of(emp,dept)))
        .flatMap(tuple -> webclient(tuple.getT1().getEmpId(), tuple.getT1().getMangagerId(), tuple.getT2().getDeptId(), tuple.getT2().getDeptName()));

IMO thats not very readable, but its also Reactor Way™ for some official operators, like zip.

Answer 3

You need to keep that information in a separate container like an array list what is accessible, and use it latter when required. You may try as follows.

// Declare the arraylist to contain the return response object.
List<Object> returnedObjectList = new ArrayList<>();

Flux<Employee> empFlux =  empRepository.findEmployees("ACTIVE"); // (1)
Flux<Department> deptFlux=  deptRepository.findDepartments(deptIds); // (2)

return Flux.zip(empFlux, deptFlux)
         .flatMap(tuple -> {
             final String empId = tuple.getT1().getEmpId();
             final String managerId = tuple.getT1().getManagerId();

             final String deptId = tuple.getT2().getDeptId();
             final String deptName = tuple.getT2().getDeptName();

             Object returnedObject = webclient(empId, managerId, deptId, deptName); // (3)
             returnedObjectList.add(returnedObject);
             return Flux.just(returnedObject);

          });