BroadcastChannel -> MutableSharedFlow in repos

CHANGELOG.md

@@ -5,6 +5,8 @@
 * `Versions`:
     * `Ktor`: `1.4.2` -> `1.4.3`
 * `Repo`
+    * `Common`
+        * In repos different usages of `BroadcastChannel`s has been replaced with `MutableSharedFlow`
     * `Exposed`
         * `asObject` open fun has been added in CRUD realization
 

repos/common/src/main/kotlin/dev/inmo/micro_utils/repos/crud/AbstractMutableAndroidCRUDRepo.kt

@@ -3,20 +3,18 @@ package dev.inmo.micro_utils.repos.crud
 import android.content.ContentValues
 import dev.inmo.micro_utils.common.mapNotNullA
 import dev.inmo.micro_utils.repos.*
-import kotlinx.coroutines.channels.BroadcastChannel
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.*
 
 abstract class AbstractMutableAndroidCRUDRepo<ObjectType, IdType, InputValueType>(
     helper: StandardSQLHelper
 ) : WriteStandardCRUDRepo<ObjectType, IdType, InputValueType>,
     AbstractAndroidCRUDRepo<ObjectType, IdType>(helper) {
-    protected val newObjectsChannel = BroadcastChannel<ObjectType>(64)
-    protected val updateObjectsChannel = BroadcastChannel<ObjectType>(64)
-    protected val deleteObjectsIdsChannel = BroadcastChannel<IdType>(64)
-    override val newObjectsFlow: Flow<ObjectType> = newObjectsChannel.asFlow()
-    override val updatedObjectsFlow: Flow<ObjectType> = updateObjectsChannel.asFlow()
-    override val deletedObjectsIdsFlow: Flow<IdType> = deleteObjectsIdsChannel.asFlow()
+    protected val newObjectsChannel = MutableSharedFlow<ObjectType>(64)
+    protected val updateObjectsChannel = MutableSharedFlow<ObjectType>(64)
+    protected val deleteObjectsIdsChannel = MutableSharedFlow<IdType>(64)
+    override val newObjectsFlow: Flow<ObjectType> = newObjectsChannel.asSharedFlow()
+    override val updatedObjectsFlow: Flow<ObjectType> = updateObjectsChannel.asSharedFlow()
+    override val deletedObjectsIdsFlow: Flow<IdType> = deleteObjectsIdsChannel.asSharedFlow()
 
     protected abstract suspend fun InputValueType.asContentValues(id: IdType? = null): ContentValues
 
@@ -42,7 +40,7 @@ abstract class AbstractMutableAndroidCRUDRepo<ObjectType, IdType, InputValueType
             }
         }.also {
             it.forEach {
-                newObjectsChannel.send(it)
+                newObjectsChannel.emit(it)
             }
         }
     }
@@ -59,7 +57,7 @@ abstract class AbstractMutableAndroidCRUDRepo<ObjectType, IdType, InputValueType
             }
         }
         deleted.forEach {
-            deleteObjectsIdsChannel.send(it)
+            deleteObjectsIdsChannel.emit(it)
         }
     }
 
@@ -76,7 +74,7 @@ abstract class AbstractMutableAndroidCRUDRepo<ObjectType, IdType, InputValueType
             }
         }
         return getById(id) ?.also {
-            updateObjectsChannel.send(it)
+            updateObjectsChannel.emit(it)
         }
     }
 
@@ -95,7 +93,7 @@ abstract class AbstractMutableAndroidCRUDRepo<ObjectType, IdType, InputValueType
             getById(it.first)
         }.also {
             it.forEach {
-                updateObjectsChannel.send(it)
+                updateObjectsChannel.emit(it)
             }
         }
     }

repos/common/src/main/kotlin/dev/inmo/micro_utils/repos/keyvalue/KeyValueStore.kt

@@ -8,10 +8,7 @@ import dev.inmo.micro_utils.pagination.PaginationResult
 import dev.inmo.micro_utils.pagination.utils.paginate
 import dev.inmo.micro_utils.pagination.utils.reverse
 import dev.inmo.micro_utils.repos.StandardKeyValueRepo
-import kotlinx.coroutines.channels.BroadcastChannel
-import kotlinx.coroutines.channels.Channel
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.*
 
 private val cache = HashMap<String, KeyValueStore<*>>()
 
@@ -37,11 +34,11 @@ class KeyValueStore<T : Any> internal constructor (
         null
     }
 
-    private val onNewValueChannel = BroadcastChannel<Pair<String, T>>(Channel.BUFFERED)
-    private val onValueRemovedChannel = BroadcastChannel<String>(Channel.BUFFERED)
+    private val onNewValueChannel = MutableSharedFlow<Pair<String, T>>()
+    private val onValueRemovedChannel = MutableSharedFlow<String>()
 
-    override val onNewValue: Flow<Pair<String, T>> = onNewValueChannel.asFlow()
-    override val onValueRemoved: Flow<String> = onValueRemovedChannel.asFlow()
+    override val onNewValue: Flow<Pair<String, T>> = onNewValueChannel.asSharedFlow()
+    override val onValueRemoved: Flow<String> = onValueRemovedChannel.asSharedFlow()
 
     init {
         cachedData ?.let {
@@ -131,7 +128,7 @@ class KeyValueStore<T : Any> internal constructor (
             }
         }
         toSet.forEach { (k, v) ->
-            onNewValueChannel.send(k to v)
+            onNewValueChannel.emit(k to v)
         }
     }
 
@@ -139,6 +136,6 @@ class KeyValueStore<T : Any> internal constructor (
         sharedPreferences.edit {
             toUnset.forEach { remove(it) }
         }
-        toUnset.forEach { onValueRemovedChannel.send(it) }
+        toUnset.forEach { onValueRemovedChannel.emit(it) }
     }
 }

repos/common/src/main/kotlin/dev/inmo/micro_utils/repos/versions/AndroidSQLStandardVersionsRepoProxy.kt

@@ -32,11 +32,11 @@ class AndroidSQLStandardVersionsRepoProxy(
         }
     }
 
-    override suspend fun getTableVersion(table: String): Int? = database.writableTransaction {
+    override suspend fun getTableVersion(tableName: String): Int? = database.writableTransaction {
         select(
-            tableName,
+            this@AndroidSQLStandardVersionsRepoProxy.tableName,
             selection = "$tableNameColumnName=?",
-            selectionArgs = arrayOf(table),
+            selectionArgs = arrayOf(tableName),
             limit = limitClause(1)
         ).use {
             if (it.moveToFirst()) {
@@ -47,16 +47,16 @@ class AndroidSQLStandardVersionsRepoProxy(
         }
     }
 
-    override suspend fun updateTableVersion(table: String, version: Int) {
+    override suspend fun updateTableVersion(tableName: String, version: Int) {
         database.writableTransaction {
             val updated = update(
-                tableName,
+                this@AndroidSQLStandardVersionsRepoProxy.tableName,
                 contentValuesOf(tableVersionColumnName to version),
                 "$tableNameColumnName=?",
-                arrayOf(table)
+                arrayOf(tableName)
             ) > 0
             if (!updated) {
-                insert(tableName, null, contentValuesOf(tableNameColumnName to table, tableVersionColumnName to version))
+                insert(this@AndroidSQLStandardVersionsRepoProxy.tableName, null, contentValuesOf(tableNameColumnName to tableName, tableVersionColumnName to version))
             }
         }
     }

repos/exposed/src/jvmMain/kotlin/dev/inmo/micro_utils/repos/exposed/AbstractExposedWriteCRUDRepo.kt

@@ -2,7 +2,6 @@ package dev.inmo.micro_utils.repos.exposed
 
 import dev.inmo.micro_utils.repos.UpdatedValuePair
 import dev.inmo.micro_utils.repos.WriteStandardCRUDRepo
-import kotlinx.coroutines.channels.BroadcastChannel
 import kotlinx.coroutines.flow.*
 import org.jetbrains.exposed.sql.*
 import org.jetbrains.exposed.sql.statements.InsertStatement