kotlin.time.Duration support

Fixes Kotlin#1402

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

@@ -6,6 +6,8 @@ package kotlinx.coroutines
 
 import kotlinx.coroutines.selects.*
 import kotlin.coroutines.*
+import kotlin.time.Duration
+import kotlin.time.ExperimentalTime
 
 /**
  * This dispatcher _feature_ is implemented by [CoroutineDispatcher] implementations that natively support
@@ -75,5 +77,18 @@ public suspend fun delay(timeMillis: Long) {
     }
 }
 
+/**
+ * Delays coroutine for a given [duration] without blocking a thread and resumes it after the specified time.
+ * This suspending function is cancellable.
+ * If the [Job] of the current coroutine is cancelled or completed while this suspending function is waiting, this function
+ * immediately resumes with [CancellationException].
+ *
+ * Note that delay can be used in [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
+ *
+ * Implementation note: how exactly time is tracked is an implementation detail of [CoroutineDispatcher] in the context.
+ */
+@ExperimentalTime
+public suspend fun delay(duration: Duration) = delay(duration.toLongMilliseconds())
+
 /** Returns [Delay] implementation of the given context */
 internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultDelay

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

@@ -10,6 +10,8 @@ import kotlinx.coroutines.selects.*
 import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.*
 import kotlin.jvm.*
+import kotlin.time.Duration
+import kotlin.time.ExperimentalTime
 
 /**
  * Runs a given suspending [block] of code inside a coroutine with a specified [timeout][timeMillis] and throws
@@ -32,6 +34,22 @@ public suspend fun <T> withTimeout(timeMillis: Long, block: suspend CoroutineSco
     }
 }
 
+/**
+ * Runs a given suspending [block] of code inside a coroutine with the specified [timeout] and throws
+ * a [TimeoutCancellationException] if the timeout was exceeded.
+ *
+ * The code that is executing inside the [block] is cancelled on timeout and the active or next invocation of
+ * the cancellable suspending function inside the block throws a [TimeoutCancellationException].
+ *
+ * The sibling function that does not throw an exception on timeout is [withTimeoutOrNull].
+ * Note that the timeout action can be specified for a [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
+ *
+ * Implementation note: how the time is tracked exactly is an implementation detail of the context's [CoroutineDispatcher].
+ */
+@ExperimentalTime
+public suspend fun <T> withTimeout(timeout: Duration, block: suspend CoroutineScope.() -> T): T =
+        withTimeout(timeout.toLongMilliseconds(), block)
+
 /**
  * Runs a given suspending block of code inside a coroutine with a specified [timeout][timeMillis] and returns
  * `null` if this timeout was exceeded.
@@ -65,6 +83,22 @@ public suspend fun <T> withTimeoutOrNull(timeMillis: Long, block: suspend Corout
     }
 }
 
+/**
+ * Runs a given suspending block of code inside a coroutine with the 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 a [TimeoutCancellationException].
+ *
+ * The sibling function that throws an exception on timeout is [withTimeout].
+ * Note that the timeout action can be specified for a [select] invocation with [onTimeout][SelectBuilder.onTimeout] clause.
+ *
+ * Implementation note: how the time is tracked exactly is an implementation detail of the context's [CoroutineDispatcher].
+ */
+@ExperimentalTime
+public suspend fun <T> withTimeoutOrNull(timeout: Duration, block: suspend CoroutineScope.() -> T): T? =
+        withTimeoutOrNull(timeout.toLongMilliseconds(), block)
+
 private fun <U, T: U> setupTimeout(
     coroutine: TimeoutCoroutine<U, T>,
     block: suspend CoroutineScope.() -> T

kotlinx-coroutines-core/common/src/selects/Select.kt

@@ -14,6 +14,8 @@ import kotlin.coroutines.*
 import kotlin.coroutines.intrinsics.*
 import kotlin.jvm.*
 import kotlin.native.concurrent.*
+import kotlin.time.Duration
+import kotlin.time.ExperimentalTime
 
 /**
  * Scope for [select] invocation.
@@ -52,6 +54,18 @@ public interface SelectBuilder<in R> {
     public fun onTimeout(timeMillis: Long, block: suspend () -> R)
 }
 
+
+/**
+ * Clause that selects the given [block] after the specified [timeout] passes.
+ * If timeout is negative or zero, [block] is selected immediately.
+ *
+ * **Note: This is an experimental api.** It may be replaced with light-weight timer/timeout channels in the future.
+ */
+@ExperimentalCoroutinesApi
+@ExperimentalTime
+public fun <R> SelectBuilder<R>.onTimeout(timeout: Duration, block: suspend () -> R) =
+        onTimeout(timeout.toLongMilliseconds(), block)
+
 /**
  * Clause for [select] expression without additional parameters that does not select any value.
  */

kotlinx-coroutines-core/common/test/DelayDurationTest.kt

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+@file:Suppress("NAMED_ARGUMENTS_NOT_ALLOWED", "DEPRECATION")
+
+// KT-21913
+
+package kotlinx.coroutines
+
+import kotlin.test.Test
+import kotlin.time.Duration
+import kotlin.time.ExperimentalTime
+import kotlin.time.milliseconds
+import kotlin.time.seconds
+
+@ExperimentalTime
+class DelayDurationTest : TestBase() {
+
+    @Test
+    fun testCancellation() = runTest(expected = { it is CancellationException }) {
+        runAndCancel(1.seconds)
+    }
+
+    @Test
+    fun testInfinite() = runTest(expected = { it is CancellationException }) {
+        runAndCancel(Duration.INFINITE)
+    }
+
+    @Test
+    fun testRegularDelay() = runTest {
+        val deferred = async {
+            expect(2)
+            delay(1.milliseconds)
+            expect(3)
+        }
+
+        expect(1)
+        yield()
+        deferred.await()
+        finish(4)
+    }
+
+    private suspend fun runAndCancel(time: Duration) = coroutineScope {
+        expect(1)
+        val deferred = async {
+            expect(2)
+            delay(time)
+            expectUnreached()
+        }
+
+        yield()
+        expect(3)
+        require(deferred.isActive)
+        deferred.cancel()
+        finish(4)
+        deferred.await()
+    }
+}

kotlinx-coroutines-core/common/test/WithTimeoutDurationTest.kt

@@ -0,0 +1,219 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+@file:Suppress("NAMED_ARGUMENTS_NOT_ALLOWED", "UNREACHABLE_CODE")
+
+// KT-21913
+
+package kotlinx.coroutines
+
+import kotlin.test.*
+import kotlin.time.Duration
+import kotlin.time.ExperimentalTime
+import kotlin.time.milliseconds
+import kotlin.time.seconds
+
+@ExperimentalTime
+class WithTimeoutDurationTest : TestBase() {
+    /**
+     * Tests a case of no timeout and no suspension inside.
+     */
+    @Test
+    fun testBasicNoSuspend() = runTest {
+        expect(1)
+        val result = withTimeout(10.seconds) {
+            expect(2)
+            "OK"
+        }
+        assertEquals("OK", result)
+        finish(3)
+    }
+
+    /**
+     * Tests a case of no timeout and one suspension inside.
+     */
+    @Test
+    fun testBasicSuspend() = runTest {
+        expect(1)
+        val result = withTimeout(10.seconds) {
+            expect(2)
+            yield()
+            expect(3)
+            "OK"
+        }
+        assertEquals("OK", result)
+        finish(4)
+    }
+
+    /**
+     * Tests proper dispatching of `withTimeout` blocks
+     */
+    @Test
+    fun testDispatch() = runTest {
+        expect(1)
+        launch {
+            expect(4)
+            yield() // back to main
+            expect(7)
+        }
+        expect(2)
+        // test that it does not yield to the above job when started
+        val result = withTimeout(1.seconds) {
+            expect(3)
+            yield() // yield only now
+            expect(5)
+            "OK"
+        }
+        assertEquals("OK", result)
+        expect(6)
+        yield() // back to launch
+        finish(8)
+    }
+
+
+    /**
+     * Tests that a 100% CPU-consuming loop will react on timeout if it has yields.
+     */
+    @Test
+    fun testYieldBlockingWithTimeout() = runTest(
+            expected = { it is CancellationException }
+    ) {
+        withTimeout(100.milliseconds) {
+            while (true) {
+                yield()
+            }
+        }
+    }
+
+    /**
+     * Tests that [withTimeout] waits for children coroutines to complete.
+     */
+    @Test
+    fun testWithTimeoutChildWait() = runTest {
+        expect(1)
+        withTimeout(100.milliseconds) {
+            expect(2)
+            // launch child with timeout
+            launch {
+                expect(4)
+            }
+            expect(3)
+            // now will wait for child before returning
+        }
+        finish(5)
+    }
+
+    @Test
+    fun testBadClass() = runTest {
+        val bad = BadClass()
+        val result = withTimeout(100.milliseconds) {
+            bad
+        }
+        assertSame(bad, result)
+    }
+
+    class BadClass {
+        override fun equals(other: Any?): Boolean = error("Should not be called")
+        override fun hashCode(): Int = error("Should not be called")
+        override fun toString(): String = error("Should not be called")
+    }
+
+    @Test
+    fun testExceptionOnTimeout() = runTest {
+        expect(1)
+        try {
+            withTimeout(100.milliseconds) {
+                expect(2)
+                delay(1000.milliseconds)
+                expectUnreached()
+                "OK"
+            }
+        } catch (e: CancellationException) {
+            assertEquals("Timed out waiting for 100 ms", e.message)
+            finish(3)
+        }
+    }
+
+    @Test
+    fun testSuppressExceptionWithResult() = runTest(
+            expected = { it is CancellationException }
+    ) {
+        expect(1)
+        withTimeout(100.milliseconds) {
+            expect(2)
+            try {
+                delay(1000.milliseconds)
+            } catch (e: CancellationException) {
+                finish(3)
+            }
+            "OK"
+        }
+        expectUnreached()
+    }
+
+    @Test
+    fun testSuppressExceptionWithAnotherException() = runTest {
+        expect(1)
+        try {
+            withTimeout(100.milliseconds) {
+                expect(2)
+                try {
+                    delay(1000.milliseconds)
+                } catch (e: CancellationException) {
+                    expect(3)
+                    throw TestException()
+                }
+                expectUnreached()
+                "OK"
+            }
+            expectUnreached()
+        } catch (e: TestException) {
+            finish(4)
+        }
+    }
+
+    @Test
+    fun testNegativeTimeout() = runTest {
+        expect(1)
+        try {
+            withTimeout(-1.milliseconds) {
+                expectUnreached()
+                "OK"
+            }
+        } catch (e: TimeoutCancellationException) {
+            assertEquals("Timed out immediately", e.message)
+            finish(2)
+        }
+    }
+
+    @Test
+    fun testExceptionFromWithinTimeout() = runTest {
+        expect(1)
+        try {
+            expect(2)
+            withTimeout(1.seconds) {
+                expect(3)
+                throw TestException()
+            }
+            expectUnreached()
+        } catch (e: TestException) {
+            finish(4)
+        }
+    }
+
+    @Test
+    fun testIncompleteWithTimeoutState() = runTest {
+        lateinit var timeoutJob: Job
+        val handle = withTimeout(Duration.INFINITE) {
+            timeoutJob = coroutineContext[Job]!!
+            timeoutJob.invokeOnCompletion { }
+        }
+
+        handle.dispose()
+        timeoutJob.join()
+        assertTrue(timeoutJob.isCompleted)
+        assertFalse(timeoutJob.isActive)
+        assertFalse(timeoutJob.isCancelled)
+    }
+}