Spring Reactive: manipulate a flux into a mono type response

Question

I'm fairly new to Spring Reactive.

I want to transform a Spring Reactive Repo Flux response into a Mono type response of another type:

@Transactional(readOnly = true)
@Override
public Mono<BaseListResponseVm<AcsTransactionVm>> transactionsBy(
        @NonNull Integer offset,
        @NonNull Integer limit,
        @NonNull String term,
        AcsTransactionRequestDto dto
) {
    var response = new BaseListResponseVm<AcsTransactionVm>();
    var list = repository.findByPan(term, PageRequest.of(offset, limit));

    return repository.findByPan(term, PageRequest.of(offset, limit))
            XXXXX.(res -> {
                response.setItems(); // List<AcsTransactionVm>
                response.setCount(); // Long
            });
}

Repository

public interface AcsTransactionRepository extends ReactiveCassandraRepository<AcsTransactionDao, Integer> {

    @AllowFiltering
    Flux<AcsTransactionDao> findByPan(String term, Pageable page);

}

I've tried the operators that are provided (transform being the closes that I need, but still it provides every single param of the repo result as reactive one), but couldn't end up with anything that I would know how how to use.

I am not sure what do you mean by transform to Mono of another type, so you have a Flux<A> and want to transform to Mono<B> ? and if this is the question what is the relation between A and B? is B = List<A> ? — JArgente

Deniss M. Deniss M. · Accepted Answer · 2021-06-16T13:06:16

This appears to be the way to do it:

@Transactional(readOnly = true)
@Override
public Mono<BaseListResponseVm<AcsTransactionVm>> transactionsBy(
        @NonNull Integer offset,
        @NonNull Integer limit,
        @NonNull String term,
        AcsTransactionRequestDto dto
) {
    log.debug("DB: fetching ACS transactions using offset {}, limit {}, term {} and request {}", offset, limit, term, dto);
    var response = new BaseListResponseVm<AcsTransactionVm>();

    return repository.findByPan(term, PageRequest.of(offset, limit))
            .collectList().map(transactions -> {
                var mappedItems = transactions.stream()
                        .map(mapper::toVm)
                        .collect(Collectors.toList());
                response.setItems(mappedItems);

                return response;
            });
}

Spring Reactive: manipulate a flux into a mono type response

1 Answers