Как из Flux собрать массив JSON?

Question

[Wan-Derer]

Есть такой контроллер (Java, Reactor Netty):

public Publisher<Void> getFilterListByType(HttpServerRequest req, HttpServerResponse resp) {
    return resp
      .status(HttpResponseStatus.OK)
      .addHeader(CONTENT_TYPE, "application/json")
      .addHeader(ACCESS_CONTROL_ALLOW_ORIGIN, "*")     // CORS
      .sendString(
        myService.getObjectsAsFlux()    // получить Flux<MyObj> 
          .map(item -> toJson(item))       // преобразовать в JSON
    );
}

который выдаёт строку вида

{"id": 123, "desc": "OLOLO"} {"id": 456, "desc": "AZAZA"}

т.е. для полноценного массива JSON не хватает квадратных скобок по краям и запятых между объектами.
Лог Netty показывает выдачу чанками (я правильно понимаю что объекты выдаются по мере их готовности и выдачи сервисом?).
Я могу добавить после .map код:

.reduce((a, b) -> a + "," + b)
.map(item -> "[" + item + "]")

и получит нормальный массив JSON. Но в этом случае я вместо Flux получаю Mono, а лог Netty показывает выдачу одним блоком. Т.е. я как бы теряю асинхронность. Или нет?

В общем, как такие вещи делать правильно?
(заглядывать в код Spring Web Flux или Vert.X я боюсь - я там ни в жисть не разберусь!)

Answer 1

[mayton2019]

При чем здесь Flux и Mono к JSON? Это разные уровни смыслов. А если-бы ты c XML работал? Ты явно указываешь getObjectsAsFlux зачит и будел флюкс. А если тебя смущает массив внутри респонса то оберни его объектом и будет тебе всегда объект.

Comments

[Wan-Derer]

Мне надо получить валидный JSON. Т.е. контроллер на вход получает Flux, а выдать должен JSON-массив. И по возможности сохранить асинхронность (т.е. выдавать объекты по мере их готовности).

[mayton2019]

Мне твоя идея не нравится. Тебе нужно либо в каждом флюх-респонсе отдавать независимый документ. И клиент должен об этом знать.

Либо вообще отказаться от реактивности и работать с обычным вебом, формируя длинный JSON-array.

[Wan-Derer]

mayton2019, не очень понятно. Что ещё можно выдать их контроллера? Единственное что я придумал - преобразовать в JSON и отдать как строку. Какие ещё есть варианты?

[mayton2019]

Ты же и так выдаешь его как строку. В чем вопрос?

[Wan-Derer]

mayton2019, Вопрос в том как выдать валидный JSON-массив, сохранив асинхронность. Т.е. чтобы эта строка выдавалась чанками по мере поступления объектов во Flux.

Answer 2

[Dmitry Roo]

Вот мое демо:

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.stream.IntStream;

@Slf4j
@RequiredArgsConstructor
@RestController
public class WebFluxDemo {

    private final ObjectMapper objectMapper;

    @GetMapping("/getMyObjs")
    Flux<MyObj> getMyObjs() {
        return Flux.fromStream(IntStream.range(1, 6).boxed()
                        .map(String::valueOf)
                )
                .map(MyObj::new)
                .delayElements(Duration.ofSeconds(1))
                .log()
                ;
    }

    @GetMapping("/getMyObjsAsString")
    Flux<String> getMyObjsAsString() {
        return Flux.fromStream(IntStream.range(1, 6).boxed()
                        .map(String::valueOf))
                .map(MyObj::new)
                .map(this::mapToObj)
                .delayElements(Duration.ofSeconds(1))
                .log()
                ;
    }

    @SneakyThrows
    private String mapToObj(MyObj myObj) {
        return objectMapper.writeValueAsString(myObj);
    }

    private record MyObj(String id) {
    }
}

1. Валидный массив открывается и закрывается скобками. Т.е. он должен быть закрытым. Если он не закрыт - это не валидный массив.
2. Поэтому, если мы отдаем flux (см. getMyObjs()) то получим валидный массив, но ответ будет отдан только на onComplete. Т.е. все сразу и когда поступят все элементы.
3. Можно отдавать строку - тогда элементы будут отданы по мере поступления, но "валидного массива" никакого не будет.

Comments

[Wan-Derer]

У Flux есть методы .startWith() и .concatWithValues(), позволяющие "обрамить" поток некоторыми значениями. Я могу в своём методе toJson() ставить запятую после закрывающей скобки. Таким образом получится собрать массив, но будет одна лишняя запятая (последняя). Можно решить так:

.startWith("[")
.concatWithValues("{}]")    // или "0]"

Получим валидный массив JSON, без потери асинхронности. Но в конце будет пустой объект (или 0), который надо будет отфильтровать на фронте. А мне не хочется без особой нужды вводить нестандартный контракт (типа "получи массив и отбрось последний элемент").
В общем, фиг его знает как делать правильно. Наверно, придётся-таки сделать демку на Spring и посмотреть (как-то) что именно они выдают, может зря я парюсь на тему асинхронности :) Может, когда она действительно нужна, уже делают на сокетах :)

Answer 3

[Wan-Derer]

Если кому интересно, придумал такой костыль:

Flux<String> items = ... здесь исходный Flux ... ;
Flux<Integer> numbers = Flux.generate(
  () -> 0,
  (state, sink) -> {
    sink.next(state);
    return state + 1;
  }
);

items.zipWith(numbers, (i,n) -> n + i)
  .map(item-> item.replace("0{", "{").replaceFirst("^[0-9]+", ", "))
  .startWith("[")
  .concatWithValues("]")
  .subscribe(System.out::println);

Т.е. мы сначала к каждой строке пририсовываем последовательный номер. Потом первый номер (0) просто затираем, а остальные меняем на запятую. Ну и потом оборачиваем всё квадратными скобками.
Не знаю насколько это хорошо, но работает :)