Как наполнять Flux по мере поступления данных?

Question

[Wan-Derer]

Допустим, я хочу создать Flux и наполнить его данными, получаемыми, например, и БД. Я могу написать так:

public Flux<Integer> getFlux(){

  List<Integer> values = new ArrayList<>();

  for (int i = 0; i < 10; i++) {
    values.add(someLongTimeProcedure());
  }
  
  return Flux.fromIterable(values);
}

private Integer someLongTimeProcedure() {
  // здесь запрос в БД
  return РЕЗУЛЬТАТ_ЗАПРОСА;
}

Но в этом случае Flux будет создан только после то как будет наполнен List. А как сделать так чтобы Flux был создан заранее, а данные "проталкивались" в него по мере их получения?
Важный момент: кол-во элементов конечно. Т.е. в какой-то момент я должен сказать Flux что всё, данных больше не будет.

Answer 1

[Dmitry Roo]

Для этого есть Sinks:

public static void main(String[] args) throws Exception {
        Sinks.Many<Integer> sinks = Sinks.many()
                .multicast() 
                .onBackpressureBuffer(); // Изучайте документацию. Тут много вариантов

        sinks.asFlux()
                .subscribe(System.out::println);

        // В очереди ничего нет. Ничего не происходит
        Thread.sleep(1000);
        
        // Наполняем очередь элементами
        sinks.tryEmitNext(1);
        sinks.tryEmitNext(2);
        sinks.tryEmitNext(3);
    }

Comments

[Wan-Derer]

Спасибо, почитаю. Пока сделал так:

public Flux<MyObj> getFlux() {

  String query = "select * from............................'";

  return Flux.create(
    sink -> sink.onRequest(r ->
      {
        try (Connection con = dataSource.getConnection()) {
          final Statement statement = con.createStatement();
          final ResultSet rs = statement.executeQuery(query);

          while (rs.next()) {
            sink.next(MyObj.builder()
              // билдим объект
              .build());
          }

          sink.complete();

        } catch (SQLException e) {
          e.printStackTrace();
        } 
      }
    ));
}

Проблема в том что try - catch получается сделать только внутри create(), иначе идут ошибки из-за необработанных исключений. Но мне бы хотелось пробрасывать исключение наружу метода getFlux() чтобы, например, давать 500-ю ошибку внешнему сервису.
Как это можно сделать?

[Dmitry Roo]

Wan-Derer,
1. Пробрасываете исключения наверх как рантайм-эксепшены или с помощью @SneakyThrows (лямбды не могут бросать чекед)
2. У флюкса есть куча методов типа .onError… с помощью которых можно реализовать требуемое поведение.

[Wan-Derer]

Dmitry Roo, Спасибо. Хочу немного пояснить для тех кто будет это читать :)
Во-первых, из лямбды действительно нельзя бросить проверяемое исключение.
Во-вторых, если написать так:

public Flux<MyObj> getFlux() {

  String query = "select * from............................'";

  Flux<MyObj> res =  Flux.create(
    sink -> sink.onRequest(r ->
      {
        try (Connection con = dataSource.getConnection()) {
          final Statement statement = con.createStatement();
          final ResultSet rs = statement.executeQuery(query);

          while (rs.next()) {
            sink.next(MyObj.builder()
              // билдим объект
              .build());
             System.out.println("создаём объект");
          }

          sink.complete();

        } catch (SQLException e) {
          e.printStackTrace();
        } 
      }
    ));

  System.out.println("выходим из метода");

  return res;
}

то мы увидим что выход их метода произошёл раньше чем созданы объекты. Мы собственно за это и боролись - за асинхронность! Мы создали "контейнер", передали его дальше, а по мере поступления новых объектов - докидываем их в "трубу", а на той стороне их сразу получают.
Отсюда важный вывод: мы не можем бросить исключение в этом методе: мы же отсюда уже ушли :)
Но мы можем сделать следующее:

} catch (SQLException e) {
  e.printStackTrace();
  sink.error(e);
// или так:
// sink.error(new Exception("Database Problem. Query: '" + query + "'. Exception: " + e.getMessage()));
}

т.е. "внедрить" в наш поток объектов сведения об ошибке. И на принимающей стороне мы можем их получить:

@Test
public void shouldGetAllMyObj() throws Exception {

  AtomicReference<String> err = new AtomicReference<>();

  final Flux<MyObj> myObjList;
  myObjList = myRepo.getAllMyObj();
  myObjList.subscribe(
    item -> { обработка валидных объектов },
    error -> err.set(error.getMessage())    // здесь ловим ошибку
  );

  System.out.println(err.get());
  if (err.get() != null) throw new Exception(err.get());     // например, можем бросить исключение

}

Добавлю что у sink есть методы для описания того что надо делать при ошибке: завершить, продолжить, сделать повторный запрос и т.д.