StateFlow implementation

StateFlow is a Flow analogue to ConflatedBroadcastChannel. Since Flow API is simpler than channels APIs, the implementation of StateFlow is simpler. It consumes and allocates less memory, while still providing full deadlock-freedom (even though it is not lock-free internally). Fixes #1973
Kotlin · Apr 29, 2020 · 55db768 · 55db768
1 parent d5766f3
commit 55db768
Show file tree

Hide file tree

Showing 8 changed files with 871 additions and 14 deletions.
diff --git a/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api b/kotlinx-coroutines-core/api/kotlinx-coroutines-core.api
@@ -974,6 +974,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun skip (Lkotlinx/coroutines/flow/Flow;I)Lkotlinx/coroutines/flow/Flow;
 	public static final fun startWith (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun startWith (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun stateIn (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/CoroutineScope;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun subscribe (Lkotlinx/coroutines/flow/Flow;)V
 	public static final fun subscribe (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)V
 	public static final fun subscribe (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/jvm/functions/Function2;)V
@@ -993,7 +994,38 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun zip (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 }
 
-public abstract class kotlinx/coroutines/flow/internal/ChannelFlow : kotlinx/coroutines/flow/Flow {
+public abstract interface class kotlinx/coroutines/flow/MutableStateFlow : kotlinx/coroutines/flow/StateFlow {
+	public abstract fun close (Ljava/lang/Throwable;)Z
+	public abstract fun getValue ()Ljava/lang/Object;
+	public abstract fun setValue (Ljava/lang/Object;)V
+}
+
+public final class kotlinx/coroutines/flow/MutableStateFlow$DefaultImpls {
+	public static synthetic fun close$default (Lkotlinx/coroutines/flow/MutableStateFlow;Ljava/lang/Throwable;ILjava/lang/Object;)Z
+}
+
+public abstract interface class kotlinx/coroutines/flow/StateFlow : kotlinx/coroutines/flow/Flow {
+	public abstract fun getValue ()Ljava/lang/Object;
+	public abstract fun isClosed ()Z
+}
+
+public abstract interface class kotlinx/coroutines/flow/StateFlowJob : kotlinx/coroutines/Job, kotlinx/coroutines/flow/StateFlow {
+}
+
+public final class kotlinx/coroutines/flow/StateFlowJob$DefaultImpls {
+	public static synthetic fun cancel (Lkotlinx/coroutines/flow/StateFlowJob;)V
+	public static fun fold (Lkotlinx/coroutines/flow/StateFlowJob;Ljava/lang/Object;Lkotlin/jvm/functions/Function2;)Ljava/lang/Object;
+	public static fun get (Lkotlinx/coroutines/flow/StateFlowJob;Lkotlin/coroutines/CoroutineContext$Key;)Lkotlin/coroutines/CoroutineContext$Element;
+	public static fun minusKey (Lkotlinx/coroutines/flow/StateFlowJob;Lkotlin/coroutines/CoroutineContext$Key;)Lkotlin/coroutines/CoroutineContext;
+	public static fun plus (Lkotlinx/coroutines/flow/StateFlowJob;Lkotlin/coroutines/CoroutineContext;)Lkotlin/coroutines/CoroutineContext;
+	public static fun plus (Lkotlinx/coroutines/flow/StateFlowJob;Lkotlinx/coroutines/Job;)Lkotlinx/coroutines/Job;
+}
+
+public final class kotlinx/coroutines/flow/StateFlowKt {
+	public static final fun StateFlow (Ljava/lang/Object;)Lkotlinx/coroutines/flow/MutableStateFlow;
+}
+
+public abstract class kotlinx/coroutines/flow/internal/ChannelFlow : kotlinx/coroutines/flow/internal/FusibleFlow {
 	public final field capacity I
 	public final field context Lkotlin/coroutines/CoroutineContext;
 	public fun <init> (Lkotlin/coroutines/CoroutineContext;I)V
@@ -1002,10 +1034,9 @@ public abstract class kotlinx/coroutines/flow/internal/ChannelFlow : kotlinx/cor
 	public fun collect (Lkotlinx/coroutines/flow/FlowCollector;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	protected abstract fun collectTo (Lkotlinx/coroutines/channels/ProducerScope;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	protected abstract fun create (Lkotlin/coroutines/CoroutineContext;I)Lkotlinx/coroutines/flow/internal/ChannelFlow;
+	public fun fuse (Lkotlin/coroutines/CoroutineContext;I)Lkotlinx/coroutines/flow/internal/FusibleFlow;
 	public fun produceImpl (Lkotlinx/coroutines/CoroutineScope;)Lkotlinx/coroutines/channels/ReceiveChannel;
 	public fun toString ()Ljava/lang/String;
-	public final fun update (Lkotlin/coroutines/CoroutineContext;I)Lkotlinx/coroutines/flow/internal/ChannelFlow;
-	public static synthetic fun update$default (Lkotlinx/coroutines/flow/internal/ChannelFlow;Lkotlin/coroutines/CoroutineContext;IILjava/lang/Object;)Lkotlinx/coroutines/flow/internal/ChannelFlow;
 }
 
 public final class kotlinx/coroutines/flow/internal/CombineKt {
@@ -1016,6 +1047,14 @@ public final class kotlinx/coroutines/flow/internal/FlowExceptions_commonKt {
 	public static final fun checkIndexOverflow (I)I
 }
 
+public abstract interface class kotlinx/coroutines/flow/internal/FusibleFlow : kotlinx/coroutines/flow/Flow {
+	public abstract fun fuse (Lkotlin/coroutines/CoroutineContext;I)Lkotlinx/coroutines/flow/internal/FusibleFlow;
+}
+
+public final class kotlinx/coroutines/flow/internal/FusibleFlow$DefaultImpls {
+	public static synthetic fun fuse$default (Lkotlinx/coroutines/flow/internal/FusibleFlow;Lkotlin/coroutines/CoroutineContext;IILjava/lang/Object;)Lkotlinx/coroutines/flow/internal/FusibleFlow;
+}
+
 public final class kotlinx/coroutines/flow/internal/SafeCollector_commonKt {
 	public static final fun unsafeFlow (Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 }