Flux into Mono List Object - project reactor

user_6126878 0
Viewed 3797

I'm using project reactor and I've the next issue:

I've one method that return Mono<CustomerResponse> that contains a CustomerDto list, each client has attributes, one of theirs attributes is a payment list. But this payment list is null.

I've another method that receive client id and returns a Flux payment Flux<PaymentDto> for that client.

This are the model

public class CustomerResponse {
    private List<CustomerDto> customers;
}

public class CustomerDto {
    private int id;
    private String fullname;
    private String documentNumber;
    private List<PaymentDto> payments;
}

These are the interfaces

public interface CustomerService {
    public Mono<CustomerResponse> customerSearch(CustomerRequest request);
}
public interface PaymentService {
    public Flux<PaymentDto> getPayments(int clientId);
}

This is my method

public Mono<CustomerResponse> getCustomer(CustomerRequest request) {
    return customerService.customerSearch(request).map(resp -> resp.getCustomers())
            .flatMap(customerList -> {
                List<CustomerDto> newCustomerList = customerList.parallelStream().map(customer -> {
                    Flux<PaymentDto> paymentFlux = 
                            paymentService.getPayments(customer.getId());
                    
                    // Here: java.lang.IllegalStateException: block()/blockFirst()/blockLast()
                    customer.setPayments(paymentFlux.collectList().block());
                    return customer;
                }).collect(Collectors.toList());
                
                return Mono.just(new CustomerResponse(newCustomerList));
            });
}

I've the next exception:

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-4
    at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:83) ~[reactor-core-3.3.6.RELEASE.jar:3.3.6.RELEASE]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException:

I would like to know if there is a non-blocking or optimal way to do it

project-reactor spring-webflux reactive-programming nonblocking java
1 Answers

You can refactor your code like this to avoid blocking call:

  public Mono<CustomerResponse> getCustomer(CustomerRequest request) {
    Flux<CustomerDto> customerDtoFluxEnriched = customerService.customerSearch(request)
        .map(CustomerResponse::getCustomers).flatMapMany(Flux::fromIterable).flatMap(customerDto -> {
          Flux<PaymentDto> paymentFlux = paymentService.getPayments(customerDto.getId());
          Mono<List<PaymentDto>> paymentListMono = paymentFlux.collectList();
          return paymentListMono.map(paymentList -> {
            customerDto.setPayments(paymentList);
            return customerDto;
          });
        });
    return customerDtoFluxEnriched.collectList().map(customerList -> {
      CustomerResponse customerResponse = new CustomerResponse();
      customerResponse.setCustomers(customerList);
      return customerResponse;
    });
  }
user_137851210
Related