You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "laglangyue (via GitHub)" <gi...@apache.org> on 2024/01/20 09:26:56 UTC
[PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
laglangyue opened a new pull request, #989:
URL: https://github.com/apache/incubator-pekko/pull/989
## motivation
#972
## includes
add forall/forallAsync for flowops/sink, runForall for source
add some document
## test
add tests and test locally
and test by CI
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1468516915
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,47 @@
+# Sink.forall
+
+A `Sink` that will test the given predicate `p` for every received element and completes with the result.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall%5BT%5D(p%3A%20T%20%3D%3E%20Boolean):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
+
+## Description
+forall applies a predicate function to assert each element received, it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+It materializes into a `Future` (in Scala) or a `CompletionStage` (in Java) that completes with the last state when the stream has finished.
+
+Notes that if source is empty, it will return true
+
+A `Sink` that will test the given predicate `p` for every received element and
+
+ - completes and returns @scala[`Future`] @java[`CompletionStage`] of `true` if the predicate is true for all elements;
+ - completes and returns @scala[`Future`] @java[`CompletionStage`] of `true` if the stream is empty (i.e. completes before signalling any elements);
+ - completes and returns @scala[`Future`] @java[`CompletionStage`] of `false` if the predicate is false for any element.
+
+The materialized value @scala[`Future`] @java[`CompletionStage`] will be completed with the value `true` or `false`
+when the input stream ends, or completed with `Failure` if there is a failure signaled in the stream.
Review Comment:
It depends on the supervision behavior, with the restart strategy, if there are two elements and one matches and one doesn't, and the later unmatched one causes an Exception, then it should well return `true`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460745471
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala:
##########
@@ -435,6 +436,24 @@ object Sink {
def foldAsync[U, T](zero: U)(f: (U, T) => Future[U]): Sink[T, Future[U]] =
Flow[T].foldAsync(zero)(f).toMat(Sink.head)(Keep.right).named("foldAsyncSink")
+ /**
+ * A `Sink` that will invoke the given function for every received element, giving it its previous
Review Comment:
`function` -> predicates , better to match the forall.md
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460744970
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala:
##########
@@ -69,6 +69,20 @@ object Sink {
f: function.Function2[U, In, CompletionStage[U]]): javadsl.Sink[In, CompletionStage[U]] =
new Sink(scaladsl.Sink.foldAsync[U, In](zero)(f(_, _).asScala).toCompletionStage())
+ /**
+ * A `Sink` that will invoke the given function for every received element, giving it its previous
Review Comment:
`function` -> `predicates`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1913464500
@pjfanning @mdedetrich ping for a review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1468444778
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,47 @@
+# Sink.forall
+
+A `Sink` that will test the given predicate `p` for every received element and completes with the result.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall%5BT%5D(p%3A%20T%20%3D%3E%20Boolean):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
+
+## Description
+forall applies a predicate function to assert each element received, it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+It materializes into a `Future` (in Scala) or a `CompletionStage` (in Java) that completes with the last state when the stream has finished.
+
+Notes that if source is empty, it will return true
+
+A `Sink` that will test the given predicate `p` for every received element and
+
+ - completes and returns @scala[`Future`] @java[`CompletionStage`] of `true` if the predicate is true for all elements;
+ - completes and returns @scala[`Future`] @java[`CompletionStage`] of `true` if the stream is empty (i.e. completes before signalling any elements);
+ - completes and returns @scala[`Future`] @java[`CompletionStage`] of `false` if the predicate is false for any element.
+
+The materialized value @scala[`Future`] @java[`CompletionStage`] will be completed with the value `true` or `false`
+when the input stream ends, or completed with `Failure` if there is a failure signaled in the stream.
Review Comment:
what is the intended behaviour if the function fails for one of the elements? Can we have tests to verify that behaviour?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1913261992
@pjfanning I think this one should be fine now, networking issue slow me done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1465359580
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala:
##########
@@ -447,6 +440,28 @@ object Sink {
def foldAsync[U, T](zero: U)(f: (U, T) => Future[U]): Sink[T, Future[U]] =
Flow[T].foldAsync(zero)(f).toMat(Sink.head)(Keep.right).named("foldAsyncSink")
+ /**
+ * A `Sink` that will test the given predicate `p` for every received element and
+ * 1. completes and returns [[scala.concurrent.Future]] of `true` if the predicate is true for all elements;
+ * 2. completes and returns [[scala.concurrent.Future]] of `true` if the stream is empty (i.e. completes before signalling any elements);
+ * 3. completes and returns [[scala.concurrent.Future]] of `false` if the predicate is false for any element.
+ *
+ * The materialized value [[scala.concurrent.Future]] will be completed with the value `true` or `false`
+ * when the input stream ends, or completed with `Failure` if there is a failure signaled in the stream.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Completes when''' upstream completes or the predicate `p` returns `false`
+ *
+ * '''Backpressures when''' the invocation of predicate `p` has not yet completed
+ *
+ * '''Cancels when''' predicate `p` returns `false`
+ */
+ def forall[T](p: T => Boolean): Sink[T, Future[Boolean]] =
+ Flow[T].foldWhile(true)(util.ConstantFun.scalaIdentityFunction)(_ && p(_))
+ .toMat(Sink.head)(Keep.right)
+ .named("forallSink")
Review Comment:
the implementation makes use of `foldWhile`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460777227
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,34 @@
+# Sink.forall
+
+forall applies a predicate assertion to each element in the stream; it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall(predicate:T=>U):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
+
+## Description
+forall, is implemented through `Fold`, applies a predicate assertion to each element in the stream; it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+It materializes into a `Future` (in Scala) or a `CompletionStage` (in Java) that completes with the last state when the stream has finished.
+
+This operator enables combining values into a result without relying on global mutable state, as it passes the state along between invocations.
+
+## Example
+
+This example reads a stream of positive integers, asserts that all numbers are greater than 0, and finally prints the assertion result.
+
+Scala
+: @@snip [snip](/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SinkSpec.scala) { #forall }
+
+Java
+: @@snip [snip](/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java) { #forall }
+
+## Reactive Streams Semantics
+
+@@@div { .callout }
+
+**cancels**: never
+
+**backpressures**: when the previous predicate function invocation has not yet completed
Review Comment:
I have tried my best to describe it again, but the theory here is not clear yet. Can you improve it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460861990
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala:
##########
@@ -445,12 +445,9 @@ object Sink {
* there is a failure signaled in the stream.
*/
def forall[T](predicate: T => Boolean): Sink[T, Future[Boolean]] = {
- var acc = true
- Flow[T].collectWhile {
- case element if acc && predicate(element) => true
- case _ if acc => acc = false; false
- }.statefulMap(() => true)((state, bool) => (state && bool, bool), bool => Some(bool))
- .toMat(Sink.last)(Keep.right).named("forallSink")
+ Flow[T].statefulMap(() => true)((state, ele) => (predicate(ele), state), bool => Some(bool))
+ .takeWhile(s => s, inclusive = true)
+ .toMat(Sink.last)(Keep.right)
Review Comment:
Cool, I have a pending PR, which will help this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460390502
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1792,6 +1792,30 @@ trait FlowOps[+Out, +Mat] {
*/
def foldAsync[T](zero: T)(f: (T, Out) => Future[T]): Repr[T] = via(new FoldAsync(zero, f))
+ /**
+ * implemented by `Fold`, only emits its result when the upstream completes,
+ * after which it also completes. Applies the given function towards initialized `true`
+ * and next predicate result, yielding the next predicate result.
+ *
+ * If the function `predicate` throws an exception and the supervision decision is
+ * [[pekko.stream.Supervision.Restart]] current value starts at `true` again
+ * the stream will continue.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * See also [[FlowOps.fold]]
+ */
+ def forall(predicate: Out => Boolean): Repr[Boolean] =
+ via(Fold(true, (acc: Boolean, element) => acc && predicate(element)))
Review Comment:
Yes, I also thought of this, so I used `acc&&predict (element)` (boolean short circuited)，I wonder if `fold` should not be used.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1902038616
> @laglangyue Thanks and welcome the pr from China scala user group. ping me if you think it's ready.
ok I will do this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460384881
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1792,6 +1792,30 @@ trait FlowOps[+Out, +Mat] {
*/
def foldAsync[T](zero: T)(f: (T, Out) => Future[T]): Repr[T] = via(new FoldAsync(zero, f))
+ /**
+ * implemented by `Fold`, only emits its result when the upstream completes,
+ * after which it also completes. Applies the given function towards initialized `true`
+ * and next predicate result, yielding the next predicate result.
+ *
+ * If the function `predicate` throws an exception and the supervision decision is
+ * [[pekko.stream.Supervision.Restart]] current value starts at `true` again
+ * the stream will continue.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * See also [[FlowOps.fold]]
+ */
+ def forall(predicate: Out => Boolean): Repr[Boolean] =
+ via(Fold(true, (acc: Boolean, element) => acc && predicate(element)))
Review Comment:
You can see: scala.collection.IterableOnceOps#forall can do a quick break.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460420837
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowForallSpec.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.scaladsl
+
+import org.apache.pekko.stream.testkit.StreamSpec
+import org.apache.pekko.stream.testkit.Utils.TE
+import org.apache.pekko.stream.{ ActorAttributes, Supervision }
+import scala.concurrent.{ Await, Future }
+import scala.concurrent.duration.DurationInt
+
+class FlowForallSpec extends StreamSpec {
+
+ "a forall" must {
+ val error = TE("Boom!")
+ val inputSource = Source(1 to 100)
+
+ "work when use source.forall = true" in {
+ // # forall
+ val predicate: Future[Boolean] = Source(1 to 100).forall(_ > 0).runWith(Sink.head)
+ // # forall
+ Await.result(predicate, 3.seconds) shouldBe true
+ }
+
+ "work when using source.forall = false" in {
+ Await.result(inputSource.forall(_ < 50).runWith(Sink.head), 3.seconds) should be(false)
+ }
+
+ "work when using sink.forall = false" in {
+ Await.result(inputSource.runWith(Sink.forall(_ < 50)), 3.second) should be(false)
+ }
+
+ "work when using flow.forall = false" in {
+ Await.result(inputSource.via(Flow[Int].forall(_ < 50)).runWith(Sink.head), 3.second) should be(false)
+ }
+
+ "work when using source.runForall = false" in {
+ Await.result(inputSource.runForall(_ < 50), 3.seconds) should be(false)
+ }
+
+ "work when using source.runForall = true" in {
+ Await.result(inputSource.runForall(_ <= 100), 3.seconds) should be(true)
+ }
+
+ "work when source empty = true" in {
+ Await.result(inputSource.filter(_ < 0).runForall(_ > 0), 3.seconds) should be(true)
+ }
+
+ "work when using flow.forallAsync = false" in {
+ Await.result(inputSource.via(Flow[Int].forall(_ < 50)).runWith(Sink.head), 3.second) should be(false)
Review Comment:
Split the forallAsync to another pr.
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,34 @@
+# Sink.forall
+
+forall applies a predicate assertion to each element in the stream; it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall(predicate:T=>U):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
+
+## Description
+forall, is implemented through `Fold`, applies a predicate assertion to each element in the stream; it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+It materializes into a `Future` (in Scala) or a `CompletionStage` (in Java) that completes with the last state when the stream has finished.
+
+This operator enables combining values into a result without relying on global mutable state, as it passes the state along between invocations.
+
+## Example
+
+This example reads a stream of positive integers, asserts that all numbers are greater than 0, and finally prints the assertion result.
+
+Scala
+: @@snip [snip](/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SinkSpec.scala) { #forall }
+
+Java
+: @@snip [snip](/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java) { #forall }
+
+## Reactive Streams Semantics
+
+@@@div { .callout }
+
+**cancels**: never
+
+**backpressures**: when the previous predicate function invocation has not yet completed
Review Comment:
Add the full semantics?
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java:
##########
@@ -224,4 +225,18 @@ public void sinkForeachMustBeDocumented()
// #foreach
assertEquals(Done.done(), done);
}
+
+ @Test
+ public void sinkForallMustBeDocumented()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ // #forall
+ Sink<Integer, CompletionStage<Boolean>> forallSink = Sink.forall((Predicate<Integer>) param -> param > 0);
+ CompletionStage<Boolean> cs = Source.from(Arrays.asList(1, 2, 3, 4)).runWith(forallSink, system);
+ Object predicate = cs.toCompletableFuture().get(100, TimeUnit.MILLISECONDS);
Review Comment:
Boolean？
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -1614,4 +1614,15 @@ public void mustBeAbleToConvertToJavaInJava() {
org.apache.pekko.stream.scaladsl.Flow.apply();
Flow<Integer, Integer, NotUsed> javaFlow = scalaFlow.asJava();
}
+
+ @Test
+ public void mustBeAbleToUseForall() throws Exception {
+ final TestKit probe = new TestKit(system);
+ // # forall
+ Flow<Integer, Boolean, NotUsed> forallFlow = Flow.of(Integer.class).forall((Predicate<Integer>) s -> s > 0);
Review Comment:
Ha, java
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowForallSpec.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.scaladsl
+
+import org.apache.pekko.stream.testkit.StreamSpec
+import org.apache.pekko.stream.testkit.Utils.TE
+import org.apache.pekko.stream.{ ActorAttributes, Supervision }
+import scala.concurrent.{ Await, Future }
+import scala.concurrent.duration.DurationInt
+
+class FlowForallSpec extends StreamSpec {
+
+ "a forall" must {
+ val error = TE("Boom!")
+ val inputSource = Source(1 to 100)
+
+ "work when use source.forall = true" in {
+ // # forall
+ val predicate: Future[Boolean] = Source(1 to 100).forall(_ > 0).runWith(Sink.head)
+ // # forall
+ Await.result(predicate, 3.seconds) shouldBe true
+ }
+
+ "work when using source.forall = false" in {
+ Await.result(inputSource.forall(_ < 50).runWith(Sink.head), 3.seconds) should be(false)
+ }
+
+ "work when using sink.forall = false" in {
+ Await.result(inputSource.runWith(Sink.forall(_ < 50)), 3.second) should be(false)
+ }
+
+ "work when using flow.forall = false" in {
+ Await.result(inputSource.via(Flow[Int].forall(_ < 50)).runWith(Sink.head), 3.second) should be(false)
+ }
+
+ "work when using source.runForall = false" in {
+ Await.result(inputSource.runForall(_ < 50), 3.seconds) should be(false)
+ }
+
+ "work when using source.runForall = true" in {
+ Await.result(inputSource.runForall(_ <= 100), 3.seconds) should be(true)
+ }
+
+ "work when source empty = true" in {
+ Await.result(inputSource.filter(_ < 0).runForall(_ > 0), 3.seconds) should be(true)
+ }
+
+ "work when using flow.forallAsync = false" in {
+ Await.result(inputSource.via(Flow[Int].forall(_ < 50)).runWith(Sink.head), 3.second) should be(false)
+ }
+
+ "function predicate should be short-circuited" in {
+ Await.result(inputSource.runForall(s => if (s < 50) false else throw error), 3.seconds) should be(false)
+ }
+
+ "complete future with failure when the predicate function throws and the supervisor strategy decides to stop" in {
+ val forall = inputSource.runForall(s => if (s == 50) throw error else true)
+ the[Exception] thrownBy Await.result(forall, 3.seconds) should be(error)
+ }
+
+ "resume with the accumulated state when the predicate function throws and the supervisor strategy decides to resume" in {
+ val forall = Sink.forall[Int](s => if (s == 50) throw error else true)
+ val future =
+ inputSource.runWith(forall.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider)))
+
+ Await.result(future, 3.seconds) should be(true)
+ }
+
+ "resume and reset the state when the predicate function throws when the supervisor strategy decides to restart" in {
+ val forall = Sink.forall[Int](s => if (s == 50) throw error else true)
+ val future =
+ inputSource.runWith(forall.withAttributes(ActorAttributes.supervisionStrategy(Supervision.restartingDecider)))
+
+ Await.result(future, 3.seconds) should be(true)
+ }
Review Comment:
Add a spec for empty source
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowForallSpec.scala:
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.scaladsl
+
+import org.apache.pekko.stream.testkit.StreamSpec
+import org.apache.pekko.stream.testkit.Utils.TE
+import org.apache.pekko.stream.{ ActorAttributes, Supervision }
+import scala.concurrent.{ Await, Future }
+import scala.concurrent.duration.DurationInt
+
+class FlowForallSpec extends StreamSpec {
+
+ "a forall" must {
+ val error = TE("Boom!")
+ val inputSource = Source(1 to 100)
+
+ "work when use source.forall = true" in {
+ // # forall
+ val predicate: Future[Boolean] = Source(1 to 100).forall(_ > 0).runWith(Sink.head)
+ // # forall
+ Await.result(predicate, 3.seconds) shouldBe true
+ }
+
+ "work when using source.forall = false" in {
+ Await.result(inputSource.forall(_ < 50).runWith(Sink.head), 3.seconds) should be(false)
+ }
+
+ "work when using sink.forall = false" in {
+ Await.result(inputSource.runWith(Sink.forall(_ < 50)), 3.second) should be(false)
+ }
+
+ "work when using flow.forall = false" in {
+ Await.result(inputSource.via(Flow[Int].forall(_ < 50)).runWith(Sink.head), 3.second) should be(false)
Review Comment:
Can use futurevalue should be in scala spec
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460744652
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java:
##########
@@ -224,4 +225,20 @@ public void sinkForeachMustBeDocumented()
// #foreach
assertEquals(Done.done(), done);
}
+
+ @Test
+ public void sinkForallMustBeDocumented()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ // #forall
+ Sink<Integer, CompletionStage<Boolean>> forallSink =
+ Sink.forall((Predicate<Integer>) param -> param > 0);
+ CompletionStage<Boolean> cs =
+ Source.from(Arrays.asList(1, 2, 3, 4)).runWith(forallSink, system);
+ Object predicate = cs.toCompletableFuture().get(100, TimeUnit.MILLISECONDS);
+ System.out.println(predicate);
+ // will print
Review Comment:
Expect prints?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1908323418
I have contacted with @laglangyue ,and will take care of this pr now, will update it in this pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460744732
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SinkSpec.scala:
##########
@@ -339,6 +339,42 @@ class SinkSpec extends StreamSpec with DefaultTimeout with ScalaFutures {
}
}
+ "The forall sink" must {
+ "always true forall = true" in {
+ // #foreach
+ val forallSink: Sink[Int, Future[Boolean]] = Sink.forall[Int](_ > 0)
+ val f = Source(1 to 4).runWith(forallSink)
+ val result = Await.result(f, 100.millis)
+ println(result)
Review Comment:
remove the `println` in spec, and for the doc, you can copy it and addback the `println`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460744732
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SinkSpec.scala:
##########
@@ -339,6 +339,42 @@ class SinkSpec extends StreamSpec with DefaultTimeout with ScalaFutures {
}
}
+ "The forall sink" must {
+ "always true forall = true" in {
+ // #foreach
+ val forallSink: Sink[Int, Future[Boolean]] = Sink.forall[Int](_ > 0)
+ val f = Source(1 to 4).runWith(forallSink)
+ val result = Await.result(f, 100.millis)
+ println(result)
Review Comment:
remove the `println` in spec.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460746593
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala:
##########
@@ -435,6 +436,24 @@ object Sink {
def foldAsync[U, T](zero: U)(f: (U, T) => Future[U]): Sink[T, Future[U]] =
Flow[T].foldAsync(zero)(f).toMat(Sink.head)(Keep.right).named("foldAsyncSink")
+ /**
+ * A `Sink` that will invoke the given function for every received element, giving it its previous
+ * output (or the given `false`) and the element as input.
+ * * The returned [[scala.concurrent.Future]] will be completed with value of the final
+ * function 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]] = {
+ var acc = true
+ Flow[T].collectWhile {
+ case element if acc && predicate(element) => true
+ case _ if acc => acc = false; false
Review Comment:
As the graph can be materialize multi time , this can cause a problem. when you run this graph several times.
Maybe you can try with `statefulMap`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460750113
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala:
##########
@@ -435,6 +436,24 @@ object Sink {
def foldAsync[U, T](zero: U)(f: (U, T) => Future[U]): Sink[T, Future[U]] =
Flow[T].foldAsync(zero)(f).toMat(Sink.head)(Keep.right).named("foldAsyncSink")
+ /**
+ * A `Sink` that will invoke the given function for every received element, giving it its previous
+ * output (or the given `false`) and the element as input.
+ * * The returned [[scala.concurrent.Future]] will be completed with value of the final
+ * function 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]] = {
+ var acc = true
+ Flow[T].collectWhile {
+ case element if acc && predicate(element) => true
+ case _ if acc => acc = false; false
Review Comment:
let me a try
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1468539657
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SinkSpec.scala:
##########
@@ -339,6 +340,47 @@ class SinkSpec extends StreamSpec with DefaultTimeout with ScalaFutures {
}
}
+ "The forall sink" must {
+
+ "completes with `ture` when all elements match" in {
+ Source(1 to 4)
+ .runWith(Sink.forall(_ > 0))
+ .futureValue shouldBe true
+ }
+
+ "completes with `false` when any element match" in {
+ Source(1 to 4)
+ .runWith(Sink.forall(_ > 2))
+ .futureValue shouldBe false
+ }
+
+ "completes with `true` if the stream is empty" in {
+ Source.empty[Int]
+ .runWith(Sink.forall(_ > 2))
+ .futureValue shouldBe true
+ }
+
+ "completes with `Failure` if the stream failed" in {
+ Source.failed[Int](new RuntimeException("Oops"))
+ .runWith(Sink.forall(_ > 2))
+ .failed.futureValue shouldBe a[RuntimeException]
+ }
+
+ "completes with `true` with restart strategy" in {
+ val sink = Sink.forall[Int](elem => {
+ if (elem == 2) {
+ throw new RuntimeException("Oops")
+ }
+ elem > 0
+ }).withAttributes(supervisionStrategy(Supervision.restartingDecider))
+
+ Source(1 to 2)
+ .runWith(sink)
+ .futureValue shouldBe true
Review Comment:
@pjfanning The requested test have been added.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1902066284
@He-Pin ready
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460383732
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1792,6 +1792,30 @@ trait FlowOps[+Out, +Mat] {
*/
def foldAsync[T](zero: T)(f: (T, Out) => Future[T]): Repr[T] = via(new FoldAsync(zero, f))
+ /**
+ * implemented by `Fold`, only emits its result when the upstream completes,
+ * after which it also completes. Applies the given function towards initialized `true`
+ * and next predicate result, yielding the next predicate result.
+ *
+ * If the function `predicate` throws an exception and the supervision decision is
+ * [[pekko.stream.Supervision.Restart]] current value starts at `true` again
+ * the stream will continue.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * See also [[FlowOps.fold]]
+ */
+ def forall(predicate: Out => Boolean): Repr[Boolean] =
+ via(Fold(true, (acc: Boolean, element) => acc && predicate(element)))
Review Comment:
add `.name("forAll")`, impelment it with `fold` has once drawback, which is we can't do a quick cancel after one elemement which doesn't match the predicate. maybe we can test it with `collectWhile and fold`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1902113695
I was so foolish. I actually implemented `forall` for the flow. I only found out after completing the test that the flow should not have `forall`. I remove them all.
It's ready now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1461300880
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,36 @@
+# Sink.forall
+
+Apply a predicate function to assert each element received, it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall(predicate:T=>U):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
Review Comment:
```suggestion
@apidoc[Sink.forall](Sink$) { scala="#forall(predicate:T=%3EU):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1468469688
##########
docs/src/test/java/jdocs/stream/operators/sink/ForAll.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package jdocs.stream.operators.sink;
+
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.stream.javadsl.Sink;
+import org.apache.pekko.stream.javadsl.Source;
+
+import java.util.concurrent.TimeUnit;
+
+public class ForAll {
+ private ActorSystem system = null;
+
+ public void forAllUsage() throws Exception {
+ // #forall
+ final boolean allMatch =
+ Source.range(1, 100)
+ .runWith(Sink.forall(elem -> elem <= 100), system)
+ .toCompletableFuture()
+ .get(3, TimeUnit.SECONDS);
+ System.out.println(allMatch);
Review Comment:
Use assert rather than STD_OUT
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,47 @@
+# Sink.forall
+
+A `Sink` that will test the given predicate `p` for every received element and completes with the result.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall%5BT%5D(p%3A%20T%20%3D%3E%20Boolean):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
Review Comment:
```suggestion
@apidoc[Sink.forall](Sink$) { scala="#forall[T](p:T=%3EBoolean):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
```
Can you update this? It can be simpler, And I have been verifyed locally.
```
[info] compiling 81 Scala sources and 1 Java source to /Users/roiocam/IdeaProject/incubator-pekko/bench-jmh/target/scala-2.13/classes ...
[warn] Duplicate anchor [#bug-fixes] on [release-notes/index.html]
[success] Total time: 240 s (04:00), completed 2024-1-27 21:36:02
```
##########
docs/src/test/scala/docs/stream/operators/sink/ForAll.scala:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package docs.stream.operators.sink
+
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.stream.scaladsl.{ Sink, Source }
+
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.{ Await, ExecutionContextExecutor, Future }
+
+object ForAll {
+ implicit val system: ActorSystem = ???
+ implicit val ec: ExecutionContextExecutor = system.dispatcher
+ def foldExample: Unit = {
+ // #forall
+ val result: Future[Boolean] =
+ Source(1 to 100)
+ .runWith(Sink.forall(_ <= 100))
+ val allMatch = Await.result(result, 3.seconds)
+ println(allMatch)
+ // Expect prints:
+ // true
+ // #forall
Review Comment:
Same assert issue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1468515300
##########
docs/src/test/java/jdocs/stream/operators/sink/ForAll.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package jdocs.stream.operators.sink;
+
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.stream.javadsl.Sink;
+import org.apache.pekko.stream.javadsl.Source;
+
+import java.util.concurrent.TimeUnit;
+
+public class ForAll {
+ private ActorSystem system = null;
+
+ public void forAllUsage() throws Exception {
+ // #forall
+ final boolean allMatch =
+ Source.range(1, 100)
+ .runWith(Sink.forall(elem -> elem <= 100), system)
+ .toCompletableFuture()
+ .get(3, TimeUnit.SECONDS);
+ System.out.println(allMatch);
Review Comment:
This is for the doc.
##########
docs/src/test/scala/docs/stream/operators/sink/ForAll.scala:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package docs.stream.operators.sink
+
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.stream.scaladsl.{ Sink, Source }
+
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.{ Await, ExecutionContextExecutor, Future }
+
+object ForAll {
+ implicit val system: ActorSystem = ???
+ implicit val ec: ExecutionContextExecutor = system.dispatcher
+ def foldExample: Unit = {
+ // #forall
+ val result: Future[Boolean] =
+ Source(1 to 100)
+ .runWith(Sink.forall(_ <= 100))
+ val allMatch = Await.result(result, 3.seconds)
+ println(allMatch)
+ // Expect prints:
+ // true
+ // #forall
Review Comment:
this is for the doc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460383732
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1792,6 +1792,30 @@ trait FlowOps[+Out, +Mat] {
*/
def foldAsync[T](zero: T)(f: (T, Out) => Future[T]): Repr[T] = via(new FoldAsync(zero, f))
+ /**
+ * implemented by `Fold`, only emits its result when the upstream completes,
+ * after which it also completes. Applies the given function towards initialized `true`
+ * and next predicate result, yielding the next predicate result.
+ *
+ * If the function `predicate` throws an exception and the supervision decision is
+ * [[pekko.stream.Supervision.Restart]] current value starts at `true` again
+ * the stream will continue.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * See also [[FlowOps.fold]]
+ */
+ def forall(predicate: Out => Boolean): Repr[Boolean] =
+ via(Fold(true, (acc: Boolean, element) => acc && predicate(element)))
Review Comment:
add `.name("forAll")`, impelment it with `fold` has once drawback, which is we can't do a quick cancel after any elemement doesn't match the predicate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1907277142
@laglangyue oh my god, sad to hear that, take care. I can help with this, I will back home soon and do my surgical operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1461300880
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,36 @@
+# Sink.forall
+
+Apply a predicate function to assert each element received, it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall(predicate:T=>U):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
Review Comment:
```suggestion
@apidoc[Sink.forall](Sink$) { scala="#forall(predicate:T=>U):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1902037198
@laglangyue Thanks and welcome the pr from China scala user group. ping me if you think it's ready.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460744210
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -0,0 +1,36 @@
+# Sink.forall
+
+Apply a predicate function to assert each element received, it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.forall](Sink$) { scala="#forall(predicate:T=>U):org.apache.pekko.stream.scaladsl.Sink[T,scala.concurrent.Future[Boolean]]" java="#forall(org.apache.pekko.japi.function.Predicate)" }
+
+## Description
+forall applies a predicate function to assert each element received, it returns true if all elements satisfy the assertion, otherwise it returns false.
+
+It materializes into a `Future` (in Scala) or a `CompletionStage` (in Java) that completes with the last state when the stream has finished.
+
+notes: if source is empty, it will return true
Review Comment:
Notes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460745082
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala:
##########
@@ -69,6 +69,20 @@ object Sink {
f: function.Function2[U, In, CompletionStage[U]]): javadsl.Sink[In, CompletionStage[U]] =
new Sink(scaladsl.Sink.foldAsync[U, In](zero)(f(_, _).asScala).toCompletionStage())
+ /**
+ * A `Sink` that will invoke the given function for every received element, giving it its previous
+ * output (or the given `false`) and the element as input.
+ * * The returned [[scala.concurrent.Future]] will be completed with value of the final
Review Comment:
Java CompeletionStage
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1902037438
If you like , we can split the `forAllAsync` to a separate PR.
and for formating, you can use `scala-cli fmt` which is quit faster than the `sbt scalafmtAll`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add forall/forallAsync [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460483475
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1792,6 +1792,30 @@ trait FlowOps[+Out, +Mat] {
*/
def foldAsync[T](zero: T)(f: (T, Out) => Future[T]): Repr[T] = via(new FoldAsync(zero, f))
+ /**
+ * implemented by `Fold`, only emits its result when the upstream completes,
+ * after which it also completes. Applies the given function towards initialized `true`
+ * and next predicate result, yielding the next predicate result.
+ *
+ * If the function `predicate` throws an exception and the supervision decision is
+ * [[pekko.stream.Supervision.Restart]] current value starts at `true` again
+ * the stream will continue.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' upstream completes
+ *
+ * '''Cancels when''' downstream cancels
+ *
+ * See also [[FlowOps.fold]]
+ */
+ def forall(predicate: Out => Boolean): Repr[Boolean] =
+ via(Fold(true, (acc: Boolean, element) => acc && predicate(element)))
Review Comment:
empty stream shoud true or false?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1907275881
Friendly ping, we are scheduling a 1.1.0-M1 release soon, would you like to work on this this weekend, thanks. @laglangyue
And you can try to use the `foldWhile` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#discussion_r1460744604
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java:
##########
@@ -224,4 +225,20 @@ public void sinkForeachMustBeDocumented()
// #foreach
assertEquals(Done.done(), done);
}
+
+ @Test
+ public void sinkForallMustBeDocumented()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ // #forall
+ Sink<Integer, CompletionStage<Boolean>> forallSink =
+ Sink.forall((Predicate<Integer>) param -> param > 0);
Review Comment:
`Sink.<Integer>(param -> param > 0)` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] [pekko-streams] feat: add Sink.forall [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1905614739
This PR has been delayed for a week, and my father passed away due to cancer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] feat: Add Sink#forall operator. [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #989:
URL: https://github.com/apache/incubator-pekko/pull/989#issuecomment-1913243521
@pjfanning I'm still working on it...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org