Get items from a single payload using a Flux

Question

I have a method which queries a remote service. This service returns a single payload which holds many items.

How do I get those items out using a Flux and a flatMapMany?

At the moment my "fetch from service" method looks like:

public Flux<Stack> listAll() {
    return this.webClient
            .get()
            .uri("/projects")
            .accept(MediaType.APPLICATION_JSON)
            .exchange()
            .flatMapMany(response -> response.bodyToFlux(Stack.class));
}

a Stack is just a POJO which looks like:

public class Stack {
    String id;
    String name;
    String title;
    String created;
}

Nothing special here, but I think my deserializer is wrong:

protected Stack deserializeObject(JsonParser jsonParser, DeserializationContext deserializationContext, ObjectCodec objectCodec, JsonNode jsonNode) throws IOException {
    log.info("JsonNode {}", jsonNode);

    return Stack.builder()
            .id(nullSafeValue(jsonNode.findValue("id"), String.class))
            .name(nullSafeValue(jsonNode.findValue("name"), String.class))
            .title(nullSafeValue(jsonNode.findValue("title"), String.class))
            .created(nullSafeValue(jsonNode.findValue("created"), String.class))
            .build();
}

What I've noticed happening is the first object is serialized correctly, but then it seems to get serialized again, rather than the next object in the payload.

The payload coming in follows standard JSON API spec and looks like:

{  
   "data":[  
      {  
         "type":"stacks",
         "id":"1",
         "attributes":{  
            "name":"name_1",
            "title":"title_1",
            "created":"2017-03-31 12:27:59",
            "created_unix":1490916479
         }
      },
      {  
         "type":"stacks",
         "id":"2",
         "attributes":{  
            "name":"name_2",
            "title":"title_2",
            "created":"2017-03-31 12:28:00",
            "created_unix":1490916480
         }
      },
      {  
         "type":"stacks",
         "id":"3",
         "attributes":{  
            "name":"name_3",
            "title":"title_3",
            "created":"2017-03-31 12:28:01",
            "created_unix":1490916481
         }
      }
   ]
}   

I've based this pattern on the spring-reactive-university

Any help as to where I've gone wrong would be awesome;

Cheers!

Answer 1

I think I solved it, still using a Flux.

public Flux<Stack> listAllStacks() {
    return this.webClient
            .get()
            .uri("/naut/projects")
            .accept(MediaType.APPLICATION_JSON)
            .exchange()
            .flatMap(response -> response.toEntity(String.class))
            .flatMapMany(this::transformPayloadToStack);
}

Converts the incoming payload to a String where I can then parse it using a jsonapi library

private Flux<Stack> transformPayloadToStack(ResponseEntity<String> payload) {
    ObjectMapper objectMapper = new ObjectMapper();
    ResourceConverter resourceConverter = new ResourceConverter(objectMapper, Stack.class);
    List<Stack> stackList = resourceConverter.readDocumentCollection(payload.getBody().getBytes(), Stack.class).get();

    return Flux.fromIterable(stackList);
}

Which returns a Flux. Thanks to the library, I don't need to create a bunch of domains either, I can still work with my simple Stack POJO

@Data
@NoArgsConstructor
@AllArgsConstructor
@JsonIgnoreProperties(ignoreUnknown = true)
@Type("stacks")
public class Stack {
    @com.github.jasminb.jsonapi.annotations.Id
    String id;
    String name;
    String title;
    String created;
}

And this in turn is called from the controller

@GetMapping("/stacks")
@ResponseBody
public Flux<Stack> findAll() {
    return this.stackService.listAllStacks();
}

I've not tested if this is blocking or not yet, but seems to work okay.