Обратные вызовы жизненного цикла Java Stream

Question

Существует ли элегантный способ регистрации обратных вызовов, когда последний элемент потока был обработан, и поток полностью «потребляется»?

Особенно, когда источник потока (например, курсор БД, Iterator<T> или пользовательский Supplier<T>) конечен и может явно знать, когда данных больше нет.

Пример:

public Stream<Row> query(String sql){
   Connection c = dataSource.openConnection();
   Stream<Row> rows = MyDB.query(sql).stream();
   c.close();
   return rows;
}

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

Я знаю, что есть onClose() API, но это зависит от потребителей, которые явно вызывают close() в потоке.

Answer 1

Вызов onClose и документ, подтверждающий, что возвращаемый поток должен быть закрыт вызывающим абонентом.

/**
 * The returned stream must be closed.
 */
public Stream<Row> query(String sql){
    Connection c = dataSource.openConnection();
    return MyDB.query(sql).stream().onClose(() -> {
        try {
            c.close();
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    });
}

Answer 2

Вы хотите зарегистрировать обработчик закрытия потока :

public Stream<Row> query(String sql){
    Connection c = dataSource.openConnection();
    return MyDB.query(sql).stream().onClose(() -> c.close());
}

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

(Примечание: я пропустилобработка исключений здесь.)

Answer 3

Утилита, которая работает в основном, за исключением случаев, когда возвращаемый поток не повторяется до конца:

public <T> Stream<T> wrap(Stream<T> stream, Runnable onEnd) {
    final Object endSignal = new Object();
    return Stream.concat(stream, Stream.of(endSignal))
            .peek(i -> {
                if(i == endSignal){onEnd.run();}
            })
            .filter(i -> i != endSignal)
            .map(i -> (T) i);
}