Apply new convention (without restricted controllers)

Author: dzharkov

kotlinx-coroutines-async-common/src/main/kotlin/runWithCurrentContinuation.kt

@@ -1,6 +1,9 @@
 package kotlinx.coroutines
 
 import java.util.concurrent.atomic.AtomicReference
+import kotlin.coroutines.*
+import kotlin.coroutines.SUSPENDED
+import kotlin.coroutines.suspendWithCurrentContinuation
 
 suspend fun <T> runWithCurrentContinuation(
         block: (Continuation<T>) -> Unit
@@ -25,7 +28,7 @@ private class SafeContinuation<in T>(val delegate: Continuation<T>) : Continuati
     }
 
     fun returnResult(): Any? {
-        if (result.get() == Undecided) result.compareAndSet(Undecided, Suspend)
+        if (result.get() == Undecided) result.compareAndSet(Undecided, SUSPENDED)
         val result = result.get()
         if (result is Fail) throw result.e else return result
     }

kotlinx-coroutines-async-example-ui/src/main/kotlin/main.kt

@@ -1,4 +1,5 @@
 import kotlinx.coroutines.asyncUI
+import kotlinx.coroutines.await
 import java.awt.Insets
 import java.util.concurrent.CompletableFuture
 import javax.swing.*

kotlinx-coroutines-async/src/main/kotlin/async.kt

@@ -9,6 +9,9 @@ import java.nio.channels.CompletionHandler
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.TimeUnit
 import javax.swing.SwingUtilities
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.ResumeInterceptor
+import kotlin.coroutines.startCoroutine
 
 /**
  * Run asynchronous computations based on [c] coroutine parameter
@@ -26,13 +29,50 @@ import javax.swing.SwingUtilities
  *
  * @return CompletableFuture object representing result of computations
  */
+
+
 fun <T> async(
         continuationWrapper: ContinuationWrapper? = null,
-        coroutine c: FutureController<T>.() -> Continuation<Unit>
+        c: suspend () -> T
 ): CompletableFuture<T> {
-    val controller = FutureController<T>(continuationWrapper)
-    c(controller).resume(Unit)
-    return controller.future
+    val future = CompletableFuture<T>()
+
+    c.startCoroutine(
+            object : Continuation<T> {
+                override fun resumeWithException(exception: Throwable) {
+                    future.completeExceptionally(exception)
+                }
+
+                override fun resume(data: T) {
+                    future.complete(data)
+                }
+            },
+
+            if (continuationWrapper != null) {
+                object: ResumeInterceptor {
+                    override fun <P> interceptResume(data: P, continuation: Continuation<P>): Boolean {
+                        continuationWrapper {
+                            continuation.resume(data)
+                        }
+
+                        return true
+                    }
+
+                    override fun interceptResumeWithException(exception: Throwable, continuation: Continuation<*>): Boolean {
+                        continuationWrapper {
+                            continuation.resumeWithException(exception)
+                        }
+
+                        return true
+                    }
+                }
+            }
+            else {
+                null
+            }
+    )
+
+    return future
 }
 
 /**
@@ -45,27 +85,14 @@ fun <T> async(
  * @See async
  */
 fun asyncUI(
-        coroutine c: FutureController<Unit>.() -> Continuation<Unit>
+        c: suspend () -> Unit
 ) {
-    if (SwingUtilities.isEventDispatchThread()) {
-        async({ SwingUtilities.invokeLater(it) }, c)
-    }
-    else {
-        SwingUtilities.invokeLater {
-            async({ SwingUtilities.invokeLater(it) }, c)
-        }
-    }
+    async({ SwingUtilities.invokeLater(it) }, c)
 }
 
 typealias ContinuationWrapper = (() -> Unit) -> Unit
 
-@AllowSuspendExtensions
-class FutureController<T>(
-        val continuationWrapper: ContinuationWrapper?
-) {
-    val future = CompletableFuture<T>()
-
-    suspend fun <V> await(f: CompletableFuture<V>): V =
+suspend fun <V> await(f: CompletableFuture<V>): V =
         runWithCurrentContinuation {
             f.whenComplete { value, throwable ->
                 if (throwable == null)
@@ -74,25 +101,6 @@ class FutureController<T>(
                     it.resumeWithException(throwable)
             }
         }
-
-    inline operator fun interceptResume(crossinline x: () -> Unit) {
-        if (continuationWrapper != null) {
-            continuationWrapper.invoke { x() }
-        }
-        else {
-            x()
-        }
-    }
-
-    operator fun handleResult(value: T, c: Continuation<Nothing>) {
-        future.complete(value)
-    }
-
-    operator fun handleException(t: Throwable, c: Continuation<Nothing>) {
-        future.completeExceptionally(t)
-    }
-}
-
 //
 // IO parts
 //

kotlinx-coroutines-generate/src/main/kotlin/generate.kt

@@ -1,41 +1,44 @@
 package kotlinx.coroutines
 
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.SUSPENDED
+import kotlin.coroutines.createCoroutine
+import kotlin.coroutines.suspendWithCurrentContinuation
+
 /**
  * Creates a Sequence object based on received coroutine [c].
  *
  * Each call of 'yield' suspend function within the coroutine lambda generates
  * next element of resulting sequence.
  */
-fun <T> generate(
-        coroutine c: GeneratorController<T>.() -> Continuation<Unit>
-): Sequence<T> =
-        object : Sequence<T> {
-            override fun iterator(): Iterator<T> {
-                val iterator = GeneratorController<T>()
-                iterator.setNextStep(c(iterator))
-                return iterator
-            }
+interface Generator<in T> {
+    suspend fun yield(value: T)
+}
+
+fun <T> generate(block: suspend Generator<T>.() -> Unit): Sequence<T> = GeneratedSequence(block)
+
+private class GeneratedSequence<out T>(private val block: suspend Generator<T>.() -> Unit) : Sequence<T> {
+    override fun iterator(): Iterator<T> = GeneratedIterator(block)
+}
+
+private class GeneratedIterator<T>(block: suspend Generator<T>.() -> Unit) : AbstractIterator<T>(), Generator<T> {
+    private var nextStep: Continuation<Unit> = block.createCoroutine(this, object : Continuation<Unit> {
+        override fun resume(data: Unit) {
+            done()
         }
 
-class GeneratorController<T> internal constructor() : AbstractIterator<T>() {
-    private lateinit var nextStep: Continuation<Unit>
+        override fun resumeWithException(exception: Throwable) {
+            throw exception
+        }
+    })
 
     override fun computeNext() {
         nextStep.resume(Unit)
     }
-
-    internal fun setNextStep(step: Continuation<Unit>) {
-        nextStep = step
-    }
-
-    suspend fun yield(value: T) = suspendWithCurrentContinuation<Unit> { c ->
+    suspend override fun yield(value: T) = suspendWithCurrentContinuation<Unit> { c ->
         setNext(value)
-        setNextStep(c)
+        nextStep = c
 
-        Suspend
+        SUSPENDED
     }
-
-    operator fun handleResult(result: Unit, c: Continuation<Nothing>) {
-        done()
-    }
-}
+}

kotlinx-coroutines-rx-example/src/main/kotlin/main.kt

@@ -1,6 +1,7 @@
 package kotlinx.coroutines.example
 
 import kotlinx.coroutines.asyncRx
+import kotlinx.coroutines.awaitSingle
 import retrofit2.Retrofit
 import retrofit2.adapter.rxjava.RxJavaCallAdapterFactory
 import retrofit2.converter.gson.GsonConverterFactory

kotlinx-coroutines-rx/src/main/kotlin/asyncRx.kt

@@ -2,6 +2,8 @@ package kotlinx.coroutines
 
 import rx.Observable
 import rx.subjects.AsyncSubject
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.startCoroutine
 
 /**
  * Run asynchronous computations based on [c] coroutine parameter
@@ -16,40 +18,39 @@ import rx.subjects.AsyncSubject
  * @return Observable with single value containing expression returned from coroutine
  */
 fun <T> asyncRx(
-        coroutine c: RxController<T>.() -> Continuation<Unit>
+        c: suspend () -> T
 ): Observable<T> {
-    val controller = RxController<T>()
-    c(controller).resume(Unit)
-
-    return controller.result
+    val result: AsyncSubject<T> = AsyncSubject.create<T>()
+
+    c.startCoroutine(
+            object: Continuation<T> {
+                override fun resumeWithException(exception: Throwable) {
+                    result.onError(exception)
+                }
+
+                override fun resume(data: T) {
+                    result.onNext(data)
+                    result.onCompleted()
+                }
+            }
+    )
+
+    return result
 }
 
-@AllowSuspendExtensions
-class RxController<T> internal constructor() {
-    internal val result: AsyncSubject<T> = AsyncSubject.create<T>()
-
-    suspend fun <V> Observable<V>.awaitFirst() = first().awaitOne()
 
-    suspend fun <V> Observable<V>.awaitLast() = last().awaitOne()
+suspend fun <V> Observable<V>.awaitFirst(): V = first().awaitOne()
 
-    suspend fun <V> Observable<V>.awaitSingle() = single().awaitOne()
+suspend fun <V> Observable<V>.awaitLast(): V = last().awaitOne()
 
-    private suspend fun <V> Observable<V>.awaitOne() = runWithCurrentContinuation<V> { x ->
-        subscribe(x::resume, x::resumeWithException)
-    }
+suspend fun <V> Observable<V>.awaitSingle(): V = single().awaitOne()
 
-    suspend fun <V> Observable<V>.applyForEachAndAwait(
-            block: (V) -> Unit
-    ) = runWithCurrentContinuation<Unit> { x->
-        this.subscribe(block, x::resumeWithException, { x.resume(Unit) })
-    }
-
-    operator fun handleResult(v: T, x: Continuation<Nothing>) {
-        result.onNext(v)
-        result.onCompleted()
-    }
+private suspend fun <V> Observable<V>.awaitOne(): V = runWithCurrentContinuation<V> { x ->
+    subscribe(x::resume, x::resumeWithException)
+}
 
-    operator fun handleException(t: Throwable, x: Continuation<Nothing>) {
-        result.onError(t)
-    }
+suspend fun <V> Observable<V>.applyForEachAndAwait(
+        block: (V) -> Unit
+) = runWithCurrentContinuation<Unit> { x->
+    this.subscribe(block, x::resumeWithException, { x.resume(Unit) })
 }