Merge pull request #423 from square/zachklipp/completable-asworker

Add Completable.asWorker() operator.

Author: zach-klippenstein

kotlin/workflow-core/src/main/java/com/squareup/workflow/Worker.kt

@@ -194,7 +194,8 @@ interface Worker<out T> {
      * Creates a [Worker] that just performs some side effects and doesn't emit anything.
      *
      * The returned [Worker] will equate to any other workers created with this function that have
-     * the same key.
+     * the same key. The key is required for this builder because there is no type information
+     * available to distinguish workers.
      *
      * E.g.:
      * ```

kotlin/workflow-rx2/src/main/java/com/squareup/workflow/rx2/RxWorkers.kt

@@ -18,6 +18,7 @@ package com.squareup.workflow.rx2
 import com.squareup.workflow.Worker
 import com.squareup.workflow.Worker.Emitter
 import com.squareup.workflow.emitAll
+import io.reactivex.Completable
 import io.reactivex.Flowable
 import io.reactivex.Maybe
 import io.reactivex.Observable
@@ -108,3 +109,15 @@ inline fun <reified T : Any> Single<out T?>.asWorker(key: String = ""): Worker<T
 // This !! works because RxJava types don't actually allow nulls, it's just that they can't
   // express that in their types because Java.
   Worker.from(key) { await()!! }
+
+/**
+ * Creates a [Worker] from this [Completable].
+ *
+ * The [Completable] will be subscribed to when the [Worker] is started, and disposed when it is
+ * cancelled.
+ *
+ * The key is required for this operator because there is no type information available to
+ * distinguish workers.
+ */
+fun Completable.asWorker(key: String) =
+  Worker.createSideEffect(key) { await() }

kotlin/workflow-rx2/src/test/java/com/squareup/workflow/rx2/RxWorkersTest.kt

@@ -21,6 +21,7 @@ import io.reactivex.Flowable
 import io.reactivex.Maybe
 import io.reactivex.Observable
 import io.reactivex.Single
+import io.reactivex.subjects.CompletableSubject
 import io.reactivex.subjects.MaybeSubject
 import io.reactivex.subjects.PublishSubject
 import io.reactivex.subjects.SingleSubject
@@ -310,4 +311,56 @@ class RxWorkersTest {
   }
 
   // endregion
+
+  // region Completable
+
+  @Test fun `completable emits`() {
+    val subject = CompletableSubject.create()
+    val worker = subject.asWorker("")
+
+    worker.test {
+      subject.onComplete()
+      assertFinished()
+    }
+  }
+
+  @Test fun `completable throws`() {
+    val subject = CompletableSubject.create()
+    val worker = subject.asWorker("")
+
+    worker.test {
+      subject.onError(ExpectedException())
+      assertTrue(getException() is ExpectedException)
+    }
+  }
+
+  @Test fun `completable is subscribed lazily`() {
+    var subscriptions = 0
+    val subject = CompletableSubject.create()
+    val worker = subject.doOnSubscribe { subscriptions++ }
+        .asWorker("")
+
+    assertEquals(0, subscriptions)
+
+    worker.test {
+      assertEquals(1, subscriptions)
+    }
+  }
+
+  @Test fun `completable is disposed when worker cancelled`() {
+    var cancels = 0
+    val subject = CompletableSubject.create()
+    val worker = subject.doOnDispose { cancels++ }
+        .asWorker("")
+
+    assertEquals(0, cancels)
+
+    worker.test {
+      assertEquals(0, cancels)
+      cancelWorker()
+      assertEquals(1, cancels)
+    }
+  }
+
+  // endregion
 }