API Review for 1.0 release
Introduces annotation to mark different maturities of coroutines API:
* ExperimentalCoroutinesApi for declarations that are not stable yet
* ObsoleteCoroutinesApi for declarations that will be reworked for sure
* InternalCoroutinesApi for declarations that should not be outside
Changes:
* Updated guide tests
* TimeUnit is deprecated, everything is in ms now
Converting small delays < 1ms to 1ms delay
Fixes #255
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-android.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-android.txt
index d8ed275..9f3922d 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-android.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-android.txt
@@ -1,8 +1,3 @@
-public final class kotlinx/coroutines/experimental/android/AndroidExceptionPreHandler : kotlin/coroutines/experimental/AbstractCoroutineContextElement, kotlinx/coroutines/experimental/CoroutineExceptionHandler {
- public fun <init> ()V
- public fun handleException (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Throwable;)V
-}
-
public final class kotlinx/coroutines/experimental/android/HandlerContext : kotlinx/coroutines/experimental/android/HandlerDispatcher, kotlinx/coroutines/experimental/Delay {
public fun <init> (Landroid/os/Handler;Ljava/lang/String;)V
public synthetic fun <init> (Landroid/os/Handler;Ljava/lang/String;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
@@ -12,9 +7,9 @@
public fun getImmediate ()Lkotlinx/coroutines/experimental/android/HandlerContext;
public synthetic fun getImmediate ()Lkotlinx/coroutines/experimental/android/HandlerDispatcher;
public fun hashCode ()I
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
public fun isDispatchNeeded (Lkotlin/coroutines/experimental/CoroutineContext;)Z
- public fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun scheduleResumeAfterDelay (JLkotlinx/coroutines/experimental/CancellableContinuation;)V
public fun toString ()Ljava/lang/String;
}
@@ -24,14 +19,17 @@
}
public abstract class kotlinx/coroutines/experimental/android/HandlerDispatcher : kotlinx/coroutines/experimental/CoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
- public fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public synthetic fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public abstract fun getImmediate ()Lkotlinx/coroutines/experimental/android/HandlerDispatcher;
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
}
public final class kotlinx/coroutines/experimental/android/HandlerDispatcherKt {
- public static final fun asCoroutineDispatcher (Landroid/os/Handler;)Lkotlinx/coroutines/experimental/android/HandlerDispatcher;
public static final fun awaitFrame (Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final fun from (Landroid/os/Handler;)Lkotlinx/coroutines/experimental/android/HandlerDispatcher;
public static final fun getMain (Lkotlinx/coroutines/experimental/Dispatchers;)Lkotlinx/coroutines/experimental/android/HandlerDispatcher;
}
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt
index f95bce0..afb2184 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt
@@ -22,22 +22,27 @@
}
public final class kotlinx/coroutines/experimental/BuildersKt {
+ public static final fun async (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Deferred;
+ public static synthetic fun async$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Deferred;
public static final fun launch (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Job;
public static final synthetic fun launch (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Job;
public static final fun launch (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Job;
public static final synthetic fun launch (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Job;
public static final fun launch (Lkotlin/coroutines/experimental/CoroutineContext;ZLkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Job;
public static final fun launch (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Job;
+ public static final fun launch (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/Job;
public static synthetic fun launch$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Job;
public static synthetic fun launch$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Job;
public static synthetic fun launch$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Job;
public static synthetic fun launch$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Job;
public static synthetic fun launch$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Job;
+ public static synthetic fun launch$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Job;
public static final synthetic fun run (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final fun run (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static synthetic fun run$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
public static final fun runBlocking (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlin/jvm/functions/Function2;)Ljava/lang/Object;
public static synthetic fun runBlocking$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/lang/Object;
+ public static final fun withContext (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final synthetic fun withContext (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final fun withContext (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static synthetic fun withContext$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
@@ -135,10 +140,6 @@
}
public final class kotlinx/coroutines/experimental/CoroutineContextKt {
- public static final field DEBUG_PROPERTY_NAME Ljava/lang/String;
- public static final field DEBUG_PROPERTY_VALUE_AUTO Ljava/lang/String;
- public static final field DEBUG_PROPERTY_VALUE_OFF Ljava/lang/String;
- public static final field DEBUG_PROPERTY_VALUE_ON Ljava/lang/String;
public static final fun getDefaultDispatcher ()Lkotlinx/coroutines/experimental/CoroutineDispatcher;
public static final fun getIO ()Lkotlinx/coroutines/experimental/CoroutineDispatcher;
public static final fun newCoroutineContext (Lkotlin/coroutines/experimental/CoroutineContext;)Lkotlin/coroutines/experimental/CoroutineContext;
@@ -151,7 +152,7 @@
public fun <init> ()V
public abstract fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
public fun dispatchYield (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
- public fun interceptContinuation (Lkotlin/coroutines/experimental/Continuation;)Lkotlin/coroutines/experimental/Continuation;
+ public final fun interceptContinuation (Lkotlin/coroutines/experimental/Continuation;)Lkotlin/coroutines/experimental/Continuation;
public fun isDispatchNeeded (Lkotlin/coroutines/experimental/CoroutineContext;)Z
public final fun plus (Lkotlinx/coroutines/experimental/CoroutineDispatcher;)Lkotlinx/coroutines/experimental/CoroutineDispatcher;
public fun toString ()Ljava/lang/String;
@@ -223,6 +224,13 @@
public static fun values ()[Lkotlinx/coroutines/experimental/CoroutineStart;
}
+public final class kotlinx/coroutines/experimental/DebugKt {
+ public static final field DEBUG_PROPERTY_NAME Ljava/lang/String;
+ public static final field DEBUG_PROPERTY_VALUE_AUTO Ljava/lang/String;
+ public static final field DEBUG_PROPERTY_VALUE_OFF Ljava/lang/String;
+ public static final field DEBUG_PROPERTY_VALUE_ON Ljava/lang/String;
+}
+
public abstract interface class kotlinx/coroutines/experimental/Deferred : kotlinx/coroutines/experimental/Job {
public abstract fun await (Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public abstract fun getCompleted ()Ljava/lang/Object;
@@ -258,20 +266,27 @@
}
public abstract interface class kotlinx/coroutines/experimental/Delay {
- public abstract fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
- public abstract fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
- public abstract fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public abstract synthetic fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public abstract fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public abstract fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public abstract synthetic fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public abstract synthetic fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public abstract fun scheduleResumeAfterDelay (JLkotlinx/coroutines/experimental/CancellableContinuation;)V
}
public final class kotlinx/coroutines/experimental/Delay$DefaultImpls {
- public static fun delay (Lkotlinx/coroutines/experimental/Delay;JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static synthetic fun delay (Lkotlinx/coroutines/experimental/Delay;JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static fun delay (Lkotlinx/coroutines/experimental/Delay;JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static synthetic fun delay$default (Lkotlinx/coroutines/experimental/Delay;JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
- public static fun invokeOnTimeout (Lkotlinx/coroutines/experimental/Delay;JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public static fun invokeOnTimeout (Lkotlinx/coroutines/experimental/Delay;JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public static synthetic fun invokeOnTimeout (Lkotlinx/coroutines/experimental/Delay;JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public static synthetic fun scheduleResumeAfterDelay (Lkotlinx/coroutines/experimental/Delay;JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
}
public final class kotlinx/coroutines/experimental/DelayKt {
- public static final fun delay (ILkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final synthetic fun delay (ILkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static synthetic fun delay$default (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
}
@@ -337,37 +352,48 @@
public abstract class kotlinx/coroutines/experimental/ExecutorCoroutineDispatcher : kotlinx/coroutines/experimental/CloseableCoroutineDispatcher, java/io/Closeable {
public fun <init> ()V
+ public abstract fun close ()V
public abstract fun getExecutor ()Ljava/util/concurrent/Executor;
}
public abstract class kotlinx/coroutines/experimental/ExecutorCoroutineDispatcherBase : kotlinx/coroutines/experimental/ExecutorCoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
public fun <init> ()V
public fun close ()V
- public fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public synthetic fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
public fun equals (Ljava/lang/Object;)Z
public fun hashCode ()I
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
- public fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun scheduleResumeAfterDelay (JLkotlinx/coroutines/experimental/CancellableContinuation;)V
public fun toString ()Ljava/lang/String;
}
public final class kotlinx/coroutines/experimental/ExecutorsKt {
- public static final fun asCoroutineDispatcher (Ljava/util/concurrent/Executor;)Lkotlinx/coroutines/experimental/CoroutineDispatcher;
public static final synthetic fun asCoroutineDispatcher (Ljava/util/concurrent/ExecutorService;)Lkotlinx/coroutines/experimental/CloseableCoroutineDispatcher;
public static final fun asCoroutineDispatcher (Ljava/util/concurrent/ExecutorService;)Lkotlinx/coroutines/experimental/ExecutorCoroutineDispatcher;
+ public static final fun from (Ljava/util/concurrent/Executor;)Lkotlinx/coroutines/experimental/CoroutineDispatcher;
public static final fun toCoroutineDispatcher (Ljava/util/concurrent/Executor;)Lkotlinx/coroutines/experimental/CoroutineDispatcher;
}
+public abstract interface annotation class kotlinx/coroutines/experimental/ExperimentalCoroutinesApi : java/lang/annotation/Annotation {
+}
+
public final class kotlinx/coroutines/experimental/GlobalScope : kotlinx/coroutines/experimental/CoroutineScope {
public static final field INSTANCE Lkotlinx/coroutines/experimental/GlobalScope;
public fun getCoroutineContext ()Lkotlin/coroutines/experimental/CoroutineContext;
public synthetic fun isActive ()Z
}
+public abstract interface annotation class kotlinx/coroutines/experimental/InternalCoroutinesApi : java/lang/annotation/Annotation {
+}
+
public abstract interface class kotlinx/coroutines/experimental/Job : kotlin/coroutines/experimental/CoroutineContext$Element {
public static final field Key Lkotlinx/coroutines/experimental/Job$Key;
public abstract fun attachChild (Lkotlinx/coroutines/experimental/Job;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public abstract fun cancel ()Z
public abstract fun cancel (Ljava/lang/Throwable;)Z
public abstract synthetic fun cancelChildren (Ljava/lang/Throwable;)V
public abstract fun getCancellationException ()Ljava/util/concurrent/CancellationException;
@@ -413,11 +439,14 @@
}
public final class kotlinx/coroutines/experimental/JobKt {
+ public static final fun DisposableHandle (Lkotlin/jvm/functions/Function0;)Lkotlinx/coroutines/experimental/DisposableHandle;
public static final fun Job (Lkotlinx/coroutines/experimental/Job;)Lkotlinx/coroutines/experimental/Job;
public static synthetic fun Job$default (Lkotlinx/coroutines/experimental/Job;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/Job;
+ public static final fun cancel (Lkotlin/coroutines/experimental/CoroutineContext;)Z
public static final fun cancel (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Throwable;)Z
public static synthetic fun cancel$default (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Throwable;ILjava/lang/Object;)Z
public static final fun cancelAndJoin (Lkotlinx/coroutines/experimental/Job;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final fun cancelChildren (Lkotlin/coroutines/experimental/CoroutineContext;)V
public static final fun cancelChildren (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Throwable;)V
public static final fun cancelChildren (Lkotlinx/coroutines/experimental/Job;Ljava/lang/Throwable;)V
public static synthetic fun cancelChildren$default (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Throwable;ILjava/lang/Object;)V
@@ -439,6 +468,7 @@
public final class kotlinx/coroutines/experimental/NonCancellable : kotlin/coroutines/experimental/AbstractCoroutineContextElement, kotlinx/coroutines/experimental/Job {
public static final field INSTANCE Lkotlinx/coroutines/experimental/NonCancellable;
public fun attachChild (Lkotlinx/coroutines/experimental/Job;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public fun cancel ()Z
public fun cancel (Ljava/lang/Throwable;)Z
public synthetic fun cancelChildren (Ljava/lang/Throwable;)V
public fun getCancellationException ()Ljava/util/concurrent/CancellationException;
@@ -463,15 +493,18 @@
public fun toString ()Ljava/lang/String;
}
+public abstract interface annotation class kotlinx/coroutines/experimental/ObsoleteCoroutinesApi : java/lang/annotation/Annotation {
+}
+
public final class kotlinx/coroutines/experimental/RunnableKt {
public static final fun Runnable (Lkotlin/jvm/functions/Function0;)Ljava/lang/Runnable;
}
public final class kotlinx/coroutines/experimental/ScheduledKt {
- public static final fun withTimeout (ILkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final synthetic fun withTimeout (ILkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final fun withTimeout (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static synthetic fun withTimeout$default (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
- public static final fun withTimeoutOrNull (ILkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final synthetic fun withTimeoutOrNull (ILkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final fun withTimeoutOrNull (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static synthetic fun withTimeoutOrNull$default (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
}
@@ -500,10 +533,12 @@
}
public final class kotlinx/coroutines/experimental/ThreadPoolDispatcherKt {
- public static final fun newFixedThreadPoolContext (ILjava/lang/String;)Lkotlinx/coroutines/experimental/ThreadPoolDispatcher;
+ public static final fun newFixedThreadPoolContext (ILjava/lang/String;)Lkotlinx/coroutines/experimental/ExecutorCoroutineDispatcher;
+ public static final synthetic fun newFixedThreadPoolContext (ILjava/lang/String;)Lkotlinx/coroutines/experimental/ThreadPoolDispatcher;
public static final fun newFixedThreadPoolContext (ILjava/lang/String;Lkotlinx/coroutines/experimental/Job;)Lkotlin/coroutines/experimental/CoroutineContext;
public static synthetic fun newFixedThreadPoolContext$default (ILjava/lang/String;Lkotlinx/coroutines/experimental/Job;ILjava/lang/Object;)Lkotlin/coroutines/experimental/CoroutineContext;
- public static final fun newSingleThreadContext (Ljava/lang/String;)Lkotlinx/coroutines/experimental/ThreadPoolDispatcher;
+ public static final fun newSingleThreadContext (Ljava/lang/String;)Lkotlinx/coroutines/experimental/ExecutorCoroutineDispatcher;
+ public static final synthetic fun newSingleThreadContext (Ljava/lang/String;)Lkotlinx/coroutines/experimental/ThreadPoolDispatcher;
public static final fun newSingleThreadContext (Ljava/lang/String;Lkotlinx/coroutines/experimental/Job;)Lkotlin/coroutines/experimental/CoroutineContext;
public static synthetic fun newSingleThreadContext$default (Ljava/lang/String;Lkotlinx/coroutines/experimental/Job;ILjava/lang/Object;)Lkotlin/coroutines/experimental/CoroutineContext;
}
@@ -512,6 +547,11 @@
public fun <init> (Ljava/lang/String;)V
}
+public final class kotlinx/coroutines/experimental/TimeoutKt {
+ public static final fun withTimeout (JLkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final fun withTimeoutOrNull (JLkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+}
+
public final class kotlinx/coroutines/experimental/Unconfined : kotlinx/coroutines/experimental/CoroutineDispatcher {
public static final field INSTANCE Lkotlinx/coroutines/experimental/Unconfined;
public fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
@@ -525,6 +565,7 @@
public abstract class kotlinx/coroutines/experimental/channels/AbstractChannel : kotlinx/coroutines/experimental/channels/AbstractSendChannel, kotlinx/coroutines/experimental/channels/Channel {
public fun <init> ()V
+ public fun cancel ()Z
public fun cancel (Ljava/lang/Throwable;)Z
protected fun cleanupSendQueueOnCancel ()V
protected final fun describeTryPoll ()Lkotlinx/coroutines/experimental/channels/AbstractChannel$TryPollDesc;
@@ -691,8 +732,9 @@
}
public final class kotlinx/coroutines/experimental/channels/ChannelKt {
- public static final fun Channel ()Lkotlinx/coroutines/experimental/channels/Channel;
+ public static final synthetic fun Channel ()Lkotlinx/coroutines/experimental/channels/Channel;
public static final fun Channel (I)Lkotlinx/coroutines/experimental/channels/Channel;
+ public static synthetic fun Channel$default (IILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/Channel;
}
public final class kotlinx/coroutines/experimental/channels/ChannelsKt {
@@ -897,11 +939,13 @@
public static final fun produce (Lkotlin/coroutines/experimental/CoroutineContext;ILkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static final synthetic fun produce (Lkotlin/coroutines/experimental/CoroutineContext;ILkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static final fun produce (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;ILkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
+ public static final fun produce (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;ILkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static synthetic fun produce$default (Lkotlin/coroutines/experimental/CoroutineContext;ILkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static synthetic fun produce$default (Lkotlin/coroutines/experimental/CoroutineContext;ILkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ProducerJob;
public static synthetic fun produce$default (Lkotlin/coroutines/experimental/CoroutineContext;ILkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static synthetic fun produce$default (Lkotlin/coroutines/experimental/CoroutineContext;ILkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static synthetic fun produce$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;ILkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
+ public static synthetic fun produce$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;ILkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
}
public abstract interface class kotlinx/coroutines/experimental/channels/ProducerJob : kotlinx/coroutines/experimental/Job, kotlinx/coroutines/experimental/channels/ReceiveChannel {
@@ -926,6 +970,7 @@
}
public abstract interface class kotlinx/coroutines/experimental/channels/ReceiveChannel {
+ public abstract fun cancel ()Z
public abstract fun cancel (Ljava/lang/Throwable;)Z
public abstract fun getOnReceive ()Lkotlinx/coroutines/experimental/selects/SelectClause1;
public abstract fun getOnReceiveOrNull ()Lkotlinx/coroutines/experimental/selects/SelectClause1;
@@ -995,7 +1040,9 @@
}
public final class kotlinx/coroutines/experimental/channels/TickerChannelsKt {
+ public static final fun ticker (JJLkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/channels/TickerMode;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static final fun ticker (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/channels/TickerMode;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
+ public static synthetic fun ticker$default (JJLkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/channels/TickerMode;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
public static synthetic fun ticker$default (JLjava/util/concurrent/TimeUnit;JLkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/channels/TickerMode;ILjava/lang/Object;)Lkotlinx/coroutines/experimental/channels/ReceiveChannel;
}
@@ -1026,10 +1073,12 @@
public abstract fun invoke (Lkotlinx/coroutines/experimental/selects/SelectClause2;Ljava/lang/Object;Lkotlin/jvm/functions/Function2;)V
public abstract fun invoke (Lkotlinx/coroutines/experimental/selects/SelectClause2;Lkotlin/jvm/functions/Function2;)V
public abstract fun onTimeout (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function1;)V
+ public abstract fun onTimeout (JLkotlin/jvm/functions/Function1;)V
}
public final class kotlinx/coroutines/experimental/selects/SelectBuilder$DefaultImpls {
public static fun invoke (Lkotlinx/coroutines/experimental/selects/SelectBuilder;Lkotlinx/coroutines/experimental/selects/SelectClause2;Lkotlin/jvm/functions/Function2;)V
+ public static fun onTimeout (Lkotlinx/coroutines/experimental/selects/SelectBuilder;JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function1;)V
public static synthetic fun onTimeout$default (Lkotlinx/coroutines/experimental/selects/SelectBuilder;JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
}
@@ -1046,6 +1095,7 @@
public fun invoke (Lkotlinx/coroutines/experimental/selects/SelectClause2;Lkotlin/jvm/functions/Function2;)V
public fun isSelected ()Z
public fun onTimeout (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function1;)V
+ public fun onTimeout (JLkotlin/jvm/functions/Function1;)V
public fun performAtomicIfNotSelected (Lkotlinx/coroutines/experimental/internal/AtomicDesc;)Ljava/lang/Object;
public fun performAtomicTrySelect (Lkotlinx/coroutines/experimental/internal/AtomicDesc;)Ljava/lang/Object;
public fun resume (Ljava/lang/Object;)V
@@ -1087,10 +1137,7 @@
public fun invoke (Lkotlinx/coroutines/experimental/selects/SelectClause2;Ljava/lang/Object;Lkotlin/jvm/functions/Function2;)V
public fun invoke (Lkotlinx/coroutines/experimental/selects/SelectClause2;Lkotlin/jvm/functions/Function2;)V
public fun onTimeout (JLjava/util/concurrent/TimeUnit;Lkotlin/jvm/functions/Function1;)V
-}
-
-public final class kotlinx/coroutines/experimental/selects/WhileSelectKt {
- public static final fun whileSelect (Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun onTimeout (JLkotlin/jvm/functions/Function1;)V
}
public abstract interface class kotlinx/coroutines/experimental/sync/Mutex {
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-guava.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-guava.txt
index efb5ce2..49f14a7 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-guava.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-guava.txt
@@ -6,10 +6,12 @@
public static final fun future (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lcom/google/common/util/concurrent/ListenableFuture;
public static final synthetic fun future (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;)Lcom/google/common/util/concurrent/ListenableFuture;
public static final fun future (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Lcom/google/common/util/concurrent/ListenableFuture;
+ public static final fun future (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;)Lcom/google/common/util/concurrent/ListenableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lcom/google/common/util/concurrent/ListenableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lcom/google/common/util/concurrent/ListenableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lcom/google/common/util/concurrent/ListenableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lcom/google/common/util/concurrent/ListenableFuture;
public static synthetic fun future$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lcom/google/common/util/concurrent/ListenableFuture;
+ public static synthetic fun future$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lcom/google/common/util/concurrent/ListenableFuture;
}
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-javafx.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-javafx.txt
index 06955a4..69c98bb 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-javafx.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-javafx.txt
@@ -1,15 +1,18 @@
-public final class kotlinx/coroutines/experimental/javafx/JavaFx : kotlinx/coroutines/experimental/javafx/JavaFxDispatcher, kotlinx/coroutines/experimental/Delay {
+public final class kotlinx/coroutines/experimental/javafx/JavaFx : kotlinx/coroutines/experimental/javafx/JavaFxDispatcher {
public static final field INSTANCE Lkotlinx/coroutines/experimental/javafx/JavaFx;
public final fun awaitPulse (Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
- public fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public fun scheduleResumeAfterDelay (JLkotlinx/coroutines/experimental/CancellableContinuation;)V
public fun toString ()Ljava/lang/String;
}
public abstract class kotlinx/coroutines/experimental/javafx/JavaFxDispatcher : kotlinx/coroutines/experimental/CoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
- public fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
}
public final class kotlinx/coroutines/experimental/javafx/JavaFxDispatcherKt {
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-jdk8.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-jdk8.txt
index 94fb357..6019c23 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-jdk8.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-jdk8.txt
@@ -16,18 +16,21 @@
public static final fun future (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Ljava/util/concurrent/CompletableFuture;
public static final synthetic fun future (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;)Ljava/util/concurrent/CompletableFuture;
public static final fun future (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;)Ljava/util/concurrent/CompletableFuture;
+ public static final fun future (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;)Ljava/util/concurrent/CompletableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
public static synthetic fun future$default (Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlinx/coroutines/experimental/Job;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
public static synthetic fun future$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
+ public static synthetic fun future$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlinx/coroutines/experimental/CoroutineStart;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Ljava/util/concurrent/CompletableFuture;
public static final fun toCompletableFuture (Lkotlinx/coroutines/experimental/Deferred;)Ljava/util/concurrent/CompletableFuture;
}
public final class kotlinx/coroutines/experimental/time/TimeKt {
public static final fun delay (Ljava/time/Duration;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
- public static final fun onTimeout (Lkotlinx/coroutines/experimental/selects/SelectBuilder;Ljava/time/Duration;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public static final fun onTimeout (Lkotlinx/coroutines/experimental/selects/SelectBuilder;Ljava/time/Duration;Lkotlin/jvm/functions/Function1;)V
+ public static final synthetic fun onTimeout (Lkotlinx/coroutines/experimental/selects/SelectBuilder;Ljava/time/Duration;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final synthetic fun withTimeout (Ljava/time/Duration;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final fun withTimeout (Ljava/time/Duration;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public static final synthetic fun withTimeoutOrNull (Ljava/time/Duration;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-reactor.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-reactor.txt
index 1dbc637..83b8dc8 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-reactor.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-reactor.txt
@@ -1,10 +1,12 @@
public final class kotlinx/coroutines/experimental/reactor/ConvertKt {
public static final fun asFlux (Lkotlinx/coroutines/experimental/channels/ReceiveChannel;Lkotlin/coroutines/experimental/CoroutineContext;)Lreactor/core/publisher/Flux;
public static synthetic fun asFlux$default (Lkotlinx/coroutines/experimental/channels/ReceiveChannel;Lkotlin/coroutines/experimental/CoroutineContext;ILjava/lang/Object;)Lreactor/core/publisher/Flux;
+ public static final fun asMono (Lkotlinx/coroutines/experimental/Deferred;)Lreactor/core/publisher/Mono;
public static final fun asMono (Lkotlinx/coroutines/experimental/Deferred;Lkotlin/coroutines/experimental/CoroutineContext;)Lreactor/core/publisher/Mono;
+ public static final fun asMono (Lkotlinx/coroutines/experimental/Job;)Lreactor/core/publisher/Mono;
public static final fun asMono (Lkotlinx/coroutines/experimental/Job;Lkotlin/coroutines/experimental/CoroutineContext;)Lreactor/core/publisher/Mono;
- public static synthetic fun asMono$default (Lkotlinx/coroutines/experimental/Deferred;Lkotlin/coroutines/experimental/CoroutineContext;ILjava/lang/Object;)Lreactor/core/publisher/Mono;
- public static synthetic fun asMono$default (Lkotlinx/coroutines/experimental/Job;Lkotlin/coroutines/experimental/CoroutineContext;ILjava/lang/Object;)Lreactor/core/publisher/Mono;
+ public static final synthetic fun asMono$default (Lkotlinx/coroutines/experimental/Deferred;Lkotlin/coroutines/experimental/CoroutineContext;ILjava/lang/Object;)Lreactor/core/publisher/Mono;
+ public static final synthetic fun asMono$default (Lkotlinx/coroutines/experimental/Job;Lkotlin/coroutines/experimental/CoroutineContext;ILjava/lang/Object;)Lreactor/core/publisher/Mono;
}
public final class kotlinx/coroutines/experimental/reactor/FluxKt {
@@ -22,14 +24,18 @@
public static synthetic fun mono$default (Lkotlinx/coroutines/experimental/CoroutineScope;Lkotlin/coroutines/experimental/CoroutineContext;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lreactor/core/publisher/Mono;
}
-public class kotlinx/coroutines/experimental/reactor/SchedulerCoroutineDispatcher : kotlinx/coroutines/experimental/CoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
+public final class kotlinx/coroutines/experimental/reactor/SchedulerCoroutineDispatcher : kotlinx/coroutines/experimental/CoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
public fun <init> (Lreactor/core/scheduler/Scheduler;)V
- public fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public synthetic fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
public fun equals (Ljava/lang/Object;)Z
+ public final fun getScheduler ()Lreactor/core/scheduler/Scheduler;
public fun hashCode ()I
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
- public fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun scheduleResumeAfterDelay (JLkotlinx/coroutines/experimental/CancellableContinuation;)V
public fun toString ()Ljava/lang/String;
}
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-rx2.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-rx2.txt
index c4dca9b..237212b 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-rx2.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-rx2.txt
@@ -69,12 +69,16 @@
public final class kotlinx/coroutines/experimental/rx2/SchedulerCoroutineDispatcher : kotlinx/coroutines/experimental/CoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
public fun <init> (Lio/reactivex/Scheduler;)V
- public fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public synthetic fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
public fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
public fun equals (Ljava/lang/Object;)Z
+ public final fun getScheduler ()Lio/reactivex/Scheduler;
public fun hashCode ()I
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
- public fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun scheduleResumeAfterDelay (JLkotlinx/coroutines/experimental/CancellableContinuation;)V
public fun toString ()Ljava/lang/String;
}
diff --git a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-swing.txt b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-swing.txt
index 1a7f44d..dcb7f3b 100644
--- a/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-swing.txt
+++ b/binary-compatibility-validator/reference-public-api/kotlinx-coroutines-swing.txt
@@ -1,14 +1,17 @@
-public final class kotlinx/coroutines/experimental/swing/Swing : kotlinx/coroutines/experimental/swing/SwingDispatcher, kotlinx/coroutines/experimental/Delay {
+public final class kotlinx/coroutines/experimental/swing/Swing : kotlinx/coroutines/experimental/swing/SwingDispatcher {
public static final field INSTANCE Lkotlinx/coroutines/experimental/swing/Swing;
public fun dispatch (Lkotlin/coroutines/experimental/CoroutineContext;Ljava/lang/Runnable;)V
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
- public fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public fun scheduleResumeAfterDelay (JLkotlinx/coroutines/experimental/CancellableContinuation;)V
public fun toString ()Ljava/lang/String;
}
public abstract class kotlinx/coroutines/experimental/swing/SwingDispatcher : kotlinx/coroutines/experimental/CoroutineDispatcher, kotlinx/coroutines/experimental/Delay {
- public fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
- public fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun delay (JLjava/util/concurrent/TimeUnit;Lkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun delay (JLkotlin/coroutines/experimental/Continuation;)Ljava/lang/Object;
+ public fun invokeOnTimeout (JLjava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun invokeOnTimeout (JLjava/util/concurrent/TimeUnit;Ljava/lang/Runnable;)Lkotlinx/coroutines/experimental/DisposableHandle;
+ public synthetic fun scheduleResumeAfterDelay (JLjava/util/concurrent/TimeUnit;Lkotlinx/coroutines/experimental/CancellableContinuation;)V
}
public final class kotlinx/coroutines/experimental/swing/SwingDispatcherKt {
diff --git a/common/kotlinx-coroutines-core-common/src/AbstractCoroutine.kt b/common/kotlinx-coroutines-core-common/src/AbstractCoroutine.kt
index b6f840b..f674a5e 100644
--- a/common/kotlinx-coroutines-core-common/src/AbstractCoroutine.kt
+++ b/common/kotlinx-coroutines-core-common/src/AbstractCoroutine.kt
@@ -27,8 +27,11 @@
* @param parentContext context of the parent coroutine.
* @param active when `true` (by default) coroutine is created in _active_ state, when `false` in _new_ state.
* See [Job] for details.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
@Suppress("EXPOSED_SUPER_CLASS")
+@InternalCoroutinesApi
public abstract class AbstractCoroutine<in T>(
private val parentContext: CoroutineContext,
active: Boolean = true
diff --git a/common/kotlinx-coroutines-core-common/src/Annotations.kt b/common/kotlinx-coroutines-core-common/src/Annotations.kt
new file mode 100644
index 0000000..7ba8b34
--- /dev/null
+++ b/common/kotlinx-coroutines-core-common/src/Annotations.kt
@@ -0,0 +1,39 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental
+
+/**
+ * Marks declarations that are still **experimental** in coroutines API, which means that the design of the
+ * corresponding declarations has open issues which may (or may not) lead to their changes in the future.
+ * Roughly speaking, there is a chance that those declarations will be deprecated in the near future or
+ * the semantics of their behavior may change in some way that may break some code.
+ */
+@MustBeDocumented
+@Retention(value = AnnotationRetention.SOURCE)
+// todo: Experimental WARNING
+public annotation class ExperimentalCoroutinesApi
+
+
+/**
+ * Marks declarations that are **obsolete** in coroutines API, which means that the design of the corresponding
+ * declarations has serious known flaws and they must be redesigned in the future.
+ * Roughly speaking, these declarations will be deprecated in the future but there is no replacement for them yet,
+ * so they cannot be deprecated right away.
+ */
+@MustBeDocumented
+@Retention(value = AnnotationRetention.SOURCE)
+// todo: Experimental WARNING
+public annotation class ObsoleteCoroutinesApi
+
+/**
+ * Marks declarations that are **internal** in coroutines API, which means that should not be used outside of
+ * `kotlinx.coroutines`, because their signatures and semantics will be changing between release without any
+ * warnings and without providing any migration aids.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+@Retention(value = AnnotationRetention.SOURCE)
+// todo: Experimental ERROR
+public annotation class InternalCoroutinesApi
\ No newline at end of file
diff --git a/common/kotlinx-coroutines-core-common/src/Builders.common.kt b/common/kotlinx-coroutines-core-common/src/Builders.common.kt
index 4bfba1d..48d2d27 100644
--- a/common/kotlinx-coroutines-core-common/src/Builders.common.kt
+++ b/common/kotlinx-coroutines-core-common/src/Builders.common.kt
@@ -9,10 +9,11 @@
import kotlinx.coroutines.experimental.internal.*
import kotlinx.coroutines.experimental.intrinsics.*
+import kotlinx.coroutines.experimental.selects.*
import kotlin.coroutines.experimental.*
import kotlin.coroutines.experimental.intrinsics.*
-// --------------- basic coroutine builders ---------------
+// --------------- launch ---------------
/**
* Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
@@ -37,25 +38,34 @@
*
* @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
* @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
* @param block the coroutine code which will be invoked in the context of the provided scope.
**/
public fun CoroutineScope.launch(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
- onCompletion: CompletionHandler? = null,
block: suspend CoroutineScope.() -> Unit
): Job {
val newContext = newCoroutineContext(context)
val coroutine = if (start.isLazy)
LazyStandaloneCoroutine(newContext, block) else
StandaloneCoroutine(newContext, active = true)
- if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
coroutine.start(start, coroutine, block)
return coroutine
}
/**
+ * @suppress **Deprecated**: onCompletion parameter is deprecated.
+ */
+@Deprecated("onCompletion parameter is deprecated")
+public fun CoroutineScope.launch(
+ context: CoroutineContext = EmptyCoroutineContext,
+ start: CoroutineStart = CoroutineStart.DEFAULT,
+ onCompletion: CompletionHandler? = null,
+ block: suspend CoroutineScope.() -> Unit
+): Job =
+ launch(context, start, block).also { if (onCompletion != null) it.invokeOnCompletion(onCompletion) }
+
+/**
* Launches new coroutine without blocking current thread and returns a reference to the coroutine as a [Job].
* @suppress **Deprecated** Use [CoroutineScope.launch] instead.
*/
@@ -107,6 +117,63 @@
): Job =
GlobalScope.launch(context, start, block = block)
+// --------------- async ---------------
+
+/**
+ * Creates new coroutine and returns its future result as an implementation of [Deferred].
+ * The running coroutine is cancelled when the resulting deferred is [cancelled][Job.cancel].
+ *
+ * Coroutine context is inherited from a [CoroutineScope], additional context elements can be specified with [context] argument.
+ * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is used.
+ * The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
+ * with corresponding [coroutineContext] element.
+ *
+ * By default, the coroutine is immediately scheduled for execution.
+ * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
+ * An optional [start] parameter can be set to [CoroutineStart.LAZY] to start coroutine _lazily_. In this case,,
+ * the resulting [Deferred] is created in _new_ state. It can be explicitly started with [start][Job.start]
+ * function and will be started implicitly on the first invocation of [join][Job.join], [await][Deferred.await] or [awaitAll].
+ *
+ * @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
+ * @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
+ * @param block the coroutine code.
+ */
+public fun <T> CoroutineScope.async(
+ context: CoroutineContext = EmptyCoroutineContext,
+ start: CoroutineStart = CoroutineStart.DEFAULT,
+ block: suspend CoroutineScope.() -> T
+): Deferred<T> {
+ val newContext = newCoroutineContext(context)
+ val coroutine = if (start.isLazy)
+ LazyDeferredCoroutine(newContext, block) else
+ DeferredCoroutine<T>(newContext, active = true)
+ coroutine.start(start, coroutine, block)
+ return coroutine
+}
+
+@Suppress("UNCHECKED_CAST")
+private open class DeferredCoroutine<T>(
+ parentContext: CoroutineContext,
+ active: Boolean
+) : AbstractCoroutine<T>(parentContext, active), Deferred<T>, SelectClause1<T> {
+ override fun getCompleted(): T = getCompletedInternal() as T
+ override suspend fun await(): T = awaitInternal() as T
+ override val onAwait: SelectClause1<T> get() = this
+ override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
+ registerSelectClause1Internal(select, block)
+}
+
+private class LazyDeferredCoroutine<T>(
+ parentContext: CoroutineContext,
+ private val block: suspend CoroutineScope.() -> T
+) : DeferredCoroutine<T>(parentContext, active = false) {
+ override fun onStart() {
+ block.startCoroutineCancellable(this, this)
+ }
+}
+
+// --------------- withContext ---------------
+
/**
* @suppress **Deprecated**: Use `start = CoroutineStart.XXX` parameter
*/
@@ -122,17 +189,24 @@
* This function immediately applies dispatcher from the new context, shifting execution of the block into the
* different thread inside the block, and back when it completes.
* The specified [context] is added onto the current coroutine context for the execution of the block.
- *
- * An optional `start` parameter is used only if the specified `context` uses a different [CoroutineDispatcher] than
- * a current one, otherwise it is ignored.
- * By default, the coroutine is immediately scheduled for execution and can be cancelled
- * while it is waiting to be executed and it can be cancelled while the result is scheduled
- * to be processed by the invoker context.
- * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
- * A value of [CoroutineStart.LAZY] is not supported and produces [IllegalArgumentException].
*/
public suspend fun <T> withContext(
context: CoroutineContext,
+ block: suspend CoroutineScope.() -> T
+): T =
+ // todo: optimize fast-path to work without allocation (when there is a already a coroutine implementing scope)
+ withContextImpl(context, start = CoroutineStart.DEFAULT) {
+ currentScope {
+ block()
+ }
+ }
+
+/**
+ * @suppress **Deprecated**: start parameter is deprecated, no replacement.
+ */
+@Deprecated("start parameter is deprecated, no replacement")
+public suspend fun <T> withContext(
+ context: CoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
block: suspend CoroutineScope.() -> T
): T =
diff --git a/common/kotlinx-coroutines-core-common/src/CancellableContinuation.kt b/common/kotlinx-coroutines-core-common/src/CancellableContinuation.kt
index 9ee6ca1..deef82c 100644
--- a/common/kotlinx-coroutines-core-common/src/CancellableContinuation.kt
+++ b/common/kotlinx-coroutines-core-common/src/CancellableContinuation.kt
@@ -73,6 +73,7 @@
*
* @suppress **This is unstable API and it is subject to change.**
*/
+ @InternalCoroutinesApi
public fun tryResume(value: T, idempotent: Any? = null): Any?
/**
@@ -82,6 +83,7 @@
*
* @suppress **This is unstable API and it is subject to change.**
*/
+ @InternalCoroutinesApi
public fun tryResumeWithException(exception: Throwable): Any?
/**
@@ -89,12 +91,14 @@
*
* @suppress **This is unstable API and it is subject to change.**
*/
+ @InternalCoroutinesApi
public fun completeResume(token: Any)
/**
* Makes this continuation cancellable. Use it with `holdCancellability` optional parameter to
* [suspendCancellableCoroutine] function. It throws [IllegalStateException] if invoked more than once.
*/
+ @InternalCoroutinesApi
public fun initCancellability()
/**
@@ -146,7 +150,10 @@
* [dispatch][CoroutineDispatcher.dispatch] function of the [CoroutineDispatcher] in the [context].
* This function is designed to be used only by the [CoroutineDispatcher] implementations themselves.
* **It should not be used in general code**.
+ *
+ * **Note: This function is experimental.** Its signature general code may be changed in the future.
*/
+ @ExperimentalCoroutinesApi
public fun CoroutineDispatcher.resumeUndispatched(value: T)
/**
@@ -154,30 +161,37 @@
* [dispatch][CoroutineDispatcher.dispatch] function of the [CoroutineDispatcher] in the [context].
* This function is designed to be used only by the [CoroutineDispatcher] implementations themselves.
* **It should not be used in general code**.
+ *
+ * **Note: This function is experimental.** Its signature general code may be changed in the future.
*/
+ @ExperimentalCoroutinesApi
public fun CoroutineDispatcher.resumeUndispatchedWithException(exception: Throwable)
}
/**
* Suspends coroutine similar to [suspendCoroutine], but provide an implementation of [CancellableContinuation] to
* the [block]. This function throws [CancellationException] if the coroutine is cancelled or completed while suspended.
- *
- * If [holdCancellability] optional parameter is `true`, then the coroutine is suspended, but it is not
- * cancellable until [CancellableContinuation.initCancellability] is invoked.
- *
- * See [suspendAtomicCancellableCoroutine] for suspending functions that need *atomic cancellation*.
*/
public suspend inline fun <T> suspendCancellableCoroutine(
- holdCancellability: Boolean = false,
crossinline block: (CancellableContinuation<T>) -> Unit
): T =
suspendCoroutineUninterceptedOrReturn { uCont ->
val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)
- if (!holdCancellability) cancellable.initCancellability()
+ cancellable.initCancellability()
block(cancellable)
cancellable.getResult()
}
+@Suppress("DeprecatedCallableAddReplaceWith")
+@Deprecated(
+ level = DeprecationLevel.ERROR,
+ message = "holdCancellability is no longer supported, no replacement"
+)
+public suspend inline fun <T> suspendCancellableCoroutine(
+ holdCancellability: Boolean = false,
+ crossinline block: (CancellableContinuation<T>) -> Unit
+): T = error("holdCancellability is no longer supported, no replacement")
+
/**
* Suspends coroutine similar to [suspendCancellableCoroutine], but with *atomic cancellation*.
*
@@ -185,7 +199,10 @@
* As a side-effect of atomic cancellation, a thread-bound coroutine (to some UI thread, for example) may
* continue to execute even after it was cancelled from the same thread in the case when the continuation
* was already resumed and was posted for execution to the thread's queue.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public suspend inline fun <T> suspendAtomicCancellableCoroutine(
holdCancellability: Boolean = false,
crossinline block: (CancellableContinuation<T>) -> Unit
@@ -241,7 +258,10 @@
* ```
* invokeOnCancellation { handle.dispose() }
* ```
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun CancellableContinuation<*>.disposeOnCancellation(handle: DisposableHandle) =
invokeOnCancellation(handler = DisposeOnCancel(handle).asHandler)
diff --git a/common/kotlinx-coroutines-core-common/src/CoroutineContext.common.kt b/common/kotlinx-coroutines-core-common/src/CoroutineContext.common.kt
index 5eed6c1..5545599 100644
--- a/common/kotlinx-coroutines-core-common/src/CoroutineContext.common.kt
+++ b/common/kotlinx-coroutines-core-common/src/CoroutineContext.common.kt
@@ -8,6 +8,10 @@
public expect fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext
+/**
+ * The default [CoroutineDispatcher] that is used by all standard builders.
+ * @suppress **Deprecated**: Use [Dispatchers.Default].
+ */
@Suppress("PropertyName")
@Deprecated(
message = "Use Dispatchers.Default",
diff --git a/common/kotlinx-coroutines-core-common/src/CoroutineDispatcher.kt b/common/kotlinx-coroutines-core-common/src/CoroutineDispatcher.kt
index 296f8f8..2940f8d 100644
--- a/common/kotlinx-coroutines-core-common/src/CoroutineDispatcher.kt
+++ b/common/kotlinx-coroutines-core-common/src/CoroutineDispatcher.kt
@@ -59,7 +59,10 @@
* However, coroutine builders like [launch][CoroutineScope.launch] and [async][CoroutineScope.async] accept an optional [CoroutineStart]
* parameter that allows one to optionally choose C#-style [CoroutineStart.UNDISPATCHED] behaviour
* whenever it is needed for efficiency.
+ *
+ * **Note: This is an experimental api.** Execution semantics of coroutines may change in the future when this function returns `false`.
*/
+ @ExperimentalCoroutinesApi
public open fun isDispatchNeeded(context: CoroutineContext): Boolean = true
/**
@@ -74,13 +77,16 @@
*
* **Implementation note** though yield marker may be passed as a part of [context], this
* is a separate method for performance reasons
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+ @InternalCoroutinesApi
public open fun dispatchYield(context: CoroutineContext, block: Runnable) = dispatch(context, block)
/**
* Returns continuation that wraps the original [continuation], thus intercepting all resumptions.
*/
- public override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
+ public final override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> =
DispatchedContinuation(this, continuation)
/**
diff --git a/common/kotlinx-coroutines-core-common/src/CoroutineExceptionHandler.kt b/common/kotlinx-coroutines-core-common/src/CoroutineExceptionHandler.kt
index 82c4963..f03adec 100644
--- a/common/kotlinx-coroutines-core-common/src/CoroutineExceptionHandler.kt
+++ b/common/kotlinx-coroutines-core-common/src/CoroutineExceptionHandler.kt
@@ -22,6 +22,7 @@
* Otherwise all instances of [CoroutineExceptionHandler] found via [ServiceLoader] and [Thread.uncaughtExceptionHandler] are invoked
*/
@JvmOverloads // binary compatibility
+@InternalCoroutinesApi // todo: review KDoc use
public fun handleCoroutineException(context: CoroutineContext, exception: Throwable, caller: Job? = null) {
// if exception handling fails, make sure the original exception is not lost
try {
diff --git a/common/kotlinx-coroutines-core-common/src/CoroutineScope.kt b/common/kotlinx-coroutines-core-common/src/CoroutineScope.kt
index f3f0b3f..97a95ff 100644
--- a/common/kotlinx-coroutines-core-common/src/CoroutineScope.kt
+++ b/common/kotlinx-coroutines-core-common/src/CoroutineScope.kt
@@ -193,6 +193,7 @@
* Provides [CoroutineScope] that is already present in the current [coroutineContext] to the given [block].
* Note, this method doesn't wait for all launched children to complete (as opposed to [coroutineScope]).
*/
+@Deprecated("No replacement, usages are discouraged. Find another solution.")
public suspend inline fun <R> currentScope(block: CoroutineScope.() -> R): R =
CoroutineScope(coroutineContext).block()
diff --git a/common/kotlinx-coroutines-core-common/src/CoroutineStart.kt b/common/kotlinx-coroutines-core-common/src/CoroutineStart.kt
index e91aea3..4ce9df4 100644
--- a/common/kotlinx-coroutines-core-common/src/CoroutineStart.kt
+++ b/common/kotlinx-coroutines-core-common/src/CoroutineStart.kt
@@ -54,7 +54,10 @@
*
* Cancellability of coroutine at suspension points depends on the particular implementation details of
* suspending functions as in [DEFAULT].
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+ @InternalCoroutinesApi
ATOMIC,
/**
@@ -67,7 +70,10 @@
*
* Cancellability of coroutine at suspension points depends on the particular implementation details of
* suspending functions as in [DEFAULT].
+ *
+ * **Note: This is an experimental api.** Execution semantics of coroutines may change in the future when this mode is used.
*/
+ @ExperimentalCoroutinesApi
UNDISPATCHED;
/**
@@ -77,7 +83,10 @@
* * [ATOMIC] uses [startCoroutine].
* * [UNDISPATCHED] uses [startCoroutineUndispatched].
* * [LAZY] does nothing.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+ @InternalCoroutinesApi
public operator fun <T> invoke(block: suspend () -> T, completion: Continuation<T>) =
when (this) {
CoroutineStart.DEFAULT -> block.startCoroutineCancellable(completion)
@@ -93,7 +102,10 @@
* * [ATOMIC] uses [startCoroutine].
* * [UNDISPATCHED] uses [startCoroutineUndispatched].
* * [LAZY] does nothing.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+ @InternalCoroutinesApi
public operator fun <R, T> invoke(block: suspend R.() -> T, receiver: R, completion: Continuation<T>) =
when (this) {
CoroutineStart.DEFAULT -> block.startCoroutineCancellable(receiver, completion)
@@ -104,6 +116,9 @@
/**
* Returns `true` when [LAZY].
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+ @InternalCoroutinesApi
public val isLazy: Boolean get() = this === LAZY
}
diff --git a/common/kotlinx-coroutines-core-common/src/Deferred.kt b/common/kotlinx-coroutines-core-common/src/Deferred.kt
index cfc2ec3..964196b 100644
--- a/common/kotlinx-coroutines-core-common/src/Deferred.kt
+++ b/common/kotlinx-coroutines-core-common/src/Deferred.kt
@@ -102,7 +102,10 @@
*
* This function is designed to be used from [invokeOnCompletion] handlers, when there is an absolute certainty that
* the value is already complete. See also [getCompletionExceptionOrNull].
+ *
+ * **Note: This is an experimental api.** This function may be removed or renamed in the future.
*/
+ @ExperimentalCoroutinesApi
public fun getCompleted(): T
/**
@@ -112,7 +115,10 @@
*
* This function is designed to be used from [invokeOnCompletion] handlers, when there is an absolute certainty that
* the value is already complete. See also [getCompleted].
+ *
+ * **Note: This is an experimental api.** This function may be removed or renamed in the future.
*/
+ @ExperimentalCoroutinesApi
public fun getCompletionExceptionOrNull(): Throwable?
/**
@@ -123,39 +129,16 @@
}
/**
- * Creates new coroutine and returns its future result as an implementation of [Deferred].
- * The running coroutine is cancelled when the resulting deferred is [cancelled][Job.cancel].
- *
- * Coroutine context is inherited from a [CoroutineScope], additional context elements can be specified with [context] argument.
- * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is used.
- * The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
- * with corresponding [coroutineContext] element.
- *
- * By default, the coroutine is immediately scheduled for execution.
- * Other options can be specified via `start` parameter. See [CoroutineStart] for details.
- * An optional [start] parameter can be set to [CoroutineStart.LAZY] to start coroutine _lazily_. In this case,,
- * the resulting [Deferred] is created in _new_ state. It can be explicitly started with [start][Job.start]
- * function and will be started implicitly on the first invocation of [join][Job.join], [await][Deferred.await] or [awaitAll].
- *
- * @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
- * @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
- * @param block the coroutine code.
+ * @suppress **Deprecated**: onCompletion parameter is deprecated.
*/
+@Deprecated("onCompletion parameter is deprecated")
public fun <T> CoroutineScope.async(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
onCompletion: CompletionHandler? = null,
block: suspend CoroutineScope.() -> T
-): Deferred<T> {
- val newContext = newCoroutineContext(context)
- val coroutine = if (start.isLazy)
- LazyDeferredCoroutine(newContext, block) else
- DeferredCoroutine<T>(newContext, active = true)
- if (onCompletion != null) coroutine.invokeOnCompletion(handler = onCompletion)
- coroutine.start(start, coroutine, block)
- return coroutine
-}
+): Deferred<T> =
+ async(context, start, block).also { if (onCompletion != null) it.invokeOnCompletion(onCompletion) }
/**
* Creates new coroutine and returns its future result as an implementation of [Deferred].
@@ -225,23 +208,3 @@
public fun <T> defer(context: CoroutineContext, block: suspend CoroutineScope.() -> T): Deferred<T> =
GlobalScope.async(context, block = block)
-@Suppress("UNCHECKED_CAST")
-private open class DeferredCoroutine<T>(
- parentContext: CoroutineContext,
- active: Boolean
-) : AbstractCoroutine<T>(parentContext, active), Deferred<T>, SelectClause1<T> {
- override fun getCompleted(): T = getCompletedInternal() as T
- override suspend fun await(): T = awaitInternal() as T
- override val onAwait: SelectClause1<T> get() = this
- override fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (T) -> R) =
- registerSelectClause1Internal(select, block)
-}
-
-private class LazyDeferredCoroutine<T>(
- parentContext: CoroutineContext,
- private val block: suspend CoroutineScope.() -> T
-) : DeferredCoroutine<T>(parentContext, active = false) {
- override fun onStart() {
- block.startCoroutineCancellable(this, this)
- }
-}
diff --git a/common/kotlinx-coroutines-core-common/src/Delay.kt b/common/kotlinx-coroutines-core-common/src/Delay.kt
index c33e6d0..96dd6a4 100644
--- a/common/kotlinx-coroutines-core-common/src/Delay.kt
+++ b/common/kotlinx-coroutines-core-common/src/Delay.kt
@@ -14,21 +14,31 @@
*
* Implementation of this interface affects operation of
* [delay][kotlinx.coroutines.experimental.delay] and [withTimeout] functions.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi // todo: Remove references from other docs
public interface Delay {
+ @Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compat")
+ suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) = delay(time.convertToMillis(unit))
+
/**
* Delays coroutine for a given time without blocking a thread and resumes it after a 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].
*/
- suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
+ suspend fun delay(time: Long) {
if (time <= 0) return // don't delay
- return suspendCancellableCoroutine { scheduleResumeAfterDelay(time, unit, it) }
+ return suspendCancellableCoroutine { scheduleResumeAfterDelay(time, it) }
}
+ @Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compat")
+ fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) =
+ scheduleResumeAfterDelay(time.convertToMillis(unit), continuation)
+
/**
- * Schedules resume of a specified [continuation] after a specified delay [time].
+ * Schedules resume of a specified [continuation] after a specified delay [timeMillis].
*
* Continuation **must be scheduled** to resume even if it is already cancelled, because a cancellation is just
* an exception that the coroutine that used `delay` might wanted to catch and process. It might
@@ -42,19 +52,27 @@
* with(continuation) { resumeUndispatched(Unit) }
* ```
*/
- fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>)
+ fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>)
+
+ @Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compat")
+ fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
+ DefaultDelay.invokeOnTimeout(time.convertToMillis(unit), block)
/**
- * Schedules invocation of a specified [block] after a specified delay [time].
+ * Schedules invocation of a specified [block] after a specified delay [timeMillis].
* The resulting [DisposableHandle] can be used to [dispose][DisposableHandle.dispose] of this invocation
* request if it is not needed anymore.
*
* This implementation uses a built-in single-threaded scheduled executor service.
*/
- fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
- DefaultDelay.invokeOnTimeout(time, unit, block)
+ fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle =
+ DefaultDelay.invokeOnTimeout(timeMillis, block)
}
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compat")
+public suspend fun delay(timeMillis: Int) =
+ delay(timeMillis.toLong(), TimeUnit.MILLISECONDS)
+
/**
* Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
* This suspending function is cancellable.
@@ -66,10 +84,14 @@
* This function delegates to [Delay.scheduleResumeAfterDelay] if the context [CoroutineDispatcher]
* implements [Delay] interface, otherwise it resumes using a built-in single-threaded scheduled executor service.
*
- * @param time time in milliseconds.
+ * @param timeMillis time in milliseconds.
*/
-public suspend fun delay(time: Int) =
- delay(time.toLong(), TimeUnit.MILLISECONDS)
+public suspend fun delay(timeMillis: Long) {
+ if (timeMillis <= 0) return // don't delay
+ return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
+ cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
+ }
+}
/**
* Delays coroutine for a given time without blocking a thread and resumes it after a specified time.
@@ -84,11 +106,23 @@
*
* @param time time in the specified [unit].
* @param unit time unit.
+ *
+ * @suppress **Deprecated**: Replace with `delay(unit.toMillis(time))`
*/
-public suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) {
- if (time <= 0) return // don't delay
- return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->
- cont.context.delay.scheduleResumeAfterDelay(time, unit, cont)
+// todo: review usage in Guides and samples
+@Deprecated(
+ message = "Replace with delay(unit.toMillis(time))",
+ replaceWith = ReplaceWith("delay(unit.toMillis(time))")
+)
+public suspend fun delay(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS) =
+ delay(time.convertToMillis(unit))
+
+internal fun Long.convertToMillis(unit: TimeUnit): Long {
+ val result = unit.toMillis(this)
+ return when {
+ result != 0L -> result
+ this > 0 -> 1L
+ else -> 0L
}
}
diff --git a/common/kotlinx-coroutines-core-common/src/Dispatchers.common.kt b/common/kotlinx-coroutines-core-common/src/Dispatchers.common.kt
index c9aea8d..5426891 100644
--- a/common/kotlinx-coroutines-core-common/src/Dispatchers.common.kt
+++ b/common/kotlinx-coroutines-core-common/src/Dispatchers.common.kt
@@ -37,8 +37,12 @@
* an optional [CoroutineStart] parameter in coroutine builders like
* [launch][CoroutineScope.launch] and [async][CoroutineScope.async] setting it to the
* the value of [CoroutineStart.UNDISPATCHED].
+ *
+ * **Note: This is an experimental api.**
+ * Semantics, order of execution, and particular implementation details of this dispatcher may change in the future.
*/
@JvmField
+ @ExperimentalCoroutinesApi
public val Unconfined: CoroutineDispatcher =
kotlinx.coroutines.experimental.Unconfined
}
\ No newline at end of file
diff --git a/common/kotlinx-coroutines-core-common/src/Exceptions.common.kt b/common/kotlinx-coroutines-core-common/src/Exceptions.common.kt
index 2ea6103..5d5689b 100644
--- a/common/kotlinx-coroutines-core-common/src/Exceptions.common.kt
+++ b/common/kotlinx-coroutines-core-common/src/Exceptions.common.kt
@@ -4,10 +4,16 @@
package kotlinx.coroutines.experimental
+@InternalCoroutinesApi
public expect class CompletionHandlerException(message: String, cause: Throwable) : RuntimeException
public expect open class CancellationException(message: String?) : IllegalStateException
+/**
+ * @suppress **Deprecated**: Replace with [CancellationException].
+ */
+@InternalCoroutinesApi // todo: review usage from docs and examples
+@Deprecated(message = "Replace with CancellationException", replaceWith = ReplaceWith("CancellationException"))
public expect class JobCancellationException(
message: String,
cause: Throwable?,
diff --git a/common/kotlinx-coroutines-core-common/src/Job.kt b/common/kotlinx-coroutines-core-common/src/Job.kt
index 2540ada..eb3c67c 100644
--- a/common/kotlinx-coroutines-core-common/src/Job.kt
+++ b/common/kotlinx-coroutines-core-common/src/Job.kt
@@ -129,7 +129,10 @@
*
* This function throws [IllegalStateException] when invoked on a job that has not
* [completed][isCompleted] nor [cancelled][isCancelled] yet.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+ @InternalCoroutinesApi
public fun getCancellationException(): CancellationException
/**
@@ -149,6 +152,14 @@
public fun start(): Boolean
/**
+ * Cancels this job.
+ * The result is `true` if this job was either cancelled as a result of this invocation
+ * or was already being cancelled.
+ * If job is already completed, method returns `false`.
+ */
+ public fun cancel(): Boolean
+
+ /**
* Cancels this job with an optional cancellation [cause].
* The result is `true` if this job was either cancelled as a result of this invocation
* or it's being cancelled and given [cause] was successfully received by the job and will be properly handled, `false` otherwise.
@@ -159,7 +170,10 @@
* When cancellation has a clear reason in the code, an instance of [CancellationException] should be created
* at the corresponding original cancellation site and passed into this method to aid in debugging by providing
* both the context of cancellation and text description of the reason.
+ *
+ * **Note: This is an experimental api.** Cancellation of a job with exception may change its semantics in the future.
*/
+ @ExperimentalCoroutinesApi
public fun cancel(cause: Throwable? = null): Boolean
// ------------ parent-child ------------
@@ -203,7 +217,7 @@
*
* @suppress This is an internal API. This method is too error prone for public API.
*/
- @Deprecated(message = "Start child coroutine with 'parent' parameter", level = DeprecationLevel.WARNING)
+ @InternalCoroutinesApi
public fun attachChild(child: Job): DisposableHandle
/**
@@ -322,7 +336,10 @@
* then the [handler] is immediately and synchronously invoked and [NonDisposableHandle] is returned;
* when `false` then [NonDisposableHandle] is returned, but the [handler] is not invoked.
* @param handler the handler.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+ @InternalCoroutinesApi
public fun invokeOnCompletion(
onCancelling: Boolean = false,
invokeImmediately: Boolean = true,
@@ -361,6 +378,18 @@
public fun dispose()
}
+/**
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+@Suppress("FunctionName")
+@InternalCoroutinesApi
+public inline fun DisposableHandle(crossinline block: () -> Unit) =
+ object : DisposableHandle {
+ override fun dispose() {
+ block()
+ }
+ }
+
// -------------------- Job extensions --------------------
/**
@@ -384,7 +413,10 @@
* ```
* invokeOnCompletion { handle.dispose() }
* ```
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun Job.disposeOnCompletion(handle: DisposableHandle): DisposableHandle =
invokeOnCompletion(handler = DisposeOnCompletion(this, handle).asHandler)
@@ -420,7 +452,10 @@
* Suspends coroutine until all [children][Job.children] of this job are complete using
* [Job.join] for all of them. Unlike [Job.join] on this job as a whole, it does not wait until
* this job is complete.
+ *
+ * @suppress **Deprecated**: No replacement. Group child in a `coroutineScope { }` block to wait for them
*/
+@Deprecated("No replacement. Group child in a coroutineScope { } block to wait for them")
public suspend fun Job.joinChildren() {
children.forEach { it.join() }
}
@@ -447,20 +482,42 @@
get() = this[Job]?.isActive == true
/**
+ * Cancels [Job] of this context. The result is `true` if the job was
+ * cancelled as a result of this invocation or was already being cancelled and
+ * `false` if there is no job in the context or if it was already completed. See [Job.cancel] for details.
+ */
+public fun CoroutineContext.cancel(): Boolean =
+ this[Job]?.cancel() ?: false
+
+/**
* Cancels [Job] of this context with an optional cancellation [cause]. The result is `true` if the job was
* cancelled as a result of this invocation and `false` if there is no job in the context or if it was already
* cancelled or completed. See [Job.cancel] for details.
+ *
+ * **Note: This is an experimental api.** Cancellation of a job with exception may change its semantics in the future.
*/
+@ExperimentalCoroutinesApi
public fun CoroutineContext.cancel(cause: Throwable? = null): Boolean =
this[Job]?.cancel(cause) ?: false
/**
- * Cancels all children of the [Job] in this context with an optional cancellation [cause].
+ * Cancels all children of the [Job] in this context, without touching the the state of this job itself.
* It does not do anything if there is no job in the context or it has no children.
- * See [Job.cancelChildren] for details.
*/
+public fun CoroutineContext.cancelChildren() {
+ this[Job]?.children?.forEach { it.cancel() }
+}
+
+/**
+ * Cancels all children of the [Job] in this context with an optional cancellation [cause],
+ * without touching the the state of this job itself.
+ * It does not do anything if there is no job in the context or it has no children.
+ *
+ * **Note: This is an experimental api.** Cancellation of a job with exception may change its semantics in the future.
+ */
+@ExperimentalCoroutinesApi
public fun CoroutineContext.cancelChildren(cause: Throwable? = null) {
- this[Job]?.cancelChildren(cause)
+ this[Job]?.children?.forEach { it.cancel(cause) }
}
/**
@@ -472,7 +529,9 @@
/**
* No-op implementation of [DisposableHandle].
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi // todo: review references from KDocs
public object NonDisposableHandle : DisposableHandle {
/** Does not do anything. */
override fun dispose() {}
diff --git a/common/kotlinx-coroutines-core-common/src/JobSupport.kt b/common/kotlinx-coroutines-core-common/src/JobSupport.kt
index 5edeeef..0a42f51 100644
--- a/common/kotlinx-coroutines-core-common/src/JobSupport.kt
+++ b/common/kotlinx-coroutines-core-common/src/JobSupport.kt
@@ -572,6 +572,8 @@
*/
internal open val onCancelMode: Int get() = ON_CANCEL_MAKE_CANCELLING
+ public override fun cancel(): Boolean = cancel(null)
+
public override fun cancel(cause: Throwable?): Boolean = when (onCancelMode) {
ON_CANCEL_MAKE_CANCELLING -> makeCancelling(cause)
ON_CANCEL_MAKE_COMPLETING -> makeCompleting(Cancelled(this, cause))
diff --git a/common/kotlinx-coroutines-core-common/src/NonCancellable.kt b/common/kotlinx-coroutines-core-common/src/NonCancellable.kt
index 1356a93..bab42d3 100644
--- a/common/kotlinx-coroutines-core-common/src/NonCancellable.kt
+++ b/common/kotlinx-coroutines-core-common/src/NonCancellable.kt
@@ -20,60 +20,127 @@
* ```
*/
public object NonCancellable : AbstractCoroutineContextElement(Job), Job {
- /** Always returns `true`. */
+ /**
+ * Always returns `true`.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override val isActive: Boolean get() = true
- /** Always returns `false`. */
+ /**
+ * Always returns `false`.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override val isCompleted: Boolean get() = false
- /** Always returns `false`. */
+ /**
+ * Always returns `false`.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override val isCancelled: Boolean get() = false
- /** Always returns `false`. */
+ /**
+ * Always returns `false`.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override fun start(): Boolean = false
- /** Always throws [UnsupportedOperationException]. */
+ /**
+ * Always throws [UnsupportedOperationException].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override suspend fun join() {
throw UnsupportedOperationException("This job is always active")
}
+ /**
+ * Always throws [UnsupportedOperationException].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
override val onJoin: SelectClause0
get() = throw UnsupportedOperationException("This job is always active")
- /** Always throws [IllegalStateException]. */
+ /**
+ * Always throws [IllegalStateException].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override fun getCancellationException(): CancellationException = throw IllegalStateException("This job is always active")
- /** Always returns [NonDisposableHandle]. */
+ /**
+ * Always returns [NonDisposableHandle].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
@Suppress("OverridingDeprecatedMember")
+ @InternalCoroutinesApi
override fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle =
NonDisposableHandle
- /** Always returns [NonDisposableHandle]. */
+ /**
+ * Always returns [NonDisposableHandle].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
@Suppress("OverridingDeprecatedMember")
+ @InternalCoroutinesApi
override fun invokeOnCompletion(handler: CompletionHandler, onCancelling: Boolean): DisposableHandle =
NonDisposableHandle
- /** Always returns [NonDisposableHandle]. */
+ /**
+ * Always returns [NonDisposableHandle].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
@Suppress("OverridingDeprecatedMember")
+ @InternalCoroutinesApi
override fun invokeOnCompletion(onCancelling_: Boolean, handler: CompletionHandler): DisposableHandle =
NonDisposableHandle
- /** Always returns [NonDisposableHandle]. */
+ /**
+ * Always returns [NonDisposableHandle].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override fun invokeOnCompletion(onCancelling: Boolean, invokeImmediately: Boolean, handler: CompletionHandler): DisposableHandle =
NonDisposableHandle
- /** Always returns `false`. */
+ /**
+ * Always returns `false`.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
+ override fun cancel(): Boolean = false
+
+ /**
+ * Always returns `false`.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override fun cancel(cause: Throwable?): Boolean = false
- /** Always returns [emptySequence]. */
+ /**
+ * Always returns [emptySequence].
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
override val children: Sequence<Job>
get() = emptySequence()
- /** Always returns [NonDisposableHandle] and does not do anything. */
+ /**
+ * Always returns [NonDisposableHandle] and does not do anything.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
@Suppress("OverridingDeprecatedMember")
+ @InternalCoroutinesApi
override fun attachChild(child: Job): DisposableHandle = NonDisposableHandle
- /** Does not do anything. */
+ /**
+ * Does not do anything.
+ * @suppress **This an internal API and should not be used from general code.**
+ */
@Suppress("OverridingDeprecatedMember")
+ @InternalCoroutinesApi
override fun cancelChildren(cause: Throwable?) {}
}
diff --git a/common/kotlinx-coroutines-core-common/src/Scheduled.kt b/common/kotlinx-coroutines-core-common/src/Scheduled.kt
index baaa9cd..10e0306 100644
--- a/common/kotlinx-coroutines-core-common/src/Scheduled.kt
+++ b/common/kotlinx-coroutines-core-common/src/Scheduled.kt
@@ -4,32 +4,12 @@
package kotlinx.coroutines.experimental
-import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.intrinsics.*
import kotlinx.coroutines.experimental.selects.*
import kotlinx.coroutines.experimental.timeunit.*
-import kotlin.coroutines.experimental.*
-import kotlin.coroutines.experimental.intrinsics.*
-/**
- * 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.
- */
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compat")
public suspend fun <T> withTimeout(time: Int, block: suspend CoroutineScope.() -> T): T =
- withTimeout(time.toLong(), TimeUnit.MILLISECONDS, block)
+ withTimeout(time.toLong(), block)
/**
* Runs a given suspending [block] of code inside a coroutine with a specified timeout and throws
@@ -48,68 +28,18 @@
*
* @param time timeout time
* @param unit timeout unit (milliseconds by default)
+ *
+ * @suppress **Deprecated**: Replace with `withTimeout(unit.toMillis(time), block)`
*/
-public suspend fun <T> withTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T {
- if (time <= 0L) throw CancellationException("Timed out immediately")
- return suspendCoroutineUninterceptedOrReturn { uCont ->
- setupTimeout(TimeoutCoroutine(time, unit, uCont), block)
- }
-}
+// todo: review usage in Guides and samples
+@Deprecated(
+ message = "Replace with withTimeout(unit.toMillis(time), block)",
+ replaceWith = ReplaceWith("withTimeout(unit.toMillis(time), block)")
+)
+public suspend fun <T> withTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T =
+ withTimeout(time.convertToMillis(unit), 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.uCont
- val context = cont.context
- coroutine.disposeOnCompletion(context.delay.invokeOnTimeout(coroutine.time, coroutine.unit, coroutine))
- // restart block using new coroutine with new job,
- // however start it as undispatched coroutine, because we are already in the proper context
- return coroutine.startUndispatchedOrReturn(coroutine, block)
-}
-
-private open class TimeoutCoroutine<U, in T: U>(
- @JvmField val time: Long,
- @JvmField val unit: TimeUnit,
- @JvmField val uCont: Continuation<U> // unintercepted continuation
-) : AbstractCoroutine<T>(uCont.context, active = true), Runnable, Continuation<T> {
- override val defaultResumeMode: Int get() = MODE_DIRECT
-
- @Suppress("LeakingThis")
- override fun run() {
- cancel(TimeoutCancellationException(time, unit, this))
- }
-
- @Suppress("UNCHECKED_CAST")
- internal override fun onCompletionInternal(state: Any?, mode: Int) {
- if (state is CompletedExceptionally)
- uCont.resumeUninterceptedWithExceptionMode(state.cause, mode)
- else
- uCont.resumeUninterceptedMode(state as T, mode)
- }
-
- override fun nameString(): String =
- "${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].
- * **NB**: this method is exception-unsafe. Even if the code in the block suppresses [TimeoutCancellationException], this
- * invocation of `withTimeoutOrNull` still returns `null` and thrown exception will be ignored.
- *
- * 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.
- */
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compat")
public suspend fun <T> withTimeoutOrNull(time: Int, block: suspend CoroutineScope.() -> T): T? =
withTimeoutOrNull(time.toLong(), TimeUnit.MILLISECONDS, block)
@@ -130,43 +60,13 @@
*
* @param time timeout time
* @param unit timeout unit (milliseconds by default)
+ * @suppress **Deprecated**: Replace with `withTimeoutOrNull(unit.toMillis(time), block)`
*/
-public suspend fun <T> withTimeoutOrNull(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T? {
- if (time <= 0L) return null
+// todo: review usage in Guides and samples
+@Deprecated(
+ message = "Replace with withTimeoutOrNull(unit.toMillis(time), block)",
+ replaceWith = ReplaceWith("withTimeoutOrNull(unit.toMillis(time), block)")
+)
+public suspend fun <T> withTimeoutOrNull(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend CoroutineScope.() -> T): T? =
+ withTimeoutOrNull(time.convertToMillis(unit), block)
- // Workaround for K/N bug
- val array = arrayOfNulls<TimeoutCoroutine<T?, T?>>(1)
- try {
- return suspendCoroutineUninterceptedOrReturn { uCont ->
- val timeoutCoroutine = TimeoutCoroutine(time, unit, uCont)
- array[0] = timeoutCoroutine
- setupTimeout<T?, T?>(timeoutCoroutine, block)
- }
- } catch (e: TimeoutCancellationException) {
- // Return null iff it's our exception, otherwise propagate it upstream (e.g. in case of nested withTimeouts)
- if (e.coroutine === array[0]) {
- return null
- }
- throw e
- }
-}
-
-/**
- * This exception is thrown by [withTimeout] to indicate timeout.
- */
-public class TimeoutCancellationException internal constructor(
- message: String,
- @JvmField internal val coroutine: Job?
-) : CancellationException(message) {
- /**
- * Creates timeout exception with a given message.
- */
- public constructor(message: String) : this(message, null)
-}
-
-@Suppress("FunctionName")
-internal fun TimeoutCancellationException(
- time: Long,
- unit: TimeUnit,
- coroutine: Job
-) : TimeoutCancellationException = TimeoutCancellationException("Timed out waiting for $time $unit", coroutine)
diff --git a/common/kotlinx-coroutines-core-common/src/Timeout.kt b/common/kotlinx-coroutines-core-common/src/Timeout.kt
new file mode 100644
index 0000000..0403d83
--- /dev/null
+++ b/common/kotlinx-coroutines-core-common/src/Timeout.kt
@@ -0,0 +1,133 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental
+
+import kotlinx.coroutines.experimental.internal.*
+import kotlinx.coroutines.experimental.intrinsics.*
+import kotlinx.coroutines.experimental.selects.*
+import kotlin.coroutines.experimental.*
+import kotlin.coroutines.experimental.intrinsics.*
+
+/**
+ * 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 timeMillis timeout time in milliseconds.
+ */
+public suspend fun <T> withTimeout(timeMillis: Long, block: suspend CoroutineScope.() -> T): T {
+ if (timeMillis <= 0L) throw CancellationException("Timed out immediately")
+ return suspendCoroutineUninterceptedOrReturn { uCont ->
+ setupTimeout(TimeoutCoroutine(timeMillis, uCont), block)
+ }
+}
+
+/**
+ * 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].
+ * **NB**: this method is exception-unsafe. Even if the code in the block suppresses [TimeoutCancellationException], this
+ * invocation of `withTimeoutOrNull` still returns `null` and thrown exception will be ignored.
+ *
+ * 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 timeMillis timeout time in milliseconds.
+ */
+public suspend fun <T> withTimeoutOrNull(timeMillis: Long, block: suspend CoroutineScope.() -> T): T? {
+ if (timeMillis <= 0L) return null
+
+ // Workaround for K/N bug
+ val array = arrayOfNulls<TimeoutCoroutine<T?, T?>>(1)
+ try {
+ return suspendCoroutineUninterceptedOrReturn { uCont ->
+ val timeoutCoroutine = TimeoutCoroutine(timeMillis, uCont)
+ array[0] = timeoutCoroutine
+ setupTimeout<T?, T?>(timeoutCoroutine, block)
+ }
+ } catch (e: TimeoutCancellationException) {
+ // Return null iff it's our exception, otherwise propagate it upstream (e.g. in case of nested withTimeouts)
+ if (e.coroutine === array[0]) {
+ return null
+ }
+ throw e
+ }
+}
+
+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.uCont
+ val context = cont.context
+ coroutine.disposeOnCompletion(context.delay.invokeOnTimeout(coroutine.time, coroutine))
+ // restart block using new coroutine with new job,
+ // however start it as undispatched coroutine, because we are already in the proper context
+ return coroutine.startUndispatchedOrReturn(coroutine, block)
+}
+
+private open class TimeoutCoroutine<U, in T: U>(
+ @JvmField val time: Long,
+ @JvmField val uCont: Continuation<U> // unintercepted continuation
+) : AbstractCoroutine<T>(uCont.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")
+ internal override fun onCompletionInternal(state: Any?, mode: Int) {
+ if (state is CompletedExceptionally)
+ uCont.resumeUninterceptedWithExceptionMode(state.cause, mode)
+ else
+ uCont.resumeUninterceptedMode(state as T, mode)
+ }
+
+ override fun nameString(): String =
+ "${super.nameString()}(timeMillis=$time)"
+}
+
+/**
+ * This exception is thrown by [withTimeout] to indicate timeout.
+ */
+public class TimeoutCancellationException internal constructor(
+ message: String,
+ @JvmField internal val coroutine: Job?
+) : CancellationException(message) {
+ /**
+ * Creates timeout exception with a given message.
+ * This constructor is needed for exception stack-traces recovery.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
+ */
+ @InternalCoroutinesApi
+ public constructor(message: String) : this(message, null)
+}
+
+@Suppress("FunctionName")
+internal fun TimeoutCancellationException(
+ time: Long,
+ coroutine: Job
+) : TimeoutCancellationException = TimeoutCancellationException("Timed out waiting for $time ms", coroutine)
+
+
diff --git a/common/kotlinx-coroutines-core-common/src/channels/AbstractChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/AbstractChannel.kt
index 2a1aa61..201449b 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/AbstractChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/AbstractChannel.kt
@@ -13,7 +13,10 @@
/**
* Abstract send channel. It is a base class for all send channel implementations.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public abstract class AbstractSendChannel<E> : SendChannel<E> {
/** @suppress **This is unstable API and it is subject to change.** */
protected val queue = LockFreeLinkedListHead()
@@ -499,7 +502,10 @@
/**
* Abstract send/receive channel. It is a base class for all channel implementations.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public abstract class AbstractChannel<E> : AbstractSendChannel<E>(), Channel<E> {
// ------ extension points for buffered channels ------
@@ -654,6 +660,8 @@
return if (result === POLL_FAILED) null else receiveOrNullResult(result)
}
+ override fun cancel(): Boolean = cancel(null)
+
override fun cancel(cause: Throwable?): Boolean =
close(cause).also {
cleanupSendQueueOnCancel()
@@ -1024,6 +1032,7 @@
* Represents sending waiter in the queue.
* @suppress **This is unstable API and it is subject to change.**
*/
+@InternalCoroutinesApi
public interface Send {
val pollResult: Any? // E | Closed
fun tryResumeSend(idempotent: Any?): Any?
@@ -1035,6 +1044,7 @@
* Represents receiver waiter in the queue or closed token.
* @suppress **This is unstable API and it is subject to change.**
*/
+@InternalCoroutinesApi
public interface ReceiveOrClosed<in E> {
val offerResult: Any // OFFER_SUCCESS | Closed
fun tryResumeReceive(value: E, idempotent: Any?): Any?
@@ -1046,6 +1056,7 @@
* @suppress **This is unstable API and it is subject to change.**
*/
@Suppress("UNCHECKED_CAST")
+@InternalCoroutinesApi
public class SendElement(
override val pollResult: Any?,
@JvmField val cont: CancellableContinuation<Unit>
@@ -1060,6 +1071,7 @@
* Represents closed channel.
* @suppress **This is unstable API and it is subject to change.**
*/
+@InternalCoroutinesApi
public class Closed<in E>(
@JvmField val closeCause: Throwable?
) : LockFreeLinkedListNode(), Send, ReceiveOrClosed<E> {
diff --git a/common/kotlinx-coroutines-core-common/src/channels/ArrayBroadcastChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/ArrayBroadcastChannel.kt
index 145bd81..5b31537 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/ArrayBroadcastChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/ArrayBroadcastChannel.kt
@@ -4,6 +4,7 @@
package kotlinx.coroutines.experimental.channels
+import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.internal.*
import kotlinx.coroutines.experimental.selects.*
@@ -20,8 +21,16 @@
* This implementation uses lock to protect the buffer, which is held only during very short buffer-update operations.
* The lock at each subscription is also used to manage concurrent attempts to receive from the same subscriber.
* The lists of suspended senders or receivers are lock-free.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
-class ArrayBroadcastChannel<E>(
+@InternalCoroutinesApi // todo: review KDoc refs
+public class ArrayBroadcastChannel<E>
+@Deprecated(
+ "Replace with BroadcastChannel factory function",
+ replaceWith = ReplaceWith("BroadcastChannel(capacity)")
+)
+constructor(
/**
* Buffer capacity.
*/
diff --git a/common/kotlinx-coroutines-core-common/src/channels/ArrayChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/ArrayChannel.kt
index 04a55be..5809a50 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/ArrayChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/ArrayChannel.kt
@@ -4,6 +4,7 @@
package kotlinx.coroutines.experimental.channels
+import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.internal.*
import kotlinx.coroutines.experimental.selects.*
@@ -15,8 +16,15 @@
*
* This implementation uses lock to protect the buffer, which is held only during very short buffer-update operations.
* The lists of suspended senders or receivers are lock-free.
+ * @suppress **This an internal API and should not be used from general code.**
*/
-public open class ArrayChannel<E>(
+@InternalCoroutinesApi // todo: review KDoc refs
+public open class ArrayChannel<E>
+@Deprecated(
+ "Replace with Channel factory function",
+ replaceWith = ReplaceWith("Channel(capacity)")
+)
+constructor(
/**
* Buffer capacity.
*/
diff --git a/common/kotlinx-coroutines-core-common/src/channels/Broadcast.kt b/common/kotlinx-coroutines-core-common/src/channels/Broadcast.kt
index c35e39a..0e72e64 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/Broadcast.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/Broadcast.kt
@@ -111,6 +111,7 @@
override val channel: SendChannel<E>
get() = this
+ override fun cancel(): Boolean = super.cancel()
public override fun cancel(cause: Throwable?): Boolean = super.cancel(cause)
override fun onCancellationInternal(exceptionally: CompletedExceptionally?) {
diff --git a/common/kotlinx-coroutines-core-common/src/channels/BroadcastChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/BroadcastChannel.kt
index 40f489f..2c16616 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/BroadcastChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/BroadcastChannel.kt
@@ -16,12 +16,16 @@
*
* See `BroadcastChannel()` factory function for the description of available
* broadcast channel implementations.
+ *
+ * **Note: This is an experimental api.** It may be changed in the future updates.
*/
+@ExperimentalCoroutinesApi
public interface BroadcastChannel<E> : SendChannel<E> {
/**
* Factory for broadcast channels.
* @suppress **Deprecated**
*/
+ @Deprecated(message = "No replacement")
public companion object Factory {
/**
* Creates a broadcast channel with the specified buffer capacity.
@@ -73,7 +77,10 @@
* **Note:** this channel looses all items that are send to it until the first subscriber appears;
* * when `capacity` is [CONFLATED] -- creates [ConflatedBroadcastChannel] that conflates back-to-back sends;
* * otherwise -- throws [IllegalArgumentException].
+ *
+ * **Note: This is an experimental api.** It may be changed in the future updates.
*/
+@ExperimentalCoroutinesApi
public fun <E> BroadcastChannel(capacity: Int): BroadcastChannel<E> =
when (capacity) {
0 -> throw IllegalArgumentException("Unsupported 0 capacity for BroadcastChannel")
diff --git a/common/kotlinx-coroutines-core-common/src/channels/Channel.kt b/common/kotlinx-coroutines-core-common/src/channels/Channel.kt
index 797a1c0..d4d5b8a 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/Channel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/Channel.kt
@@ -7,6 +7,7 @@
import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.Channel.Factory.CONFLATED
import kotlinx.coroutines.experimental.channels.Channel.Factory.UNLIMITED
+import kotlinx.coroutines.experimental.internal.*
import kotlinx.coroutines.experimental.selects.*
/**
@@ -16,13 +17,19 @@
/**
* Returns `true` if this channel was closed by invocation of [close] and thus
* the [send] and [offer] attempts throws exception.
+ *
+ * **Note: This is an experimental api.** This property may change its semantics and/or name in the future.
*/
+ @ExperimentalCoroutinesApi
public val isClosedForSend: Boolean
/**
* Returns `true` if the channel is full (out of capacity) and the [send] attempt will suspend.
* This function returns `false` for [isClosedForSend] channel.
+ *
+ * **Note: This is an experimental api.** This property may change its semantics and/or name in the future.
*/
+ @ExperimentalCoroutinesApi
public val isFull: Boolean
/**
@@ -68,7 +75,7 @@
public fun offer(element: E): Boolean
/**
- * Closes this channel with an optional exceptional [cause].
+ * Closes this channel.
* This is an idempotent operation -- repeated invocations of this function have no effect and return `false`.
* Conceptually, its sends a special "close token" over this channel.
*
@@ -110,11 +117,14 @@
*
* ```
*
+ * **Note: This is an experimental api.** This function may change its semantics, parameters or return type in the future.
+ *
* @throws UnsupportedOperationException if underlying channel doesn't support [invokeOnClose].
* Implementation note: currently, [invokeOnClose] is unsupported only by Rx-like integrations
*
* @throws IllegalStateException if another handler was already registered
*/
+ @ExperimentalCoroutinesApi
public fun invokeOnClose(handler: (cause: Throwable?) -> Unit)
}
@@ -128,13 +138,19 @@
* throws [ClosedReceiveChannelException]. If the channel was closed because of the exception, it
* is considered closed, too, but it is called a _failed_ channel. All suspending attempts to receive
* an element from a failed channel throw the original [close][SendChannel.close] cause exception.
+ *
+ * **Note: This is an experimental api.** This property may change its semantics and/or name in the future.
*/
+ @ExperimentalCoroutinesApi
public val isClosedForReceive: Boolean
/**
* Returns `true` if the channel is empty (contains no elements) and the [receive] attempt will suspend.
* This function returns `false` for [isClosedForReceive] channel.
+ *
+ * **Note: This is an experimental api.** This property may change its semantics and/or name in the future.
*/
+ @ExperimentalCoroutinesApi
public val isEmpty: Boolean
/**
@@ -187,7 +203,10 @@
*
* This function can be used in [select] invocation with [onReceiveOrNull] clause.
* Use [poll] to try receiving from this channel without waiting.
+ *
+ * **Note: This is an experimental api.** This function may be replaced with a better on in the future.
*/
+ @ExperimentalCoroutinesApi
public suspend fun receiveOrNull(): E?
/**
@@ -195,7 +214,10 @@
* is received from the channel or selects with `null` if if the channel
* [isClosedForReceive] without cause. The [select] invocation fails with
* the original [close][SendChannel.close] cause exception if the channel has _failed_.
+ *
+ * **Note: This is an experimental api.** This function may be replaced with a better on in the future.
*/
+ @ExperimentalCoroutinesApi
public val onReceiveOrNull: SelectClause1<E?>
/**
@@ -213,6 +235,19 @@
public operator fun iterator(): ChannelIterator<E>
/**
+ * Cancels reception of remaining elements from this channel. This function closes the channel
+ * and removes all buffered sent elements from it.
+ * This function returns `true` if the channel was not closed previously, or `false` otherwise.
+ *
+ * Immediately after invocation of this function [isClosedForReceive] and
+ * [isClosedForSend][SendChannel.isClosedForSend]
+ * on the side of [SendChannel] start returning `true`, so all attempts to send to this channel
+ * afterwards will throw [ClosedSendChannelException], while attempts to receive will throw
+ * [ClosedReceiveChannelException].
+ */
+ public fun cancel(): Boolean
+
+ /**
* Cancels reception of remaining elements from this channel. This function closes the channel with
* the specified cause (unless it was already closed) and removes all buffered sent elements from it.
* This function returns `true` if the channel was not closed previously, or `false` otherwise.
@@ -224,7 +259,11 @@
* [ClosedReceiveChannelException] if it was cancelled without a cause.
* A channel that was cancelled with non-null [cause] is called a _failed_ channel. Attempts to send or
* receive on a failed channel throw the specified [cause] exception.
+ *
+ * **Note: This is an experimental api.** Semantics of _cancelling_ a channel with exception may
+ * change in the future or this feature may be completely removed.
*/
+ @ExperimentalCoroutinesApi
public fun cancel(cause: Throwable? = null): Boolean
}
@@ -313,6 +352,7 @@
/**
* Creates a channel without a buffer -- [RendezvousChannel].
*/
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "binary compat")
public fun <E> Channel(): Channel<E> = RendezvousChannel<E>()
/**
@@ -325,7 +365,8 @@
* * when `capacity` is positive, but less than [UNLIMITED] -- creates [ArrayChannel] with a buffer of the specified `capacity`;
* * otherwise -- throws [IllegalArgumentException].
*/
-public fun <E> Channel(capacity: Int): Channel<E> =
+// todo: docs
+public fun <E> Channel(capacity: Int = 0): Channel<E> =
when (capacity) {
0 -> RendezvousChannel()
UNLIMITED -> LinkedListChannel()
@@ -345,4 +386,5 @@
* channel that was closed without a cause. A _failed_ channel rethrows the original [close][SendChannel.close] cause
* exception on receive attempts.
*/
+// todo: explain when this exception is thrown
public class ClosedReceiveChannelException(message: String?) : NoSuchElementException(message)
diff --git a/common/kotlinx-coroutines-core-common/src/channels/ChannelCoroutine.kt b/common/kotlinx-coroutines-core-common/src/channels/ChannelCoroutine.kt
index 598063e..1e3c4d0 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/ChannelCoroutine.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/ChannelCoroutine.kt
@@ -14,5 +14,6 @@
) : AbstractCoroutine<Unit>(parentContext, active), Channel<E> by _channel {
val channel: Channel<E> get() = this
+ override fun cancel(): Boolean = super.cancel()
override fun cancel(cause: Throwable?): Boolean = super.cancel(cause)
}
diff --git a/common/kotlinx-coroutines-core-common/src/channels/Channels.common.kt b/common/kotlinx-coroutines-core-common/src/channels/Channels.common.kt
index 8ad0db7..f52a3ea 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/Channels.common.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/Channels.common.kt
@@ -18,6 +18,7 @@
/**
* Returns a channel to read all element of the [Iterable].
*/
+@Deprecated("No replacement")
public fun <E> Iterable<E>.asReceiveChannel(context: CoroutineContext = Dispatchers.Unconfined): ReceiveChannel<E> =
GlobalScope.produce(context) {
for (element in this@asReceiveChannel)
@@ -27,6 +28,7 @@
/**
* Returns a channel to read all element of the [Sequence].
*/
+@Deprecated("No replacement")
public fun <E> Sequence<E>.asReceiveChannel(context: CoroutineContext = Dispatchers.Unconfined): ReceiveChannel<E> =
GlobalScope.produce(context) {
for (element in this@asReceiveChannel)
@@ -38,7 +40,11 @@
/**
* Opens subscription to this [BroadcastChannel] and makes sure that the given [block] consumes all elements
* from it by always invoking [cancel][ReceiveChannel.cancel] after the execution of the block.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public inline fun <E, R> BroadcastChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
val channel = openSubscription()
try {
@@ -50,7 +56,11 @@
/**
* Subscribes to this [BroadcastChannel] and performs the specified action for each received element.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> BroadcastChannel<E>.consumeEach(action: (E) -> Unit) =
consume {
for (element in this) action(element)
@@ -74,7 +84,11 @@
* immediately throws an [IllegalStateException].
* See [this issue](https://github.com/Kotlin/kotlinx.coroutines/issues/167)
* for details.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public fun ReceiveChannel<*>.consumes(): CompletionHandler =
{ cause: Throwable? -> cancel(cause) }
@@ -82,7 +96,11 @@
* Returns a [CompletionHandler] that invokes [cancel][ReceiveChannel.cancel] on all the
* specified [ReceiveChannel] instances with the corresponding cause.
* See also [ReceiveChannel.consumes()] for a version on one channel.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public fun consumesAll(vararg channels: ReceiveChannel<*>): CompletionHandler =
{ cause: Throwable? ->
var exception: Throwable? = null
@@ -110,7 +128,11 @@
* for details.
*
* The operation is _terminal_.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
var cause: Throwable? = null
try {
@@ -134,7 +156,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit) =
consume {
for (e in this) action(e)
@@ -152,7 +178,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.consumeEachIndexed(action: (IndexedValue<E>) -> Unit) {
var index = 0
consumeEach {
@@ -165,7 +195,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.elementAt(index: Int): E =
elementAtOrElse(index) { throw IndexOutOfBoundsException("ReceiveChannel doesn't contain element at index $index.") }
@@ -174,7 +208,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.elementAtOrElse(index: Int, defaultValue: (Int) -> E): E =
consume {
if (index < 0)
@@ -192,7 +230,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.elementAtOrNull(index: Int): E? =
consume {
if (index < 0)
@@ -210,7 +252,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.find(predicate: (E) -> Boolean): E? =
firstOrNull(predicate)
@@ -219,7 +265,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.findLast(predicate: (E) -> Boolean): E? =
lastOrNull(predicate)
@@ -229,7 +279,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.first(): E =
consume {
val iterator = iterator()
@@ -244,7 +298,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.first(predicate: (E) -> Boolean): E {
consumeEach {
if (predicate(it)) return it
@@ -257,7 +315,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.firstOrNull(): E? =
consume {
val iterator = iterator()
@@ -271,7 +333,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public inline suspend fun <E> ReceiveChannel<E>.firstOrNull(predicate: (E) -> Boolean): E? {
consumeEach {
if (predicate(it)) return it
@@ -284,7 +350,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.indexOf(element: E): Int {
var index = 0
consumeEach {
@@ -300,7 +370,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.indexOfFirst(predicate: (E) -> Boolean): Int {
var index = 0
consumeEach {
@@ -316,7 +390,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public inline suspend fun <E> ReceiveChannel<E>.indexOfLast(predicate: (E) -> Boolean): Int {
var lastIndex = -1
var index = 0
@@ -334,7 +412,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.last(): E =
consume {
val iterator = iterator()
@@ -352,7 +434,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.last(predicate: (E) -> Boolean): E {
var last: E? = null
var found = false
@@ -372,7 +458,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.lastIndexOf(element: E): Int {
var lastIndex = -1
var index = 0
@@ -389,7 +479,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.lastOrNull(): E? =
consume {
val iterator = iterator()
@@ -406,7 +500,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.lastOrNull(predicate: (E) -> Boolean): E? {
var last: E? = null
consumeEach {
@@ -422,7 +520,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.single(): E =
consume {
val iterator = iterator()
@@ -439,7 +541,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.single(predicate: (E) -> Boolean): E {
var single: E? = null
var found = false
@@ -460,7 +566,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.singleOrNull(): E? =
consume {
val iterator = iterator()
@@ -477,7 +587,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.singleOrNull(predicate: (E) -> Boolean): E? {
var single: E? = null
var found = false
@@ -497,7 +611,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public fun <E> ReceiveChannel<E>.drop(n: Int, context: CoroutineContext = Dispatchers.Unconfined): ReceiveChannel<E> =
GlobalScope.produce(context, onCompletion = consumes()) {
require(n >= 0) { "Requested element count $n is less than zero." }
@@ -518,7 +636,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.dropWhile(context: CoroutineContext = Dispatchers.Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
GlobalScope.produce(context, onCompletion = consumes()) {
@@ -538,7 +660,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.filter(context: CoroutineContext = Dispatchers.Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
GlobalScope.produce(context, onCompletion = consumes()) {
@@ -554,7 +680,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.filterIndexed(context: CoroutineContext = Dispatchers.Unconfined, predicate: suspend (index: Int, E) -> Boolean): ReceiveChannel<E> =
GlobalScope.produce(context, onCompletion = consumes()) {
@@ -571,7 +701,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterIndexedTo(destination: C, predicate: (index: Int, E) -> Boolean): C {
consumeEachIndexed { (index, element) ->
if (predicate(index, element)) destination.add(element)
@@ -586,7 +720,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterIndexedTo(destination: C, predicate: (index: Int, E) -> Boolean): C {
consumeEachIndexed { (index, element) ->
if (predicate(index, element)) destination.send(element)
@@ -599,7 +737,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.filterNot(context: CoroutineContext = Dispatchers.Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
filter(context) { !predicate(it) }
@@ -615,7 +757,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
@Suppress("UNCHECKED_CAST")
public fun <E : Any> ReceiveChannel<E?>.filterNotNull(): ReceiveChannel<E> =
filter { it != null } as ReceiveChannel<E>
@@ -625,7 +771,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E : Any, C : MutableCollection<in E>> ReceiveChannel<E?>.filterNotNullTo(destination: C): C {
consumeEach {
if (it != null) destination.add(it)
@@ -638,7 +788,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E : Any, C : SendChannel<E>> ReceiveChannel<E?>.filterNotNullTo(destination: C): C {
consumeEach {
if (it != null) destination.send(it)
@@ -651,7 +805,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterNotTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (!predicate(it)) destination.add(it)
@@ -664,7 +822,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterNotTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (!predicate(it)) destination.send(it)
@@ -677,7 +839,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.filterTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (predicate(it)) destination.add(it)
@@ -690,7 +856,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, C : SendChannel<E>> ReceiveChannel<E>.filterTo(destination: C, predicate: (E) -> Boolean): C {
consumeEach {
if (predicate(it)) destination.send(it)
@@ -703,7 +873,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public fun <E> ReceiveChannel<E>.take(n: Int, context: CoroutineContext = Dispatchers.Unconfined): ReceiveChannel<E> =
GlobalScope.produce(context, onCompletion = consumes()) {
if (n == 0) return@produce
@@ -722,7 +896,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E> ReceiveChannel<E>.takeWhile(context: CoroutineContext = Dispatchers.Unconfined, predicate: suspend (E) -> Boolean): ReceiveChannel<E> =
GlobalScope.produce(context, onCompletion = consumes()) {
@@ -742,7 +920,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, V> ReceiveChannel<E>.associate(transform: (E) -> Pair<K, V>): Map<K, V> =
associateTo(LinkedHashMap(), transform)
@@ -756,7 +938,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K> ReceiveChannel<E>.associateBy(keySelector: (E) -> K): Map<K, E> =
associateByTo(LinkedHashMap(), keySelector)
@@ -769,7 +955,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, V> ReceiveChannel<E>.associateBy(keySelector: (E) -> K, valueTransform: (E) -> V): Map<K, V> =
associateByTo(LinkedHashMap(), keySelector, valueTransform)
@@ -782,7 +972,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, M : MutableMap<in K, in E>> ReceiveChannel<E>.associateByTo(destination: M, keySelector: (E) -> K): M {
consumeEach {
destination.put(keySelector(it), it)
@@ -799,7 +993,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, V, M : MutableMap<in K, in V>> ReceiveChannel<E>.associateByTo(destination: M, keySelector: (E) -> K, valueTransform: (E) -> V): M {
consumeEach {
destination.put(keySelector(it), valueTransform(it))
@@ -815,7 +1013,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, V, M : MutableMap<in K, in V>> ReceiveChannel<E>.associateTo(destination: M, transform: (E) -> Pair<K, V>): M {
consumeEach {
destination += transform(it)
@@ -829,7 +1031,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E, C : SendChannel<E>> ReceiveChannel<E>.toChannel(destination: C): C {
consumeEach {
destination.send(it)
@@ -842,7 +1048,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E, C : MutableCollection<in E>> ReceiveChannel<E>.toCollection(destination: C): C {
consumeEach {
destination.add(it)
@@ -864,7 +1074,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <K, V> ReceiveChannel<Pair<K, V>>.toMap(): Map<K, V> =
toMap(LinkedHashMap())
@@ -873,7 +1087,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <K, V, M : MutableMap<in K, in V>> ReceiveChannel<Pair<K, V>>.toMap(destination: M): M {
consumeEach {
destination += it
@@ -886,7 +1104,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.toMutableList(): MutableList<E> =
toCollection(ArrayList())
@@ -897,7 +1119,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.toSet(): Set<E> =
this.toMutableSet()
@@ -906,7 +1132,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R> ReceiveChannel<E>.flatMap(context: CoroutineContext = Dispatchers.Unconfined, transform: suspend (E) -> ReceiveChannel<R>): ReceiveChannel<R> =
GlobalScope.produce(context, onCompletion = consumes()) {
@@ -923,7 +1153,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K> ReceiveChannel<E>.groupBy(keySelector: (E) -> K): Map<K, List<E>> =
groupByTo(LinkedHashMap(), keySelector)
@@ -936,7 +1170,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, V> ReceiveChannel<E>.groupBy(keySelector: (E) -> K, valueTransform: (E) -> V): Map<K, List<V>> =
groupByTo(LinkedHashMap(), keySelector, valueTransform)
@@ -948,7 +1186,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, M : MutableMap<in K, MutableList<E>>> ReceiveChannel<E>.groupByTo(destination: M, keySelector: (E) -> K): M {
consumeEach {
val key = keySelector(it)
@@ -967,7 +1209,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, K, V, M : MutableMap<in K, MutableList<V>>> ReceiveChannel<E>.groupByTo(destination: M, keySelector: (E) -> K, valueTransform: (E) -> V): M {
consumeEach {
val key = keySelector(it)
@@ -1000,7 +1246,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R> ReceiveChannel<E>.mapIndexed(context: CoroutineContext = Dispatchers.Unconfined, transform: suspend (index: Int, E) -> R): ReceiveChannel<R> =
GlobalScope.produce(context, onCompletion = consumes()) {
@@ -1018,7 +1268,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R : Any> ReceiveChannel<E>.mapIndexedNotNull(context: CoroutineContext = Dispatchers.Unconfined, transform: suspend (index: Int, E) -> R?): ReceiveChannel<R> =
mapIndexed(context, transform).filterNotNull()
@@ -1031,7 +1285,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R : Any, C : MutableCollection<in R>> ReceiveChannel<E>.mapIndexedNotNullTo(destination: C, transform: (index: Int, E) -> R?): C {
consumeEachIndexed { (index, element) ->
transform(index, element)?.let { destination.add(it) }
@@ -1047,7 +1305,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R : Any, C : SendChannel<R>> ReceiveChannel<E>.mapIndexedNotNullTo(destination: C, transform: (index: Int, E) -> R?): C {
consumeEachIndexed { (index, element) ->
transform(index, element)?.let { destination.send(it) }
@@ -1063,7 +1325,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R, C : MutableCollection<in R>> ReceiveChannel<E>.mapIndexedTo(destination: C, transform: (index: Int, E) -> R): C {
var index = 0
consumeEach {
@@ -1080,7 +1346,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R, C : SendChannel<R>> ReceiveChannel<E>.mapIndexedTo(destination: C, transform: (index: Int, E) -> R): C {
var index = 0
consumeEach {
@@ -1095,7 +1365,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R : Any> ReceiveChannel<E>.mapNotNull(context: CoroutineContext = Dispatchers.Unconfined, transform: suspend (E) -> R?): ReceiveChannel<R> =
map(context, transform).filterNotNull()
@@ -1106,7 +1380,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R : Any, C : MutableCollection<in R>> ReceiveChannel<E>.mapNotNullTo(destination: C, transform: (E) -> R?): C {
consumeEach {
transform(it)?.let { destination.add(it) }
@@ -1120,7 +1398,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R : Any, C : SendChannel<R>> ReceiveChannel<E>.mapNotNullTo(destination: C, transform: (E) -> R?): C {
consumeEach {
transform(it)?.let { destination.send(it) }
@@ -1134,7 +1416,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R, C : MutableCollection<in R>> ReceiveChannel<E>.mapTo(destination: C, transform: (E) -> R): C {
consumeEach {
destination.add(transform(it))
@@ -1148,7 +1434,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R, C : SendChannel<R>> ReceiveChannel<E>.mapTo(destination: C, transform: (E) -> R): C {
consumeEach {
destination.send(transform(it))
@@ -1161,7 +1451,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public fun <E> ReceiveChannel<E>.withIndex(context: CoroutineContext = Dispatchers.Unconfined): ReceiveChannel<IndexedValue<E>> =
GlobalScope.produce(context, onCompletion = consumes()) {
var index = 0
@@ -1177,7 +1471,11 @@
*
* The operation is _intermediate_ and _stateful_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public fun <E> ReceiveChannel<E>.distinct(): ReceiveChannel<E> =
this.distinctBy { it }
@@ -1189,7 +1487,11 @@
*
* The operation is _intermediate_ and _stateful_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark predicate with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, K> ReceiveChannel<E>.distinctBy(context: CoroutineContext = Dispatchers.Unconfined, selector: suspend (E) -> K): ReceiveChannel<E> =
GlobalScope.produce(context, onCompletion = consumes()) {
@@ -1210,7 +1512,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.toMutableSet(): MutableSet<E> =
toCollection(LinkedHashSet())
@@ -1219,7 +1525,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.all(predicate: (E) -> Boolean): Boolean {
consumeEach {
if (!predicate(it)) return false
@@ -1232,7 +1542,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.any(): Boolean =
consume {
return iterator().hasNext()
@@ -1243,7 +1557,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.any(predicate: (E) -> Boolean): Boolean {
consumeEach {
if (predicate(it)) return true
@@ -1256,7 +1574,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.count(): Int {
var count = 0
consumeEach { count++ }
@@ -1268,7 +1590,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.count(predicate: (E) -> Boolean): Int {
var count = 0
consumeEach {
@@ -1282,7 +1608,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R> ReceiveChannel<E>.fold(initial: R, operation: (acc: R, E) -> R): R {
var accumulator = initial
consumeEach {
@@ -1299,7 +1629,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R> ReceiveChannel<E>.foldIndexed(initial: R, operation: (index: Int, acc: R, E) -> R): R {
var index = 0
var accumulator = initial
@@ -1314,7 +1648,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R : Comparable<R>> ReceiveChannel<E>.maxBy(selector: (E) -> R): E? =
consume {
val iterator = iterator()
@@ -1337,7 +1675,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.maxWith(comparator: Comparator<in E>): E? =
consume {
val iterator = iterator()
@@ -1355,7 +1697,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E, R : Comparable<R>> ReceiveChannel<E>.minBy(selector: (E) -> R): E? =
consume {
val iterator = iterator()
@@ -1378,7 +1724,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.minWith(comparator: Comparator<in E>): E? =
consume {
val iterator = iterator()
@@ -1396,7 +1746,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend fun <E> ReceiveChannel<E>.none(): Boolean =
consume {
return !iterator().hasNext()
@@ -1407,7 +1761,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.none(predicate: (E) -> Boolean): Boolean {
consumeEach {
if (predicate(it)) return false
@@ -1420,7 +1778,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <S, E : S> ReceiveChannel<E>.reduce(operation: (acc: S, E) -> S): S =
consume {
val iterator = this.iterator()
@@ -1440,7 +1802,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark operation with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public suspend inline fun <S, E : S> ReceiveChannel<E>.reduceIndexed(operation: (index: Int, acc: S, E) -> S): S =
consume {
@@ -1459,7 +1825,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.sumBy(selector: (E) -> Int): Int {
var sum = 0
consumeEach {
@@ -1473,7 +1843,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.sumByDouble(selector: (E) -> Double): Double {
var sum = 0.0
consumeEach {
@@ -1487,7 +1861,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public fun <E : Any> ReceiveChannel<E?>.requireNoNulls(): ReceiveChannel<E> =
map { it ?: throw IllegalArgumentException("null element found in $this.") }
@@ -1498,7 +1876,11 @@
*
* The operation is _terminal_.
* This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public suspend inline fun <E> ReceiveChannel<E>.partition(predicate: (E) -> Boolean): Pair<List<E>, List<E>> {
val first = ArrayList<E>()
val second = ArrayList<E>()
@@ -1518,7 +1900,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][ReceiveChannel.consume] all elements of both the original [ReceiveChannel] and the `other` one.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
public infix fun <E, R> ReceiveChannel<E>.zip(other: ReceiveChannel<R>): ReceiveChannel<Pair<E, R>> =
zip(other) { t1, t2 -> t1 to t2 }
@@ -1527,7 +1913,11 @@
*
* The operation is _intermediate_ and _stateless_.
* This function [consumes][consume] all elements of both the original [ReceiveChannel] and the `other` one.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
// todo: mark transform with crossinline modifier when it is supported: https://youtrack.jetbrains.com/issue/KT-19159
public fun <E, R, V> ReceiveChannel<E>.zip(other: ReceiveChannel<R>, context: CoroutineContext = Dispatchers.Unconfined, transform: (a: E, b: R) -> V): ReceiveChannel<V> =
GlobalScope.produce(context, onCompletion = consumesAll(this, other)) {
diff --git a/common/kotlinx-coroutines-core-common/src/channels/ConflatedBroadcastChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/ConflatedBroadcastChannel.kt
index 3bdd4c4..a91129d 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/ConflatedBroadcastChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/ConflatedBroadcastChannel.kt
@@ -5,6 +5,7 @@
package kotlinx.coroutines.experimental.channels
import kotlinx.atomicfu.*
+import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.internal.*
import kotlinx.coroutines.experimental.intrinsics.*
import kotlinx.coroutines.experimental.selects.*
@@ -23,7 +24,10 @@
* This implementation is fully lock-free. In this implementation
* [opening][openSubscription] and [closing][ReceiveChannel.cancel] subscription takes O(N) time, where N is the
* number of subscribers.
+ *
+ * **Note: This API is experimental.** It may be changed in the future updates.
*/
+@ExperimentalCoroutinesApi
public class ConflatedBroadcastChannel<E>() : BroadcastChannel<E> {
/**
* Creates an instance of this class that already holds a value.
diff --git a/common/kotlinx-coroutines-core-common/src/channels/ConflatedChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/ConflatedChannel.kt
index 5208378..16e01d7 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/ConflatedChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/ConflatedChannel.kt
@@ -4,6 +4,7 @@
package kotlinx.coroutines.experimental.channels
+import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.selects.*
/**
@@ -16,8 +17,16 @@
* This channel is created by `Channel(Channel.CONFLATED)` factory function invocation.
*
* This implementation is fully lock-free.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
-public open class ConflatedChannel<E> : AbstractChannel<E>() {
+@InternalCoroutinesApi // todo: review KDoc usage
+public open class ConflatedChannel<E>
+@Deprecated(
+ "Replace with Channel factory function",
+ replaceWith = ReplaceWith("Channel(Channel.CONFLATED)")
+)
+constructor(): AbstractChannel<E>() {
protected final override val isBufferAlwaysEmpty: Boolean get() = true
protected final override val isBufferEmpty: Boolean get() = true
protected final override val isBufferAlwaysFull: Boolean get() = false
diff --git a/common/kotlinx-coroutines-core-common/src/channels/LinkedListChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/LinkedListChannel.kt
index 885ba7a..4455d1a 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/LinkedListChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/LinkedListChannel.kt
@@ -4,6 +4,7 @@
package kotlinx.coroutines.experimental.channels
+import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.selects.*
/**
@@ -13,8 +14,16 @@
* This channel is created by `Channel(Channel.UNLIMITED)` factory function invocation.
*
* This implementation is fully lock-free.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
-public open class LinkedListChannel<E> : AbstractChannel<E>() {
+@InternalCoroutinesApi // todo: review KDoc usage
+public open class LinkedListChannel<E>
+@Deprecated(
+ "Replace with Channel factory function",
+ replaceWith = ReplaceWith("Channel(Channel.UNLIMITED)")
+)
+constructor() : AbstractChannel<E>() {
protected final override val isBufferAlwaysEmpty: Boolean get() = true
protected final override val isBufferEmpty: Boolean get() = true
protected final override val isBufferAlwaysFull: Boolean get() = false
diff --git a/common/kotlinx-coroutines-core-common/src/channels/Produce.kt b/common/kotlinx-coroutines-core-common/src/channels/Produce.kt
index 9f234aa..63726cf 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/Produce.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/Produce.kt
@@ -10,7 +10,11 @@
/**
* Scope for [produce][CoroutineScope.produce] coroutine builder.
+ *
+ * **Note: This is an experimental api.** Behaviour of producers that work as children in a parent scope with respect
+ * to cancellation and error handling may change in the future.
*/
+@ExperimentalCoroutinesApi
public interface ProducerScope<in E> : CoroutineScope, SendChannel<E> {
/**
* A reference to the channel that this coroutine [sends][send] elements to.
@@ -60,11 +64,31 @@
*
* See [newCoroutineContext] for a description of debugging facilities that are available for newly created coroutine.
*
+ * **Note: This is an experimental api.** Behaviour of producers that work as children in a parent scope with respect
+ * to cancellation and error handling may change in the future.
+ *
* @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
* @param capacity capacity of the channel's buffer (no buffer by default).
- * @param onCompletion optional completion handler for the producer coroutine (see [Job.invokeOnCompletion]).
* @param block the coroutine code.
*/
+@ExperimentalCoroutinesApi
+public fun <E> CoroutineScope.produce(
+ context: CoroutineContext = EmptyCoroutineContext,
+ capacity: Int = 0,
+ block: suspend ProducerScope<E>.() -> Unit
+): ReceiveChannel<E> {
+ val channel = Channel<E>(capacity)
+ val newContext = newCoroutineContext(context)
+ val coroutine = ProducerCoroutine(newContext, channel)
+ coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
+ return coroutine
+}
+
+/**
+ * @suppress **This an internal API and should not be used from general code.**
+ * onCompletion parameter will be redesigned.
+ */
+@InternalCoroutinesApi
public fun <E> CoroutineScope.produce(
context: CoroutineContext = EmptyCoroutineContext,
capacity: Int = 0,
diff --git a/common/kotlinx-coroutines-core-common/src/channels/RendezvousChannel.kt b/common/kotlinx-coroutines-core-common/src/channels/RendezvousChannel.kt
index cac7d5b..a47f34f 100644
--- a/common/kotlinx-coroutines-core-common/src/channels/RendezvousChannel.kt
+++ b/common/kotlinx-coroutines-core-common/src/channels/RendezvousChannel.kt
@@ -4,6 +4,8 @@
package kotlinx.coroutines.experimental.channels
+import kotlinx.coroutines.experimental.*
+
/**
* Rendezvous channel. This channel does not have any buffer at all. An element is transferred from sender
* to receiver only when [send] and [receive] invocations meet in time (rendezvous), so [send] suspends
@@ -12,8 +14,16 @@
* Use `Channel()` factory function to conveniently create an instance of rendezvous channel.
*
* This implementation is fully lock-free.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
-public open class RendezvousChannel<E> : AbstractChannel<E>() {
+@InternalCoroutinesApi // todo: review KDoc usage
+public open class RendezvousChannel<E>
+@Deprecated(
+ "Replace with Channel factory function",
+ replaceWith = ReplaceWith("Channel()")
+)
+constructor() : AbstractChannel<E>() {
protected final override val isBufferAlwaysEmpty: Boolean get() = true
protected final override val isBufferEmpty: Boolean get() = true
protected final override val isBufferAlwaysFull: Boolean get() = true
diff --git a/common/kotlinx-coroutines-core-common/src/intrinsics/Cancellable.kt b/common/kotlinx-coroutines-core-common/src/intrinsics/Cancellable.kt
index 4564c2c..5d263f1 100644
--- a/common/kotlinx-coroutines-core-common/src/intrinsics/Cancellable.kt
+++ b/common/kotlinx-coroutines-core-common/src/intrinsics/Cancellable.kt
@@ -11,13 +11,19 @@
/**
* Use this function to start coroutine in a cancellable way, so that it can be cancelled
* while waiting to be dispatched.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>) =
createCoroutineUnchecked(completion).resumeCancellable(Unit)
/**
* Use this function to start coroutine in a cancellable way, so that it can be cancelled
* while waiting to be dispatched.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <R, T> (suspend (R) -> T).startCoroutineCancellable(receiver: R, completion: Continuation<T>) =
createCoroutineUnchecked(receiver, completion).resumeCancellable(Unit)
diff --git a/common/kotlinx-coroutines-core-common/src/intrinsics/Undispatched.kt b/common/kotlinx-coroutines-core-common/src/intrinsics/Undispatched.kt
index 80048a8..8daf5da 100644
--- a/common/kotlinx-coroutines-core-common/src/intrinsics/Undispatched.kt
+++ b/common/kotlinx-coroutines-core-common/src/intrinsics/Undispatched.kt
@@ -12,7 +12,10 @@
* Use this function to restart coroutine directly from inside of [suspendCoroutine],
* when the code is already in the context of this coroutine.
* It does not use [ContinuationInterceptor] and does not update context of the current thread.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <T> (suspend () -> T).startCoroutineUnintercepted(completion: Continuation<T>) {
startDirect(completion) {
startCoroutineUninterceptedOrReturn(completion)
@@ -23,7 +26,10 @@
* Use this function to restart coroutine directly from inside of [suspendCoroutine],
* when the code is already in the context of this coroutine.
* It does not use [ContinuationInterceptor] and does not update context of the current thread.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <R, T> (suspend (R) -> T).startCoroutineUnintercepted(receiver: R, completion: Continuation<T>) {
startDirect(completion) {
startCoroutineUninterceptedOrReturn(receiver, completion)
@@ -34,7 +40,10 @@
* Use this function to start new coroutine in [CoroutineStart.UNDISPATCHED] mode —
* immediately execute coroutine in the current thread until next suspension.
* It does not use [ContinuationInterceptor], but updates the context of the current thread for the new coroutine.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <T> (suspend () -> T).startCoroutineUndispatched(completion: Continuation<T>) {
startDirect(completion) {
withCoroutineContext(completion.context) {
@@ -47,7 +56,10 @@
* Use this function to start new coroutine in [CoroutineStart.UNDISPATCHED] mode —
* immediately execute coroutine in the current thread until next suspension.
* It does not use [ContinuationInterceptor], but updates the context of the current thread for the new coroutine.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <R, T> (suspend (R) -> T).startCoroutineUndispatched(receiver: R, completion: Continuation<T>) {
startDirect(completion) {
withCoroutineContext(completion.context) {
@@ -76,7 +88,10 @@
*
* First, this function initializes parent job from the `parentContext` of this coroutine that was passed to it
* during construction. Second, it starts the coroutine using [startCoroutineUninterceptedOrReturn].
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <T> AbstractCoroutine<T>.startUndispatchedOrReturn(block: suspend () -> T): Any? {
initParentJob()
return undispatchedResult { block.startCoroutineUninterceptedOrReturn(this) }
@@ -89,7 +104,10 @@
*
* First, this function initializes parent job from the `parentContext` of this coroutine that was passed to it
* during construction. Second, it starts the coroutine using [startCoroutineUninterceptedOrReturn].
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun <T, R> AbstractCoroutine<T>.startUndispatchedOrReturn(receiver: R, block: suspend R.() -> T): Any? {
initParentJob()
return undispatchedResult { block.startCoroutineUninterceptedOrReturn(receiver, this) }
diff --git a/common/kotlinx-coroutines-core-common/src/selects/Select.kt b/common/kotlinx-coroutines-core-common/src/selects/Select.kt
index 23b6752..ee9fe73 100644
--- a/common/kotlinx-coroutines-core-common/src/selects/Select.kt
+++ b/common/kotlinx-coroutines-core-common/src/selects/Select.kt
@@ -42,10 +42,22 @@
* Clause that selects the given [block] after a specified timeout passes.
* If timeout is negative or zero, [block] is selected immediately.
*
- * @param time timeout time
- * @param unit timeout unit (milliseconds by default)
+ * **Note: This is an experimental api.** It may be replaced with light-weight timer/timeout channels in the future.
+ *
+ * @param timeMillis timeout time in milliseconds.
*/
- public fun onTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend () -> R)
+ @ExperimentalCoroutinesApi
+ public fun onTimeout(timeMillis: Long, block: suspend () -> R)
+
+ /**
+ * @suppress **Deprecated**: onTimeout(unit.toMillis(time), block)`
+ */
+ @Deprecated(
+ message = "Replace with onTimeout(unit.toMillis(time), block)",
+ replaceWith = ReplaceWith("onTimeout(unit.toMillis(time), block)")
+ )
+ public fun onTimeout(time: Long, unit: TimeUnit = TimeUnit.MILLISECONDS, block: suspend () -> R) =
+ onTimeout(time.convertToMillis(unit), block)
}
/**
@@ -56,6 +68,7 @@
* Registers this clause with the specified [select] instance and [block] of code.
* @suppress **This is unstable API and it is subject to change.**
*/
+ @InternalCoroutinesApi
public fun <R> registerSelectClause0(select: SelectInstance<R>, block: suspend () -> R)
}
@@ -67,6 +80,7 @@
* Registers this clause with the specified [select] instance and [block] of code.
* @suppress **This is unstable API and it is subject to change.**
*/
+ @InternalCoroutinesApi
public fun <R> registerSelectClause1(select: SelectInstance<R>, block: suspend (Q) -> R)
}
@@ -78,6 +92,7 @@
* Registers this clause with the specified [select] instance and [block] of code.
* @suppress **This is unstable API and it is subject to change.**
*/
+ @InternalCoroutinesApi
public fun <R> registerSelectClause2(select: SelectInstance<R>, param: P, block: suspend (Q) -> R)
}
@@ -87,6 +102,7 @@
*
* @suppress **This is unstable API and it is subject to change.**
*/
+@InternalCoroutinesApi
public interface SelectInstance<in R> {
/**
* Returns `true` when this [select] statement had already picked a clause to execute.
@@ -404,8 +420,8 @@
registerSelectClause2(this@SelectBuilderImpl, param, block)
}
- override fun onTimeout(time: Long, unit: TimeUnit, block: suspend () -> R) {
- if (time <= 0L) {
+ override fun onTimeout(timeMillis: Long, block: suspend () -> R) {
+ if (timeMillis <= 0L) {
if (trySelect(null))
block.startCoroutineUnintercepted(completion)
return
@@ -416,7 +432,7 @@
if (trySelect(null))
block.startCoroutineCancellable(completion) // shall be cancellable while waits for dispatch
}
- disposeOnSelect(context.delay.invokeOnTimeout(time, unit, action))
+ disposeOnSelect(context.delay.invokeOnTimeout(timeMillis, action))
}
private class DisposeNode(
diff --git a/common/kotlinx-coroutines-core-common/src/selects/SelectUnbiased.kt b/common/kotlinx-coroutines-core-common/src/selects/SelectUnbiased.kt
index 05e1bc2..2c84d47 100644
--- a/common/kotlinx-coroutines-core-common/src/selects/SelectUnbiased.kt
+++ b/common/kotlinx-coroutines-core-common/src/selects/SelectUnbiased.kt
@@ -4,7 +4,6 @@
package kotlinx.coroutines.experimental.selects
-import kotlinx.coroutines.experimental.timeunit.*
import kotlin.coroutines.experimental.*
import kotlin.coroutines.experimental.intrinsics.*
@@ -64,7 +63,7 @@
clauses += { registerSelectClause2(instance, param, block) }
}
- override fun onTimeout(time: Long, unit: TimeUnit, block: suspend () -> R) {
- clauses += { instance.onTimeout(time, unit, block) }
+ override fun onTimeout(timeMillis: Long, block: suspend () -> R) {
+ clauses += { instance.onTimeout(timeMillis, block) }
}
}
\ No newline at end of file
diff --git a/common/kotlinx-coroutines-core-common/src/selects/WhileSelect.kt b/common/kotlinx-coroutines-core-common/src/selects/WhileSelect.kt
index 8358c19..f9260b9 100644
--- a/common/kotlinx-coroutines-core-common/src/selects/WhileSelect.kt
+++ b/common/kotlinx-coroutines-core-common/src/selects/WhileSelect.kt
@@ -4,6 +4,8 @@
package kotlinx.coroutines.experimental.selects
+import kotlinx.coroutines.experimental.*
+
/**
* Loops while [select] expression returns `true`.
*
@@ -21,7 +23,10 @@
* while(select<Boolean> {
* /*body*/
* }) {}
+ *
+ * **Note: This is an experimental api.** It may be replaced with a higher-performance DSL for selection from loops.
*/
-suspend fun whileSelect(builder: SelectBuilder<Boolean>.() -> Unit) {
+@ExperimentalCoroutinesApi
+public suspend inline fun whileSelect(crossinline builder: SelectBuilder<Boolean>.() -> Unit) {
while(select<Boolean>(builder)) {}
}
diff --git a/common/kotlinx-coroutines-core-common/src/sync/Mutex.kt b/common/kotlinx-coroutines-core-common/src/sync/Mutex.kt
index 997be13..896e682 100644
--- a/common/kotlinx-coroutines-core-common/src/sync/Mutex.kt
+++ b/common/kotlinx-coroutines-core-common/src/sync/Mutex.kt
@@ -76,8 +76,9 @@
*/
public fun holdsLock(owner: Any): Boolean
- /**
- * Unlocks this mutex. Throws [IllegalStateException] if invoked on a mutex that is not locked.
+ /**
+ * Unlocks this mutex. Throws [IllegalStateException] if invoked on a mutex that is not locked or
+ * was locked with a different owner token (by identity).
*
* @param owner Optional owner token for debugging. When `owner` is specified (non-null value) and this mutex
* was locked with the different token (by identity), this function throws [IllegalStateException].
diff --git a/common/kotlinx-coroutines-core-common/src/timeunit/TimeUnit.common.kt b/common/kotlinx-coroutines-core-common/src/timeunit/TimeUnit.common.kt
index 3c868df..f10609e 100644
--- a/common/kotlinx-coroutines-core-common/src/timeunit/TimeUnit.common.kt
+++ b/common/kotlinx-coroutines-core-common/src/timeunit/TimeUnit.common.kt
@@ -4,6 +4,11 @@
package kotlinx.coroutines.experimental.timeunit
+/*
+ * @suppress **Deprecated** No replacement
+ */
+@Suppress("ACTUAL_WITHOUT_EXPECT")
+@Deprecated(message = "No replacement")
public expect enum class TimeUnit {
MILLISECONDS,
SECONDS;
diff --git a/common/kotlinx-coroutines-core-common/test/WithTimeoutTest.kt b/common/kotlinx-coroutines-core-common/test/WithTimeoutTest.kt
index 40f0e71..2408f77 100644
--- a/common/kotlinx-coroutines-core-common/test/WithTimeoutTest.kt
+++ b/common/kotlinx-coroutines-core-common/test/WithTimeoutTest.kt
@@ -124,7 +124,7 @@
"OK"
}
} catch (e: CancellationException) {
- assertEquals("Timed out waiting for 100 MILLISECONDS", e.message)
+ assertEquals("Timed out waiting for 100 ms", e.message)
finish(3)
}
}
diff --git a/common/kotlinx-coroutines-core-common/test/channels/ProduceTest.kt b/common/kotlinx-coroutines-core-common/test/channels/ProduceTest.kt
index 0f958af..5c455e4 100644
--- a/common/kotlinx-coroutines-core-common/test/channels/ProduceTest.kt
+++ b/common/kotlinx-coroutines-core-common/test/channels/ProduceTest.kt
@@ -106,7 +106,7 @@
try {
source.receive()
// TODO shouldn't it be ClosedReceiveChannelException ?
- } catch (e: JobCancellationException) {
+ } catch (e: CancellationException) {
finish(4)
}
}
diff --git a/common/kotlinx-coroutines-core-common/test/channels/TestChannelKind.kt b/common/kotlinx-coroutines-core-common/test/channels/TestChannelKind.kt
index 7af18b8..7798b6b 100644
--- a/common/kotlinx-coroutines-core-common/test/channels/TestChannelKind.kt
+++ b/common/kotlinx-coroutines-core-common/test/channels/TestChannelKind.kt
@@ -59,6 +59,7 @@
override suspend fun receiveOrNull(): E? = sub.receiveOrNull()
override fun poll(): E? = sub.poll()
override fun iterator(): ChannelIterator<E> = sub.iterator()
+ override fun cancel(): Boolean = sub.cancel()
override fun cancel(cause: Throwable?): Boolean = sub.cancel(cause)
override val onReceive: SelectClause1<E>
get() = sub.onReceive
diff --git a/core/kotlinx-coroutines-core/README.md b/core/kotlinx-coroutines-core/README.md
index f2b9977..5093fd1 100644
--- a/core/kotlinx-coroutines-core/README.md
+++ b/core/kotlinx-coroutines-core/README.md
@@ -114,7 +114,7 @@
[newSingleThreadContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/new-single-thread-context.html
[newFixedThreadPoolContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/new-fixed-thread-pool-context.html
[java.util.concurrent.Executor.asCoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/java.util.concurrent.-executor/as-coroutine-dispatcher.html
-[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable/index.html
+[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable.html
[CoroutineExceptionHandler]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-exception-handler/index.html
[delay]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/delay.html
[yield]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/yield.html
diff --git a/core/kotlinx-coroutines-core/src/Builders.kt b/core/kotlinx-coroutines-core/src/Builders.kt
index 43521c2..03ae637 100644
--- a/core/kotlinx-coroutines-core/src/Builders.kt
+++ b/core/kotlinx-coroutines-core/src/Builders.kt
@@ -15,14 +15,13 @@
* This function should not be used from coroutine. It is designed to bridge regular blocking code
* to libraries that are written in suspending style, to be used in `main` functions and in tests.
*
- * The default [CoroutineDispatcher] for this builder in an implementation of [EventLoop] that processes continuations
+ * The default [CoroutineDispatcher] for this builder in an internal implementation of event loop that processes continuations
* in this blocked thread until the completion of this coroutine.
* See [CoroutineDispatcher] for the other implementations that are provided by `kotlinx.coroutines`.
*
* When [CoroutineDispatcher] is explicitly specified in the [context], then the new coroutine runs in the context of
- * the specified dispatcher while the current thread is blocked. If the specified dispatcher implements [EventLoop]
- * interface and this `runBlocking` invocation is performed from inside of the this event loop's thread, then
- * this event loop is processed using its [processNextEvent][EventLoop.processNextEvent] method until coroutine completes.
+ * the specified dispatcher while the current thread is blocked. If the specified dispatcher is an event loop of another `runBlocking`,
+ * the this invocation uses an outer event loop.
*
* If this blocked thread is interrupted (see [Thread.interrupt]), then the coroutine job is cancelled and
* this `runBlocking` invocation throws [InterruptedException].
diff --git a/core/kotlinx-coroutines-core/src/CoroutineContext.kt b/core/kotlinx-coroutines-core/src/CoroutineContext.kt
index 402b504..c2af973 100644
--- a/core/kotlinx-coroutines-core/src/CoroutineContext.kt
+++ b/core/kotlinx-coroutines-core/src/CoroutineContext.kt
@@ -9,35 +9,6 @@
import java.util.concurrent.atomic.*
import kotlin.coroutines.experimental.*
-/**
- * Name of the property that controls coroutine debugging. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
- */
-public const val DEBUG_PROPERTY_NAME = "kotlinx.coroutines.debug"
-
-/**
- * Automatic debug configuration value for [DEBUG_PROPERTY_NAME]. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
- */
-public const val DEBUG_PROPERTY_VALUE_AUTO = "auto"
-
-/**
- * Debug turned on value for [DEBUG_PROPERTY_NAME]. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
- */
-public const val DEBUG_PROPERTY_VALUE_ON = "on"
-
-/**
- * Debug turned on value for [DEBUG_PROPERTY_NAME]. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
- */
-public const val DEBUG_PROPERTY_VALUE_OFF = "off"
-
-internal val DEBUG = systemProp(DEBUG_PROPERTY_NAME).let { value ->
- when (value) {
- DEBUG_PROPERTY_VALUE_AUTO, null -> CoroutineId::class.java.desiredAssertionStatus()
- DEBUG_PROPERTY_VALUE_ON, "" -> true
- DEBUG_PROPERTY_VALUE_OFF -> false
- else -> error("System property '$DEBUG_PROPERTY_NAME' has unrecognized value '$value'")
- }
-}
-
private val COROUTINE_ID = AtomicLong()
// for tests only
@@ -101,7 +72,11 @@
*
* Coroutine name can be explicitly assigned using [CoroutineName] context element.
* The string "coroutine" is used as a default name.
+ *
+ * **Note: This is an experimental api.**
+ * Behavior of this function may change in the future with respect to its support for debugging facilities.
*/
+@ExperimentalCoroutinesApi
public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
val combined = coroutineContext + context
val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
diff --git a/core/kotlinx-coroutines-core/src/Debug.kt b/core/kotlinx-coroutines-core/src/Debug.kt
index d2e0da0..6a31687 100644
--- a/core/kotlinx-coroutines-core/src/Debug.kt
+++ b/core/kotlinx-coroutines-core/src/Debug.kt
@@ -4,8 +4,38 @@
package kotlinx.coroutines.experimental
+import kotlinx.coroutines.experimental.internal.*
import kotlin.coroutines.experimental.Continuation
+/**
+ * Name of the property that controls coroutine debugging. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
+ */
+public const val DEBUG_PROPERTY_NAME = "kotlinx.coroutines.debug"
+
+/**
+ * Automatic debug configuration value for [DEBUG_PROPERTY_NAME]. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
+ */
+public const val DEBUG_PROPERTY_VALUE_AUTO = "auto"
+
+/**
+ * Debug turned on value for [DEBUG_PROPERTY_NAME]. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
+ */
+public const val DEBUG_PROPERTY_VALUE_ON = "on"
+
+/**
+ * Debug turned on value for [DEBUG_PROPERTY_NAME]. See [newCoroutineContext][CoroutineScope.newCoroutineContext].
+ */
+public const val DEBUG_PROPERTY_VALUE_OFF = "off"
+
+internal val DEBUG = systemProp(DEBUG_PROPERTY_NAME).let { value ->
+ when (value) {
+ DEBUG_PROPERTY_VALUE_AUTO, null -> CoroutineId::class.java.desiredAssertionStatus()
+ DEBUG_PROPERTY_VALUE_ON, "" -> true
+ DEBUG_PROPERTY_VALUE_OFF -> false
+ else -> error("System property '$DEBUG_PROPERTY_NAME' has unrecognized value '$value'")
+ }
+}
+
// internal debugging tools
internal actual val Any.hexAddress: String
diff --git a/core/kotlinx-coroutines-core/src/DefaultExecutor.kt b/core/kotlinx-coroutines-core/src/DefaultExecutor.kt
index 322914b..cdec915 100644
--- a/core/kotlinx-coroutines-core/src/DefaultExecutor.kt
+++ b/core/kotlinx-coroutines-core/src/DefaultExecutor.kt
@@ -4,7 +4,7 @@
package kotlinx.coroutines.experimental
-import kotlinx.coroutines.experimental.timeunit.*
+import java.util.concurrent.*
internal actual val DefaultDelay: Delay = DefaultExecutor
@@ -22,6 +22,7 @@
DEFAULT_KEEP_ALIVE
})
+ @Suppress("ObjectPropertyName")
@Volatile
private var _thread: Thread? = null
@@ -47,8 +48,8 @@
* Livelock is possible only if `runBlocking` is called on internal default executed (which is used by default [delay]),
* but it's not exposed as public API.
*/
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
- DelayedRunnableTask(time, unit, block).also { schedule(it) }
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle =
+ DelayedRunnableTask(timeMillis, block).also { schedule(it) }
override fun run() {
timeSource.registerTimeLoopThread()
diff --git a/core/kotlinx-coroutines-core/src/EventLoop.kt b/core/kotlinx-coroutines-core/src/EventLoop.kt
index 8aa533d..4e21abd 100644
--- a/core/kotlinx-coroutines-core/src/EventLoop.kt
+++ b/core/kotlinx-coroutines-core/src/EventLoop.kt
@@ -6,7 +6,6 @@
import kotlinx.atomicfu.*
import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.timeunit.*
import java.util.concurrent.locks.*
import kotlin.coroutines.experimental.*
@@ -16,7 +15,10 @@
*
* It may optionally implement [Delay] interface and support time-scheduled tasks. It is used by [runBlocking] to
* continue processing events when invoked from the event dispatch thread.
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi // todo: review KDoc references to this interface
public interface EventLoop: ContinuationInterceptor {
/**
* Processes next event in this event loop.
@@ -52,8 +54,11 @@
* LockSupport.parkNanos(eventLoop.processNextEvent()) // event loop will unpark
* }
* ```
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
@Suppress("FunctionName")
+@InternalCoroutinesApi
public fun EventLoop(thread: Thread = Thread.currentThread(), parentJob: Job? = null): EventLoop =
EventLoopImpl(thread).apply {
if (parentJob != null) initParentJob(parentJob)
@@ -71,6 +76,18 @@
private const val REMOVED = 1
private const val RESCHEDULED = 2
+private const val MS_TO_NS = 1_000_000L
+private const val MAX_MS = Long.MAX_VALUE / MS_TO_NS
+
+internal fun delayToNanos(timeMillis: Long): Long = when {
+ timeMillis <= 0 -> 0L
+ timeMillis >= MAX_MS -> Long.MAX_VALUE
+ else -> timeMillis * MS_TO_NS
+}
+
+internal fun delayNanosToMillis(timeNanos: Long): Long =
+ timeNanos / MS_TO_NS
+
@Suppress("PrivatePropertyName")
private val CLOSED_EMPTY = Symbol("CLOSED_EMPTY")
@@ -121,8 +138,8 @@
override fun dispatch(context: CoroutineContext, block: Runnable) =
execute(block)
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) =
- schedule(DelayedResumeTask(time, unit, continuation))
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) =
+ schedule(DelayedResumeTask(timeMillis, continuation))
override fun processNextEvent(): Long {
if (!isCorrectThread()) return Long.MAX_VALUE
@@ -260,11 +277,11 @@
}
internal abstract inner class DelayedTask(
- time: Long, timeUnit: TimeUnit
+ timeMillis: Long
) : Runnable, Comparable<DelayedTask>, DisposableHandle, ThreadSafeHeapNode {
override var index: Int = -1
- var state = DELAYED // Guarded by by lock on this task for reschedule/dispose purposes
- @JvmField val nanoTime: Long = timeSource.nanoTime() + timeUnit.toNanos(time)
+ @JvmField var state = DELAYED // Guarded by by lock on this task for reschedule/dispose purposes
+ @JvmField val nanoTime: Long = timeSource.nanoTime() + delayToNanos(timeMillis)
override fun compareTo(other: DelayedTask): Int {
val dTime = nanoTime - other.nanoTime
@@ -302,10 +319,9 @@
}
private inner class DelayedResumeTask(
- time: Long, timeUnit: TimeUnit,
+ timeMillis: Long,
private val cont: CancellableContinuation<Unit>
- ) : DelayedTask(time, timeUnit) {
-
+ ) : DelayedTask(timeMillis) {
init {
// Note that this operation isn't lock-free, but very short
cont.disposeOnCancellation(this)
@@ -318,9 +334,9 @@
// Cannot be moved to DefaultExecutor due to BE bug
internal inner class DelayedRunnableTask(
- time: Long, timeUnit: TimeUnit,
+ time: Long,
private val block: Runnable
- ) : DelayedTask(time, timeUnit) {
+ ) : DelayedTask(time) {
override fun run() { block.run() }
override fun toString(): String = super.toString() + block.toString()
}
diff --git a/core/kotlinx-coroutines-core/src/Exceptions.kt b/core/kotlinx-coroutines-core/src/Exceptions.kt
index 9a9ac06..43889c0 100644
--- a/core/kotlinx-coroutines-core/src/Exceptions.kt
+++ b/core/kotlinx-coroutines-core/src/Exceptions.kt
@@ -6,7 +6,10 @@
/**
* This exception gets thrown if an exception is caught while processing [CompletionHandler] invocation for [Job].
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public actual class CompletionHandlerException actual constructor(
message: String,
cause: Throwable
@@ -24,7 +27,11 @@
* Thrown by cancellable suspending functions if the [Job] of the coroutine is cancelled or completed
* without cause, or with a cause or exception that is not [CancellationException]
* (see [Job.getCancellationException]).
+ *
+ * @suppress **Deprecated**: Replace with [CancellationException].
*/
+@InternalCoroutinesApi // todo: review usage from docs and examples
+@Deprecated(message = "Replace with CancellationException", replaceWith = ReplaceWith("CancellationException"))
public actual class JobCancellationException public actual constructor(
message: String,
cause: Throwable?,
diff --git a/core/kotlinx-coroutines-core/src/Executors.kt b/core/kotlinx-coroutines-core/src/Executors.kt
index d393564..178ee39 100644
--- a/core/kotlinx-coroutines-core/src/Executors.kt
+++ b/core/kotlinx-coroutines-core/src/Executors.kt
@@ -4,7 +4,6 @@
package kotlinx.coroutines.experimental
-import kotlinx.coroutines.experimental.timeunit.TimeUnit
import java.io.*
import java.util.concurrent.*
import kotlin.coroutines.experimental.*
@@ -17,6 +16,12 @@
* asynchronous API which requires instance of the [Executor].
*/
public abstract class ExecutorCoroutineDispatcher: CloseableCoroutineDispatcher(), Closeable {
+ /**
+ * Closes this coroutine dispatcher and shuts down its executor.
+ *
+ * It may throw an exception if this dispatcher is global and cannot be closed.
+ */
+ public abstract override fun close()
/**
* Underlying executor of current [CoroutineDispatcher].
@@ -42,6 +47,7 @@
/**
* Converts an instance of [Executor] to an implementation of [CoroutineDispatcher].
*/
+@JvmName("from") // this is for a nice Java API, see issue #255
public fun Executor.asCoroutineDispatcher(): CoroutineDispatcher =
ExecutorCoroutineDispatcherImpl(this)
@@ -68,6 +74,7 @@
/**
* @suppress **This is unstable API and it is subject to change.**
*/
+@InternalCoroutinesApi
public abstract class ExecutorCoroutineDispatcherBase : ExecutorCoroutineDispatcher(), Delay {
override fun dispatch(context: CoroutineContext, block: Runnable) =
@@ -77,26 +84,26 @@
DefaultExecutor.execute(block)
}
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
val timeout =
try { (executor as? ScheduledExecutorService)
- ?.schedule(ResumeUndispatchedRunnable(this, continuation), time, unit) }
+ ?.schedule(ResumeUndispatchedRunnable(this, continuation), timeMillis, TimeUnit.MILLISECONDS) }
catch (e: RejectedExecutionException) { null }
if (timeout != null)
continuation.cancelFutureOnCancellation(timeout)
else
- DefaultExecutor.scheduleResumeAfterDelay(time, unit, continuation)
+ DefaultExecutor.scheduleResumeAfterDelay(timeMillis, continuation)
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
val timeout =
try { (executor as? ScheduledExecutorService)
- ?.schedule(block, time, unit) }
+ ?.schedule(block, timeMillis, TimeUnit.MILLISECONDS) }
catch (e: RejectedExecutionException) { null }
return if (timeout != null)
DisposableFutureHandle(timeout)
else
- DefaultExecutor.invokeOnTimeout(time, unit, block)
+ DefaultExecutor.invokeOnTimeout(timeMillis, block)
}
override fun close() {
diff --git a/core/kotlinx-coroutines-core/src/Future.kt b/core/kotlinx-coroutines-core/src/Future.kt
index b198404..398885f 100644
--- a/core/kotlinx-coroutines-core/src/Future.kt
+++ b/core/kotlinx-coroutines-core/src/Future.kt
@@ -15,7 +15,10 @@
* ```
* invokeOnCompletion { future.cancel(false) }
* ```
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
public fun Job.cancelFutureOnCompletion(future: Future<*>): DisposableHandle =
invokeOnCompletion(handler = CancelFutureOnCompletion(this, future)) // TODO make it work only on cancellation as well?
diff --git a/core/kotlinx-coroutines-core/src/ThreadPoolDispatcher.kt b/core/kotlinx-coroutines-core/src/ThreadPoolDispatcher.kt
index b13a731..08e2d1e 100644
--- a/core/kotlinx-coroutines-core/src/ThreadPoolDispatcher.kt
+++ b/core/kotlinx-coroutines-core/src/ThreadPoolDispatcher.kt
@@ -9,16 +9,24 @@
import kotlin.coroutines.experimental.*
/**
- * Creates a new coroutine execution context using a single thread with built-in [yield] and [delay] support.
- * **NOTE: The resulting [ThreadPoolDispatcher] owns native resources (its thread).
- * Resources are reclaimed by [ThreadPoolDispatcher.close].**
+ * Creates a new coroutine execution context using a single thread with built-in [yield] support.
+ * **NOTE: The resulting [ExecutorCoroutineDispatcher] owns native resources (its thread).
+ * Resources are reclaimed by [ExecutorCoroutineDispatcher.close].**
*
* @param name the base name of the created thread.
*/
-fun newSingleThreadContext(name: String): ThreadPoolDispatcher =
+fun newSingleThreadContext(name: String): ExecutorCoroutineDispatcher =
newFixedThreadPoolContext(1, name)
/**
+ * @suppress binary compatibility
+ */
+@JvmName("newSingleThreadContext")
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "Binary compatibility only")
+fun newSingleThreadContext0(name: String): ThreadPoolDispatcher =
+ newSingleThreadContext(name) as ThreadPoolDispatcher
+
+/**
* @suppress **Deprecated**: Parent job is no longer supported.
*/
@Deprecated(message = "Parent job is no longer supported, `close` the resulting ThreadPoolDispatcher to release resources",
@@ -27,19 +35,27 @@
newFixedThreadPoolContext(1, name)
/**
- * Creates new coroutine execution context with the fixed-size thread-pool and built-in [yield] and [delay] support.
- * **NOTE: The resulting [ThreadPoolDispatcher] owns native resources (its threads).
- * Resources are reclaimed by [ThreadPoolDispatcher.close].**
+ * Creates new coroutine execution context with the fixed-size thread-pool and built-in [yield] support.
+ * **NOTE: The resulting [ExecutorCoroutineDispatcher] owns native resources (its threads).
+ * Resources are reclaimed by [ExecutorCoroutineDispatcher.close].**
*
* @param nThreads the number of threads.
* @param name the base name of the created threads.
*/
-fun newFixedThreadPoolContext(nThreads: Int, name: String): ThreadPoolDispatcher {
+fun newFixedThreadPoolContext(nThreads: Int, name: String): ExecutorCoroutineDispatcher {
require(nThreads >= 1) { "Expected at least one thread, but $nThreads specified" }
return ThreadPoolDispatcher(nThreads, name)
}
/**
+ * @suppress binary compatibility
+ */
+@JvmName("newFixedThreadPoolContext")
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "Binary compatibility only")
+fun newFixedThreadPoolContext0(nThreads: Int, name: String): ThreadPoolDispatcher =
+ newFixedThreadPoolContext(nThreads, name) as ThreadPoolDispatcher
+
+/**
* @suppress **Deprecated**: Parent job is no longer supported.
*/
@Deprecated(message = "Parent job is no longer supported, `close` the resulting ThreadPoolDispatcher to release resources",
@@ -57,7 +73,11 @@
/**
* Dispatches coroutine execution to a thread pool of a fixed size. Instances of this dispatcher are
* created with [newSingleThreadContext] and [newFixedThreadPoolContext].
+ *
+ * @suppress **This an internal API and should not be used from general code.**
*/
+@InternalCoroutinesApi
+@Deprecated("Replace with ExecutorCoroutineDispatcher", replaceWith = ReplaceWith("ExecutorCoroutineDispatcher"))
public class ThreadPoolDispatcher internal constructor(
private val nThreads: Int,
private val name: String
diff --git a/core/kotlinx-coroutines-core/src/channels/Actor.kt b/core/kotlinx-coroutines-core/src/channels/Actor.kt
index dd933e3..824799a 100644
--- a/core/kotlinx-coroutines-core/src/channels/Actor.kt
+++ b/core/kotlinx-coroutines-core/src/channels/Actor.kt
@@ -12,7 +12,11 @@
/**
* Scope for [actor][GlobalScope.actor] coroutine builder.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of complex actors.**
+ * See [issue #87](https://github.com/Kotlin/kotlinx.coroutines/issues/87).
*/
+@ObsoleteCoroutinesApi
public interface ActorScope<E> : CoroutineScope, ReceiveChannel<E> {
/**
* A reference to the mailbox channel that this coroutine [receives][receive] messages from.
@@ -105,12 +109,16 @@
* "`for (msg in channel)`" and other cancellable suspending functions throw [CancellationException] and actor
* completes without processing remaining messages.
*
+ * **Note: This API will become obsolete in future updates with introduction of complex actors.**
+ * See [issue #87](https://github.com/Kotlin/kotlinx.coroutines/issues/87).
+ *
* @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
* @param capacity capacity of the channel's buffer (no buffer by default).
* @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
* @param onCompletion optional completion handler for the actor coroutine (see [Job.invokeOnCompletion]).
* @param block the coroutine code.
*/
+@ObsoleteCoroutinesApi
public fun <E> CoroutineScope.actor(
context: CoroutineContext = EmptyCoroutineContext,
capacity: Int = 0,
diff --git a/core/kotlinx-coroutines-core/src/channels/TickerChannels.kt b/core/kotlinx-coroutines-core/src/channels/TickerChannels.kt
index 155aa6f..9d3b1f3 100644
--- a/core/kotlinx-coroutines-core/src/channels/TickerChannels.kt
+++ b/core/kotlinx-coroutines-core/src/channels/TickerChannels.kt
@@ -10,7 +10,10 @@
/**
* Mode for [ticker] function.
+ *
+ * **Note: Ticker channels are not currently integrated with structured concurrency and their api will change in the future.**
*/
+@ObsoleteCoroutinesApi
enum class TickerMode {
/**
* Adjust delay to maintain fixed period if consumer cannot keep up or is otherwise slow.
@@ -49,38 +52,52 @@
*
* **Note** producer to this channel is dispatched via [Dispatchers.Unconfined] by default and started eagerly.
*
- * @param delay delay between each element.
- * @param unit unit of time that applies to [initialDelay] and [delay] (in milliseconds by default).
- * @param initialDelay delay after which the first element will be produced (it is equal to [delay] by default).
+ * **Note: Ticker channels are not currently integrated with structured concurrency and their api will change in the future.**
+ *
+ * @param delayMillis delay between each element in milliseconds.
+ * @param initialDelayMillis delay after which the first element will be produced (it is equal to [delayMillis] by default) in milliseconds.
* @param context context of the producing coroutine.
* @param mode specifies behavior when elements are not received ([FIXED_PERIOD][TickerMode.FIXED_PERIOD] by default).
*/
+@ObsoleteCoroutinesApi
public fun ticker(
- delay: Long,
- unit: TimeUnit = TimeUnit.MILLISECONDS,
- initialDelay: Long = delay,
+ delayMillis: Long,
+ initialDelayMillis: Long = delayMillis,
context: CoroutineContext = EmptyCoroutineContext,
mode: TickerMode = TickerMode.FIXED_PERIOD
): ReceiveChannel<Unit> {
- require(delay >= 0) { "Expected non-negative delay, but has $delay" }
- require(initialDelay >= 0) { "Expected non-negative initial delay, but has $initialDelay" }
+ require(delayMillis >= 0) { "Expected non-negative delay, but has $delayMillis ms" }
+ require(initialDelayMillis >= 0) { "Expected non-negative initial delay, but has $initialDelayMillis ms" }
return GlobalScope.produce(Dispatchers.Unconfined + context, capacity = 0) {
when (mode) {
- TickerMode.FIXED_PERIOD -> fixedPeriodTicker(delay, unit, initialDelay, channel)
- TickerMode.FIXED_DELAY -> fixedDelayTicker(delay, unit, initialDelay, channel)
+ TickerMode.FIXED_PERIOD -> fixedPeriodTicker(delayMillis, initialDelayMillis, channel)
+ TickerMode.FIXED_DELAY -> fixedDelayTicker(delayMillis, initialDelayMillis, channel)
}
}
}
-private suspend fun fixedPeriodTicker(
+/** @suppress **Deprecated**: TimeUnit is not longer supported */
+@Deprecated(
+ message = "TimeUnit is no longer supported",
+ replaceWith = ReplaceWith("ticker(unit.toMillis(delay), unit.toMillis(initialDelay), context, mode)")
+)
+public fun ticker(
delay: Long,
- unit: TimeUnit,
- initialDelay: Long,
+ unit: TimeUnit = TimeUnit.MILLISECONDS, // todo: remove
+ initialDelay: Long = delay,
+ context: CoroutineContext = EmptyCoroutineContext,
+ mode: TickerMode = TickerMode.FIXED_PERIOD
+): ReceiveChannel<Unit> =
+ ticker(unit.toMillis(delay), unit.toMillis(initialDelay), context, mode)
+
+private suspend fun fixedPeriodTicker(
+ delayMillis: Long,
+ initialDelayMillis: Long,
channel: SendChannel<Unit>
) {
- var deadline = timeSource.nanoTime() + unit.toNanos(initialDelay)
- delay(initialDelay, unit)
- val delayNs = unit.toNanos(delay)
+ var deadline = timeSource.nanoTime() + delayToNanos(initialDelayMillis)
+ delay(initialDelayMillis)
+ val delayNs = delayToNanos(delayMillis)
while (true) {
deadline += delayNs
channel.send(Unit)
@@ -89,22 +106,21 @@
if (nextDelay == 0L && delayNs != 0L) {
val adjustedDelay = delayNs - (now - deadline) % delayNs
deadline = now + adjustedDelay
- delay(adjustedDelay, java.util.concurrent.TimeUnit.NANOSECONDS)
+ delay(delayNanosToMillis(adjustedDelay))
} else {
- delay(nextDelay, java.util.concurrent.TimeUnit.NANOSECONDS)
+ delay(delayNanosToMillis(nextDelay))
}
}
}
private suspend fun fixedDelayTicker(
- delay: Long,
- unit: TimeUnit,
- initialDelay: Long,
+ delayMillis: Long,
+ initialDelayMillis: Long,
channel: SendChannel<Unit>
) {
- delay(initialDelay, unit)
+ delay(initialDelayMillis)
while (true) {
channel.send(Unit)
- delay(delay, unit)
+ delay(delayMillis)
}
}
diff --git a/core/kotlinx-coroutines-core/src/scheduling/ExperimentalCoroutineDispatcher.kt b/core/kotlinx-coroutines-core/src/scheduling/ExperimentalCoroutineDispatcher.kt
index 7abbf90..776b7af 100644
--- a/core/kotlinx-coroutines-core/src/scheduling/ExperimentalCoroutineDispatcher.kt
+++ b/core/kotlinx-coroutines-core/src/scheduling/ExperimentalCoroutineDispatcher.kt
@@ -25,7 +25,7 @@
private val corePoolSize: Int,
private val maxPoolSize: Int,
private val idleWorkerKeepAliveNs: Long
-) : ExecutorCoroutineDispatcher(), Delay {
+) : ExecutorCoroutineDispatcher() {
constructor(
corePoolSize: Int = CORE_POOL_SIZE,
maxPoolSize: Int = MAX_POOL_SIZE
@@ -47,9 +47,6 @@
override fun dispatchYield(context: CoroutineContext, block: Runnable): Unit =
coroutineScheduler.dispatch(block, fair = true)
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>): Unit =
- DefaultExecutor.scheduleResumeAfterDelay(time, unit, continuation)
-
// TODO throw error when this API becomes public and close it in tests via another method
override fun close() = coroutineScheduler.close()
@@ -101,7 +98,7 @@
val dispatcher: ExperimentalCoroutineDispatcher,
val parallelism: Int,
override val taskMode: TaskMode
-) : ExecutorCoroutineDispatcher(), Delay, TaskContext, Executor {
+) : ExecutorCoroutineDispatcher(), TaskContext, Executor {
private val queue = ConcurrentLinkedQueue<Runnable>()
private val inFlightTasks = atomic(0)
@@ -193,7 +190,4 @@
next = queue.poll() ?: return
dispatch(next, true)
}
-
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) =
- dispatcher.scheduleResumeAfterDelay(time, unit, continuation)
}
diff --git a/core/kotlinx-coroutines-core/src/test_/TestCoroutineContext.kt b/core/kotlinx-coroutines-core/src/test_/TestCoroutineContext.kt
index 8251d80..deb3e1a 100644
--- a/core/kotlinx-coroutines-core/src/test_/TestCoroutineContext.kt
+++ b/core/kotlinx-coroutines-core/src/test_/TestCoroutineContext.kt
@@ -25,8 +25,12 @@
* This dispatcher's virtual time will be automatically advanced based based on the delayed actions
* within the Coroutine(s).
*
+ * **Note: This API will become obsolete in future updates due to integration with structured concurrency.**
+ * See [issue #541](https://github.com/Kotlin/kotlinx.coroutines/issues/541).
+ *
* @param name A user-readable name for debugging purposes.
*/
+@ObsoleteCoroutinesApi
class TestCoroutineContext(private val name: String? = null) : CoroutineContext {
private val uncaughtExceptions = mutableListOf<Throwable>()
@@ -209,14 +213,14 @@
private inner class Dispatcher : CoroutineDispatcher(), Delay, EventLoop {
override fun dispatch(context: CoroutineContext, block: Runnable) = post(block)
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
postDelayed(Runnable {
with(continuation) { resumeUndispatched(Unit) }
- }, unit.toMillis(time))
+ }, timeMillis)
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
- val node = postDelayed(block, unit.toMillis(time))
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
+ val node = postDelayed(block, timeMillis)
return object : DisposableHandle {
override fun dispose() {
queue.remove(node)
@@ -260,10 +264,14 @@
* [TestCoroutineContext.assertExceptions], the list of unhandled exceptions will have been cleared and this method will
* not throw an [AssertionError].
*
+ * **Note: This API will become obsolete in future updates due to integration with structured concurrency.**
+ * See [issue #541](https://github.com/Kotlin/kotlinx.coroutines/issues/541).
+ *
* @param testContext The provided [TestCoroutineContext]. If not specified, a default [TestCoroutineContext] will be
* provided instead.
* @param testBody The code of the unit-test.
*/
+@ObsoleteCoroutinesApi
public fun withTestContext(testContext: TestCoroutineContext = TestCoroutineContext(), testBody: TestCoroutineContext.() -> Unit) {
with (testContext) {
testBody()
diff --git a/core/kotlinx-coroutines-core/src/timeunit/TimeUnit.kt b/core/kotlinx-coroutines-core/src/timeunit/TimeUnit.kt
index fc32f90..e8bf546 100644
--- a/core/kotlinx-coroutines-core/src/timeunit/TimeUnit.kt
+++ b/core/kotlinx-coroutines-core/src/timeunit/TimeUnit.kt
@@ -6,6 +6,9 @@
/**
* Time unit type alias for writing multiplatform code.
+ *
+ * @suppress **Deprecated** No replacement
*/
@Suppress("ACTUAL_WITHOUT_EXPECT")
+@Deprecated(message = "No replacement")
public actual typealias TimeUnit = java.util.concurrent.TimeUnit
\ No newline at end of file
diff --git a/core/kotlinx-coroutines-core/test/Threads.kt b/core/kotlinx-coroutines-core/test/Threads.kt
index c0c5435..ca69586 100644
--- a/core/kotlinx-coroutines-core/test/Threads.kt
+++ b/core/kotlinx-coroutines-core/test/Threads.kt
@@ -37,7 +37,7 @@
println("===")
}
-fun ThreadPoolDispatcher.dumpThreads(header: String) =
+fun ExecutorCoroutineDispatcher.dumpThreads(header: String) =
currentThreads().filter { it is PoolThread && it.dispatcher == this@dumpThreads }.dumpThreads(header)
fun checkTestThreads(threadsBefore: Set<Thread>) {
diff --git a/core/kotlinx-coroutines-core/test/channels/BroadcastChannelMultiReceiveStressTest.kt b/core/kotlinx-coroutines-core/test/channels/BroadcastChannelMultiReceiveStressTest.kt
index f234d04..8b9c346 100644
--- a/core/kotlinx-coroutines-core/test/channels/BroadcastChannelMultiReceiveStressTest.kt
+++ b/core/kotlinx-coroutines-core/test/channels/BroadcastChannelMultiReceiveStressTest.kt
@@ -6,7 +6,6 @@
import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.selects.*
-import kotlinx.coroutines.experimental.timeunit.*
import org.junit.*
import org.junit.runner.*
import org.junit.runners.*
@@ -87,7 +86,7 @@
println(" Sent $total events, waiting for receivers")
stopOnReceive.set(total)
try {
- withTimeout(5, TimeUnit.SECONDS) {
+ withTimeout(5000) {
receivers.forEachIndexed { index, receiver ->
if (lastReceived[index].get() == total)
receiver.cancel()
diff --git a/core/kotlinx-coroutines-core/test/channels/TickerChannelCommonTest.kt b/core/kotlinx-coroutines-core/test/channels/TickerChannelCommonTest.kt
index b413097..9ad8ae3 100644
--- a/core/kotlinx-coroutines-core/test/channels/TickerChannelCommonTest.kt
+++ b/core/kotlinx-coroutines-core/test/channels/TickerChannelCommonTest.kt
@@ -23,12 +23,12 @@
enum class Channel {
FIXED_PERIOD {
override fun invoke(delay: Long, initialDelay: Long) =
- ticker(delay, initialDelay = initialDelay, mode = TickerMode.FIXED_PERIOD)
+ ticker(delay, initialDelayMillis = initialDelay, mode = TickerMode.FIXED_PERIOD)
},
FIXED_DELAY {
override fun invoke(delay: Long, initialDelay: Long) =
- ticker(delay, initialDelay = initialDelay, mode = TickerMode.FIXED_DELAY)
+ ticker(delay, initialDelayMillis = initialDelay, mode = TickerMode.FIXED_DELAY)
};
abstract operator fun invoke(delay: Long, initialDelay: Long = 0): ReceiveChannel<Unit>
diff --git a/core/kotlinx-coroutines-core/test/channels/TickerChannelTest.kt b/core/kotlinx-coroutines-core/test/channels/TickerChannelTest.kt
index 8bf9f4b..37c1c6e 100644
--- a/core/kotlinx-coroutines-core/test/channels/TickerChannelTest.kt
+++ b/core/kotlinx-coroutines-core/test/channels/TickerChannelTest.kt
@@ -11,7 +11,7 @@
@Test
fun testFixedDelayChannelBackpressure() = withVirtualTimeSource {
runTest {
- val delayChannel = ticker(delay = 1000, initialDelay = 0, mode = TickerMode.FIXED_DELAY)
+ val delayChannel = ticker(delayMillis = 1000, initialDelayMillis = 0, mode = TickerMode.FIXED_DELAY)
delayChannel.checkNotEmpty()
delayChannel.checkEmpty()
@@ -28,7 +28,7 @@
@Test
fun testDelayChannelBackpressure() = withVirtualTimeSource {
runTest {
- val delayChannel = ticker(delay = 1000, initialDelay = 0)
+ val delayChannel = ticker(delayMillis = 1000, initialDelayMillis = 0)
delayChannel.checkNotEmpty()
delayChannel.checkEmpty()
@@ -47,7 +47,7 @@
@Test
fun testDelayChannelBackpressure2() = withVirtualTimeSource {
runTest {
- val delayChannel = ticker(delay = 1000, initialDelay = 0)
+ val delayChannel = ticker(delayMillis = 1000, initialDelayMillis = 0)
delayChannel.checkNotEmpty()
delayChannel.checkEmpty()
diff --git a/core/kotlinx-coroutines-core/test/exceptions/JobExceptionsStressTest.kt b/core/kotlinx-coroutines-core/test/exceptions/JobExceptionsStressTest.kt
index 760cc05..878e3ba 100644
--- a/core/kotlinx-coroutines-core/test/exceptions/JobExceptionsStressTest.kt
+++ b/core/kotlinx-coroutines-core/test/exceptions/JobExceptionsStressTest.kt
@@ -13,7 +13,7 @@
class JobExceptionsStressTest : TestBase() {
- private val executor: ThreadPoolDispatcher by lazy { newFixedThreadPoolContext(5, "JobExceptionsStressTest") }
+ private val executor = newFixedThreadPoolContext(5, "JobExceptionsStressTest")
@After
fun tearDown() {
diff --git a/core/kotlinx-coroutines-core/test/guide/test/CancellationTimeOutsGuideTest.kt b/core/kotlinx-coroutines-core/test/guide/test/CancellationTimeOutsGuideTest.kt
index 17b0998..36a7875 100644
--- a/core/kotlinx-coroutines-core/test/guide/test/CancellationTimeOutsGuideTest.kt
+++ b/core/kotlinx-coroutines-core/test/guide/test/CancellationTimeOutsGuideTest.kt
@@ -71,7 +71,7 @@
"I'm sleeping 0 ...",
"I'm sleeping 1 ...",
"I'm sleeping 2 ...",
- "Exception in thread \"main\" kotlinx.coroutines.experimental.TimeoutCancellationException: Timed out waiting for 1300 MILLISECONDS"
+ "Exception in thread \"main\" kotlinx.coroutines.experimental.TimeoutCancellationException: Timed out waiting for 1300 ms"
)
}
diff --git a/docs/cancellation-and-timeouts.md b/docs/cancellation-and-timeouts.md
index 9522ca3..b2373f4 100644
--- a/docs/cancellation-and-timeouts.md
+++ b/docs/cancellation-and-timeouts.md
@@ -274,7 +274,7 @@
I'm sleeping 0 ...
I'm sleeping 1 ...
I'm sleeping 2 ...
-Exception in thread "main" kotlinx.coroutines.experimental.TimeoutCancellationException: Timed out waiting for 1300 MILLISECONDS
+Exception in thread "main" kotlinx.coroutines.experimental.TimeoutCancellationException: Timed out waiting for 1300 ms
```
<!--- TEST STARTS_WITH -->
@@ -327,7 +327,7 @@
[isActive]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/is-active.html
[CoroutineScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-scope/index.html
[withContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/with-context.html
-[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable/index.html
+[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable.html
[withTimeout]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/with-timeout.html
[withTimeoutOrNull]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/with-timeout-or-null.html
<!--- END -->
diff --git a/docs/coroutine-context-and-dispatchers.md b/docs/coroutine-context-and-dispatchers.md
index e9247cf..b900db5 100644
--- a/docs/coroutine-context-and-dispatchers.md
+++ b/docs/coroutine-context-and-dispatchers.md
@@ -105,7 +105,7 @@
[newSingleThreadContext] creates a new thread for the coroutine to run.
A dedicated thread is a very expensive resource.
-In a real application it must be either released, when no longer needed, using [close][ThreadPoolDispatcher.close]
+In a real application it must be either released, when no longer needed, using [close][ExecutorCoroutineDispatcher.close]
function, or stored in a top-level variable and reused throughout the application.
### Unconfined vs confined dispatcher
@@ -608,7 +608,7 @@
[GlobalScope]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-global-scope/index.html
[Dispatchers.Default]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-dispatchers/-default.html
[newSingleThreadContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/new-single-thread-context.html
-[ThreadPoolDispatcher.close]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-thread-pool-dispatcher/close.html
+[ExecutorCoroutineDispatcher.close]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-executor-coroutine-dispatcher/close.html
[runBlocking]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/run-blocking.html
[delay]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/delay.html
[newCoroutineContext]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/new-coroutine-context.html
diff --git a/integration/kotlinx-coroutines-guava/src/ListenableFuture.kt b/integration/kotlinx-coroutines-guava/src/ListenableFuture.kt
index bb77ad0..2a7f6a2 100644
--- a/integration/kotlinx-coroutines-guava/src/ListenableFuture.kt
+++ b/integration/kotlinx-coroutines-guava/src/ListenableFuture.kt
@@ -28,12 +28,29 @@
*
* @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
* @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
* @param block the coroutine code.
*/
public fun <T> CoroutineScope.future(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
+ block: suspend CoroutineScope.() -> T
+): ListenableFuture<T> {
+ require(!start.isLazy) { "$start start is not supported" }
+ val newContext = newCoroutineContext(context)
+ val job = Job(newContext[Job])
+ val future = ListenableFutureCoroutine<T>(newContext + job)
+ job.cancelFutureOnCompletion(future)
+ start(block, receiver=future, completion=future) // use the specified start strategy
+ return future
+}
+
+/**
+ * @suppress **Deprecated**: onCompletion parameter is deprecated.
+ */
+@Deprecated("onCompletion parameter is deprecated")
+public fun <T> CoroutineScope.future(
+ context: CoroutineContext = EmptyCoroutineContext,
+ start: CoroutineStart = CoroutineStart.DEFAULT,
onCompletion: CompletionHandler? = null,
block: suspend CoroutineScope.() -> T
): ListenableFuture<T> {
diff --git a/integration/kotlinx-coroutines-jdk8/src/channels8/Channels.kt b/integration/kotlinx-coroutines-jdk8/src/channels8/Channels.kt
index c199328..96390bd 100644
--- a/integration/kotlinx-coroutines-jdk8/src/channels8/Channels.kt
+++ b/integration/kotlinx-coroutines-jdk8/src/channels8/Channels.kt
@@ -17,6 +17,7 @@
/**
* Creates a [ProducerJob] to read all element of the [Stream].
*/
+@Deprecated("No replacement")
public fun <E> Stream<E>.asReceiveChannel(context: CoroutineContext = EmptyCoroutineContext): ReceiveChannel<E> =
GlobalScope.produce(context) {
for (element in this@asReceiveChannel)
@@ -26,6 +27,7 @@
/**
* Creates a [Stream] of elements in this [ReceiveChannel].
*/
+@Deprecated(message = "Use toList().stream()", replaceWith = ReplaceWith("toList().stream()"))
public fun <E : Any> ReceiveChannel<E>.asStream(): Stream<E> = StreamSupport.stream<E>(SpliteratorAdapter(this), false)
/**
diff --git a/integration/kotlinx-coroutines-jdk8/src/future/Future.kt b/integration/kotlinx-coroutines-jdk8/src/future/Future.kt
index ebb1c84..b2809fb 100644
--- a/integration/kotlinx-coroutines-jdk8/src/future/Future.kt
+++ b/integration/kotlinx-coroutines-jdk8/src/future/Future.kt
@@ -28,15 +28,33 @@
*
* @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
* @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
* @param block the coroutine code.
*/
public fun <T> CoroutineScope.future(
+ context: CoroutineContext = EmptyCoroutineContext,
+ start: CoroutineStart = CoroutineStart.DEFAULT,
+ block: suspend CoroutineScope.() -> T
+) : CompletableFuture<T> {
+ require(!start.isLazy) { "$start start is not supported" }
+ val newContext = this.newCoroutineContext(context)
+ val job = Job(newContext[Job])
+ val future = CompletableFutureCoroutine<T>(newContext + job)
+ job.cancelFutureOnCompletion(future)
+ future.whenComplete { _, exception -> job.cancel(exception) }
+ start(block, receiver = future, completion = future) // use the specified start strategy
+ return future
+}
+
+/**
+ * @suppress **Deprecated**: onCompletion parameter is deprecated.
+ */
+@Deprecated("onCompletion parameter is deprecated")
+public fun <T> CoroutineScope.future(
context: CoroutineContext = Dispatchers.Default,
start: CoroutineStart = CoroutineStart.DEFAULT,
onCompletion: CompletionHandler? = null,
block: suspend CoroutineScope.() -> T
-): CompletableFuture<T> {
+) : CompletableFuture<T> {
require(!start.isLazy) { "$start start is not supported" }
val newContext = this.newCoroutineContext(context)
val job = Job(newContext[Job])
diff --git a/integration/kotlinx-coroutines-jdk8/src/time/Time.kt b/integration/kotlinx-coroutines-jdk8/src/time/Time.kt
index de30e5a..2f05a68 100644
--- a/integration/kotlinx-coroutines-jdk8/src/time/Time.kt
+++ b/integration/kotlinx-coroutines-jdk8/src/time/Time.kt
@@ -12,36 +12,44 @@
* "java.time" adapter method for [kotlinx.coroutines.experimental.delay]
*/
public suspend fun delay(duration: Duration) =
- kotlinx.coroutines.experimental.delay(duration.toNanos(), TimeUnit.NANOSECONDS)
+ kotlinx.coroutines.experimental.delay(duration.toMillis())
/**
* "java.time" adapter method for [SelectBuilder.onTimeout]
*/
-public suspend fun <R> SelectBuilder<R>.onTimeout(duration: Duration, block: suspend () -> R) =
- onTimeout(duration.toNanos(), TimeUnit.NANOSECONDS, block)
+public fun <R> SelectBuilder<R>.onTimeout(duration: Duration, block: suspend () -> R) =
+ onTimeout(duration.toMillis(), block)
+
+/**
+ * @suppress
+ */
+@Deprecated(level = DeprecationLevel.HIDDEN, message = "binary")
+@JvmName("onTimeout")
+public suspend fun <R> SelectBuilder<R>.onTimeout0(duration: Duration, block: suspend () -> R) =
+ onTimeout(duration.toMillis(), block)
/**
* "java.time" adapter method for [kotlinx.coroutines.experimental.withTimeout]
*/
public suspend fun <T> withTimeout(duration: Duration, block: suspend CoroutineScope.() -> T): T =
- kotlinx.coroutines.experimental.withTimeout(duration.toNanos(), TimeUnit.NANOSECONDS, block)
+ kotlinx.coroutines.experimental.withTimeout(duration.toMillis(), block)
/**
* @suppress **Deprecated**: for binary compatibility only
*/
@Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
public suspend fun <T> withTimeout(duration: Duration, block: suspend () -> T): T =
- kotlinx.coroutines.experimental.withTimeout(duration.toNanos(), TimeUnit.NANOSECONDS) { block() }
+ kotlinx.coroutines.experimental.withTimeout(duration.toNanos(), TimeUnit.NANOSECONDS) { block() }
/**
* "java.time" adapter method for [kotlinx.coroutines.experimental.withTimeoutOrNull]
*/
public suspend fun <T> withTimeoutOrNull(duration: Duration, block: suspend CoroutineScope.() -> T): T? =
- kotlinx.coroutines.experimental.withTimeoutOrNull(duration.toNanos(), TimeUnit.NANOSECONDS, block)
+ kotlinx.coroutines.experimental.withTimeoutOrNull(duration.toMillis(), block)
/**
* @suppress **Deprecated**: for binary compatibility only
*/
@Deprecated("for binary compatibility only", level=DeprecationLevel.HIDDEN)
public suspend fun <T> withTimeoutOrNull(duration: Duration, block: suspend () -> T): T? =
- kotlinx.coroutines.experimental.withTimeoutOrNull(duration.toNanos(), TimeUnit.NANOSECONDS) { block() }
+ kotlinx.coroutines.experimental.withTimeoutOrNull(duration.toNanos(), TimeUnit.NANOSECONDS) { block() }
diff --git a/integration/kotlinx-coroutines-slf4j/src/MDCContext.kt b/integration/kotlinx-coroutines-slf4j/src/MDCContext.kt
index 6976b36..e70c8a5 100644
--- a/integration/kotlinx-coroutines-slf4j/src/MDCContext.kt
+++ b/integration/kotlinx-coroutines-slf4j/src/MDCContext.kt
@@ -50,12 +50,14 @@
*/
companion object Key : CoroutineContext.Key<MDCContext>
+ /** @suppress */
override fun updateThreadContext(context: CoroutineContext): MDCContextMap {
val oldState = MDC.getCopyOfContextMap()
setCurrent(contextMap)
return oldState
}
+ /** @suppress */
override fun restoreThreadContext(context: CoroutineContext, oldState: MDCContextMap) {
setCurrent(oldState)
}
diff --git a/js/kotlinx-coroutines-core-js/README.md b/js/kotlinx-coroutines-core-js/README.md
index a4dff92..21e6b7b 100644
--- a/js/kotlinx-coroutines-core-js/README.md
+++ b/js/kotlinx-coroutines-core-js/README.md
@@ -69,7 +69,7 @@
[CoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-dispatcher/index.html
[Dispatchers.Default]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-dispatchers/-default.html
[Dispatchers.Unconfined]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-dispatchers/-unconfined.html
-[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable/index.html
+[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable.html
[CoroutineExceptionHandler]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-exception-handler/index.html
[delay]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/delay.html
[yield]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/yield.html
diff --git a/js/kotlinx-coroutines-core-js/src/JSDispatcher.kt b/js/kotlinx-coroutines-core-js/src/JSDispatcher.kt
index 8317cae..e94bcb4 100644
--- a/js/kotlinx-coroutines-core-js/src/JSDispatcher.kt
+++ b/js/kotlinx-coroutines-core-js/src/JSDispatcher.kt
@@ -4,17 +4,21 @@
package kotlinx.coroutines.experimental
-import kotlinx.coroutines.experimental.timeunit.TimeUnit
import kotlin.coroutines.experimental.*
import org.w3c.dom.*
+private const val MAX_DELAY = Int.MAX_VALUE.toLong()
+
+private fun delayToInt(timeMillis: Long): Int =
+ timeMillis.coerceIn(0, MAX_DELAY).toInt()
+
internal class NodeDispatcher : CoroutineDispatcher(), Delay {
override fun dispatch(context: CoroutineContext, block: Runnable) {
setTimeout({ block.run() }, 0)
}
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
- val handle = setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, time.toIntMillis(unit))
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
+ val handle = setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, delayToInt(timeMillis))
// Actually on cancellation, but clearTimeout is idempotent
continuation.invokeOnCancellation(handler = ClearTimeout(handle).asHandler)
}
@@ -25,8 +29,8 @@
override fun toString(): String = "ClearTimeout[$handle]"
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
- val handle = setTimeout({ block.run() }, time.toIntMillis(unit))
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
+ val handle = setTimeout({ block.run() }, delayToInt(timeMillis))
return ClearTimeout(handle)
}
}
@@ -53,12 +57,12 @@
queue.enqueue(block)
}
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
- window.setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, time.toIntMillis(unit))
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
+ window.setTimeout({ with(continuation) { resumeUndispatched(Unit) } }, delayToInt(timeMillis))
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
- val handle = window.setTimeout({ block.run() }, time.toIntMillis(unit))
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
+ val handle = window.setTimeout({ block.run() }, delayToInt(timeMillis))
return object : DisposableHandle {
override fun dispose() {
window.clearTimeout(handle)
@@ -99,9 +103,6 @@
}
}
-private fun Long.toIntMillis(unit: TimeUnit): Int =
- unit.toMillis(this).coerceIn(0L, Int.MAX_VALUE.toLong()).toInt()
-
internal open class Queue<T : Any> {
private var queue = arrayOfNulls<Any?>(8)
private var head = 0
diff --git a/js/kotlinx-coroutines-core-js/src/Promise.kt b/js/kotlinx-coroutines-core-js/src/Promise.kt
index 929cd70..a963347 100644
--- a/js/kotlinx-coroutines-core-js/src/Promise.kt
+++ b/js/kotlinx-coroutines-core-js/src/Promise.kt
@@ -20,16 +20,26 @@
*
* @param context additional to [CoroutineScope.coroutineContext] context of the coroutine.
* @param start coroutine start option. The default value is [CoroutineStart.DEFAULT].
- * @param onCompletion optional completion handler for the coroutine (see [Job.invokeOnCompletion]).
* @param block the coroutine code.
*/
public fun <T> CoroutineScope.promise(
context: CoroutineContext = EmptyCoroutineContext,
start: CoroutineStart = CoroutineStart.DEFAULT,
+ block: suspend CoroutineScope.() -> T
+): Promise<T> =
+ async(context, start, block).asPromise()
+
+/**
+ * @suppress **Deprecated**: onCompletion parameter is deprecated.
+ */
+@Deprecated("onCompletion parameter is deprecated")
+public fun <T> CoroutineScope.promise(
+ context: CoroutineContext = EmptyCoroutineContext,
+ start: CoroutineStart = CoroutineStart.DEFAULT,
onCompletion: CompletionHandler? = null,
block: suspend CoroutineScope.() -> T
): Promise<T> =
- async(context, start, onCompletion, block = block).asPromise()
+ async(context, start, block).also { if (onCompletion != null) it.invokeOnCompletion(onCompletion) }.asPromise()
/**
* Starts new coroutine and returns its result as an implementation of [Promise].
diff --git a/native/kotlinx-coroutines-core-native/README.md b/native/kotlinx-coroutines-core-native/README.md
index 824268a..de0bd06 100644
--- a/native/kotlinx-coroutines-core-native/README.md
+++ b/native/kotlinx-coroutines-core-native/README.md
@@ -72,7 +72,7 @@
[CoroutineDispatcher]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-dispatcher/index.html
[Dispatchers.Default]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-dispatchers/-default.html
[Dispatchers.Unconfined]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-dispatchers/-unconfined.html
-[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable/index.html
+[NonCancellable]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-non-cancellable.html
[CoroutineExceptionHandler]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/-coroutine-exception-handler/index.html
[delay]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/delay.html
[yield]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental/yield.html
diff --git a/native/kotlinx-coroutines-core-native/src/CoroutineContext.kt b/native/kotlinx-coroutines-core-native/src/CoroutineContext.kt
index 6fca7ff..cb54cb8 100644
--- a/native/kotlinx-coroutines-core-native/src/CoroutineContext.kt
+++ b/native/kotlinx-coroutines-core-native/src/CoroutineContext.kt
@@ -5,7 +5,6 @@
package kotlinx.coroutines.experimental
import kotlin.coroutines.experimental.*
-import kotlinx.coroutines.experimental.timeunit.*
internal val currentEventLoop = ArrayList<BlockingEventLoop>()
@@ -15,10 +14,10 @@
internal object DefaultExecutor : CoroutineDispatcher(), Delay {
override fun dispatch(context: CoroutineContext, block: Runnable) =
takeEventLoop().dispatch(context, block)
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) =
- takeEventLoop().scheduleResumeAfterDelay(time, unit, continuation)
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
- takeEventLoop().invokeOnTimeout(time, unit, block)
+ override fun scheduleResumeAfterDelay(time: Long, continuation: CancellableContinuation<Unit>) =
+ takeEventLoop().scheduleResumeAfterDelay(time, continuation)
+ override fun invokeOnTimeout(time: Long, block: Runnable): DisposableHandle =
+ takeEventLoop().invokeOnTimeout(time, block)
fun execute(task: Runnable) {
error("Cannot execute task because event loop was shut down")
diff --git a/native/kotlinx-coroutines-core-native/src/EventLoop.kt b/native/kotlinx-coroutines-core-native/src/EventLoop.kt
index e3a9227..021db70 100644
--- a/native/kotlinx-coroutines-core-native/src/EventLoop.kt
+++ b/native/kotlinx-coroutines-core-native/src/EventLoop.kt
@@ -7,7 +7,6 @@
import kotlinx.atomicfu.*
import kotlinx.cinterop.*
import kotlinx.coroutines.experimental.internal.*
-import kotlinx.coroutines.experimental.timeunit.*
import platform.posix.*
import kotlin.coroutines.experimental.*
import kotlin.system.*
@@ -44,6 +43,15 @@
private const val REMOVED = 1
private const val RESCHEDULED = 2
+private const val MS_TO_NS = 1_000_000L
+private const val MAX_MS = Long.MAX_VALUE / MS_TO_NS
+
+private fun delayToNanos(timeMillis: Long): Long = when {
+ timeMillis <= 0 -> 0L
+ timeMillis >= MAX_MS -> Long.MAX_VALUE
+ else -> timeMillis * MS_TO_NS
+}
+
@Suppress("PrivatePropertyName")
private val CLOSED_EMPTY = Symbol("CLOSED_EMPTY")
@@ -86,11 +94,11 @@
override fun dispatch(context: CoroutineContext, block: Runnable) =
execute(block)
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) =
- schedule(DelayedResumeTask(time, unit, continuation))
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) =
+ schedule(DelayedResumeTask(timeMillis, continuation))
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
- DelayedRunnableTask(time, unit, block).also { schedule(it) }
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle =
+ DelayedRunnableTask(timeMillis, block).also { schedule(it) }
override fun processNextEvent(): Long {
// queue all delayed tasks that are due to be executed
@@ -235,11 +243,11 @@
}
internal abstract inner class DelayedTask(
- time: Long, timeUnit: TimeUnit
+ timeMillis: Long
) : Runnable, Comparable<DelayedTask>, DisposableHandle, ThreadSafeHeapNode {
override var index: Int = -1
var state = DELAYED // Guarded by by lock on this task for reschedule/dispose purposes
- val nanoTime: Long = nanoTime() + timeUnit.toNanos(time)
+ val nanoTime: Long = nanoTime() + delayToNanos(timeMillis)
override fun compareTo(other: DelayedTask): Int {
val dTime = nanoTime - other.nanoTime
@@ -274,18 +282,18 @@
}
private inner class DelayedResumeTask(
- time: Long, timeUnit: TimeUnit,
+ timeMillis: Long,
private val cont: CancellableContinuation<Unit>
- ) : DelayedTask(time, timeUnit) {
+ ) : DelayedTask(timeMillis) {
override fun run() {
with(cont) { resumeUndispatched(Unit) }
}
}
private inner class DelayedRunnableTask(
- time: Long, timeUnit: TimeUnit,
+ timeMillis: Long,
private val block: Runnable
- ) : DelayedTask(time, timeUnit) {
+ ) : DelayedTask(timeMillis) {
override fun run() { block.run() }
override fun toString(): String = super.toString() + block.toString()
}
diff --git a/reactive/coroutines-guide-reactive.md b/reactive/coroutines-guide-reactive.md
index e4b05bf..29cd34f 100644
--- a/reactive/coroutines-guide-reactive.md
+++ b/reactive/coroutines-guide-reactive.md
@@ -531,7 +531,9 @@
<!--- TEST -->
-Another implementation of [BroadcastChannel] is [ArrayBroadcastChannel]. It delivers every event to every
+Another implementation of [BroadcastChannel] is `ArrayBroadcastChannel` with an array-based buffer of
+a specified `capacity`. It can be created with `BroadcastChannel(capacity)`.
+It delivers every event to every
subscriber since the moment the corresponding subscription is open. It corresponds to
[PublishSubject](http://reactivex.io/RxJava/2.x/javadoc/io/reactivex/subjects/PublishSubject.html) in Rx.
The capacity of the buffer in the constructor of `ArrayBroadcastChannel` controls the numbers of elements
@@ -1069,7 +1071,6 @@
[SendChannel.send]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/send.html
[BroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-broadcast-channel/index.html
[ConflatedBroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-conflated-broadcast-channel/index.html
-[ArrayBroadcastChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-array-broadcast-channel/index.html
<!--- INDEX kotlinx.coroutines.experimental.selects -->
[select]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/select.html
[whileSelect]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.selects/while-select.html
diff --git a/reactive/kotlinx-coroutines-reactive/src/Channel.kt b/reactive/kotlinx-coroutines-reactive/src/Channel.kt
index a3fe5f1..42556db 100644
--- a/reactive/kotlinx-coroutines-reactive/src/Channel.kt
+++ b/reactive/kotlinx-coroutines-reactive/src/Channel.kt
@@ -5,14 +5,20 @@
package kotlinx.coroutines.experimental.reactive
import kotlinx.atomicfu.*
+import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.*
import org.reactivestreams.*
/**
* Subscribes to this [Publisher] and returns a channel to receive elements emitted by it.
* The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this publisher.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
+ *
* @param request how many items to request from publisher in advance (optional, on-demand request by default).
*/
+@ObsoleteCoroutinesApi
@Suppress("CONFLICTING_OVERLOADS")
public fun <T> Publisher<T>.openSubscription(request: Int = 0): ReceiveChannel<T> {
val channel = SubscriptionChannel<T>(request)
diff --git a/reactive/kotlinx-coroutines-reactive/src/Convert.kt b/reactive/kotlinx-coroutines-reactive/src/Convert.kt
index 5fa826a..f6487db 100644
--- a/reactive/kotlinx-coroutines-reactive/src/Convert.kt
+++ b/reactive/kotlinx-coroutines-reactive/src/Convert.kt
@@ -15,8 +15,12 @@
* Every subscriber receives values from this channel in **fan-out** fashion. If the are multiple subscribers,
* they'll receive values in round-robin way.
*
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
+ *
* @param context -- the coroutine context from which the resulting observable is going to be signalled
*/
+@ObsoleteCoroutinesApi
public fun <T> ReceiveChannel<T>.asPublisher(context: CoroutineContext = EmptyCoroutineContext): Publisher<T> = GlobalScope.publish(context) {
for (t in this@asPublisher)
send(t)
diff --git a/reactive/kotlinx-coroutines-reactive/src/Publish.kt b/reactive/kotlinx-coroutines-reactive/src/Publish.kt
index 7222208..f86b5d8 100644
--- a/reactive/kotlinx-coroutines-reactive/src/Publish.kt
+++ b/reactive/kotlinx-coroutines-reactive/src/Publish.kt
@@ -31,9 +31,13 @@
* The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
* with corresponding [coroutineContext] element.
*
+ * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect
+ * to cancellation and error handling may change in the future.
+ *
* @param context context of the coroutine.
* @param block the coroutine code.
*/
+@ExperimentalCoroutinesApi
public fun <T> CoroutineScope.publish(
context: CoroutineContext = EmptyCoroutineContext,
block: suspend ProducerScope<T>.() -> Unit
@@ -62,6 +66,7 @@
private const val CLOSED = -1L // closed, but have not signalled onCompleted/onError yet
private const val SIGNALLED = -2L // already signalled subscriber onCompleted/onError
+@Suppress("CONFLICTING_JVM_DECLARATIONS", "RETURN_TYPE_MISMATCH_ON_INHERITANCE")
private class PublisherCoroutine<in T>(
parentContext: CoroutineContext,
private val subscriber: Subscriber<T>
@@ -216,7 +221,7 @@
}
// Subscription impl
- override fun cancel() {
- cancel(cause = null)
- }
+ @JvmName("cancel")
+ @Suppress("NOTHING_TO_OVERRIDE", "ACCIDENTAL_OVERRIDE")
+ override fun cancelSubscription() = super.cancel(null)
}
\ No newline at end of file
diff --git a/reactive/kotlinx-coroutines-reactor/src/Convert.kt b/reactive/kotlinx-coroutines-reactor/src/Convert.kt
index 822a64a..d4d565c 100644
--- a/reactive/kotlinx-coroutines-reactor/src/Convert.kt
+++ b/reactive/kotlinx-coroutines-reactor/src/Convert.kt
@@ -16,9 +16,27 @@
* Every subscriber gets the signal at the same time.
* Unsubscribing from the resulting mono **does not** affect the original job in any way.
*
+ * **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
+ * in the future to account for the concept of structured concurrency.
+ *
* @param context -- the coroutine context from which the resulting mono is going to be signalled
*/
-public fun Job.asMono(context: CoroutineContext = EmptyCoroutineContext): Mono<Unit> = GlobalScope.mono(context) { this@asMono.join() }
+@ExperimentalCoroutinesApi
+public fun Job.asMono(context: CoroutineContext): Mono<Unit> = GlobalScope.mono(context) { this@asMono.join() }
+
+/**
+ * @suppress **Deprecated**: Specify explicit context
+ */
+@Deprecated("Specify explicit context", level = DeprecationLevel.HIDDEN)
+@JvmName("asMono\$default")
+public fun Job.asMono0(context: CoroutineContext?, flags: Int, obj: Any?): Mono<Unit> =
+ asMono(context ?: EmptyCoroutineContext)
+
+/**
+ * @suppress **Deprecated**: Specify explicit context
+ */
+@Deprecated("Specify explicit context", replaceWith = ReplaceWith("asMono(EmptyCoroutineContext)"))
+public fun Job.asMono(): Mono<Unit> = asMono(EmptyCoroutineContext)
/**
* Converts this deferred value to the hot reactive mono that signals
@@ -27,9 +45,27 @@
* Every subscriber gets the same completion value.
* Unsubscribing from the resulting mono **does not** affect the original deferred value in any way.
*
+ * **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
+ * in the future to account for the concept of structured concurrency.
+ *
* @param context -- the coroutine context from which the resulting mono is going to be signalled
*/
-public fun <T> Deferred<T?>.asMono(context: CoroutineContext = EmptyCoroutineContext): Mono<T> = GlobalScope.mono(context) { this@asMono.await() }
+@ExperimentalCoroutinesApi
+public fun <T> Deferred<T?>.asMono(context: CoroutineContext): Mono<T> = GlobalScope.mono(context) { this@asMono.await() }
+
+/**
+ * @suppress **Deprecated**: Specify explicit context
+ */
+@Deprecated("Specify explicit context", level = DeprecationLevel.HIDDEN)
+@JvmName("asMono\$default")
+public fun <T> Deferred<T?>.asMono0(context: CoroutineContext?, flags: Int, obj: Any?): Mono<T> =
+ asMono(context ?: EmptyCoroutineContext)
+
+/**
+ * @suppress **Deprecated**: Specify explicit context
+ */
+@Deprecated("Specify explicit context", replaceWith = ReplaceWith("asMono(EmptyCoroutineContext)"))
+public fun <T> Deferred<T?>.asMono(): Mono<T> = asMono(EmptyCoroutineContext)
/**
* Converts a stream of elements received from the channel to the hot reactive flux.
@@ -37,8 +73,12 @@
* Every subscriber receives values from this channel in **fan-out** fashion. If the are multiple subscribers,
* they'll receive values in round-robin way.
*
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
+ *
* @param context -- the coroutine context from which the resulting flux is going to be signalled
*/
+@ObsoleteCoroutinesApi
public fun <T> ReceiveChannel<T>.asFlux(context: CoroutineContext = EmptyCoroutineContext): Flux<T> = GlobalScope.flux(context) {
for (t in this@asFlux)
send(t)
diff --git a/reactive/kotlinx-coroutines-reactor/src/Flux.kt b/reactive/kotlinx-coroutines-reactor/src/Flux.kt
index 73e1a6a..1873d3b 100644
--- a/reactive/kotlinx-coroutines-reactor/src/Flux.kt
+++ b/reactive/kotlinx-coroutines-reactor/src/Flux.kt
@@ -28,7 +28,11 @@
* | `send` | `onNext`
* | Normal completion or `close` without cause | `onComplete`
* | Failure with exception or `close` with cause | `onError`
+ *
+ * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect
+ * to cancellation and error handling may change in the future.
*/
+@ExperimentalCoroutinesApi
fun <T> CoroutineScope.flux(
context: CoroutineContext = EmptyCoroutineContext,
block: suspend ProducerScope<T>.() -> Unit
diff --git a/reactive/kotlinx-coroutines-reactor/src/Scheduler.kt b/reactive/kotlinx-coroutines-reactor/src/Scheduler.kt
index 6496732..1819b36 100644
--- a/reactive/kotlinx-coroutines-reactor/src/Scheduler.kt
+++ b/reactive/kotlinx-coroutines-reactor/src/Scheduler.kt
@@ -7,35 +7,46 @@
import kotlinx.coroutines.experimental.*
import reactor.core.Disposable
import reactor.core.scheduler.Scheduler
-import java.util.concurrent.TimeUnit
+import java.util.concurrent.*
import kotlin.coroutines.experimental.CoroutineContext
/**
* Converts an instance of [Scheduler] to an implementation of [CoroutineDispatcher].
*/
-fun Scheduler.asCoroutineDispatcher() = SchedulerCoroutineDispatcher(this)
+fun Scheduler.asCoroutineDispatcher(): SchedulerCoroutineDispatcher = SchedulerCoroutineDispatcher(this)
/**
* Implements [CoroutineDispatcher] on top of an arbitrary [Scheduler].
* @param scheduler a scheduler.
*/
-open class SchedulerCoroutineDispatcher(private val scheduler: Scheduler) : CoroutineDispatcher(), Delay {
+public class SchedulerCoroutineDispatcher(
+ /**
+ * Underlying scheduler of current [CoroutineDispatcher].
+ */
+ public val scheduler: Scheduler
+) : CoroutineDispatcher(), Delay {
+ /** @suppress */
override fun dispatch(context: CoroutineContext, block: Runnable) {
scheduler.schedule(block)
}
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
+ /** @suppress */
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
val disposable = scheduler.schedule({
with(continuation) { resumeUndispatched(Unit) }
- }, time, unit)
+ }, timeMillis, TimeUnit.MILLISECONDS)
continuation.disposeOnCancellation(disposable.asDisposableHandle())
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle =
- scheduler.schedule(block, time, unit).asDisposableHandle()
+ /** @suppress */
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle =
+ scheduler.schedule(block, timeMillis, TimeUnit.MILLISECONDS).asDisposableHandle()
+ /** @suppress */
override fun toString(): String = scheduler.toString()
+ /** @suppress */
override fun equals(other: Any?): Boolean = other is SchedulerCoroutineDispatcher && other.scheduler === scheduler
+ /** @suppress */
override fun hashCode(): Int = System.identityHashCode(scheduler)
}
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxAwait.kt b/reactive/kotlinx-coroutines-rx2/src/RxAwait.kt
index a124494..eb41658 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxAwait.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxAwait.kt
@@ -5,11 +5,8 @@
package kotlinx.coroutines.experimental.rx2
import io.reactivex.*
-import io.reactivex.disposables.Disposable
-import kotlinx.coroutines.experimental.CancellableContinuation
-import kotlinx.coroutines.experimental.CancellationException
-import kotlinx.coroutines.experimental.Job
-import kotlinx.coroutines.experimental.suspendCancellableCoroutine
+import io.reactivex.disposables.*
+import kotlinx.coroutines.experimental.*
// ------------------------ CompletableSource ------------------------
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxCancellable.kt b/reactive/kotlinx-coroutines-rx2/src/RxCancellable.kt
new file mode 100644
index 0000000..be353bf
--- /dev/null
+++ b/reactive/kotlinx-coroutines-rx2/src/RxCancellable.kt
@@ -0,0 +1,14 @@
+/*
+ * Copyright 2016-2018 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.experimental.rx2
+
+import io.reactivex.functions.*
+import kotlinx.coroutines.experimental.*
+
+internal class RxCancellable(private val job: Job) : Cancellable {
+ override fun cancel() {
+ job.cancel()
+ }
+}
\ No newline at end of file
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxChannel.kt b/reactive/kotlinx-coroutines-rx2/src/RxChannel.kt
index 28bebc2..00e3a02 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxChannel.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxChannel.kt
@@ -6,12 +6,17 @@
import io.reactivex.*
import io.reactivex.disposables.*
+import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.*
/**
* Subscribes to this [MaybeSource] and returns a channel to receive elements emitted by it.
* The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this source.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
@Suppress("CONFLICTING_OVERLOADS")
public fun <T> MaybeSource<T>.openSubscription(): ReceiveChannel<T> {
val channel = SubscriptionChannel<T>()
@@ -27,7 +32,11 @@
/**
* Subscribes to this [ObservableSource] and returns a channel to receive elements emitted by it.
* The resulting channel shall be [cancelled][ReceiveChannel.cancel] to unsubscribe from this source.
+ *
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
*/
+@ObsoleteCoroutinesApi
@Suppress("CONFLICTING_OVERLOADS")
public fun <T> ObservableSource<T>.openSubscription(): ReceiveChannel<T> {
val channel = SubscriptionChannel<T>()
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxCompletable.kt b/reactive/kotlinx-coroutines-rx2/src/RxCompletable.kt
index 50dcb9f..159bc3c 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxCompletable.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxCompletable.kt
@@ -33,7 +33,7 @@
): Completable = Completable.create { subscriber ->
val newContext = newCoroutineContext(context)
val coroutine = RxCompletableCoroutine(newContext, subscriber)
- subscriber.setCancellable(coroutine)
+ subscriber.setCancellable(RxCancellable(coroutine))
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
}
@@ -55,7 +55,7 @@
private class RxCompletableCoroutine(
parentContext: CoroutineContext,
private val subscriber: CompletableEmitter
-) : AbstractCoroutine<Unit>(parentContext, true), Cancellable {
+) : AbstractCoroutine<Unit>(parentContext, true) {
override fun onCompleted(value: Unit) {
if (!subscriber.isDisposed) subscriber.onComplete()
}
@@ -63,7 +63,4 @@
override fun onCompletedExceptionally(exception: Throwable) {
if (!subscriber.isDisposed) subscriber.onError(exception)
}
-
- // Cancellable impl
- override fun cancel() { cancel(cause = null) }
}
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxConvert.kt b/reactive/kotlinx-coroutines-rx2/src/RxConvert.kt
index d0bade4..72de46c 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxConvert.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxConvert.kt
@@ -16,8 +16,12 @@
* Every subscriber gets the signal at the same time.
* Unsubscribing from the resulting completable **does not** affect the original job in any way.
*
+ * **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
+ * in the future to account for the concept of structured concurrency.
+ *
* @param context -- the coroutine context from which the resulting completable is going to be signalled
*/
+@ExperimentalCoroutinesApi
public fun Job.asCompletable(context: CoroutineContext): Completable = GlobalScope.rxCompletable(context) {
this@asCompletable.join()
}
@@ -29,8 +33,12 @@
* Every subscriber gets the same completion value.
* Unsubscribing from the resulting maybe **does not** affect the original deferred value in any way.
*
+ * **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
+ * in the future to account for the concept of structured concurrency.
+ *
* @param context -- the coroutine context from which the resulting maybe is going to be signalled
*/
+@ExperimentalCoroutinesApi
public fun <T> Deferred<T?>.asMaybe(context: CoroutineContext): Maybe<T> = GlobalScope.rxMaybe(context) {
this@asMaybe.await()
}
@@ -42,8 +50,12 @@
* Every subscriber gets the same completion value.
* Unsubscribing from the resulting single **does not** affect the original deferred value in any way.
*
+ * **Note: This is an experimental api.** Conversion of coroutines primitives to reactive entities may change
+ * in the future to account for the concept of structured concurrency.
+ *
* @param context -- the coroutine context from which the resulting single is going to be signalled
*/
+@ExperimentalCoroutinesApi
public fun <T> Deferred<T>.asSingle(context: CoroutineContext): Single<T> = GlobalScope.rxSingle(context) {
this@asSingle.await()
}
@@ -54,8 +66,12 @@
* Every subscriber receives values from this channel in **fan-out** fashion. If the are multiple subscribers,
* they'll receive values in round-robin way.
*
+ * **Note: This API will become obsolete in future updates with introduction of lazy asynchronous streams.**
+ * See [issue #254](https://github.com/Kotlin/kotlinx.coroutines/issues/254).
+ *
* @param context -- the coroutine context from which the resulting observable is going to be signalled
*/
+@ObsoleteCoroutinesApi
public fun <T> ReceiveChannel<T>.asObservable(context: CoroutineContext): Observable<T> = GlobalScope.rxObservable(context) {
for (t in this@asObservable)
send(t)
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxFlowable.kt b/reactive/kotlinx-coroutines-rx2/src/RxFlowable.kt
index ad3b53d..4ce4325 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxFlowable.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxFlowable.kt
@@ -29,9 +29,13 @@
* The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
* with corresponding [coroutineContext] element.
*
+ * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect
+ * to cancellation and error handling may change in the future.
+ *
* @param context context of the coroutine.
* @param block the coroutine code.
*/
+@ExperimentalCoroutinesApi
public fun <T> CoroutineScope.rxFlowable(
context: CoroutineContext = EmptyCoroutineContext,
block: suspend ProducerScope<T>.() -> Unit
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxMaybe.kt b/reactive/kotlinx-coroutines-rx2/src/RxMaybe.kt
index 44aee95..4b9496b 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxMaybe.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxMaybe.kt
@@ -34,7 +34,7 @@
): Maybe<T> = Maybe.create { subscriber ->
val newContext = newCoroutineContext(context)
val coroutine = RxMaybeCoroutine(newContext, subscriber)
- subscriber.setCancellable(coroutine)
+ subscriber.setCancellable(RxCancellable(coroutine))
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
}
@@ -56,7 +56,7 @@
private class RxMaybeCoroutine<T>(
parentContext: CoroutineContext,
private val subscriber: MaybeEmitter<T>
-) : AbstractCoroutine<T>(parentContext, true), Cancellable {
+) : AbstractCoroutine<T>(parentContext, true) {
override fun onCompleted(value: T) {
if (!subscriber.isDisposed) {
if (value == null) subscriber.onComplete() else subscriber.onSuccess(value)
@@ -66,7 +66,4 @@
override fun onCompletedExceptionally(exception: Throwable) {
if (!subscriber.isDisposed) subscriber.onError(exception)
}
-
- // Cancellable impl
- override fun cancel() { cancel(cause = null) }
}
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxObservable.kt b/reactive/kotlinx-coroutines-rx2/src/RxObservable.kt
index 1744968..23b3f69 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxObservable.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxObservable.kt
@@ -32,16 +32,20 @@
* The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
* with corresponding [coroutineContext] element.
*
+ * **Note: This is an experimental api.** Behaviour of publishers that work as children in a parent scope with respect
+ * to cancellation and error handling may change in the future.
+ *
* @param context context of the coroutine.
* @param block the coroutine code.
*/
+@ExperimentalCoroutinesApi
public fun <T> CoroutineScope.rxObservable(
context: CoroutineContext = EmptyCoroutineContext,
block: suspend ProducerScope<T>.() -> Unit
): Observable<T> = Observable.create { subscriber ->
val newContext = newCoroutineContext(context)
val coroutine = RxObservableCoroutine(newContext, subscriber)
- subscriber.setCancellable(coroutine) // do it first (before starting coroutine), to await unnecessary suspensions
+ subscriber.setCancellable(RxCancellable(coroutine)) // do it first (before starting coroutine), to await unnecessary suspensions
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
}
@@ -67,7 +71,7 @@
private class RxObservableCoroutine<T>(
parentContext: CoroutineContext,
private val subscriber: ObservableEmitter<T>
-) : AbstractCoroutine<Unit>(parentContext, true), ProducerScope<T>, Cancellable, SelectClause2<T, SendChannel<T>> {
+) : AbstractCoroutine<Unit>(parentContext, true), ProducerScope<T>, SelectClause2<T, SendChannel<T>> {
override val channel: SendChannel<T> get() = this
// Mutex is locked when while subscriber.onXXX is being invoked
@@ -167,7 +171,4 @@
if (mutex.tryLock()) // if we can acquire the lock
doLockedSignalCompleted()
}
-
- // Cancellable impl
- override fun cancel() { cancel(cause = null) }
}
\ No newline at end of file
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxScheduler.kt b/reactive/kotlinx-coroutines-rx2/src/RxScheduler.kt
index 66747db..d2ade17 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxScheduler.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxScheduler.kt
@@ -16,34 +16,41 @@
* Converts an instance of [Scheduler] to an implementation of [CoroutineDispatcher]
* and provides native [delay][Delay.delay] support.
*/
-public fun Scheduler.asCoroutineDispatcher() = SchedulerCoroutineDispatcher(this)
+public fun Scheduler.asCoroutineDispatcher(): SchedulerCoroutineDispatcher = SchedulerCoroutineDispatcher(this)
/**
* Implements [CoroutineDispatcher] on top of an arbitrary [Scheduler].
* @param scheduler a scheduler.
*/
-public class SchedulerCoroutineDispatcher(private val scheduler: Scheduler) : CoroutineDispatcher(), Delay {
+public class SchedulerCoroutineDispatcher(
+ /**
+ * Underlying scheduler of current [CoroutineDispatcher].
+ */
+ public val scheduler: Scheduler
+) : CoroutineDispatcher(), Delay {
+ /** @suppress */
override fun dispatch(context: CoroutineContext, block: Runnable) {
scheduler.scheduleDirect(block)
}
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
+ /** @suppress */
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
val disposable = scheduler.scheduleDirect({
with(continuation) { resumeUndispatched(Unit) }
- }, time, unit)
+ }, timeMillis, TimeUnit.MILLISECONDS)
continuation.disposeOnCancellation(disposable)
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
- val disposable = scheduler.scheduleDirect(block, time, unit)
- return object : DisposableHandle {
- override fun dispose() {
- disposable.dispose()
- }
- }
+ /** @suppress */
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
+ val disposable = scheduler.scheduleDirect(block, timeMillis, TimeUnit.MILLISECONDS)
+ return DisposableHandle { disposable.dispose() }
}
+ /** @suppress */
override fun toString(): String = scheduler.toString()
+ /** @suppress */
override fun equals(other: Any?): Boolean = other is SchedulerCoroutineDispatcher && other.scheduler === scheduler
+ /** @suppress */
override fun hashCode(): Int = System.identityHashCode(scheduler)
}
diff --git a/reactive/kotlinx-coroutines-rx2/src/RxSingle.kt b/reactive/kotlinx-coroutines-rx2/src/RxSingle.kt
index 00bb9fe..7ffe1cd 100644
--- a/reactive/kotlinx-coroutines-rx2/src/RxSingle.kt
+++ b/reactive/kotlinx-coroutines-rx2/src/RxSingle.kt
@@ -33,7 +33,7 @@
): Single<T> = Single.create { subscriber ->
val newContext = newCoroutineContext(context)
val coroutine = RxSingleCoroutine(newContext, subscriber)
- subscriber.setCancellable(coroutine)
+ subscriber.setCancellable(RxCancellable(coroutine))
coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
}
@@ -55,7 +55,7 @@
private class RxSingleCoroutine<T>(
parentContext: CoroutineContext,
private val subscriber: SingleEmitter<T>
-) : AbstractCoroutine<T>(parentContext, true), Cancellable {
+) : AbstractCoroutine<T>(parentContext, true) {
override fun onCompleted(value: T) {
if (!subscriber.isDisposed) subscriber.onSuccess(value)
}
@@ -63,7 +63,4 @@
override fun onCompletedExceptionally(exception: Throwable) {
if (!subscriber.isDisposed) subscriber.onError(exception)
}
-
- // Cancellable impl
- override fun cancel() { cancel(cause = null) }
}
diff --git a/ui/coroutines-guide-ui.md b/ui/coroutines-guide-ui.md
index be1e214..5eae387 100644
--- a/ui/coroutines-guide-ui.md
+++ b/ui/coroutines-guide-ui.md
@@ -339,7 +339,7 @@
Try clicking repeatedly on a circle in this version of the code. The clicks are just ignored while the countdown
animation is running. This happens because the actor is busy with an animation and does not receive from its channel.
-By default, an actor's mailbox is backed by [RendezvousChannel], whose `offer` operation succeeds only when
+By default, an actor's mailbox is backed by `RendezvousChannel`, whose `offer` operation succeeds only when
the `receive` is active.
> On Android, there is `View` sent in OnClickListener, so we send the `View` to the actor as a signal.
@@ -368,7 +368,7 @@
controls the implementation of the channel that this actor is using for its mailbox. The description of all
the available choices is given in documentation of the [`Channel()`][Channel] factory function.
-Let us change the code to use [ConflatedChannel] by passing [Channel.CONFLATED] capacity value. The
+Let us change the code to use `ConflatedChannel` by passing [Channel.CONFLATED] capacity value. The
change is only to the line that creates an actor:
```kotlin
@@ -393,10 +393,10 @@
This is also a desired behaviour for UI applications that have to react to incoming high-frequency
event streams by updating their UI based on the most recently received update. A coroutine that is using
-[ConflatedChannel] avoids delays that are usually introduced by buffering of events.
+`ConflatedChannel` avoids delays that are usually introduced by buffering of events.
You can experiment with `capacity` parameter in the above line to see how it affects the behaviour of the code.
-Setting `capacity = Channel.UNLIMITED` creates a coroutine with [LinkedListChannel] mailbox that buffers all
+Setting `capacity = Channel.UNLIMITED` creates a coroutine with `LinkedListChannel` mailbox that buffers all
events. In this case, the animation runs as many times as the circle is clicked.
## Blocking operations
@@ -715,11 +715,8 @@
[actor]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/actor.html
[SendChannel.offer]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/offer.html
[SendChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/index.html
-[RendezvousChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-rendezvous-channel/index.html
[Channel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-channel/index.html
-[ConflatedChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-conflated-channel/index.html
[Channel.CONFLATED]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-channel/-c-o-n-f-l-a-t-e-d.html
-[LinkedListChannel]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-linked-list-channel/index.html
<!--- MODULE kotlinx-coroutines-javafx -->
<!--- INDEX kotlinx.coroutines.experimental.javafx -->
[kotlinx.coroutines.experimental.Dispatchers.JavaFx]: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-javafx/kotlinx.coroutines.experimental.javafx/kotlinx.coroutines.experimental.-dispatchers/-java-fx.html
diff --git a/ui/kotlinx-coroutines-android/src/AndroidExceptionPreHandler.kt b/ui/kotlinx-coroutines-android/src/AndroidExceptionPreHandler.kt
index e7129f4..3776d1c 100644
--- a/ui/kotlinx-coroutines-android/src/AndroidExceptionPreHandler.kt
+++ b/ui/kotlinx-coroutines-android/src/AndroidExceptionPreHandler.kt
@@ -25,7 +25,8 @@
* @suppress This is an internal impl class.
*/
@Keep
-class AndroidExceptionPreHandler :
+@InternalCoroutinesApi
+internal class AndroidExceptionPreHandler :
AbstractCoroutineContextElement(CoroutineExceptionHandler), CoroutineExceptionHandler
{
override fun handleException(context: CoroutineContext, exception: Throwable) {
diff --git a/ui/kotlinx-coroutines-android/src/HandlerDispatcher.kt b/ui/kotlinx-coroutines-android/src/HandlerDispatcher.kt
index dcd3e15..88f14c4 100644
--- a/ui/kotlinx-coroutines-android/src/HandlerDispatcher.kt
+++ b/ui/kotlinx-coroutines-android/src/HandlerDispatcher.kt
@@ -11,14 +11,13 @@
import android.view.*
import kotlinx.coroutines.experimental.*
import java.lang.reflect.Constructor
-import java.util.concurrent.*
import kotlin.coroutines.experimental.*
/**
* Dispatches execution onto Android main thread and provides native [delay][Delay.delay] support.
*/
public val Dispatchers.Main: HandlerDispatcher
- get() = mainDispatcher
+ get() = kotlinx.coroutines.experimental.android.Main
/**
* Dispatches execution onto Android [Handler].
@@ -30,19 +29,23 @@
* Returns dispatcher that executes coroutines immediately when it is already in the right handler context
* (current looper is the same as this handler's looper). See [isDispatchNeeded] documentation on
* why this should not be done by default.
+ *
+ * **Note: This is an experimental api.** Semantics of this dispatcher may change in the future.
*/
+ @ExperimentalCoroutinesApi
public abstract val immediate: HandlerDispatcher
}
/**
* Represents an arbitrary [Handler] as a implementation of [CoroutineDispatcher].
*/
+@JvmName("from") // this is for a nice Java API, see issue #255
public fun Handler.asCoroutineDispatcher(): HandlerDispatcher =
HandlerContext(this)
private const val MAX_DELAY = Long.MAX_VALUE / 2 // cannot delay for too long on Android
-private val mainHandler = Looper.getMainLooper().asHandler(async = true)
+internal val mainHandler = Looper.getMainLooper().asHandler(async = true)
@VisibleForTesting
internal fun Looper.asHandler(async: Boolean): Handler {
@@ -68,7 +71,8 @@
return constructor.newInstance(this, null, true)
}
-private val mainDispatcher = HandlerContext(mainHandler, "Main")
+@JvmField // this is for a nice Java API, see issue #255
+internal val Main: HandlerDispatcher = HandlerContext(mainHandler, "Main")
/**
* Implements [CoroutineDispatcher] on top of an arbitrary Android [Handler].
@@ -109,14 +113,14 @@
handler.post(block)
}
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
handler.postDelayed({
with(continuation) { resumeUndispatched(Unit) }
- }, unit.toMillis(time).coerceAtMost(MAX_DELAY))
+ }, timeMillis.coerceAtMost(MAX_DELAY))
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
- handler.postDelayed(block, unit.toMillis(time).coerceAtMost(MAX_DELAY))
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
+ handler.postDelayed(block, timeMillis.coerceAtMost(MAX_DELAY))
return object : DisposableHandle {
override fun dispose() {
handler.removeCallbacks(block)
@@ -171,6 +175,6 @@
private fun postFrameCallback(choreographer: Choreographer, cont: CancellableContinuation<Long>) {
choreographer.postFrameCallback { nanos ->
- with(cont) { mainDispatcher.resumeUndispatched(nanos) }
+ with(cont) { Main.resumeUndispatched(nanos) }
}
}
diff --git a/ui/kotlinx-coroutines-javafx/src/JavaFxDispatcher.kt b/ui/kotlinx-coroutines-javafx/src/JavaFxDispatcher.kt
index cbeba40..4d40345 100644
--- a/ui/kotlinx-coroutines-javafx/src/JavaFxDispatcher.kt
+++ b/ui/kotlinx-coroutines-javafx/src/JavaFxDispatcher.kt
@@ -9,7 +9,6 @@
import javafx.event.*
import javafx.util.*
import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.javafx.JavaFx.delay
import java.util.concurrent.*
import kotlin.coroutines.experimental.*
@@ -24,8 +23,7 @@
*
* This class provides type-safety and a point for future extensions.
*/
-public sealed class JavaFxDispatcher : CoroutineDispatcher(), Delay {
-}
+public sealed class JavaFxDispatcher : CoroutineDispatcher(), Delay
/**
* Dispatches execution onto JavaFx application thread and provides native [delay] support.
@@ -37,7 +35,7 @@
imports = ["kotlinx.coroutines.experimental.Dispatchers", "kotlinx.coroutines.experimental.javafx.JavaFx"])
)
// todo: it will become an internal implementation object
-object JavaFx : JavaFxDispatcher(), Delay {
+object JavaFx : JavaFxDispatcher() {
init {
// :kludge: to make sure Toolkit is initialized if we use JavaFx dispatcher outside of JavaFx app
initPlatform()
@@ -59,15 +57,15 @@
suspend fun awaitPulse(): Long =
kotlinx.coroutines.experimental.javafx.awaitPulse()
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
- val timeline = schedule(time, unit, EventHandler {
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
+ val timeline = schedule(timeMillis, TimeUnit.MILLISECONDS, EventHandler {
with(continuation) { resumeUndispatched(Unit) }
})
continuation.invokeOnCancellation { timeline.stop() }
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
- val timeline = schedule(time, unit, EventHandler {
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
+ val timeline = schedule(timeMillis, TimeUnit.MILLISECONDS, EventHandler {
block.run()
})
return object : DisposableHandle {
@@ -90,7 +88,7 @@
/**
* Suspends coroutine until next JavaFx pulse and returns time of the pulse on resumption.
* If the [Job] of the current coroutine is completed while this suspending function is waiting, this function
- * immediately resumes with [CancellationException] .
+ * immediately resumes with [CancellationException][kotlinx.coroutines.experimental.CancellationException].
*/
public suspend fun awaitPulse(): Long = suspendCancellableCoroutine { cont ->
pulseTimer.onNext(cont)
diff --git a/ui/kotlinx-coroutines-swing/src/SwingDispatcher.kt b/ui/kotlinx-coroutines-swing/src/SwingDispatcher.kt
index 8ee7761..a78e2ed 100644
--- a/ui/kotlinx-coroutines-swing/src/SwingDispatcher.kt
+++ b/ui/kotlinx-coroutines-swing/src/SwingDispatcher.kt
@@ -5,7 +5,6 @@
package kotlinx.coroutines.experimental.swing
import kotlinx.coroutines.experimental.*
-import kotlinx.coroutines.experimental.swing.Swing.delay
import java.awt.event.*
import java.util.concurrent.*
import javax.swing.*
@@ -34,18 +33,18 @@
imports = ["kotlinx.coroutines.experimental.Dispatchers", "kotlinx.coroutines.experimental.swing.Swing"])
)
// todo: it will become an internal implementation object
-object Swing : SwingDispatcher(), Delay {
+object Swing : SwingDispatcher() {
override fun dispatch(context: CoroutineContext, block: Runnable) = SwingUtilities.invokeLater(block)
- override fun scheduleResumeAfterDelay(time: Long, unit: TimeUnit, continuation: CancellableContinuation<Unit>) {
- val timer = schedule(time, unit, ActionListener {
+ override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
+ val timer = schedule(timeMillis, TimeUnit.MILLISECONDS, ActionListener {
with(continuation) { resumeUndispatched(Unit) }
})
continuation.invokeOnCancellation { timer.stop() }
}
- override fun invokeOnTimeout(time: Long, unit: TimeUnit, block: Runnable): DisposableHandle {
- val timer = schedule(time, unit, ActionListener {
+ override fun invokeOnTimeout(timeMillis: Long, block: Runnable): DisposableHandle {
+ val timer = schedule(timeMillis, TimeUnit.MILLISECONDS, ActionListener {
block.run()
})
return object : DisposableHandle {