Merge pull request #282 from square/zachklipp/workflowhost-inputs

Add the ability to update the input for a WorkflowHost.

Author: rjrjr

kotlin/samples/tictactoe/common/src/test/java/com/squareup/sample/gameworkflow/TakeTurnsWorkflowTest.kt

@@ -69,6 +69,6 @@ class TakeTurnsWorkflowTest {
   }
 }
 
-private fun WorkflowTester<*, GamePlayScreen>.takeSquare(event: TakeSquare) {
+private fun WorkflowTester<*, *, GamePlayScreen>.takeSquare(event: TakeSquare) {
   withNextRendering { it.onEvent(event) }
 }

kotlin/samples/tictactoe/common/src/test/java/com/squareup/sample/mainworkflow/MainWorkflowTest.kt

@@ -40,6 +40,9 @@ class MainWorkflowTest {
     }
 
     MainWorkflow(authWorkflow, runGameWorkflow()).testFromStart { tester ->
+      tester.withNextRendering { screen ->
+        assertThat(screen.panels).containsOnly(DEFAULT_AUTH)
+      }
       tester.withNextRendering { screen ->
         assertThat(screen.panels).isEmpty()
         assertThat(screen.body).isEqualTo(DEFAULT_RUN_GAME)

kotlin/settings.gradle

@@ -23,6 +23,7 @@ include ':samples:tictactoe:common'
 include ':workflow-core'
 include ':workflow-runtime'
 include ':workflow-rx2'
+include ':workflow-rx2-runtime'
 include ':workflow-testing'
 include ':workflow-ui-core'
 include ':workflow-ui-android'

kotlin/workflow-core/src/main/java/com/squareup/workflow/WorkflowContext.kt

@@ -21,6 +21,7 @@ import com.squareup.workflow.util.ChannelUpdate
 import com.squareup.workflow.util.ChannelUpdate.Value
 import com.squareup.workflow.util.KTypes
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.CoroutineStart.UNDISPATCHED
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.produce
@@ -259,6 +260,9 @@ fun <T, StateT : Any, OutputT : Any> WorkflowContext<StateT, OutputT>.onDeferred
  * This function is provided as a helper for writing [WorkflowContext] extension functions, it
  * should not be used by general application code.
  *
+ * The suspending function will be executed in the current stack frame ([UNDISPATCHED]). When this
+ * workflow is being torn down, the coroutine running the function will be cancelled.
+ *
  * @param type The [KType] that represents both the type of data source (e.g. `Deferred`) and the
  * element type [T].
  * @param key An optional string key that is used to distinguish between subscriptions of the same
@@ -289,10 +293,9 @@ fun <T, StateT : Any, OutputT : Any> WorkflowContext<StateT, OutputT>.onSuspendi
  */
 private fun <T> CoroutineScope.wrapInNeverClosingChannel(
   function: suspend () -> T
-): ReceiveChannel<T> =
-  produce {
-    send(function())
-    // We explicitly don't want to close the channel, because that would trigger an infinite loop.
-    // Instead, just suspend forever.
-    suspendCancellableCoroutine<Nothing> { }
-  }
+): ReceiveChannel<T> = produce {
+  send(function())
+  // We explicitly don't want to close the channel, because that would trigger an infinite loop.
+  // Instead, just suspend forever.
+  suspendCancellableCoroutine<Nothing> { }
+}

kotlin/workflow-runtime/src/main/java/com/squareup/workflow/WorkflowHost.kt

@@ -19,24 +19,29 @@ package com.squareup.workflow
 
 import com.squareup.workflow.WorkflowHost.Factory
 import com.squareup.workflow.WorkflowHost.Update
-import com.squareup.workflow.internal.WorkflowId
 import com.squareup.workflow.internal.WorkflowNode
 import com.squareup.workflow.internal.id
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.InternalCoroutinesApi
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.NonCancellable.isActive
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.produce
-import kotlinx.coroutines.isActive
 import kotlinx.coroutines.selects.select
 import org.jetbrains.annotations.TestOnly
 import kotlin.coroutines.CoroutineContext
 import kotlin.coroutines.EmptyCoroutineContext
+import kotlin.coroutines.coroutineContext
 
 /**
  * Provides a stream of [updates][Update] from a tree of [Workflow]s.
  *
  * Create these by injecting a [Factory] and calling [run][Factory.run].
  */
-interface WorkflowHost<out OutputT : Any, out RenderingT : Any> {
+interface WorkflowHost<in InputT : Any, out OutputT : Any, out RenderingT : Any> {
 
   /**
    * Output from a [WorkflowHost]. Emitted from [WorkflowHost.updates] after every compose pass.
@@ -62,88 +67,134 @@ interface WorkflowHost<out OutputT : Any, out RenderingT : Any> {
     /**
      * Creates a [WorkflowHost] to run [workflow].
      *
-     * The workflow's initial state is determined by passing [input] to
-     * [StatefulWorkflow.initialState].
+     * The workflow's initial state is determined by passing the first value emitted by [inputs] to
+     * [StatefulWorkflow.initialState]. Subsequent values emitted from [inputs] will be used to
+     * re-render the workflow.
      *
      * @param workflow The workflow to start.
-     * @param input Passed to [StatefulWorkflow.initialState] to determine the root workflow's
-     * initial state. If [InputT] is `Unit`, you can just omit this argument.
+     * @param inputs Passed to [StatefulWorkflow.initialState] to determine the root workflow's
+     * initial state, and to pass input updates to the root workflow.
+     * If [InputT] is `Unit`, you can just omit this argument.
      * @param snapshot If not null, used to restore the workflow.
      * @param context The [CoroutineContext] used to run the workflow tree. Added to the [Factory]'s
      * context.
      */
     fun <InputT : Any, OutputT : Any, RenderingT : Any> run(
       workflow: Workflow<InputT, OutputT, RenderingT>,
-      input: InputT,
+      inputs: ReceiveChannel<InputT>,
       snapshot: Snapshot? = null,
       context: CoroutineContext = EmptyCoroutineContext
-    ): WorkflowHost<OutputT, RenderingT> = run(workflow.id(), workflow, input, snapshot, context)
+    ): WorkflowHost<InputT, OutputT, RenderingT> =
+      object : WorkflowHost<InputT, OutputT, RenderingT> {
+        private val scope = CoroutineScope(context)
+
+        override val updates: ReceiveChannel<Update<OutputT, RenderingT>> =
+          scope.produce(capacity = 0) {
+            runWorkflowTree(
+                workflow = workflow.asStatefulWorkflow(),
+                inputs = inputs,
+                initialSnapshot = snapshot,
+                onUpdate = ::send
+            )
+          }
+      }
 
     fun <OutputT : Any, RenderingT : Any> run(
       workflow: Workflow<Unit, OutputT, RenderingT>,
       snapshot: Snapshot? = null,
       context: CoroutineContext = EmptyCoroutineContext
-    ): WorkflowHost<OutputT, RenderingT> = run(workflow.id(), workflow, Unit, snapshot, context)
+    ): WorkflowHost<Unit, OutputT, RenderingT> = run(workflow, channelOf(Unit), snapshot, context)
 
     /**
      * Creates a [WorkflowHost] that runs [workflow] starting from [initialState].
      *
      * **Don't call this directly.**
      *
-     * Instead, your module should have a test dependency on `pure-v2-testing` and you should call the
-     * testing extension method defined there on your workflow itself.
+     * Instead, your module should have a test dependency on `pure-v2-testing` and you should call
+     * the testing extension method defined there on your workflow itself.
      */
     @TestOnly
     fun <InputT : Any, StateT : Any, OutputT : Any, RenderingT : Any> runTestFromState(
       workflow: StatefulWorkflow<InputT, StateT, OutputT, RenderingT>,
       input: InputT,
       initialState: StateT
-    ): WorkflowHost<OutputT, RenderingT> {
-      val workflowId = workflow.id()
-      return object : WorkflowHost<OutputT, RenderingT> {
-        val node = WorkflowNode(workflowId, workflow, input, null, baseContext, initialState)
+    ): WorkflowHost<InputT, OutputT, RenderingT> =
+      object : WorkflowHost<InputT, OutputT, RenderingT> {
         override val updates: ReceiveChannel<Update<OutputT, RenderingT>> =
-          node.start(workflow, input)
+          GlobalScope.produce(capacity = 0, context = baseContext) {
+            runWorkflowTree(
+                workflow = workflow.asStatefulWorkflow(),
+                inputs = channelOf(input),
+                initialSnapshot = null,
+                initialState = initialState,
+                onUpdate = ::send
+            )
+          }
       }
-    }
 
-    internal fun <InputT : Any, OutputT : Any, RenderingT : Any> run(
-      id: WorkflowId<InputT, OutputT, RenderingT>,
-      workflow: Workflow<InputT, OutputT, RenderingT>,
-      input: InputT,
-      snapshot: Snapshot?,
-      context: CoroutineContext
-    ): WorkflowHost<OutputT, RenderingT> = object : WorkflowHost<OutputT, RenderingT> {
-      val node = WorkflowNode(
-          id = id,
-          workflow = workflow.asStatefulWorkflow(),
-          initialInput = input,
-          snapshot = snapshot,
-          baseContext = baseContext + context
-      )
-      override val updates: ReceiveChannel<Update<OutputT, RenderingT>> =
-        node.start(workflow.asStatefulWorkflow(), input)
-    }
+    private fun <T> channelOf(value: T) = Channel<T>(capacity = 1)
+        .apply { offer(value) }
   }
 }
 
 /**
- * Starts the coroutine that runs the coroutine loop.
+ * Loops forever, or until the coroutine is cancelled, processing the workflow tree and emitting
+ * updates by calling [onUpdate].
+ *
+ * This function is the lowest-level entry point into the runtime. Don't call this directly, instead
+ * use [WorkflowHost.Factory] to create a [WorkflowHost], or one of the stream operators for your
+ * favorite Rx library to map a stream of [InputT]s into [Update]s.
  */
-internal fun <I : Any, O : Any, R : Any> WorkflowNode<I, *, O, R>.start(
-  workflow: StatefulWorkflow<I, *, O, R>,
-  input: I
-): ReceiveChannel<Update<O, R>> = produce(capacity = 0) {
+@UseExperimental(InternalCoroutinesApi::class)
+suspend fun <InputT : Any, StateT : Any, OutputT : Any, RenderingT : Any> runWorkflowTree(
+  workflow: StatefulWorkflow<InputT, StateT, OutputT, RenderingT>,
+  inputs: ReceiveChannel<InputT>,
+  initialSnapshot: Snapshot?,
+  initialState: StateT? = null,
+  onUpdate: suspend (Update<OutputT, RenderingT>) -> Unit
+): Nothing {
+  var output: OutputT? = null
+  var input: InputT = inputs.receive()
+  var inputsClosed = false
+  val rootNode = WorkflowNode(
+      id = workflow.id(),
+      workflow = workflow,
+      initialInput = input,
+      snapshot = initialSnapshot,
+      baseContext = coroutineContext,
+      initialState = initialState
+  )
+
   try {
-    var output: O? = null
-    while (isActive) {
-      val rendering = compose(workflow, input)
-      val snapshot = snapshot(workflow)
-      send(Update(rendering, snapshot, output))
+    while (true) {
+      // Manually implement `ensureActive()` until we're on coroutines 1.2.x.
+      if (!isActive) throw coroutineContext[Job]!!.getCancellationException()
+
+      val rendering = rootNode.compose(workflow, input)
+      val snapshot = rootNode.snapshot(workflow)
+
+      onUpdate(Update(rendering, snapshot, output))
+
       // Tick _might_ return an output, but if it returns null, it means the state or a child
       // probably changed, so we should re-compose/snapshot and emit again.
       output = select {
-        tick(this) { it }
+        // Stop trying to read from the inputs channel after it's closed.
+        if (!inputsClosed) {
+          @Suppress("EXPERIMENTAL_API_USAGE")
+          inputs.onReceiveOrNull { newInput ->
+            if (newInput == null) {
+              inputsClosed = true
+            } else {
+              input = newInput
+            }
+            // No output. Returning from the select will go to the top of the loop to do another
+            // compose pass.
+            return@onReceiveOrNull null
+          }
+        }
+
+        // Tick the workflow tree.
+        rootNode.tick(this) { it }
       }
     }
   } catch (e: Throwable) {
@@ -154,9 +205,9 @@ internal fun <I : Any, O : Any, R : Any> WorkflowNode<I, *, O, R>.start(
     coroutineContext.cancel(e)
     throw e
   } finally {
-    // There's a potential race condition if the producer coroutine is cancelled before it has a chance
-    // to enter the try block, since we can't use CoroutineStart.ATOMIC. However, until we actually
-    // see this cause problems, I'm not too worried about it.
-    cancel()
+    // There's a potential race condition if the producer coroutine is cancelled before it has a
+    // chance to enter the try block, since we can't use CoroutineStart.ATOMIC. However, until we
+    // actually see this cause problems, I'm not too worried about it.
+    rootNode.cancel()
   }
 }

kotlin/workflow-rx2-runtime/README.md

@@ -0,0 +1,3 @@
+# workflow-rx2-runtime
+
+This module contains adapters to use the Workflow runtime with RxJava2.

kotlin/workflow-rx2-runtime/build.gradle

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2019 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.
+ */
+apply plugin: 'java-library'
+apply plugin: 'kotlin'
+apply plugin: 'com.vanniktech.maven.publish'
+apply plugin: 'org.jetbrains.dokka'
+
+sourceCompatibility = JavaVersion.VERSION_1_7
+targetCompatibility = JavaVersion.VERSION_1_7
+
+dokka rootProject.ext.defaultDokkaConfig
+
+dependencies {
+  compileOnly deps.annotations.intellij
+
+  api project(':workflow-runtime')
+  api deps.kotlin.stdLib.jdk6
+  api deps.kotlin.coroutines.core
+  api deps.rxjava2.rxjava2
+
+  implementation deps.kotlin.coroutines.rx2
+
+  testImplementation project(':workflow-testing')
+  testImplementation deps.kotlin.test.jdk
+}

kotlin/workflow-rx2-runtime/gradle.properties

@@ -0,0 +1,18 @@
+#
+# Copyright 2019 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.
+#
+POM_ARTIFACT_ID=workflow-rx2-runtime
+POM_NAME=Workflow RxJava2 Runtime
+POM_PACKAGING=jar