akka-persistence prototype

The most prominent changes compared to eventsourced are:

- No central processor and channel registry any more
- Auto-recovery of processors on start and restart (can be disabled)
- Recovery of processor networks doesn't require coordination
- Explicit channel activation not needed any more
- Message sequence numbers generated per processor (no gaps)
- Sender references are journaled along with messages
- Processors can determine their recovery status
- No custom API on extension object, only messages
- Journal created by extension from config, not by application
- Applications only interact with processors and channels via messages
- Internal design prepared for having processor-specific journal actors (for later optimization possibilities)

Further additions and changes during review:

- Allow processor implementation classes to use inherited stash
- Channel support to resolve (potentially invalid) sender references
- Logical intead of physical deletion of messages
- Pinned dispatcher for LevelDB journal
- Processor can handle failures during recovery
- Message renamed to Persistent

This prototype has the following limitations:

- Serialization of persistent messages and their payload via JavaSerializer only (will be configurable later)
- The LevelDB journal implementation based on a LevelDB Java port, not the native LevelDB (will be configurable later)

The following features will be added later using separate tickets:

- Snapshot-based recovery
- Reliable channels
- Journal plugin API
- Optimizations
- ...

akka-actor/src/main/scala/akka/actor/Actor.scala

@@ -469,6 +469,16 @@ trait Actor {
   def receive: Actor.Receive
   //#receive
 
+  /**
+   * INTERNAL API.
+   *
+   * Can be overridden to intercept calls to this actor's current behavior.
+   *
+   * @param receive current behavior.
+   * @param msg current message.
+   */
+  protected[akka] def aroundReceive(receive: Actor.Receive, msg: Any): Unit = receive.applyOrElse(msg, unhandled)
+
   /**
    * User overridable definition the strategy to use for supervising
    * child actors.

akka-actor/src/main/scala/akka/actor/ActorCell.scala

@@ -495,7 +495,7 @@ private[akka] class ActorCell(
     }
   }
 
-  final def receiveMessage(msg: Any): Unit = behaviorStack.head.applyOrElse(msg, actor.unhandled)
+  final def receiveMessage(msg: Any): Unit = actor.aroundReceive(behaviorStack.head, msg)
 
   /*
    * ACTOR CONTEXT IMPLEMENTATION

akka-actor/src/main/scala/akka/actor/Stash.scala

@@ -77,10 +77,13 @@ trait UnrestrictedStash extends Actor {
     config.getInt("stash-capacity")
   }
 
-  /* The actor's deque-based message queue.
+  /**
+   * INTERNAL API.
+   *
+   * The actor's deque-based message queue.
    * `mailbox.queue` is the underlying `Deque`.
    */
-  private val mailbox: DequeBasedMessageQueueSemantics = {
+  protected[akka] val mailbox: DequeBasedMessageQueueSemantics = {
     context.asInstanceOf[ActorCell].mailbox.messageQueue match {
       case queue: DequeBasedMessageQueueSemantics ⇒ queue
       case other ⇒ throw ActorInitializationException(self, s"DequeBasedMailbox required, got: ${other.getClass.getName}\n" +
@@ -116,9 +119,26 @@ trait UnrestrictedStash extends Actor {
    *
    *  The stash is guaranteed to be empty after calling `unstashAll()`.
    */
-  def unstashAll(): Unit = {
+  def unstashAll(): Unit = unstashAll(_ ⇒ true)
+
+  /**
+   * INTERNAL API.
+   *
+   *  Prepends selected messages in the stash, applying `filterPredicate`,  to the
+   *  mailbox, and then clears the stash.
+   *
+   *  Messages from the stash are enqueued to the mailbox until the capacity of the
+   *  mailbox (if any) has been reached. In case a bounded mailbox overflows, a
+   *  `MessageQueueAppendFailedException` is thrown.
+   *
+   *  The stash is guaranteed to be empty after calling `unstashAll(Any => Boolean)`.
+   *
+   *  @param filterPredicate only stashed messages selected by this predicate are
+   *                         prepended to the mailbox.
+   */
+  protected[akka] def unstashAll(filterPredicate: Any ⇒ Boolean): Unit = {
     try {
-      val i = theStash.reverseIterator
+      val i = theStash.reverseIterator.filter(envelope ⇒ filterPredicate(envelope.message))
       while (i.hasNext) mailbox.enqueueFirst(self, i.next())
     } finally {
       theStash = Vector.empty[Envelope]

akka-docs/rst/java/code/docs/persistence/PersistenceDocTest.java

@@ -0,0 +1,172 @@
+package docs.persistence;
+
+import scala.Option;
+
+import akka.actor.*;
+import akka.persistence.*;
+
+public class PersistenceDocTest {
+
+    public interface ProcessorMethods {
+        //#processor-id
+        public String processorId();
+        //#processor-id
+        //#recovery-status
+        public boolean recoveryRunning();
+        public boolean recoveryFinished();
+        //#recovery-status
+        //#current-message
+        public Persistent getCurrentPersistentMessage();
+        //#current-message
+    }
+
+    static Object o1 = new Object() {
+        //#definition
+        class MyProcessor extends UntypedProcessor {
+            public void onReceive(Object message) throws Exception {
+                if (message instanceof Persistent) {
+                    // message has been written to journal
+                    Persistent persistent = (Persistent)message;
+                    Object payload = persistent.payload();
+                    Long sequenceNr = persistent.sequenceNr();
+                    // ...
+                } else {
+                    // message has not been written to journal
+                }
+            }
+        }
+        //#definition
+
+        class MyActor extends UntypedActor {
+            ActorRef processor;
+
+            public MyActor() {
+                //#usage
+                processor = getContext().actorOf(Props.create(MyProcessor.class), "myProcessor");
+
+                processor.tell(Persistent.create("foo"), null);
+                processor.tell("bar", null);
+                //#usage
+            }
+
+            public void onReceive(Object message) throws Exception {
+                // ...
+            }
+
+            private void recover() {
+                //#recover-explicit
+                processor.tell(Recover.create(), null);
+                //#recover-explicit
+            }
+        }
+    };
+
+    static Object o2 = new Object() {
+        abstract class MyProcessor1 extends UntypedProcessor {
+            //#recover-on-start-disabled
+            @Override
+            public void preStartProcessor() {}
+            //#recover-on-start-disabled
+
+            //#recover-on-restart-disabled
+            @Override
+            public void preRestartProcessor(Throwable reason, Option<Object> message) {}
+            //#recover-on-restart-disabled
+        }
+
+        abstract class MyProcessor2 extends UntypedProcessor {
+            //#recover-on-start-custom
+            @Override
+            public void preStartProcessor() {
+                getSelf().tell(Recover.create(457L), null);
+            }
+            //#recover-on-start-custom
+        }
+
+        abstract class MyProcessor3 extends UntypedProcessor {
+            //#deletion
+            @Override
+            public void preRestartProcessor(Throwable reason, Option<Object> message) throws Exception {
+                if (message.isDefined() && message.get() instanceof Persistent) {
+                    delete((Persistent) message.get());
+                }
+                super.preRestartProcessor(reason, message);
+            }
+            //#deletion
+        }
+
+        class MyProcessor4 extends UntypedProcessor implements ProcessorMethods {
+            //#processor-id-override
+            @Override
+            public String processorId() {
+                return "my-stable-processor-id";
+            }
+            //#processor-id-override
+            @Override
+            public void onReceive(Object message) throws Exception {}
+        }
+    };
+
+    static Object o3 = new Object() {
+        //#channel-example
+        class MyProcessor extends UntypedProcessor {
+            private final ActorRef destination;
+            private final ActorRef channel;
+
+            public MyProcessor() {
+                this.destination = getContext().actorOf(Props.create(MyDestination.class));
+                this.channel = getContext().actorOf(Channel.props(), "myChannel");
+            }
+
+            public void onReceive(Object message) throws Exception {
+                if (message instanceof Persistent) {
+                    Persistent p = (Persistent)message;
+                    Persistent out = p.withPayload("done " + p.payload());
+                    channel.tell(Deliver.create(out, destination), getSelf());
+                }
+            }
+        }
+
+        class MyDestination extends UntypedActor {
+            public void onReceive(Object message) throws Exception {
+                if (message instanceof Persistent) {
+                    Persistent p = (Persistent)message;
+                    System.out.println("received " + p.payload());
+                    p.confirm();
+                }
+            }
+        }
+        //#channel-example
+
+        class MyProcessor2 extends UntypedProcessor {
+            private final ActorRef destination;
+            private final ActorRef channel;
+
+            public MyProcessor2(ActorRef destination) {
+                this.destination = getContext().actorOf(Props.create(MyDestination.class));
+                //#channel-id-override
+                this.channel = getContext().actorOf(Channel.props("my-stable-channel-id"));
+                //#channel-id-override
+            }
+
+            public void onReceive(Object message) throws Exception {
+                if (message instanceof Persistent) {
+                    Persistent p = (Persistent)message;
+                    Persistent out = p.withPayload("done " + p.payload());
+                    channel.tell(Deliver.create(out, destination), getSelf());
+
+                    //#channel-example-reply
+                    channel.tell(Deliver.create(out, getSender()), getSelf());
+                    //#channel-example-reply
+                    //#resolve-destination
+                    channel.tell(Deliver.create(out, getSender(), Resolve.destination()), getSelf());
+                    //#resolve-destination
+                    //#resolve-sender
+                    channel.tell(Deliver.create(out, destination, Resolve.sender()), getSender());
+                    //#resolve-sender
+
+                }
+            }
+        }
+    };
+}

akka-docs/rst/java/index-actors.rst

@@ -11,4 +11,5 @@ Actors
    mailboxes
    routing
    fsm
+   persistence
    testing