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 947459288d docs: fill stream operator doc gaps and add missing
overloads (#3590)
947459288d is described below
commit 947459288d6412cbd3e046e989009ce12f55ffdc
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 9 07:44:58 2026 +0100
docs: fill stream operator doc gaps and add missing overloads (#3590)
* docs: fill stream operator doc gaps and add missing overloads
Motivation:
Some stream operator pages have empty sections or stubs, several pages
list only some of an operator's overloads, and the operator index
generator still lists removed operators as pending.
Modification:
- Unzip.md, UnzipWith.md: fill the empty Signature sections
- Source/items.md: remove the empty Examples heading (no snippet exists)
- stream-flows-and-basics.md, stream-dynamic.md: remove empty
Introduction headings
- stream-refs.md: remove the Bulk Stream References stub; no such API
exists
- StreamOperatorsIndexGenerator.scala: drop the removed throttleEven,
actorPublisher and actorSubscriber from the pending lists
- Add missing overloads and describe their extra parameters:
dropRepeated(p), flatMapConcat(parallelism, f),
groupedWeightedWithin(maxWeight, maxNumber, d)(costFn),
takeWhile(p, inclusive), zipLatestWith(..., eagerComplete),
Compression.gzip/deflate level and autoFlush variants, and the
onErrorContinue predicate and Java variants
Result:
Operator pages list all overloads and have no empty sections.
Tests:
- Not run - docs only (plus removing stale names from the docs index
generator's pending lists)
References:
Refs #3589
* docs: add Signature section to the Balance operator page
Motivation:
operators/Balance.md has no Signature section, unlike the other
fan-out operator pages.
Modification:
Add a Signature section linking stream.*.Balance, in the same form as
Broadcast.md.
Result:
The Balance page links to its API docs.
Tests:
- Not run - docs only
References:
Refs #3589
* docs: fix stream docs whose text contradicts the code
Motivation:
A number of stream guide and operator pages describe behaviour,
parameters or types that do not match the code, often because text was
copied from a neighbouring operator.
Modification:
- Guides: Unzip name and MergeSequence output (stream-graphs), pull()
precondition (stream-customize), KillSwitch is a BidiFlow
(stream-dynamic), Java failure wording (stream-error), Scala vs Java
framing truncation (stream-io), NotUsed package (stream-quickstart),
drop the "founding member" claim (stream-introduction,
stream-flows-and-basics)
- Source pages: asSubscriber/fromPublisher signatures and links,
iterate parameter names, lazySource, maybe (completionStage), queue
completion
- Sink pages: queue backpressure, collect, completionStageSink,
futureSink, lazyCompletionStageSink, lazyFutureSink (sink not flow)
- Flow pages: flattenOptional description; remove the wrong second
"completes" line from completionStageFlow, futureFlow, lazyFlow,
lazyFutureFlow and lazyCompletionStageFlow, add missing "cancels"
- asJavaStream: elements come from upstream
- Source-or-Flow pages: alsoToAll attaches Sinks, batch has no costFn,
buffer fail case, delay(With) DelayOverflowStrategy, foldWhile
completion, interleave(All) downstream backpressure, limitWeighted
accumulated cost, mergeLatest example and eagerComplete,
onErrorContinue fails on unhandled errors, wireTap Sink variant
signatures, zipLatestWith combine function, Source.sliding signature
Result:
These pages match the code.
Tests:
- Not run - docs only
References:
Refs #3589
* docs: describe transferring large amounts of data with stream refs
Motivation:
An earlier commit in this PR removed the two-line "Bulk Stream
References" section on the grounds that no bulk stream ref API
exists. The section never described a separate API: it described
using SourceRef/SinkRef as a side-channel for large amounts of data,
which is a valid use case, but it was an unexplained stub.
Modification:
Add a "Transferring large amounts of data" subsection under Stream
References that explains the use case with the existing
StreamRefs.sourceRef()/sinkRef() API and lists the points that matter
for large transfers: back-pressure across the network, single-use
refs, no persistence or automatic resumption, elements must fit in
the remoting frame size (pekko.remote.artery.advanced.maximum-frame-size),
and the subscription timeout.
Result:
The use case is documented instead of being a stub or removed.
Tests:
- Not run - docs only
References:
Refs #3589
* docs: align iterate.md prose with the real p/f parameter names
---------
Co-authored-by: 虎鸣 <[email protected]>
---
docs/src/main/paradox/stream/operators/Balance.md | 4 ++++
.../stream/operators/Compression/deflate.md | 7 ++++++
.../paradox/stream/operators/Compression/gzip.md | 7 ++++++
.../stream/operators/Flow/completionStageFlow.md | 6 +++--
.../stream/operators/Flow/flattenOptional.md | 4 ++--
.../paradox/stream/operators/Flow/futureFlow.md | 2 --
.../operators/Flow/lazyCompletionStageFlow.md | 6 +++--
.../main/paradox/stream/operators/Flow/lazyFlow.md | 2 --
.../stream/operators/Flow/lazyFutureFlow.md | 2 --
.../main/paradox/stream/operators/Sink/collect.md | 2 +-
.../stream/operators/Sink/completionStageSink.md | 2 +-
.../paradox/stream/operators/Sink/futureSink.md | 2 +-
.../operators/Sink/lazyCompletionStageSink.md | 6 ++---
.../stream/operators/Sink/lazyFutureSink.md | 6 ++---
.../main/paradox/stream/operators/Sink/queue.md | 2 +-
.../stream/operators/Source-or-Flow/alsoToAll.md | 4 ++--
.../stream/operators/Source-or-Flow/batch.md | 4 ++--
.../stream/operators/Source-or-Flow/buffer.md | 4 +++-
.../stream/operators/Source-or-Flow/delay.md | 2 +-
.../stream/operators/Source-or-Flow/delayWith.md | 2 +-
.../operators/Source-or-Flow/dropRepeated.md | 5 ++++
.../operators/Source-or-Flow/flatMapConcat.md | 6 +++++
.../stream/operators/Source-or-Flow/foldWhile.md | 6 ++---
.../Source-or-Flow/groupedWeightedWithin.md | 2 ++
.../stream/operators/Source-or-Flow/interleave.md | 2 +-
.../operators/Source-or-Flow/interleaveAll.md | 2 +-
.../operators/Source-or-Flow/limitWeighted.md | 4 ++--
.../stream/operators/Source-or-Flow/mergeLatest.md | 6 ++---
.../operators/Source-or-Flow/onErrorContinue.md | 18 ++++++++++++++-
.../stream/operators/Source-or-Flow/sliding.md | 1 +
.../stream/operators/Source-or-Flow/takeWhile.md | 9 +++++---
.../stream/operators/Source-or-Flow/wireTap.md | 2 ++
.../operators/Source-or-Flow/zipLatestWith.md | 12 +++++++---
.../stream/operators/Source/asSubscriber.md | 10 ++++----
.../stream/operators/Source/fromPublisher.md | 8 +++----
.../main/paradox/stream/operators/Source/items.md | 2 --
.../paradox/stream/operators/Source/iterate.md | 12 +++++-----
.../paradox/stream/operators/Source/lazySource.md | 2 +-
.../main/paradox/stream/operators/Source/maybe.md | 4 ++--
.../main/paradox/stream/operators/Source/queue.md | 2 +-
.../operators/StreamConverters/asJavaStream.md | 2 +-
docs/src/main/paradox/stream/operators/Unzip.md | 2 ++
.../src/main/paradox/stream/operators/UnzipWith.md | 2 ++
docs/src/main/paradox/stream/stream-customize.md | 3 ++-
docs/src/main/paradox/stream/stream-dynamic.md | 4 +---
docs/src/main/paradox/stream/stream-error.md | 2 +-
.../main/paradox/stream/stream-flows-and-basics.md | 4 +---
docs/src/main/paradox/stream/stream-graphs.md | 4 ++--
.../src/main/paradox/stream/stream-introduction.md | 4 ++--
docs/src/main/paradox/stream/stream-io.md | 6 +++--
docs/src/main/paradox/stream/stream-quickstart.md | 2 +-
docs/src/main/paradox/stream/stream-refs.md | 27 ++++++++++++++++++----
project/StreamOperatorsIndexGenerator.scala | 3 ---
53 files changed, 163 insertions(+), 93 deletions(-)
diff --git a/docs/src/main/paradox/stream/operators/Balance.md
b/docs/src/main/paradox/stream/operators/Balance.md
index a5f3e1a3f8..c4601a3018 100644
--- a/docs/src/main/paradox/stream/operators/Balance.md
+++ b/docs/src/main/paradox/stream/operators/Balance.md
@@ -4,6 +4,10 @@ Fan-out the stream to several streams.
@ref[Fan-out operators](index.md#fan-out-operators)
+## Signature
+
+@apidoc[stream.*.Balance]
+
## Description
Fan-out the stream to several streams. Each upstream element is emitted to the
first available downstream consumer.
diff --git a/docs/src/main/paradox/stream/operators/Compression/deflate.md
b/docs/src/main/paradox/stream/operators/Compression/deflate.md
index 8276e88c8f..0d1bbdd150 100644
--- a/docs/src/main/paradox/stream/operators/Compression/deflate.md
+++ b/docs/src/main/paradox/stream/operators/Compression/deflate.md
@@ -7,6 +7,8 @@ Creates a flow that deflate-compresses a stream of ByteStrings.
## Signature
@apidoc[Compression.deflate](stream.*.Compression$) {
scala="#deflate:org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#deflate()" }
+@apidoc[Compression.deflate](stream.*.Compression$) {
scala="#deflate(level:Int,nowrap:Boolean):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#deflate(int,boolean)" }
+@apidoc[Compression.deflate](stream.*.Compression$) {
scala="#deflate(level:Int,nowrap:Boolean,autoFlush:Boolean):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#deflate(int,boolean,boolean)" }
## Description
@@ -17,6 +19,11 @@ come at a compression performance cost for very small chunks.
Use the overload method with parameters to control the compression level and
compatibility with GZip.
+Since 1.3.0, an overload with an additional `autoFlush` parameter is also
available. If `autoFlush` is `true` (the default
+for the other overloads), the compressor flushes after every single element in
the stream. If it is `false`, the
+compressor only emits output when the compression algorithm produces it, which
can improve the compression ratio for
+small chunks, but an emitted `ByteString` may not be fully decompressible
until more data arrives.
+
## Reactive Streams semantics
@@@div { .callout }
diff --git a/docs/src/main/paradox/stream/operators/Compression/gzip.md
b/docs/src/main/paradox/stream/operators/Compression/gzip.md
index 40958dc4bd..8a5a4702c2 100644
--- a/docs/src/main/paradox/stream/operators/Compression/gzip.md
+++ b/docs/src/main/paradox/stream/operators/Compression/gzip.md
@@ -7,6 +7,8 @@ Creates a flow that gzip-compresses a stream of ByteStrings.
## Signature
@apidoc[Compression.gzip](stream.*.Compression$) {
scala="#gzip:org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#gzip()" }
+@apidoc[Compression.gzip](stream.*.Compression$) {
scala="#gzip(level:Int):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#gzip(int)" }
+@apidoc[Compression.gzip](stream.*.Compression$) {
scala="#gzip(level:Int,autoFlush:Boolean):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#gzip(int,boolean)" }
## Description
@@ -17,6 +19,11 @@ come at a compression performance cost for very small chunks.
Use the overload method to control the compression level.
+Since 1.3.0, an overload with an `autoFlush` parameter is also available. If
`autoFlush` is `true` (the default for the
+other overloads), the compressor flushes after every single element in the
stream. If it is `false`, the compressor
+only emits output when the compression algorithm produces it, which can
improve the compression ratio for small
+chunks, but an emitted `ByteString` may not be fully decompressible until more
data arrives.
+
## Reactive Streams semantics
@@@div { .callout }
diff --git a/docs/src/main/paradox/stream/operators/Flow/completionStageFlow.md
b/docs/src/main/paradox/stream/operators/Flow/completionStageFlow.md
index 2e273be828..54187e0c4b 100644
--- a/docs/src/main/paradox/stream/operators/Flow/completionStageFlow.md
+++ b/docs/src/main/paradox/stream/operators/Flow/completionStageFlow.md
@@ -33,7 +33,9 @@ Scala
**completes** when upstream completes and all elements have been emitted from
the internal flow
-**completes** when upstream completes and all futures have been completed and
all elements have been emitted
-
+**cancels** when downstream cancels (keep reading)
+ The operator's default behavior in case of downstream cancellation before
nested flow materialization (`CompletionStage` completion) is to cancel
immediately.
+ This behavior can be controlled by setting the
[[org.apache.pekko.stream.Attributes.NestedMaterializationCancellationPolicy.PropagateToNested]]
attribute,
+ this will delay downstream cancellation until nested flow's
materialization which is then immediately cancelled (with the original
cancellation cause).
@@@
diff --git a/docs/src/main/paradox/stream/operators/Flow/flattenOptional.md
b/docs/src/main/paradox/stream/operators/Flow/flattenOptional.md
index 1ad5d8a983..91dd4c1bff 100644
--- a/docs/src/main/paradox/stream/operators/Flow/flattenOptional.md
+++ b/docs/src/main/paradox/stream/operators/Flow/flattenOptional.md
@@ -11,8 +11,8 @@ Collect the value of `Optional` from all the elements passing
through this flow
## Description
-Streams the elements through the given future flow once it successfully
completes.
-If the future fails the stream is failed.
+Creates a `Flow` that emits the value of each present
@javadoc[Optional](java.util.Optional) element passing through it,
+while empty `Optional` elements are filtered out (dropped).
## Reactive Streams semantics
diff --git a/docs/src/main/paradox/stream/operators/Flow/futureFlow.md
b/docs/src/main/paradox/stream/operators/Flow/futureFlow.md
index 9900472a61..c5ef3de2f0 100644
--- a/docs/src/main/paradox/stream/operators/Flow/futureFlow.md
+++ b/docs/src/main/paradox/stream/operators/Flow/futureFlow.md
@@ -34,8 +34,6 @@ Scala
**completes** when upstream completes and all elements have been emitted from
the internal flow
-**completes** when upstream completes and all futures have been completed and
all elements have been emitted
-
**cancels** when downstream cancels (keep reading)
The operator's default behavior in case of downstream cancellation before
nested flow materialization (future completion) is to cancel immediately.
This behavior can be controlled by setting the
[[org.apache.pekko.stream.Attributes.NestedMaterializationCancellationPolicy.PropagateToNested]]
attribute,
diff --git
a/docs/src/main/paradox/stream/operators/Flow/lazyCompletionStageFlow.md
b/docs/src/main/paradox/stream/operators/Flow/lazyCompletionStageFlow.md
index 52b8dcc517..8ea8ec83cd 100644
--- a/docs/src/main/paradox/stream/operators/Flow/lazyCompletionStageFlow.md
+++ b/docs/src/main/paradox/stream/operators/Flow/lazyCompletionStageFlow.md
@@ -29,7 +29,9 @@ See @ref:[lazyFlow](lazyFlow.md) for sample.
**completes** when upstream completes and all elements have been emitted from
the internal flow
-**completes** when upstream completes and all futures have been completed and
all elements have been emitted
-
+**cancels** when downstream cancels (keep reading)
+ The operator's default behavior in case of downstream cancellation before
nested flow materialization (`CompletionStage` completion) is to cancel
immediately.
+ This behavior can be controlled by setting the
[[org.apache.pekko.stream.Attributes.NestedMaterializationCancellationPolicy.PropagateToNested]]
attribute,
+ this will delay downstream cancellation until nested flow's
materialization which is then immediately cancelled (with the original
cancellation cause).
@@@
diff --git a/docs/src/main/paradox/stream/operators/Flow/lazyFlow.md
b/docs/src/main/paradox/stream/operators/Flow/lazyFlow.md
index 66972dbc47..1effeee481 100644
--- a/docs/src/main/paradox/stream/operators/Flow/lazyFlow.md
+++ b/docs/src/main/paradox/stream/operators/Flow/lazyFlow.md
@@ -61,8 +61,6 @@ all materialization and what is even worse, unsafely across
threads.
**completes** when upstream completes and all elements have been emitted from
the internal flow
-**completes** when upstream completes and all futures have been completed and
all elements have been emitted
-
**cancels** when downstream cancels (keep reading)
The operator's default behavior in case of downstream cancellation before
nested flow materialization (future completion) is to cancel immediately.
This behavior can be controlled by setting the
[[org.apache.pekko.stream.Attributes.NestedMaterializationCancellationPolicy.PropagateToNested]]
attribute,
diff --git a/docs/src/main/paradox/stream/operators/Flow/lazyFutureFlow.md
b/docs/src/main/paradox/stream/operators/Flow/lazyFutureFlow.md
index 537c0362dc..7a3ae5e063 100644
--- a/docs/src/main/paradox/stream/operators/Flow/lazyFutureFlow.md
+++ b/docs/src/main/paradox/stream/operators/Flow/lazyFutureFlow.md
@@ -33,8 +33,6 @@ See @ref:[lazyFlow](lazyFlow.md) for sample.
**completes** when upstream completes and all elements have been emitted from
the internal flow
-**completes** when upstream completes and all futures have been completed and
all elements have been emitted
-
**cancels** when downstream cancels (keep reading)
The operator's default behavior in case of downstream cancellation before
nested flow materialization (future completion) is to cancel immediately.
This behavior can be controlled by setting the
[[org.apache.pekko.stream.Attributes.NestedMaterializationCancellationPolicy.PropagateToNested]]
attribute,
diff --git a/docs/src/main/paradox/stream/operators/Sink/collect.md
b/docs/src/main/paradox/stream/operators/Sink/collect.md
index 55d4c1a694..6b8fc41259 100644
--- a/docs/src/main/paradox/stream/operators/Sink/collect.md
+++ b/docs/src/main/paradox/stream/operators/Sink/collect.md
@@ -15,7 +15,7 @@ which will be completed with a result of the Java
@javadoc[Collector](java.util.
## Example
-Given a stream of numbers we can collect the numbers into a collection with
the `seq` operator
+Given a stream of numbers we can collect the numbers into a `List` with the
`collect` operator and `Collectors.toList()`
Java
: @@snip
[SinkTest.java](/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java)
{ #collect-to-list }
diff --git a/docs/src/main/paradox/stream/operators/Sink/completionStageSink.md
b/docs/src/main/paradox/stream/operators/Sink/completionStageSink.md
index 16a1b2f5eb..916e8a2688 100644
--- a/docs/src/main/paradox/stream/operators/Sink/completionStageSink.md
+++ b/docs/src/main/paradox/stream/operators/Sink/completionStageSink.md
@@ -7,7 +7,7 @@ Streams the elements to the given future sink once it
successfully completes.
## Description
-Streams the elements through the given future flow once it successfully
completes.
+Streams the elements through the given future sink once it successfully
completes.
If the future fails the stream is failed.
`completionStageSink` uses the same lazy materialization semantics as
diff --git a/docs/src/main/paradox/stream/operators/Sink/futureSink.md
b/docs/src/main/paradox/stream/operators/Sink/futureSink.md
index 45e54fab76..52942ccaa3 100644
--- a/docs/src/main/paradox/stream/operators/Sink/futureSink.md
+++ b/docs/src/main/paradox/stream/operators/Sink/futureSink.md
@@ -11,7 +11,7 @@ Streams the elements to the given future sink once it
successfully completes.
## Description
-Streams the elements through the given future flow once it successfully
completes.
+Streams the elements through the given future sink once it successfully
completes.
If the future fails the stream is failed.
`futureSink` uses the same lazy materialization semantics as
@ref:[lazyFutureSink](lazyFutureSink.md): the nested sink
diff --git
a/docs/src/main/paradox/stream/operators/Sink/lazyCompletionStageSink.md
b/docs/src/main/paradox/stream/operators/Sink/lazyCompletionStageSink.md
index 14b66218e1..545378427f 100644
--- a/docs/src/main/paradox/stream/operators/Sink/lazyCompletionStageSink.md
+++ b/docs/src/main/paradox/stream/operators/Sink/lazyCompletionStageSink.md
@@ -9,10 +9,10 @@ Defers creation and materialization of a `Sink` until there
is a first element.
When the first element comes from upstream the `CompletionStage<Sink>` is
created. When that completes successfully with a sink
that is materialized and inserted in the stream.
-The internal `Sink` will not be created if the stream completes of fails
before any element got through.
+The internal `Sink` will not be created if the stream completes or fails
before any element got through.
-The materialized value of the `Sink` will be the materialized value of the
created internal flow if it is materialized
-and failed with a `org.apache.pekko.stream.NeverMaterializedException` if the
stream fails or completes without the flow being materialized.
+The materialized value of the `Sink` will be the materialized value of the
created internal sink if it is materialized
+and failed with a `org.apache.pekko.stream.NeverMaterializedException` if the
stream fails or completes without the sink being materialized.
Can be combined with @ref:[prefixAndTail](../Source-or-Flow/prefixAndTail.md)
to base the sink on the first element.
diff --git a/docs/src/main/paradox/stream/operators/Sink/lazyFutureSink.md
b/docs/src/main/paradox/stream/operators/Sink/lazyFutureSink.md
index 0be83b8005..4a93387350 100644
--- a/docs/src/main/paradox/stream/operators/Sink/lazyFutureSink.md
+++ b/docs/src/main/paradox/stream/operators/Sink/lazyFutureSink.md
@@ -13,10 +13,10 @@ Defers creation and materialization of a `Sink` until there
is a first element.
When the first element comes from upstream the `Future[Sink]` is created. When
that completes successfully with a sink
that is materialized and inserted in the stream.
-The internal `Sink` will not be created if the stream completes of fails
before any element got through.
+The internal `Sink` will not be created if the stream completes or fails
before any element got through.
-The materialized value of the `Sink` will be the materialized value of the
created internal flow if it is materialized
-and failed with a `org.apache.pekko.stream.NeverMaterializedException` if the
stream fails or completes without the flow being materialized.
+The materialized value of the `Sink` will be the materialized value of the
created internal sink if it is materialized
+and failed with a `org.apache.pekko.stream.NeverMaterializedException` if the
stream fails or completes without the sink being materialized.
Can be combined with @ref:[prefixAndTail](../Source-or-Flow/prefixAndTail.md)
to base the sink on the first element.
diff --git a/docs/src/main/paradox/stream/operators/Sink/queue.md
b/docs/src/main/paradox/stream/operators/Sink/queue.md
index 4531d56073..0c0f231ac9 100644
--- a/docs/src/main/paradox/stream/operators/Sink/queue.md
+++ b/docs/src/main/paradox/stream/operators/Sink/queue.md
@@ -21,7 +21,7 @@ a buffer in case stream emitting elements faster than queue
pulling them.
**cancels** when `SinkQueue.cancel` is called
-**backpressures** when buffer has some space
+**backpressures** when the buffer is full
@@@
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoToAll.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoToAll.md
index 8a75a87a2f..756851d8c5 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoToAll.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/alsoToAll.md
@@ -1,6 +1,6 @@
# alsoToAll
-Attaches the given @apidoc[Source]s to this @apidoc[Flow], meaning that
elements that pass through this @apidoc[Flow] will also be sent to all those
@apidoc[Sink]s.
+Attaches the given @apidoc[Sink]s to this @apidoc[Flow], meaning that elements
that pass through this @apidoc[Flow] will also be sent to all those
@apidoc[Sink]s.
@ref[Fan-out operators](../index.md#fan-out-operators)
@@ -11,7 +11,7 @@ Attaches the given @apidoc[Source]s to this @apidoc[Flow],
meaning that elements
## Description
-Attaches the given @apidoc[Source] s to this @apidoc[Flow], meaning that
elements that pass through this @apidoc[Flow]
+Attaches the given @apidoc[Sink]s to this @apidoc[Flow], meaning that elements
that pass through this @apidoc[Flow]
will also be sent to all those @apidoc[Sink]s.
## Reactive Streams semantics
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/batch.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/batch.md
index 08646f7c40..0b7fe56c8a 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/batch.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/batch.md
@@ -23,8 +23,8 @@ to the summary type.
Will eagerly pull elements, this behavior may result in a single pending (i.e.
buffered) element which cannot be
aggregated to the batched value.
-This operator adheres to the `ActorAttributes.SupervisionStrategy` attribute
for exceptions thrown by the `costFn`,
-`seed`, or `aggregate` functions. On `Supervision.Stop` (the default) the
stream fails. On `Supervision.Resume` the
+This operator adheres to the `ActorAttributes.SupervisionStrategy` attribute
for exceptions thrown by the `seed` or
+`aggregate` functions. On `Supervision.Stop` (the default) the stream fails.
On `Supervision.Resume` the
offending element is dropped and the accumulated batch (if any) is preserved.
On `Supervision.Restart` the offending
element is dropped, the accumulated batch is discarded, and the operator
starts fresh.
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/buffer.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/buffer.md
index 44d78ba26e..858c0b9bca 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/buffer.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/buffer.md
@@ -28,7 +28,9 @@ handled according to the specified `OverflowStrategy`:
**backpressures** when `OverflowStrategy` is `backpressure` and buffer is full
-**completes** when upstream completes and buffered elements has been drained,
or when `OverflowStrategy` is `fail`, the buffer is full and a new element
arrives
+**completes** when upstream completes and buffered elements has been drained
+
+**fails** when `OverflowStrategy` is `fail`, the buffer is full and a new
element arrives
@@@
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/delay.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/delay.md
index a621920f95..d43c6a7a39 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/delay.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/delay.md
@@ -20,7 +20,7 @@ Delay every element passed through with a specific duration.
**emits** when there is a pending element in the buffer and configured time
for this element elapsed
-**backpressures** differs, depends on `OverflowStrategy` set
+**backpressures** differs, depends on `DelayOverflowStrategy` set
**completes** when upstream completes and buffered elements have been drained
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/delayWith.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/delayWith.md
index 216d861bfe..f45cd2d8f7 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/delayWith.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/delayWith.md
@@ -21,7 +21,7 @@ This operator adheres to the
ActorAttributes.SupervisionStrategy attribute. On `
**emits** when there is a pending element in the buffer and configured time
for this element elapsed
-**backpressures** differs, depends on `OverflowStrategy` set
+**backpressures** differs, depends on `DelayOverflowStrategy` set
**completes** when upstream completes and buffered elements have been drained
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/dropRepeated.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/dropRepeated.md
index 7299419b24..61af7b1f0f 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/dropRepeated.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/dropRepeated.md
@@ -8,12 +8,17 @@ Only pass on those elements that are distinct from the
previous element.
@apidoc[Source.dropRepeated](Source) {
scala="#dropRepeated():FlowOps.this.Repr[Out]" java="#dropRepeated()" }
@apidoc[Flow.dropRepeated](Flow) {
scala="#dropRepeated():FlowOps.this.Repr[Out]" java="#dropRepeated()" }
+@apidoc[Source.dropRepeated](Source) {
scala="#dropRepeated(p:(Out,Out)=>Boolean):FlowOps.this.Repr[Out]"
java="#dropRepeated(org.apache.pekko.japi.function.Function2)" }
+@apidoc[Flow.dropRepeated](Flow) {
scala="#dropRepeated(p:(Out,Out)=>Boolean):FlowOps.this.Repr[Out]"
java="#dropRepeated(org.apache.pekko.japi.function.Function2)" }
## Description
Only pass on those elements that are distinct from the previous element.
+By default elements are compared using equality. An overload accepting a
predicate `p` can be used instead,
+in which case an element is dropped when `p` returns `true` for the previous
element and the current element.
+
## Example
For example, given a `Source` of numbers, we just want to pass distinct
numbers downstream:
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/flatMapConcat.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/flatMapConcat.md
index 5ba258c2d6..8b331af614 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/flatMapConcat.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/flatMapConcat.md
@@ -7,13 +7,19 @@ Transform each input element into a `Source` whose elements
are then flattened i
## Signature
+@apidoc[Source.flatMapConcat](Source) {
scala="#flatMapConcat[T,M](f:Out=%3Eorg.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[T],M]):FlowOps.this.Repr[T]"
java="#flatMapConcat(org.apache.pekko.japi.function.Function)" }
+@apidoc[Source.flatMapConcat](Source) {
scala="#flatMapConcat[T,M](parallelism:Int,f:Out=%3Eorg.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[T],M]):FlowOps.this.Repr[T]"
java="#flatMapConcat(int,org.apache.pekko.japi.function.Function)" }
@apidoc[Flow.flatMapConcat](Flow) {
scala="#flatMapConcat[T,M](f:Out=%3Eorg.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[T],M]):FlowOps.this.Repr[T]"
java="#flatMapConcat(org.apache.pekko.japi.function.Function)" }
+@apidoc[Flow.flatMapConcat](Flow) {
scala="#flatMapConcat[T,M](parallelism:Int,f:Out=%3Eorg.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[T],M]):FlowOps.this.Repr[T]"
java="#flatMapConcat(int,org.apache.pekko.japi.function.Function)" }
## Description
Transform each input element into a `Source` whose elements are then flattened
into the output stream through
concatenation. This means each source is fully consumed before consumption of
the next source starts.
+Since 1.2.0, an overload accepting a `parallelism` parameter can be used to
configure the maximum number of in-flight
+sources that are materialized at the same time. The elements are still emitted
in order, one source after the other.
+
See also: @ref:[flatMapMerge](flatMapMerge.md), @ref:[mapConcat](mapConcat.md)
## Example
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/foldWhile.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/foldWhile.md
index 1de0907726..3409d5f22d 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/foldWhile.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/foldWhile.md
@@ -12,7 +12,7 @@ Start with current value `zero` and then apply the current
and next value to the
## Description
Start with current value `zero` and then apply the current and next value to
the given function. When upstream
-completes, the current value is emitted downstream.
+completes or the predicate `p` returns `false`, the current value is emitted
downstream and the stream completes.
@@@ warning
@@ -38,11 +38,11 @@ Java
@@@div { .callout }
-**emits** when upstream completes
+**emits** when upstream completes or the predicate `p` returns `false`
**backpressures** when downstream backpressures
-**completes** when upstream completes
+**completes** when upstream completes or the predicate `p` returns `false`
@@@
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/groupedWeightedWithin.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/groupedWeightedWithin.md
index fa3337dd43..a66fb47402 100644
---
a/docs/src/main/paradox/stream/operators/Source-or-Flow/groupedWeightedWithin.md
+++
b/docs/src/main/paradox/stream/operators/Source-or-Flow/groupedWeightedWithin.md
@@ -8,6 +8,8 @@ Chunk up this stream into groups of elements received within a
time window, or l
@apidoc[Source.groupedWeightedWithin](Source) {
scala="#groupedWeightedWithin(maxWeight:Long,d:scala.concurrent.duration.FiniteDuration)(costFn:Out=>Long):FlowOps.this.Repr[Seq[Out]]"
java="#groupedWeightedWithin(long,org.apache.pekko.japi.function.Function,java.time.Duration)"
}
@apidoc[Flow.groupedWeightedWithin](Flow) {
scala="#groupedWeightedWithin(maxWeight:Long,d:scala.concurrent.duration.FiniteDuration)(costFn:Out=>Long):FlowOps.this.Repr[Seq[Out]]"
java="#groupedWeightedWithin(long,org.apache.pekko.japi.function.Function,java.time.Duration)"
}
+@apidoc[Source.groupedWeightedWithin](Source) {
scala="#groupedWeightedWithin(maxWeight:Long,maxNumber:Int,d:scala.concurrent.duration.FiniteDuration)(costFn:Out=>Long):FlowOps.this.Repr[Seq[Out]]"
java="#groupedWeightedWithin(long,int,org.apache.pekko.japi.function.Function,java.time.Duration)"
}
+@apidoc[Flow.groupedWeightedWithin](Flow) {
scala="#groupedWeightedWithin(maxWeight:Long,maxNumber:Int,d:scala.concurrent.duration.FiniteDuration)(costFn:Out=>Long):FlowOps.this.Repr[Seq[Out]]"
java="#groupedWeightedWithin(long,int,org.apache.pekko.japi.function.Function,java.time.Duration)"
}
## Description
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/interleave.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/interleave.md
index 61e6e283c9..6fd169b55c 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/interleave.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/interleave.md
@@ -28,7 +28,7 @@ Java
**emits** when element is available from the currently consumed upstream
-**backpressures** when upstream backpressures
+**backpressures** when downstream backpressures
**completes** when both upstreams have completed
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/interleaveAll.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/interleaveAll.md
index ced164f3fd..29f56ba7e3 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/interleaveAll.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/interleaveAll.md
@@ -29,7 +29,7 @@ Java
**emits** when element is available from the currently consumed upstream
-**backpressures** when upstream backpressures
+**backpressures** when downstream backpressures
**completes** when all upstreams have completed if `eagerClose` is false, or
any upstream completes if `eagerClose` is true.
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/limitWeighted.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/limitWeighted.md
index 3712ba6c54..c1b775ad8a 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/limitWeighted.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/limitWeighted.md
@@ -30,10 +30,10 @@ Java
@@@div { .callout }
-**emits** when upstream emits and the number of emitted elements has not
reached max
+**emits** when upstream emits and the accumulated cost has not reached max
**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
@@@
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/mergeLatest.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/mergeLatest.md
index 9f7197cbe0..cbff4f091e 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/mergeLatest.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/mergeLatest.md
@@ -19,8 +19,8 @@ completes otherwise when all upstreams complete.
## Example
-This example takes a stream of prices and quantities and emits the price each
time the
-price of quantity changes:
+This example takes a stream of prices and quantities and emits the price
multiplied by the quantity each time
+the price or the quantity changes:
Scala
: @@snip
[MergeLatest.scala](/docs/src/test/scala/docs/stream/operators/sourceorflow/MergeLatest.scala)
{ #mergeLatest }
@@ -35,6 +35,6 @@ Java
**emits** when element is available from some input and each input emits at
least one element from stream start
-**completes** all upstreams complete (eagerClose=false) or one upstream
completes (eagerClose=true)
+**completes** when all upstreams complete (`eagerComplete=false`) or one
upstream completes (`eagerComplete=true`)
@@@
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/onErrorContinue.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/onErrorContinue.md
index 8fb257e534..35c744bb9c 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/onErrorContinue.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/onErrorContinue.md
@@ -9,10 +9,24 @@ Continues the stream when an upstream error occurs.
@apidoc[Source.onErrorContinue](Source) {
scala="#onErrorContinue[T%3C:Throwable](errorConsumer:Throwable=%3EUnit)(implicittag:scala.reflect.ClassTag[T]):FlowOps.this.Repr[Out]"
java="#onErrorContinue(org.apache.pekko.japi.function.Procedure)" }
@apidoc[Flow.onErrorContinue](Flow) {
scala="#onErrorContinue[T%3C:Throwable](errorConsumer:Throwable=%3EUnit)(implicittag:scala.reflect.ClassTag[T]):FlowOps.this.Repr[Out]"
java="#onErrorContinue(java.lang.Class,org.apache.pekko.japi.function.Procedure)"
}
+@apidoc[Source.onErrorContinue](Source) {
scala="#onErrorContinue(p:Throwable=%3EBoolean)(errorConsumer:Throwable=%3EUnit):FlowOps.this.Repr[Out]"
java="#onErrorContinue(org.apache.pekko.japi.function.Predicate,org.apache.pekko.japi.function.Procedure)"
}
+@apidoc[Source.onErrorContinue](Source) {
java="#onErrorContinue(java.lang.Class,org.apache.pekko.japi.function.Procedure)"
}
+@apidoc[Flow.onErrorContinue](Flow) {
scala="#onErrorContinue(p:Throwable=%3EBoolean)(errorConsumer:Throwable=%3EUnit):FlowOps.this.Repr[Out]"
java="#onErrorContinue(org.apache.pekko.japi.function.Predicate,org.apache.pekko.japi.function.Procedure)"
}
+@apidoc[Flow.onErrorContinue](Flow) {
java="#onErrorContinue(org.apache.pekko.japi.function.Procedure)" }
+
## Description
Continues the stream when an upstream error occurs.
+When an error is signaled from upstream, the `errorConsumer` function is
invoked with the `Throwable`, and the stream
+resumes processing subsequent elements. The element that caused the error is
dropped.
+
+Which errors are handled can be restricted @scala[by the type parameter
`T`]@java[by passing a `Class`], or by passing
+a predicate `p` that decides for each error whether it should be handled.
Errors that are not handled fail the
+stream as usual.
+
+This operator relies on supervision, so it has no effect on operators that do
not support supervision.
+
## Reactive Streams semantics
@@@div { .callout }
@@ -21,7 +35,9 @@ Continues the stream when an upstream error occurs.
**backpressures** downstream backpressures
-**completes** upstream completes or upstream failed with exception this
operator can't handle
+**completes** upstream completes
+
+**fails** upstream failed with exception this operator can't handle
**Cancels when** downstream cancels
@@@
\ No newline at end of file
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md
index c3aeed8fd8..bd485b4a70 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/sliding.md
@@ -6,6 +6,7 @@ Provide a sliding window over the incoming stream and pass the
windows as groups
## Signature
+@apidoc[Source.sliding](Source) {
scala="#sliding(n:Int,step:Int):FlowOps.this.Repr[Seq[Out]]"
java="#sliding(int,int)" }
@apidoc[Flow.sliding](Flow) {
scala="#sliding(n:Int,step:Int):FlowOps.this.Repr[Seq[Out]]"
java="#sliding(int,int)" }
## Description
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/takeWhile.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/takeWhile.md
index 8765fed3c5..e7d1460bc2 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/takeWhile.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/takeWhile.md
@@ -6,14 +6,17 @@ Pass elements downstream as long as a predicate function
returns true and then c
## Signature
-@apidoc[Source.takeWhile](Source) {
scala="#takeWhile(p:Out=>Boolean,inclusive:Boolean):FlowOps.this.Repr[Out]"
java="#takeWhile(org.apache.pekko.japi.function.Predicate)" }
-@apidoc[Flow.takeWhile](Flow) {
scala="#takeWhile(p:Out=>Boolean,inclusive:Boolean):FlowOps.this.Repr[Out]"
java="#takeWhile(org.apache.pekko.japi.function.Predicate)" }
+@apidoc[Source.takeWhile](Source) {
scala="#takeWhile(p:Out=>Boolean):FlowOps.this.Repr[Out]"
java="#takeWhile(org.apache.pekko.japi.function.Predicate)" }
+@apidoc[Source.takeWhile](Source) {
scala="#takeWhile(p:Out=>Boolean,inclusive:Boolean):FlowOps.this.Repr[Out]"
java="#takeWhile(org.apache.pekko.japi.function.Predicate,boolean)" }
+@apidoc[Flow.takeWhile](Flow) {
scala="#takeWhile(p:Out=>Boolean):FlowOps.this.Repr[Out]"
java="#takeWhile(org.apache.pekko.japi.function.Predicate)" }
+@apidoc[Flow.takeWhile](Flow) {
scala="#takeWhile(p:Out=>Boolean,inclusive:Boolean):FlowOps.this.Repr[Out]"
java="#takeWhile(org.apache.pekko.japi.function.Predicate,boolean)" }
## Description
Pass elements downstream as long as a predicate function returns true and then
complete.
-The element for which the predicate returns false is not emitted.
+The element for which the predicate returns false is not emitted, unless the
overload with `inclusive` set to `true`
+is used, in which case that first failing element is also emitted before
completing.
## Example
diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/wireTap.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/wireTap.md
index 3533119c8a..9e611fb0d1 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/wireTap.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/wireTap.md
@@ -8,6 +8,8 @@ Attaches the given `Sink` to this `Flow` as a wire tap, meaning
that elements th
@apidoc[Source.wireTap](Source) {
scala="#wireTap(f:Out=>Unit):FlowOps.this.Repr[Out]"
java="#wireTap(org.apache.pekko.japi.function.Procedure)" }
@apidoc[Flow.wireTap](Flow) {
scala="#wireTap(f:Out=>Unit):FlowOps.this.Repr[Out]"
java="#wireTap(org.apache.pekko.japi.function.Procedure)" }
+@apidoc[Source.wireTap](Source) {
scala="#wireTap(that:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SinkShape[Out],_]):FlowOps.this.Repr[Out]"
java="#wireTap(org.apache.pekko.stream.Graph)" }
+@apidoc[Flow.wireTap](Flow) {
scala="#wireTap(that:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SinkShape[Out],_]):FlowOps.this.Repr[Out]"
java="#wireTap(org.apache.pekko.stream.Graph)" }
## Description
diff --git
a/docs/src/main/paradox/stream/operators/Source-or-Flow/zipLatestWith.md
b/docs/src/main/paradox/stream/operators/Source-or-Flow/zipLatestWith.md
index 975ee80de0..3b0de281f7 100644
--- a/docs/src/main/paradox/stream/operators/Source-or-Flow/zipLatestWith.md
+++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/zipLatestWith.md
@@ -8,14 +8,20 @@ Combines elements from multiple sources through a `combine`
function and passes
@apidoc[Source.zipLatestWith](Source) {
scala="#zipLatestWith[Out2,Out3](that:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[Out2],_])(combine:(Out,Out2)=>Out3):FlowOps.this.Repr[Out3]"
java="#zipLatestWith(org.apache.pekko.stream.Graph,org.apache.pekko.japi.function.Function2)"
}
@apidoc[Flow.zipLatestWith](Flow) {
scala="#zipLatestWith[Out2,Out3](that:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[Out2],_])(combine:(Out,Out2)=>Out3):FlowOps.this.Repr[Out3]"
java="#zipLatestWith(org.apache.pekko.stream.Graph,org.apache.pekko.japi.function.Function2)"
}
+@apidoc[Source.zipLatestWith](Source) {
scala="#zipLatestWith[Out2,Out3](that:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[Out2],_],eagerComplete:Boolean)(combine:(Out,Out2)=>Out3):FlowOps.this.Repr[Out3]"
java="#zipLatestWith(org.apache.pekko.stream.Graph,boolean,org.apache.pekko.japi.function.Function2)"
}
+@apidoc[Flow.zipLatestWith](Flow) {
scala="#zipLatestWith[Out2,Out3](that:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[Out2],_],eagerComplete:Boolean)(combine:(Out,Out2)=>Out3):FlowOps.this.Repr[Out3]"
java="#zipLatestWith(org.apache.pekko.stream.Graph,boolean,org.apache.pekko.japi.function.Function2)"
}
## Description
-Combines elements from each of multiple sources into @scala[tuples]
@java[*Pair*] and passes the @scala[tuples] @java[pairs] downstream, picking
always the latest element of each.
+Combines elements from each of multiple sources through a `combine` function
and passes the returned value downstream, picking always the latest element of
each.
No element is emitted until at least one element from each Source becomes
available. Whenever a new
-element appears, the zipping function is invoked with a tuple containing the
new element and the last seen element of the other stream.
+element appears, the `combine` function is invoked with the new element and
the last seen element of the other stream.
+
+By default the stream completes as soon as any upstream completes. Use the
overload with `eagerComplete` set to `false`
+to keep running until all upstreams have completed (the stream still completes
immediately if an upstream
+completes before it has emitted any element).
## Reactive Streams semantics
@@ -26,7 +32,7 @@ element appears, the zipping function is invoked with a tuple
containing the new
**backpressures** when downstream backpressures
-**completes** when any upstream completes
+**completes** when any upstream completes, or when all upstreams complete if
`eagerComplete` is `false`
**cancels** when downstream cancels
diff --git a/docs/src/main/paradox/stream/operators/Source/asSubscriber.md
b/docs/src/main/paradox/stream/operators/Source/asSubscriber.md
index 85b499b375..ca6aa01b45 100644
--- a/docs/src/main/paradox/stream/operators/Source/asSubscriber.md
+++ b/docs/src/main/paradox/stream/operators/Source/asSubscriber.md
@@ -6,18 +6,14 @@ Integration with Reactive Streams, materializes into a
`org.reactivestreams.Subs
## Signature
-Scala
-:
@@snip[JavaFlowSupport.scala](/stream/src/main/scala/org/apache/pekko/stream/scaladsl/JavaFlowSupport.scala)
{ #asSubscriber }
-
-Java
-:
@@snip[AsSubscriber.java](/docs/src/test/java/jdocs/stream/operators/source/AsSubscriber.java)
{ #api }
+@apidoc[Source.asSubscriber](Source$) {
scala="#asSubscriber[T]:org.apache.pekko.stream.scaladsl.Source[T,org.reactivestreams.Subscriber[T]]"
java="#asSubscriber()" }
## Description
If you want to create a @apidoc[Source] that gets its elements from another
library that supports
[Reactive Streams](https://www.reactive-streams.org/), you can use
`Source.asSubscriber`.
Each time this @apidoc[Source] is materialized, it produces a materialized
value of type
-@javadoc[org.reactivestreams.Subscriber](java.util.concurrent.Flow.Subscriber).
+`org.reactivestreams.Subscriber`.
This `org.reactivestreams.Subscriber` can be attached to a
[Reactive Streams](https://www.reactive-streams.org/)
`org.reactivestreams.Publisher`
to populate it.
@@ -30,6 +26,8 @@ through
@scala[`org.apache.pekko.stream.scaladsl.JavaFlowSupport.Source#asSubscr
## Example
+The examples below use the `JavaFlowSupport` variant, which works with the
`java.util.concurrent.Flow` interfaces.
+
Suppose we use a database client that supports [Reactive
Streams](https://www.reactive-streams.org/),
we could create a @apidoc[Source] that queries the database for its rows. That
@apidoc[Source] can then
be used for further processing, for example creating a @apidoc[Source] that
contains the names of the
diff --git a/docs/src/main/paradox/stream/operators/Source/fromPublisher.md
b/docs/src/main/paradox/stream/operators/Source/fromPublisher.md
index 8f6ccb58cd..2a82294011 100644
--- a/docs/src/main/paradox/stream/operators/Source/fromPublisher.md
+++ b/docs/src/main/paradox/stream/operators/Source/fromPublisher.md
@@ -6,11 +6,7 @@ Integration with Reactive Streams, subscribes to a
`org.reactivestreams.Publishe
## Signature
-Scala
-:
@@snip[JavaFlowSupport.scala](/stream/src/main/scala/org/apache/pekko/stream/scaladsl/JavaFlowSupport.scala)
{ #fromPublisher }
-
-Java
-:
@@snip[FromPublisher.java](/docs/src/test/java/jdocs/stream/operators/source/FromPublisher.java)
{ #api }
+@apidoc[Source.fromPublisher](Source$) {
scala="#fromPublisher[T](publisher:org.reactivestreams.Publisher[T]):org.apache.pekko.stream.scaladsl.Source[T,org.apache.pekko.NotUsed]"
java="#fromPublisher(org.reactivestreams.Publisher)" }
## Description
@@ -28,6 +24,8 @@ through
@scala[`org.apache.pekko.stream.scaladsl.JavaFlowSupport.Source#fromPubl
## Example
+The examples below use the `JavaFlowSupport` variant, which works with the
`java.util.concurrent.Flow` interfaces.
+
Suppose we use a database client that supports [Reactive
Streams](https://www.reactive-streams.org/),
we could create a @apidoc[Source] that queries the database for its rows. That
@apidoc[Source] can then
be used for further processing, for example creating a @apidoc[Source] that
contains the names of the
diff --git a/docs/src/main/paradox/stream/operators/Source/items.md
b/docs/src/main/paradox/stream/operators/Source/items.md
index 31cf9a0415..2e80534e0b 100644
--- a/docs/src/main/paradox/stream/operators/Source/items.md
+++ b/docs/src/main/paradox/stream/operators/Source/items.md
@@ -13,8 +13,6 @@ Create a `Source` from the given items.
Create a `Source` from the given items.
-## Examples
-
## Reactive Streams semantics
@@@div { .callout }
diff --git a/docs/src/main/paradox/stream/operators/Source/iterate.md
b/docs/src/main/paradox/stream/operators/Source/iterate.md
index 1daa161635..6191405c6f 100644
--- a/docs/src/main/paradox/stream/operators/Source/iterate.md
+++ b/docs/src/main/paradox/stream/operators/Source/iterate.md
@@ -6,17 +6,17 @@ Creates a sequential `Source` by iterating with the given
predicate, function an
## Signature
-@apidoc[Source.iterate](Source$) {
scala="#iterate%5BT](seed:T)(hasNext:T=>Boolean,next:T=>T):org.apache.pekko.stream.scaladsl.Source%5BT,org.apache.pekko.NotUsed]"
java="#iterate(java.lang.Object,org.apache.pekko.japi.function.Predicate,org.apache.pekko.japi.function.Function)"
}
+@apidoc[Source.iterate](Source$) {
scala="#iterate%5BT](seed:T)(p:T=>Boolean,f:T=>T):org.apache.pekko.stream.scaladsl.Source%5BT,org.apache.pekko.NotUsed]"
java="#iterate(java.lang.Object,org.apache.pekko.japi.function.Predicate,org.apache.pekko.japi.function.Function)"
}
## Description
-Creates a sequential Source by iterating with the given hasNext predicate and
next function,
-starting with the given seed value. If the hasNext function returns false for
the seed, the Source completes with empty.
+Creates a sequential Source by iterating with the given predicate `p` and
function `f`,
+starting with the given seed value. If the predicate `p` returns false for the
seed, the Source completes with empty.
@@@ warning
-The same `seed` value will be used for every materialization of the `Source`
so it is **mandatory** that the state is immutable. For example a
`java.util.Iterator`, `Array` or Java standard library collection would not be
safe as the fold operation could mutate the value. If you must use a mutable
value, combining with @ref:[Source.lazySource](lazySource.md) to make sure a
new mutable `zero` value is created for each materialization is one solution.
+The same `seed` value will be used for every materialization of the `Source`
so it is **mandatory** that the state is immutable. For example a
`java.util.Iterator`, `Array` or Java standard library collection would not be
safe as the `f` function could mutate the value. If you must use a mutable
value, combining with @ref:[Source.lazySource](lazySource.md) to make sure a
new mutable `seed` value is created for each materialization is one solution.
@@@
@@ -35,9 +35,9 @@ Java
@@@div { .callout }
-**emits** when there is demand and the `next` function returns.
+**emits** when there is demand and the `f` function returns.
-**completes** when the `hasNext` predicate returns false.
+**completes** when the `p` predicate returns false.
@@@
diff --git a/docs/src/main/paradox/stream/operators/Source/lazySource.md
b/docs/src/main/paradox/stream/operators/Source/lazySource.md
index 92b93a7a64..aa8a3cd147 100644
--- a/docs/src/main/paradox/stream/operators/Source/lazySource.md
+++ b/docs/src/main/paradox/stream/operators/Source/lazySource.md
@@ -45,7 +45,7 @@ In this example we make use of that by unfolding a mutable
object that works lik
there are more elements and one that produces the next and moves to the next
element.
If the `IteratorLikeThing` was used directly in a `Source.unfold` the same
instance would end up being unsafely shared
-across all three materializations of the stream, but wrapping it with
`Source.lazy` ensures we create a separate instance
+across all three materializations of the stream, but wrapping it with
`Source.lazySource` ensures we create a separate instance
for each of the started streams:
Scala
diff --git a/docs/src/main/paradox/stream/operators/Source/maybe.md
b/docs/src/main/paradox/stream/operators/Source/maybe.md
index 00f819c1e5..ccefc3fab9 100644
--- a/docs/src/main/paradox/stream/operators/Source/maybe.md
+++ b/docs/src/main/paradox/stream/operators/Source/maybe.md
@@ -23,10 +23,10 @@ after creation.
* If the downstream of this source cancels or fails before the promise has
been completed, then the promise will be completed
with @scala[`None`]@java[empty `Optional`].
-`Source.maybe` has some similarities with
@scala[@ref:[`Source.future`](future.md)]@java[@ref:[`Source.completionStageSource`](completionStageSource.md)].
+`Source.maybe` has some similarities with
@scala[@ref:[`Source.future`](future.md)]@java[@ref:[`Source.completionStage`](completionStage.md)].
One difference is that a new @scala[`Promise`]@java[`CompletableFuture`] is
materialized from `Source.maybe` each time
the stream is run while the @scala[`Future`]@java[`CompletionStage`] given to
-@scala[`Source.future`]@java[`Source.completionStageSource`] can only be
completed once.
+@scala[`Source.future`]@java[`Source.completionStage`] can only be completed
once.
@ref:[`Source.queue`](queue.md) is an alternative for emitting more than one
element.
diff --git a/docs/src/main/paradox/stream/operators/Source/queue.md
b/docs/src/main/paradox/stream/operators/Source/queue.md
index 4fc5235162..02d706ba80 100644
--- a/docs/src/main/paradox/stream/operators/Source/queue.md
+++ b/docs/src/main/paradox/stream/operators/Source/queue.md
@@ -79,7 +79,7 @@ Java
**emits** when there is demand and the queue contains elements
-**completes** when downstream completes
+**completes** when the queue is completed (via `complete()` on the
materialized queue) and all buffered elements have been emitted
@@@
diff --git
a/docs/src/main/paradox/stream/operators/StreamConverters/asJavaStream.md
b/docs/src/main/paradox/stream/operators/StreamConverters/asJavaStream.md
index 52fa8182c3..46d764f560 100644
--- a/docs/src/main/paradox/stream/operators/StreamConverters/asJavaStream.md
+++ b/docs/src/main/paradox/stream/operators/StreamConverters/asJavaStream.md
@@ -16,7 +16,7 @@ Elements emitted through the stream will be available for
reading through the Ja
The Java 8 `Stream` will be ended when the stream flowing into this `Sink`
completes, and closing the Java
`Stream` will cancel the inflow of this `Sink`. If the Java `Stream` throws an
exception, the Pekko stream is cancelled.
-Be aware that Java `Stream` blocks current thread while waiting on next
element from downstream.
+Be aware that Java `Stream` blocks current thread while waiting on next
element from upstream.
## Example
diff --git a/docs/src/main/paradox/stream/operators/Unzip.md
b/docs/src/main/paradox/stream/operators/Unzip.md
index 77999834ad..1f071731d7 100644
--- a/docs/src/main/paradox/stream/operators/Unzip.md
+++ b/docs/src/main/paradox/stream/operators/Unzip.md
@@ -6,6 +6,8 @@ Takes a stream of two element tuples and unzips the two
elements into two differ
## Signature
+@apidoc[stream.*.Unzip$]
+
## Description
Takes a stream of two element tuples and unzips the two elements into two
different downstreams.
diff --git a/docs/src/main/paradox/stream/operators/UnzipWith.md
b/docs/src/main/paradox/stream/operators/UnzipWith.md
index c1e0833578..5d1ac8f70f 100644
--- a/docs/src/main/paradox/stream/operators/UnzipWith.md
+++ b/docs/src/main/paradox/stream/operators/UnzipWith.md
@@ -6,6 +6,8 @@ Splits each element of input into multiple downstreams using a
function
## Signature
+@apidoc[stream.*.UnzipWith$]
+
## Description
Splits each element of input into multiple downstreams using a function
diff --git a/docs/src/main/paradox/stream/stream-customize.md
b/docs/src/main/paradox/stream/stream-customize.md
index 34d0aa5587..cdc9bc04e4 100644
--- a/docs/src/main/paradox/stream/stream-customize.md
+++ b/docs/src/main/paradox/stream/stream-customize.md
@@ -133,7 +133,8 @@ in that state.
The following operations are available for *input* ports:
- * `pull(in)` requests a new element from an input port. This is only possible
after the port has been pushed by upstream.
+ * `pull(in)` requests a new element from an input port. This is only possible
if the port is not closed and has not already been pulled
+(i.e. after a previous element has been pushed by upstream, or initially
before any element has been requested).
* `grab(in)` acquires the element that has been received during an
`onPush()`. It cannot be called again until the
port is pushed again by the upstream.
* `cancel(in)` closes the input port.
diff --git a/docs/src/main/paradox/stream/stream-dynamic.md
b/docs/src/main/paradox/stream/stream-dynamic.md
index 7470af3b80..a57235fd90 100644
--- a/docs/src/main/paradox/stream/stream-dynamic.md
+++ b/docs/src/main/paradox/stream/stream-dynamic.md
@@ -13,8 +13,6 @@ To use Pekko Streams, add the module to your project:
version=PekkoVersion
}
-## Introduction
-
<a id="kill-switch"></a>
## Controlling stream completion with KillSwitch
@@ -165,7 +163,7 @@ Java
We now wrap the @apidoc[stream.*.Sink] and @apidoc[stream.*.Source] in a
@apidoc[stream.*.Flow] using @apidoc[Flow.fromSinkAndSource](stream.*.Flow$)
{scala="#fromSinkAndSource[I,O](sink:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SinkShape[I],_],source:org.apache.pekko.stream.Graph[org.apache.pekko.stream.SourceShape[O],_]):org.apache.pekko.stream.scaladsl.Flow[I,O,org.apache.pekko.NotUsed]"
java="#fromSinkAndSource(org.apache.pekko.stream.Graph,org.apache.pekko.stream.Graph)"
[...]
up the two sides of the channel into one and forces users of it to always
define a publisher and subscriber side
(even if the subscriber side is dropping). It also allows us to attach a
@apidoc[stream.KillSwitch] as
-a `BidiStage` which in turn makes it possible to close both the original
`Sink` and `Source` at the
+a bidirectional operator (with `BidiShape`, created by
`KillSwitches.singleBidi`) which in turn makes it possible to close both the
original `Sink` and `Source` at the
same time.
Finally, we add `backpressureTimeout` on the consumer side to ensure that
subscribers that block the channel for more
than 3 seconds are forcefully removed (and their stream failed).
diff --git a/docs/src/main/paradox/stream/stream-error.md
b/docs/src/main/paradox/stream/stream-error.md
index dd1f13f2a7..92f3070ba6 100644
--- a/docs/src/main/paradox/stream/stream-error.md
+++ b/docs/src/main/paradox/stream/stream-error.md
@@ -271,7 +271,7 @@ Scala
Java
: @@snip
[IntegrationDocTest.java](/docs/src/test/java/jdocs/stream/IntegrationDocTest.java)
{ #email-address-lookup2 }
-The @scala[@scaladoc[Future](scala.concurrent.Future)]
@java[@javadoc[CompletionStage](java.util.concurrent.CompletionStage)] is
completed @scala[with `Failure`] @java[normally] if the email is not found.
+The @scala[@scaladoc[Future](scala.concurrent.Future)]
@java[@javadoc[CompletionStage](java.util.concurrent.CompletionStage)] is
completed @scala[with `Failure`] @java[exceptionally] if the email is not found.
Transforming the stream of authors to a stream of email addresses by using the
`lookupEmail`
service can be done with @apidoc[mapAsync](stream.*.Source)
{scala="#mapAsync[T](parallelism:Int)(f:Out=%3Escala.concurrent.Future[T]):FlowOps.this.Repr[T]"
java="#mapAsync(int,org.apache.pekko.japi.function.Function)"} and we use
@scala[@scaladoc[Supervision.resumingDecider](pekko.stream.Supervision$#resumingDecider:org.apache.pekko.stream.Supervision.Deciderwithorg.apache.pekko.japi.function.Function[Throwable,org.apache.pekko.stream.Supervision.Directive])]
@java[@javadoc[Supervision. [...]
diff --git a/docs/src/main/paradox/stream/stream-flows-and-basics.md
b/docs/src/main/paradox/stream/stream-flows-and-basics.md
index 940b85f5b7..2bb05d8994 100644
--- a/docs/src/main/paradox/stream/stream-flows-and-basics.md
+++ b/docs/src/main/paradox/stream/stream-flows-and-basics.md
@@ -13,8 +13,6 @@ To use Pekko Streams, add the module to your project:
version=PekkoVersion
}
-## Introduction
-
## Core concepts
Pekko Streams is a library to process and transfer a sequence of elements
using bounded buffer space. This
@@ -196,7 +194,7 @@ of absence of a value we recommend using
@scala[@scaladoc[scala.Option](scala.Op
## Back-pressure explained
Pekko Streams implement an asynchronous non-blocking back-pressure protocol
standardised by the [Reactive Streams](https://www.reactive-streams.org/)
-specification, which Pekko is a founding member of.
+specification.
The user of the library does not have to write any explicit back-pressure
handling code — it is built in
and dealt with automatically by all of the provided Pekko Streams operators.
It is possible however to add
diff --git a/docs/src/main/paradox/stream/stream-graphs.md
b/docs/src/main/paradox/stream/stream-graphs.md
index 04016c3fe0..eebd6266e5 100644
--- a/docs/src/main/paradox/stream/stream-graphs.md
+++ b/docs/src/main/paradox/stream/stream-graphs.md
@@ -41,7 +41,7 @@ Pekko Streams currently provide these junctions (for a
detailed list see the @re
* @scala[`Balance[T]`]@java[`Balance<T>`] – *(1 input, N outputs)* given
an input element emits to one of its output ports
* @scala[`Partition[T]`]@java[`Partition<T>`] – *(1 input, N outputs)*
given an input element emits to specified output based on a partition function
* @scala[`UnzipWith[In,A,B,...]`]@java[`UnzipWith<In,A,B,...>`] – *(1
input, N outputs)* takes a function of 1 input that given a value for each
input emits N output elements (where N <= 20)
- * @scala[`UnZip[A,B]`]@java[`UnZip<A,B>`] – *(1 input, 2 outputs)* splits
a stream of @scala[`(A,B)`]@java[`Pair<A,B>`] tuples into two streams, one of
type `A` and one of type `B`
+ * @scala[`Unzip[A,B]`]@java[`Unzip<A,B>`] – *(1 input, 2 outputs)* splits
a stream of @scala[`(A,B)`]@java[`Pair<A,B>`] tuples into two streams, one of
type `A` and one of type `B`
* **Fan-in**
@@ -49,7 +49,7 @@ Pekko Streams currently provide these junctions (for a
detailed list see the @re
* @scala[`MergePreferred[In]`]@java[`MergePreferred<In>`] – like `Merge`
but if elements are available on `preferred` port, it picks from it, otherwise
randomly from `others`
* @scala[`MergePrioritized[In]`]@java[`MergePrioritized<In>`] – like
`Merge` but if elements are available on all input ports, it picks from them
randomly based on their `priority`
* @scala[`MergeLatest[In]`]@java[`MergeLatest<In>`] – *(N inputs, 1
output)* emits `List[In]`, when i-th input stream emits element, then i-th
element in emitted list is updated
- * @scala[`MergeSequence[In]`]@java[`MergeSequence<In>`] – *(N inputs, 1
output)* emits `List[In]`, where the input streams must represent a partitioned
sequence that must be merged back together in order
+ * @scala[`MergeSequence[In]`]@java[`MergeSequence<In>`] – *(N inputs, 1
output)* emits elements of type `In` in sequence order, where the input streams
must represent a partitioned sequence that must be merged back together in order
* @scala[`ZipWith[A,B,...,Out]`]@java[`ZipWith<A,B,...,Out>`] – *(N
inputs, 1 output)* which takes a function of N inputs that given a value for
each input emits 1 output element
* @scala[`Zip[A,B]`]@java[`Zip<A,B>`] – *(2 inputs, 1 output)* is a
`ZipWith` specialised to zipping input streams of `A` and `B` into a
@scala[`(A,B)`]@java[`Pair(A,B)`] tuple stream
* @scala[`Concat[A]`]@java[`Concat<A>`] – *(2 inputs, 1 output)*
concatenates two streams (first consume one, then the second one)
diff --git a/docs/src/main/paradox/stream/stream-introduction.md
b/docs/src/main/paradox/stream/stream-introduction.md
index 31025b8239..b1e6500300 100644
--- a/docs/src/main/paradox/stream/stream-introduction.md
+++ b/docs/src/main/paradox/stream/stream-introduction.md
@@ -28,8 +28,8 @@ efficiently and with bounded resource usage—no more
OutOfMemoryErrors. In orde
to achieve this our streams need to be able to limit the buffering that they
employ, they need to be able to slow down producers if the consumers cannot
keep up. This feature is called back-pressure and is at the core of the
-[Reactive Streams](https://www.reactive-streams.org/) initiative of which
Pekko is a
-founding member. For you this means that the hard problem of propagating and
+[Reactive Streams](https://www.reactive-streams.org/) initiative, which Pekko
Streams
+implements. For you this means that the hard problem of propagating and
reacting to back-pressure has been incorporated in the design of Pekko Streams
already, so you have one less thing to worry about; it also means that Pekko
Streams interoperate seamlessly with all other Reactive Streams implementations
diff --git a/docs/src/main/paradox/stream/stream-io.md
b/docs/src/main/paradox/stream/stream-io.md
index 9343e398a4..d5ed9cf4c7 100644
--- a/docs/src/main/paradox/stream/stream-io.md
+++ b/docs/src/main/paradox/stream/stream-io.md
@@ -38,8 +38,10 @@ Java
Next, we handle *each* incoming connection using a @apidoc[stream.*.Flow]
which will be used as the operator
to handle and emit @apidoc[util.ByteString] s from and to the TCP Socket.
Since one `ByteString` does not have to necessarily
correspond to exactly one line of text (the client might be sending the line
in chunks) we use the @apidoc[Framing.delimiter](stream.*.Framing$)
{scala="#delimiter(delimiter:org.apache.pekko.util.ByteString,maximumFrameLength:Int,allowTruncation:Boolean):org.apache.pekko.stream.scaladsl.Flow[org.apache.pekko.util.ByteString,org.apache.pekko.util.ByteString,org.apache.pekko.NotUsed]"
java="#delimiter(org.apache.pekko.util.ByteString,int,org.apache.pekko.stream.javadsl.FramingTruncation)"}
-helper Flow to chunk the inputs up into actual lines of text. The last boolean
-argument indicates that we require an explicit line ending even for the last
message before the connection is closed.
+helper Flow to chunk the inputs up into actual lines of text. The last
argument controls whether an explicit line ending
+is required for the last message before the connection is closed:
@scala[`allowTruncation = true` means that the last message
+is emitted even without a trailing line
ending]@java[`FramingTruncation.DISALLOW` means that an explicit line ending is
required
+even for the last message].
In this example we add exclamation marks to each incoming text message and
push it through the flow:
Scala
diff --git a/docs/src/main/paradox/stream/stream-quickstart.md
b/docs/src/main/paradox/stream/stream-quickstart.md
index 6e78229ce3..4dd4d53c09 100644
--- a/docs/src/main/paradox/stream/stream-quickstart.md
+++ b/docs/src/main/paradox/stream/stream-quickstart.md
@@ -251,7 +251,7 @@ more advanced operators to finally be consumed by a
@scala[`Sink[In,M3]`]@java[`
and `M3` for now, they are not relevant to the types of the elements
produced/consumed by these classes – they are
"materialized types", which we'll talk about
@ref:[below](#materialized-values-quick))]@java[. The first type
parameter—`Tweet` in this case—designates the kind of elements produced
by the source while the `M` type parameters describe the object that is
created during
-materialization ([see below](#materialized-values-quick))—`NotUsed` (from the
`scala.runtime`
+materialization ([see below](#materialized-values-quick))—`NotUsed` (from the
`org.apache.pekko`
package) means that no value is produced, it is the generic equivalent of
`void`.]
diff --git a/docs/src/main/paradox/stream/stream-refs.md
b/docs/src/main/paradox/stream/stream-refs.md
index 071de31f73..349705ecc7 100644
--- a/docs/src/main/paradox/stream/stream-refs.md
+++ b/docs/src/main/paradox/stream/stream-refs.md
@@ -162,12 +162,29 @@ Stream refs utilise normal actor messaging for their
transport, and therefore pr
- messages can be lost, however:
- a *dropped demand signal* will be re-delivered automatically (similar to
system messages)
- a *dropped element signal* will cause the stream to *fail*
-
-## Bulk Stream References
-
-Bulk stream refs can be used to create simple side-channels to transfer
humongous amounts
-of data such as huge log files, messages or even media, with as much ease as
if it was a trivial local stream.
+### Transferring large amounts of data
+
+Stream refs can also be used to create simple side-channels to transfer large
amounts of data, such as big log
+files, other files or media, between nodes, with as much ease as if it was a
local stream. This uses the same
+`SourceRef` and `SinkRef` described above; there is no separate API for it.
For example, a file can be offered to a
+remote node by running `FileIO.fromPath(path)` into `StreamRefs.sourceRef()`
and sending the materialized `SourceRef`
+to a remote actor, which runs its `source` into a `Sink` of its own, such as
`FileIO.toPath`.
+
+Things to keep in mind for large transfers:
+
+- Back-pressure is maintained across the network, so the origin does not send
data faster than the target can process it.
+- Each `SourceRef` and `SinkRef` can be materialized only once.
+- Stream refs are not persistent and, as described in @ref:[Delivery
guarantees](#delivery-guarantees), a lost element
+ or a failure of one of the nodes fails the stream. There is no automatic
resumption, so for very large transfers
+ consider adding your own resumption protocol, for example by tracking an
offset in the actor messages that set up
+ the stream.
+- Each stream element is sent as a remoting message, so a serialized element
must fit within the remoting frame size
+ (`pekko.remote.artery.advanced.maximum-frame-size`, 256 KiB by default).
When streaming files, keep the `ByteString`
+ chunk size well below that.
+- The target side must attach to the stream ref within the
+ @ref:[subscription timeout](#stream-reference-subscription-timeouts).
+
## Serialization of SourceRef and SinkRef
StreamRefs require serialization, since the whole point is to send them
between nodes of a cluster. A built in serializer
diff --git a/project/StreamOperatorsIndexGenerator.scala
b/project/StreamOperatorsIndexGenerator.scala
index daf88cc4e4..8c2ed8d524 100644
--- a/project/StreamOperatorsIndexGenerator.scala
+++ b/project/StreamOperatorsIndexGenerator.scala
@@ -63,8 +63,6 @@ object StreamOperatorsIndexGenerator extends AutoPlugin {
"runForeach",
"runReduce",
"named",
- "throttleEven",
- "actorPublisher",
"addAttributes",
"mapMaterializedValue",
// for comprehensions
@@ -109,7 +107,6 @@ object StreamOperatorsIndexGenerator extends AutoPlugin {
"shape",
"traversalBuilder",
"fromGraph",
- "actorSubscriber",
"foldAsync",
"newOnCompleteStage"))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
