!per #3704 Persistence improvements

- Channel enhancements (#3773):
- Live read models (#3776):
- Batch-oriented journal plugin API (#3804):
- Batching of confirmations and deletions
- Message deletion enhancements (more efficient range deletions)

akka-persistence/src/main/scala/akka/persistence/journal/AsyncWriteJournal.scala

@@ -12,69 +12,73 @@ import scala.util._
 import akka.actor._
 import akka.pattern.pipe
 import akka.persistence._
-import akka.persistence.JournalProtocol._
 
 /**
  * Abstract journal, optimized for asynchronous, non-blocking writes.
  */
-trait AsyncWriteJournal extends Actor with AsyncReplay {
+trait AsyncWriteJournal extends Actor with AsyncRecovery {
+  import JournalProtocol._
   import AsyncWriteJournal._
   import context.dispatcher
 
   private val extension = Persistence(context.system)
+  private val publish = extension.settings.internal.publishPluginCommands
 
   private val resequencer = context.actorOf(Props[Resequencer])
   private var resequencerCounter = 1L
 
   def receive = {
-    case WriteBatch(persistentBatch, processor) ⇒
+    case WriteMessages(persistentBatch, processor) ⇒
       val cctr = resequencerCounter
       def resequence(f: PersistentRepr ⇒ Any) = persistentBatch.zipWithIndex.foreach {
         case (p, i) ⇒ resequencer ! Desequenced(f(p), cctr + i + 1, processor, p.sender)
       }
-      writeAsync(persistentBatch.map(_.prepareWrite())) onComplete {
+      asyncWriteMessages(persistentBatch.map(_.prepareWrite())) onComplete {
         case Success(_) ⇒
-          resequencer ! Desequenced(WriteBatchSuccess, cctr, processor, self)
-          resequence(WriteSuccess(_))
+          resequencer ! Desequenced(WriteMessagesSuccess, cctr, processor, self)
+          resequence(WriteMessageSuccess(_))
         case Failure(e) ⇒
-          resequencer ! Desequenced(WriteBatchFailure(e), cctr, processor, self)
-          resequence(WriteFailure(_, e))
+          resequencer ! Desequenced(WriteMessagesFailure(e), cctr, processor, self)
+          resequence(WriteMessageFailure(_, e))
       }
       resequencerCounter += persistentBatch.length + 1
-    case Replay(fromSequenceNr, toSequenceNr, processorId, processor) ⇒
+    case ReplayMessages(fromSequenceNr, toSequenceNr, max, processorId, processor, replayDeleted) ⇒
       // Send replayed messages and replay result to processor directly. No need
       // to resequence replayed messages relative to written and looped messages.
-      replayAsync(processorId, fromSequenceNr, toSequenceNr) { p ⇒
-        if (!p.deleted) processor.tell(Replayed(p), p.sender)
+      asyncReplayMessages(processorId, fromSequenceNr, toSequenceNr, max) { p ⇒
+        if (!p.deleted || replayDeleted) processor.tell(ReplayedMessage(p), p.sender)
       } map {
-        maxSnr ⇒ ReplaySuccess(maxSnr)
+        case _ ⇒ ReplayMessagesSuccess
       } recover {
-        case e ⇒ ReplayFailure(e)
+        case e ⇒ ReplayMessagesFailure(e)
       } pipeTo (processor)
-    case c @ Confirm(processorId, messageSequenceNr, channelId, wrapperSequenceNr, channelEndpoint) ⇒
-      val op = if (wrapperSequenceNr == 0L) {
-        // A wrapperSequenceNr == 0L means that the corresponding message was delivered by a
-        // transient channel. We can now write a delivery confirmation for this message.
-        confirmAsync(processorId, messageSequenceNr, channelId)
-      } else {
-        // A wrapperSequenceNr != 0L means that the corresponding message was delivered by a
-        // persistent channel. We can now safely delete the wrapper message (that contains the
-        // delivered message).
-        deleteAsync(channelId, wrapperSequenceNr, wrapperSequenceNr, true)
+    case ReadHighestSequenceNr(fromSequenceNr, processorId, processor) ⇒
+      // Send read highest sequence number to processor directly. No need
+      // to resequence the result relative to written and looped messages.
+      asyncReadHighestSequenceNr(processorId, fromSequenceNr).map {
+        highest ⇒ ReadHighestSequenceNrSuccess(highest)
+      } recover {
+        case e ⇒ ReadHighestSequenceNrFailure(e)
+      } pipeTo (processor)
+    case c @ WriteConfirmations(confirmationsBatch, requestor) ⇒
+      asyncWriteConfirmations(confirmationsBatch) onComplete {
+        case Success(_) ⇒ requestor ! WriteConfirmationsSuccess(confirmationsBatch)
+        case Failure(e) ⇒ requestor ! WriteConfirmationsFailure(e)
       }
-      op onComplete {
+    case d @ DeleteMessages(messageIds, permanent, requestorOption) ⇒
+      asyncDeleteMessages(messageIds, permanent) onComplete {
         case Success(_) ⇒
-          if (extension.publishPluginCommands) context.system.eventStream.publish(c)
-          if (channelEndpoint != null) channelEndpoint ! c
-        case Failure(e) ⇒ // TODO: publish failure to event stream
+          requestorOption.foreach(_ ! DeleteMessagesSuccess(messageIds))
+          if (publish) context.system.eventStream.publish(d)
+        case Failure(e) ⇒
       }
-    case d @ Delete(processorId, fromSequenceNr, toSequenceNr, permanent) ⇒
-      deleteAsync(processorId, fromSequenceNr, toSequenceNr, permanent) onComplete {
-        case Success(_) ⇒ if (extension.publishPluginCommands) context.system.eventStream.publish(d)
-        case Failure(e) ⇒ // TODO: publish failure to event stream
+    case d @ DeleteMessagesTo(processorId, toSequenceNr, permanent) ⇒
+      asyncDeleteMessagesTo(processorId, toSequenceNr, permanent) onComplete {
+        case Success(_) ⇒ if (publish) context.system.eventStream.publish(d)
+        case Failure(e) ⇒
       }
-    case Loop(message, processor) ⇒
-      resequencer ! Desequenced(LoopSuccess(message), resequencerCounter, processor, sender)
+    case LoopMessage(message, processor) ⇒
+      resequencer ! Desequenced(LoopMessageSuccess(message), resequencerCounter, processor, sender)
       resequencerCounter += 1
   }
 
@@ -84,22 +88,26 @@ trait AsyncWriteJournal extends Actor with AsyncReplay {
    * The batch write must be atomic i.e. either all persistent messages in the batch
    * are written or none.
    */
-  def writeAsync(persistentBatch: immutable.Seq[PersistentRepr]): Future[Unit]
+  def asyncWriteMessages(messages: immutable.Seq[PersistentRepr]): Future[Unit]
 
   /**
-   * Plugin API: asynchronously deletes all persistent messages within the range from
-   * `fromSequenceNr` to `toSequenceNr` (both inclusive). If `permanent` is set to
-   * `false`, the persistent messages are marked as deleted, otherwise they are
-   * permanently deleted.
-   *
-   * @see [[AsyncReplay]]
+   * Plugin API: asynchronously writes a batch of delivery confirmations to the journal.
    */
-  def deleteAsync(processorId: String, fromSequenceNr: Long, toSequenceNr: Long, permanent: Boolean): Future[Unit]
+  def asyncWriteConfirmations(confirmations: immutable.Seq[PersistentConfirmation]): Future[Unit]
 
   /**
-   * Plugin API: asynchronously writes a delivery confirmation to the journal.
+   * Plugin API: asynchronously deletes messages identified by `messageIds` from the
+   * journal. If `permanent` is set to `false`, the persistent messages are marked as
+   * deleted, otherwise they are permanently deleted.
    */
-  def confirmAsync(processorId: String, sequenceNr: Long, channelId: String): Future[Unit]
+  def asyncDeleteMessages(messageIds: immutable.Seq[PersistentId], permanent: Boolean): Future[Unit]
+
+  /**
+   * Plugin API: asynchronously deletes all persistent messages up to `toSequenceNr`
+   * (inclusive). If `permanent` is set to `false`, the persistent messages are marked
+   * as deleted, otherwise they are permanently deleted.
+   */
+  def asyncDeleteMessagesTo(processorId: String, toSequenceNr: Long, permanent: Boolean): Future[Unit]
   //#journal-plugin-api
 }