Have FlowConcurrentMap backed by MutableStateFlow

build.gradle

@@ -9,6 +9,7 @@ buildscript {
 plugins {
     id 'org.jetbrains.kotlin.jvm' version '1.3.71' apply false
     id 'org.jmailen.kotlinter' version '2.2.0' apply false
+    id 'binary-compatibility-validator' version '0.2.3'
 }
 
 subprojects {

collections/api/collections.api

@@ -0,0 +1,30 @@
+public final class com/juul/tuulbox/collections/FlowConcurrentMap : java/util/concurrent/ConcurrentMap {
+	public fun <init> ()V
+	public fun <init> (Ljava/util/concurrent/ConcurrentMap;)V
+	public fun clear ()V
+	public fun containsKey (Ljava/lang/Object;)Z
+	public fun containsValue (Ljava/lang/Object;)Z
+	public final fun entrySet ()Ljava/util/Set;
+	public fun get (Ljava/lang/Object;)Ljava/lang/Object;
+	public fun getEntries ()Ljava/util/Set;
+	public fun getKeys ()Ljava/util/Set;
+	public final fun getOnChanged ()Lkotlinx/coroutines/flow/Flow;
+	public fun getSize ()I
+	public fun getValues ()Ljava/util/Collection;
+	public fun isEmpty ()Z
+	public final fun keySet ()Ljava/util/Set;
+	public fun put (Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
+	public fun putAll (Ljava/util/Map;)V
+	public fun putIfAbsent (Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
+	public fun remove (Ljava/lang/Object;)Ljava/lang/Object;
+	public fun remove (Ljava/lang/Object;Ljava/lang/Object;)Z
+	public fun replace (Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;
+	public fun replace (Ljava/lang/Object;Ljava/lang/Object;Ljava/lang/Object;)Z
+	public final fun size ()I
+	public final fun values ()Ljava/util/Collection;
+}
+
+public final class com/juul/tuulbox/collections/FlowConcurrentMapKt {
+	public static final fun withFlow (Ljava/util/concurrent/ConcurrentMap;)Lcom/juul/tuulbox/collections/FlowConcurrentMap;
+}
+

collections/src/main/kotlin/FlowConcurrentMap.kt

@@ -2,10 +2,9 @@ package com.juul.tuulbox.collections
 
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.ConcurrentMap
-import kotlinx.coroutines.channels.BroadcastChannel
-import kotlinx.coroutines.channels.Channel.Factory.CONFLATED
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.filterNotNull
 
 fun <K, V> ConcurrentMap<K, V>.withFlow() = FlowConcurrentMap(this)
 
@@ -19,8 +18,8 @@ class FlowConcurrentMap<K, V>(
 
     constructor() : this(ConcurrentHashMap())
 
-    private val _onChanged = BroadcastChannel<Map<K, V>>(CONFLATED)
-    val onChanged = _onChanged.asFlow()
+    private val _onChanged = MutableStateFlow<Map<K, V>?>(null)
+    val onChanged: Flow<Map<K, V>> = _onChanged.filterNotNull()
 
     /** @see ConcurrentMap.put */
     override fun put(
@@ -42,7 +41,7 @@ class FlowConcurrentMap<K, V>(
         key: K,
         value: V
     ): Boolean = map.remove(key, value).also { didRemove ->
-        if (didRemove) _onChanged.offer(map.toMap())
+        if (didRemove) _onChanged.value = map.toMap()
     }
 
     /** @see ConcurrentMap.clear */
@@ -62,7 +61,7 @@ class FlowConcurrentMap<K, V>(
         key: K,
         value: V
     ): V? = map.putIfAbsent(key, value).also { previousValue ->
-        if (previousValue == null) _onChanged.offer(map.toMap())
+        if (previousValue == null) _onChanged.value = map.toMap()
     }
 
     /**
@@ -75,7 +74,7 @@ class FlowConcurrentMap<K, V>(
         oldValue: V,
         newValue: V
     ): Boolean = map.replace(key, oldValue, newValue).also { didReplace ->
-        if (didReplace) _onChanged.offer(map.toMap())
+        if (didReplace) _onChanged.value = map.toMap()
     }
 
     /**
@@ -87,7 +86,7 @@ class FlowConcurrentMap<K, V>(
         key: K,
         value: V
     ): V? = map.replace(key, value).also { previousValue ->
-        if (previousValue != null) _onChanged.offer(map.toMap())
+        if (previousValue != null) _onChanged.value = map.toMap()
     }
 
     /** @throws UnsupportedOperationException */
@@ -106,7 +105,7 @@ class FlowConcurrentMap<K, V>(
         action: ConcurrentMap<K, V>.() -> T
     ): T {
         val result = action.invoke(this)
-        _onChanged.offer(toMap())
+        _onChanged.value = toMap()
         return result
     }
 }

gradle/dependencies.gradle

@@ -1,7 +1,7 @@
 ext.deps = [
     kotlin: [
         stdlib: "org.jetbrains.kotlin:kotlin-stdlib",
-        coroutines: "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.5",
+        coroutines: "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.7",
         junit: "org.jetbrains.kotlin:kotlin-test-junit",
     ],
 ]

settings.gradle

@@ -1 +1,17 @@
+pluginManagement {
+    repositories {
+        gradlePluginPortal()
+    }
+
+    resolutionStrategy {
+        eachPlugin {
+            // Support using `binary-compatibility-validator` via Gradle's plugin lambda.
+            // https://medium.com/@StefMa/its-time-to-ditch-the-buildscript-block-a1ab12e0d9ce
+            if (requested.id.id == "binary-compatibility-validator") {
+                useModule("org.jetbrains.kotlinx:binary-compatibility-validator:${requested.version}")
+            }
+        }
+    }
+}
+
 include ':collections'