+per akka#15035 Reactive-stream producers for persistent messages written by akka-persistence processors

Author: krasserm

akka-persistence/src/main/scala/akka/persistence/View.scala

@@ -126,7 +126,7 @@ trait View extends Actor with Recovery {
   /**
    * Switches to `idle` state and schedules the next update if `autoUpdate` returns `true`.
    */
-  private def onReplayComplete(await: Boolean): Unit = {
+  private[persistence] def onReplayComplete(await: Boolean): Unit = {
     _currentState = idle
     if (autoUpdate) schedule = Some(context.system.scheduler.scheduleOnce(autoUpdateInterval, self, Update(await = false, autoUpdateReplayMax)))
     if (await) receiverStash.unstashAll()

akka-persistence/src/main/scala/akka/persistence/ViewProducer.scala

@@ -0,0 +1,206 @@
+package akka.persistence
+
+import scala.util.control.NonFatal
+import scala.concurrent.duration._
+
+import org.reactivestreams.api.Producer
+import org.reactivestreams.spi.Subscriber
+
+import akka.actor._
+import akka.stream._
+import akka.stream.impl._
+import akka.stream.impl.Ast.ProducerNode
+import akka.stream.scaladsl.Flow
+
+object PersistentFlow {
+  /**
+   * Starts a new [[Persistent]] message flow from the given processor,
+   * identified by `processorId`. Elements are pulled from the processor's
+   * journal (using a [[View]]) in accordance with the demand coming from
+   * the downstream transformation steps.
+   *
+   * Elements pulled from the processor's journal are buffered in memory so that
+   * fine-grained demands (requests) from downstream can be served efficiently.
+   * Reads from the journal are done in (coarse-grained) batches of configurable
+   * size (which correspond to the configurable maximum buffer size).
+   *
+   * @see [[ProducerSettings]]
+   */
+  def fromProcessor(processorId: String, producerSettings: ProducerSettings = ProducerSettings()): Flow[Persistent] =
+    FlowImpl(ViewProducerNode(processorId, producerSettings), Nil)
+}
+
+/**
+ * Configuration object for a [[Persistent]] stream producer.
+ *
+ * @param maxBufferSize maximum number of persistent messages to be buffered in memory (per producer).
+ */
+case class ProducerSettings(maxBufferSize: Int = 20)
+
+private[akka] object ViewProducer {
+  def props(processorId: String, producerSettings: ProducerSettings, settings: MaterializerSettings): Props =
+    Props(classOf[ViewProducerImpl], processorId, producerSettings, settings)
+}
+
+private[akka] case class ViewProducerNode(processorId: String, producerSettings: ProducerSettings) extends ProducerNode[Persistent] {
+  def createProducer(settings: MaterializerSettings, context: ActorRefFactory): Producer[Persistent] =
+    new ActorProducer(context.actorOf(ViewProducer.props(processorId, producerSettings, settings)))
+}
+
+private[akka] class ViewProducerImpl(processorId: String, producerSettings: ProducerSettings, materializerSettings: MaterializerSettings)
+  extends Actor
+  with ActorLogging
+  with SubscriberManagement[Persistent]
+  with SoftShutdown {
+
+  import ActorBasedFlowMaterializer._
+  import ViewBuffer._
+
+  type S = ActorSubscription[Persistent]
+
+  private val view = context.actorOf(Props(classOf[ViewBuffer], processorId, producerSettings.maxBufferSize, self))
+
+  private var pub: ActorPublisher[Persistent] = _
+  private var shutdownReason: Option[Throwable] = ActorPublisher.NormalShutdownReason
+
+  context.setReceiveTimeout(materializerSettings.downstreamSubscriptionTimeout)
+
+  final def receive = {
+    case ExposedPublisher(pub) ⇒
+      this.pub = pub.asInstanceOf[ActorPublisher[Persistent]]
+      context.become(waitingForSubscribers)
+  }
+
+  final def waitingForSubscribers: Receive = {
+    case SubscribePending ⇒
+      pub.takePendingSubscribers() foreach registerSubscriber
+      context.setReceiveTimeout(Duration.Undefined)
+      context.become(active)
+  }
+
+  final def active: Receive = {
+    case SubscribePending ⇒
+      pub.takePendingSubscribers() foreach registerSubscriber
+    case RequestMore(sub, elements) ⇒
+      moreRequested(sub.asInstanceOf[S], elements)
+    case Cancel(sub) ⇒
+      unregisterSubscription(sub.asInstanceOf[S])
+    case r @ Response(ps) ⇒
+      try {
+        ps.foreach(p ⇒ pushToDownstream(withCtx(context)(p)))
+      } catch {
+        case Stop        ⇒ { completeDownstream(); shutdownReason = None }
+        case NonFatal(e) ⇒ { abortDownstream(e); shutdownReason = Some(e) }
+      }
+  }
+
+  override def requestFromUpstream(elements: Int): Unit =
+    view ! Request(elements)
+
+  override def initialBufferSize =
+    materializerSettings.initialFanOutBufferSize
+
+  override def maxBufferSize =
+    materializerSettings.maxFanOutBufferSize
+
+  override def createSubscription(subscriber: Subscriber[Persistent]): ActorSubscription[Persistent] =
+    new ActorSubscription(self, subscriber)
+
+  override def cancelUpstream(): Unit = {
+    pub.shutdown(shutdownReason)
+    softShutdown()
+  }
+  override def shutdown(completed: Boolean): Unit = {
+    pub.shutdown(shutdownReason)
+    softShutdown()
+  }
+
+  override def postStop(): Unit = {
+    pub.shutdown(shutdownReason)
+  }
+}
+
+private object ViewBuffer {
+  case class Request(num: Int)
+  case class Response(messages: Vector[Persistent])
+
+  case object Fill
+  case object Filled
+}
+
+/**
+ * A view that buffers up to `maxBufferSize` persistent messages in memory. Downstream demands
+ * (requests) are served if the buffer is non-empty either while filling the buffer or after
+ * having filled the buffer. When the buffer becomes empty new persistent messages are loaded
+ * from the journal (in batches up to `maxBufferSize`).
+ */
+private[akka] class ViewBuffer(val processorId: String, maxBufferSize: Int, producer: ActorRef) extends View {
+  import ViewBuffer._
+  import context.dispatcher
+
+  private var replayed = 0
+  private var requested = 0
+  private var buffer: Vector[Persistent] = Vector.empty
+
+  val filling: Receive = {
+    case p: Persistent ⇒
+      buffer = buffer :+ p
+      replayed += 1
+      if (requested > 0) respond(requested)
+    case Filled ⇒
+      if (buffer.nonEmpty && requested > 0) respond(requested)
+      if (buffer.nonEmpty) pause()
+      else if (requested == 0) pause()
+      else if (replayed > 0) fill()
+      else schedule()
+    case Request(num) ⇒
+      requested += num
+      if (buffer.nonEmpty) respond(requested)
+  }
+
+  val pausing: Receive = {
+    case Request(num) ⇒
+      requested += num
+      respond(requested)
+      if (buffer.isEmpty) fill()
+  }
+
+  val scheduled: Receive = {
+    case Fill ⇒
+      fill()
+    case Request(num) ⇒
+      requested += num
+  }
+
+  def receive = filling
+
+  def fill(): Unit = {
+    replayed = 0
+    context.become(filling)
+    self ! Update(false, maxBufferSize)
+  }
+
+  def pause(): Unit = {
+    context.become(pausing)
+  }
+
+  def schedule(): Unit = {
+    context.become(scheduled)
+    context.system.scheduler.scheduleOnce(autoUpdateInterval, self, Fill)
+  }
+
+  def respond(num: Int): Unit = {
+    val (res, buf) = buffer.splitAt(num)
+    producer ! Response(res)
+    buffer = buf
+    requested -= res.size
+  }
+
+  private[persistence] override def onReplayComplete(await: Boolean): Unit = {
+    super.onReplayComplete(await)
+    self ! Filled
+  }
+
+  override def autoUpdateReplayMax: Long = maxBufferSize
+  override def autoUpdate: Boolean = false
+}

akka-samples/akka-sample-persistence-scala/src/main/scala/sample/persistence/StreamExample.scala

@@ -0,0 +1,48 @@
+package sample.persistence
+
+import org.reactivestreams.api._
+
+import akka.actor._
+import akka.persistence.{ Persistent, PersistentFlow, Processor }
+import akka.stream._
+import akka.stream.scaladsl._
+
+/**
+ * This example demonstrates how akka-persistence Views can be used as reactive-stream Producers. A
+ * View-based Producer is created with PersistentFlow.fromProcessor(processorId: String).toProducer().
+ * This Producer produces Persistent messages as they are written by its corresponding akka-persistence
+ * Processor. The PersistentFlow object is an extension to the akka-stream DSL.
+ */
+object StreamExample extends App {
+  implicit val system = ActorSystem("example")
+
+  class ExampleProcessor(pid: String) extends Processor {
+    override def processorId = pid
+    def receive = {
+      case Persistent(payload, _) =>
+    }
+  }
+
+  val p1 = system.actorOf(Props(classOf[ExampleProcessor], "p1"))
+  val p2 = system.actorOf(Props(classOf[ExampleProcessor], "p2"))
+
+  val materializer = FlowMaterializer(MaterializerSettings())
+
+  // 1 view-backed producer and 2 consumers:
+  val producer1: Producer[Persistent] = PersistentFlow.fromProcessor("p1").toProducer(materializer)
+  Flow(producer1).foreach { p => println(s"consumer-1: ${p.payload}") }.consume(materializer)
+  Flow(producer1).foreach { p => println(s"consumer-2: ${p.payload}") }.consume(materializer)
+
+  // 2 view-backed producers (merged) and 1 consumer:
+  // This is an example how message/event streams from multiple processors can be merged into a single stream.
+  val producer2: Producer[Persistent] = PersistentFlow.fromProcessor("p1").toProducer(materializer)
+  val merged: Producer[Persistent] = PersistentFlow.fromProcessor("p2").merge(producer2).toProducer(materializer)
+  Flow(merged).foreach { p => println(s"consumer-3: ${p.payload}") }.consume(materializer)
+
+  while (true) {
+    p1 ! Persistent("a-" + System.currentTimeMillis())
+    p2 ! Persistent("b-" + System.currentTimeMillis())
+    Thread.sleep(500)
+  }
+}
+

project/AkkaBuild.scala

@@ -284,7 +284,7 @@ object AkkaBuild extends Build {
   lazy val persistence = Project(
     id = "akka-persistence-experimental",
     base = file("akka-persistence"),
-    dependencies = Seq(actor, remote % "test->test", testkit % "test->test"),
+    dependencies = Seq(actor, stream, remote % "test->test", testkit % "test->test"),
     settings = defaultSettings ++ formatSettings ++ scaladocSettings ++ experimentalSettings ++ javadocSettings ++ OSGi.persistence ++ Seq(
       fork in Test := true,
       javaOptions in Test := defaultMultiJvmOptions,