MPP: withTimeout, delay in int number of milliseconds

Author: elizarov

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonDelay.kt

@@ -2,4 +2,4 @@ package kotlinx.coroutines.experimental
 
 public expect interface Delay
 
-public expect suspend fun delay(time: Long)
+public expect suspend fun delay(time: Int)

common/kotlinx-coroutines-core-common/src/main/kotlin/kotlinx/coroutines/experimental/CommonScheduled.kt

@@ -1,2 +1,7 @@
 package kotlinx.coroutines.experimental
 
+public expect suspend fun <T> withTimeout(time: Int, block: suspend CoroutineScope.() -> T): T
+
+public expect suspend fun <T> withTimeoutOrNull(time: Int, block: suspend CoroutineScope.() -> T): T?
+
+public expect class TimeoutCancellationException public constructor(message: String)

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt

@@ -84,13 +84,8 @@ public actual interface Delay {
  *
  * @param time time in milliseconds.
  */
-public actual suspend fun delay(time: Long) {
-    kotlin.require(time >= 0) { "Delay time $time cannot be negative" }
-    if (time <= 0) return // don't delay
-    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
-        cont.context.delay.scheduleResumeAfterDelay(time, TimeUnit.MILLISECONDS, cont)
-    }
-}
+public actual suspend fun delay(time: Int) =
+    delay(time.toLong(), TimeUnit.MILLISECONDS)
 
 /**
  * Delays coroutine for a given time without blocking a thread and resumes it after a specified time.

core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt

@@ -25,6 +25,26 @@ import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
 import kotlin.coroutines.experimental.intrinsics.startCoroutineUninterceptedOrReturn
 import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
 
+/**
+ * Runs a given suspending [block] of code inside a coroutine with a specified timeout and throws
+ * [TimeoutCancellationException] if timeout was exceeded.
+ *
+ * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
+ * cancellable suspending function inside the block throws [TimeoutCancellationException].
+ * Even if the code in the block suppresses [TimeoutCancellationException], it
+ * is still thrown by `withTimeout` invocation.
+ *
+ * The sibling function that does not throw exception on timeout is [withTimeoutOrNull].
+ * Note, that timeout action can be specified for [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
+ *
+ * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
+ * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
+ *
+ * @param time timeout time in milliseconds.
+ */
+public actual suspend fun <T> withTimeout(time: Int, block: suspend CoroutineScope.() -> T): T =
+    withTimeout(time.toLong(), TimeUnit.MILLISECONDS, block)
+
 /**
  * Runs a given suspending [block] of code inside a coroutine with a specified timeout and throws
  * [TimeoutCancellationException] if timeout was exceeded.
@@ -107,6 +127,26 @@ private open class TimeoutCoroutine<U, in T: U>(
         "${super.nameString()}($time $unit)"
 }
 
+/**
+ * Runs a given suspending block of code inside a coroutine with a specified timeout and returns
+ * `null` if this timeout was exceeded.
+ *
+ * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
+ * cancellable suspending function inside the block throws [TimeoutCancellationException].
+ * Even if the code in the block suppresses [TimeoutCancellationException], this
+ * invocation of `withTimeoutOrNull` still returns `null`.
+ *
+ * The sibling function that throws exception on timeout is [withTimeout].
+ * Note, that timeout action can be specified for [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
+ *
+ * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
+ * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
+ *
+ * @param time timeout time in milliseconds.
+ */
+public actual suspend fun <T> withTimeoutOrNull(time: Int, block: suspend CoroutineScope.() -> T): T? =
+    withTimeoutOrNull(time.toLong(), TimeUnit.MILLISECONDS, block)
+
 /**
  * Runs a given suspending block of code inside a coroutine with a specified timeout and returns
  * `null` if this timeout was exceeded.
@@ -161,14 +201,14 @@ private class TimeoutOrNullCoroutine<T>(
  * This exception is thrown by [withTimeout] to indicate timeout.
  */
 @Suppress("DEPRECATION")
-public class TimeoutCancellationException internal constructor(
+public actual class TimeoutCancellationException internal constructor(
     message: String,
     @JvmField internal val coroutine: Job?
 ) : TimeoutException(message) {
     /**
      * Creates timeout exception with a given message.
      */
-    public constructor(message: String) : this(message, null)
+    public actual constructor(message: String) : this(message, null)
 }
 
 /**

js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/CoroutineContext.kt

@@ -44,12 +44,12 @@ internal object DefaultExecutor : CoroutineDispatcher(), Delay {
         window.setTimeout({ block.run() }, 0)
     }
 
-    override fun scheduleResumeAfterDelay(time: Double, continuation: CancellableContinuation<Unit>) {
-        window.setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, time.timeToInt())
+    override fun scheduleResumeAfterDelay(time: Int, continuation: CancellableContinuation<Unit>) {
+        window.setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, time.coerceAtLeast(0))
     }
 
-    override fun invokeOnTimeout(time: Double, block: Runnable): DisposableHandle {
-        val handle = window.setTimeout({ block.run() }, time.timeToInt())
+    override fun invokeOnTimeout(time: Int, block: Runnable): DisposableHandle {
+        val handle = window.setTimeout({ block.run() }, time.coerceAtLeast(0))
         return object : DisposableHandle {
             override fun dispose() {
                 window.clearTimeout(handle)

js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Delay.kt

@@ -42,14 +42,14 @@ public actual interface Delay {
      * with(continuation) { resumeUndispatched(Unit) }
      * ```
      */
-    fun scheduleResumeAfterDelay(time: Double, continuation: CancellableContinuation<Unit>)
+    fun scheduleResumeAfterDelay(time: Int, continuation: CancellableContinuation<Unit>)
 
     /**
      * Schedules invocation of a specified [block] after a specified delay [time].
      * The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] of this invocation
      * request if it is not needed anymore.
      */
-    fun invokeOnTimeout(time: Double, block: Runnable): DisposableHandle
+    fun invokeOnTimeout(time: Int, block: Runnable): DisposableHandle
 }
 
 /**
@@ -63,12 +63,11 @@ public actual interface Delay {
  *
  * @param time time in milliseconds.
  */
-public actual suspend fun delay(time: Long) {
-    val dt = time.toDouble()
-    kotlin.require(dt >= 0) { "Delay time $time cannot be negative" }
-    if (dt <= 0) return // don't delay
+public actual suspend fun delay(time: Int) {
+    kotlin.require(time >= 0) { "Delay time $time cannot be negative" }
+    if (time <= 0) return // don't delay
     return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
-        cont.context.delay.scheduleResumeAfterDelay(dt, cont)
+        cont.context.delay.scheduleResumeAfterDelay(time, cont)
     }
 }
 

js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/EventLoop.kt

@@ -58,10 +58,10 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
     override fun dispatch(context: CoroutineContext, block: Runnable) =
         enqueue(block.toQueuedTask())
 
-    override fun scheduleResumeAfterDelay(time: Double, continuation: CancellableContinuation<Unit>) =
+    override fun scheduleResumeAfterDelay(time: Int, continuation: CancellableContinuation<Unit>) =
         schedule(DelayedResumeTask(time, continuation))
 
-    override fun invokeOnTimeout(time: Double, block: Runnable): DisposableHandle =
+    override fun invokeOnTimeout(time: Int, block: Runnable): DisposableHandle =
         DelayedRunnableTask(time, block).also { schedule(it) }
 
     override fun processNextEvent(): Double {
@@ -129,7 +129,7 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
     }
 
     internal abstract inner class DelayedTask(
-        delay: Double
+        delay: Int
     ) : QueuedTask(), Comparable<DelayedTask>, DisposableHandle, HeapNode {
         override var index: Int = -1
         var state = DELAYED
@@ -169,7 +169,7 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
     }
 
     private inner class DelayedResumeTask(
-        time: Double,
+        time: Int,
         private val cont: CancellableContinuation<Unit>
     ) : DelayedTask(time) {
         override fun run() {
@@ -178,7 +178,7 @@ internal abstract class EventLoopBase: CoroutineDispatcher(), Delay, EventLoop {
     }
 
     private inner class DelayedRunnableTask(
-        time: Double,
+        time: Int,
         private val block: Runnable
     ) : DelayedTask(time) {
         override fun run() { block.run() }

js/kotlinx-coroutines-core-js/src/main/kotlin/kotlinx/coroutines/experimental/Scheduled.kt

@@ -1,2 +1,143 @@
 package kotlinx.coroutines.experimental
 
+import kotlin.coroutines.experimental.Continuation
+import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED
+import kotlin.coroutines.experimental.intrinsics.startCoroutineUninterceptedOrReturn
+import kotlin.coroutines.experimental.intrinsics.suspendCoroutineOrReturn
+
+/**
+ * Runs a given suspending [block] of code inside a coroutine with a specified timeout and throws
+ * [TimeoutCancellationException] if timeout was exceeded.
+ *
+ * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
+ * cancellable suspending function inside the block throws [TimeoutCancellationException].
+ * Even if the code in the block suppresses [TimeoutCancellationException], it
+ * is still thrown by `withTimeout` invocation.
+ *
+ * The sibling function that does not throw exception on timeout is [withTimeoutOrNull].
+ *
+ * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
+ * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
+ *
+ * @param time timeout time in milliseconds.
+ */
+public actual suspend fun <T> withTimeout(time: Int, block: suspend CoroutineScope.() -> T): T {
+    require(time >= 0) { "Timeout time $time cannot be negative" }
+    if (time <= 0L) throw CancellationException("Timed out immediately")
+    return suspendCoroutineOrReturn { cont: Continuation<T> ->
+        setupTimeout(TimeoutCoroutine(time, cont), block)
+    }
+}
+
+private fun <U, T: U> setupTimeout(
+    coroutine: TimeoutCoroutine<U, T>,
+    block: suspend CoroutineScope.() -> T
+): Any? {
+    // schedule cancellation of this coroutine on time
+    val cont = coroutine.cont
+    val context = cont.context
+    coroutine.disposeOnCompletion(context.delay.invokeOnTimeout(coroutine.time, coroutine))
+    coroutine.initParentJob(context[Job])
+    // restart block using new coroutine with new job,
+    // however start it as undispatched coroutine, because we are already in the proper context
+    val result = try {
+        block.startCoroutineUninterceptedOrReturn(receiver = coroutine, completion = coroutine)
+    } catch (e: Throwable) {
+        JobSupport.CompletedExceptionally(e)
+    }
+    return when {
+        result == COROUTINE_SUSPENDED -> COROUTINE_SUSPENDED
+        coroutine.makeCompleting(result, MODE_IGNORE) -> {
+            if (result is JobSupport.CompletedExceptionally) throw result.exception else result
+        }
+        else -> COROUTINE_SUSPENDED
+    }
+}
+
+private open class TimeoutCoroutine<U, in T: U>(
+    val time: Int,
+    val cont: Continuation<U>
+) : AbstractCoroutine<T>(cont.context, active = true), Runnable, Continuation<T> {
+    override val defaultResumeMode: Int get() = MODE_DIRECT
+
+    @Suppress("LeakingThis")
+    override fun run() {
+        cancel(TimeoutCancellationException(time, this))
+    }
+
+    @Suppress("UNCHECKED_CAST")
+    override fun afterCompletion(state: Any?, mode: Int) {
+        if (state is CompletedExceptionally)
+            cont.resumeWithExceptionMode(state.exception, mode)
+        else
+            cont.resumeMode(state as T, mode)
+    }
+
+    override fun toString(): String  =
+        "TimeoutCoroutine($time)"
+}
+
+/**
+ * Runs a given suspending block of code inside a coroutine with a specified timeout and returns
+ * `null` if this timeout was exceeded.
+ *
+ * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
+ * cancellable suspending function inside the block throws [TimeoutCancellationException].
+ * Even if the code in the block suppresses [TimeoutCancellationException], this
+ * invocation of `withTimeoutOrNull` still returns `null`.
+ *
+ * The sibling function that throws exception on timeout is [withTimeout].
+ *
+ * This function delegates to [Delay.invokeOnTimeout] if the context [CoroutineDispatcher]
+ * implements [Delay] interface, otherwise it tracks time using a built-in single-threaded scheduled executor service.
+ *
+ * @param time timeout time in milliseconds.
+ */
+public actual suspend fun <T> withTimeoutOrNull(time: Int, block: suspend CoroutineScope.() -> T): T? {
+    require(time >= 0) { "Timeout time $time cannot be negative" }
+    if (time <= 0L) return null
+    return suspendCoroutineOrReturn { cont: Continuation<T?> ->
+        setupTimeout(TimeoutOrNullCoroutine(time, cont), block)
+    }
+}
+
+private class TimeoutOrNullCoroutine<T>(
+    time: Int,
+    cont: Continuation<T?>
+) : TimeoutCoroutine<T?, T>(time, cont) {
+    @Suppress("UNCHECKED_CAST")
+    override fun afterCompletion(state: Any?, mode: Int) {
+        if (state is CompletedExceptionally) {
+            val exception = state.exception
+            if (exception is TimeoutCancellationException && exception.coroutine === this)
+                cont.resumeMode(null, mode) else
+                cont.resumeWithExceptionMode(exception, mode)
+        } else
+            cont.resumeMode(state as T, mode)
+    }
+
+    override fun toString(): String  =
+        "TimeoutOrNullCoroutine($time)"
+}
+
+/**
+ * This exception is thrown by [withTimeout] to indicate timeout.
+ */
+@Suppress("DEPRECATION")
+public actual class TimeoutCancellationException internal constructor(
+    message: String,
+    internal val coroutine: Job?
+) : CancellationException(message) {
+    /**
+     * Creates timeout exception with a given message.
+     */
+    public actual constructor(message: String) : this(message, null)
+}
+
+@Suppress("FunctionName")
+private fun TimeoutCancellationException(
+    time: Int,
+    coroutine: Job
+) : TimeoutCancellationException = TimeoutCancellationException("Timed out waiting for $time", coroutine)
+
+