You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "GreyPlane (via GitHub)" <gi...@apache.org> on 2024/01/20 09:36:11 UTC
[PR] feat: +Flow.exists & Sink.exists [incubator-pekko]
GreyPlane opened a new pull request, #990:
URL: https://github.com/apache/incubator-pekko/pull/990
#973
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460791095
##########
stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala:
##########
@@ -70,6 +70,8 @@ class DslConsistencySpec extends AnyWordSpec with Matchers {
"andThenMat",
"isIdentity",
"withAttributes",
+ // We box Scala Boolean to Java Boolean for better user experience
+ "exists",
Review Comment:
Good!
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1465360502
##########
stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala:
##########
@@ -70,6 +70,8 @@ class DslConsistencySpec extends AnyWordSpec with Matchers {
"andThenMat",
"isIdentity",
"withAttributes",
+ // We box Scala Boolean to Java Boolean for better user experience
+ "exists",
Review Comment:
remove it, no longer needed
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460406664
##########
docs/src/main/paradox/stream/operators/Flow/exists.md:
##########
@@ -0,0 +1,37 @@
+# exists
+
+Emit true and complete the stream as soon as there's an element satisfy the *predicate*, or emit false when upstream
+complete and no element satisfied the *predicate*.
+
+@ref[Simple operators](../index.md#simple-operators)
+
+## Signature
+
+@apidoc[Flow.exists](Flow$) { scala="#exists[I](p:Out%20=%3E%20Boolean)" java="#exists[I](p:org.apache.pekko.japi.function.Predicate[I])" }
+
+## Description
+
+Emit true and complete the stream as soon as there's an element satisfy the *predicate*, or emit false when upstream
+complete and no element satisfied the *predicate*.
Review Comment:
Emits completes
##########
docs/src/main/paradox/stream/operators/Flow/exists.md:
##########
@@ -0,0 +1,37 @@
+# exists
+
+Emit true and complete the stream as soon as there's an element satisfy the *predicate*, or emit false when upstream
+complete and no element satisfied the *predicate*.
+
+@ref[Simple operators](../index.md#simple-operators)
+
+## Signature
+
+@apidoc[Flow.exists](Flow$) { scala="#exists[I](p:Out%20=%3E%20Boolean)" java="#exists[I](p:org.apache.pekko.japi.function.Predicate[I])" }
+
+## Description
+
+Emit true and complete the stream as soon as there's an element satisfy the *predicate*, or emit false when upstream
+complete and no element satisfied the *predicate*.
+
+## Examples
+
+Scala
+: @@snip [Exists.scala](/docs/src/test/scala/docs/stream/operators/flow/DiMap.scala) { #imports #exists }
+
+Java
+: @@snip [Exists.java](/docs/src/test/java/jdocs/stream/operators/flow/DiMap.java) { #imports #exists }
Review Comment:
Another file instead of Dimap.java？
##########
docs/src/main/paradox/stream/operators/Flow/exists.md:
##########
@@ -0,0 +1,37 @@
+# exists
+
+Emit true and complete the stream as soon as there's an element satisfy the *predicate*, or emit false when upstream
+complete and no element satisfied the *predicate*.
+
+@ref[Simple operators](../index.md#simple-operators)
+
+## Signature
+
+@apidoc[Flow.exists](Flow$) { scala="#exists[I](p:Out%20=%3E%20Boolean)" java="#exists[I](p:org.apache.pekko.japi.function.Predicate[I])" }
+
+## Description
+
+Emit true and complete the stream as soon as there's an element satisfy the *predicate*, or emit false when upstream
+complete and no element satisfied the *predicate*.
+
+## Examples
+
+Scala
+: @@snip [Exists.scala](/docs/src/test/scala/docs/stream/operators/flow/DiMap.scala) { #imports #exists }
+
+Java
+: @@snip [Exists.java](/docs/src/test/java/jdocs/stream/operators/flow/DiMap.java) { #imports #exists }
+
+## Reactive Streams semantics
+
+@@@div { .callout }
+
+**emits** when predicate returned true or upstream completes
+
+**backpressures** '''Backpressures when'' downstream backpressures
+
+**completes** when predicate returned true or upstream completes
Review Comment:
Return false
##########
docs/src/main/paradox/stream/operators/Flow/exists.md:
##########
@@ -0,0 +1,37 @@
+# exists
+
+Emit true and complete the stream as soon as there's an element satisfy the *predicate*, or emit false when upstream
Review Comment:
Emits ,complete.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460388271
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1527,6 +1527,28 @@ trait FlowOps[+Out, +Mat] {
*/
def dropWhile(p: Out => Boolean): Repr[Out] = via(DropWhile(p))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: Out => Boolean): Repr[Boolean] = map(p).takeWhile(!_, inclusive = true).fold(true) {
Review Comment:
Sorry, the `collectFirst` is not merged yet.refs : https://github.com/apache/incubator-pekko/pull/984
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1465369034
##########
project/StreamOperatorsIndexGenerator.scala:
##########
@@ -111,7 +112,8 @@ object StreamOperatorsIndexGenerator extends AutoPlugin {
"fromGraph",
"actorSubscriber",
"foldAsync",
- "newOnCompleteStage"))
+ "newOnCompleteStage",
+ "exists"))
Review Comment:
I think this is not needed, as 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.exists operator [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1470620029
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -44,4 +44,4 @@ Java
**backpressures** when the invocation of predicate `p` has not yet completed
-@@@
\ No newline at end of file
+@@@
Review Comment:
A small issue, seemingly unrelated change
--
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.exists operator [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1470624960
##########
docs/src/main/paradox/stream/operators/Sink/forall.md:
##########
@@ -44,4 +44,4 @@ Java
**backpressures** when the invocation of predicate `p` has not yet completed
-@@@
\ No newline at end of file
+@@@
Review Comment:
Add a bland endline here.
--
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.exists operator [incubator-pekko]
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1465824483
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -318,6 +318,28 @@ public boolean test(Integer elem) {
future.toCompletableFuture().get(3, TimeUnit.SECONDS);
}
+ @Test
+ public void mustBeAbleToUseExists() throws Exception {
+ final TestKit probe = new TestKit(system);
+ final Source<Integer, NotUsed> source = Source.from(Arrays.asList(3, 4, 5));
+ final Flow<Integer, Boolean, NotUsed> flow =
+ Flow.of(Integer.class)
+ .exists(
+ new Predicate<Integer>() {
+ public boolean test(Integer elem) {
+ return elem > 3;
+ }
+ });
Review Comment:
```suggestion
final Flow<Integer, Boolean, NotUsed> flow = Flow.of(Integer.class).exists(elem -> elem > 3);
```
Would be great that use lambda?
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#issuecomment-1914555137
I will take care of it, as I discussed with @GreyPlane.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460386525
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1527,6 +1527,28 @@ trait FlowOps[+Out, +Mat] {
*/
def dropWhile(p: Out => Boolean): Repr[Out] = via(DropWhile(p))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: Out => Boolean): Repr[Boolean] = map(p).takeWhile(!_, inclusive = true).fold(true) {
Review Comment:
`map` and `takeWhile` can be expressed with `collectWhile`/`collectFirst`?
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "GreyPlane (via GitHub)" <gi...@apache.org>.
GreyPlane commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460729901
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExistsSpec.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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._
+import org.apache.pekko.stream.testkit.scaladsl._
+import org.scalatest.concurrent.ScalaFutures
+
+class FlowExistsSpec extends StreamSpec with ScalaFutures {
+
+ "An Exists" must {
+
+ def greaterThan3(i: Int) = i > 3
+
+ "emit true for empty Source" in {
Review Comment:
After talking to fs2 people on discord, it seems like that fs2 returns true was a historical mistake of implementation restriction.
So yes, we should change our implementation to return false for empty stream.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExistsSpec.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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._
+import org.apache.pekko.stream.testkit.scaladsl._
+import org.scalatest.concurrent.ScalaFutures
+
+class FlowExistsSpec extends StreamSpec with ScalaFutures {
+
+ "An Exists" must {
+
+ def greaterThan3(i: Int) = i > 3
+
+ "emit true for empty Source" in {
Review Comment:
After talking to fs2 people on discord, it seems like that fs2 returns true was a historical mistake or implementation restriction.
So yes, we should change our implementation to return false for empty stream.
--
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.exists operator [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#issuecomment-1916123759
LGTM, although has an irrelevant 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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#issuecomment-1907275495
Friendly ping, we are scheduling a 1.1.0-M1 release soon, would you like to work on this this weekend, thanks. @GreyPlane
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460386759
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1527,6 +1527,28 @@ trait FlowOps[+Out, +Mat] {
*/
def dropWhile(p: Out => Boolean): Repr[Out] = via(DropWhile(p))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: Out => Boolean): Repr[Boolean] = map(p).takeWhile(!_, inclusive = true).fold(true) {
+ case (_, incoming) => incoming
Review Comment:
need a `name("exists")` for this subFlow.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "GreyPlane (via GitHub)" <gi...@apache.org>.
GreyPlane commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1461316029
##########
stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala:
##########
@@ -41,8 +41,9 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
"futureFlow", // completionStageFlow
"futureSink", // completionStageSink
"lazyFutureSink", // lazyCompletionStageSink
- "createGraph" // renamed/overload of create for getting type inference working in Scala 3
- )
+ "createGraph", // renamed/overload of create for getting type inference working in Scala 3
+ // We box Scala Boolean to Java Boolean for better user experience
+ "exists")
Review Comment:
Sure, but it feels like a separate PR just like these works on doc IMO
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1468806419
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -1835,6 +1835,27 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
*/
def dropWhile(p: function.Predicate[Out]): javadsl.Flow[In, Out, Mat] = new Flow(delegate.dropWhile(p.test))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: function.Predicate[Out]): javadsl.Flow[In, java.lang.Boolean, Mat] =
+ new Flow(delegate.exists(p.test).map(Boolean.box))
Review Comment:
if we plan to add this to `flow/source`, then `forall` operator will need to do the same.
Currently, in ZIO, these method only exists on `ZSink`, but in fs2, these method exists on `Stream`.
Maybe we can just have these method on Sink for now too, wdyt, @GreyPlane @laglangyue @pjfanning @mdedetrich
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "GreyPlane (via GitHub)" <gi...@apache.org>.
GreyPlane commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1468805850
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -3543,6 +3543,27 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
*/
def dropWhile(p: function.Predicate[Out]): javadsl.Source[Out, Mat] = new Source(delegate.dropWhile(p.test))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: function.Predicate[Out]): Source[java.lang.Boolean, Mat] =
Review Comment:
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] feat: Add Sink.exists operator [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "GreyPlane (via GitHub)" <gi...@apache.org>.
GreyPlane commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460729901
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExistsSpec.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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._
+import org.apache.pekko.stream.testkit.scaladsl._
+import org.scalatest.concurrent.ScalaFutures
+
+class FlowExistsSpec extends StreamSpec with ScalaFutures {
+
+ "An Exists" must {
+
+ def greaterThan3(i: Int) = i > 3
+
+ "emit true for empty Source" in {
Review Comment:
After talking to fs2 people on discord, it seems like that fs2 returns false was a historical mistake of implementation restriction.
So yes, we should change our implementation to return false for empty stream.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#issuecomment-1902038538
First:
Thanks and welcome contributions from China user group.
What still missing, the `exists.md` , the JavaTest, the Demo code, which is needed in the `exists.md`. and in `exists.md`, will need an urlencoded for the scaladoc function signature.
For formating, you can try `scala-cli fmt` , which is fast.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "GreyPlane (via GitHub)" <gi...@apache.org>.
GreyPlane commented on PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#issuecomment-1902553374
docs was removed for now, in favor to fire another PR for adding Sink-or-Flow category for document.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1468804246
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -3543,6 +3543,27 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
*/
def dropWhile(p: function.Predicate[Out]): javadsl.Source[Out, Mat] = new Source(delegate.dropWhile(p.test))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: function.Predicate[Out]): Source[java.lang.Boolean, Mat] =
Review Comment:
`@since 1.1.0` needed on al new public methods
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460409620
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExistsSpec.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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._
+import org.apache.pekko.stream.testkit.scaladsl._
+import org.scalatest.concurrent.ScalaFutures
+
+class FlowExistsSpec extends StreamSpec with ScalaFutures {
+
+ "An Exists" must {
+
+ def greaterThan3(i: Int) = i > 3
+
+ "emit true for empty Source" in {
Review Comment:
I just checked fs2 where return true for empty stream.
--
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.exists operator [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#issuecomment-1916163782
As I have rewritten most of the code, I think we can bring this in, and @GreyPlane Thank you for your time.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "laglangyue (via GitHub)" <gi...@apache.org>.
laglangyue commented on PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#issuecomment-1902038312
add test for javadsl, add document
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "GreyPlane (via GitHub)" <gi...@apache.org>.
GreyPlane commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460386646
##########
docs/src/test/java/jdocs/stream/operators/flow/Exists.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.flow;
+
+// #imports
+
+import org.apache.pekko.NotUsed;
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.japi.function.Predicate;
+import org.apache.pekko.japi.function.Procedure;
+import org.apache.pekko.stream.javadsl.Flow;
+import org.apache.pekko.stream.javadsl.Source;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+// #imports
+
+public class Exists {
+ private static final ActorSystem system = null;
+
+ private void detectAnomaly() {
+ // #exists
+ final Source<String, NotUsed> source = Source.from(Arrays.asList("Sun is shining", "Unidentified Object", "River is flowing"));
+
+ List<String> anomalies = Collections.singletonList("Unidentified Object");
+ Predicate<String> isAnomaly = new Predicate<String>() {
+ @Override
+ public boolean test(String phenomenon) {
+ return anomalies.contains(phenomenon);
+ }
+ };
+
+ source
+ .via(Flow.<String>create().exists(isAnomaly))
+ .runForeach(new Procedure<Boolean>() {
+ @Override
+ public void apply(Boolean param) throws Exception {
+ System.out.println("Anomaly detected!");
+ }
+ }, system);
+ // expected prints:
+ // Anomaly detected!
+ // #exists
Review Comment:
Thanks for point it out, I didn't know that is a thing.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460385960
##########
docs/src/test/java/jdocs/stream/operators/flow/Exists.java:
##########
@@ -0,0 +1,61 @@
+/*
+ * 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.flow;
+
+// #imports
+
+import org.apache.pekko.NotUsed;
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.japi.function.Predicate;
+import org.apache.pekko.japi.function.Procedure;
+import org.apache.pekko.stream.javadsl.Flow;
+import org.apache.pekko.stream.javadsl.Source;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+// #imports
+
+public class Exists {
+ private static final ActorSystem system = null;
+
+ private void detectAnomaly() {
+ // #exists
+ final Source<String, NotUsed> source = Source.from(Arrays.asList("Sun is shining", "Unidentified Object", "River is flowing"));
+
+ List<String> anomalies = Collections.singletonList("Unidentified Object");
+ Predicate<String> isAnomaly = new Predicate<String>() {
+ @Override
+ public boolean test(String phenomenon) {
+ return anomalies.contains(phenomenon);
+ }
+ };
+
+ source
+ .via(Flow.<String>create().exists(isAnomaly))
+ .runForeach(new Procedure<Boolean>() {
+ @Override
+ public void apply(Boolean param) throws Exception {
+ System.out.println("Anomaly detected!");
+ }
+ }, system);
+ // expected prints:
+ // Anomaly detected!
+ // #exists
Review Comment:
need format with `javafmtAll`?
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1468806674
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1527,6 +1527,28 @@ trait FlowOps[+Out, +Mat] {
*/
def dropWhile(p: Out => Boolean): Repr[Out] = via(DropWhile(p))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: Out => Boolean): Repr[Boolean] = map(p).takeWhile(!_, inclusive = true).fold(true) {
+ case (_, incoming) => incoming
Review Comment:
in the Flow/Source, the name attributes should placed in `DefaultAttributes`.
And you will need via( new Fold...withAttributes(...)`.
--
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.exists operator [incubator-pekko]
Posted by "Roiocam (via GitHub)" <gi...@apache.org>.
Roiocam commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1465824483
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -318,6 +318,28 @@ public boolean test(Integer elem) {
future.toCompletableFuture().get(3, TimeUnit.SECONDS);
}
+ @Test
+ public void mustBeAbleToUseExists() throws Exception {
+ final TestKit probe = new TestKit(system);
+ final Source<Integer, NotUsed> source = Source.from(Arrays.asList(3, 4, 5));
+ final Flow<Integer, Boolean, NotUsed> flow =
+ Flow.of(Integer.class)
+ .exists(
+ new Predicate<Integer>() {
+ public boolean test(Integer elem) {
+ return elem > 3;
+ }
+ });
Review Comment:
```suggestion
final Flow<Integer, Boolean, NotUsed> flow = Flow.of(Integer.class).exists(elem -> elem > 3);
```
Would be great that use lambda?
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "GreyPlane (via GitHub)" <gi...@apache.org>.
GreyPlane commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460389801
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -1527,6 +1527,28 @@ trait FlowOps[+Out, +Mat] {
*/
def dropWhile(p: Out => Boolean): Repr[Out] = via(DropWhile(p))
+ /**
+ * Emit true and complete the stream as soon as there's an element satisfy the predicate.
+ * Or emit false when upstream complete and no element satisfied the predicate.
+ *
+ * The stream will produce true for empty stream (i.e. completes before signalling any elements).
+ *
+ * The stream will never complete if there's no element satisfied the predicate and upstream isn't completed.
+ *
+ * Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
+ *
+ * '''Emits when''' predicate returned true or upstream completes
+ *
+ * '''Backpressures when''' downstream backpressures
+ *
+ * '''Completes when''' predicate returned true or upstream completes
+ *
+ * '''Cancels when''' predicate returned true or downstream cancels
+ */
+ def exists(p: Out => Boolean): Repr[Boolean] = map(p).takeWhile(!_, inclusive = true).fold(true) {
Review Comment:
Can't see obvious reason for use `collectWhile` I guess? `inclusive` makes `takeWhile` appropriate for reaching the semantic demand.
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460973722
##########
stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala:
##########
@@ -41,8 +41,9 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
"futureFlow", // completionStageFlow
"futureSink", // completionStageSink
"lazyFutureSink", // lazyCompletionStageSink
- "createGraph" // renamed/overload of create for getting type inference working in Scala 3
- )
+ "createGraph", // renamed/overload of create for getting type inference working in Scala 3
+ // We box Scala Boolean to Java Boolean for better user experience
+ "exists")
Review Comment:
Can this be fixed with a mapping in `scala -> java types`
##########
stream-tests/src/test/scala/org/apache/pekko/stream/DslFactoriesConsistencySpec.scala:
##########
@@ -41,8 +41,9 @@ class DslFactoriesConsistencySpec extends AnyWordSpec with Matchers {
"futureFlow", // completionStageFlow
"futureSink", // completionStageSink
"lazyFutureSink", // lazyCompletionStageSink
- "createGraph" // renamed/overload of create for getting type inference working in Scala 3
- )
+ "createGraph", // renamed/overload of create for getting type inference working in Scala 3
+ // We box Scala Boolean to Java Boolean for better user experience
+ "exists")
Review Comment:
Can this be fixed with a mapping in `scala -> java types`?
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460791095
##########
stream-tests/src/test/scala/org/apache/pekko/stream/DslConsistencySpec.scala:
##########
@@ -70,6 +70,8 @@ class DslConsistencySpec extends AnyWordSpec with Matchers {
"andThenMat",
"isIdentity",
"withAttributes",
+ // We box Scala Boolean to Java Boolean for better user experience
+ "exists",
Review Comment:
Good!
--
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: +Flow.exists & Sink.exists [incubator-pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #990:
URL: https://github.com/apache/incubator-pekko/pull/990#discussion_r1460409051
##########
docs/src/test/scala/docs/stream/operators/sink/Exists.scala:
##########
@@ -0,0 +1,46 @@
+/*
+ * 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
+
+//#imports
+import org.apache.pekko.actor.ActorSystem
+import org.apache.pekko.stream.scaladsl._
+
+import scala.concurrent.ExecutionContext
+//#imports
+
+object Exists {
+
+ implicit val system: ActorSystem = null
+ implicit val ec: ExecutionContext = system.dispatcher
+
+ def detectAnomaly(): Unit = {
+ // #exists
+ val source = Source(Seq("Sun is shining", "Unidentified Object", "River is flowing"))
+
+ val anomalies = Seq("Unidentified Object")
+ def isAnomaly(phenomenon: String): Boolean = anomalies.contains(phenomenon)
+
+ val result = source.runWith(Sink.exists(isAnomaly))
+ result.map(println)
Review Comment:
May need await? The thread may.exsits so quickly
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExistsSpec.scala:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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._
+import org.apache.pekko.stream.testkit.scaladsl._
+import org.scalatest.concurrent.ScalaFutures
+
+class FlowExistsSpec extends StreamSpec with ScalaFutures {
+
+ "An Exists" must {
+
+ def greaterThan3(i: Int) = i > 3
+
+ "emit true for empty Source" in {
Review Comment:
false？
##########
stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java:
##########
@@ -306,6 +306,28 @@ public boolean test(Integer elem) {
future.toCompletableFuture().get(3, TimeUnit.SECONDS);
}
+ @Test
+ public void mustBeAbleToUseExists() throws Exception {
+ final TestKit probe = new TestKit(system);
+ final Source<Integer, NotUsed> source = Source.from(Arrays.asList(3, 4, 5));
+ final Flow<Integer, Boolean, NotUsed> flow =
+ Flow.of(Integer.class)
+ .exists(
+ new Predicate<Integer>() {
+ public boolean test(Integer elem) {
+ return elem > 3;
+ }
+ });
+
+ final CompletionStage<Done> future =
+ source
+ .via(flow)
+ .runWith(Sink.foreach(elem -> probe.getRef().tell(elem, ActorRef.noSender())), system);
+
+ probe.expectMsgEquals(true);
+ future.toCompletableFuture().get(3, TimeUnit.SECONDS);
Review Comment:
Need an assertions
--
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