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 that will be emitted downstream.

Actor interop operators

Dependency

This operator is included in:

sbt
Maven
Gradle
val PekkoVersion = "1.1.3"
libraryDependencies += "org.apache.pekko" %% "pekko-stream-typed" % PekkoVersion
<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.apache.pekko</groupId>
      <artifactId>pekko-bom_${scala.binary.version}</artifactId>
      <version>1.1.3</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-stream-typed_${scala.binary.version}</artifactId>
  </dependency>
</dependencies>
def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation platform("org.apache.pekko:pekko-bom_${versions.ScalaBinary}:1.1.3")

  implementation "org.apache.pekko:pekko-stream-typed_${versions.ScalaBinary}"
}

Signature

ActorFlow.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 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
  • a timeout.

See also:

  • Flow.ask for the classic actors variant

Examples

The ActorFlow.ask sends a message to the actor. The actor expects Asking messages which contain the actor ref for replies of type Reply. When the actor for replies receives a reply, the ActorFlow.ask stream stage emits the reply and the map extracts the message String.

Scala
Java
sourceimport org.apache.pekko
import pekko.stream.scaladsl.{ Flow, Sink, Source }
import pekko.stream.typed.scaladsl.ActorFlow
import pekko.actor.typed.ActorRef
import pekko.actor.typed.scaladsl.Behaviors
import pekko.util.Timeout

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

final case class AskingWithStatus(s: String, replyTo: ActorRef[StatusReply[String]])

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

implicit val timeout: 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)
sourceimport org.apache.pekko.actor.typed.ActorRef;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.pattern.StatusReply;
import org.apache.pekko.stream.javadsl.Flow;
import org.apache.pekko.stream.javadsl.Sink;
import org.apache.pekko.stream.javadsl.Source;
import org.apache.pekko.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;
  }
}

static class AskingWithStatus {
  final String payload;
  final ActorRef<StatusReply<String>> replyTo;

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

class Reply {
  public final String msg;

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

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

Flow<String, String, NotUsed> askFlowExplicitWithStatus =
    ActorFlow.askWithStatus(
        actorWithStatusRef, timeout, (msg, replyTo) -> new AskingWithStatus(msg, replyTo));

Source.repeat("hello").via(askFlow).map(reply -> reply.msg).runWith(Sink.seq(), system);

Reactive Streams semantics

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 when any of the asks exceed a timeout

cancels when downstream cancels

1.1.3