#1311: Start side effects DEFAULT instead of LAZY.

settings.gradle.kts

@@ -64,6 +64,7 @@ include(
   ":workflow-config:config-jvm",
   ":workflow-core",
   ":workflow-runtime",
+  ":workflow-runtime-android",
   ":workflow-rx2",
   ":workflow-testing",
   ":workflow-tracing",

workflow-runtime-android/README.md

@@ -0,0 +1,4 @@
+# Module Workflow Runtime Android
+
+This module is an Android library that is used to test the Workflow Runtime with Android specific
+coroutine dispatchers. These are headless android-tests that run on device without UI.

workflow-runtime-android/build.gradle.kts

@@ -0,0 +1,25 @@
+plugins {
+  id("com.android.library")
+  id("kotlin-android")
+  id("android-defaults")
+  id("android-ui-tests")
+}
+
+android {
+  namespace = "com.squareup.workflow1"
+  testNamespace = "$namespace.test"
+}
+
+dependencies {
+  api(project(":workflow-runtime"))
+  implementation(project(":workflow-core"))
+
+  androidTestImplementation(libs.androidx.test.core)
+  androidTestImplementation(libs.androidx.test.truth)
+  androidTestImplementation(libs.kotlin.test.core)
+  androidTestImplementation(libs.kotlin.test.jdk)
+  androidTestImplementation(libs.kotlinx.coroutines.android)
+  androidTestImplementation(libs.kotlinx.coroutines.core)
+  androidTestImplementation(libs.kotlinx.coroutines.test)
+  androidTestImplementation(libs.truth)
+}

workflow-runtime-android/gradle.properties

@@ -0,0 +1,3 @@
+POM_ARTIFACT_ID=workflow-runtime-android
+POM_NAME=Workflow Runtime Android
+POM_PACKAGING=aar

workflow-runtime-android/src/androidTest/AndroidManifest.xml

@@ -0,0 +1,6 @@
+<manifest xmlns:android="http://schemas.android.com/apk/res/android">
+
+  <application>
+    <activity android:name="androidx.activity.ComponentActivity"/>
+  </application>
+</manifest>

workflow-runtime-android/src/androidTest/java/com/squareup/workflow1/AndroidRenderWorkflowInTest.kt

@@ -0,0 +1,121 @@
+package com.squareup.workflow1
+
+import com.squareup.workflow1.RuntimeConfigOptions.CONFLATE_STALE_RENDERINGS
+import com.squareup.workflow1.WorkflowInterceptor.RenderPassesComplete
+import com.squareup.workflow1.WorkflowInterceptor.RuntimeLoopOutcome
+import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.receiveAsFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.plus
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.test.UnconfinedTestDispatcher
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+@OptIn(WorkflowExperimentalRuntime::class, ExperimentalCoroutinesApi::class)
+class AndroidRenderWorkflowInTest {
+
+  // @Ignore("#1311: Does not yet work with immediate dispatcher.")
+  @Test
+  fun with_conflate_we_conflate_stacked_actions_into_one_rendering() =
+    runTest(UnconfinedTestDispatcher()) {
+
+      var childHandlerActionExecuted = false
+      val trigger1 = Channel<String>(capacity = 1)
+      val trigger2 = Channel<String>(capacity = 1)
+      val emitted = mutableListOf<String>()
+      var renderingsPassed = 0
+      val countInterceptor = object : WorkflowInterceptor {
+        override fun onRuntimeLoopTick(outcome: RuntimeLoopOutcome) {
+          if (outcome is RenderPassesComplete<*>) {
+            renderingsPassed++
+          }
+        }
+      }
+
+      val childWorkflow = Workflow.stateful<String, String, String>(
+        initialState = "unchanging state",
+        render = { renderState ->
+          runningWorker(
+            trigger1.receiveAsFlow().asWorker()
+          ) {
+            action("") {
+              state = it
+              setOutput(it)
+            }
+          }
+          renderState
+        }
+      )
+      val workflow = Workflow.stateful<String, String, String>(
+        initialState = "unchanging state",
+        render = { renderState ->
+          renderChild(childWorkflow) { childOutput ->
+            action("childHandler") {
+              childHandlerActionExecuted = true
+              state = childOutput
+            }
+          }
+          runningWorker(
+            trigger2.receiveAsFlow().asWorker()
+          ) {
+            action("") {
+              // Update the rendering in order to show conflation.
+              state = "$it+update"
+              setOutput(state)
+            }
+          }
+          renderState
+        }
+      )
+      val props = MutableStateFlow(Unit)
+      // Render this on the Main.immediate dispatcher from Android.
+      val renderings = renderWorkflowIn(
+        workflow = workflow,
+        scope = backgroundScope +
+          Dispatchers.Main.immediate,
+        props = props,
+        runtimeConfig = setOf(CONFLATE_STALE_RENDERINGS),
+        workflowTracer = null,
+        interceptors = listOf(countInterceptor)
+      ) { }
+
+      val renderedMutex = Mutex(locked = true)
+
+      val collectionJob = launch(context = Dispatchers.Main.immediate, start = UNDISPATCHED) {
+        // Collect this unconfined so we can get all the renderings faster than actions can
+        // be processed.
+        renderings.collect {
+          emitted += it.rendering
+          if (it.rendering == "changed state 2+update") {
+            renderedMutex.unlock()
+          }
+        }
+      }
+
+      launch(context = Dispatchers.Main.immediate, start = UNDISPATCHED) {
+        trigger1.trySend("changed state 1")
+        trigger2.trySend("changed state 2")
+      }.join()
+
+      renderedMutex.lock()
+
+      collectionJob.cancel()
+
+      // 2 renderings (initial and then the update.) Not *3* renderings.
+      assertEquals(2, emitted.size, "Expected only 2 renderings when conflating actions.")
+      assertEquals(
+        2,
+        renderingsPassed,
+        "Expected only 2 renderings passed when conflating actions."
+      )
+      assertEquals("changed state 2+update", emitted.last())
+      assertTrue(childHandlerActionExecuted)
+    }
+}

workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/RealRenderContext.kt

@@ -100,10 +100,23 @@ internal class RealRenderContext<out PropsT, StateT, OutputT>(
    * Freezes this context so that any further calls to this context will throw.
    */
   fun freeze() {
-    checkNotFrozen("freeze") { "freeze" }
+    // checkNotFrozen("freeze") { "freeze" }
     frozen = true
   }
 
+  /**
+   * Freezes this context if it is not currently frozen.
+   *
+   * @return Boolean of whether or not the context was already frozen.
+   */
+  internal fun freezeIfNotFrozen(): Boolean {
+    if (!frozen) {
+      freeze()
+      return false
+    }
+    return true
+  }
+
   /**
    * Unfreezes when the node is about to render() again.
    */

workflow-runtime/src/commonMain/kotlin/com/squareup/workflow1/internal/WorkflowNode.kt

@@ -26,7 +26,7 @@ import com.squareup.workflow1.trace
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.CoroutineStart.LAZY
+import kotlinx.coroutines.CoroutineStart.DEFAULT
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.Job
@@ -36,7 +36,10 @@ import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.plus
 import kotlinx.coroutines.selects.SelectBuilder
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.ContinuationInterceptor
 import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.CoroutineContext.Key
 import kotlin.reflect.KType
 
 /**
@@ -279,9 +282,6 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
       workflowTracer.trace("UpdateRuntimeTree") {
         // Tear down workflows and workers that are obsolete.
         subtreeManager.commitRenderedChildren()
-        // Side effect jobs are launched lazily, since they can send actions to the sink, and can only
-        // be started after context is frozen.
-        sideEffects.forEachStaging { it.job.start() }
         sideEffects.commitStaging { it.job.cancel() }
         remembered.commitStaging { /* Nothing to clean up. */ }
       }
@@ -351,13 +351,42 @@ internal class WorkflowNode<PropsT, StateT, OutputT, RenderingT>(
     }
   }
 
+  inner class FrozenContextContinuationInterceptor : ContinuationInterceptor {
+    override val key: Key<*>
+      get() = ContinuationInterceptor.Key
+
+    private var originallyFrozen = false
+
+    override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
+      object : Continuation<T> {
+        override val context: CoroutineContext
+          get() = continuation.context
+
+        override fun resumeWith(result: Result<T>) {
+          originallyFrozen = baseRenderContext.freezeIfNotFrozen()
+          continuation.resumeWith(result)
+        }
+      }
+
+    override fun releaseInterceptedContinuation(continuation: Continuation<*>) {
+      if (!originallyFrozen) {
+        baseRenderContext.unfreeze()
+      }
+    }
+  }
+
   private fun createSideEffectNode(
     key: String,
     sideEffect: suspend CoroutineScope.() -> Unit
   ): SideEffectNode {
     return workflowTracer.trace("CreateSideEffectNode") {
-      val scope = this + CoroutineName("sideEffect[$key] for $id")
-      val job = scope.launch(start = LAZY, block = sideEffect)
+      val scope = this +
+        CoroutineName("sideEffect[$key] for $id") +
+        // Adds the ContinuationInterceptor that freezes whenever we dispatch the continuation
+        // for the side effect. This allows us to schedule side effects during the render
+        // pass.
+        FrozenContextContinuationInterceptor()
+      val job = scope.launch(start = DEFAULT, block = sideEffect)
       SideEffectNode(key, job)
     }
   }

workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/RenderWorkflowInTest.kt

@@ -198,7 +198,7 @@ class RenderWorkflowInTest {
   }
 
   @Test
-  fun `side_effects_from_initial_rendering_in_root_workflow_are_never_started_when_scope_cancelled_before_start`() {
+  fun `side_effects_from_initial_rendering_in_root_workflow_are_started_when_scope_cancelled_before_start`() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       before = ::setup,
@@ -222,13 +222,13 @@ class RenderWorkflowInTest {
         ) {}
         advanceIfStandard(dispatcher)
 
-        assertFalse(sideEffectWasRan)
+        assertTrue(sideEffectWasRan)
       }
     }
   }
 
   @Test
-  fun `side_effects_from_initial_rendering_in_non_root_workflow_are_never_started_when_scope_cancelled_before_start`() {
+  fun `side_effects_from_initial_rendering_in_non_root_workflow_are_started_when_scope_cancelled_before_start`() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       before = ::setup,
@@ -255,7 +255,7 @@ class RenderWorkflowInTest {
         ) {}
         advanceIfStandard(dispatcher)
 
-        assertFalse(sideEffectWasRan)
+        assertTrue(sideEffectWasRan)
       }
     }
   }
@@ -765,7 +765,7 @@ class RenderWorkflowInTest {
   }
 
   @Test
-  fun `side_effects_from_initial_rendering_in_root_workflow_are_never_started_when_initial_render_of_root_workflow_fails`() {
+  fun `side_effects_from_initial_rendering_in_root_workflow_are_started_when_initial_render_of_root_workflow_fails`() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       before = ::setup,
@@ -788,7 +788,7 @@ class RenderWorkflowInTest {
             workflowTracer = workflowTracer,
           ) {}
         }
-        assertFalse(sideEffectWasRan)
+        assertTrue(sideEffectWasRan)
       }
     }
   }
@@ -838,7 +838,7 @@ class RenderWorkflowInTest {
   }
 
   @Test
-  fun `side_effects_from_initial_rendering_in_non_root_workflow_are_never_started_when_initial_render_of_non_root_workflow_fails`() {
+  fun `side_effects_from_initial_rendering_in_non_root_workflow_are_started_when_initial_render_of_non_root_workflow_fails`() {
     runtimeTestRunner.runParametrizedTest(
       paramSource = runtimeOptions,
       before = ::setup,
@@ -864,7 +864,7 @@ class RenderWorkflowInTest {
             workflowTracer = workflowTracer,
           ) {}
         }
-        assertFalse(sideEffectWasRan)
+        assertTrue(sideEffectWasRan)
       }
     }
   }

workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/RealRenderContextTest.kt

@@ -220,7 +220,6 @@ internal class RealRenderContextTest {
 
     val child = Workflow.stateless<Unit, Nothing, Unit> { fail() }
     assertFailsWith<IllegalStateException> { context.renderChild(child) }
-    assertFailsWith<IllegalStateException> { context.freeze() }
     assertFailsWith<IllegalStateException> { context.remember("key", typeOf<String>()) {} }
   }
 

workflow-runtime/src/commonTest/kotlin/com/squareup/workflow1/internal/WorkflowNodeTest.kt

@@ -276,13 +276,14 @@ internal class WorkflowNodeTest {
     sink.send(action("") { setOutput("event2") })
   }
 
-  @Test fun sideEffect_is_not_started_until_after_render_completes() {
+  @Test fun sideEffect_is_started_immediately() {
     var started = false
     val workflow = Workflow.stateless<Unit, Nothing, Unit> {
       runningSideEffect("key") {
         started = true
       }
-      assertFalse(started)
+      // This is because context uses and Unconfined Dispatcher, so is eagerly entered.
+      assertTrue(started)
     }
     val node = WorkflowNode(
       workflow.id(),
@@ -466,8 +467,8 @@ internal class WorkflowNodeTest {
 
   @Test fun multiple_sideEffects_with_same_key_throws() {
     val workflow = Workflow.stateless<Unit, Nothing, Unit> {
-      runningSideEffect("same") { fail() }
-      runningSideEffect("same") { fail() }
+      runningSideEffect("same") { }
+      runningSideEffect("same") { }
     }
     val node = WorkflowNode(
       workflow.id(),