ActorFlow.askWithContext

Use the “Ask Pattern” to send each stream element (without the context) as an ask to the target actor (of the new actors API), and expect a reply of Type StatusReply[T]StatusReply<T> where the T will be unwrapped and emitted downstream.

Actor interop operators

Dependency

The Akka dependencies are available from Akka’s library repository. To access them there, you need to configure the URL for this repository.

sbt
resolvers += "Akka library repository".at("https://repo.akka.io/maven")
Maven
<project>
  ...
  <repositories>
    <repository>
      <id>akka-repository</id>
      <name>Akka library repository</name>
      <url>https://repo.akka.io/maven</url>
    </repository>
  </repositories>
</project>
Gradle
repositories {
    mavenCentral()
    maven {
        url "https://repo.akka.io/maven"
    }
}

This operator is included in:

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

  implementation "com.typesafe.akka:akka-stream-typed_${versions.ScalaBinary}"
}

Signature

ActorFlow.askWithStatusAndContextActorFlow.askWithStatusAndContext { scala=“#askWithStatusAndContextI,Q,A,Ctx(ref:akka.actor.typed.ActorRef[Q])(makeMessage:(I,akka.actor.typed.ActorRef[akka.pattern.StatusReply[A]])=>Q)(implicittimeout:akka.util.Timeout):akka.stream.scaladsl.Flow[(I,Ctx),(A,Ctx),akka.NotUsed]” java =“#askWithStatusAndContextI,Q,A,Ctx” }

Description

Use the Ask pattern to send a request-reply message to the target ref actor when you expect the reply to be akka.pattern.StatusReply. The stream context is not sent, instead it is locally recombined to the actor’s reply.

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

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

Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.