Make renderAsState public, make WorkflowContainer take a ViewEnvironment.

This narrows the scope of `WorkflowContainer` to be only for showing a workflow's renderings
using a `ViewEnvironment`. This elimnates the need to have separate overloads for workflows
with rendering type `ComposeRendering`, and is more idiomatic – the old `WorkflowContainer`
wasn't a "container" in any Compose-y sense of the word, and it was really weird that it took
a content function. Now this composable is actually a container for all workflow-related
composables, and so I think the name is actually appropriate (closes #22).

To render a workflow without a `ViewEnvironment`, `renderAsState` is now public. This is also
much more idiomatic, as it resembles APIs like `Flow<T>.collectAsState` and
`Observable.subscribeAsState`.

Author: zach-klippenstein

core-compose/src/main/java/com/squareup/workflow/ui/compose/RenderAsState.kt

@@ -0,0 +1,159 @@
+/*
+ * 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.
+ */
+@file:Suppress("NOTHING_TO_INLINE")
+
+package com.squareup.workflow.ui.compose
+
+import androidx.annotation.VisibleForTesting
+import androidx.compose.Composable
+import androidx.compose.Pivotal
+import androidx.compose.State
+import androidx.compose.onDispose
+import androidx.compose.remember
+import androidx.compose.state
+import androidx.ui.core.CoroutineContextAmbient
+import androidx.ui.savedinstancestate.Saver
+import androidx.ui.savedinstancestate.SaverScope
+import androidx.ui.savedinstancestate.UiSavedStateRegistryAmbient
+import androidx.ui.savedinstancestate.savedInstanceState
+import com.squareup.workflow.Snapshot
+import com.squareup.workflow.Workflow
+import com.squareup.workflow.diagnostic.WorkflowDiagnosticListener
+import com.squareup.workflow.launchWorkflowIn
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.distinctUntilChanged
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import okio.ByteString
+import kotlin.coroutines.CoroutineContext
+
+/**
+ * TODO write documentation
+ */
+@Composable
+fun <PropsT, OutputT : Any, RenderingT> Workflow<PropsT, OutputT, RenderingT>.renderAsState(
+  props: PropsT,
+  onOutput: (OutputT) -> Unit,
+  diagnosticListener: WorkflowDiagnosticListener? = null
+): State<RenderingT> {
+  @Suppress("DEPRECATION")
+  val coroutineContext = CoroutineContextAmbient.current + Dispatchers.Main.immediate
+  return renderAsStateImpl(this, props, onOutput, diagnosticListener, coroutineContext)
+}
+
+/**
+ * TODO write documentation
+ */
+@Composable
+inline fun <OutputT : Any, RenderingT> Workflow<Unit, OutputT, RenderingT>.renderAsState(
+  noinline onOutput: (OutputT) -> Unit,
+  diagnosticListener: WorkflowDiagnosticListener? = null
+): State<RenderingT> = renderAsState(Unit, onOutput, diagnosticListener)
+
+/**
+ * TODO write documentation
+ */
+@Composable
+inline fun <PropsT, RenderingT> Workflow<PropsT, Nothing, RenderingT>.renderAsState(
+  props: PropsT,
+  diagnosticListener: WorkflowDiagnosticListener? = null
+): State<RenderingT> = renderAsState(props, {}, diagnosticListener)
+
+/**
+ * TODO write documentation
+ */
+@Composable
+inline fun <RenderingT> Workflow<Unit, Nothing, RenderingT>.renderAsState(
+  diagnosticListener: WorkflowDiagnosticListener? = null
+): State<RenderingT> = renderAsState(Unit, {}, diagnosticListener)
+
+/**
+ * @param snapshotKey Allows tests to pass in a custom key to use to save/restore the snapshot from
+ * the [UiSavedStateRegistryAmbient]. If null, will use the default key based on source location.
+ */
+@Suppress("EXPERIMENTAL_API_USAGE")
+@VisibleForTesting(otherwise = VisibleForTesting.PRIVATE)
+@Composable internal fun <PropsT, OutputT : Any, RenderingT> renderAsStateImpl(
+  @Pivotal workflow: Workflow<PropsT, OutputT, RenderingT>,
+  props: PropsT,
+  onOutput: (OutputT) -> Unit,
+  @Pivotal diagnosticListener: WorkflowDiagnosticListener?,
+  @Pivotal coroutineContext: CoroutineContext,
+  snapshotKey: String? = null
+): State<RenderingT> {
+  // This can be a StateFlow once coroutines is upgraded to 1.3.6.
+  val propsChannel = remember { Channel<PropsT>(capacity = Channel.CONFLATED) }
+  propsChannel.offer(props)
+
+  // Need a mutable holder for onOutput so the outputs subscriber created in the onActive block
+  // will always be able to see the latest value.
+  val outputCallback = remember { OutputCallback(onOutput) }
+  outputCallback.onOutput = onOutput
+
+  val renderingState = state<RenderingT?> { null }
+  val snapshotState = savedInstanceState(key = snapshotKey, saver = SnapshotSaver) { null }
+
+  // We can't use onActive/on(Pre)Commit because they won't run their callback until after this
+  // function returns, and we need to run this immediately so we get the rendering synchronously.
+  val workflowScope = remember {
+    val coroutineScope = CoroutineScope(coroutineContext + Dispatchers.Main.immediate)
+    val propsFlow = propsChannel.consumeAsFlow()
+        .distinctUntilChanged()
+
+    launchWorkflowIn(coroutineScope, workflow, propsFlow, snapshotState.value) { session ->
+      session.diagnosticListener = diagnosticListener
+
+      // Don't call onOutput directly, since out captured reference won't be changed if the
+      // if a different argument is passed to observeWorkflow.
+      session.outputs.onEach { outputCallback.onOutput(it) }
+          .launchIn(this)
+
+      session.renderingsAndSnapshots
+          .onEach { (rendering, snapshot) ->
+            renderingState.value = rendering
+            snapshotState.value = snapshot
+          }
+          .launchIn(this)
+    }
+
+    return@remember coroutineScope
+  }
+
+  onDispose {
+    workflowScope.cancel()
+  }
+
+  // The value is guaranteed to be set before returning, so this cast is fine.
+  @Suppress("UNCHECKED_CAST")
+  return renderingState as State<RenderingT>
+}
+
+private object SnapshotSaver : Saver<Snapshot?, ByteArray> {
+  override fun SaverScope.save(value: Snapshot?): ByteArray {
+    return value?.bytes?.toByteArray() ?: ByteArray(0)
+  }
+
+  override fun restore(value: ByteArray): Snapshot? {
+    return value.takeUnless { it.isEmpty() }
+        ?.let { bytes -> Snapshot.of(ByteString.of(*bytes)) }
+  }
+}
+
+private class OutputCallback<OutputT>(var onOutput: (OutputT) -> Unit)