Copilot commented on code in PR #3185:
URL: https://github.com/apache/pekko/pull/3185#discussion_r3487566344
##########
stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala:
##########
@@ -2501,12 +2501,16 @@ trait FlowOps[+Out, +Mat] {
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the upstream it will be
backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the
upstream"), which is grammatically incorrect and unclear. It should compare
upstream speed to downstream speed.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala:
##########
@@ -3969,12 +3969,16 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the upstream it will be
backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the
upstream"), which is grammatically incorrect and unclear. It should compare
upstream speed to downstream speed.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala:
##########
@@ -2566,12 +2566,16 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the upstream it will be
backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the
upstream"), which is grammatically incorrect and unclear. It should compare
upstream speed to downstream speed.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala:
##########
@@ -1859,12 +1859,16 @@ final class SubSource[Out, Mat](
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the upstream it will be
backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the
upstream"), which is grammatically incorrect and unclear. It should compare
upstream speed to downstream speed.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/FlowExpandSpec.scala:
##########
@@ -150,6 +151,108 @@ class FlowExpandSpec extends StreamSpec("""
source.sendNext(2).sendComplete()
sink.expectNext(2 -> 0).expectComplete()
}
+
+ "fail stream when expander throws and supervision is Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 5)
+ .expand(i => if (i == 3) throw ex else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.stoppingDecider))
+ .runWith(Sink.ignore)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "fail stream when expander throws and supervision defaults to Stop" in {
+ val ex = new RuntimeException("boom")
+ val result = Source(1 to 5)
+ .expand(i => if (i == 3) throw ex else Iterator.single(i))
+ .runWith(Sink.ignore)
+ result.failed.futureValue shouldBe ex
+ }
+
+ "resume and keep current extrapolation when expander throws" in {
+ val ex = new RuntimeException("boom")
+ val publisher = TestPublisher.probe[Int]()
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(publisher)
+ .expand(i => if (i == 1) Iterator(1, 10, 11) else if (i == 2) throw ex
else Iterator.single(i))
+
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ publisher.sendNext(1)
+ subscriber.requestNext(1)
+ publisher.sendNext(2) // throws in expander, element 2 is dropped under
Resume
+ subscriber.requestNext(10) // continue from current extrapolation
+ subscriber.cancel()
+ }
+
+ "complete immediately on upstream finish if expanded is true" in {
+ val subscriber = TestSubscriber.probe[Int]()
+
+ Source.single(1)
+ .expand(_ => Iterator(1, 2, 3))
+ .to(Sink.fromSubscriber(subscriber))
+ .run()
+
+ subscriber.requestNext(1) // expanded becomes true
+ subscriber.expectComplete() // Does it complete immediately?
Review Comment:
This inline test comment is phrased as a question, which makes the test
intent look uncertain. Consider removing it or rephrasing as a statement.
##########
stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala:
##########
@@ -1891,12 +1891,16 @@ final class SubFlow[In, Out, Mat](
* element until new element comes from the upstream. For example an expand
step might repeat the last element for
* the subscriber until it receives an update from upstream.
*
- * This element will never "drop" upstream elements as all elements go
through at least one extrapolation step.
- * This means that if the upstream is actually faster than the upstream it
will be backpressured by the downstream
- * subscriber.
+ * Under normal operation all upstream elements go through at least one
extrapolation step.
+ * If supervision drops a failed element, that element is not emitted.
+ * If the upstream is actually faster than the upstream it will be
backpressured by the downstream subscriber.
Review Comment:
The Scaladoc sentence compares "upstream" to itself ("faster than the
upstream"), which is grammatically incorrect and unclear. It should compare
upstream speed to downstream speed.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
