ActorFlow.ask

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.

Actor interop operators

Dependency

This operator is included in:

sbt

val AkkaVersion = "2.6.8"
libraryDependencies += "com.typesafe.akka" %% "akka-stream-typed" % AkkaVersion

Maven

<properties>
  <akka.version>2.6.8</akka.version>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-stream-typed_${scala.binary.version}</artifactId>
  <version>${akka.version}</version>
</dependency>

Gradle

versions += [
  AkkaVersion: "2.6.8",
  ScalaBinary: "2.13"
]
dependencies {
  compile group: 'com.typesafe.akka', name: "akka-stream-typed_${versions.ScalaBinary}", version: versions.AkkaVersion
}

Signature

ActorFlow.askActorFlow.ask

Description

Use the Ask pattern 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 AskTimeoutExceptionAskTimeoutException.

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:

• Flow.ask 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

import akka.stream.scaladsl.{ Flow, Sink, Source }
import akka.stream.typed.scaladsl.ActorFlow
import akka.actor.typed.ActorRef
import akka.actor.typed.scaladsl.Behaviors

final case class Asking(s: String, replyTo: ActorRef[Reply])
final case class Reply(msg: String)

    val ref = spawn(Behaviors.receiveMessage[Asking] { asking =>
      asking.replyTo ! Reply(asking.s + "!!!")
      Behaviors.same
    })

implicit val timeout: akka.util.Timeout = 1.second

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)

Java

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;

class Asking {
  final String payload;
  final ActorRef<Reply> replyTo;

  public Asking(String payload, ActorRef<Reply> replyTo) {
    this.payload = payload;
    this.replyTo = replyTo;
  }
}

class Reply {
  public final String msg;

  public Reply(String msg) {
    this.msg = msg;
  }
}

final ActorRef<Asking> actorRef = // ???
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);

Reactive Streams semantics