Support 'suspend' methods with cooperative cancellation via AbortController/AbortSignal.

Author: Gregory Lureau

compiler/build.gradle.kts

@@ -13,6 +13,7 @@ val kspVersion: String by project
 dependencies {
 
     implementation(project(":lib"))
+    implementation(project(":lib-coroutines"))
     implementation("com.squareup:kotlinpoet:1.10.2") {
         exclude(module = "kotlin-reflect")
     }

compiler/src/main/kotlin/deezer/kustomexport/compiler/js/CoroutinesExt.kt

@@ -25,5 +25,14 @@ import com.squareup.kotlinpoet.TypeName
 val coroutinesGlobalScope = ClassName("kotlinx.coroutines", "GlobalScope")
 val coroutinesPromiseFunc = MemberName("kotlinx.coroutines", "promise")
 val coroutinesAwait = MemberName("kotlinx.coroutines", "await")
+val coroutinesJob = ClassName("kotlinx.coroutines", "Job")
+
+//val coroutinesScope = ClassName("kotlinx.coroutines", "CoroutineScope")
+val coroutinesContext = MemberName("kotlin.coroutines", "coroutineContext")
+val coroutinesContextJob = MemberName("kotlinx.coroutines", "job")
+val coroutinesCancellationException = ClassName("kotlinx.coroutines", "CancellationException")
 val coroutinesPromise = ClassName("kotlin.js", "Promise")
+val abortController = ClassName("", "AbortController")
+val abortSignal = ClassName("", "AbortSignal")
+
 fun TypeName.asCoroutinesPromise() = coroutinesPromise.parameterizedBy(this)

compiler/src/main/kotlin/deezer/kustomexport/compiler/js/pattern/SharedTransformer.kt

@@ -25,8 +25,12 @@ import deezer.kustomexport.compiler.js.FormatString
 import deezer.kustomexport.compiler.js.FunctionDescriptor
 import deezer.kustomexport.compiler.js.MethodNameDisambiguation
 import deezer.kustomexport.compiler.js.PropertyDescriptor
+import deezer.kustomexport.compiler.js.abortController
+import deezer.kustomexport.compiler.js.abortSignal
 import deezer.kustomexport.compiler.js.asCoroutinesPromise
 import deezer.kustomexport.compiler.js.coroutinesAwait
+import deezer.kustomexport.compiler.js.coroutinesContext
+import deezer.kustomexport.compiler.js.coroutinesContextJob
 import deezer.kustomexport.compiler.js.coroutinesGlobalScope
 import deezer.kustomexport.compiler.js.coroutinesPromiseFunc
 import deezer.kustomexport.compiler.js.mapping.INDENTATION
@@ -65,14 +69,35 @@ fun FunctionDescriptor.buildWrappingFunction(
             fb.addParameter(param.name, param.type.exportedTypeName)
         }
     }
+    if (!import && isSuspend) {
+        fb.addParameter("abortSignal", abortSignal)
+    }
 
     if (body) {
-        if (!import && isSuspend)fb.addCode("return %T.%M·{\n", coroutinesGlobalScope, coroutinesPromiseFunc)
+        if (!import && isSuspend) {
+            fb.addCode("return %T.%M·{\n", coroutinesGlobalScope, coroutinesPromiseFunc)
+        }
+        if (import && isSuspend) {
+            fb.addStatement("val abortController = %T()", abortController)
+            fb.addStatement("val abortSignal = abortController.signal")
+            fb.addStatement(
+                "%M.%M.invokeOnCompletion { abortController.abort() }",
+                coroutinesContext,
+                coroutinesContextJob
+            )
+        }
+        if (!import && isSuspend) {
+            fb.addStatement("abortSignal.onabort = { %M.%M.cancel() }", coroutinesContext, coroutinesContextJob)
+        }
 
         val funcName = if (import) funExportedName else name
-        val params = parameters.fold(FormatString("")) { acc, item ->
+        var params = parameters.fold(FormatString("")) { acc, item ->
             acc + "$INDENTATION${item.name} = ".toFormatString() + item.portMethod(!import) + ",\n"
         }
+        if (import && isSuspend) {
+            params += FormatString("${INDENTATION}abortSignal = abortSignal")
+        }
+
         //TODO: Opti : could save the local "result" variable here
         fb.addCode(
             ("val result = $delegateName.$funcName(".toFormatString() +
@@ -86,6 +111,7 @@ fun FunctionDescriptor.buildWrappingFunction(
                 returnType.portMethod(import, "result".toFormatString()) +
                 (if (import && isSuspend) ".%M()".toFormatString(coroutinesAwait) else "".toFormatString())).asCode()
         )
+
         if (!import && isSuspend) fb.addCode("\n}")
     }
     return fb.build()

lib-coroutines/build.gradle.kts

@@ -0,0 +1,29 @@
+plugins {
+    kotlin("multiplatform")
+}
+
+kotlin {
+    js(IR) {
+        browser()
+    }
+    jvm()
+    ios()
+    iosSimulatorArm64()
+    tvos()
+    watchos()
+
+    sourceSets {
+        all {
+            languageSettings.optIn("kotlin.RequiresOptIn")
+            languageSettings.optIn("kotlin.js.ExperimentalJsExport")
+        }
+    }
+
+    targets.all {
+        compilations.all {
+            // Cannot enable rn due to native issue (stdlib included more than once)
+            // may be related to https://youtrack.jetbrains.com/issue/KT-46636
+            kotlinOptions.allWarningsAsErrors = false
+        }
+    }
+}

lib-coroutines/src/jsMain/kotlin/org/w3c/fetch/AbortController.kt

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2022 Deezer.
+ *
+ * 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 org.w3c.fetch // I want to rely on the external type, no need for namespace here
+
+// Inspired from https://github.com/JetBrains/kotlin-wrappers/blob/master/kotlin-browser/src/main/kotlin/org/w3c/fetch/AbortController.kt
+
+external class AbortController {
+    /**
+     * Returns the AbortSignal object associated with this object.
+     */
+    val signal: AbortSignal
+
+    /**
+     * Invoking this method will set this object's AbortSignal's aborted flag and signal to any observers that the associated activity is to be aborted.
+     */
+    fun abort()
+}
+
+/** A signal object that allows you to communicate with a DOM request (such as a Fetch) and abort it if required via an AbortController object. */
+external class AbortSignal {
+    /**
+     * Returns true if this AbortSignal's AbortController has signaled to abort, and false otherwise.
+     */
+    val aborted: Boolean
+    var onabort: ((event: dynamic) -> Unit)?
+}

samples/build.gradle.kts

@@ -27,6 +27,7 @@ kotlin {
 
             dependencies {
                 implementation(project(":lib"))
+                implementation(project(":lib-coroutines"))
                 implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.6.0")
             }
         }

samples/src/commonMain/kotlin/sample/coroutines/Coroutines.kt

@@ -18,31 +18,29 @@
 package sample.coroutines
 
 import deezer.kustomexport.KustomExport
-import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.async
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.ensureActive
-import kotlinx.coroutines.isActive
 import kotlinx.coroutines.job
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.withContext
 import kotlinx.coroutines.withTimeout
 import kotlin.coroutines.coroutineContext
-import kotlin.js.Date
 
 @KustomExport
 interface IComputer {
-    suspend fun longCompute(abortSignal: AbortSignal): Int
+    suspend fun longCompute(): Int
 }
 
 @KustomExport
 class Computer : IComputer {
-    override suspend fun longCompute(abortSignal: AbortSignal): Int {
+    var completed = false
+    override suspend fun longCompute(): Int {
         // listen abortSignal , if aborted => throw
         delay(1000)
+        completed = true
         return 42
     }
 }
@@ -51,31 +49,24 @@ class Computer : IComputer {
 class ComputerTester(private val computer: IComputer) {
     suspend fun testAsync(): Int {
         return withContext(Dispatchers.Unconfined) {
-            println(Date())
-            val task1 = async { computer.longCompute() } //coroutineContext.job::isActive) }
-            val task2 = async { computer.longCompute() } //coroutineContext.job::isActive) }
-
-            coroutineContext.job.invokeOnCompletion { error ->
-                if (error is CancellationException) {
-                    //ctrler.abort()
-                }
-            }
+            val task1 = async { computer.longCompute() }
+            val task2 = async { computer.longCompute() }
             return@withContext task1.await() + task2.await()
         }
     }
 
     suspend fun startAndCancelAfter(duration: Long) {
         withContext(Dispatchers.Unconfined) {
             withTimeout(duration) {
-                computer.longCompute()// { coroutineContext.job.isActive }
+                computer.longCompute()
             }
         }
     }
 
     suspend fun startCancellable(): Cancellable {
         val coroutineScope = CoroutineScope(Dispatchers.Unconfined + SupervisorJob())
         val job = coroutineScope.launch {
-            computer.longCompute() //coroutineContext.job::isActive)
+            computer.longCompute()
         }
 
         return object : Cancellable {

samples/src/commonMain/kotlin/sample/coroutines/Coroutines.ts

@@ -3,12 +3,14 @@ import { assert, assertEquals, assertQuiet, assertEqualsQuiet } from "../shared_
 import { sample } from '@kustom/Samples'
 
 runTest("Coroutines", async () : Promise<void> => {
-    console.log("start at " + Date())
+    var neverAbortController = new AbortController()
+    var neverAbortSignal = neverAbortController.signal
+
     var computer = new sample.coroutines.js.Computer()
-    var res = await computer.longCompute()
+    var res = await computer.longCompute(neverAbortSignal)
     assertEquals(42, res, "execute Kotlin coroutines")
 
-    var p = computer.longCompute()
+    var p = computer.longCompute(neverAbortSignal)
         .then((res) => {
             return res
         })
@@ -25,71 +27,96 @@ runTest("Coroutines", async () : Promise<void> => {
     }
 
     var tester = new sample.coroutines.js.ComputerTester(new MyComputer())
-    assertEquals(1998, await tester.testAsync(), "parallel setTimeout")
+    assertEquals(1998, await tester.testAsync(neverAbortSignal), "parallel setTimeout")
 
     class RejectComputer implements sample.coroutines.js.IComputer {
         async longCompute() {
             return new Promise<void>((resolve, reject) => {
                 setTimeout(() => {
-                    //reject(new Error("bug"));
-                    //reject(undefined); // null, undefined, <nothing> == resolve()
-                    resolve()
-                    //resolve(55)
+                    reject(new Error("bug"));
                 }, 1000);
               });
         }
     }
 
     var rejectTester = new sample.coroutines.js.ComputerTester(new RejectComputer())
-    await rejectTester.testAsync()
-    assertEquals(0, await rejectTester.testAsync(), "parallel reject")
+    try {
+        await rejectTester.testAsync(neverAbortSignal)
+    } catch(e) {
+        assert(e instanceof Error, "reject throws an Error")
+    }
+
+    cancelTypescriptPromiseFromKotlin()
+
+    cancelKotlinCoroutinesFromTypescript()
+})
 
+async function cancelTypescriptPromiseFromKotlin() {
+    var neverAbortController = new AbortController()
+    var neverAbortSignal = neverAbortController.signal
 
     class CancellableComputer implements sample.coroutines.js.IComputer {
+        workAborted: boolean
         // Should be passed in parameters in the longCompute method for cooperative cancellation
         // Cancellation is not available yet.
-        isJobActive() {
-            return true
-        }
-        async longCompute() {
+        async longCompute(abortSignal) {
             return new Promise((resolve, reject) => {
                 var timeout = 5000
                 var currTime = 0
-                this.rec(this.isJobActive, resolve, reject, currTime, timeout)
+                this.rec(abortSignal, resolve, reject, currTime, timeout)
               });
         }
-        rec(isJobActive, resolve, reject, currTime, timeout) {
+        rec(abortSignal, resolve, reject, currTime, timeout) {
             setTimeout(() => {
-                console.log("rec " + currTime + " / " + timeout)
                 if (currTime >= timeout) {
                     resolve(1234)
                 } else {
-                    if (isJobActive()) {
-                        this.rec(isJobActive, resolve, reject, currTime + 100, timeout)
+                    if (!abortSignal.aborted) {
+                        this.rec(abortSignal, resolve, reject, currTime + 100, timeout)
+                    } else {
+                        this.workAborted = true
                     }
                 }
             }, 100);
         }
     }
 
-    var cancellableTester = new sample.coroutines.js.ComputerTester(new CancellableComputer())
+    var cancellableComputer = new CancellableComputer()
+    var cancellableTester = new sample.coroutines.js.ComputerTester(cancellableComputer)
     try {
-        console.log(Date() + " startAndCancelAfter")
-        await cancellableTester.startAndCancelAfter(1000)
+        var abortController = new AbortController()
+        var abortSignal = neverAbortController.signal
+        await cancellableTester.startAndCancelAfter(1000, abortSignal)
+        assertQuiet(false, "TimeoutCancellationException should have been thrown")
     } catch (tce) {
         // this check is ok, but the Promise is not really cancelled, so not good enough...
-        // assert(tce.name === "TimeoutCancellationException", "can cancel (timeout)")
+        assert(tce.name === "TimeoutCancellationException", "can cancel (timeout throw CancellationException)")
+        // Exception is thrown immediately when the signal is emitted, work is really stopped a few ms after that
+        setTimeout(() => {
+            assertEquals(true, cancellableComputer.workAborted, "can cancel (work is aborted)")
+        }, 100);
     }
-    //assertEquals(91, await infiniteTester.testAsync(), "parallel reject")
+}
 
-    //console.log(await new RejectComputer().longCompute())
+async function cancelKotlinCoroutinesFromTypescript() {
+    var abortController = new AbortController()
+    var abortSignal = abortController.signal
+    var computer = new sample.coroutines.js.Computer()
+    var promise = computer.longCompute(abortSignal) // Trigger promise, duration = 1s
+    // Catch on promise is required, or else the test will stop before the end
+    var cancellationException = null
+    promise.catch((e) => {
+        cancellationException = e
+        //console.log("Catching the cancellation exception: " + e.name + " - " + e.message)
+    })
+
+    setTimeout(() => {
+        abortController.abort()
+    }, 100);
 
-    //var cancellableTester = new sample.coroutines.js.ComputerTester(new InfiniteComputer())
-    //var cancellable = cancellableTester.startCancellable()
-    /*
     setTimeout(() => {
-        console.log("time !")
-        //cancellable.cancel()
-    }, 500);
-    */
-})
+        assertEquals(false, computer.completed, "can cancel Kotlin coroutines from Typescript (work has been cancelled and will never complete)")
+        assertEqualsQuiet("JobCancellationException", cancellationException.name, "cancellation exception")
+        assertEqualsQuiet("DeferredCoroutine was cancelled", cancellationException.message, "cancellation exception")
+    }, 1200);
+}