Limitar um fluxo por um predicado

Existe uma operação de fluxo do Java 8 que limita um (potencialmente infinito) Streamaté que o primeiro elemento falhe ao corresponder a um predicado?

No Java 9, podemos usar takeWhilecomo no exemplo abaixo para imprimir todos os números menores que 10.

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

Como não existe essa operação no Java 8, qual é a melhor maneira de implementá-la de maneira geral?

Essa operação deve ser possível com um Java 8 Stream, mas não pode necessariamente ser feita com eficiência - por exemplo, você não pode necessariamente paralelizar uma operação, pois é necessário examinar os elementos em ordem.

A API não fornece uma maneira fácil de fazê-lo, mas provavelmente a maneira mais simples é seguir Stream.iterator(), agrupar Iteratorpara ter uma implementação "demorada" e, em seguida, voltar para ae Spliteratordepois a Stream. Ou - talvez - encapsule o Spliterator, embora ele realmente não possa mais ser dividido nesta implementação.

Aqui está uma implementação não testada de takeWhileem um 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);
}

Operações takeWhilee dropWhileforam adicionados ao JDK 9. Seu código de exemplo

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

se comportará exatamente como você espera quando compilado e executado no JDK 9.

O JDK 9 foi lançado. Está disponível para download aqui: http://jdk.java.net/9/

allMatch()é uma função de curto-circuito, para que você possa usá-la para interromper o processamento. A principal desvantagem é que você deve fazer o teste duas vezes: uma vez para ver se deve processá-lo e outra vez para continuar.

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

Como acompanhamento da resposta do @StuartMarks . Minha biblioteca StreamEx possui a takeWhileoperação que é compatível com a implementação atual do JDK-9. Quando executado no JDK-9, ele apenas delega para a implementação do JDK (através da MethodHandle.invokeExactqual é realmente rápido). Ao executar no JDK-8, a implementação "polyfill" será usada. Portanto, usando minha biblioteca, o problema pode ser resolvido assim:

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

takeWhileé uma das funções fornecidas pela biblioteca 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));

Atualização: o Java 9 Streamagora vem com um takeWhile método .

Não há necessidade de hacks ou outras soluções. Apenas use isso!

Estou certo de que isso pode ser bastante aprimorado: (talvez alguém possa torná-lo seguro para threads)

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

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

Um hack com certeza ... Não é elegante - mas funciona ~: 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);
    }

}

Você pode usar o 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));

Na verdade, existem 2 maneiras de fazer isso no Java 8 sem bibliotecas extras ou usando o Java 9.

Se você quiser imprimir números de 2 a 20 no console, faça o seguinte:

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

ou

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

A saída é nos dois casos:

2
4
6
8
10
12
14
16
18
20

Ninguém mencionou anyMatch ainda. Esta é a razão para este post.

Essa é a fonte copiada do JDK 9 java.util.stream.Stream.takeWhile (Predicate). Uma pequena diferença para trabalhar com o 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);
}

Aqui está uma versão feita em ints - conforme solicitado na pergunta.

Uso:

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

Aqui está o código para o 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;
        }
    }
}

Vá para a biblioteca AbacusUtil . Ele fornece a API exata que você deseja e muito mais:

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

Declaração: Sou o desenvolvedor do AbacusUtil.

Você não pode abortar um fluxo, exceto por uma operação de terminal em curto-circuito, o que deixaria alguns valores de fluxo não processados, independentemente de seu valor. Mas se você quiser apenas evitar operações em um fluxo, poderá adicionar uma transformação e um filtro ao fluxo:

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

Isso transforma o fluxo de coisas em nulos quando as coisas atendem a alguma condição e depois filtra nulos. Se você estiver disposto a entrar em efeitos colaterais, poderá definir o valor da condição como verdadeiro assim que alguma coisa for encontrada, para que todas as coisas subsequentes sejam filtradas independentemente do seu valor. Mas, se não, você pode economizar muito (se não todo) o processamento, filtrando valores do fluxo que não deseja processar.

Até eu estava tendo um requisito semelhante - invoque o serviço da Web, se falhar, tente novamente três vezes. Se falhar, mesmo após essas várias tentativas, envie uma notificação por email. Depois de pesquisar muito, anyMatch()veio como um salvador. Meu código de exemplo da seguinte maneira. No exemplo a seguir, se o método webServiceCall retornar true na primeira iteração, o fluxo não iterará mais como chamamos anyMatch(). Eu acredito que é isso que você está procurando.

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;
}

Se você souber a quantidade exata de repetições que serão executadas, poderá fazer

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();

em vez de pico, você pode usar o mapToObj para retornar o objeto ou mensagem final

    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);

Se você tiver um problema diferente, uma solução diferente pode ser necessária, mas para o seu problema atual, eu simplesmente diria:

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

Pode ser um pouco estranho, mas é para isso que temos, List<T>e nãoStream<T> .

Primeiro você precisa ter um takemétodo util. Este método leva os primeiros nelementos:

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);
    }
}

simplesmente funciona como 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));

agora será bastante simples escrever um takeWhilemétodo baseado emtake

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
}

funciona assim:

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

essa implementação itera a lista parcialmente por algumas vezes, mas não adiciona O(n^2)operações de adição . Espero que seja aceitável.

Eu tenho outra solução rápida implementando isso (que é realmente imundo, mas você entendeu):

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();
            }
        };
    }
}

Aqui está minha tentativa de usar apenas a biblioteca Java Stream.

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