-
Notifications
You must be signed in to change notification settings - Fork 70
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Replace WorkflowHost with the runWorkflow function.
Closes #439.
- Loading branch information
1 parent
8786269
commit 950136c
Showing
12 changed files
with
910 additions
and
851 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
178 changes: 178 additions & 0 deletions
178
kotlin/workflow-runtime/src/main/java/com/squareup/workflow/LaunchWorkflow.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,178 @@ | ||
| /* | ||
| * 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.internal.RealWorkflowLoop | ||
| import com.squareup.workflow.internal.WorkflowLoop | ||
| import com.squareup.workflow.internal.unwrapCancellationCause | ||
| import kotlinx.coroutines.CoroutineScope | ||
| import kotlinx.coroutines.ExperimentalCoroutinesApi | ||
| import kotlinx.coroutines.FlowPreview | ||
| import kotlinx.coroutines.Job | ||
| import kotlinx.coroutines.channels.BroadcastChannel | ||
| import kotlinx.coroutines.channels.ConflatedBroadcastChannel | ||
| import kotlinx.coroutines.flow.Flow | ||
| import kotlinx.coroutines.flow.asFlow | ||
| import kotlinx.coroutines.launch | ||
| import kotlinx.coroutines.plus | ||
| import org.jetbrains.annotations.TestOnly | ||
|
|
||
| /** | ||
| * Don't use this typealias for the public API, better to just use the function directly so it's | ||
| * more obvious how to use it. | ||
| */ | ||
| @UseExperimental(ExperimentalCoroutinesApi::class) | ||
| internal typealias Configurator <O, R, T> = CoroutineScope.( | ||
| renderingsAndSnapshots: Flow<RenderingAndSnapshot<R>>, | ||
| outputs: Flow<O> | ||
| ) -> T | ||
|
|
||
| /** | ||
| * Launches the [workflow] in a new coroutine in [scope]. The workflow tree is seeded with | ||
| * [initialSnapshot] and the first value emitted by [inputs]. Subsequent values emitted from | ||
| * [inputs] will be used to re-render the workflow. | ||
| * | ||
| * ## Initialization | ||
| * | ||
| * Before starting the actual workflow runtime, this function will invoke [beforeStart] and pass | ||
| * it the [Flow]s of renderings, snapshots, and outputs, as well as a [CoroutineScope] that the | ||
| * runtime will be hosted in. The workflow runtime will not be started until after this function | ||
| * returns. This is to allow the output flow to start being collected before any outputs can | ||
| * actually be emitted. Collectors that start _after_ [beforeStart] returns may not receive outputs | ||
| * that are emitted very quickly. The value returned by [beforeStart] will be returned from this | ||
| * function after the runtime is launched. The [CoroutineScope] passed to [beforeStart] can be used | ||
| * to do more than just cancel the runtime – it can also be used to start coroutines that will run | ||
| * until the workflow is cancelled, namely to collect the rendering and output [Flow]s. | ||
| * | ||
| * ## Scoping | ||
| * | ||
| * The workflow runtime makes use of | ||
| * [structured concurrency](https://medium.com/@elizarov/structured-concurrency-722d765aa952). | ||
| * The runtime is started in a specific [CoroutineScope], which defines the context for the entire | ||
| * workflow tree – most importantly, the [Job] that governs the runtime's lifetime and exception | ||
| * reporting path, and the [CoroutineDispatcher][kotlinx.coroutines.CoroutineDispatcher] that | ||
| * decides what on what threads to run workflow code. | ||
| * | ||
| * This function creates a child [Job] in [scope] and uses it as the parent for the workflow | ||
| * runtime, and as the job passed to the [beforeStart] function. This means that if [beforeStart] | ||
| * calls `CoroutineScope.cancel`, it will cancel the workflow runtime, but will not cancel the | ||
| * [scope] passed into this function. | ||
| * | ||
| * @param scope The [CoroutineScope] in which to launch the workflow runtime. Any exceptions thrown | ||
| * in any workflows will be reported to this scope, and cancelling this scope will cancel the | ||
| * workflow runtime. | ||
| * @param workflow The root workflow to start. | ||
| * @param inputs Stream of input values for the root workflow. The first value emitted is passed to | ||
| * the root workflow's [StatefulWorkflow.initialState], and subsequent emissions are passed as | ||
| * input updates to the root workflow. | ||
| * @param initialSnapshot If not null, used to restore the workflow. | ||
| * @param beforeStart Called exactly once with the flows for renderings/snapshots and outputs. | ||
| * Note that the receiver scope will already be cancelled, and the flows will complete immediately | ||
| * if this function was invoked from a cancelled [scope]. | ||
| * | ||
| * @return The value returned by [beforeStart]. | ||
| */ | ||
| @UseExperimental(ExperimentalCoroutinesApi::class, FlowPreview::class) | ||
| fun <InputT, OutputT : Any, RenderingT, ResultT> launchWorkflowIn( | ||
| scope: CoroutineScope, | ||
| workflow: Workflow<InputT, OutputT, RenderingT>, | ||
| inputs: Flow<InputT>, | ||
| initialSnapshot: Snapshot? = null, | ||
| beforeStart: CoroutineScope.( | ||
| renderingsAndSnapshots: Flow<RenderingAndSnapshot<RenderingT>>, | ||
| outputs: Flow<OutputT> | ||
| ) -> ResultT | ||
| ): ResultT = launchWorkflowImpl( | ||
| scope, | ||
| RealWorkflowLoop, | ||
| workflow.asStatefulWorkflow(), | ||
| inputs, | ||
| initialSnapshot = initialSnapshot, | ||
| initialState = null, | ||
| beforeStart = beforeStart | ||
| ) | ||
|
|
||
| /** | ||
| * Launches the [workflow] in a new coroutine in [scope]. The workflow tree is seeded with | ||
| * [initialState] and the first value emitted by [inputs]. Subsequent values emitted from | ||
| * [inputs] will be used to re-render the workflow. | ||
| * | ||
| * See [launchWorkflowIn] for documentation about most of the parameters and behavior. | ||
| * | ||
| * @param initialState The [StateT] in which to start the root workflow. | ||
| */ | ||
| @TestOnly | ||
| @UseExperimental(ExperimentalCoroutinesApi::class, FlowPreview::class) | ||
| fun <InputT, StateT, OutputT : Any, RenderingT, ResultT> launchWorkflowForTestFromStateIn( | ||
| scope: CoroutineScope, | ||
| workflow: StatefulWorkflow<InputT, StateT, OutputT, RenderingT>, | ||
| inputs: Flow<InputT>, | ||
| initialState: StateT, | ||
| beforeStart: CoroutineScope.( | ||
| renderingsAndSnapshots: Flow<RenderingAndSnapshot<RenderingT>>, | ||
| outputs: Flow<OutputT> | ||
| ) -> ResultT | ||
| ): ResultT = launchWorkflowImpl( | ||
| scope, | ||
| RealWorkflowLoop, | ||
| workflow, | ||
| inputs, | ||
| initialState = initialState, | ||
| initialSnapshot = null, | ||
| beforeStart = beforeStart | ||
| ) | ||
|
|
||
| @UseExperimental(ExperimentalCoroutinesApi::class, FlowPreview::class) | ||
| internal fun <InputT, StateT, OutputT : Any, RenderingT, ResultT> launchWorkflowImpl( | ||
| scope: CoroutineScope, | ||
| workflowLoop: WorkflowLoop, | ||
| workflow: StatefulWorkflow<InputT, StateT, OutputT, RenderingT>, | ||
| inputs: Flow<InputT>, | ||
| initialSnapshot: Snapshot?, | ||
| initialState: StateT?, | ||
| beforeStart: Configurator<OutputT, RenderingT, ResultT> | ||
| ): ResultT { | ||
| val renderingsAndSnapshots = ConflatedBroadcastChannel<RenderingAndSnapshot<RenderingT>>() | ||
| val outputs = BroadcastChannel<OutputT>(capacity = 1) | ||
| val workflowJob = Job(parent = scope.coroutineContext[Job]) | ||
| val workflowScope = scope + workflowJob | ||
|
|
||
| // Ensure we close the channels when we're done, so that they propagate errors. | ||
| workflowJob.invokeOnCompletion { cause -> | ||
| // We need to unwrap the cancellation exception so that we *complete* the channels instead | ||
| // of cancelling them if our coroutine was merely cancelled. | ||
| val realCause = cause?.unwrapCancellationCause() | ||
| renderingsAndSnapshots.close(realCause) | ||
| outputs.close(realCause) | ||
| } | ||
|
|
||
| // Give the caller a chance to start collecting outputs. | ||
| val result = beforeStart(workflowScope, renderingsAndSnapshots.asFlow(), outputs.asFlow()) | ||
|
|
||
| workflowScope.launch { | ||
| // Run the workflow processing loop forever, or until it fails or is cancelled. | ||
| workflowLoop.runWorkflowLoop( | ||
| workflow, | ||
| inputs, | ||
| initialSnapshot = initialSnapshot, | ||
| initialState = initialState, | ||
| onRendering = renderingsAndSnapshots::send, | ||
| onOutput = outputs::send | ||
| ) | ||
| } | ||
|
|
||
| return result | ||
| } |
Oops, something went wrong.