How to use a Flux inside an object as JSON response?

Question

In my Spring Webflux API gateway I am receiving a Flux from a microservice via REST:

public Flux<MyObject> getMyObjects(String id) {

    Flux<MyObject> myObjects = webClient.get().uri(nextServerUrl + "/myobject" + issueId)
            .accept(MediaType.APPLICATION_JSON)
            .retrieve()
            .bodyToFlux(MyObject.class);

    return myObjects;
}

I have to rearrange the information received by the microservice in the API gateway for the response to the client. I tried to do it in two ways:

1. Use the Flux as far as possible:

    private Rearranged createClientResponse(String id) {
        Rearranged rearranged = new Rearranged();

        Flux<MyObject> myObjects = myObjectService.getMyObjects(id);

        rearranged.setMyObjects(myObjects);

        myObjects.map(myObject -> {
            rearranged.setInfo(myObject.getInfo());

            //set more

            return myObjects;
        });

        return rearranged;
    }

    public class Rearranged {
        private Flux<MyObject> myObjects;
        //more attributes
    }

Result: Following empty object:

{
    "information": null,
    "myObjects": {
        "scanAvailable": true,
        "prefetch": -1
    }
}

2. Block the Flux and work with synchronous objects

    private Rearranged createClientResponse(String id) {
        Rearranged rearranged = new Rearranged();

        List<MyObject> myObjects = myObjectService.getMyObjects(id).collectList().block();

        rearranged.setMyObjects(myObjects);

        rearranged.setInfo(myObjects.get(0).getInfo());

        return rearranged;
    }

    public class Rearranged {
        private List<MyObject> myObjects;
        //more attributes
    }

Result: receiving the exception block()/blockFirst()/blockLast() are blocking which is not supported in thread

What would be the right way to achieve the possibility of rearranging the information from the microservice response to respond to the client? How would I be able to block for the Flux to complete? I understand that a block is possible when I am returning a "synchronous" object (like I am doing but still getting the exception)?

Answer 1

First of all, your model should not countains reactive stream. Use plain object or list.

public class Rearranged {
        private MyObject myObject;       
}

Or

public class Rearranged {
        private List<MyObject> myObjects;       
}

If you block the thread, reactor threads will exhausted in a moments. If your getMyObjects method only receives one object (if not, look at the end of the comment), then you should handle it as a Mono. Then in the createClientResponse, you have to return with Mono<Rearranged> Now you can easily map from one Mono to another using the .map method.

 private Mono<Rearranged> createClientResponse(String id) {

       Mono<MyObject> myObjects = myObjectService.getMyObjects(id);

       return myObjects.map(myObject -> {
           retrun new Rearranged(myObject)
            //create the proper object here
        });

    }

If you need more object, you can use the same method, for example, the collectList() collect the elements from the Flux<> into Mono<List<>>, then the same method can be accepted.