WIP: Add the ability to update the input for a WorkflowHost.

Closes #247.

Author: zach-klippenstein

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/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-runtime/src/main/java/com/squareup/workflow/WorkflowHost.kt

@@ -19,25 +19,28 @@ 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.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.GlobalScope
 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.ensureActive
 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.
@@ -75,76 +78,118 @@ interface WorkflowHost<out OutputT : Any, out RenderingT : Any> {
      */
     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) {
+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 workflowNode = 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) {
+      coroutineContext.ensureActive()
+
+      val rendering = workflowNode.compose(workflow, input)
+      val snapshot = workflowNode.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.
+        workflowNode.tick(this) { it }
       }
     }
   } catch (e: Throwable) {
@@ -153,9 +198,9 @@ internal fun <I : Any, O : Any, R : Any> WorkflowNode<I, *, O, R>.start(
     coroutineContext.cancel(if (e is CancellationException) e else CancellationException(null, 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.
+    workflowNode.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

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

@@ -0,0 +1,83 @@
+/*
+ * 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.
+ */
+@file:Suppress("EXPERIMENTAL_API_USAGE")
+
+package com.squareup.workflow.rx2
+
+import com.squareup.workflow.Snapshot
+import com.squareup.workflow.Workflow
+import com.squareup.workflow.WorkflowHost.Update
+import com.squareup.workflow.runWorkflowTree
+import io.reactivex.Flowable
+import io.reactivex.Observable
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.channels.consume
+import kotlinx.coroutines.reactive.openSubscription
+import kotlinx.coroutines.rx2.rxFlowable
+import kotlin.DeprecationLevel.ERROR
+
+/**
+ * Given a stream of [InputT] values to use as inputs for the top-level [workflow], returns a
+ * [Flowable] that, when subscribed to, will start [workflow] and emit its [Update]s. When the
+ * subscription is disposed, the workflow will be terminated.
+ *
+ * The returned [Flowable] is _not_ multicasted – **every subscription will start a new workflow
+ * session.** It is recommended to use a multicasting operator on the resulting stream, such as
+ * [Flowable.replay], to share the updates from a single workflow session.
+ *
+ * The workflow's logic will run in whatever threading context the source [Flowable] is being
+ * observed on.
+ *
+ * This operator is an alternative to using
+ * [WorkflowHost.Factory][com.squareup.workflow.WorkflowHost.Factory] that is more convenient to
+ * use with a stream of inputs.
+ *
+ * This function operates on [Flowable] instead of [Observable] because the workflow runtime
+ * inherently supports backpressure. [Flowable] supports backpressure, [Observable] does not.
+ * RxJava provides operators to adapt between [Flowable]s and [Observable]s by explicitly specifying
+ * how to use backpressure. By operating on [Flowable], this operator leaves it up to the caller to
+ * specify strategies for handling backpressure, instead of assuming any particular behavior.
+ */
+fun <InputT : Any, OutputT : Any, RenderingT : Any> Flowable<InputT>.flatMapWorkflow(
+  workflow: Workflow<InputT, OutputT, RenderingT>,
+  initialSnapshot: Snapshot? = null
+): Flowable<Update<OutputT, RenderingT>> = GlobalScope.rxFlowable(
+    // We use the Unconfined dispatcher here because we want to execute on whatever Rx scheduler the
+    // source Flowable is being observed on.
+    context = Dispatchers.Unconfined
+) {
+  // Convert the input stream into a channel.
+  openSubscription().consume {
+    runWorkflowTree(
+        workflow = workflow.asStatefulWorkflow(),
+        inputs = this,
+        initialSnapshot = initialSnapshot,
+        onUpdate = ::send
+    )
+  }
+}
+
+/**
+ * Use [Flowable.flatMapWorkflow] instead. This function is just defined to be discoverable when
+ * working with [Observable]s instead of [Flowable]s.
+ *
+ * See [Observable.toFlowable] and [Flowable.toObservable] to convert between stream types.
+ */
+@Suppress("DeprecatedCallableAddReplaceWith", "unused")
+@Deprecated("Use Flowable.flatMapWorkflow instead.", level = ERROR)
+fun Observable<*>.flatMapWorkflow(): Nothing =
+  throw UnsupportedOperationException("Use Flowable.flatMapWorkflow instead.")

kotlin/workflow-rx2-runtime/src/test/java/com/squareup/workflow/rx2/FlatMapWorkflowTest.kt

@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+package com.squareup.workflow.rx2
+
+import kotlin.test.Test
+import kotlin.test.fail
+
+class FlatMapWorkflowTest {
+
+  @Test fun stuff() {
+    fail()
+  }
+}