Wie verknüpfe ich Ströme fließend?

Question

Ich versuche, einen Stream zu implementieren, der eine andere Instanz seiner selbst in seiner Implementierung verwendet. Dem Stream sind einige konstante Elemente vorangestellt (mit IntStream.concat), daher sollte dies so lange funktionieren, wie der verkettete Stream den nicht konstanten Teil träge erzeugt. Ich denke, die Verwendung der StreamSupport.intStream overload taking a Supplier mit IntStream.concat (die "creates a lazily concatenated stream") sollte faul genug sein, nur den zweiten Spliterator zu erstellen, wenn Elemente von ihm angefordert werden, aber sogar den Stream erstellen (nicht evaluieren) überläuft den Stapel. Wie kann ich Ströme fließend verketten?

---

Ich bin versucht, das Streaming-Primzahl Sieb von this answer in Java zu portieren. Dieses Sieb verwendet eine andere Instanz von sich selbst (ps = postponed_sieve() im Python-Code). Wenn ich die ersten vier konstanten Elemente (yield 2; yield 3; yield 5; yield 7;) in ihren eigenen Strom zu brechen, ist es einfach, den Generator als spliterator zu implementieren:

/** 
* based on https://stackoverflow.com/a/10733621/3614835 
*/ 
static class PrimeSpliterator extends Spliterators.AbstractIntSpliterator { 
    private static final int CHARACTERISTICS = Spliterator.DISTINCT | Spliterator.IMMUTABLE | Spliterator.NONNULL | Spliterator.ORDERED | Spliterator.SORTED; 
    private final Map<Integer, Supplier<IntStream>> sieve = new HashMap<>(); 
    private final PrimitiveIterator.OfInt postponedSieve = primes().iterator(); 
    private int p, q, c = 9; 
    private Supplier<IntStream> s; 
    PrimeSpliterator() { 
     super(105097564 /* according to Wolfram Alpha */ - 4 /* in prefix */, 
       CHARACTERISTICS); 
     //p = next(ps) and next(ps) (that's Pythonic?) 
     postponedSieve.nextInt(); 
     this.p = postponedSieve.nextInt(); 
     this.q = p*p; 
    } 

    @Override 
    public boolean tryAdvance(IntConsumer action) { 
     for (; c > 0 /* overflow */; c += 2) { 
      Supplier<IntStream> maybeS = sieve.remove(c); 
      if (maybeS != null) 
       s = maybeS; 
      else if (c < q) { 
       action.accept(c); 
       return true; //continue 
      } else { 
       s =() -> IntStream.iterate(q+2*p, x -> x + 2*p); 
       p = postponedSieve.nextInt(); 
       q = p*p; 
      } 
      int m = s.get().filter(x -> !sieve.containsKey(x)).findFirst().getAsInt(); 
      sieve.put(m, s); 
     } 
     return false; 
    } 
} 

Mein erster Versuch, die Primzahlen() Methode gibt ein IntStream einen konstanten Strom verketten mit ein neues PrimeSpliterator:

public static IntStream primes() { 
    return IntStream.concat(IntStream.of(2, 3, 5, 7), 
      StreamSupport.intStream(new PrimeSpliterator())); 
} 

Aufruf Primzahlen() führt zu einem Stackoverflow weil Primzahlen() immer ein PrimeSpliterator instanziiert, aber PrimeSpliterator des Feldinitialisierer immer ruft Primzahlen(). Allerdings gibt es eine Überlastung von StreamSupport.intStream, die einen Lieferanten nimmt, die träge erlauben sollten den PrimeSpliterator erstellen:

public static IntStream primes() { 
    return IntStream.concat(IntStream.of(2, 3, 5, 7), 
      StreamSupport.intStream(PrimeSpliterator::new, PrimeSpliterator.CHARACTERISTICS, false)); 
} 

Aber ich stattdessen eine Stackoverflow mit einer anderen Rückverfolgung erhalten (getrimmt, wie es wiederholt). Beachten Sie, dass die Rekursion vollständig im Aufruf von primes() enthalten ist - der terminal operation iterator() wird nie für einen zurückgegebenen Stream aufgerufen.

Exception in thread "main" java.lang.StackOverflowError 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator$OfInt.<init>(StreamSpliterators.java:582) 
    at java.util.stream.IntPipeline.lazySpliterator(IntPipeline.java:155) 
    at java.util.stream.IntPipeline$Head.lazySpliterator(IntPipeline.java:514) 
    at java.util.stream.AbstractPipeline.spliterator(AbstractPipeline.java:352) 
    at java.util.stream.IntPipeline.spliterator(IntPipeline.java:181) 
    at java.util.stream.IntStream.concat(IntStream.java:851) 
    at com.jeffreybosboom.projecteuler.util.Primes.primes(Primes.java:22) 
    at com.jeffreybosboom.projecteuler.util.Primes$PrimeSpliterator.<init>(Primes.java:32) 
    at com.jeffreybosboom.projecteuler.util.Primes$$Lambda$1/834600351.get(Unknown Source) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.get(StreamSpliterators.java:513) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.estimateSize(StreamSpliterators.java:536) 
    at java.util.stream.Streams$ConcatSpliterator.<init>(Streams.java:713) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:789) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:785) 
    at java.util.stream.Streams$ConcatSpliterator$OfInt.<init>(Streams.java:819) 
    at java.util.stream.IntStream.concat(IntStream.java:851) 
    at com.jeffreybosboom.projecteuler.util.Primes.primes(Primes.java:22) 
    at com.jeffreybosboom.projecteuler.util.Primes$PrimeSpliterator.<init>(Primes.java:32) 
    at com.jeffreybosboom.projecteuler.util.Primes$$Lambda$1/834600351.get(Unknown Source) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.get(StreamSpliterators.java:513) 
    at java.util.stream.StreamSpliterators$DelegatingSpliterator.estimateSize(StreamSpliterators.java:536) 
    at java.util.stream.Streams$ConcatSpliterator.<init>(Streams.java:713) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:789) 
    at java.util.stream.Streams$ConcatSpliterator$OfPrimitive.<init>(Streams.java:785) 
    at java.util.stream.Streams$ConcatSpliterator$OfInt.<init>(Streams.java:819) 
    at java.util.stream.IntStream.concat(IntStream.java:851) 
    at com.jeffreybosboom.projecteuler.util.Primes.primes(Primes.java:22) 

Wie kann ich Streams verketten faul genug, um einen Strom zu ermöglichen, eine weitere Kopie von sich selbst in seiner Umsetzung zu benutzen?

Answer 1

Sie gehen offenbar davon aus, dass die Streams-API ihre Faulheitsgarantie sogar auf die Instantiierung von Spliteratoren ausdehnt; das ist nicht richtig. Es erwartet, dass der Spliterator des Streams jederzeit vor dem tatsächlichen Verbrauch instanziiert werden kann, beispielsweise um die Eigenschaften des Streams und die gemeldete Größe zu ermitteln. Der Verbrauch beginnt erst mit dem Aufruf von trySplit, tryAdvance oder forEachRemaining.

In diesem Sinne initialisieren Sie das verschobene Sieb früher als Sie es benötigen. Sie können keines der Ergebnisse bis zum else if Teil in tryAdvance verwenden. So verschieben Sie den Code zum letzten Moment der Richtigkeit gibt:

@Override 
public boolean tryAdvance(IntConsumer action) { 
    for (; c > 0 /* overflow */; c += 2) { 
     Supplier<IntStream> maybeS = sieve.remove(c); 
     if (maybeS != null) 
      s = maybeS; 
     else { 
      if (postponedSieve == null) { 
       postponedSieve = primes().iterator(); 
       postponedSieve.nextInt(); 
       this.p = postponedSieve.nextInt(); 
       this.q = p*p; 
      } 
      if (c < q) { 
       action.accept(c); 
       return true; //continue 

Ich denke, dass mit dieser Änderung auch Ihr erster Versuch primes() funktionieren soll.

Wenn Sie mit Ihrem aktuellen Ansatz bleiben wollen, können Sie das folgende Idiom beinhalten:

Stream.<Supplier<IntStream>>of(
()->IntStream.of(2, 3, 5, 7), 
()->intStream(new PrimeSpliterator())) 
.flatMap(Supplier::get); 

Sie können feststellen, dass diese Ihnen so viel Faulheit gibt, wie Sie benötigen.