Source.actorRef

Materialize an ActorRef of the classic actors API; sending messages to it will emit them on the stream.

Actor interop operators

Signature

Source.actorRefSource.actorRef

Description

Materialize an ActorRef, sending messages to it will emit them on the stream. The actor contains a buffer but since communication is one way, there is no back pressure. Handling overflow is done by either dropping elements or failing the stream; the strategy is chosen by the user.

The stream can be completed successfully by sending the actor reference a message that is matched by the completionMatcher. If the matcher returns org.apache.pekko.stream.CompletionStrategy.immediately the completion will be signaled immediately. If it returns org.apache.pekko.stream.CompletionStrategy.draining, already buffered elements will be sent out before signaling completion. The stream can be failed by sending a message that is matched by the failureMatcher; the extracted Throwable is used to fail the stream. org.apache.pekko.actor.PoisonPill and org.apache.pekko.actor.Kill messages are ignored (a warning is logged) and do not complete the stream. Using org.apache.pekko.actor.ActorSystem.stop to stop the actor and complete the stream is not supported.

See also:

• Source.actorRefWithBackpressure This operator, but with backpressure control
• ActorSource.actorRef The corresponding operator for the new actors API
• ActorSource.actorRefWithBackpressure The operator for the new actors API with backpressure control
• Source.queue Materialize a SourceQueue onto which elements can be pushed for emitting from the source

Examples

Scala

copysourceimport org.apache.pekko
import pekko.Done
import pekko.actor.ActorRef
import pekko.stream.OverflowStrategy
import pekko.stream.CompletionStrategy
import pekko.stream.scaladsl._

val source: Source[Any, ActorRef] = Source.actorRef(
  completionMatcher = {
    case Done =>
      // complete stream immediately if we send it Done
      CompletionStrategy.immediately
  },
  // never fail the stream because of a message
  failureMatcher = PartialFunction.empty,
  bufferSize = 100,
  overflowStrategy = OverflowStrategy.dropHead)
val actorRef: ActorRef = source.to(Sink.foreach(println)).run()

actorRef ! "hello"
actorRef ! "hello"

// The stream completes successfully with the following message
actorRef ! Done

Java

copysourceimport org.apache.pekko.actor.ActorRef;
import org.apache.pekko.stream.OverflowStrategy;
import org.apache.pekko.stream.CompletionStrategy;
import org.apache.pekko.stream.javadsl.Sink;
import org.apache.pekko.testkit.TestProbe;

int bufferSize = 100;
Source<Object, ActorRef> source =
    Source.actorRef(
        elem -> {
          // complete stream immediately if we send it Done
          if (elem == Done.done()) return Optional.of(CompletionStrategy.immediately());
          else return Optional.empty();
        },
        // never fail the stream because of a message
        elem -> Optional.empty(),
        bufferSize,
        OverflowStrategy.dropHead());

ActorRef actorRef = source.to(Sink.foreach(System.out::println)).run(system);
actorRef.tell("hello", ActorRef.noSender());
actorRef.tell("hello", ActorRef.noSender());

// The stream completes successfully with the following message
actorRef.tell(Done.done(), ActorRef.noSender());

Reactive Streams semantics