Introduce JobSupport.cancelCoroutine for AbstractCoroutine-derived classes

Author: elizarov

integration/kotlinx-coroutines-jdk8/test/examples/ExplicitJob-example.kt

@@ -26,7 +26,7 @@ fun main(args: Array<String>) {
         log("g should not execute this line")
     }
     log("Started futures f && g... will not wait -- cancel them!!!")
-    job.cancel(CancellationException("I don't want it"))
+    job.cancel()
     check(f.isCancelled)
     check(g.isCancelled)
     log("f result = ${Try<Unit> { f.get() }}")

kotlinx-coroutines-core/common/src/JobSupport.kt

@@ -584,6 +584,10 @@ public open class JobSupport constructor(active: Boolean) : Job, ChildJob, Paren
     public open fun childCancelled(cause: Throwable): Boolean =
         cancelImpl(cause) && handlesException
 
+    // For AbstractCoroutine implementations
+    protected fun cancelCoroutine(cause: Throwable?) =
+        cancelImpl(cause)
+
     // cause is Throwable or ParentJob when cancelChild was invoked
     // returns true is exception was handled, false otherwise
     private fun cancelImpl(cause: Any?): Boolean {

kotlinx-coroutines-core/common/src/Timeout.kt

@@ -91,7 +91,7 @@ private open class TimeoutCoroutine<U, in T: U>(
 
     @Suppress("LeakingThis", "Deprecation")
     override fun run() {
-        cancel(TimeoutCancellationException(time, this))
+        cancelCoroutine(TimeoutCancellationException(time, this))
     }
 
     @Suppress("UNCHECKED_CAST")

kotlinx-coroutines-core/common/src/channels/Broadcast.kt

@@ -98,7 +98,7 @@ private open class BroadcastCoroutine<E>(
     override fun cancel(cause: Throwable?): Boolean {
         val wasCancelled = _channel.cancel(cause)
         @Suppress("DEPRECATION")
-        if (wasCancelled) super.cancel(cause) // cancel the job
+        if (wasCancelled) cancelCoroutine(cause) // cancel the job
         return wasCancelled
     }
 

kotlinx-coroutines-core/common/src/channels/ChannelCoroutine.kt

@@ -23,7 +23,7 @@ internal open class ChannelCoroutine<E>(
 
     override fun cancel(cause: Throwable?): Boolean {
         val wasCancelled = _channel.cancel(cause)
-        if (wasCancelled) super.cancel(cause) // cancel the job
+        if (wasCancelled) cancelCoroutine(cause) // cancel the job
         return wasCancelled
     }
 }

kotlinx-coroutines-core/jvm/src/Builders.kt

@@ -75,7 +75,7 @@ private class BlockingCoroutine<T>(
             try {
                 while (true) {
                     @Suppress("DEPRECATION")
-                    if (Thread.interrupted()) throw InterruptedException().also { cancel(it) }
+                    if (Thread.interrupted()) throw InterruptedException().also { cancelCoroutine(it) }
                     val parkNanos = eventLoop?.processNextEvent() ?: Long.MAX_VALUE
                     // note: process next even may loose unpark flag, so check if completed before parking
                     if (isCompleted) break

reactive/kotlinx-coroutines-reactive/src/Publish.kt

@@ -72,7 +72,7 @@ private class PublisherCoroutine<in T>(
 
     override val isClosedForSend: Boolean get() = isCompleted
     override val isFull: Boolean = mutex.isLocked
-    override fun close(cause: Throwable?): Boolean = cancel(cause)
+    override fun close(cause: Throwable?): Boolean = cancelCoroutine(cause)
     override fun invokeOnClose(handler: (Throwable?) -> Unit) =
         throw UnsupportedOperationException("PublisherCoroutine doesn't support invokeOnClose")
 
@@ -134,7 +134,7 @@ private class PublisherCoroutine<in T>(
             // If onNext fails with exception, then we cancel coroutine (with this exception) and then rethrow it
             // to abort the corresponding send/offer invocation. From the standpoint of coroutines machinery,
             // this failure is essentially equivalent to a failure of a child coroutine.
-            childCancelled(e)
+            cancelCoroutine(e)
             unlockAndCheckCompleted()
             throw e
         }
@@ -201,7 +201,7 @@ private class PublisherCoroutine<in T>(
     override fun request(n: Long) {
         if (n <= 0) {
             // Specification requires IAE for n <= 0
-            cancel(IllegalArgumentException("non-positive subscription request $n"))
+            cancelCoroutine(IllegalArgumentException("non-positive subscription request $n"))
             return
         }
         while (true) { // lock-free loop for nRequested

reactive/kotlinx-coroutines-rx2/src/RxObservable.kt

@@ -65,7 +65,7 @@ private class RxObservableCoroutine<T: Any>(
 
     override val isClosedForSend: Boolean get() = isCompleted
     override val isFull: Boolean = mutex.isLocked
-    override fun close(cause: Throwable?): Boolean = cancel(cause)
+    override fun close(cause: Throwable?): Boolean = cancelCoroutine(cause)
     override fun invokeOnClose(handler: (Throwable?) -> Unit) =
         throw UnsupportedOperationException("RxObservableCoroutine doesn't support invokeOnClose")
 
@@ -113,7 +113,7 @@ private class RxObservableCoroutine<T: Any>(
             // If onNext fails with exception, then we cancel coroutine (with this exception) and then rethrow it
             // to abort the corresponding send/offer invocation. From the standpoint of coroutines machinery,
             // this failure is essentially equivalent to a failure of a child coroutine.
-            childCancelled(e)
+            cancelCoroutine(e)
             doLockedSignalCompleted()
             throw e
         }