Limiter un flux par un prédicat

Existe-t-il une opération de flux Java 8 qui limite un (potentiellement infini) Stream jusqu'à ce que le premier élément ne corresponde pas à un prédicat?

En Java 9, nous pouvons utiliser takeWhilecomme dans l'exemple ci-dessous pour imprimer tous les nombres inférieurs à 10.

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

Comme il n'y a pas une telle opération dans Java 8, quelle est la meilleure façon de l'implémenter de manière générale?

Une telle opération devrait être possible avec un Java 8 Stream, mais elle ne peut pas nécessairement être effectuée efficacement - par exemple, vous ne pouvez pas nécessairement paralléliser une telle opération, car vous devez regarder les éléments dans l'ordre.

L'API ne fournit pas un moyen facile de le faire, mais le moyen le plus simple est probablement de prendre Stream.iterator(), d'encapsuler le Iteratorpour avoir une implémentation «à prendre en main», puis de revenir à a Spliterator, puis a Stream. Ou - peut-être - enveloppez le Spliterator, bien qu'il ne puisse plus vraiment être divisé dans cette implémentation.

Voici une implémentation non testée de takeWhilesur un 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);
}

Operations takeWhileet dropWhileont été ajoutés au JDK 9. Votre exemple de code

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

se comportera exactement comme vous vous y attendez lorsqu'il sera compilé et exécuté sous JDK 9.

Le JDK 9 a été publié. Il est disponible en téléchargement ici: http://jdk.java.net/9/

allMatch()est une fonction de court-circuit, vous pouvez donc l'utiliser pour arrêter le traitement. Le principal inconvénient est que vous devez faire votre test deux fois: une fois pour voir si vous devez le traiter, et encore une fois pour voir s'il faut continuer.

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

Dans le prolongement de la réponse @StuartMarks . Ma bibliothèque StreamEx a un takeWhilefonctionnement compatible avec l'implémentation actuelle du JDK-9. Lors de l'exécution sous JDK-9, il déléguera simplement à l'implémentation JDK (via MethodHandle.invokeExactlaquelle est vraiment rapide). Lors de l'exécution sous JDK-8, l'implémentation "polyfill" sera utilisée. Donc, en utilisant ma bibliothèque, le problème peut être résolu comme ceci:

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

takeWhileest l'une des fonctions fournies par la bibliothèque 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));

Mise à jour: Java 9 Streamest désormais livré avec une méthode takeWhile .

Pas besoin de hacks ou d'autres solutions. Utilisez juste ça!

Je suis sûr que cela peut être grandement amélioré: (quelqu'un pourrait peut-être le rendre thread-safe)

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

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

Un hack bien sûr ... Pas élégant - mais ça marche ~: 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);
    }

}

Vous pouvez utiliser 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));

En fait, il existe 2 façons de le faire dans Java 8 sans aucune bibliothèque supplémentaire ou en utilisant Java 9.

Si vous souhaitez imprimer les nombres de 2 à 20 sur la console, vous pouvez le faire:

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

La sortie est dans les deux cas:

2
4
6
8
10
12
14
16
18
20

Personne n'a encore mentionné anyMatch . C'est la raison de cet article.

Il s'agit de la source copiée à partir du JDK 9 java.util.stream.Stream.takeWhile (Predicate). Une petite différence pour travailler avec 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);
}

Voici une version réalisée sur ints - comme demandé dans la question.

Usage:

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

Voici le code pour 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;
        }
    }
}

Allez chercher la bibliothèque AbacusUtil . Il fournit l'API exacte que vous souhaitez et plus encore:

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

Déclaration ： Je suis le développeur d'AbacusUtil.

Vous ne pouvez pas abandonner un flux sauf par une opération de terminal en court-circuit, ce qui laisserait certaines valeurs de flux non traitées quelle que soit leur valeur. Mais si vous souhaitez simplement éviter les opérations sur un flux, vous pouvez ajouter une transformation et un filtre au flux:

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

Cela transforme le flux d'objets en valeurs nulles lorsque les éléments remplissent une condition, puis filtre les valeurs nulles. Si vous êtes prêt à vous adonner aux effets secondaires, vous pouvez définir la valeur de la condition sur true une fois que quelque chose est rencontré, de sorte que toutes les choses suivantes sont filtrées quelle que soit leur valeur. Mais même si ce n'est pas le cas, vous pouvez économiser beaucoup (sinon la totalité) du traitement en filtrant les valeurs du flux que vous ne souhaitez pas traiter.

Même moi, j'avais une exigence similaire: invoquez le service Web, s'il échoue, réessayez 3 fois. S'il échoue même après ces nombreux essais, envoyez une notification par e-mail. Après avoir beaucoup cherché sur Google, anyMatch()est venu en sauveur. Mon exemple de code comme suit. Dans l'exemple suivant, si la méthode webServiceCall renvoie true dans la première itération elle-même, stream n'itère pas plus loin comme nous l'avons appelé anyMatch(). Je crois que c'est ce que vous recherchez.

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

Si vous connaissez le nombre exact de répétitions qui seront effectuées, vous pouvez faire

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

au lieu de peak, vous pouvez utiliser mapToObj pour renvoyer l'objet ou le message 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);

Si vous avez un problème différent, une solution différente peut être nécessaire, mais pour votre problème actuel, j'irais simplement avec:

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

Peut-être un peu hors sujet, mais c'est ce que nous avons pour List<T>plutôt queStream<T> .

Vous devez d'abord avoir une takeméthode util. Cette méthode prend les premiers néléments:

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

ça marche juste comme 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));

maintenant, il sera assez simple d'écrire une takeWhileméthode basée surtake

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
}

ça marche comme ça:

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

cette implémentation itère partiellement la liste plusieurs fois mais n'ajoute pas d' O(n^2)opérations d' ajout . J'espère que c'est acceptable.

J'ai une autre solution rapide en mettant en œuvre ceci (ce qui est plutôt impur en fait, mais vous voyez l'idée):

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

Voici ma tentative en utilisant uniquement la bibliothèque Java Stream.

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