Java Stream `generate ()` как «включить» первый «исключенный» элемент

Question

Предположите этот сценарий использования для потока Java, где данные добавляются из источника данных. Источником данных может быть список значений, как в примере ниже, или API-интерфейс REST с разбивкой по страницам. В данный момент это не имеет значения.

import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class Main {
    public static void main(String[] args) {
        final List<Boolean> dataSource = List.of(true, true, true, false, false, false, false);
        final AtomicInteger index = new AtomicInteger();
        Stream
            .generate(() -> {
                boolean value = dataSource.get(index.getAndIncrement());
                System.out.format("--> Executed expensive operation to retrieve data: %b\n", value);
                return value;
            })
            .takeWhile(value -> value == true)
            .forEach(data -> System.out.printf("--> Using: %b\n", data));
    }
}

Если вы запустите этот код, ваш вывод будет

--> Executed expensive operation to retrieve data: true
--> Using: true
--> Executed expensive operation to retrieve data: true
--> Using: true
--> Executed expensive operation to retrieve data: true
--> Using: true
--> Executed expensive operation to retrieve data: false

Как вы можете видеть, последний элемент, который оценивается как false, не был добавлен в поток, как ожидалось.

Теперь предположим, что метод generate() загружает страницы данных из API REST. В этом случае значение true/false является значением на странице N, указывающим, существует ли страница N + 1, что-то вроде поля has_more. Теперь я хочу, чтобы последняя страница, возвращенная API, была добавлена ​​в поток, но я не хочу выполнять другую дорогостоящую операцию для чтения пустой страницы, потому что я уже знаю, что страниц больше нет.

Какой самый идиоматичный способ сделать это с помощью API Java Stream? Каждый обходной путь, который я могу придумать, требует выполнения вызова API.

---

UPDATE

В дополнение к подходам, перечисленным в Включая takeWhile () для потоков , есть еще один уродливый способ добиться этого.

public static void main(String[] args) {
    final List<Boolean> dataSource = List.of(true, true, true, false, false, false, false);
    final AtomicInteger index = new AtomicInteger();
    final AtomicBoolean hasMore = new AtomicBoolean(true);
    Stream
        .generate(() -> {
            if (!hasMore.get()) {
                return null;
            }
            boolean value = dataSource.get(index.getAndIncrement());
            hasMore.set(value);
            System.out.format("--> Executed expensive operation to retrieve data: %b\n", value);
            return value;
        })
        .takeWhile(Objects::nonNull)
        .forEach(data -> System.out.printf("--> Using: %b\n", data));
}

Answer 1

Вы используете неправильный инструмент для своей работы. Как уже было заметно в вашем примере кода, Supplier, переданный Stream.generate, должен идти на все, чтобы поддерживать индекс, необходимый для извлечения страниц.

Что еще хуже, так это то, что Stream.generate создает неупорядоченный поток :

Возвращает бесконечный последовательный неупорядоченный поток, в котором каждый элемент генерируется предоставленным Поставщиком. Это подходит для генерации постоянных потоков, потоков случайных элементов и т. Д.

Вы не возвращаете постоянные или случайные значения, а также все, что не зависит от порядка.

Это оказывает существенное влияние на семантику takeWhile:

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

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

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

На практике вы вряд ли столкнетесь с такими эффектами для последовательного потока, но это не меняет того факта, что Stream.generate(…) .takeWhile(…) семантически неверен для вашей задачи.

---

Исходя из вашего примера кода, я заключаю, что страницы не содержат ни своего собственного номера, ни метода "getNext", поэтому мы должны сохранить номер и состояние "hasNext" для создания потока.

Предполагается пример настройки, такой как

class Page {
    private String data;
    private boolean hasNext;

    public Page(String data, boolean hasNext) {
        this.data = data;
        this.hasNext = hasNext;
    }

    public String getData() {
        return data;
    }

    public boolean hasNext() {
        return hasNext;
    }

}

private static String[] SAMPLE_PAGES = { "foo", "bar", "baz" };
public static Page getPage(int index) {
    Objects.checkIndex(index, SAMPLE_PAGES.length);
    return new Page(SAMPLE_PAGES[index], index + 1 < SAMPLE_PAGES.length);
}

Вы можете реализовать правильный поток, такой как

Stream.iterate(Map.entry(0, getPage(0)), Objects::nonNull,
        e -> e.getValue().hasNext()? Map.entry(e.getKey()+1, getPage(e.getKey()+1)): null)
    .map(Map.Entry::getValue)
    .forEach(page -> System.out.println(page.getData()));

Обратите внимание, что Stream.iterate создает заказанный поток :

Возвращает последовательный упорядоченный поток, созданный итеративным применением данной следующей функции к начальному элементу, при условии выполнения заданного предиката hasNext.

Конечно, было бы намного проще, если бы страница знала свой номер, например,

Stream.iterate(getPage(0), Objects::nonNull,
               p -> p.hasNext()? getPage(p.getPageNumber()+1): null)
    .forEach(page -> System.out.println(page.getData()));

или если существовал метод для перехода с существующей страницы на следующую страницу, например,

Stream.iterate(getPage(0), Objects::nonNull, p -> p.hasNext()? p.getNextPage(): null)
    .forEach(page -> System.out.println(page.getData()));