Ограничить поток предикатом

Существует ли потоковая операция Java 8, которая ограничивает (потенциально бесконечный) Streamдо тех пор, пока первый элемент не будет соответствовать предикату?

В Java 9 мы можем использовать, takeWhileкак в примере ниже, чтобы напечатать все числа меньше 10.

IntStream
    .iterate(1, n -> n + 1)
    .takeWhile(n -> n < 10)
    .forEach(System.out::println);

Поскольку в Java 8 такой операции нет, каков наилучший способ ее реализации в общем виде?

Такая операция должна быть возможной с Java 8 Stream, но она не обязательно может быть выполнена эффективно - например, вы не можете обязательно распараллелить такую ​​операцию, поскольку вы должны смотреть на элементы по порядку.

API не предоставляет простой способ сделать это, но, вероятно, самый простой способ - это взять Stream.iterator(), обернуть, Iteratorчтобы иметь реализацию "take-while", а затем вернуться к a Spliteratorи затем a Stream. Или - может быть - обернуть Spliterator, хотя это не может быть больше разделено в этой реализации.

Вот непроверенная реализация takeWhileна Spliterator:

static <T> Spliterator<T> takeWhile(
    Spliterator<T> splitr, Predicate<? super T> predicate) {
  return new Spliterators.AbstractSpliterator<T>(splitr.estimateSize(), 0) {
    boolean stillGoing = true;
    @Override public boolean tryAdvance(Consumer<? super T> consumer) {
      if (stillGoing) {
        boolean hadNext = splitr.tryAdvance(elem -> {
          if (predicate.test(elem)) {
            consumer.accept(elem);
          } else {
            stillGoing = false;
          }
        });
        return hadNext && stillGoing;
      }
      return false;
    }
  };
}

static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> predicate) {
   return StreamSupport.stream(takeWhile(stream.spliterator(), predicate), false);
}

Операции takeWhileи dropWhileбыли добавлены в JDK 9. Ваш пример кода

IntStream
    .iterate(1, n -> n + 1)
    .takeWhile(n -> n < 10)
    .forEach(System.out::println);

будет вести себя точно так, как вы ожидаете, при компиляции и запуске под JDK 9.

JDK 9 был выпущен. Он доступен для скачивания здесь: http://jdk.java.net/9/

allMatch()является функцией короткого замыкания, так что вы можете использовать ее, чтобы остановить обработку. Основным недостатком является то, что вы должны сделать свой тест дважды: один раз, чтобы увидеть, нужно ли его обрабатывать, и еще раз, чтобы увидеть, стоит ли продолжать.

IntStream
    .iterate(1, n -> n + 1)
    .peek(n->{if (n<10) System.out.println(n);})
    .allMatch(n->n < 10);

В качестве продолжения ответа @StuartMarks . В моей библиотеке StreamEx есть takeWhileоперация, совместимая с текущей реализацией JDK-9. При работе под JDK-9 он просто делегирует реализацию JDK (благодаря MethodHandle.invokeExactкоторой это действительно быстро). При работе под JDK-8 будет использоваться реализация «polyfill». Таким образом, используя мою библиотеку, проблема может быть решена так:

IntStreamEx.iterate(1, n -> n + 1)
           .takeWhile(n -> n < 10)
           .forEach(System.out::println);

takeWhileявляется одной из функций, предоставляемых библиотекой protonpack .

Stream<Integer> infiniteInts = Stream.iterate(0, i -> i + 1);
Stream<Integer> finiteInts = StreamUtils.takeWhile(infiniteInts, i -> i < 10);

assertThat(finiteInts.collect(Collectors.toList()),
           hasSize(10));

Обновление: Java 9 Streamтеперь поставляется с методом takeWhile .

Нет необходимости для взлома или других решений. Просто используйте это!

Я уверен, что это может быть значительно улучшено: (кто-то может сделать это потокобезопасным, может быть)

Stream<Integer> stream = Stream.iterate(0, n -> n + 1);

TakeWhile.stream(stream, n -> n < 10000)
         .forEach(n -> System.out.print((n == 0 ? "" + n : "," + n)));

Взломать наверняка ... Не элегантно - но это работает ~: D

class TakeWhile<T> implements Iterator<T> {

    private final Iterator<T> iterator;
    private final Predicate<T> predicate;
    private volatile T next;
    private volatile boolean keepGoing = true;

    public TakeWhile(Stream<T> s, Predicate<T> p) {
        this.iterator = s.iterator();
        this.predicate = p;
    }

    @Override
    public boolean hasNext() {
        if (!keepGoing) {
            return false;
        }
        if (next != null) {
            return true;
        }
        if (iterator.hasNext()) {
            next = iterator.next();
            keepGoing = predicate.test(next);
            if (!keepGoing) {
                next = null;
            }
        }
        return next != null;
    }

    @Override
    public T next() {
        if (next == null) {
            if (!hasNext()) {
                throw new NoSuchElementException("Sorry. Nothing for you.");
            }
        }
        T temp = next;
        next = null;
        return temp;
    }

    public static <T> Stream<T> stream(Stream<T> s, Predicate<T> p) {
        TakeWhile tw = new TakeWhile(s, p);
        Spliterator split = Spliterators.spliterator(tw, Integer.MAX_VALUE, Spliterator.ORDERED);
        return StreamSupport.stream(split, false);
    }

}

Вы можете использовать java8 + rxjava .

import java.util.stream.IntStream;
import rx.Observable;


// Example 1)
IntStream intStream  = IntStream.iterate(1, n -> n + 1);
Observable.from(() -> intStream.iterator())
    .takeWhile(n ->
          {
                System.out.println(n);
                return n < 10;
          }
    ).subscribe() ;


// Example 2
IntStream intStream  = IntStream.iterate(1, n -> n + 1);
Observable.from(() -> intStream.iterator())
    .takeWhile(n -> n < 10)
    .forEach( n -> System.out.println(n));

На самом деле есть два способа сделать это в Java 8 без каких-либо дополнительных библиотек или с помощью Java 9.

Если вы хотите напечатать числа от 2 до 20 на консоли, вы можете сделать это:

IntStream.iterate(2, (i) -> i + 2).peek(System.out::println).allMatch(i -> i < 20);

или

IntStream.iterate(2, (i) -> i + 2).peek(System.out::println).anyMatch(i -> i >= 20);

Выход в обоих случаях:

2
4
6
8
10
12
14
16
18
20

Никто не упомянул что- либо еще. Это причина этого поста.

Это источник, скопированный из JDK 9 java.util.stream.Stream.takeWhile (Predicate). Небольшая разница для работы с JDK 8.

static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> p) {
    class Taking extends Spliterators.AbstractSpliterator<T> implements Consumer<T> {
        private static final int CANCEL_CHECK_COUNT = 63;
        private final Spliterator<T> s;
        private int count;
        private T t;
        private final AtomicBoolean cancel = new AtomicBoolean();
        private boolean takeOrDrop = true;

        Taking(Spliterator<T> s) {
            super(s.estimateSize(), s.characteristics() & ~(Spliterator.SIZED | Spliterator.SUBSIZED));
            this.s = s;
        }

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            boolean test = true;
            if (takeOrDrop &&               // If can take
                    (count != 0 || !cancel.get()) && // and if not cancelled
                    s.tryAdvance(this) &&   // and if advanced one element
                    (test = p.test(t))) {   // and test on element passes
                action.accept(t);           // then accept element
                return true;
            } else {
                // Taking is finished
                takeOrDrop = false;
                // Cancel all further traversal and splitting operations
                // only if test of element failed (short-circuited)
                if (!test)
                    cancel.set(true);
                return false;
            }
        }

        @Override
        public Comparator<? super T> getComparator() {
            return s.getComparator();
        }

        @Override
        public void accept(T t) {
            count = (count + 1) & CANCEL_CHECK_COUNT;
            this.t = t;
        }

        @Override
        public Spliterator<T> trySplit() {
            return null;
        }
    }
    return StreamSupport.stream(new Taking(stream.spliterator()), stream.isParallel()).onClose(stream::close);
}

Вот версия, сделанная для ints - как задано в вопросе.

Использование:

StreamUtil.takeWhile(IntStream.iterate(1, n -> n + 1), n -> n < 10);

Вот код для StreamUtil:

import java.util.PrimitiveIterator;
import java.util.Spliterators;
import java.util.function.IntConsumer;
import java.util.function.IntPredicate;
import java.util.stream.IntStream;
import java.util.stream.StreamSupport;

public class StreamUtil
{
    public static IntStream takeWhile(IntStream stream, IntPredicate predicate)
    {
        return StreamSupport.intStream(new PredicateIntSpliterator(stream, predicate), false);
    }

    private static class PredicateIntSpliterator extends Spliterators.AbstractIntSpliterator
    {
        private final PrimitiveIterator.OfInt iterator;
        private final IntPredicate predicate;

        public PredicateIntSpliterator(IntStream stream, IntPredicate predicate)
        {
            super(Long.MAX_VALUE, IMMUTABLE);
            this.iterator = stream.iterator();
            this.predicate = predicate;
        }

        @Override
        public boolean tryAdvance(IntConsumer action)
        {
            if (iterator.hasNext()) {
                int value = iterator.nextInt();
                if (predicate.test(value)) {
                    action.accept(value);
                    return true;
                }
            }

            return false;
        }
    }
}

Пойдите, чтобы получить библиотеку AbacusUtil . Он предоставляет точный API, который вы хотите, и многое другое:

IntStream.iterate(1, n -> n + 1).takeWhile(n -> n < 10).forEach(System.out::println);

Декларация: я разработчик AbacusUtil.

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

import java.util.Objects;

class ThingProcessor
{
    static Thing returnNullOnCondition(Thing thing)
    {    return( (*** is condition met ***)? null : thing);    }

    void processThings(Collection<Thing> thingsCollection)
    {
        thingsCollection.stream()
        *** regular stream processing ***
        .map(ThingProcessor::returnNullOnCondition)
        .filter(Objects::nonNull)
        *** continue stream processing ***
    }
} // class ThingProcessor

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

Даже у меня было похожее требование - запусти веб-сервис, если не получится, повторите его 3 раза. Если это не удается даже после этих многочисленных испытаний, отправьте уведомление по электронной почте. Погуглив много, anyMatch()пришел как спаситель. Мой пример кода выглядит следующим образом. В следующем примере, если метод webServiceCall возвращает true в самой первой итерации, поток не выполняет дальнейшую итерацию, как мы ее вызвали anyMatch(). Я считаю, это то, что вы ищете.

import java.util.stream.IntStream;

import io.netty.util.internal.ThreadLocalRandom;

class TrialStreamMatch {

public static void main(String[] args) {        
    if(!IntStream.range(1,3).anyMatch(integ -> webServiceCall(integ))){
         //Code for sending email notifications
    }
}

public static boolean webServiceCall(int i){
    //For time being, I have written a code for generating boolean randomly
    //This whole piece needs to be replaced by actual web-service client code
    boolean bool = ThreadLocalRandom.current().nextBoolean();
    System.out.println("Iteration index :: "+i+" bool :: "+bool);

    //Return success status -- true or false
    return bool;
}

Если вы знаете точное количество повторений, которые будут выполнены, вы можете сделать

IntStream
          .iterate(1, n -> n + 1)
          .limit(10)
          .forEach(System.out::println);

    IntStream.iterate(1, n -> n + 1)
    .peek(System.out::println) //it will be executed 9 times
    .filter(n->n>=9)
    .findAny();

вместо пика вы можете использовать mapToObj для возврата конечного объекта или сообщения

    IntStream.iterate(1, n -> n + 1)
    .mapToObj(n->{   //it will be executed 9 times
            if(n<9)
                return "";
            return "Loop repeats " + n + " times";});
    .filter(message->!message.isEmpty())
    .findAny()
    .ifPresent(System.out::println);

Если у вас другая проблема, может потребоваться другое решение, но для вашей текущей проблемы я бы просто сказал:

IntStream
    .iterate(1, n -> n + 1)
    .limit(10)
    .forEach(System.out::println);

Может быть, немного не по теме, но это то, что у нас есть, List<T>а не Stream<T>.

Для начала вам нужно использовать takeметод util. Этот метод принимает первые nэлементы:

static <T> List<T> take(List<T> l, int n) {
    if (n <= 0) {
        return newArrayList();
    } else {
        int takeTo = Math.min(Math.max(n, 0), l.size());
        return l.subList(0, takeTo);
    }
}

это просто работает как scala.List.take

    assertEquals(newArrayList(1, 2, 3), take(newArrayList(1, 2, 3, 4, 5), 3));
    assertEquals(newArrayList(1, 2, 3), take(newArrayList(1, 2, 3), 5));

    assertEquals(newArrayList(), take(newArrayList(1, 2, 3), -1));
    assertEquals(newArrayList(), take(newArrayList(1, 2, 3), 0));

Теперь будет довольно просто написать takeWhileметод, основанный наtake

static <T> List<T> takeWhile(List<T> l, Predicate<T> p) {
    return l.stream().
            filter(p.negate()).findFirst(). // find first element when p is false
            map(l::indexOf).        // find the index of that element
            map(i -> take(l, i)).   // take up to the index
            orElse(l);  // return full list if p is true for all elements
}

это работает так:

    assertEquals(newArrayList(1, 2, 3), takeWhile(newArrayList(1, 2, 3, 4, 3, 2, 1), i -> i < 4));

эта реализация частично повторяет список несколько раз, но не добавляет O(n^2)операций добавления . Надеюсь, что это приемлемо.

У меня есть другое быстрое решение путем реализации этого (который на самом деле нечист, но вы поняли):

public static void main(String[] args) {
    System.out.println(StreamUtil.iterate(1, o -> o + 1).terminateOn(15)
            .map(o -> o.toString()).collect(Collectors.joining(", ")));
}

static interface TerminatedStream<T> {
    Stream<T> terminateOn(T e);
}

static class StreamUtil {
    static <T> TerminatedStream<T> iterate(T seed, UnaryOperator<T> op) {
        return new TerminatedStream<T>() {
            public Stream<T> terminateOn(T e) {
                Builder<T> builder = Stream.<T> builder().add(seed);
                T current = seed;
                while (!current.equals(e)) {
                    current = op.apply(current);
                    builder.add(current);
                }
                return builder.build();
            }
        };
    }
}

Вот моя попытка использовать только библиотеку Java Stream.

        IntStream.iterate(0, i -> i + 1)
        .filter(n -> {
                if (n < 10) {
                    System.out.println(n);
                    return false;
                } else {
                    return true;
                }
            })
        .findAny();