[FIXED] Wie kann ich eine Mono (als Nebeneffekt?) von Mono.subscribe() zurückgeben?

Ausgabe

Ich habe den folgenden Code, der “funktioniert” … so weit. Mit “funktioniert” meine ich, dass das Flux<DemoPOJO>von zurückgegeben service.getAll()wird und das ” hasElements().subscribe(this::foo)” dazu führt , dass eine foo()Ausgabe generiert wird, die korrekt widerspiegelt, ob das Flux<DemoPOJO>irgendwelche Elemente hat.

Der gewünschte Endzustand besteht darin, ein ServerResponseObjekt zurückzugeben, das Flux<DemoPOJO>die umschließt, was widerspiegelt, ob der zurückgegebene Flux leer oder “hasElements” ist.

Mein Problem ist, dass Mono.subscribe()a zurückgegeben reactor.core.Disposablewird, und ich möchte irgendwie zu a gelangen Mono<ServerResponse>. Oder bin ich “auf dem falschen Baum gebellt”?

Anmerkung hinzufügen: Ich habe einige Beispiele mit gesehen Flux.flatMap(), aber dies scheint problematisch zu sein, wenn der zurückgegebene Flux viele Elemente enthält (dh die Überprüfung hasElements()scheint viel besser zu sein, als möglicherweise alle Elemente flach abzubilden).

@Component
public class DemoPOJOHandler {

    public static final String PATH_VAR_ID = "id";

    @Autowired
    private DemoPOJOService service;

    public Mono<ServerResponse> getAll(ServerRequest request) {
        Mono<ServerResponse> response = null;
        Flux<DemoPOJO>       entities = service.getAll();

        entities.hasElements().subscribe(this::foo);
        // just return something, for now
        return ServerResponse.ok().build();
    }

    private Mono<ServerRequest> foo(Boolean hasElements) {
        System.out.println("DEBUG >> Mono has elements -> " + hasElements);
        return Mono.empty();
    }
}

Hier ist die DemoPOJOService-Implementierung …

@Component
public class DemoPOJOService {

    @Autowired
    private DemoPOJORepo demoPOJORepo;

    public Flux<DemoPOJO> getAll() {
        return Flux.fromArray(demoPOJORepo.getAll());
    }

    // more implementation, omitted for brevity
}

Und hier ist die DemoPOJORepo-Implementierung …

@Component
public class DemoPOJORepo {

    private static final int NUM_OBJS =20;

    private static DemoPOJORepo demoRepo = null;

    private Map<Integer, DemoPOJO> demoPOJOMap;

    private DemoPOJORepo() {
        initMap();
    }

    public static DemoPOJORepo getInstance() {
        if (demoRepo == null) {
            demoRepo = new DemoPOJORepo();
        }
        return demoRepo;
    }

    public DemoPOJO[] getAll() {
        return demoPOJOMap.values().toArray(new DemoPOJO[demoPOJOMap.size()]);
    }

    // more implementation, omitted for brevity

    private void initMap() {
        demoPOJOMap = new TreeMap<Integer, DemoPOJO>();

        for(int ndx=1; ndx<( NUM_OBJS + 1 ); ndx++) {
            demoPOJOMap.put(ndx, new DemoPOJO(ndx, "foo_" + ndx, ndx+100));
        }
    }
}

Lösung

@SoCal Ihre Antwort scheint zu funktionieren, leidet jedoch unter einem Nachteil: getAll()DB-Aufruf wird zweimal getätigt.

Die Schwierigkeit besteht darin, dass Sie sich erst für den Statuscode entscheiden können, wenn Sie mit dem Empfang von Daten begonnen haben.

Aber da Sie die asynchrone Natur des Körpers anscheinend nicht wirklich benötigen (Sie streamen nicht die einzelnen Elemente, sondern erzeugen eine One-Shot-JSON-Antwort), könnten Sie in diesem Fall den gesamten Ergebnissatz sammeln und einer Antwort zuordnen .

Rufen Sie also die DB auf, sammeln Sie die Elemente in einer Mono<List>, mapdie zu A) einer leeren 404-Antwort, wenn die Liste leer ist, oder B) einer 200 erfolgreichen JSON-Antwort andernfalls (beachten Sie die Verwendung von syncBody):

@Component
public class DemoPOJOHandler {

    public static final String PATH_VAR_ID = "id";

    @Autowired
    private DemoPOJOService service;

    public Mono<ServerResponse> getAll(ServerRequest request) {
        Flux<DemoPOJO> entities = service.getAll();
        Mono<List<DemoPOJO>> collected = entities.collectList();

        return collected.map(list -> list.isEmpty() ? 
            ServerResponse.noContent().build() :
            ServerResponse.ok()
                .contentType(MediaType.APPLICATION_JSON)
                .syncBody(list)
        );
    }
}

Randnotiz: Ich denke , dass ResponseEntitydies der bevorzugte Typ für annotierte Controller ServerResponseist, siehe https://docs.spring.io/spring/docs/current/spring-framework-reference/web-reactive.html#webflux-ann-responseentity .

Beantwortet von – Simon Baslé

Antwort geprüft von – Robin (FixError Admin)