ActorSource.actorRefWithBackpressure

Materialize an ActorRef<T>ActorRef[T] of the new actors API; 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.

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.10.0+31-e778606c-SNAPSHOT"
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.10.0+31-e778606c-SNAPSHOT</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.10.0+31-e778606c-SNAPSHOT")

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

Signature

ActorSource.actorRefWithBackpressureActorSource.actorRefWithBackpressure

Description

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

See also:

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

Example

With actorRefWithBackpressure two actors get into play:

1. An actor that is materialized when the stream runs. It feeds the stream.
2. An actor provided by the user. It gets the ack signal when an element is emitted into the stream.

For the ack signal we create an Emitted objectempty Emitted class.

For “feeding” the stream we use the Event traitinterface.

In this example we create the stream in an actor which itself reacts on the demand of the stream and sends more messages.

Scala

copysourceimport akka.actor.typed.ActorRef
import akka.stream.CompletionStrategy
import akka.stream.scaladsl.Sink
import akka.stream.typed.scaladsl.ActorSource

object StreamFeeder {

  /** Signals that the latest element is emitted into the stream */
  case object Emitted

  sealed trait Event
  case class Element(content: String) extends Event
  case object ReachedEnd extends Event
  case class FailureOccured(ex: Exception) extends Event

  def apply(): Behavior[Emitted.type] =
    Behaviors.setup { context =>
      val streamActor = runStream(context.self)(context.system)
      streamActor ! Element("first")
      sender(streamActor, 0)
    }

  private def runStream(ackReceiver: ActorRef[Emitted.type])(implicit system: ActorSystem[_]): ActorRef[Event] = {
    val source =
      ActorSource.actorRefWithBackpressure[Event, Emitted.type](
        // get demand signalled to this actor receiving Ack
        ackTo = ackReceiver,
        ackMessage = Emitted,
        // complete when we send ReachedEnd
        completionMatcher = {
          case ReachedEnd => CompletionStrategy.draining
        },
        failureMatcher = {
          case FailureOccured(ex) => ex
        })

    val streamActor: ActorRef[Event] = source
      .collect {
        case Element(msg) => msg
      }
      .to(Sink.foreach(println))
      .run()

    streamActor
  }

  private def sender(streamSource: ActorRef[Event], counter: Int): Behavior[Emitted.type] =
    Behaviors.receiveMessage {
      case Emitted if counter < 5 =>
        streamSource ! Element(counter.toString)
        sender(streamSource, counter + 1)
      case _ =>
        streamSource ! ReachedEnd
        Behaviors.stopped
    }
}

ActorSystem(StreamFeeder(), "stream-feeder")

// Will print:
// first
// 0
// 1
// 2
// 3
// 4

Java

copysourceimport akka.actor.typed.ActorRef;
import akka.actor.typed.ActorSystem;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import akka.stream.CompletionStrategy;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import akka.stream.typed.javadsl.ActorSource;
import java.util.Optional;

class StreamFeeder extends AbstractBehavior<StreamFeeder.Emitted> {
  /** Signals that the latest element is emitted into the stream */
  public enum Emitted {
    INSTANCE;
  }

  public interface Event {}

  public static class Element implements Event {
    public final String content;

    public Element(String content) {
      this.content = content;
    }

    @Override
    public String toString() {
      return "Element(" + content + ")";
    }
  }

  public enum ReachedEnd implements Event {
    INSTANCE;
  }

  public static class FailureOccured implements Event {
    public final Exception ex;

    public FailureOccured(Exception ex) {
      this.ex = ex;
    }
  }

  public static Behavior<Emitted> create() {
    return Behaviors.setup(StreamFeeder::new);
  }

  private int counter = 0;
  private final ActorRef<Event> streamSource;

  private StreamFeeder(ActorContext<Emitted> context) {
    super(context);
    streamSource = runStream(context.getSelf(), context.getSystem());
    streamSource.tell(new Element("first"));
  }

  @Override
  public Receive<Emitted> createReceive() {
    return newReceiveBuilder().onMessage(Emitted.class, this::onEmitted).build();
  }

  private static ActorRef<Event> runStream(ActorRef<Emitted> ackReceiver, ActorSystem<?> system) {
    Source<Event, ActorRef<Event>> source =
        ActorSource.actorRefWithBackpressure(
            ackReceiver,
            Emitted.INSTANCE,
            // complete when we send ReachedEnd
            (msg) -> {
              if (msg == ReachedEnd.INSTANCE) return Optional.of(CompletionStrategy.draining());
              else return Optional.empty();
            },
            (msg) -> {
              if (msg instanceof FailureOccured) return Optional.of(((FailureOccured) msg).ex);
              else return Optional.empty();
            });

    return source.to(Sink.foreach(System.out::println)).run(system);
  }

  private Behavior<Emitted> onEmitted(Emitted message) {
    if (counter < 5) {
      streamSource.tell(new Element(String.valueOf(counter)));
      counter++;
      return this;
    } else {
      streamSource.tell(ReachedEnd.INSTANCE);
      return Behaviors.stopped();
    }
  }
}
    ActorSystem<StreamFeeder.Emitted> system =
        ActorSystem.create(StreamFeeder.create(), "stream-feeder");

    // will print:
    // Element(first)
    // Element(0)
    // Element(1)
    // Element(2)
    // Element(3)
    // Element(4)

Reactive Streams semantics