GUWT step 1: Implement runningSideEffect API

First part of Kotlin implementation of #1021.

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,21 @@ interface RenderContext<StateT, in OutputT : Any> {
     key: String = "",
     handler: (T) -> WorkflowAction<StateT, OutputT>
   )
+
+  /**
+   * Ensures [sideEffect] is running with the given [key].
+   *
+   * @param key The string key that is used to distinguish between side effects.
+   * @param sideEffect A suspending lambda that will be started the first time this method called
+   * with the given key, and
+   * [cancelled](https://kotlinlang.org/docs/reference/coroutines/cancellation-and-timeouts.html)
+   * after it is no longer called with the given key (or this workflow is torn down, whichever
+   * happens first).
+   */
+  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,30 @@
+/*
+ * 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
+
+/**
+ * TODO write documentation
+ */
+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

@@ -25,14 +25,18 @@ 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 must not
+    // be started until after context is frozen.
+    sideEffects.forEachStaging { it.job.start() }
+    sideEffects.commitStaging { it.job.cancel() }
 
     return rendering
   }
@@ -282,6 +308,16 @@ internal class WorkflowNode<PropsT, StateT, OutputT : Any, RenderingT>(
     return WorkerChildNode(worker, key, workerChannel, handler = handler)
   }
 
+  private fun createSideEffectNode(
+    key: String,
+    sideEffect: suspend () -> Unit
+  ): SideEffectNode {
+    // workerContext is guaranteed to not contain a Job.
+    val scope = this + CoroutineName("sideEffect[$key] for $id") + workerContext
+    val job = scope.launch(start = LAZY) { sideEffect() }
+    return SideEffectNode(key, job)
+  }
+
   private fun ByteString.restoreState(): Snapshot? {
     val (snapshotToRestoreFrom, childSnapshots) = parseTreeSnapshot(this)
     subtreeManager.restoreChildrenFromSnapshots(childSnapshots)