You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by fa...@apache.org on 2025/11/10 18:46:37 UTC
(pekko) branch 1.3.x updated: rework reactivestreams docs (#2468) (#2470)
This is an automated email from the ASF dual-hosted git repository.
fanningpj pushed a commit to branch 1.3.x
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/1.3.x by this push:
new 47d64b4a6c rework reactivestreams docs (#2468) (#2470)
47d64b4a6c is described below
commit 47d64b4a6cea46d7f93dbba9e397a28a891c1473
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Mon Nov 10 19:46:31 2025 +0100
rework reactivestreams docs (#2468) (#2470)
* rework reactivestreams docs
* tidy up
* link issues
add back operators
---
.../paradox/stream/operators/Sink/asPublisher.md | 3 ++-
.../stream/operators/Sink/fromSubscriber.md | 4 ++-
.../stream/operators/Source/asSubscriber.md | 29 ++++++++++------------
.../stream/operators/Source/fromPublisher.md | 25 ++++++++-----------
docs/src/main/paradox/stream/operators/index.md | 4 +--
5 files changed, 31 insertions(+), 34 deletions(-)
diff --git a/docs/src/main/paradox/stream/operators/Sink/asPublisher.md b/docs/src/main/paradox/stream/operators/Sink/asPublisher.md
index 49be6139f9..01759996ce 100644
--- a/docs/src/main/paradox/stream/operators/Sink/asPublisher.md
+++ b/docs/src/main/paradox/stream/operators/Sink/asPublisher.md
@@ -15,8 +15,9 @@ Integration with Reactive Streams, materializes into a `org.reactivestreams.Publ
This method gives you the capability to publish the data from the `Sink` through a Reactive Streams [Publisher](https://www.reactive-streams.org/reactive-streams-1.0.3-javadoc/org/reactivestreams/Publisher.html).
Generally, in Pekko Streams a `Sink` is considered a subscriber, which consumes the data from source. To integrate with other Reactive Stream implementations `Sink.asPublisher` provides a `Publisher` materialized value when run.
Now, the data from this publisher can be consumed by subscribing to it. We can control if we allow more than one downstream subscriber from the single running Pekko stream through the `fanout` parameter.
+
In Java 9, the Reactive Stream API was included in the JDK, and `Publisher` is available through [Flow.Publisher](https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/Flow.Publisher.html).
-Since those APIs are identical but exist at different package namespaces and does not depend on the Reactive Streams package a separate publisher sink for those is available
+Since those APIs are identical but exist at different package namespaces and does not depend on the Reactive Streams package a separate API for those is available
through @scala[`org.apache.pekko.stream.scaladsl.JavaFlowSupport.Sink#asPublisher`]@java[`org.apache.pekko.stream.javadsl.JavaFlowSupport.Sink#asPublisher`].
diff --git a/docs/src/main/paradox/stream/operators/Sink/fromSubscriber.md b/docs/src/main/paradox/stream/operators/Sink/fromSubscriber.md
index 637b0d4e27..bc93195abb 100644
--- a/docs/src/main/paradox/stream/operators/Sink/fromSubscriber.md
+++ b/docs/src/main/paradox/stream/operators/Sink/fromSubscriber.md
@@ -11,4 +11,6 @@ Integration with Reactive Streams, wraps a `org.reactivestreams.Subscriber` as a
## Description
-TODO: We would welcome help on contributing descriptions and examples, see: https://github.com/akka/akka/issues/25646
+In Java 9, the Reactive Stream API was included in the JDK, and `Subscriber` is available through [Flow.Subscriber](https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/Flow.Subscriber.html).
+Since those APIs are identical but exist at different package namespaces and does not depend on the Reactive Streams package a separate API for those is available
+through @scala[`org.apache.pekko.stream.scaladsl.JavaFlowSupport.Sink#fromSubscriber`]@java[`org.apache.pekko.stream.javadsl.JavaFlowSupport.Sink#fromSubscriber`].
diff --git a/docs/src/main/paradox/stream/operators/Source/asSubscriber.md b/docs/src/main/paradox/stream/operators/Source/asSubscriber.md
index 0b07e72447..c0cf7d2d7c 100644
--- a/docs/src/main/paradox/stream/operators/Source/asSubscriber.md
+++ b/docs/src/main/paradox/stream/operators/Source/asSubscriber.md
@@ -1,35 +1,32 @@
# Source.asSubscriber
-Integration with Reactive Streams, materializes into a @javadoc[Subscriber](java.util.concurrent.Flow.Subscriber).
+Integration with Reactive Streams, materializes into a `org.reactivestreams.Subscriber`.
@ref[Source operators](../index.md#source-operators)
## Signature
Scala
-: @@snip[JavaFlowSupport.scala](/stream/src/main/scala-jdk-9/org/apache/pekko/stream/scaladsl/JavaFlowSupport.scala) { #asSubscriber }
+: @@snip[JavaFlowSupport.scala](/stream/src/main/scala/org/apache/pekko/stream/scaladsl/JavaFlowSupport.scala) { #asSubscriber }
Java
-: @@snip[JavaFlowSupport.java](/docs/src/test/java-jdk9-only/jdocs/stream/operators/source/AsSubscriber.java) { #api }
+: @@snip[AsSubscriber.java](/docs/src/test/java/jdocs/stream/operators/source/AsSubscriber.java) { #api }
## 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 `JavaFlowSupport.Source.asSubscriber`.
+[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[java.util.concurrent.Flow.Subscriber](java.util.concurrent.Flow.Subscriber).
-This @javadoc[Subscriber](java.util.concurrent.Flow.Subscriber) can be attached to a
-[Reactive Streams](https://www.reactive-streams.org/) @javadoc[Publisher](java.util.concurrent.Flow.Publisher)
+@javadoc[org.reactivestreams.Subscriber](java.util.concurrent.Flow.Subscriber).
+This `org.reactivestreams.Subscriber` can be attached to a
+[Reactive Streams](https://www.reactive-streams.org/) `org.reactivestreams.Publisher`
to populate it.
-If the API you want to consume elements from provides a @javadoc[Publisher](java.util.concurrent.Flow.Publisher) instead of accepting a @javadoc[Subscriber](java.util.concurrent.Flow.Subscriber), see @ref[fromPublisher](fromPublisher.md).
+If the API you want to consume elements from provides a `org.reactivestreams.Publisher` instead of accepting a `org.reactivestreams.Subscriber`, see @ref[fromPublisher](fromPublisher.md).
-@@@ note
-
-For JDK 8 users: since @javadoc[java.util.concurrent.Flow](java.util.concurrent.Flow) was introduced in JDK version 9,
-if you are still on version 8 you may use the [org.reactivestreams](https://github.com/reactive-streams/reactive-streams-jvm#reactive-streams) library with @apidoc[Source.asSubscriber](Source$) { scala="#asSubscriber[T]:org.apache.pekko.stream.scaladsl.Source[T,org.reactivestreams.Subscriber[T]]" java="#asSubscriber()" }.
-
-@@@
+In Java 9, the Reactive Stream API was included in the JDK, and `Subscriber` is available through [Flow.Subscriber](https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/Flow.Subscriber.html).
+Since those APIs are identical but exist at different package namespaces and does not depend on the Reactive Streams package a separate API for those is available
+through @scala[`org.apache.pekko.stream.scaladsl.JavaFlowSupport.Source#asSubscriber`]@java[`org.apache.pekko.stream.javadsl.JavaFlowSupport.Source#asSubscriber`].
## Example
@@ -44,7 +41,7 @@ backpressure is applied throughout the stream, preventing us from running out of
rows are consumed slower than they are produced by the database.
Scala
-: @@snip [AsSubscriber.scala](/docs/src/test/scala-jdk9-only/docs/stream/operators/source/AsSubscriber.scala) { #imports #example }
+: @@snip [AsSubscriber.scala](/docs/src/test/scala/docs/stream/operators/source/AsSubscriber.scala) { #imports #example }
Java
-: @@snip [AsSubscriber.java](/docs/src/test/java-jdk9-only/jdocs/stream/operators/source/AsSubscriber.java) { #imports #example }
+: @@snip [AsSubscriber.java](/docs/src/test/java/jdocs/stream/operators/source/AsSubscriber.java) { #imports #example }
diff --git a/docs/src/main/paradox/stream/operators/Source/fromPublisher.md b/docs/src/main/paradox/stream/operators/Source/fromPublisher.md
index 69565eb604..1109337f99 100644
--- a/docs/src/main/paradox/stream/operators/Source/fromPublisher.md
+++ b/docs/src/main/paradox/stream/operators/Source/fromPublisher.md
@@ -1,33 +1,30 @@
# Source.fromPublisher
-Integration with Reactive Streams, subscribes to a @javadoc[Publisher](java.util.concurrent.Flow.Publisher).
+Integration with Reactive Streams, subscribes to a `org.reactivestreams.Publisher`.
@ref[Source operators](../index.md#source-operators)
## Signature
Scala
-: @@snip[JavaFlowSupport.scala](/stream/src/main/scala-jdk-9/org/apache/pekko/stream/scaladsl/JavaFlowSupport.scala) { #fromPublisher }
+: @@snip[JavaFlowSupport.scala](/stream/src/main/scala/org/apache/pekko/stream/scaladsl/JavaFlowSupport.scala) { #fromPublisher }
Java
-: @@snip[JavaFlowSupport.java](/docs/src/test/java-jdk9-only/jdocs/stream/operators/source/FromPublisher.java) { #api }
+: @@snip[JavaFlowSupport.java](/docs/src/test/java/jdocs/stream/operators/source/FromPublisher.java) { #api }
## 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 `JavaFlowSupport.Source.fromPublisher`.
-This source will produce the elements from the @javadoc[Publisher](java.util.concurrent.Flow.Publisher),
+[Reactive Streams](https://www.reactive-streams.org/), you can use `Source.fromPublisher`.
+This source will produce the elements from the `org.reactivestreams.Publisher`,
and coordinate backpressure as needed.
-If the API you want to consume elements from accepts a @javadoc[Subscriber](java.util.concurrent.Flow.Subscriber) instead of providing a @javadoc[Publisher](java.util.concurrent.Flow.Publisher), see @ref[asSubscriber](asSubscriber.md).
+If the API you want to consume elements from accepts a `org.reactivestreams.Subscriber` instead of providing a `org.reactivestreams.Publisher`, see @ref[asSubscriber](asSubscriber.md).
-@@@ note
-
-For JDK 8 users: since @javadoc[java.util.concurrent.Flow](java.util.concurrent.Flow) was introduced in JDK version 9,
-if you are still on version 8 you may use the [org.reactivestreams](https://github.com/reactive-streams/reactive-streams-jvm#reactive-streams) library with @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)" }.
-
-@@@
+In Java 9, the Reactive Stream API was included in the JDK, and `Publisher` is available through [Flow.Publisher](https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/Flow.Publisher.html).
+Since those APIs are identical but exist at different package namespaces and does not depend on the Reactive Streams package a separate API for those is available
+through @scala[`org.apache.pekko.stream.scaladsl.JavaFlowSupport.Source#fromPublisher`]@java[`org.apache.pekko.stream.javadsl.JavaFlowSupport.Source#fromPublisher`].
## Example
@@ -41,7 +38,7 @@ backpressure is applied throughout the stream, preventing us from running out of
rows are consumed slower than they are produced by the database.
Scala
-: @@snip [FromPublisher.scala](/docs/src/test/scala-jdk9-only/docs/stream/operators/source/FromPublisher.scala) { #imports #example }
+: @@snip [FromPublisher.scala](/docs/src/test/scala/docs/stream/operators/source/FromPublisher.scala) { #imports #example }
Java
-: @@snip [FromPublisher.java](/docs/src/test/java-jdk9-only/jdocs/stream/operators/source/FromPublisher.java) { #imports #example }
+: @@snip [FromPublisher.java](/docs/src/test/java/jdocs/stream/operators/source/FromPublisher.java) { #imports #example }
diff --git a/docs/src/main/paradox/stream/operators/index.md b/docs/src/main/paradox/stream/operators/index.md
index 8bbf89a16f..059349da52 100644
--- a/docs/src/main/paradox/stream/operators/index.md
+++ b/docs/src/main/paradox/stream/operators/index.md
@@ -8,7 +8,7 @@ These built-in sources are available from @scala[`org.apache.pekko.stream.scalad
| |Operator|Description|
|--|--|--|
|Source|<a name="assourcewithcontext"></a>@ref[asSourceWithContext](Source/asSourceWithContext.md)|Extracts context data from the elements of a `Source` so that it can be turned into a `SourceWithContext` which can propagate that context per element along a stream.|
-|Source|<a name="assubscriber"></a>@ref[asSubscriber](Source/asSubscriber.md)|Integration with Reactive Streams, materializes into a @javadoc[Subscriber](java.util.concurrent.Flow.Subscriber).|
+|Source|<a name="assubscriber"></a>@ref[asSubscriber](Source/asSubscriber.md)|Integration with Reactive Streams, materializes into a `org.reactivestreams.Subscriber`.|
|Source|<a name="combine"></a>@ref[combine](Source/combine.md)|Combine several sources, using a given strategy such as merge or concat, into one source.|
|Source|<a name="completionstage"></a>@ref[completionStage](Source/completionStage.md)|Send the single value of the `CompletionStage` when it completes and there is demand.|
|Source|<a name="completionstagesource"></a>@ref[completionStageSource](Source/completionStageSource.md)|Streams the elements of an asynchronous source once its given *completion* operator completes.|
@@ -23,7 +23,7 @@ These built-in sources are available from @scala[`org.apache.pekko.stream.scalad
|Source|<a name="fromiterator"></a>@ref[fromIterator](Source/fromIterator.md)|Stream the values from an `Iterator`, requesting the next value when there is demand.|
|Source|<a name="fromjavastream"></a>@ref[fromJavaStream](Source/fromJavaStream.md)|Stream the values from a Java 8 `Stream`, requesting the next value when there is demand.|
|Source|<a name="fromoption"></a>@ref[fromOption](Source/fromOption.md)|Create a `Source` from an @scala[`Option[T]`] @java[`Optional<T>`] value, emitting the value if it is present.|
-|Source|<a name="frompublisher"></a>@ref[fromPublisher](Source/fromPublisher.md)|Integration with Reactive Streams, subscribes to a @javadoc[Publisher](java.util.concurrent.Flow.Publisher).|
+|Source|<a name="frompublisher"></a>@ref[fromPublisher](Source/fromPublisher.md)|Integration with Reactive Streams, subscribes to a `org.reactivestreams.Publisher`.|
|Source|<a name="fromsourcecompletionstage"></a>@ref[fromSourceCompletionStage](Source/fromSourceCompletionStage.md)|Deprecated by @ref[`Source.completionStageSource`](Source/completionStageSource.md).|
|Source|<a name="future"></a>@ref[future](Source/future.md)|Send the single value of the `Future` when it completes and there is demand.|
|Source|<a name="futuresource"></a>@ref[futureSource](Source/futureSource.md)|Streams the elements of the given future source once it successfully completes.|
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org