ReactiveCosmosRepository is not being invoked with webFlux and netty

Question

I have the following reactive repository:

@Repository
public interface FooCosmosRepository extends ReactiveCosmosRepository<Foo, String> {
}

I am using it as following:

    @Override
    public Mono<FooResponse> getFooDetails() {

        FooResponse fooResponse = new FooResponse();
        fooResponse.setCount(1000);

        List<Foo> fooList = new ArrayList<>();

        repository.findAll().collectList().flatMap(e ->{

       //This is not invoked. findAll return Flux<T> in this case Flux<Foo>
            for (Foo foo : e) {
                fooList.add(foo);
            }
            return null;
        });

        fooResponse.setFooList(fooList);

        return Mono.just(fooResponse);
    }

FooResponse is defined as follows:

@NoArgsConstructor
@Data
@FieldDefaults(level = AccessLevel.PRIVATE)
public class FooResponse {
    int rowCount;
    List<Foo> fooList;
}

I cant block cause I get error Iterating over a toIterable() / toStream() is blocking, which is not supported in thread reactor-http-nio-6

I cant return Flux<T> from the method also. I need to return Mono<FooResponse>. How can i query repository, actually get/collect response and add to the list?

Any ideas?

Answer 1

It's because you are coding imperatively instead of reactively. and you are breaking the chain which means that Reactor can't complete the assembly phase, and then execute during the subscription phase.

@Override
public Mono<FooResponse> getFooDetails() {
    return repository.findAll()
        .collectList()
        .map(list -> {
            FooResponse fooResponse = new FooResponse();
            fooResponse.setCount(1000);
            fooResponse.setList(list);
            return fooResponse;
        });
    }

This is basic reactor and I recommend the following links:

Reactor Core Features

Flight of the flux