EventsourcedRunning, internalPersist, internalPersistAll

* I don't think the JournalInteractions should be conserned with
  next behavior

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventsourcedJournalInteractions.scala

@@ -34,10 +34,7 @@ private[akka] trait EventsourcedJournalInteractions[C, E, S] {
   type EventOrTagged = Any // `Any` since can be `E` or `Tagged`
   protected def internalPersist(
     state: EventsourcedRunning.EventsourcedState[S],
-    event:       EventOrTagged,
+    event: EventOrTagged): EventsourcedRunning.EventsourcedState[S] = {
-    sideEffects: immutable.Seq[ChainableEffect[_, S]])(handler: Any ⇒ Unit): Behavior[InternalProtocol] = {
-    // pendingInvocations addLast StashingHandlerInvocation(event, handler.asInstanceOf[Any ⇒ Unit])
-    val pendingInvocations = StashingHandlerInvocation(event, handler.asInstanceOf[Any ⇒ Unit]) :: Nil
 
     val newState = state.nextSequenceNr()
 
@@ -53,35 +50,28 @@ private[akka] trait EventsourcedJournalInteractions[C, E, S] {
     val eventBatch = AtomicWrite(repr) :: Nil // batching not used, since no persistAsync
     setup.journal.tell(JournalProtocol.WriteMessages(eventBatch, setup.selfUntypedAdapted, setup.writerIdentity.instanceId), setup.selfUntypedAdapted)
 
-    EventsourcedRunning.PersistingEvents[C, E, S](setup, state, pendingInvocations, sideEffects)
+    newState
   }
 
   protected def internalPersistAll(
     events: immutable.Seq[EventOrTagged],
-    state:       EventsourcedRunning.EventsourcedState[S],
+    state:  EventsourcedRunning.EventsourcedState[S]): EventsourcedRunning.EventsourcedState[S] = {
-    sideEffects: immutable.Seq[ChainableEffect[_, S]])(handler: Any ⇒ Unit): Behavior[InternalProtocol] = {
     if (events.nonEmpty) {
-
-      val pendingInvocations = events map { event ⇒
-        // pendingInvocations addLast StashingHandlerInvocation(event, handler.asInstanceOf[Any ⇒ Unit])
-        StashingHandlerInvocation(event, handler.asInstanceOf[Any ⇒ Unit])
-      }
-
       val newState = state.nextSequenceNr()
 
       val senderNotKnownBecauseAkkaTyped = null
       val write = AtomicWrite(events.map(event ⇒ PersistentRepr(
         event,
         persistenceId = setup.persistenceId,
-        sequenceNr = newState.seqNr,
+        sequenceNr = newState.seqNr, // FIXME increment for each event
         writerUuid = setup.writerIdentity.writerUuid,
         sender = senderNotKnownBecauseAkkaTyped)
       ))
 
       setup.journal.tell(JournalProtocol.WriteMessages(write :: Nil, setup.selfUntypedAdapted, setup.writerIdentity.instanceId), setup.selfUntypedAdapted)
 
-      EventsourcedRunning.PersistingEvents(setup, state, pendingInvocations, sideEffects)
+      newState
-    } else Behaviors.same
+    } else state
   }
 
   protected def replayEvents(fromSeqNr: Long, toSeqNr: Long): Unit = {

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventsourcedRecoveringEvents.scala

@@ -177,7 +177,7 @@ private[persistence] class EventsourcedRecoveringEvents[C, E, S](
     returnRecoveryPermit(setup, "recovery completed successfully")
     setup.recoveryCompleted(setup.commandContext, state.state)
 
-    val running = EventsourcedRunning.HandlingCommands[C, E, S](
+    val running = EventsourcedRunning[C, E, S](
       setup,
       EventsourcedRunning.EventsourcedState[S](state.seqNr, state.state)
     )

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventsourcedRunning.scala

@@ -65,18 +65,18 @@ import scala.collection.immutable
     }
   }
 
-  // ===============================================
-
-  object HandlingCommands {
   def apply[C, E, S](setup: EventsourcedSetup[C, E, S], state: EventsourcedState[S]): Behavior[InternalProtocol] =
-      new HandlingCommands(setup).createBehavior(state)
+    new EventsourcedRunning(setup).handlingCommands(state)
 }
 
-  class HandlingCommands[C, E, S](override val setup: EventsourcedSetup[C, E, S])
+// ===============================================
-    extends EventsourcedJournalInteractions[C, E, S] with EventsourcedStashManagement[C, E, S] {
 
-    def createBehavior(state: EventsourcedState[S]): Behavior[InternalProtocol] = {
+@InternalApi private[akka] class EventsourcedRunning[C, E, S](override val setup: EventsourcedSetup[C, E, S])
-      withMdc {
+  extends EventsourcedJournalInteractions[C, E, S] with EventsourcedStashManagement[C, E, S] {
+  import EventsourcedRunning.EventsourcedState
+
+  def handlingCommands(state: EventsourcedState[S]): Behavior[InternalProtocol] = {
+    withMdc("run-cmnds") {
       Behaviors.immutable[EventsourcedBehavior.InternalProtocol] {
         case (_, SnapshotterResponse(r))           ⇒ Behaviors.unhandled
         case (_, JournalResponse(r))               ⇒ Behaviors.unhandled
@@ -85,10 +85,10 @@ import scala.collection.immutable
     }
   }
 
-    private def withMdc(wrapped: Behavior[InternalProtocol]) = {
+  private def withMdc(phase: String)(wrapped: Behavior[InternalProtocol]) = {
     val mdc = Map(
       "persistenceId" → setup.persistenceId,
-        "phase" → "run-cmnds"
+      "phase" → phase
     )
 
     Behaviors.withMdc((_: Any) ⇒ mdc, wrapped)
@@ -122,10 +122,17 @@ import scala.collection.immutable
         val newState = state.applyEvent(setup, event)
         val eventToPersist = tagEvent(event)
 
-          internalPersist(state, eventToPersist, sideEffects) { _ ⇒
+        val newState2 = internalPersist(newState, eventToPersist)
-            if (setup.snapshotWhen(newState.state, event, newState.seqNr))
+
+        val handler: Any ⇒ Unit = { x ⇒ // TODO is x the new state?
+          if (setup.snapshotWhen(newState2.state, event, newState2.seqNr))
             internalSaveSnapshot(state)
         }
+        val pendingInvocations = StashingHandlerInvocation(event, handler) :: Nil
+
+        // FIXME applySideEffects is missing
+
+        persistingEvents(newState2, pendingInvocations, sideEffects)
 
       case PersistAll(events) ⇒
         if (events.nonEmpty) {
@@ -138,26 +145,33 @@ import scala.collection.immutable
             events.foldLeft((state, false)) {
               case ((currentState, snapshot), event) ⇒
                 val value = currentState
-                    .nextSequenceNr()
+                  .nextSequenceNr() // FIXME seqNr is also incremented in internalPersistAll
                   .applyEvent(setup, event)
 
                 val shouldSnapshot = snapshot || setup.snapshotWhen(value.state, event, value.seqNr)
                 (value, shouldSnapshot)
             }
-            // state = newState
 
           val eventsToPersist = events.map(tagEvent)
 
-            internalPersistAll(eventsToPersist, newState, sideEffects) { _ ⇒
+          val newState2 = internalPersistAll(eventsToPersist, newState)
+
+          val handler: Any ⇒ Unit = { _ ⇒
             count -= 1
             if (count == 0) {
-                // FIXME the result of applying side effects is ignored
+              //                // FIXME the result of applying side effects is ignored
-                val b = applySideEffects(sideEffects, newState)
+              //                val b = applySideEffects(sideEffects, newState)
-
               if (shouldSnapshotAfterPersist)
                 internalSaveSnapshot(newState)
             }
           }
+
+          val pendingInvocations = events map { event ⇒
+            StashingHandlerInvocation(event, handler)
+          }
+
+          persistingEvents(newState2, pendingInvocations, sideEffects)
+
         } else {
           // run side-effects even when no events are emitted
           tryUnstash(applySideEffects(sideEffects, state))
@@ -180,25 +194,9 @@ import scala.collection.immutable
     if (tags.isEmpty) event else Tagged(event, tags)
   }
 
-  }
-
   // ===============================================
 
-  object PersistingEvents {
+  def persistingEvents(
-    def apply[C, E, S](
-      setup:              EventsourcedSetup[C, E, S],
-      state:              EventsourcedState[S],
-      pendingInvocations: immutable.Seq[PendingHandlerInvocation],
-      sideEffects:        immutable.Seq[ChainableEffect[_, S]]
-    ): Behavior[InternalProtocol] =
-      new PersistingEvents(setup).createBehavior(state, pendingInvocations, sideEffects)
-
-  }
-
-  class PersistingEvents[C, E, S](override val setup: EventsourcedSetup[C, E, S])
-    extends EventsourcedJournalInteractions[C, E, S] with EventsourcedStashManagement[C, E, S] {
-
-    def createBehavior(
     state:              EventsourcedState[S],
     pendingInvocations: immutable.Seq[PendingHandlerInvocation],
     sideEffects:        immutable.Seq[ChainableEffect[_, S]]
@@ -304,11 +302,9 @@ import scala.collection.immutable
     }
   }
 
-  }
-
   // --------------------------
 
-  def applySideEffects[S](effects: immutable.Seq[ChainableEffect[_, S]], state: EventsourcedState[S]): Behavior[InternalProtocol] = {
+  def applySideEffects(effects: immutable.Seq[ChainableEffect[_, S]], state: EventsourcedState[S]): Behavior[InternalProtocol] = {
     var res: Behavior[InternalProtocol] = Behaviors.same
     val it = effects.iterator
 
@@ -325,7 +321,7 @@ import scala.collection.immutable
     res
   }
 
-  def applySideEffect[S](effect: ChainableEffect[_, S], state: EventsourcedState[S]): Behavior[InternalProtocol] = effect match {
+  def applySideEffect(effect: ChainableEffect[_, S], state: EventsourcedState[S]): Behavior[InternalProtocol] = effect match {
     case _: Stop.type @unchecked ⇒
       Behaviors.stopped