WIP Reimplemented Workers using side effects.

This is the Kotlin half of square/workflow#1021.

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

@@ -101,19 +101,6 @@ interface RenderContext<StateT, in OutputT> {
     handler: (ChildOutputT) -> WorkflowAction<StateT, OutputT>
   ): ChildRenderingT
 
-  /**
-   * Ensures [worker] is running. When the [Worker] emits an output, [handler] is called
-   * to determine the [WorkflowAction] to take. When the worker finishes, nothing happens (although
-   * another render pass may be triggered).
-   *
-   * @param key An optional string key that is used to distinguish between identical [Worker]s.
-   */
-  fun <T> runningWorker(
-    worker: Worker<T>,
-    key: String = "",
-    handler: (T) -> WorkflowAction<StateT, OutputT>
-  )
-
   /**
    * Ensures [sideEffect] is running with the given [key].
    *
@@ -195,6 +182,22 @@ fun <StateT, OutputT> RenderContext<StateT, OutputT>.runningWorker(
   }
 }
 
+/**
+ * Ensures [worker] is running. When the [Worker] emits an output, [handler] is called
+ * to determine the [WorkflowAction] to take. When the worker finishes, nothing happens (although
+ * another render pass may be triggered).
+ *
+ * @param key An optional string key that is used to distinguish between identical [Worker]s.
+ */
+fun <T, StateT, OutputT : Any> RenderContext<StateT, OutputT>.runningWorker(
+  worker: Worker<T>,
+  key: String = "",
+  handler: (T) -> WorkflowAction<StateT, OutputT>
+) {
+  val workerWorkflow = WorkerWorkflow<T>()
+  renderChild(workerWorkflow, props = worker, key = key, handler = handler)
+}
+
 /**
  * Alternative to [RenderContext.actionSink] that allows externally defined
  * event types to be mapped to anonymous [WorkflowAction]s.

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

@@ -0,0 +1,36 @@
+package com.squareup.workflow
+
+/**
+ * TODO write documentation
+ */
+// TODO remove upper bound
+@OptIn(ExperimentalWorkflowApi::class)
+internal class WorkerWorkflow<OutputT>(
+) : StatefulWorkflow<Worker<OutputT>, Int, OutputT, Unit>(), ImpostorWorkflow {
+
+  override fun initialState(
+    props: Worker<OutputT>,
+    snapshot: Snapshot?
+  ): Int = 0
+
+  override fun onPropsChanged(
+    old: Worker<OutputT>,
+    new: Worker<OutputT>,
+    state: Int
+  ): Int = if (!old.doesSameWorkAs(new)) state + 1 else state
+
+  override fun render(
+    props: Worker<OutputT>,
+    state: Int,
+    context: RenderContext<Int, OutputT>
+  ) {
+    context.runningSideEffect(state.toString()) {
+      props.run()
+          .collectToSink(context.actionSink) { output ->
+            action { setOutput(output) }
+          }
+    }
+  }
+
+  override fun snapshotState(state: Int): Snapshot = Snapshot.EMPTY
+}