WIP: Write a Flow operator that runs a Workflow.

Closes #263.

Author: zach-klippenstein

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

@@ -0,0 +1,51 @@
+/*
+ * 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
+
+import com.squareup.workflow.WorkflowHost.Update
+import com.squareup.workflow.internal.runWorkflowTree
+import kotlinx.coroutines.FlowPreview
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.produceIn
+
+/**
+ * Runs a [Workflow] by passing upstream items to it as input and emitting [Update]s as output.
+ * The returned [Flow] will not terminate until cancelled or the workflow throws an exception.
+ *
+ * Remember, [Flow]s are cold streams, so every subscriber to the resulting [Flow] will spin up
+ * a new instance of the tree.
+ *
+ * @param snapshot If non-null, used to restore the workflow tree from a previous [Snapshot] emitted
+ * in an [Update].
+ */
+@FlowPreview
+fun <InputT : Any, OutputT : Any, RenderingT : Any> Flow<InputT>.flatMapWorkflow(
+  workflow: Workflow<InputT, OutputT, RenderingT>,
+  snapshot: Snapshot? = null
+): Flow<Update<OutputT, RenderingT>> = flow {
+  coroutineScope {
+    val inputFlow = this@flatMapWorkflow
+    val inputs = inputFlow.produceIn(this)
+    runWorkflowTree(
+        workflow = workflow.asStatefulWorkflow(),
+        inputs = inputs,
+        initialSnapshot = snapshot,
+        onUpdate = ::emit
+    )
+  }
+}

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

@@ -26,7 +26,7 @@ internal class RealWorkflowHost<InputT : Any, StateT : Any, OutputT : Any, Rende
   workflow: StatefulWorkflow<InputT, StateT, OutputT, RenderingT>,
   context: CoroutineContext,
   initialInput: InputT,
-  private val initialSnapshot: Snapshot?,
+  initialSnapshot: Snapshot?,
   private val initialState: StateT? = null
 ) : WorkflowHost<InputT, OutputT, RenderingT> {
 
@@ -38,74 +38,80 @@ internal class RealWorkflowHost<InputT : Any, StateT : Any, OutputT : Any, Rende
     scope.produce(capacity = 0) {
       runWorkflowTree(
           workflow = workflow,
+          inputs = inputs,
+          initialSnapshot = initialSnapshot,
+          initialState = initialState,
           onUpdate = ::send
       )
     }
 
   override fun setInput(input: InputT) {
     inputs.offer(input)
   }
+}
 
-  /**
-   * Loops forever, or until the coroutine is cancelled, processing the workflow tree and emitting
-   * updates by calling [onUpdate].
-   */
-  private suspend fun runWorkflowTree(
-    workflow: StatefulWorkflow<InputT, StateT, OutputT, RenderingT>,
-    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
-    )
+/**
+ * Loops forever, or until the coroutine is cancelled, processing the workflow tree and emitting
+ * updates by calling [onUpdate].
+ */
+internal suspend fun <I : Any, S : Any, O : Any, R : Any> runWorkflowTree(
+  workflow: StatefulWorkflow<I, S, O, R>,
+  inputs: ReceiveChannel<I>,
+  initialSnapshot: Snapshot?,
+  initialState: S?,
+  onUpdate: suspend (Update<O, R>) -> Unit
+): Nothing {
+  var output: O? = null
+  var input: I = inputs.receive()
+  var inputsClosed = false
+  val workflowNode = WorkflowNode(
+      id = workflow.id(),
+      workflow = workflow,
+      initialInput = input,
+      snapshot = initialSnapshot,
+      baseContext = coroutineContext,
+      initialState = initialState
+  )
 
-    try {
-      while (true) {
-        coroutineContext.ensureActive()
+  try {
+    while (true) {
+      coroutineContext.ensureActive()
 
-        val rendering = workflowNode.compose(workflow, input)
-        val snapshot = workflowNode.snapshot(workflow)
+      val rendering = workflowNode.compose(workflow, input)
+      val snapshot = workflowNode.snapshot(workflow)
 
-        onUpdate(Update(rendering, snapshot, output))
+      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 {
-          // While the inputs channel is still open, select on it so we can detect new inputs.
-          if (!inputsClosed) {
-            @Suppress("EXPERIMENTAL_API_USAGE")
-            inputs.onReceiveOrNull { newInput ->
-              if (newInput == null) {
-                inputsClosed = true
-              } else {
-                input = newInput
-              }
-              // No output.
-              return@onReceiveOrNull null
+      // 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 {
+        // While the inputs channel is still open, select on it so we can detect new inputs.
+        if (!inputsClosed) {
+          @Suppress("EXPERIMENTAL_API_USAGE")
+          inputs.onReceiveOrNull { newInput ->
+            if (newInput == null) {
+              inputsClosed = true
+            } else {
+              input = newInput
             }
+            // No output.
+            return@onReceiveOrNull null
           }
-
-          // Tick the workflow tree.
-          workflowNode.tick(this) { it }
         }
+
+        // Tick the workflow tree.
+        workflowNode.tick(this) { it }
       }
-    } catch (e: Throwable) {
-      // For some reason the exception gets masked if we don't explicitly pass it to cancel the
-      // producer coroutine ourselves here.
-      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.
-      workflowNode.cancel()
     }
+  } catch (e: Throwable) {
+    // For some reason the exception gets masked if we don't explicitly pass it to cancel the
+    // producer coroutine ourselves here.
+    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.
+    workflowNode.cancel()
   }
 }