Получайте элементы из одной полезной нагрузки с помощью Flux

У меня есть метод, который запрашивает удаленную службу. Эта служба возвращает одну полезную нагрузку, содержащую множество элементов.

Как получить эти элементы с помощью Flux и flatMapMany?

На данный момент мой метод «выборки из службы» выглядит так:

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

стек — это просто POJO, который выглядит так:

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

Здесь ничего особенного, но я думаю, что мой десериализатор неверен:

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();
}

Что я заметил, так это то, что первый объект сериализуется правильно, но затем он, кажется, снова сериализуется, а не следующий объект в полезной нагрузке.

Полезная нагрузка соответствует стандартной спецификации JSON API и выглядит так:

{  
   "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
         }
      }
   ]
}

Я создал этот шаблон на основе весенний реактивный университет

Любая помощь относительно того, где я ошибся, была бы потрясающей;

Ваше здоровье!


spring flux spring-webflux project-reactor
person Chris Turner    schedule 25.01.2018    source источник
comment
этот веб-сервис возвращает поток или только одну сущность?   -  person pvpkiran    schedule 26.01.2018

Ответы (2)


arrow_upward
2
arrow_downward

Я думаю, что решил это, все еще используя 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);
}

Преобразует входящую полезную нагрузку в String, где я могу затем проанализировать ее с помощью библиотеки jsonapi

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);
}

Который возвращает Flux. Благодаря библиотеке мне тоже не нужно создавать кучу доменов, я по-прежнему могу работать со своим простым 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;
}

А это в свою очередь вызывается из контроллера

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

Я еще не проверял, блокируется это или нет, но, похоже, работает нормально.

person Chris Turner    schedule 26.01.2018

arrow_upward
0
arrow_downward

Ваш json не совсем соответствует вашему классу модели, то есть стеку. Вместе со стеком создайте еще один класс, подобный этому

public class Data {
  List<Stack> data;
   // Getters and Setters.... 
}

Теперь в вашем веб-клиенте вы можете сделать так

Mono<Data> listMono = webClient
                .get()
                .uri("/product/projects")
                .exchange()
                .flatMap(clientResponse -> clientResponse.bodyToMono(Data.class));

Теперь, если вы сделаете listMono.block(), вы получите объект Data, в котором будут все объекты Stack.

person pvpkiran    schedule 26.01.2018
comment
Спасибо за ответ, попробую и отчитаюсь. Можете ли вы объяснить, почему пример, который я привел, работает с потоком? Он также получает одну полезную нагрузку от github. Кроме того, из всего, что я читал, вы не должны блокировать ничего, кроме тестов? - person Chris Turner; 26.01.2018
comment
Что касается блока, вы правы, не рекомендуется делать блок при выполнении реактивного программирования. Но в тестах нет другого способа выполнить код без блока. Что касается того, почему он получает один элемент в полезной нагрузке, хммм, это немного сложно, нужно исследовать больше. Но да, странно, почему он выбирает первый элемент. - person pvpkiran; 26.01.2018