small refactor in SpecialMutableStateFlow

0.20.17

CHANGELOG.md

@@ -1,9 +1,25 @@
 # Changelog
 
+## 0.20.18
+
+* `Coroutines`:
+    * `SpecialMutableStateFlow` now extends `MutableStateFlow`
+    * `Compose`:
+        * Deprecate `FlowState` due to its complexity in fixes
+
+## 0.20.17
+
+* `Versions`:
+    * `Serialization`: `1.6.1` -> `1.6.2`
+
 ## 0.20.16
 
+* `Versions`:
+    * `Exposed`: `0.44.1` -> `0.45.0`
 * `Coroutines`:
-    * Add `SpecialStateFlow`
+    * Add `SpecialMutableStateFlow`
+    * `Compose`:
+        * Add `FlowState`
 
 ## 0.20.15
 

coroutines/compose/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/compose/FlowState.kt

@@ -1,43 +1,46 @@
 package dev.inmo.micro_utils.coroutines.compose
 
 import androidx.compose.runtime.MutableState
-import androidx.compose.runtime.State
-import androidx.compose.runtime.derivedStateOf
 import dev.inmo.micro_utils.coroutines.SpecialMutableStateFlow
-import dev.inmo.micro_utils.coroutines.doInUI
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 
+/**
+ * This type works like [MutableState], [kotlinx.coroutines.flow.StateFlow] and [kotlinx.coroutines.flow.MutableSharedFlow].
+ * Based on [SpecialMutableStateFlow]
+ */
+@Deprecated("Will be removed soon")
 class FlowState<T>(
-    private val state: MutableState<T>,
-    internalScope: CoroutineScope = CoroutineScope(Dispatchers.Main)
+    initial: T,
+    internalScope: CoroutineScope = CoroutineScope(Dispatchers.Default)
 ) : MutableState<T>,
-    SpecialMutableStateFlow.Default<T>(state.value, internalScope) {
+    SpecialMutableStateFlow<T>(initial, internalScope) {
+    private var internalValue: T = initial
     override var value: T
-        get() = super.value
+        get() = internalValue
         set(value) {
+            internalValue = value
             tryEmit(value)
         }
 
-    override fun component1(): T = state.component1()
-
-    override fun component2(): (T) -> Unit = state.component2()
-
-    private val stateState = derivedStateOf {
-        if (value != state.value) {
-            value = state.value
-        }
+    override fun onChangeWithoutSync(value: T) {
+        internalValue = value
+        super.onChangeWithoutSync(value)
     }
 
-    override suspend fun onChange(value: T) {
-        super.onChange(value)
+    override fun component1(): T = value
 
-        if (state.value != value) {
-            doInUI {
-                state.value = value
-            }
-        }
+    override fun component2(): (T) -> Unit = { tryEmit(it) }
+
+    override fun tryEmit(value: T): Boolean {
+        internalValue = value
+        return super.tryEmit(value)
+    }
+
+    override suspend fun emit(value: T) {
+        internalValue = value
+        super.emit(value)
     }
 }
 
-fun <T> MutableState<T>.asFlowState(scope: CoroutineScope = CoroutineScope(Dispatchers.Main)) = FlowState(this, scope)
+//fun <T> MutableState<T>.asFlowState(scope: CoroutineScope = CoroutineScope(Dispatchers.Main)) = FlowState(this, scope)

coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SpecialMutableStateFlow.kt

@@ -0,0 +1,86 @@
+package dev.inmo.micro_utils.coroutines
+
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.InternalCoroutinesApi
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.FlowCollector
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.internal.SynchronizedObject
+import kotlinx.coroutines.internal.synchronized
+
+/**
+ * Works like [StateFlow], but guarantee that latest value update will always be delivered to
+ * each active subscriber
+ */
+open class SpecialMutableStateFlow<T>(
+    initialValue: T,
+    internalScope: CoroutineScope = CoroutineScope(Dispatchers.Default)
+) : MutableStateFlow<T>, FlowCollector<T>, MutableSharedFlow<T> {
+    @OptIn(InternalCoroutinesApi::class)
+    private val syncObject = SynchronizedObject()
+    protected val internalSharedFlow: MutableSharedFlow<T> = MutableSharedFlow(
+        replay = 0,
+        extraBufferCapacity = 2,
+        onBufferOverflow = BufferOverflow.DROP_OLDEST
+    )
+    protected val publicSharedFlow: MutableSharedFlow<T> = MutableSharedFlow(
+        replay = 1,
+        extraBufferCapacity = 1,
+        onBufferOverflow = BufferOverflow.DROP_OLDEST
+    )
+
+    protected var _value: T = initialValue
+    override var value: T
+        get() = _value
+        set(value) {
+            doOnChangeAction(value)
+        }
+    protected val job = internalSharedFlow.subscribe(internalScope) {
+        doOnChangeAction(it)
+    }
+
+    override val replayCache: List<T>
+        get() = publicSharedFlow.replayCache
+    override val subscriptionCount: StateFlow<Int>
+        get() = publicSharedFlow.subscriptionCount
+
+    @OptIn(InternalCoroutinesApi::class)
+    override fun compareAndSet(expect: T, update: T): Boolean {
+        return synchronized(syncObject) {
+            if (expect == _value && update != _value) {
+                doOnChangeAction(update)
+            }
+            expect == _value
+        }
+    }
+
+    protected open fun onChangeWithoutSync(value: T) {
+        _value = value
+        publicSharedFlow.tryEmit(value)
+    }
+    @OptIn(InternalCoroutinesApi::class)
+    protected open fun doOnChangeAction(value: T) {
+        synchronized(syncObject) {
+            if (_value != value) {
+                onChangeWithoutSync(value)
+            }
+        }
+    }
+
+    @ExperimentalCoroutinesApi
+    override fun resetReplayCache() = publicSharedFlow.resetReplayCache()
+
+    override fun tryEmit(value: T): Boolean {
+        return internalSharedFlow.tryEmit(value)
+    }
+
+    override suspend fun emit(value: T) {
+        internalSharedFlow.emit(value)
+    }
+
+    override suspend fun collect(collector: FlowCollector<T>) = publicSharedFlow.collect(collector)
+}

coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/SpecialStateFlow.kt

@@ -1,59 +0,0 @@
-package dev.inmo.micro_utils.coroutines
-
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.flow.FlowCollector
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.StateFlow
-
-interface SpecialMutableStateFlow<T> : StateFlow<T>, FlowCollector<T>, MutableSharedFlow<T> {
-    open class Default<T>(
-        initialValue: T,
-        internalScope: CoroutineScope = CoroutineScope(Dispatchers.Default)
-    ) : SpecialMutableStateFlow<T> {
-        protected val internalSharedFlow: MutableSharedFlow<T> = MutableSharedFlow(
-            replay = 0,
-            extraBufferCapacity = 1,
-            onBufferOverflow = BufferOverflow.DROP_OLDEST
-        )
-        protected val publicSharedFlow: MutableSharedFlow<T> = MutableSharedFlow(
-            replay = 1,
-            extraBufferCapacity = 1,
-            onBufferOverflow = BufferOverflow.DROP_OLDEST
-        )
-
-        protected var _value: T = initialValue
-        override val value: T
-            get() = _value
-        protected open suspend fun onChange(value: T) {
-            _value = value
-            publicSharedFlow.emit(value)
-        }
-        protected val job = internalSharedFlow.subscribeSafelyWithoutExceptions(internalScope) {
-            if (_value != it) {
-                onChange(it)
-            }
-        }
-
-        override val replayCache: List<T>
-            get() = publicSharedFlow.replayCache
-        override val subscriptionCount: StateFlow<Int>
-            get() = publicSharedFlow.subscriptionCount
-
-        @ExperimentalCoroutinesApi
-        override fun resetReplayCache() = publicSharedFlow.resetReplayCache()
-
-        override fun tryEmit(value: T): Boolean {
-            return internalSharedFlow.tryEmit(value)
-        }
-
-        override suspend fun emit(value: T) {
-            internalSharedFlow.emit(value)
-        }
-
-        override suspend fun collect(collector: FlowCollector<T>) = publicSharedFlow.collect(collector)
-    }
-}

gradle.properties

@@ -15,5 +15,5 @@ crypto_js_version=4.1.1
 # Project data
 
 group=dev.inmo
-version=0.20.16
-android_code_version=222
+version=0.20.18
+android_code_version=224

gradle/libs.versions.toml

@@ -1,7 +1,7 @@
 [versions]
 
 kt = "1.9.21"
-kt-serialization = "1.6.1"
+kt-serialization = "1.6.2"
 kt-coroutines = "1.7.3"
 
 kslog = "1.3.1"
@@ -22,11 +22,11 @@ koin = "3.5.0"
 okio = "3.6.0"
 
 ksp = "1.9.21-1.0.15"
-kotlin-poet = "1.15.1"
+kotlin-poet = "1.15.2"
 
 versions = "0.50.0"
 
-android-gradle = "8.1.4"
+android-gradle = "8.2.0"
 dexcount = "4.0.0"
 
 android-coreKtx = "1.12.0"

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.4-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.5-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists