Updates

- include feedback
- more config parameters
   * idle time
   * fromSequenceNr
- user docs
- tests

Author: krasserm

akka-docs/rst/java/persistence.rst

@@ -56,6 +56,10 @@ Architecture
   processor. A view itself does not journal new messages, instead, it updates internal state only from a processor's
   replicated message stream.
 
+* *Producer*: A producer can be thought of a view that implements the `Reactive Streams`_ ``Producer`` interface. In
+  contrast to a view, a producer only reads from a processor's message stream if explicitly requested from downstream
+  consumers.
+
 * *Channel*: Channels are used by processors and views to communicate with other actors. They prevent that replayed
   messages are redundantly delivered to these actors and provide at-least-once message delivery semantics, also in
   case of sender and receiver JVM crashes.
@@ -73,6 +77,7 @@ Architecture
   development of event sourced applications (see section :ref:`event-sourcing-java`)
 
 .. _Community plugins: http://akka.io/community/
+.. _Reactive Streams: http://www.reactive-streams.org/
 
 .. _processors-java:
 
@@ -238,6 +243,13 @@ name in its actor hierarchy and hence influences only part of the view id. To fu
 The ``viewId`` must differ from the referenced ``processorId``, unless :ref:`snapshots-java` of a view and its
 processor shall be shared (which is what applications usually do not want).
 
+.. _producers-java:
+
+Producers
+=========
+
+Java API coming soon. See also Scala :ref:`producers` documentation.
+
 .. _channels-java:
 
 Channels

akka-docs/rst/scala/code/docs/persistence/PersistenceDocSpec.scala

@@ -21,7 +21,7 @@ trait PersistenceDocSpec {
       //#auto-update
     """
 
-  val system: ActorSystem
+  implicit val system: ActorSystem
 
   import system._
 
@@ -355,4 +355,41 @@ trait PersistenceDocSpec {
     view ! Update(await = true)
     //#view-update
   }
+
+  new AnyRef {
+    //#producer-creation
+    import org.reactivestreams.api.Producer
+
+    import akka.persistence.{ Persistent, PersistentFlow, ProducerSettings }
+    import akka.stream.{ FlowMaterializer, MaterializerSettings }
+    import akka.stream.scaladsl.Flow
+
+    val materializer = FlowMaterializer(MaterializerSettings())
+
+    val flow: Flow[Persistent] = PersistentFlow.fromProcessor("some-processor-id")
+    val producer: Producer[Persistent] = flow.toProducer(materializer)
+    //#producer-creation
+
+    //#producer-buffer-size
+    PersistentFlow.fromProcessor("some-processor-id", ProducerSettings(maxBufferSize = 200))
+    //#producer-buffer-size
+
+    //#producer-examples
+    // 1 producer and 2 consumers:
+    val producer1: Producer[Persistent] =
+      PersistentFlow.fromProcessor("processor-1").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 producers (merged) and 1 consumer:
+    val producer2: Producer[Persistent] =
+      PersistentFlow.fromProcessor("processor-2").toProducer(materializer)
+    val producer3: Producer[Persistent] =
+      PersistentFlow.fromProcessor("processor-3").toProducer(materializer)
+    Flow(producer2).merge(producer3).foreach { p =>
+      println(s"consumer-3: ${p.payload}")
+    }.consume(materializer)
+    //#producer-examples
+  }
+
 }

akka-docs/rst/scala/persistence.rst

@@ -44,6 +44,10 @@ Architecture
   processor. A view itself does not journal new messages, instead, it updates internal state only from a processor's
   replicated message stream.
 
+* *Producer*: A producer can be thought of a view that implements the `Reactive Streams`_ ``Producer`` interface. In
+  contrast to a view, a producer only reads from a processor's message stream if explicitly requested from downstream
+  consumers.
+
 * *Channel*: Channels are used by processors and views to communicate with other actors. They prevent that replayed
   messages are redundantly delivered to these actors and provide at-least-once message delivery semantics, also in
   case of sender and receiver JVM crashes.
@@ -61,6 +65,7 @@ Architecture
   development of event sourced applications (see section :ref:`event-sourcing`)
 
 .. _Community plugins: http://akka.io/community/
+.. _Reactive Streams: http://www.reactive-streams.org/
 
 .. _processors:
 
@@ -228,6 +233,42 @@ name in its actor hierarchy and hence influences only part of the view id. To fu
 The ``viewId`` must differ from the referenced ``processorId``, unless :ref:`snapshots` of a view and its
 processor shall be shared (which is what applications usually do not want).
 
+.. _producers:
+
+Producers
+=========
+
+A `Reactive Streams`_ ``Producer`` can be created from a processor's message stream via the ``PersistentFlow``
+extension of the Akka Streams Scala DSL:
+
+.. includecode:: code/docs/persistence/PersistenceDocSpec.scala#producer-creation
+
+The created ``flow`` object is of type ``Flow[Persistent]`` and can be composed with other flows using ``Flow``
+combinators (= methods defined on ``Flow``). Calling the ``toProducer`` method on ``flow`` creates a producer
+of type ``Producer[Persistent]``.
+
+A persistent message producer only reads from a processor's journal when explicitly requested by downstream
+consumers. In order to avoid frequent, fine grained read access to a processor's journal, the producer tries
+to buffer persistent messages in memory from which it serves downstream requests. The maximum buffer size per
+producer is configurable with a ``ProducerSettings`` configuration object.
+
+.. includecode:: code/docs/persistence/PersistenceDocSpec.scala#producer-buffer-size
+
+Other ``ProducerSettings`` parameters are:
+
+* ``fromSequenceNr``: specifies from which sequence number the persistent message stream shall start (defaults
+  to ``1L``). Please note that specifying ``fromSequenceNr`` is much more efficient than using the ``drop(Int)``
+  combinator, especially for larger sequence numbers.
+
+* ``idle``: an optional parameter that specifies how long a producer shall wait after a journal read attempt didn't return
+  any new persistent messages. If not defined, the producer uses the ``akka.persistence.view.auto-update-interval``
+  configuration parameter, otherwise, it uses the defined ``idle`` parameter.
+
+Here are two examples how persistent message producers can be connected to downstream consumers using the Akka
+Streams Scala DSL and its ``PersistentFlow`` extension.
+
+.. includecode:: code/docs/persistence/PersistenceDocSpec.scala#producer-examples
+
 .. _channels:
 
 Channels

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[persistence] def onReplayComplete(await: Boolean): Unit = {
+  private 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

@@ -1,3 +1,6 @@
+/**
+ * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
+ */
 package akka.persistence
 
 import scala.util.control.NonFatal
@@ -33,21 +36,30 @@ object PersistentFlow {
 /**
  * Configuration object for a [[Persistent]] stream producer.
  *
- * @param maxBufferSize maximum number of persistent messages to be buffered in memory (per producer).
+ * @param fromSequenceNr Sequence number where the produced stream shall start (inclusive).
+ *                       Default is `1L`.
+ * @param maxBufferSize Maximum number of persistent messages to be buffered in memory (per producer).
+ *                      Default is `100`.
+ * @param idle Optional duration to wait if no more persistent messages can be pulled from the journal
+ *             before attempting the next pull. Default is `None` which causes the producer to take
+ *             the value defined by the `akka.persistence.view.auto-update-interval` configuration
+ *             key. If defined, the `idle` value is taken directly.
  */
-case class ProducerSettings(maxBufferSize: Int = 20)
+case class ProducerSettings(fromSequenceNr: Long = 1L, maxBufferSize: Int = 100, idle: Option[FiniteDuration] = None) {
+  require(fromSequenceNr > 0L, "fromSequenceNr must be > 0")
+}
 
-private[akka] object ViewProducer {
+private 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] {
+private 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)
+private class ViewProducerImpl(processorId: String, producerSettings: ProducerSettings, materializerSettings: MaterializerSettings)
   extends Actor
   with ActorLogging
   with SubscriberManagement[Persistent]
@@ -58,13 +70,11 @@ private[akka] class ViewProducerImpl(processorId: String, producerSettings: Prod
 
   type S = ActorSubscription[Persistent]
 
-  private val view = context.actorOf(Props(classOf[ViewBuffer], processorId, producerSettings.maxBufferSize, self))
+  private val view = context.actorOf(Props(classOf[ViewBuffer], processorId, producerSettings, self), "view-buffer")
 
   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]]
@@ -74,7 +84,6 @@ private[akka] class ViewProducerImpl(processorId: String, producerSettings: Prod
   final def waitingForSubscribers: Receive = {
     case SubscribePending ⇒
       pub.takePendingSubscribers() foreach registerSubscriber
-      context.setReceiveTimeout(Duration.Undefined)
       context.become(active)
   }
 
@@ -129,22 +138,22 @@ private object ViewBuffer {
 }
 
 /**
- * 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`).
+ * A view that buffers up to `producerSettings.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 `producerSettings.maxBufferSize`).
  */
-private[akka] class ViewBuffer(val processorId: String, maxBufferSize: Int, producer: ActorRef) extends View {
+private class ViewBuffer(val processorId: String, producerSettings: ProducerSettings, 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 = {
+  private val filling: Receive = {
     case p: Persistent ⇒
-      buffer = buffer :+ p
+      buffer :+= p
       replayed += 1
       if (requested > 0) respond(requested)
     case Filled ⇒
@@ -158,14 +167,14 @@ private[akka] class ViewBuffer(val processorId: String, maxBufferSize: Int, prod
       if (buffer.nonEmpty) respond(requested)
   }
 
-  val pausing: Receive = {
+  private val pausing: Receive = {
     case Request(num) ⇒
       requested += num
       respond(requested)
       if (buffer.isEmpty) fill()
   }
 
-  val scheduled: Receive = {
+  private val scheduled: Receive = {
     case Fill ⇒
       fill()
     case Request(num) ⇒
@@ -174,33 +183,47 @@ private[akka] class ViewBuffer(val processorId: String, maxBufferSize: Int, prod
 
   def receive = filling
 
-  def fill(): Unit = {
+  override def onReplaySuccess(receive: Receive, await: Boolean): Unit = {
+    super.onReplaySuccess(receive, await)
+    self ! Filled
+  }
+
+  override def onReplayFailure(receive: Receive, await: Boolean, cause: Throwable): Unit = {
+    super.onReplayFailure(receive, await, cause)
+    self ! Filled
+  }
+
+  override def lastSequenceNr: Long =
+    math.max(producerSettings.fromSequenceNr - 1L, super.lastSequenceNr)
+
+  override def autoUpdateInterval: FiniteDuration =
+    producerSettings.idle.getOrElse(super.autoUpdateInterval)
+
+  override def autoUpdateReplayMax: Long =
+    producerSettings.maxBufferSize
+
+  override def autoUpdate: Boolean =
+    false
+
+  private def fill(): Unit = {
     replayed = 0
     context.become(filling)
-    self ! Update(false, maxBufferSize)
+    self ! Update(false, autoUpdateReplayMax)
   }
 
-  def pause(): Unit = {
+  private def pause(): Unit = {
     context.become(pausing)
   }
 
-  def schedule(): Unit = {
+  private def schedule(): Unit = {
     context.become(scheduled)
     context.system.scheduler.scheduleOnce(autoUpdateInterval, self, Fill)
   }
 
-  def respond(num: Int): Unit = {
+  private 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
 }