Merge pull request #28762 from ennru/stream-docs-ask
This commit is contained in:
Ignasi Marimon-Clos
GitHub
commit
fcd960ad78
6 changed files with 138 additions and 99 deletions
|
|
@ -1,6 +1,6 @@
|
|||
# ActorFlow.ask
|
||||
|
||||
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.
|
||||
Use the "Ask Pattern" to send each stream element as an `ask` to the target actor (of the new actors API), and expect a reply back that will be emitted downstream.
|
||||
|
||||
@ref[Actor interop operators](../index.md#actor-interop-operators)
|
||||
|
||||
|
|
@ -14,25 +14,48 @@ This operator is included in:
|
|||
version="$akka.version$"
|
||||
}
|
||||
|
||||
@@@div { .group-scala }
|
||||
|
||||
## Signature
|
||||
|
||||
@@signature [ActorFlow.scala](/akka-stream-typed/src/main/scala/akka/stream/typed/scaladsl/ActorFlow.scala) { #ask }
|
||||
|
||||
@@@
|
||||
@apidoc[ActorFlow.ask](ActorFlow$) { scala="#ask%5BI,Q,A](ref:akka.actor.typed.ActorRef%5BQ])(makeMessage:(I,akka.actor.typed.ActorRef%5BA])=%3EQ)(implicittimeout:akka.util.Timeout):akka.stream.scaladsl.Flow%5BI,A,akka.NotUsed]" java="#ask(akka.actor.typed.ActorRef,java.time.Duration,java.util.function.BiFunction)" }
|
||||
|
||||
## Description
|
||||
|
||||
Emit the contents of a file, as `ByteString`s, materializes into a @scala[`Future`] @java[`CompletionStage`] which will be completed with
|
||||
a `IOResult` upon reaching the end of the file or if there is a failure.
|
||||
Use the @ref[Ask pattern](../../../typed/interaction-patterns.md#request-response-with-ask-from-outside-an-actor) to send a request-reply message to the target `ref` actor.
|
||||
If any of the asks times out it will fail the stream with an @apidoc[AskTimeoutException].
|
||||
|
||||
The `ask` operator requires
|
||||
|
||||
* the actor `ref`,
|
||||
* a `makeMessage` function to create the message sent to the actor from the incoming element and the actor ref accepting the actor's reply message,
|
||||
* and a timeout.
|
||||
|
||||
See also:
|
||||
|
||||
* @ref[Flow.ask](../Source-or-Flow/ask.md) for the classic actors variant
|
||||
|
||||
## Examples
|
||||
|
||||
The `ActorFlow.ask` sends a message to the actor which expects `Asking` messages which contain the actor ref for replies of type `Reply`. The replies are emitted when received and the `map` extracts the message `String`.
|
||||
|
||||
Scala
|
||||
: @@snip [ask.scala](/akka-stream-typed/src/test/scala/akka/stream/typed/scaladsl/ActorFlowSpec.scala) { #imports #ask-actor #ask }
|
||||
: @@snip [ask.scala](/akka-stream-typed/src/test/scala/docs/scaladsl/ActorFlowSpec.scala) { #imports #ask-actor #ask }
|
||||
|
||||
Java
|
||||
: @@snip [ask.java](/akka-stream-typed/src/test/java/akka/stream/typed/javadsl/ActorFlowCompileTest.java) { #ask-actor #ask }
|
||||
: @@snip [ask.java](/akka-stream-typed/src/test/java/docs/javadsl/ActorFlowCompileTest.java) { #ask-actor #ask }
|
||||
|
||||
|
||||
## Reactive Streams semantics
|
||||
|
||||
@@@div { .callout }
|
||||
|
||||
**emits** when the futures (in submission order) created by the ask pattern internally are completed
|
||||
|
||||
**backpressures** when the number of futures reaches the configured parallelism and the downstream backpressures
|
||||
|
||||
**completes** when upstream completes and all futures have been completed and all elements have been emitted
|
||||
|
||||
**fails** when the passed in actor terminates, or a timeout is exceeded in any of the asks performed
|
||||
|
||||
**cancels** when downstream cancels
|
||||
|
||||
@@@
|
||||
|
|
|
|||
|
|
@ -1,27 +1,28 @@
|
|||
# Flow.ask
|
||||
|
||||
Use the `ask` pattern to send a request-reply message to the target `ref` actor.
|
||||
Use the "Ask Pattern" to send a request-reply message to the target `ref` actor (of the classic actors API).
|
||||
|
||||
@ref[Asynchronous operators](../index.md#asynchronous-operators)
|
||||
@ref[Actor interop operators](../index.md#actor-interop-operators)
|
||||
|
||||
@@@ div { .group-scala }
|
||||
## Signature
|
||||
|
||||
@@signature [Flow.scala](/akka-stream/src/main/scala/akka/stream/scaladsl/Flow.scala) { #ask }
|
||||
@@@
|
||||
@apidoc[Flow.ask](Flow$) { scala="#ask%5BS](ref:akka.actor.ActorRef)(implicittimeout:akka.util.Timeout,implicittag:scala.reflect.ClassTag%5BS]):FlowOps.this.Repr%5BS]" java="#ask(akka.actor.ActorRef,java.lang.Class,akka.util.Timeout)" }
|
||||
|
||||
## Description
|
||||
|
||||
Use the `ask` pattern to send a request-reply message to the target `ref` actor.
|
||||
Use the @ref[Ask Pattern](../../../actors.md#ask-send-and-receive-future) to send a request-reply message to the target `ref` actor.
|
||||
If any of the asks times out it will fail the stream with a @apidoc[AskTimeoutException].
|
||||
|
||||
The @java[`mapTo` class]@scala[`S` generic] parameter is used to cast the responses from the actor to the expected outgoing flow type.
|
||||
|
||||
Similar to the plain ask pattern, the target actor is allowed to reply with `akka.util.Status`.
|
||||
An `akka.util.Status#Failure` will cause the operator to fail with the cause carried in the `Failure` message.
|
||||
Similar to the plain ask pattern, the target actor is allowed to reply with @apidoc[akka.actor.Status].
|
||||
An @apidoc[akka.actor.Status.Failure] will cause the operator to fail with the cause carried in the `Failure` message.
|
||||
|
||||
Adheres to the @scala[@scaladoc[`ActorAttributes.SupervisionStrategy`](akka.stream.ActorAttributes$$SupervisionStrategy)]
|
||||
@java[`ActorAttributes.SupervisionStrategy`] attribute.
|
||||
Adheres to the @apidoc[ActorAttributes.SupervisionStrategy] attribute.
|
||||
|
||||
See also:
|
||||
|
||||
* @ref[ActorFlow.ask](../ActorFlow/ask.md) for the `akka.actor.typed.ActorRef[_]` variant
|
||||
|
||||
## Reactive Streams semantics
|
||||
|
||||
|
|
|
|||
|
|
@ -198,7 +198,6 @@ operation at the same time (usually handling the completion of a @scala[`Future`
|
|||
|
||||
| |Operator|Description|
|
||||
|--|--|--|
|
||||
|Source/Flow|<a name="ask"></a>@ref[ask](Source-or-Flow/ask.md)|Use the `ask` pattern to send a request-reply message to the target `ref` actor.|
|
||||
|Source/Flow|<a name="mapasync"></a>@ref[mapAsync](Source-or-Flow/mapAsync.md)|Pass incoming elements to a function that return a @scala[`Future`] @java[`CompletionStage`] result.|
|
||||
|Source/Flow|<a name="mapasyncunordered"></a>@ref[mapAsyncUnordered](Source-or-Flow/mapAsyncUnordered.md)|Like `mapAsync` but @scala[`Future`] @java[`CompletionStage`] results are passed downstream as they arrive regardless of the order of the elements that triggered them.|
|
||||
|
||||
|
|
@ -322,7 +321,8 @@ Operators meant for inter-operating between Akka Streams and Actors:
|
|||
|ActorSink|<a name="actorref"></a>@ref[actorRef](ActorSink/actorRef.md)|Sends the elements of the stream to the given @java[`ActorRef<T>`]@scala[`ActorRef[T]`], without considering backpressure.|
|
||||
|ActorSource|<a name="actorrefwithbackpressure"></a>@ref[actorRefWithBackpressure](ActorSource/actorRefWithBackpressure.md)|Materialize an @java[`ActorRef<T>`]@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|<a name="actorrefwithbackpressure"></a>@ref[actorRefWithBackpressure](ActorSink/actorRefWithBackpressure.md)|Sends the elements of the stream to the given @java[`ActorRef<T>`]@scala[`ActorRef[T]`] with backpressure, to be able to signal demand when the actor is ready to receive more elements.|
|
||||
|ActorFlow|<a name="ask"></a>@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.|
|
||||
|Source/Flow|<a name="ask"></a>@ref[ask](Source-or-Flow/ask.md)|Use the "Ask Pattern" to send a request-reply message to the target `ref` actor (of the classic actors API).|
|
||||
|ActorFlow|<a name="ask"></a>@ref[ask](ActorFlow/ask.md)|Use the "Ask Pattern" to send each stream element as an `ask` to the target actor (of the new actors API), and expect a reply back that will be emitted downstream.|
|
||||
|Source/Flow|<a name="watch"></a>@ref[watch](Source-or-Flow/watch.md)|Watch a specific `ActorRef` and signal a failure downstream once the actor terminates.|
|
||||
|
||||
## Compression operators
|
||||
|
|
|
|||
|
|
@ -1,63 +0,0 @@
|
|||
/*
|
||||
* Copyright (C) 2018-2020 Lightbend Inc. <https://www.lightbend.com>
|
||||
*/
|
||||
|
||||
package akka.stream.typed.javadsl;
|
||||
|
||||
import akka.actor.typed.ActorRef;
|
||||
import akka.actor.typed.ActorSystem;
|
||||
import akka.stream.ActorMaterializer;
|
||||
import akka.stream.javadsl.Sink;
|
||||
import akka.stream.javadsl.Source;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
|
||||
public class ActorFlowCompileTest {
|
||||
|
||||
interface Protocol {}
|
||||
|
||||
class Init implements Protocol {}
|
||||
|
||||
class Msg implements Protocol {}
|
||||
|
||||
class Complete implements Protocol {}
|
||||
|
||||
class Failure implements Protocol {
|
||||
public Exception ex;
|
||||
}
|
||||
|
||||
{
|
||||
final ActorSystem<String> system = null;
|
||||
}
|
||||
|
||||
static
|
||||
// #ask-actor
|
||||
class AskMe {
|
||||
final String payload;
|
||||
final ActorRef<String> replyTo;
|
||||
|
||||
AskMe(String payload, ActorRef<String> replyTo) {
|
||||
this.payload = payload;
|
||||
this.replyTo = replyTo;
|
||||
}
|
||||
}
|
||||
|
||||
// #ask-actor
|
||||
|
||||
{
|
||||
final ActorRef<AskMe> ref = null;
|
||||
|
||||
// #ask
|
||||
Duration timeout = Duration.of(1, ChronoUnit.SECONDS);
|
||||
|
||||
Source.repeat("hello").via(ActorFlow.ask(ref, timeout, AskMe::new)).to(Sink.ignore());
|
||||
|
||||
Source.repeat("hello")
|
||||
.via(
|
||||
ActorFlow.<String, AskMe, String>ask(
|
||||
ref, timeout, (msg, replyTo) -> new AskMe(msg, replyTo)))
|
||||
.to(Sink.ignore());
|
||||
// #ask
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,67 @@
|
|||
/*
|
||||
* Copyright (C) 2018-2020 Lightbend Inc. <https://www.lightbend.com>
|
||||
*/
|
||||
|
||||
package docs.javadsl;
|
||||
|
||||
import akka.NotUsed;
|
||||
// #ask-actor
|
||||
import akka.actor.typed.ActorRef;
|
||||
import akka.actor.typed.ActorSystem;
|
||||
import akka.stream.javadsl.Flow;
|
||||
import akka.stream.javadsl.Sink;
|
||||
import akka.stream.javadsl.Source;
|
||||
import akka.stream.typed.javadsl.ActorFlow;
|
||||
|
||||
// #ask-actor
|
||||
import java.time.Duration;
|
||||
|
||||
public class ActorFlowCompileTest {
|
||||
|
||||
final ActorSystem<String> system = null;
|
||||
|
||||
static
|
||||
// #ask-actor
|
||||
class Asking {
|
||||
final String payload;
|
||||
final ActorRef<Reply> replyTo;
|
||||
|
||||
public Asking(String payload, ActorRef<Reply> replyTo) {
|
||||
this.payload = payload;
|
||||
this.replyTo = replyTo;
|
||||
}
|
||||
}
|
||||
|
||||
// #ask-actor
|
||||
static
|
||||
// #ask-actor
|
||||
class Reply {
|
||||
public final String msg;
|
||||
|
||||
public Reply(String msg) {
|
||||
this.msg = msg;
|
||||
}
|
||||
}
|
||||
|
||||
// #ask-actor
|
||||
|
||||
{
|
||||
// #ask
|
||||
final ActorRef<Asking> actorRef = // ???
|
||||
// #ask
|
||||
null;
|
||||
|
||||
// #ask
|
||||
Duration timeout = Duration.ofSeconds(1);
|
||||
|
||||
// method reference notation
|
||||
Flow<String, Reply, NotUsed> askFlow = ActorFlow.ask(actorRef, timeout, Asking::new);
|
||||
|
||||
// explicit creation of the sent message
|
||||
Flow<String, Reply, NotUsed> askFlowExplicit =
|
||||
ActorFlow.ask(actorRef, timeout, (msg, replyTo) -> new Asking(msg, replyTo));
|
||||
|
||||
Source.repeat("hello").via(askFlow).map(reply -> reply.msg).runWith(Sink.seq(), system);
|
||||
// #ask
|
||||
}
|
||||
}
|
||||
|
|
@ -2,26 +2,30 @@
|
|||
* Copyright (C) 2018-2020 Lightbend Inc. <https://www.lightbend.com>
|
||||
*/
|
||||
|
||||
package akka.stream.typed.scaladsl
|
||||
package docs.scaladsl
|
||||
|
||||
import akka.NotUsed
|
||||
//#imports
|
||||
import akka.stream.scaladsl._
|
||||
import akka.stream.scaladsl.{ Flow, Sink, Source }
|
||||
import akka.stream.typed.scaladsl.ActorFlow
|
||||
import akka.actor.typed.ActorRef
|
||||
import akka.actor.typed.scaladsl.Behaviors
|
||||
|
||||
import scala.concurrent.duration._
|
||||
//#imports
|
||||
import akka.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
|
||||
import akka.stream.testkit.TestSubscriber
|
||||
import org.scalatest.wordspec.AnyWordSpecLike
|
||||
|
||||
//#imports
|
||||
import akka.stream.testkit.TestSubscriber
|
||||
|
||||
import scala.collection.immutable
|
||||
import scala.concurrent.duration._
|
||||
import scala.concurrent.{ Await, Future }
|
||||
|
||||
object ActorFlowSpec {
|
||||
//#ask-actor
|
||||
final case class Asking(s: String, replyTo: ActorRef[Reply])
|
||||
final case class Reply(s: String)
|
||||
final case class Reply(msg: String)
|
||||
|
||||
//#ask-actor
|
||||
}
|
||||
|
||||
class ActorFlowSpec extends ScalaTestWithActorTestKit with AnyWordSpecLike {
|
||||
|
|
@ -59,14 +63,21 @@ class ActorFlowSpec extends ScalaTestWithActorTestKit with AnyWordSpecLike {
|
|||
//#ask-actor
|
||||
|
||||
//#ask
|
||||
val in: Future[immutable.Seq[Reply]] =
|
||||
Source(1 to 50)
|
||||
.map(_.toString)
|
||||
.via(ActorFlow.ask(ref)((el, replyTo: ActorRef[Reply]) => Asking(el, replyTo)))
|
||||
.runWith(Sink.seq)
|
||||
//#ask
|
||||
implicit val timeout: akka.util.Timeout = 1.second
|
||||
|
||||
in.futureValue shouldEqual List.tabulate(51)(i => Reply(s"$i!!!")).drop(1)
|
||||
val askFlow: Flow[String, Reply, NotUsed] =
|
||||
ActorFlow.ask(ref)(Asking.apply)
|
||||
|
||||
// explicit creation of the sent message
|
||||
val askFlowExplicit: Flow[String, Reply, NotUsed] =
|
||||
ActorFlow.ask(ref)(makeMessage = (el, replyTo: ActorRef[Reply]) => Asking(el, replyTo))
|
||||
|
||||
val in: Future[immutable.Seq[String]] =
|
||||
Source(1 to 50).map(_.toString).via(askFlow).map(_.msg).runWith(Sink.seq)
|
||||
//#ask
|
||||
askFlowExplicit.map(identity)
|
||||
|
||||
in.futureValue shouldEqual List.tabulate(51)(i => s"$i!!!").drop(1)
|
||||
}
|
||||
|
||||
"signal ask timeout failure" in {
|
||||
Loading…
Add table
Add a link
Reference in a new issue