Coroutines migration adapters
This commit is contained in:
Roman Elizarov
committed by
Denis Zharkov
Denis Zharkov
parent
ebe3fca45c
commit
20c255948a
+187
@@ -0,0 +1,187 @@
|
||||
/*
|
||||
* Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
|
||||
* that can be found in the license/LICENSE.txt file.
|
||||
*/
|
||||
|
||||
package kotlin.coroutines.experimental.migration
|
||||
|
||||
import kotlin.coroutines.experimental.Continuation as ExperimentalContinuation
|
||||
import kotlin.coroutines.experimental.CoroutineContext as ExperimentalCoroutineContext
|
||||
import kotlin.coroutines.experimental.AbstractCoroutineContextElement as ExperimentalAbstractCoroutineContextElement
|
||||
import kotlin.coroutines.experimental.EmptyCoroutineContext as ExperimentalEmptyCoroutineContext
|
||||
import kotlin.coroutines.experimental.ContinuationInterceptor as ExperimentalContinuationInterceptor
|
||||
import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED as EXPERIMENTAL_COROUTINE_SUSPENDED
|
||||
import kotlin.coroutines.*
|
||||
import kotlin.coroutines.intrinsics.*
|
||||
|
||||
|
||||
/**
|
||||
* Adapter to invoke experimental suspending function.
|
||||
*
|
||||
* To invoke experimental suspending function `foo(args)` that returns value of some type `Result`
|
||||
* from Kotlin 1.3 use the following code:
|
||||
*
|
||||
* ```
|
||||
* invokeExperimentalSuspend<Result> { foo(args, it) }
|
||||
* ```
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public suspend inline fun <T> invokeExperimentalSuspend(crossinline invocation: (ExperimentalContinuation<T>) -> Any?): T =
|
||||
suspendCoroutineUninterceptedOrReturn { continuation ->
|
||||
val result = invocation(continuation.toExperimentalContinuation())
|
||||
@Suppress("UNCHECKED_CAST")
|
||||
if (result === EXPERIMENTAL_COROUTINE_SUSPENDED) COROUTINE_SUSPENDED else result as T
|
||||
}
|
||||
|
||||
/**
|
||||
* Coverts reference to suspending function to experimental suspending function.
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun <T> (suspend () -> T).toExperimentalSuspendFunction(): (ExperimentalContinuation<T>) -> Any? =
|
||||
ExperimentalSuspendFunction0Migration(this)
|
||||
|
||||
/**
|
||||
* Coverts reference to experimental suspending function to suspending function.
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun <T> ((ExperimentalContinuation<T>) -> Any?).toSuspendFunction(): suspend () -> T =
|
||||
(this as? ExperimentalSuspendFunction0Migration)?.function ?: SuspendFunction0Migration(this)::invoke
|
||||
|
||||
/**
|
||||
* Coverts reference to suspending function with receiver to experimental suspending function.
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun <R, T> (suspend (R) -> T).toExperimentalSuspendFunction(): (R, ExperimentalContinuation<T>) -> Any? =
|
||||
ExperimentalSuspendFunction1Migration(this)
|
||||
|
||||
/**
|
||||
* Coverts reference to experimental suspending function with receiver to suspending function.
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun <R, T> ((R, ExperimentalContinuation<T>) -> Any?).toSuspendFunction(): suspend (R) -> T =
|
||||
(this as? ExperimentalSuspendFunction1Migration)?.function ?: SuspendFunction1Migration(this)::invoke
|
||||
|
||||
/**
|
||||
* Converts [Continuation] to [ExperimentalContinuation].
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun <T> Continuation<T>.toExperimentalContinuation(): ExperimentalContinuation<T> =
|
||||
(this as? ContinuationMigration<T>)?.continuation ?: ExperimentalContinuationMigration(this)
|
||||
|
||||
/**
|
||||
* Converts [ExperimentalContinuation] to [Continuation].
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun <T> ExperimentalContinuation<T>.toContinuation(): Continuation<T> =
|
||||
(this as? ExperimentalContinuationMigration<T>)?.continuation ?: ContinuationMigration(this)
|
||||
|
||||
/**
|
||||
* Converts [CoroutineContext] to [ExperimentalCoroutineContext].
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun CoroutineContext.toExperimentalCoroutineContext(): ExperimentalCoroutineContext {
|
||||
val interceptor = get(ContinuationInterceptor.Key)
|
||||
val migration = get(ContextMigration.Key)
|
||||
val remainder = minusKey(ContinuationInterceptor.Key).minusKey(ContextMigration.Key)
|
||||
val original = migration?.context ?: ExperimentalEmptyCoroutineContext
|
||||
val result = if (remainder === EmptyCoroutineContext) original else original + ExperimentalContextMigration(remainder)
|
||||
return if (interceptor == null) result else result + interceptor.toExperimentalContinuationInterceptor()
|
||||
}
|
||||
|
||||
/**
|
||||
* Converts [ExperimentalCoroutineContext] to [CoroutineContext].
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun ExperimentalCoroutineContext.toCoroutineContext(): CoroutineContext {
|
||||
val interceptor = get(ExperimentalContinuationInterceptor.Key)
|
||||
val migration = get(ExperimentalContextMigration.Key)
|
||||
val remainder = minusKey(ExperimentalContinuationInterceptor.Key).minusKey(ExperimentalContextMigration.Key)
|
||||
val original = migration?.context ?: EmptyCoroutineContext
|
||||
val result = if (remainder === ExperimentalEmptyCoroutineContext) original else original + ContextMigration(remainder)
|
||||
return if (interceptor == null) result else result + interceptor.toContinuationInterceptor()
|
||||
}
|
||||
|
||||
/**
|
||||
* Converts [ContinuationInterceptor] to [ExperimentalContinuationInterceptor].
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun ContinuationInterceptor.toExperimentalContinuationInterceptor(): ExperimentalContinuationInterceptor =
|
||||
(this as? ContinuationInterceptorMigration)?.interceptor ?: ExperimentalContinuationInterceptorMigration(this)
|
||||
|
||||
/**
|
||||
* Converts [ExperimentalContinuationInterceptor] to [ContinuationInterceptor].
|
||||
*/
|
||||
@SinceKotlin("1.3")
|
||||
public fun ExperimentalContinuationInterceptor.toContinuationInterceptor(): ContinuationInterceptor =
|
||||
(this as? ExperimentalContinuationInterceptorMigration)?.interceptor ?: ContinuationInterceptorMigration(this)
|
||||
|
||||
// ------------------ converter classes ------------------
|
||||
// Their name starts with "Experimental" if they implement the corresponding Experimental interfaces
|
||||
|
||||
private class ExperimentalSuspendFunction0Migration<T>(val function: suspend () -> T) : (ExperimentalContinuation<T>) -> Any? {
|
||||
override fun invoke(continuation: ExperimentalContinuation<T>): Any? {
|
||||
val result = function.startCoroutineUninterceptedOrReturn(continuation.toContinuation())
|
||||
return if (result === COROUTINE_SUSPENDED) EXPERIMENTAL_COROUTINE_SUSPENDED else result
|
||||
}
|
||||
}
|
||||
|
||||
private class SuspendFunction0Migration<T>(val function: (ExperimentalContinuation<T>) -> Any?) {
|
||||
suspend fun invoke(): T = suspendCoroutineUninterceptedOrReturn { continuation ->
|
||||
val result = function(continuation.toExperimentalContinuation())
|
||||
if (result === EXPERIMENTAL_COROUTINE_SUSPENDED) COROUTINE_SUSPENDED else result
|
||||
}
|
||||
}
|
||||
|
||||
private class ExperimentalSuspendFunction1Migration<R, T>(val function: suspend (R) -> T) : (R, ExperimentalContinuation<T>) -> Any? {
|
||||
override fun invoke(receiver: R, continuation: ExperimentalContinuation<T>): Any? {
|
||||
val result = function.startCoroutineUninterceptedOrReturn(receiver, continuation.toContinuation())
|
||||
return if (result === COROUTINE_SUSPENDED) EXPERIMENTAL_COROUTINE_SUSPENDED else result
|
||||
}
|
||||
}
|
||||
|
||||
private class SuspendFunction1Migration<R, T>(val function: (R, ExperimentalContinuation<T>) -> Any?) {
|
||||
suspend fun invoke(receiver: R): T = suspendCoroutineUninterceptedOrReturn { continuation ->
|
||||
val result = function(receiver, continuation.toExperimentalContinuation())
|
||||
if (result === EXPERIMENTAL_COROUTINE_SUSPENDED) COROUTINE_SUSPENDED else result
|
||||
}
|
||||
}
|
||||
|
||||
private class ExperimentalContinuationMigration<T>(val continuation: Continuation<T>): ExperimentalContinuation<T> {
|
||||
override val context = continuation.context.toExperimentalCoroutineContext()
|
||||
override fun resume(value: T) = continuation.resume(value)
|
||||
override fun resumeWithException(exception: Throwable) = continuation.resumeWithException(exception)
|
||||
}
|
||||
|
||||
private class ContinuationMigration<T>(val continuation: ExperimentalContinuation<T>): Continuation<T> {
|
||||
override val context: CoroutineContext = continuation.context.toCoroutineContext()
|
||||
override fun resumeWith(result: SuccessOrFailure<T>) {
|
||||
result
|
||||
.onSuccess { continuation.resume(it) }
|
||||
.onFailure { continuation.resumeWithException(it) }
|
||||
}
|
||||
}
|
||||
|
||||
private class ExperimentalContextMigration(val context: CoroutineContext): ExperimentalAbstractCoroutineContextElement(Key) {
|
||||
companion object Key : ExperimentalCoroutineContext.Key<ExperimentalContextMigration>
|
||||
}
|
||||
|
||||
private class ContextMigration(val context: ExperimentalCoroutineContext): AbstractCoroutineContextElement(Key) {
|
||||
companion object Key : CoroutineContext.Key<ContextMigration>
|
||||
}
|
||||
|
||||
private class ExperimentalContinuationInterceptorMigration(val interceptor: ContinuationInterceptor) : ExperimentalContinuationInterceptor {
|
||||
override val key: ExperimentalCoroutineContext.Key<*>
|
||||
get() = ExperimentalContinuationInterceptor.Key
|
||||
|
||||
override fun <T> interceptContinuation(continuation: ExperimentalContinuation<T>): ExperimentalContinuation<T> =
|
||||
interceptor.interceptContinuation(continuation.toContinuation()).toExperimentalContinuation()
|
||||
}
|
||||
|
||||
private class ContinuationInterceptorMigration(val interceptor: ExperimentalContinuationInterceptor) : ContinuationInterceptor {
|
||||
override val key: CoroutineContext.Key<*>
|
||||
get() = ContinuationInterceptor.Key
|
||||
|
||||
override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
|
||||
interceptor.interceptContinuation(continuation.toExperimentalContinuation()).toContinuation()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
/*
|
||||
* Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
|
||||
* that can be found in the license/LICENSE.txt file.
|
||||
*/
|
||||
|
||||
@file:Suppress("INVISIBLE_MEMBER", "INVISIBLE_REFERENCE")
|
||||
|
||||
package test.kotlin.coroutines.experimental.migration
|
||||
|
||||
import test.kotlin.coroutines.TestDispatcher
|
||||
import java.util.concurrent.Semaphore
|
||||
import kotlin.coroutines.experimental.buildSequence as experimentalBuildSequence
|
||||
import kotlin.coroutines.experimental.SequenceBuilder as ExperimentalSequenceBuilder
|
||||
import kotlin.coroutines.experimental.Continuation as ExperimentalContinuation
|
||||
import kotlin.coroutines.experimental.CoroutineContext as ExperimentalCoroutineContext
|
||||
import kotlin.coroutines.experimental.AbstractCoroutineContextElement as ExperimentalAbstractCoroutineContextElement
|
||||
import kotlin.coroutines.experimental.EmptyCoroutineContext as ExperimentalEmptyCoroutineContext
|
||||
import kotlin.coroutines.experimental.ContinuationInterceptor as ExperimentalContinuationInterceptor
|
||||
import kotlin.coroutines.experimental.intrinsics.COROUTINE_SUSPENDED as EXPERIMENTAL_COROUTINE_SUSPENDED
|
||||
import kotlin.coroutines.experimental.migration.*
|
||||
import kotlin.coroutines.*
|
||||
import kotlin.coroutines.experimental.startCoroutine
|
||||
import kotlin.coroutines.jvm.internal.runSuspend
|
||||
import kotlin.test.*
|
||||
|
||||
/**
|
||||
* Tests on coroutines migrations utilities.
|
||||
*/
|
||||
class CoroutinesMigrationTest {
|
||||
@Test
|
||||
fun testContextMigration() {
|
||||
assertTrue(EmptyCoroutineContext === ExperimentalEmptyCoroutineContext.toCoroutineContext())
|
||||
assertTrue(ExperimentalEmptyCoroutineContext === EmptyCoroutineContext.toExperimentalCoroutineContext())
|
||||
MyElement().let { e ->
|
||||
assertTrue(e === e.toExperimentalCoroutineContext().toCoroutineContext())
|
||||
val ee = MyExperimentalElement()
|
||||
val ctx = (e.toExperimentalCoroutineContext() + ee).toCoroutineContext()
|
||||
assertTrue(e === ctx[MyElement.Key])
|
||||
assertTrue(ee === ctx.toExperimentalCoroutineContext()[MyExperimentalElement.Key])
|
||||
}
|
||||
MyExperimentalElement().let { ee ->
|
||||
assertTrue(ee === ee.toCoroutineContext().toExperimentalCoroutineContext())
|
||||
val e = MyElement()
|
||||
val ctx = (ee.toCoroutineContext() + e).toExperimentalCoroutineContext()
|
||||
assertTrue(ee === ctx[MyExperimentalElement.Key])
|
||||
assertTrue(e === ctx.toCoroutineContext()[MyElement.Key])
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testFunctionMigration() {
|
||||
var fooCnt = 0
|
||||
suspend fun foo() {
|
||||
fooCnt++
|
||||
}
|
||||
runSuspend {
|
||||
foo()
|
||||
}
|
||||
assertEquals(1, fooCnt)
|
||||
val foo2 = ::foo.toExperimentalSuspendFunction().toSuspendFunction()
|
||||
runSuspend {
|
||||
foo2()
|
||||
}
|
||||
assertEquals(2, fooCnt)
|
||||
var barCnt = 0
|
||||
suspend fun bar(x: Int) {
|
||||
barCnt += x
|
||||
}
|
||||
runSuspend {
|
||||
bar(2)
|
||||
}
|
||||
assertEquals(2, barCnt)
|
||||
val bar2 = ::bar.toExperimentalSuspendFunction().toSuspendFunction()
|
||||
runSuspend {
|
||||
bar2(3)
|
||||
}
|
||||
assertEquals(5, barCnt)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testInvokeExperimental0() {
|
||||
testInvokeExperimentalImpl(false)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testInvokeExperimental1() {
|
||||
testInvokeExperimentalImpl(true)
|
||||
}
|
||||
|
||||
private fun testInvokeExperimentalImpl(shallSuspend: Boolean) {
|
||||
TestDispatcher("testInvokeExperimental").use { dispatcher ->
|
||||
val semaphore = Semaphore(0)
|
||||
val foo: suspend (Int) -> String = {
|
||||
dispatcher.assertThread()
|
||||
if (shallSuspend) {
|
||||
dispatcher.yield()
|
||||
}
|
||||
"OK$it"
|
||||
}
|
||||
val fooExp = foo.toExperimentalSuspendFunction()
|
||||
val testCode: suspend () -> String = {
|
||||
dispatcher.assertThread()
|
||||
val result = invokeExperimentalSuspend<String> {
|
||||
fooExp(42, it)
|
||||
}
|
||||
assertEquals("OK42", result)
|
||||
dispatcher.assertThread()
|
||||
"DONE"
|
||||
}
|
||||
testCode.startCoroutine(Continuation(dispatcher) { result ->
|
||||
dispatcher.assertThread()
|
||||
// todo: below does not work due to a bug in inline classes
|
||||
// assertEquals("DONE", result.getOrThrow())
|
||||
semaphore.release()
|
||||
})
|
||||
semaphore.acquire()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testExperimentalBuildSequence() {
|
||||
suspend fun action(builder: ExperimentalSequenceBuilder<Int>) {
|
||||
invokeExperimentalSuspend<Unit> { builder.yield(1, it) }
|
||||
invokeExperimentalSuspend<Unit> { builder.yield(2, it) }
|
||||
invokeExperimentalSuspend<Unit> { builder.yield(3, it) }
|
||||
}
|
||||
val seq = experimentalBuildSequence(::action.toExperimentalSuspendFunction())
|
||||
assertEquals(listOf(1, 2, 3), seq.toList())
|
||||
}
|
||||
|
||||
class MyElement : AbstractCoroutineContextElement(Key) {
|
||||
companion object Key : CoroutineContext.Key<MyElement>
|
||||
}
|
||||
|
||||
class MyExperimentalElement : ExperimentalAbstractCoroutineContextElement(Key) {
|
||||
companion object Key : ExperimentalCoroutineContext.Key<MyExperimentalElement>
|
||||
}
|
||||
}
|
||||
|
||||
@@ -52,35 +52,3 @@ class DispatcherSwitcher(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class TestDispatcher(
|
||||
private val name: String
|
||||
) : AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor, Closeable {
|
||||
private lateinit var thread: Thread
|
||||
|
||||
val executor: ExecutorService = Executors.newSingleThreadExecutor { runnable ->
|
||||
Thread(runnable, name).also { thread = it }
|
||||
}
|
||||
|
||||
fun assertThread() {
|
||||
assertEquals(thread, Thread.currentThread())
|
||||
}
|
||||
|
||||
override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
|
||||
DispatchedContinuation(continuation)
|
||||
|
||||
override fun close() {
|
||||
executor.shutdown()
|
||||
}
|
||||
|
||||
inner class DispatchedContinuation<T>(val delegate: Continuation<T>) : Continuation<T> {
|
||||
override val context: CoroutineContext = delegate.context
|
||||
|
||||
override fun resumeWith(result: SuccessOrFailure<T>) {
|
||||
executor.execute {
|
||||
delegate.resumeWith(result)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
/*
|
||||
* Copyright 2010-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license
|
||||
* that can be found in the license/LICENSE.txt file.
|
||||
*/
|
||||
|
||||
package test.kotlin.coroutines
|
||||
|
||||
import java.io.Closeable
|
||||
import java.util.concurrent.ExecutorService
|
||||
import java.util.concurrent.Executors
|
||||
import kotlin.coroutines.*
|
||||
import kotlin.test.assertEquals
|
||||
|
||||
class TestDispatcher(
|
||||
private val name: String
|
||||
) : AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor, Closeable {
|
||||
private lateinit var thread: Thread
|
||||
|
||||
val executor: ExecutorService = Executors.newSingleThreadExecutor { runnable ->
|
||||
Thread(runnable, name).also { thread = it }
|
||||
}
|
||||
|
||||
fun assertThread() {
|
||||
assertEquals(thread, Thread.currentThread())
|
||||
}
|
||||
|
||||
override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
|
||||
DispatchedContinuation(continuation)
|
||||
|
||||
override fun close() {
|
||||
executor.shutdown()
|
||||
}
|
||||
|
||||
suspend fun yield() = suspendCoroutine<Unit> { cont ->
|
||||
executor.execute {
|
||||
assertThread()
|
||||
cont.resume(Unit)
|
||||
}
|
||||
}
|
||||
|
||||
inner class DispatchedContinuation<T>(val delegate: Continuation<T>) : Continuation<T> {
|
||||
override val context: CoroutineContext = delegate.context
|
||||
|
||||
override fun resumeWith(result: SuccessOrFailure<T>) {
|
||||
executor.execute {
|
||||
delegate.resumeWith(result)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user