GUWT step 1: Implement runningSideEffect API

First part of Kotlin implementation of square/workflow#1021.

Kotlin counterpart to square/workflow#1174.

This implementation intentionally does not run the side effect coroutine
on the `workerContext` `CoroutineContext` that is threaded through the
runtime for testing infrastructure. Initially, workers ran in the same
context as the workflow runtime. The behavior of running workers on a
different dispatcher by default (`Unconfined`) was introduced in
square/workflow#851 as an optimization to reduce
the overhead for running workers that only perform wiring tasks with
other async libraries. This was a theoretical optimization, since running
on the `Unconfined` dispatcher inherently involves less dispatching work,
but the overhead of dispatching wiring coroutines was never actually shown
to be a problem. Additionally, because tests often need to be in full
control of dispatchers, the ability to override the context used for
workers was introduced in square/workflow#940,
which introduced `workerContext`. I am dropping that complexity here
because it adds a decent amount of complexity to worker/side effect
machinery without any proven value. It is also complexity that needs to
be documented, and is probably just more confusing than anything. The
old behavior for workers is maintained for now to reduce the risk of this
change, but side effects will always run in the workflow runtime's
context. This is nice and simple and unsurprising and easy to reason
about.

Author: zach-klippenstein

workflow-core/api/workflow-core.api

@@ -27,6 +27,7 @@ public abstract interface class com/squareup/workflow/RenderContext {
 	public abstract fun makeActionSink ()Lcom/squareup/workflow/Sink;
 	public abstract fun onEvent (Lkotlin/jvm/functions/Function1;)Lkotlin/jvm/functions/Function1;
 	public abstract fun renderChild (Lcom/squareup/workflow/Workflow;Ljava/lang/Object;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)Ljava/lang/Object;
+	public abstract fun runningSideEffect (Ljava/lang/String;Lkotlin/jvm/functions/Function1;)V
 	public abstract fun runningWorker (Lcom/squareup/workflow/Worker;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)V
 }
 

workflow-core/src/main/java/com/squareup/workflow/RenderContext.kt

@@ -116,6 +116,31 @@ interface RenderContext<StateT, in OutputT : Any> {
     key: String = "",
     handler: (T) -> WorkflowAction<StateT, OutputT>
   )
+
+  /**
+   * Ensures [sideEffect] is running with the given [key].
+   *
+   * The first render pass in which this method is called, [sideEffect] will be launched in a new
+   * coroutine that will be scoped to the rendering [Workflow]. Subsequent render passes that invoke
+   * this method with the same [key] will not launch the coroutine again, but let it keep running.
+   * Note that if a different function is passed with the same key, the side effect will *not* be
+   * restarted, the new function will simply be ignored. The next render pass in which the
+   * workflow does not call this method with the same key, the coroutine running [sideEffect] will
+   * be
+   * [cancelled](https://kotlinlang.org/docs/reference/coroutines/cancellation-and-timeouts.html).
+   *
+   * The coroutine will run with the same [CoroutineContext][kotlin.coroutines.CoroutineContext]
+   * that the workflow runtime is running in. The side effect coroutine will not be started until
+   * _after_ the first render call than runs it returns.
+   *
+   * @param key The string key that is used to distinguish between side effects.
+   * @param sideEffect The suspend function that will be launched in a coroutine to perform the
+   * side effect.
+   */
+  fun runningSideEffect(
+    key: String,
+    sideEffect: suspend () -> Unit
+  )
 }
 
 /**

workflow-runtime/api/workflow-runtime.api

@@ -235,13 +235,14 @@ public final class com/squareup/workflow/diagnostic/WorkflowUpdateDebugInfo$Sour
 }
 
 public final class com/squareup/workflow/internal/RealRenderContext : com/squareup/workflow/RenderContext, com/squareup/workflow/Sink {
-	public fun <init> (Lcom/squareup/workflow/internal/RealRenderContext$Renderer;Lcom/squareup/workflow/internal/RealRenderContext$WorkerRunner;Lkotlinx/coroutines/channels/SendChannel;)V
+	public fun <init> (Lcom/squareup/workflow/internal/RealRenderContext$Renderer;Lcom/squareup/workflow/internal/RealRenderContext$WorkerRunner;Lcom/squareup/workflow/internal/RealRenderContext$SideEffectRunner;Lkotlinx/coroutines/channels/SendChannel;)V
 	public final fun freeze ()V
 	public fun getActionSink ()Lcom/squareup/workflow/Sink;
 	public fun makeActionSink ()Lcom/squareup/workflow/Sink;
 	public fun onEvent (Lkotlin/jvm/functions/Function1;)Lcom/squareup/workflow/EventHandler;
 	public synthetic fun onEvent (Lkotlin/jvm/functions/Function1;)Lkotlin/jvm/functions/Function1;
 	public fun renderChild (Lcom/squareup/workflow/Workflow;Ljava/lang/Object;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)Ljava/lang/Object;
+	public fun runningSideEffect (Ljava/lang/String;Lkotlin/jvm/functions/Function1;)V
 	public fun runningWorker (Lcom/squareup/workflow/Worker;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)V
 	public fun send (Lcom/squareup/workflow/WorkflowAction;)V
 	public synthetic fun send (Ljava/lang/Object;)V
@@ -251,6 +252,10 @@ public abstract interface class com/squareup/workflow/internal/RealRenderContext
 	public abstract fun render (Lcom/squareup/workflow/Workflow;Ljava/lang/Object;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)Ljava/lang/Object;
 }
 
+public abstract interface class com/squareup/workflow/internal/RealRenderContext$SideEffectRunner {
+	public abstract fun runningSideEffect (Ljava/lang/String;Lkotlin/jvm/functions/Function1;)V
+}
+
 public abstract interface class com/squareup/workflow/internal/RealRenderContext$WorkerRunner {
 	public abstract fun runningWorker (Lcom/squareup/workflow/Worker;Ljava/lang/String;Lkotlin/jvm/functions/Function1;)V
 }

workflow-runtime/src/main/java/com/squareup/workflow/internal/RealRenderContext.kt

@@ -33,6 +33,7 @@ import kotlinx.coroutines.channels.SendChannel
 class RealRenderContext<StateT, OutputT : Any>(
   private val renderer: Renderer<StateT, OutputT>,
   private val workerRunner: WorkerRunner<StateT, OutputT>,
+  private val sideEffectRunner: SideEffectRunner,
   private val eventActionsChannel: SendChannel<WorkflowAction<StateT, OutputT>>
 ) : RenderContext<StateT, OutputT>, Sink<WorkflowAction<StateT, OutputT>> {
 
@@ -53,6 +54,13 @@ class RealRenderContext<StateT, OutputT : Any>(
     )
   }
 
+  interface SideEffectRunner {
+    fun runningSideEffect(
+      key: String,
+      sideEffect: suspend () -> Unit
+    )
+  }
+
   /**
    * False during the current render call, set to true once this node is finished rendering.
    *
@@ -104,6 +112,14 @@ class RealRenderContext<StateT, OutputT : Any>(
     workerRunner.runningWorker(worker, key, handler)
   }
 
+  override fun runningSideEffect(
+    key: String,
+    sideEffect: suspend () -> Unit
+  ) {
+    checkNotFrozen()
+    sideEffectRunner.runningSideEffect(key, sideEffect)
+  }
+
   /**
    * Freezes this context so that any further calls to this context will throw.
    */

workflow-runtime/src/main/java/com/squareup/workflow/internal/SideEffectNode.kt

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2020 Square Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.squareup.workflow.internal
+
+import com.squareup.workflow.internal.InlineLinkedList.InlineListNode
+import kotlinx.coroutines.Job
+
+/**
+ * Holds a [Job] that represents a running [side effect][RealRenderContext.runningSideEffect], as
+ * well as the key used to identify that side effect.
+ */
+internal class SideEffectNode(
+  val key: String,
+  val job: Job
+) : InlineListNode<SideEffectNode> {
+
+  override var nextListNode: SideEffectNode? = null
+}

workflow-runtime/src/main/java/com/squareup/workflow/internal/WorkflowNode.kt

@@ -15,24 +15,28 @@
  */
 package com.squareup.workflow.internal
 
+import com.squareup.workflow.ExperimentalWorkflow
 import com.squareup.workflow.Snapshot
 import com.squareup.workflow.StatefulWorkflow
-import com.squareup.workflow.ExperimentalWorkflow
 import com.squareup.workflow.Worker
 import com.squareup.workflow.Workflow
 import com.squareup.workflow.WorkflowAction
 import com.squareup.workflow.applyTo
 import com.squareup.workflow.diagnostic.IdCounter
 import com.squareup.workflow.diagnostic.WorkflowDiagnosticListener
 import com.squareup.workflow.diagnostic.createId
+import com.squareup.workflow.internal.RealRenderContext.SideEffectRunner
 import com.squareup.workflow.internal.RealRenderContext.WorkerRunner
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart.LAZY
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.plus
 import kotlinx.coroutines.selects.SelectBuilder
 import okio.ByteString
 import kotlin.coroutines.CoroutineContext
@@ -63,7 +67,7 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
   private val idCounter: IdCounter? = null,
   initialState: StateT? = null,
   private val workerContext: CoroutineContext = EmptyCoroutineContext
-) : CoroutineScope, WorkerRunner<StateT, OutputT> {
+) : CoroutineScope, WorkerRunner<StateT, OutputT>, SideEffectRunner {
 
   /**
    * Context that has a job that will live as long as this node.
@@ -84,6 +88,8 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
 
   private val workers = ActiveStagingList<WorkerChildNode<*, *, *>>()
 
+  private val sideEffects = ActiveStagingList<SideEffectNode>()
+
   private var state: StateT
 
   private var lastProps: PropsT = initialProps
@@ -144,7 +150,7 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
     key: String,
     handler: (T) -> WorkflowAction<StateT, OutputT>
   ) {
-    // Prevent duplicate workflows with the same key.
+    // Prevent duplicate workers with the same key.
     workers.forEachStaging {
       require(!(it.matches(worker, key))) {
         "Expected keys to be unique for $worker: key=$key"
@@ -159,6 +165,21 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
     stagedWorker.setHandler(handler)
   }
 
+  override fun runningSideEffect(
+    key: String,
+    sideEffect: suspend () -> Unit
+  ) {
+    // Prevent duplicate side effects with the same key.
+    sideEffects.forEachStaging {
+      require(key != it.key) { "Expected side effect keys to be unique: $key" }
+    }
+
+    sideEffects.retainOrCreate(
+        predicate = { key == it.key },
+        create = { createSideEffectNode(key, sideEffect) }
+    )
+  }
+
   /**
    * Gets the next [output][OutputT] from the state machine.
    *
@@ -229,6 +250,7 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
     val context = RealRenderContext(
         renderer = subtreeManager,
         workerRunner = this,
+        sideEffectRunner = this,
         eventActionsChannel = eventActionsChannel
     )
     diagnosticListener?.onBeforeWorkflowRendered(diagnosticId, props, state)
@@ -239,6 +261,10 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
     // Tear down workflows and workers that are obsolete.
     subtreeManager.commitRenderedChildren()
     workers.commitStaging { it.channel.cancel() }
+    // 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() }
 
     return rendering
   }
@@ -282,6 +308,15 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
     return WorkerChildNode(worker, key, workerChannel, handler = handler)
   }
 
+  private fun createSideEffectNode(
+    key: String,
+    sideEffect: suspend () -> Unit
+  ): SideEffectNode {
+    val scope = this + CoroutineName("sideEffect[$key] for $id")
+    val job = scope.launch(start = LAZY) { sideEffect() }
+    return SideEffectNode(key, job)
+  }
+
   private fun ByteString.restoreState(): Snapshot? {
     val (snapshotToRestoreFrom, childSnapshots) = parseTreeSnapshot(this)
     subtreeManager.restoreChildrenFromSnapshots(childSnapshots)