This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 6e32da242f chore: fix limit and limitWeighted Scaladoc (#3591)
6e32da242f is described below
commit 6e32da242f7636ba284859abd8d53269d8d4c8e6
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 9 06:42:56 2026 +0100
chore: fix limit and limitWeighted Scaladoc (#3591)
Motivation:
The Scaladoc of limit and limitWeighted does not match the code:
- the javadsl Flow, Source, SubFlow and SubSource versions were copied
from take: they say the stream completes when "the defined number of
elements has been taken" and completes without producing elements
when n is zero or negative, but these operators fail the stream with
StreamLimitReachedException once the limit is exceeded
- the scaladsl limitWeighted "Completes when" line talks about the
number of emitted elements rather than the accumulated cost
- all of them name the exception StreamLimitException, which does not
exist; LimitWeighted fails with StreamLimitReachedException
Modification:
Describe the actual behaviour in the Emits/Completes/Errors/Cancels
lines of limit and limitWeighted in scaladsl Flow and javadsl Flow,
Source, SubFlow and SubSource, drop the copied zero-or-negative
sentence, add the missing Errors lines and name
StreamLimitReachedException.
Result:
The Scaladoc matches the LimitWeighted stage (limit is limitWeighted
with a cost of 1 per element).
Tests:
- Not run - Scaladoc only
References:
Refs #3590
---
.../org/apache/pekko/stream/javadsl/Flow.scala | 26 +++++++++-------------
.../org/apache/pekko/stream/javadsl/Source.scala | 26 ++++++++++------------
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 26 ++++++++++------------
.../apache/pekko/stream/javadsl/SubSource.scala | 26 ++++++++++------------
.../org/apache/pekko/stream/scaladsl/Flow.scala | 10 ++++-----
5 files changed, 51 insertions(+), 63 deletions(-)
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
index 0a0d4cfdad..cab6d61922 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
@@ -1397,24 +1397,21 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
/**
* Ensure stream boundedness by limiting the number of elements from
upstream.
* If the number of incoming elements exceeds max, it will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the number of emitted elements has
not reached max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the number of emitted
elements has not reached max
*
- * '''Errors when''' the total number of incoming element exceeds max
+ * '''Errors when''' the total number of incoming elements exceeds max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
@@ -1424,26 +1421,23 @@ final class Flow[In, Out, Mat](delegate:
scaladsl.Flow[In, Out, Mat]) extends Gr
* Ensure stream boundedness by evaluating the cost of incoming elements
* using a cost function. Exactly how many elements will be allowed to
travel downstream depends on the
* evaluated cost of each element. If the accumulated cost exceeds max, it
will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
* Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
*
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the accumulated cost has not reached
max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the accumulated cost has not
reached max
*
- * '''Errors when''' when the accumulated cost exceeds max
+ * '''Errors when''' the accumulated cost exceeds max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 8ed3c33043..e172347ffe 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
@@ -3356,22 +3356,21 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
/**
* Ensure stream boundedness by limiting the number of elements from
upstream.
* If the number of incoming elements exceeds max, it will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the number of emitted elements has
not reached max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the number of emitted
elements has not reached max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Errors when''' the total number of incoming elements exceeds max
+ *
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
@@ -3381,24 +3380,23 @@ final class Source[Out, Mat](delegate:
scaladsl.Source[Out, Mat]) extends Graph[
* Ensure stream boundedness by evaluating the cost of incoming elements
* using a cost function. Exactly how many elements will be allowed to
travel downstream depends on the
* evaluated cost of each element. If the accumulated cost exceeds max, it
will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
* Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
*
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the accumulated cost has not reached
max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the accumulated cost has not
reached max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Errors when''' the accumulated cost exceeds max
+ *
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
index b4ecccb16c..23026c8bc3 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
@@ -790,22 +790,21 @@ final class SubFlow[In, Out, Mat](
/**
* Ensure stream boundedness by limiting the number of elements from
upstream.
* If the number of incoming elements exceeds max, it will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the number of emitted elements has
not reached max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the number of emitted
elements has not reached max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Errors when''' the total number of incoming elements exceeds max
+ *
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
@@ -815,24 +814,23 @@ final class SubFlow[In, Out, Mat](
* Ensure stream boundedness by evaluating the cost of incoming elements
* using a cost function. Exactly how many elements will be allowed to
travel downstream depends on the
* evaluated cost of each element. If the accumulated cost exceeds max, it
will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
* Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
*
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the accumulated cost has not reached
max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the accumulated cost has not
reached max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Errors when''' the accumulated cost exceeds max
+ *
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
index 724faa65df..b6227b0b20 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
@@ -795,22 +795,21 @@ final class SubSource[Out, Mat](
/**
* Ensure stream boundedness by limiting the number of elements from
upstream.
* If the number of incoming elements exceeds max, it will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the number of emitted elements has
not reached max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the number of emitted
elements has not reached max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Errors when''' the total number of incoming elements exceeds max
+ *
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
@@ -820,24 +819,23 @@ final class SubSource[Out, Mat](
* Ensure stream boundedness by evaluating the cost of incoming elements
* using a cost function. Exactly how many elements will be allowed to
travel downstream depends on the
* evaluated cost of each element. If the accumulated cost exceeds max, it
will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
* of this step.
*
- * The stream will be completed without producing any elements if `n` is zero
- * or negative.
- *
* Adheres to the [[ActorAttributes.SupervisionStrategy]] attribute.
*
- * '''Emits when''' the specified number of elements to take has not yet
been reached
+ * '''Emits when''' upstream emits and the accumulated cost has not reached
max
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' the defined number of elements has been taken or
upstream completes
+ * '''Completes when''' upstream completes and the accumulated cost has not
reached max
*
- * '''Cancels when''' the defined number of elements has been taken or
downstream cancels
+ * '''Errors when''' the accumulated cost exceeds max
+ *
+ * '''Cancels when''' downstream cancels
*
* See also [[Flow.take]], [[Flow.takeWithin]], [[Flow.takeWhile]]
*/
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
index 3021bb0e25..7a77d78996 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
@@ -1906,7 +1906,7 @@ trait FlowOps[+Out, +Mat] {
/**
* Ensure stream boundedness by limiting the number of elements from
upstream.
* If the number of incoming elements exceeds max, it will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
@@ -1918,7 +1918,7 @@ trait FlowOps[+Out, +Mat] {
*
* '''Completes when''' upstream completes and the number of emitted
elements has not reached max
*
- * '''Errors when''' the total number of incoming element exceeds max
+ * '''Errors when''' the total number of incoming elements exceeds max
*
* '''Cancels when''' downstream cancels
*
@@ -1930,7 +1930,7 @@ trait FlowOps[+Out, +Mat] {
* Ensure stream boundedness by evaluating the cost of incoming elements
* using a cost function. Exactly how many elements will be allowed to
travel downstream depends on the
* evaluated cost of each element. If the accumulated cost exceeds max, it
will signal
- * upstream failure `StreamLimitException` downstream.
+ * upstream failure `StreamLimitReachedException` downstream.
*
* Due to input buffering some elements may have been
* requested from upstream publishers that will then not be processed
downstream
@@ -1942,9 +1942,9 @@ trait FlowOps[+Out, +Mat] {
*
* '''Backpressures when''' downstream backpressures
*
- * '''Completes when''' upstream completes and the number of emitted
elements has not reached max
+ * '''Completes when''' upstream completes and the accumulated cost has not
reached max
*
- * '''Errors when''' when the accumulated cost exceeds max
+ * '''Errors when''' the accumulated cost exceeds max
*
* '''Cancels when''' downstream cancels
*
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
