Combining mono with flux

Question

I have simple task. I want to add to order products.

public Mono<OrderDTO> addProductsToOrder(String orderId, String[] productIds)
{

    final List<String> productIdsList = Arrays.asList(productIds);

    final Mono<Order> order = orderRepository.findById(orderId)
                                             .switchIfEmpty(orderShouldExists(orderId));

    final Flux<BaseProductInfo> products = fromIterable(productIdsList)
        .flatMap(productId -> productRepository.findById(productId)
                                               .switchIfEmpty(productShouldExists(productId)));

    return order.then(combineLatest(order, products, Tuples::of)
                          .map(objects -> objects.getT1().addProduct(objects.getT2()))
                          .take(productIdsList.size())
                          .last()
                          .flatMap(orderRepository::save)
                          .map(orderMapper::mapToDTO));
}

What I also want to achive is to return updated order. Code below doesn't work as I would like to. Sometimes all products are saved, some time none of them. The database I am using is reactive mongo.

Answer 1

I solved problem with such solution.

public Mono<OrderDTO> addProductsToOrder(String orderId, String[] productIds)
{
    return orderRepository.findById(orderId)
                          .switchIfEmpty(orderShouldExists(orderId))
                          .flatMap(order -> Flux
                                       .fromArray(productIds)
                                       .flatMap(productId -> productRepository.findById(productId)
                                                                              .switchIfEmpty(productShouldExists(productId))
                                               )
                                       .doOnNext(order::addProduct)
                                       .then(Mono.just(order))
                                  )
                          .flatMap(orderRepository::save)
                          .map(orderMapper::mapToDTO);
}

Answer 2

From your code, it looks like you are trying to add a list of BaseProductInfo to the Order and save. You can achieve this by doing something like:

    public Mono<OrderDTO> addProductsToOrder(String orderId, String[] productIds) {

    final List<String> productIdsList = Arrays.asList(productIds);

    final Mono<Order> order = orderRepository.findById(orderId)
                                             .switchIfEmpty(orderShouldExists(orderId));

    final Mono<List<BaseProductInfo>> products = fromIterable(productIdsList)
        .flatMap(productId -> productRepository.findById(productId)
                                               .switchIfEmpty(productShouldExists(productId)))
.collectList();

    return order.zipWith(products).map(tuple -> tuple.getT1().addProducts(tuple.getT2)).flatMap(orderRepository::save)
                          .map(orderMapper::mapToDTO));
}