Implement coroutines interop module

README.md

@@ -188,6 +188,9 @@ use D8 (Android Gradle Plugin 3.2+) or desugar as needed (depending on the build
 
 Kotlin extensions are bundled with almost every artifact.
 
+For coroutines - there is an `autodispose-coroutines-interop` artifact for interoperability between
+`CoroutineScope` and `ScopeProvider`/`Completable` types.
+
 ##### RxLifecycle
 
 As of 0.4.0 there is an RxLifecycle interop module under `autodispose-rxlifecycle`. This is for interop

autodispose-coroutines-interop/Module.md

@@ -0,0 +1,9 @@
+# Module autodispose-courtines-interop
+
+Extension functions to interop `ScopeProvider`/`Completable` and `CoroutineScope`, as well as
+`autoDispose(CoroutineScope)` extension functions on RxJava types.
+
+# Package com.uber.autodispose.coroutinesinterop
+
+Extension functions to interop `ScopeProvider`/`Completable` and `CoroutineScope`, as well as
+`autoDispose(CoroutineScope)` extension functions on RxJava types.

autodispose-coroutines-interop/build.gradle

@@ -0,0 +1,29 @@
+/*
+ * Copyright (C) 2019. Uber Technologies
+ *
+ * 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.
+ */
+
+plugins {
+  id 'ru.vyarus.animalsniffer'
+}
+
+dependencies {
+  api deps.rx.java
+  api project(':autodispose')
+  api deps.kotlin.coroutines
+
+  testImplementation project(':test-utils')
+}
+
+apply from: rootProject.file('gradle/gradle-mvn-push.gradle')

autodispose-coroutines-interop/gradle.properties

@@ -0,0 +1,20 @@
+#
+# Copyright (C) 2019. Uber Technologies
+#
+# 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.
+#
+
+POM_NAME=AutoDispose (Coroutines Interop)
+POM_ARTIFACT_ID=autodispose-coroutines-interop
+POM_PACKAGING=jar
+AUTOMATIC_MODULE_NAME=com.uber.autodispose.coroutinesinterop

autodispose-coroutines-interop/src/main/kotlin/com/uber/autodispose/coroutinesinterop/AutoDisposeCoroutinesInterop.kt

@@ -0,0 +1,118 @@
+/*
+ * Copyright (C) 2019. Uber Technologies
+ *
+ * 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", "unused")
+
+package com.uber.autodispose.coroutinesinterop
+
+import com.uber.autodispose.CompletableSubscribeProxy
+import com.uber.autodispose.FlowableSubscribeProxy
+import com.uber.autodispose.MaybeSubscribeProxy
+import com.uber.autodispose.ObservableSubscribeProxy
+import com.uber.autodispose.ScopeProvider
+import com.uber.autodispose.SingleSubscribeProxy
+import com.uber.autodispose.autoDispose
+import io.reactivex.Completable
+import io.reactivex.CompletableSource
+import io.reactivex.Flowable
+import io.reactivex.Maybe
+import io.reactivex.Observable
+import io.reactivex.Single
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlin.coroutines.CoroutineContext
+
+/** Extension that proxies to the normal [autoDispose] extension function with a [ScopeProvider]. */
+inline fun <T> Flowable<T>.autoDispose(scope: CoroutineScope): FlowableSubscribeProxy<T> {
+  return autoDispose(scope.asScopeProvider())
+}
+
+/** Extension that proxies to the normal [autoDispose] extension function with a [ScopeProvider]. */
+inline fun <T> Observable<T>.autoDispose(scope: CoroutineScope): ObservableSubscribeProxy<T> {
+  return autoDispose(scope.asScopeProvider())
+}
+
+/** Extension that proxies to the normal [autoDispose] extension function with a [ScopeProvider]. */
+inline fun <T> Single<T>.autoDispose(scope: CoroutineScope): SingleSubscribeProxy<T> {
+  return autoDispose(scope.asScopeProvider())
+}
+
+/** Extension that proxies to the normal [autoDispose] extension function with a [ScopeProvider]. */
+inline fun <T> Maybe<T>.autoDispose(scope: CoroutineScope): MaybeSubscribeProxy<T> {
+  return autoDispose(scope.asScopeProvider())
+}
+
+/** Extension that proxies to the normal [autoDispose] extension function with a [ScopeProvider]. */
+inline fun Completable.autoDispose(scope: CoroutineScope): CompletableSubscribeProxy {
+  return autoDispose(scope.asScopeProvider())
+}
+
+/**
+ * @return a [ScopeProvider] representation of this [CoroutineScope]. This scope will complete when
+ *         [this] coroutine scope completes.
+ */
+fun CoroutineScope.asScopeProvider(): ScopeProvider = ScopeProvider { asUndeferredCompletable() }
+
+/**
+ * @return a [Completable] representation of this [CoroutineScope]. This will complete when [this]
+ *         coroutine scope completes. Note that the returned [Completable] is deferred.
+ */
+fun CoroutineScope.asCompletable(): Completable {
+  return Completable.defer { asUndeferredCompletable() }
+}
+
+private fun CoroutineScope.asUndeferredCompletable(): Completable {
+  return Completable.create { emitter ->
+    val job = coroutineContext[Job] ?: error(
+        "Scope cannot be created because it does not have a job: ${this@asUndeferredCompletable}")
+    job.invokeOnCompletion {
+      when (it) {
+        null, is CancellationException -> emitter.onComplete()
+        else -> emitter.onError(it)
+      }
+    }
+  }
+}
+
+/**
+ * @param context an optional [CoroutineContext] to use for this scope. Default is a new
+ *                [SupervisorJob].
+ * @return a [CoroutineScope] representation of this [ScopeProvider]. This scope will cancel when
+ *         [this] scope provider completes.
+ */
+fun ScopeProvider.asCoroutineScope(context: CoroutineContext = SupervisorJob()): CoroutineScope {
+  return requestScope().asCoroutineScope(context)
+}
+
+/**
+ * @param context an optional [CoroutineContext] to use for this scope. Default is a new
+ *                [SupervisorJob].
+ * @return a [CoroutineScope] representation of this [CompletableSource]. This scope will cancel
+ *         when [this] scope provider completes.
+ */
+fun CompletableSource.asCoroutineScope(context: CoroutineContext = SupervisorJob()): CoroutineScope {
+  val scope = CoroutineScope(context)
+
+  // Bind to the scope, so if the scope is manually canceled before our scope provider emits, we
+  // clean up here.
+  Completable.wrap(this)
+      .autoDispose(scope)
+      .subscribe({ scope.cancel() }) { e -> scope.cancel("OnError", e) }
+
+  return scope
+}