From 023b379e3c8dffd75b31b0b918c97f9929de7685 Mon Sep 17 00:00:00 2001 From: Helena Edelson Date: Fri, 4 Oct 2019 10:37:56 -0700 Subject: [PATCH] Migrate streams doc under typed to operators (#27901) --- .../stream/operators/ActorSink/actorRef.md | 6 +-- .../ActorSink/actorRefWithBackpressure.md | 35 +++++++++++++ .../stream/operators/ActorSource/actorRef.md | 35 +++++++++++++ .../ActorSource/actorRefWithBackpressure.md | 19 +++++-- .../main/paradox/stream/operators/index.md | 8 ++- .../paradox/stream/stream-integrations.md | 35 +++++++++++++ akka-docs/src/main/paradox/typed/index.md | 2 - akka-docs/src/main/paradox/typed/stream.md | 51 ------------------- project/StreamOperatorsIndexGenerator.scala | 4 +- 9 files changed, 130 insertions(+), 65 deletions(-) create mode 100644 akka-docs/src/main/paradox/stream/operators/ActorSink/actorRefWithBackpressure.md create mode 100644 akka-docs/src/main/paradox/stream/operators/ActorSource/actorRef.md delete mode 100644 akka-docs/src/main/paradox/typed/stream.md diff --git a/akka-docs/src/main/paradox/stream/operators/ActorSink/actorRef.md b/akka-docs/src/main/paradox/stream/operators/ActorSink/actorRef.md index 93c58b661a..924f1f1ccf 100644 --- a/akka-docs/src/main/paradox/stream/operators/ActorSink/actorRef.md +++ b/akka-docs/src/main/paradox/stream/operators/ActorSink/actorRef.md @@ -1,11 +1,9 @@ # ActorSink.actorRef -Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`]. +Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`], without considering backpressure. @ref[Actor interop operators](../index.md#actor-interop-operators) -@@@div { .group-scala } - ## Dependency This operator is included in: @@ -16,6 +14,8 @@ This operator is included in: version="$akka.version$" } +@@@div { .group-scala } + ## Signature @@signature [ActorSink.scala](/akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSink.scala) { #actorRef } diff --git a/akka-docs/src/main/paradox/stream/operators/ActorSink/actorRefWithBackpressure.md b/akka-docs/src/main/paradox/stream/operators/ActorSink/actorRefWithBackpressure.md new file mode 100644 index 0000000000..8f7c828fc0 --- /dev/null +++ b/akka-docs/src/main/paradox/stream/operators/ActorSink/actorRefWithBackpressure.md @@ -0,0 +1,35 @@ +# ActorSink.actorRefWithBackpressure + +Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`] with backpressure, to be able to signal demand when the actor is ready to receive more elements. + +@ref[Actor interop operators](../index.md#actor-interop-operators) + +## Dependency + +This operator is included in: + +@@dependency[sbt,Maven,Gradle] { + group="com.typesafe.akka" + artifact="akka-stream-typed_$scala.binary_version$" + version="$akka.version$" +} + +@@@div { .group-scala } + +## Signature + +@@signature [ActorSink.scala](/akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSink.scala) { #actorRefWithBackpressure } + +@@@ + +## Description + +Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`] with backpressure, to be able to signal demand when the actor is ready to receive more elements. + +## Examples + +Scala +: @@snip [ActorSourceSinkExample.scala](/akka-stream-typed/src/test/scala/docs/akka/stream/typed/ActorSourceSinkExample.scala) { #actor-sink-ref-with-backpressure } + +Java +: @@snip [ActorSinkWithAckExample.java](/akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSinkWithAckExample.java) { #actor-sink-ref-with-backpressure } diff --git a/akka-docs/src/main/paradox/stream/operators/ActorSource/actorRef.md b/akka-docs/src/main/paradox/stream/operators/ActorSource/actorRef.md new file mode 100644 index 0000000000..ed16e84aee --- /dev/null +++ b/akka-docs/src/main/paradox/stream/operators/ActorSource/actorRef.md @@ -0,0 +1,35 @@ +# ActorSource.actorRef + +Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`]; sending messages to it will emit them on the stream only if they are of the same type as the stream. + +@ref[Actor interop operators](../index.md#actor-interop-operators) + +## Dependency + +This operator is included in: + +@@dependency[sbt,Maven,Gradle] { + group="com.typesafe.akka" + artifact="akka-stream-typed_$scala.binary_version$" + version="$akka.version$" +} + +@@@div { .group-scala } + +## Signature + +@@signature [ActorSource.scala](/akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSource.scala) { #actorRef } + +@@@ + +## Description + +Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`] which only accepts messages that are of the same type as the stream. + +## Examples + +Scala +: @@snip [ActorSourceSinkExample.scala](/akka-stream-typed/src/test/scala/docs/akka/stream/typed/ActorSourceSinkExample.scala) { #actor-source-ref } + +Java +: @@snip [ActorSourceExample.java](/akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSourceExample.java) { #actor-source-ref } diff --git a/akka-docs/src/main/paradox/stream/operators/ActorSource/actorRefWithBackpressure.md b/akka-docs/src/main/paradox/stream/operators/ActorSource/actorRefWithBackpressure.md index 0aa08dcf25..b2c52d4eb6 100644 --- a/akka-docs/src/main/paradox/stream/operators/ActorSource/actorRefWithBackpressure.md +++ b/akka-docs/src/main/paradox/stream/operators/ActorSource/actorRefWithBackpressure.md @@ -1,13 +1,24 @@ -# actorRefWithBackpressure +# ActorSource.actorRefWithBackpressure Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`]; sending messages to it will emit them on the stream. The source acknowledges reception after emitting a message, to provide back pressure from the source. -@ref[Source operators](../index.md#source-operators) +@ref[Actor interop operators](../index.md#actor-interop-operators) + +## Dependency + +This operator is included in: + +@@dependency[sbt,Maven,Gradle] { + group="com.typesafe.akka" + artifact="akka-stream-typed_$scala.binary_version$" + version="$akka.version$" +} + +@@@div { .group-scala } -@@@ div { .group-scala } ## Signature -@@signature [Source.scala](/akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSource.scala) { #actorRefWithBackpressure } +@@signature [ActorSource.scala](/akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSource.scala) { #actorRefWithBackpressure } @@@ ## Description diff --git a/akka-docs/src/main/paradox/stream/operators/index.md b/akka-docs/src/main/paradox/stream/operators/index.md index 5fc885d36e..69e5df1ed5 100644 --- a/akka-docs/src/main/paradox/stream/operators/index.md +++ b/akka-docs/src/main/paradox/stream/operators/index.md @@ -9,7 +9,6 @@ These built-in sources are available from @scala[`akka.stream.scaladsl.Source`] |--|--|--| |Source|@ref[actorRef](Source/actorRef.md)|Materialize an `ActorRef`; sending messages to it will emit them on the stream.| |Source|@ref[actorRefWithBackpressure](Source/actorRefWithBackpressure.md)|Materialize an `ActorRef`; sending messages to it will emit them on the stream. The source acknowledges reception after emitting a message, to provide back pressure from the source.| -|ActorSource|@ref[actorRefWithBackpressure](ActorSource/actorRefWithBackpressure.md)|Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`]; sending messages to it will emit them on the stream. The source acknowledges reception after emitting a message, to provide back pressure from the source.| |Source|@ref[asSourceWithContext](Source/asSourceWithContext.md)|Turns a Source into a SourceWithContext which can propagate a context per element along a stream.| |Source|@ref[asSubscriber](Source/asSubscriber.md)|Integration with Reactive Streams, materializes into a `org.reactivestreams.Subscriber`.| |Source|@ref[combine](Source/combine.md)|Combine several sources, using a given strategy such as merge or concat, into one source.| @@ -295,7 +294,10 @@ Operators meant for inter-operating between Akka Streams and Actors: | |Operator|Description| |--|--|--| -|ActorSink|@ref[actorRef](ActorSink/actorRef.md)|Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`].| +|ActorSource|@ref[actorRef](ActorSource/actorRef.md)|Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`]; sending messages to it will emit them on the stream only if they are of the same type as the stream.| +|ActorSink|@ref[actorRef](ActorSink/actorRef.md)|Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`], without considering backpressure.| +|ActorSource|@ref[actorRefWithBackpressure](ActorSource/actorRefWithBackpressure.md)|Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`]; sending messages to it will emit them on the stream. The source acknowledges reception after emitting a message, to provide back pressure from the source.| +|ActorSink|@ref[actorRefWithBackpressure](ActorSink/actorRefWithBackpressure.md)|Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`] with backpressure, to be able to signal demand when the actor is ready to receive more elements.| |ActorFlow|@ref[ask](ActorFlow/ask.md)|Use the `AskPattern` to send each element as an `ask` to the target actor, and expect a reply back that will be sent further downstream.| ## Error handling @@ -464,9 +466,11 @@ For more background see the @ref[Error Handling in Streams](../stream-error.md) * [withBackoff](RestartFlow/withBackoff.md) * [onFailuresWithBackoff](RestartFlow/onFailuresWithBackoff.md) * [withBackoff](RestartSink/withBackoff.md) +* [actorRef](ActorSource/actorRef.md) * [actorRefWithBackpressure](ActorSource/actorRefWithBackpressure.md) * [ask](ActorFlow/ask.md) * [actorRef](ActorSink/actorRef.md) +* [actorRefWithBackpressure](ActorSink/actorRefWithBackpressure.md) * [Partition](Partition.md) * [Broadcast](Broadcast.md) * [Balance](Balance.md) diff --git a/akka-docs/src/main/paradox/stream/stream-integrations.md b/akka-docs/src/main/paradox/stream/stream-integrations.md index c976ee972c..e5084dcd06 100644 --- a/akka-docs/src/main/paradox/stream/stream-integrations.md +++ b/akka-docs/src/main/paradox/stream/stream-integrations.md @@ -18,6 +18,8 @@ For piping the elements of a stream as messages to an ordinary actor you can use Messages can be sent to a stream with `Source.queue` or via the `ActorRef` that is materialized by `Source.actorRef`. +Additionally you can use `ActorSource.actorRef`, `ActorSource.actorRefWithBackpressure`, `ActorSink.actorRef` and `ActorSink.actorRefWithBackpressure` shown below. + ### ask @@@ note @@ -181,6 +183,39 @@ Scala Java : @@snip [IntegrationDocTest.java](/akka-docs/src/test/java/jdocs/stream/IntegrationDocTest.java) { #source-actorRef } + +### ActorSource.actorRef + +Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`]; sending messages to it will emit them on the stream only if they are of the same type as the stream. + +@@@ note + See also: @ref[ActorSource.actorRef operator reference docs](operators/ActorSource/actorRef.md) +@@@ + +### ActorSource.actorRefWithBackpressure + +Materialize an @java[`ActorRef`]@scala[`ActorRef[T]`]; sending messages to it will emit them on the stream. The source acknowledges reception after emitting a message, to provide back pressure from the source. + +@@@ note + See also: @ref[ActorSource.actorRefWithBackpressure operator reference docs](operators/ActorSource/actorRefWithBackpressure.md) +@@@ + +### ActorSink.actorRef + +Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`], without considering backpressure. + +@@@ note + See also: @ref[ActorSink.actorRef operator reference docs](operators/ActorSink/actorRef.md) +@@@ + +### ActorSink.actorRefWithBackpressure + +Sends the elements of the stream to the given @java[`ActorRef`]@scala[`ActorRef[T]`] with backpressure, to be able to signal demand when the actor is ready to receive more elements. + +@@@ note + See also: @ref[ActorSink.actorRefWithBackpressure operator reference docs](operators/ActorSink/actorRefWithBackpressure.md) +@@@ + ## Integrating with External Services Stream transformations and side effects involving external non-stream based services can be diff --git a/akka-docs/src/main/paradox/typed/index.md b/akka-docs/src/main/paradox/typed/index.md index 826b4c413a..b99a7fb59a 100644 --- a/akka-docs/src/main/paradox/typed/index.md +++ b/akka-docs/src/main/paradox/typed/index.md @@ -8,14 +8,12 @@ project.description: Using Akka to build reliable multi-core applications distri @@@ index * [actors](actors.md) - * [actor-lifecycle](actor-lifecycle.md) * [interaction patterns](interaction-patterns.md) * [fault-tolerance](fault-tolerance.md) * [actor-discovery](actor-discovery.md) * [routers](routers.md) * [stash](stash.md) -* [stream](stream.md) * [fsm](fsm.md) * [dispatchers](dispatchers.md) * [mailboxes](mailboxes.md) diff --git a/akka-docs/src/main/paradox/typed/stream.md b/akka-docs/src/main/paradox/typed/stream.md deleted file mode 100644 index 08cc14af59..0000000000 --- a/akka-docs/src/main/paradox/typed/stream.md +++ /dev/null @@ -1,51 +0,0 @@ -# Streams - -@@project-info{ projectId="akka-stream-typed" } - -## Dependency - -To use Akka Streams Typed, add the module to your project: - -@@dependency [sbt,Maven,Gradle] { - group=com.typesafe.akka - artifact=akka-stream-typed_$scala.binary_version$ - version=$akka.version$ -} - -## Introduction - -@ref:[Akka Streams](../stream/index.md) make it easy to model type-safe message processing pipelines. With typed actors it is possible to connect streams to actors without losing the type information. - -This module contains typed alternatives to the @ref:[already existing `ActorRef` sources and sinks](../stream/stream-integrations.md) together with a factory methods for -@scala[@scaladoc[`ActorMaterializer`](akka.stream.typed.scaladsl.ActorMaterializer)]@java[@javadoc[`ActorMaterializerFactory`](akka.stream.typed.javadsl.ActorMaterializerFactory)] which takes a typed `ActorSystem`. - -The materializer created from these factory methods and sources together with sinks contained in this module can be mixed and matched with the original Akka Streams building blocks from the original module. - -## Actor Source - -A stream that is driven by messages sent to a particular actor can be started with @scala[@scaladoc[`ActorSource.actorRef`](akka.stream.typed.scaladsl.ActorSource#actorRef)]@java[@javadoc[`ActorSource.actorRef`](akka.stream.typed.javadsl.ActorSource#actorRef)]. This source materializes to a typed `ActorRef` which only accepts messages that are of the same type as the stream. - -Scala -: @@snip [ActorSourceSinkExample.scala](/akka-stream-typed/src/test/scala/docs/akka/stream/typed/ActorSourceSinkExample.scala) { #actor-source-ref } - -Java -: @@snip [ActorSourceExample.java](/akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSourceExample.java) { #actor-source-ref } - - -## Actor Sink - -There are two sinks available that accept typed `ActorRef`s. To send all of the messages from a stream to an actor without considering backpressure, use @scala[@scaladoc[`ActorSink.actorRef`](akka.stream.typed.scaladsl.ActorSink#actorRef)]@java[@javadoc[`ActorSink.actorRef`](akka.stream.typed.javadsl.ActorSink#actorRef)]. - -Scala -: @@snip [ActorSourceSinkExample.scala](/akka-stream-typed/src/test/scala/docs/akka/stream/typed/ActorSourceSinkExample.scala) { #actor-sink-ref } - -Java -: @@snip [ActorSinkExample.java](/akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSinkExample.java) { #actor-sink-ref } - -For an actor to be able to react to backpressure, a protocol needs to be introduced between the actor and the stream. Use @scala[@scaladoc[`ActorSink.actorRefWithBackpressure`](akka.stream.typed.scaladsl.ActorSink#actorRefWithBackpressure)]@java[@javadoc[`ActorSink.actorRefWithBackpressure`](akka.stream.typed.javadsl.ActorSink#actorRefWithBackpressure)] to be able to signal demand when the actor is ready to receive more elements. - -Scala -: @@snip [ActorSourceSinkExample.scala](/akka-stream-typed/src/test/scala/docs/akka/stream/typed/ActorSourceSinkExample.scala) { #actor-sink-ref-with-backpressure } - -Java -: @@snip [ActorSinkWithAckExample.java](/akka-stream-typed/src/test/java/docs/akka/stream/typed/ActorSinkWithAckExample.java) { #actor-sink-ref-with-backpressure } diff --git a/project/StreamOperatorsIndexGenerator.scala b/project/StreamOperatorsIndexGenerator.scala index 0a2962c325..a09b04c032 100644 --- a/project/StreamOperatorsIndexGenerator.scala +++ b/project/StreamOperatorsIndexGenerator.scala @@ -109,11 +109,9 @@ object StreamOperatorsIndexGenerator extends AutoPlugin { "actorRefWithAck" // deprecated ), "ActorSink" -> Seq( - "actorRefWithBackpressure", "actorRefWithAck" // deprecated ), "ActorSource" -> Seq( - "actorRef", "actorRefWithAck" // deprecated ) ) @@ -164,7 +162,7 @@ object StreamOperatorsIndexGenerator extends AutoPlugin { "akka-stream-typed/src/main/scala/akka/stream/typed/javadsl/ActorFlow.scala", "akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorFlow.scala", "akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorSink.scala", - "akka-stream-typed/src/main/scala/akka/stream/typed/javadsl/ActorSink.scala", + "akka-stream-typed/src/main/scala/akka/stream/typed/javadsl/ActorSink.scala" ).flatMap{ f => val slashesNr = f.count(_ == '/') val element = f.split("/")(slashesNr).split("\\.")(0)