Allow calling processEvent() from notification callbacks

nsk90 · nsk90 · commit 9fa46d073e50 · 2022-08-10T23:06:24.000+03:00
Such calls are recorded by machine and processed when current processing completes.

Do not trigger onStop() when stop() is called multiple times (machine is already stopped).

Fix bug that allowed to call processEvent() without exception from notification callback that was triggered from start() method.
diff --git a/kstatemachine/src/main/kotlin/ru/nsk/kstatemachine/PendingEventHandler.kt b/kstatemachine/src/main/kotlin/ru/nsk/kstatemachine/PendingEventHandler.kt
@@ -0,0 +1,30 @@
+package ru.nsk.kstatemachine
+
+import kotlin.collections.ArrayDeque
+
+/**
+ * Returns [StateMachine.PendingEventHandler] implementation that throws exception. This is an old default behaviour.
+ */
+fun StateMachine.throwingPendingEventHandler() = StateMachine.PendingEventHandler { pendingEvent, _ ->
+    error(
+        "$this can not process pending $pendingEvent as event processing is already running. " +
+                "Do not call processEvent() from notification listeners or use queuePendingEventHandler()"
+    )
+}
+
+fun StateMachine.queuePendingEventHandler(): StateMachine.PendingEventHandler = QueuePendingEventHandler(this)
+
+internal class QueuePendingEventHandler(private val machine: StateMachine): StateMachine.PendingEventHandler {
+    private val queue = ArrayDeque<EventAndArgument<*>>()
+
+    fun checkEmpty() = check(queue.isEmpty()) { "Event queue is not empty, internal error" }
+
+    override fun onPendingEvent(pendingEvent: Event, argument: Any?) {
+        machine.log { "$machine queued event $pendingEvent with argument $argument " }
+        queue.add(EventAndArgument(pendingEvent, argument))
+    }
+
+    fun nextEventAndArgument() = queue.removeFirstOrNull()
+
+    fun clear() = queue.clear()
+}
diff --git a/kstatemachine/src/main/kotlin/ru/nsk/kstatemachine/StateMachineImpl.kt b/kstatemachine/src/main/kotlin/ru/nsk/kstatemachine/StateMachineImpl.kt
@@ -13,24 +13,18 @@ abstract class InternalStateMachine(name: String?, childMode: ChildMode) : State
 
 internal class StateMachineImpl(name: String?, childMode: ChildMode, override val autoDestroyOnStatesReuse: Boolean) :
     InternalStateMachine(name, childMode) {
-    /** Access to this field must be thread safe. */
     private val _machineListeners = mutableSetOf<StateMachine.Listener>()
     override val machineListeners: Collection<StateMachine.Listener> get() = _machineListeners
     override var logger: StateMachine.Logger = NullLogger
     override var ignoredEventHandler = StateMachine.IgnoredEventHandler { _, _ -> }
-    override var pendingEventHandler = StateMachine.PendingEventHandler { pendingEvent, _ ->
-        error(
-            "$this can not process pending $pendingEvent as event processing is already running. " +
-                    "Do not call processEvent() from notification listeners."
-        )
-    }
+    override var pendingEventHandler = queuePendingEventHandler()
     override var listenerExceptionHandler = StateMachine.ListenerExceptionHandler { throw it }
     private var _isDestroyed: Boolean = false
     override val isDestroyed get() = _isDestroyed
 
     /**
-     * Help to check that [processEvent] is not called from state machine notification method.
-     * Access to this field must be thread safe.
+     * Flag for event processing mechanism, which takes place in [processEvent] and during [start]/[startFrom].
+     * It is not possible to process new event while previous processing is incomplete.
      */
     private var isProcessingEvent = false
 
@@ -61,25 +55,30 @@ internal class StateMachineImpl(name: String?, childMode: ChildMode, override va
         accept(CheckUniqueNamesVisitor())
         checkBeforeRunMachine()
 
-        runCheckingExceptions {
-            runMachine(makeStartTransitionParams(this, argument = argument))
-            recursiveEnterInitialStates(argument)
+        eventProcessingScope {
+            runCheckingExceptions {
+                runMachine(makeStartTransitionParams(this, argument = argument))
+                recursiveEnterInitialStates(argument)
+            }
         }
     }
 
     override fun startFrom(state: IState, argument: Any?) {
         checkBeforeRunMachine()
 
-        runCheckingExceptions {
-            val transitionParams = makeStartTransitionParams(this, state, argument)
-            runMachine(transitionParams)
-            switchToTargetState(state as InternalState, this, transitionParams)
+        eventProcessingScope {
+            runCheckingExceptions {
+                val transitionParams = makeStartTransitionParams(this, state, argument)
+                runMachine(transitionParams)
+                switchToTargetState(state as InternalState, this, transitionParams)
+            }
         }
     }
 
     private fun checkBeforeRunMachine() {
         check(!isDestroyed) { "$this is already destroyed" }
         check(!isRunning) { "$this is already started" }
+        check(!isProcessingEvent) { "$this is already processing event, this is internal error, please report a bug" }
         if (childMode == ChildMode.EXCLUSIVE)
             checkNotNull(initialState) { "Initial state is not set, call setInitialState() first" }
     }
@@ -93,6 +92,7 @@ internal class StateMachineImpl(name: String?, childMode: ChildMode, override va
 
     override fun stop() {
         check(!isDestroyed) { "$this is already destroyed" }
+        if (!_isRunning) return
 
         runCheckingExceptions {
             _isRunning = false
@@ -109,26 +109,60 @@ internal class StateMachineImpl(name: String?, childMode: ChildMode, override va
         if (isProcessingEvent) {
             pendingEventHandler.onPendingEvent(event, argument)
             // pending event cannot be processed while previous event is still processing
-            // even if PendingEventHandler does not throw
+            // even if PendingEventHandler does not throw. QueuePendingEventHandler implementation stores such events
+            // to be processed later.
             return
         }
 
+        eventProcessingScope {
+            process(EventAndArgument(event, argument))
+        }
+    }
+
+    private fun process(eventAndArgument: EventAndArgument<*>) {
         var eventProcessed: Boolean? = null
+
+        runCheckingExceptions {
+            eventProcessed = doProcessEvent(eventAndArgument)
+        }
+
+        if (eventProcessed == false) {
+            log { "$this ignored ${eventAndArgument.event::class.simpleName}" }
+            ignoredEventHandler.onIgnoredEvent(eventAndArgument.event, eventAndArgument.argument)
+        }
+    }
+
+    /**
+     * Runs block of code that processes event, and processes all pending events from queue after it if
+     * [QueuePendingEventHandler] is used.
+     */
+    private fun eventProcessingScope(block: () -> Unit) {
+        val queue = pendingEventHandler as? QueuePendingEventHandler
+        queue?.checkEmpty()
+
         isProcessingEvent = true
         try {
-            runCheckingExceptions {
-                eventProcessed = doProcessEvent(EventAndArgument(event, argument))
-            }
+            block()
 
-            if (eventProcessed == false) {
-                log { "$this ignored ${event::class.simpleName}" }
-                ignoredEventHandler.onIgnoredEvent(event, argument)
+            queue?.let {
+                var eventAndArgument = it.nextEventAndArgument()
+                while (eventAndArgument != null) {
+                    process(eventAndArgument)
+
+                    eventAndArgument = it.nextEventAndArgument()
+                }
             }
+        } catch (e: Exception) {
+            queue?.clear()
+            throw e
         } finally {
             isProcessingEvent = false
         }
     }
 
+    /**
+     * Runs block of code that triggers notification listeners
+     */
     private fun runCheckingExceptions(block: () -> Unit) {
         try {
             block()
diff --git a/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/GuardedTransitionTest.kt b/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/GuardedTransitionTest.kt
@@ -20,7 +20,8 @@ class GuardedTransitionTest : StringSpec({
                 transition<SwitchEvent> {
                     guard = {
                         this@initialState.machine.log { "$event $argument" }
-                        value == "value2" }
+                        value == "value2"
+                    }
                     targetState = second
                     callbacks.listen(this)
                 }
diff --git a/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/ListenerExceptionHandlerTest.kt b/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/ListenerExceptionHandlerTest.kt
@@ -83,6 +83,7 @@ class ListenerExceptionHandlerTest : StringSpec({
         }
 
         shouldThrow<TestException> { machine.stop() }
+        machine.stop() // does nothing
         machine.isDestroyed shouldBe false
     }
 
diff --git a/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/PendingEventHandlerTest.kt b/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/PendingEventHandlerTest.kt
@@ -0,0 +1,99 @@
+package ru.nsk.kstatemachine
+
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.core.spec.style.StringSpec
+import io.kotest.matchers.shouldBe
+import io.mockk.verifySequence
+
+class PendingEventHandlerTest : StringSpec({
+    "queue event in QueuePendingEventHandler" {
+        val machine = createStateMachine {
+            logger = StateMachine.Logger { println(it) }
+
+            val third = state("third")
+            val second = state("second") {
+                transition<SecondEvent>(targetState = third)
+            }
+            initialState("first") {
+                transition<FirstEvent> {
+                    targetState = second
+                    onTriggered { this@createStateMachine.processEvent(SecondEvent) }
+                }
+            }
+        }
+
+        machine.processEvent(FirstEvent)
+    }
+
+    "queue event on machine start" {
+        val callbacks = mockkCallbacks()
+        createStateMachine {
+            logger = StateMachine.Logger { println(it) }
+
+            val second = state("second")
+
+            initialState("first") {
+                onEntry { machine.processEvent(SwitchEvent) }
+                initialState("first internal")
+
+                transition<SwitchEvent> {
+                    targetState = second
+                    callbacks.listen(this)
+                }
+            }
+        }
+
+        verifySequence { callbacks.onTriggeredTransition(SwitchEvent) }
+    }
+
+    "pending event queue is cleared on processing error" {
+        val machine = createStateMachine(start = false) {
+            logger = StateMachine.Logger { println(it) }
+
+            val second = state("second") {
+                transition<SecondEvent>()
+            }
+
+            initialState("first") {
+                onEntry {
+                    machine.processEvent(FirstEvent)
+                    machine.processEvent(FirstEvent)
+                    machine.processEvent(FirstEvent)
+                }
+                initialState("first internal")
+
+                transition<FirstEvent>(targetState = second)
+            }
+            ignoredEventHandler = StateMachine.IgnoredEventHandler { _, _ ->
+                throw TestException("test")
+            }
+        }
+
+        shouldThrow<TestException> { machine.start() }
+
+        machine.processEvent(SecondEvent)
+    }
+
+    "throwing PendingEventHandler does not destroy machine" {
+        val machine = createStateMachine {
+            logger = StateMachine.Logger { println(it) }
+
+            val second = state("second")
+            initialState("first") {
+                transition<SwitchEvent> {
+                    targetState = second
+                    onTriggered {
+                        shouldThrow<TestException> { this@createStateMachine.processEvent(SwitchEvent) }
+                    }
+                }
+            }
+
+            pendingEventHandler = StateMachine.PendingEventHandler { _, _ ->
+                testError("Already processing")
+            }
+        }
+
+        machine.processEvent(SwitchEvent)
+        machine.isDestroyed shouldBe false
+    }
+})
diff --git a/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/StateMachineTest.kt b/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/StateMachineTest.kt
@@ -167,29 +167,6 @@ class StateMachineTest : StringSpec({
         shouldThrowUnit<IllegalStateException> { machine.setInitialState(first) }
     }
 
-    "throwing PendingEventHandler does not destroy machine" {
-        val machine = createStateMachine {
-            logger = StateMachine.Logger { println(it) }
-
-            val second = state("second")
-            initialState("first") {
-                transition<SwitchEvent> {
-                    targetState = second
-                    onTriggered {
-                        shouldThrow<TestException> { this@createStateMachine.processEvent(SwitchEvent) }
-                    }
-                }
-            }
-
-            pendingEventHandler = StateMachine.PendingEventHandler { _, _ ->
-                testError("Already processing")
-            }
-        }
-
-        machine.processEvent(SwitchEvent)
-        machine.isDestroyed shouldBe false
-    }
-
     "process event before started" {
         createStateMachine {
             initialState("first")
@@ -273,36 +250,6 @@ class StateMachineTest : StringSpec({
         }
     }
 
-    "startFrom()" {
-        val callbacks = mockkCallbacks()
-
-        lateinit var state2: State
-        lateinit var state22: State
-
-        val machine = createStateMachine(start = false) {
-            callbacks.listen(this)
-
-            initialState("state1") { callbacks.listen(this) }
-            state2 = state("state2") {
-                callbacks.listen(this)
-
-                initialState("state2_1") { callbacks.listen(this) }
-                state22 = state("state2_2") { callbacks.listen(this) }
-            }
-
-            onStarted { callbacks.onStarted(this) }
-        }
-
-        machine.startFrom(state22)
-
-        verifySequence {
-            callbacks.onStarted(machine)
-            callbacks.onEntryState(machine)
-            callbacks.onEntryState(state2)
-            callbacks.onEntryState(state22)
-        }
-    }
-
     "restart machine" {
         val callbacks = mockkCallbacks()
 
@@ -329,6 +276,7 @@ class StateMachineTest : StringSpec({
         }
 
         machine.stop()
+        machine.stop() // does nothing
         verifySequenceAndClear(callbacks) { callbacks.onStopped(machine) }
 
         machine.start()
diff --git a/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/TestingStartFromTest.kt b/kstatemachine/src/test/kotlin/ru/nsk/kstatemachine/TestingStartFromTest.kt

Original file line number	Diff line number	Diff line change
`@@ -83,6 +83,7 @@ class ListenerExceptionHandlerTest : StringSpec({`
`83`	`83`	`}`
`84`	`84`
`85`	`85`	`shouldThrow<TestException> { machine.stop() }`
	`86`	`+ machine.stop() // does nothing`
`86`	`87`	`machine.isDestroyed shouldBe false`
`87`	`88`	`}`
`88`	`89`