actorRefWithAck

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.

Source operators

def actorRefWithAck[T](ackMessage: Any): Source[T, ActorRef]

Description

Materialize an ActorRef, sending messages to it will emit them on the stream. The actor responds with the provided ack message once the element could be emitted alowing for backpressure from the source. Sending another message before the previous one has been acknowledged will fail the stream.

Examples

Scala

copysource
import akka.actor.Status.Success
import akka.actor.ActorRef
import akka.stream.CompletionStrategy
import akka.stream.scaladsl._

implicit val system: ActorSystem = ActorSystem()
implicit val materializer: ActorMaterializer = ActorMaterializer()
val probe = TestProbe()

val source: Source[Any, ActorRef] = Source.actorRefWithAck[Any]("ack")
val actorRef: ActorRef = source.to(Sink.foreach(println)).run()

probe.send(actorRef, "hello")
probe.expectMsg("ack")
probe.send(actorRef, "hello")
probe.expectMsg("ack")

// The stream completes successfully with the following message
actorRef ! Success(CompletionStrategy.immediately)

Java

copysourceimport akka.actor.ActorRef;
import akka.actor.Status.Success;
import akka.stream.OverflowStrategy;
import akka.stream.CompletionStrategy;
import akka.stream.javadsl.Sink;
import akka.testkit.TestProbe;
final ActorSystem system = ActorSystem.create();
final Materializer materializer = ActorMaterializer.create(system);

Source<Object, ActorRef> source = Source.actorRefWithAck("ack");

ActorRef actorRef = source.to(Sink.foreach(System.out::println)).run(materializer);
probe.send(actorRef, "hello");
probe.expectMsg("ack");
probe.send(actorRef, "hello");
probe.expectMsg("ack");

// The stream completes successfully with the following message
actorRef.tell(new Success(CompletionStrategy.draining()), ActorRef.noSender());