Как проверить, нет ли потока Java 8?

Question

Как я могу проверить, является ли Stream пустым и выдает исключение, если это не так, как операция без терминала?

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

public Stream<Thing> getFilteredThings() { 
    Stream<Thing> stream = getThings().stream() 
       .filter(Thing::isFoo) 
       .filter(Thing::isBar); 
    return nonEmptyStream(stream,() -> { 
     throw new RuntimeException("No foo bar things available") 
    }); 
} 

private static <T> Stream<T> nonEmptyStream(Stream<T> stream, Supplier<T> defaultValue) { 
    List<T> list = stream.collect(Collectors.toList()); 
    if (list.isEmpty()) list.add(defaultValue.get()); 
    return list.stream(); 
} 

Answer 1

Если вы можете жить с ограниченным параллельным capablilities, следующее решение будет работать:

private static <T> Stream<T> nonEmptyStream(
    Stream<T> stream, Supplier<RuntimeException> e) { 

    Spliterator<T> it=stream.spliterator(); 
    return StreamSupport.stream(new Spliterator<T>() { 
     boolean seen; 
     public boolean tryAdvance(Consumer<? super T> action) { 
      boolean r=it.tryAdvance(action); 
      if(!seen && !r) throw e.get(); 
      seen=true; 
      return r; 
     } 
     public Spliterator<T> trySplit() { return null; } 
     public long estimateSize() { return it.estimateSize(); } 
     public int characteristics() { return it.characteristics(); } 
    }, false); 
} 

Вот несколько примеров коды с помощью его:

List<String> l=Arrays.asList("hello", "world"); 
nonEmptyStream(l.stream(),()->new RuntimeException("No strings available")) 
    .forEach(System.out::println); 
nonEmptyStream(l.stream().filter(s->s.startsWith("x")), 
       ()->new RuntimeException("No strings available")) 
    .forEach(System.out::println); 

Проблемы (эффективного) параллельного выполнения что поддерживающее разделение Spliterator требует поточно-безопасного способа узнать, видел ли какой-либо из фрагментов какое-либо значение поточно-безопасным способом. Тогда последний из фрагментов, выполняющих tryAdvance, должен понять, что он является последним (и он также не мог продвинуться), чтобы выбросить соответствующее исключение. Поэтому я не добавил поддержку для разделения здесь.

Answer 2

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

Лучшее, что вы можете сделать, это прекратить поток с помощью операции терминала findAny(), которая остановится, когда найдет какой-либо элемент, но если их нет, ему придется перебирать все входные данные, чтобы найти это.

Это поможет вам только в том случае, если в списке входных данных много элементов, а одно из первых проходит фильтры, так как только небольшое подмножество списка должно быть использовано до того, как вы знаете, что поток не пуст.

Конечно, вам нужно будет создать новый поток для создания выходного списка.

Answer 3

Другие ответы и комментарии верны в том, что для проверки содержимого потока необходимо добавить операцию терминала, тем самым «потребляя» поток. Однако это можно сделать и вернуть результат обратно в поток без буферизации всего содержимого потока. Вот несколько примеров:

static <T> Stream<T> throwIfEmpty(Stream<T> stream) { 
    Iterator<T> iterator = stream.iterator(); 
    if (iterator.hasNext()) { 
     return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, 0), false); 
    } else { 
     throw new NoSuchElementException("empty stream"); 
    } 
} 

static <T> Stream<T> defaultIfEmpty(Stream<T> stream, Supplier<T> supplier) { 
    Iterator<T> iterator = stream.iterator(); 
    if (iterator.hasNext()) { 
     return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator, 0), false); 
    } else { 
     return Stream.of(supplier.get()); 
    } 
} 

В основном превратить поток в Iterator, чтобы вызвать hasNext() на него, и если это правда, включите Iterator обратно в Stream. Это неэффективно, так как все последующие операции над потоком будут проходить через методы Iterator hasNext() и next(), что также подразумевает, что поток эффективно обрабатывается последовательно (даже если он позже стал параллельным). Однако это позволяет тестировать поток без буферизации всех его элементов.

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

Answer 4

После идеи Стюарта, это может быть сделано с Spliterator, как это:

static <T> Stream<T> defaultIfEmpty(Stream<T> stream, Stream<T> defaultStream) { 
    final Spliterator<T> spliterator = stream.spliterator(); 
    final AtomicReference<T> reference = new AtomicReference<>(); 
    if (spliterator.tryAdvance(reference::set)) { 
     return Stream.concat(Stream.of(reference.get()), StreamSupport.stream(spliterator, stream.isParallel())); 
    } else { 
     return defaultStream; 
    } 
} 

Я думаю, что это работает с параллельными потоками, как stream.spliterator() операции прервет поток, а затем восстановить его в соответствии с требованиями

В моем случае использования мне понадобилось значение по умолчанию Stream, а не значение по умолчанию. это довольно легко изменить, если это не то, что вам нужно