chore: Make use of foldWhile and format.

Author: He-Pin

Diff for: docs/src/main/paradox/stream/operators/index.md

@@ -60,7 +60,8 @@ These built-in sinks are available from @scala[`org.apache.pekko.stream.scaladsl
 |Sink|<a name="collection"></a>@ref[collection](Sink/collection.md)|@scala[Collect all values emitted from the stream into a collection.]@java[Operator only available in the Scala API. The closest operator in the Java API is @ref[`Sink.seq`](Sink/seq.md)].|
 |Sink|<a name="combine"></a>@ref[combine](Sink/combine.md)|Combine several sinks into one using a user specified strategy|
 |Sink|<a name="completionstagesink"></a>@ref[completionStageSink](Sink/completionStageSink.md)|Streams the elements to the given future sink once it successfully completes. |
-|Sink|<a name="fold"></a>@ref[fold](Sink/fold.md)|Fold over emitted element with a function, where each invocation will get the new element and the result from the previous fold invocation.|
+|Sink|<a name="fold"></a>@ref[fold](Sink/fold.md)|Fold over emitted elements with a function, where each invocation will get the new element and the result from the previous fold invocation.|
+|Sink|<a name="foldwhile"></a>@ref[foldWhile](Sink/foldWhile.md)|Fold over emitted elements with a function, where each invocation will get the new element and the result from the previous fold invocation.|
 |Sink|<a name="forall"></a>@ref[forall](Sink/forall.md)|Apply a predicate function to assert each element received, it returns true if all elements satisfy the assertion, otherwise it returns false.|
 |Sink|<a name="foreach"></a>@ref[foreach](Sink/foreach.md)|Invoke a given procedure for each element received.|
 |Sink|<a name="foreachasync"></a>@ref[foreachAsync](Sink/foreachAsync.md)|Invoke a given procedure asynchronously for each element received.|
@@ -145,6 +146,7 @@ depending on being backpressured by downstream or not.
 |--|--|--|
 |Flow|<a name="asflowwithcontext"></a>@ref[asFlowWithContext](Flow/asFlowWithContext.md)|Extracts context data from the elements of a `Flow` so that it can be turned into a `FlowWithContext` which can propagate that context per element along a stream.|
 |Source/Flow|<a name="collect"></a>@ref[collect](Source-or-Flow/collect.md)|Apply a partial function to each incoming element, if the partial function is defined for a value the returned value is passed downstream.|
+|Source/Flow|<a name="collectfirst"></a>@ref[collectFirst](Source-or-Flow/collectFirst.md)|Transform this stream by applying the given partial function to the first element on which the function is defined as it pass through this processing step, and cancel the upstream publisher after the first element is emitted.|
 |Source/Flow|<a name="collecttype"></a>@ref[collectType](Source-or-Flow/collectType.md)|Transform this stream by testing the type of each of the elements on which the element is an instance of the provided type as they pass through this processing step.|
 |Source/Flow|<a name="collectwhile"></a>@ref[collectWhile](Source-or-Flow/collectWhile.md)|Transform this stream by applying the given partial function to each of the elements on which the function is defined as they pass through this processing step, and cancel the upstream publisher after the partial function is not applied.|
 |Flow|<a name="completionstageflow"></a>@ref[completionStageFlow](Flow/completionStageFlow.md)|Streams the elements through the given future flow once it successfully completes.|
@@ -158,6 +160,7 @@ depending on being backpressured by downstream or not.
 |Flow|<a name="flattenoptional"></a>@ref[flattenOptional](Flow/flattenOptional.md)|Collect the value of `Optional` from all the elements passing through this flow , empty `Optional` is filtered out.|
 |Source/Flow|<a name="fold"></a>@ref[fold](Source-or-Flow/fold.md)|Start with current value `zero` and then apply the current and next value to the given function. When upstream completes, the current value is emitted downstream.|
 |Source/Flow|<a name="foldasync"></a>@ref[foldAsync](Source-or-Flow/foldAsync.md)|Just like `fold` but receives a function that results in a @scala[`Future`] @java[`CompletionStage`] to the next value.|
+|Source/Flow|<a name="foldwhile"></a>@ref[foldWhile](Source-or-Flow/foldWhile.md)|Start with current value `zero` and then apply the current and next value to the given function. When upstream completes or the predicate `p` returns `false`, the current value is emitted downstream.|
 |Source/Flow|<a name="frommaterializer"></a>@ref[fromMaterializer](Source-or-Flow/fromMaterializer.md)|Defer the creation of a `Source/Flow` until materialization and access `Materializer` and `Attributes`|
 |Flow|<a name="futureflow"></a>@ref[futureFlow](Flow/futureFlow.md)|Streams the elements through the given future flow once it successfully completes.|
 |Source/Flow|<a name="grouped"></a>@ref[grouped](Source-or-Flow/grouped.md)|Accumulate incoming events until the specified number of elements have been accumulated and then pass the collection of elements downstream.|
@@ -413,6 +416,7 @@ For more background see the @ref[Error Handling in Streams](../stream-error.md)
 * [cancelled](Sink/cancelled.md)
 * [collect](Source-or-Flow/collect.md)
 * [collect](Sink/collect.md)
+* [collectFirst](Source-or-Flow/collectFirst.md)
 * [collection](Sink/collection.md)
 * [collectType](Source-or-Flow/collectType.md)
 * [collectWhile](Source-or-Flow/collectWhile.md)
@@ -452,6 +456,8 @@ For more background see the @ref[Error Handling in Streams](../stream-error.md)
 * [fold](Source-or-Flow/fold.md)
 * [fold](Sink/fold.md)
 * [foldAsync](Source-or-Flow/foldAsync.md)
+* [foldWhile](Source-or-Flow/foldWhile.md)
+* [foldWhile](Sink/foldWhile.md)
 * [forall](Sink/forall.md)
 * [foreach](Sink/foreach.md)
 * [foreachAsync](Sink/foreachAsync.md)

Diff for: stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java

@@ -17,8 +17,6 @@
 import org.apache.pekko.NotUsed;
 import org.apache.pekko.japi.Pair;
 import org.apache.pekko.japi.function.Function;
-import org.apache.pekko.japi.function.Predicate;
-import org.apache.pekko.stream.*;
 import org.apache.pekko.stream.Attributes;
 import org.apache.pekko.stream.Graph;
 import org.apache.pekko.stream.StreamTest;

Diff for: stream/src/main/scala/org/apache/pekko/stream/impl/fusing/Ops.scala

@@ -14,6 +14,7 @@
 package org.apache.pekko.stream.impl.fusing
 
 import java.util.concurrent.TimeUnit.NANOSECONDS
+
 import scala.annotation.nowarn
 import scala.annotation.tailrec
 import scala.collection.immutable

Diff for: stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala

@@ -88,11 +88,10 @@ object Sink {
    * the final predicate evaluation when the input stream ends, or completed with `Failure` if
    * there is a failure signaled in the stream.
    */
-  def forall[T](predicate: function.Predicate[T]): javadsl.Sink[T, CompletionStage[java.lang.Boolean]] = {
+  def forall[T](predicate: function.Predicate[T]): javadsl.Sink[T, CompletionStage[java.lang.Boolean]] =
     scaladsl.Sink.forall(predicate.test)
       .mapMaterializedValue(_.map(Boolean.box)(ExecutionContexts.parasitic))
       .toCompletionStage().asJava
-  }
 
   /**
    * Creates a sink which materializes into a ``CompletionStage`` which will be completed with a result of the Java ``Collector``

Diff for: stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala

@@ -16,30 +16,22 @@ package org.apache.pekko.stream.scaladsl
 import scala.annotation.{ nowarn, tailrec }
 import scala.annotation.unchecked.uncheckedVariance
 import scala.collection.immutable
-import scala.concurrent.ExecutionContext
-import scala.concurrent.Future
-import scala.Option
-import scala.util.Failure
-import scala.util.Success
-import scala.util.Try
+import scala.concurrent.{ ExecutionContext, Future }
+import scala.util.{ Failure, Success, Try }
 
 import org.apache.pekko
-import pekko.Done
-import pekko.NotUsed
-import pekko.actor.ActorRef
-import pekko.actor.Status
+import pekko.{ util, Done, NotUsed }
+import pekko.actor.{ ActorRef, Status }
 import pekko.annotation.InternalApi
 import pekko.dispatch.ExecutionContexts
 import pekko.stream._
 import pekko.stream.impl._
 import pekko.stream.impl.Stages.DefaultAttributes
 import pekko.stream.impl.fusing.GraphStages
-import pekko.stream.javadsl
 import pekko.stream.stage._
 import pekko.util.ccompat._
 
-import org.reactivestreams.Publisher
-import org.reactivestreams.Subscriber
+import org.reactivestreams.{ Publisher, Subscriber }
 
 /**
  * A `Sink` is a set of stream processing steps that has one open input.
@@ -456,11 +448,10 @@ object Sink {
    * the final predicate evaluation when the input stream ends, or completed with `Failure` if
    * there is a failure signaled in the stream.
    */
-  def forall[T](predicate: T => Boolean): Sink[T, Future[Boolean]] = {
-    Flow[T].statefulMap(() => true)((state, ele) => (predicate(ele), state), bool => Some(bool))
-      .takeWhile(s => s, inclusive = true)
-      .toMat(Sink.last)(Keep.right)
-  }
+  def forall[T](predicate: T => Boolean): Sink[T, Future[Boolean]] =
+    Flow[T].foldWhile(true)(util.ConstantFun.scalaIdentityFunction)(_ && predicate(_))
+      .toMat(Sink.head)(Keep.right)
+      .named("forallSink")
 
   /**
    * A `Sink` that will invoke the given function for every received element, giving it its previous