-
Notifications
You must be signed in to change notification settings - Fork 80
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Workers: Simpler stateless helper workflows.
This should make it relatively easy to write network calls that need to inject services. The main problem is that the API is not very discoverable. It uses some syntactic magic to have the launcher act like a callable function. The workflow interface is a little too general for this – we need to have a state type and output some sort of state (can't use `Nothing`) but this type of workflow doesn't need to have any state emitted.
- Loading branch information
1 parent
dae10e6
commit dd81f57
Showing
7 changed files
with
337 additions
and
15 deletions.
There are no files selected for viewing
97 changes: 97 additions & 0 deletions
97
workflow-core/src/main/java/com/squareup/workflow/Worker.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,97 @@ | ||
| package com.squareup.workflow | ||
|
|
||
| import com.squareup.workflow.WorkflowPool.Id | ||
| import com.squareup.workflow.WorkflowPool.Type | ||
| import kotlinx.coroutines.experimental.Deferred | ||
|
|
||
| /** | ||
| * Defines a discrete task that a [Workflow] can execute asynchronously via [WorkflowPool], | ||
| * possibly across state changes. | ||
| * | ||
| * # Defining Workers | ||
| * | ||
| * There are a few ways to define a worker: | ||
| * - Call [worker] and pass a suspending block: | ||
| * ``` | ||
| * val echoWorker = worker { input -> | ||
| * withContext(IO) { | ||
| * echoService.echo(input) | ||
| * } | ||
| * } | ||
| * ``` | ||
| * | ||
| * - Convert a [Deferred] to a worker directly: | ||
| * ``` | ||
| * val deferredWorker = deferred.asWorker() | ||
| * ``` | ||
| * | ||
| * - Implement the `Worker` interface and implement [call]: | ||
| * ``` | ||
| * class EchoWorker( | ||
| * private val echoService: EchoService, | ||
| * private val ioContext: CoroutineContext | ||
| * ) : Worker<String, String> { | ||
| * override suspend fun call(input: String): String = | ||
| * withContext(ioContext) { | ||
| * echoService.echo(input) | ||
| * } | ||
| * } | ||
| * ``` | ||
| * | ||
| * # Running Workers | ||
| * | ||
| * To use, pass an instance of your [Worker] to [WorkflowPool.awaitWorkerResult] along with its | ||
| * input value. You may also pass an optional `name` `String` if you need to distinguish between | ||
| * multiple invocations. E.g.: | ||
| * | ||
| * ``` | ||
| * class MainReactor(private val echo: EchoWorker) : Reactor<…> { | ||
| * override suspend fun onReact( | ||
| * …, | ||
| * workflows: WorkflowPool | ||
| * ): Reaction<…, …> { | ||
| * … | ||
| * workflows.awaitWorkerResult(echo, "hello world") | ||
| * .let { result: String -> | ||
| * // React to result. | ||
| * } | ||
| * } | ||
| * } | ||
| * ``` | ||
| */ | ||
| interface Worker<in I : Any, out O : Any> { | ||
| /** | ||
| * Perform some asynchronous work. | ||
| */ | ||
| suspend fun call(input: I): O | ||
| } | ||
|
|
||
| /** | ||
| * Creates a [Worker] that passes [block] its input value and uses its return value as the result. | ||
| */ | ||
| fun <I : Any, O : Any> worker(block: suspend (I) -> O): Worker<I, O> = object : Worker<I, O> { | ||
| override suspend fun call(input: I): O = block(input) | ||
| } | ||
|
|
||
| /** | ||
| * Creates a [Worker] that will report the [Deferred]'s eventual value as its result. | ||
| */ | ||
| fun <T : Any> Deferred<T>.asWorker(): Worker<Unit, T> = worker { await() } | ||
|
|
||
| /** | ||
| * Uniquely identifies the [Worker] across the [WorkflowPool]. | ||
| * See [WorkflowPool.Type.makeId] for details. | ||
| */ | ||
| @Suppress("unused") | ||
| inline fun <reified I : Any, reified O : Any> Worker<I, O>.makeId( | ||
| name: String = "" | ||
| ): Id<I, Nothing, O> = workflowType.makeId(name) | ||
|
|
||
| /** | ||
| * Returns the [Type] of the [Worker] for the [WorkflowPool] | ||
| */ | ||
| @PublishedApi | ||
| @Suppress("unused") | ||
| internal inline val <reified I : Any, reified O : Any> | ||
| Worker<I, O>.workflowType: Type<I, Nothing, O> | ||
| get() = Type(I::class, Nothing::class, O::class) |
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
55 changes: 55 additions & 0 deletions
55
workflow-core/src/test/java/com/squareup/workflow/WorkerTest.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,55 @@ | ||
| package com.squareup.workflow | ||
|
|
||
| import kotlinx.coroutines.experimental.CancellationException | ||
| import kotlinx.coroutines.experimental.CompletableDeferred | ||
| import org.junit.Test | ||
| import java.io.IOException | ||
| import kotlin.test.assertEquals | ||
| import kotlin.test.assertFailsWith | ||
| import kotlin.test.assertFalse | ||
| import kotlin.test.assertTrue | ||
|
|
||
| class WorkerTest { | ||
|
|
||
| private val pool = WorkflowPool() | ||
| private val deferred = CompletableDeferred<Unit>() | ||
| private val worker = deferred.asWorker() | ||
|
|
||
| @Test fun whenCallSucceeds() { | ||
| val reaction = pool.workerResult(worker, Unit) | ||
| assertFalse(reaction.isCompleted) | ||
|
|
||
| deferred.complete(Unit) | ||
|
|
||
| assertEquals(Unit, reaction.getCompleted()) | ||
| } | ||
|
|
||
| @Test fun whenCallFails() { | ||
| val reaction = pool.workerResult(worker, Unit) | ||
| assertFalse(reaction.isCompleted) | ||
|
|
||
| deferred.completeExceptionally(IOException("network fail")) | ||
|
|
||
| val failure = reaction.getCompletionExceptionOrNull()!! | ||
| assertTrue(failure is ReactorException) | ||
| assertTrue(failure.cause is IOException) | ||
| } | ||
|
|
||
| @Test fun whenInternalCoroutineCancelled() { | ||
| val reaction = pool.workerResult(worker, Unit) | ||
| assertFalse(reaction.isCompleted) | ||
|
|
||
| deferred.cancel() | ||
|
|
||
| assertFailsWith<CancellationException> { reaction.getCompleted() } | ||
| } | ||
|
|
||
| @Test fun whenWorkflowCancelled() { | ||
| val reaction = pool.workerResult(worker, Unit) | ||
| assertFalse(reaction.isCompleted) | ||
|
|
||
| pool.abandonDelegate(worker.makeId()) | ||
|
|
||
| assertFailsWith<CancellationException> { reaction.getCompleted() } | ||
| } | ||
| } |
18 changes: 18 additions & 0 deletions
18
workflow-rx2/src/main/java/com/squareup/workflow/rx2/Workers.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,18 @@ | ||
| package com.squareup.workflow.rx2 | ||
|
|
||
| import com.squareup.workflow.Worker | ||
| import com.squareup.workflow.worker | ||
| import io.reactivex.Single | ||
| import kotlinx.coroutines.experimental.rx2.await | ||
|
|
||
| /** | ||
| * Creates a [Worker] that will pass its input value to [block], then subscribe to the returned | ||
| * [Single] and report the value it emits as the worker result. | ||
| */ | ||
| fun <I : Any, O : Any> singleWorker(block: (I) -> Single<O>): Worker<I, O> = | ||
| worker { block(it).await() } | ||
|
|
||
| /** | ||
| * Creates a [Worker] that will report the [Single]'s eventual value as its result. | ||
| */ | ||
| fun <T : Any> Single<T>.asWorker(): Worker<Unit, T> = worker { await() } |
Oops, something went wrong.