mirror of
https://github.com/InsanusMokrassar/MicroUtils.git
synced 2025-09-08 17:49:44 +00:00
update dependencies and make a lot of fixes in repos and smart lockers/semaphors
This commit is contained in:
@@ -61,18 +61,28 @@ open class WriteCRUDCacheRepo<ObjectType, IdType, InputValueType>(
|
||||
override val deletedObjectsIdsFlow: Flow<IdType> by parentRepo::deletedObjectsIdsFlow
|
||||
|
||||
val createdObjectsFlowJob = parentRepo.newObjectsFlow.onEach {
|
||||
locker.withWriteLock { kvCache.set(idGetter(it), it) }
|
||||
locker.withWriteLock {
|
||||
kvCache.set(idGetter(it), it)
|
||||
}
|
||||
}.launchIn(scope)
|
||||
|
||||
val updatedObjectsFlowJob = parentRepo.updatedObjectsFlow.onEach {
|
||||
locker.withWriteLock { kvCache.set(idGetter(it), it) }
|
||||
locker.withWriteLock {
|
||||
kvCache.set(idGetter(it), it)
|
||||
}
|
||||
}.launchIn(scope)
|
||||
|
||||
val deletedObjectsFlowJob = parentRepo.deletedObjectsIdsFlow.onEach {
|
||||
locker.withWriteLock { kvCache.unset(it) }
|
||||
locker.withWriteLock {
|
||||
kvCache.unset(it)
|
||||
}
|
||||
}.launchIn(scope)
|
||||
|
||||
override suspend fun deleteById(ids: List<IdType>) = parentRepo.deleteById(ids)
|
||||
override suspend fun deleteById(ids: List<IdType>) = parentRepo.deleteById(ids).also {
|
||||
locker.withWriteLock {
|
||||
kvCache.unset(ids)
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun update(values: List<UpdatedValuePair<IdType, InputValueType>>): List<ObjectType> {
|
||||
val updated = parentRepo.update(values)
|
||||
|
@@ -43,7 +43,7 @@ open class FullReadKeyValueCacheRepo<Key,Value>(
|
||||
protected open suspend fun actualizeAll() {
|
||||
locker.withWriteLock {
|
||||
kvCache.clear()
|
||||
kvCache.set(parentRepo.getAll { keys(it) }.toMap())
|
||||
kvCache.set(parentRepo.getAll())
|
||||
}
|
||||
}
|
||||
|
||||
|
@@ -65,12 +65,12 @@ abstract class WriteMapCRUDRepo<ObjectType, IdType, InputValueType>(
|
||||
protected val map: MutableMap<IdType, ObjectType> = mutableMapOf(),
|
||||
protected val locker: SmartRWLocker = SmartRWLocker()
|
||||
) : WriteCRUDRepo<ObjectType, IdType, InputValueType> {
|
||||
protected val _newObjectsFlow: MutableSharedFlow<ObjectType> = MutableSharedFlow()
|
||||
override val newObjectsFlow: Flow<ObjectType> = _newObjectsFlow.asSharedFlow()
|
||||
protected val _updatedObjectsFlow: MutableSharedFlow<ObjectType> = MutableSharedFlow()
|
||||
override val updatedObjectsFlow: Flow<ObjectType> = _updatedObjectsFlow.asSharedFlow()
|
||||
protected val _deletedObjectsIdsFlow: MutableSharedFlow<IdType> = MutableSharedFlow()
|
||||
override val deletedObjectsIdsFlow: Flow<IdType> = _deletedObjectsIdsFlow.asSharedFlow()
|
||||
protected open val _newObjectsFlow: MutableSharedFlow<ObjectType> = MapsReposDefaultMutableSharedFlow()
|
||||
override val newObjectsFlow: Flow<ObjectType> by lazy { _newObjectsFlow.asSharedFlow() }
|
||||
protected open val _updatedObjectsFlow: MutableSharedFlow<ObjectType> = MapsReposDefaultMutableSharedFlow()
|
||||
override val updatedObjectsFlow: Flow<ObjectType> by lazy { _updatedObjectsFlow.asSharedFlow() }
|
||||
protected open val _deletedObjectsIdsFlow: MutableSharedFlow<IdType> = MapsReposDefaultMutableSharedFlow()
|
||||
override val deletedObjectsIdsFlow: Flow<IdType> by lazy { _deletedObjectsIdsFlow.asSharedFlow() }
|
||||
|
||||
protected abstract suspend fun updateObject(newValue: InputValueType, id: IdType, old: ObjectType): ObjectType
|
||||
protected abstract suspend fun createObject(newValue: InputValueType): Pair<IdType, ObjectType>
|
||||
@@ -80,10 +80,10 @@ abstract class WriteMapCRUDRepo<ObjectType, IdType, InputValueType>(
|
||||
values.map {
|
||||
val (id, newObject) = createObject(it)
|
||||
map[id] = newObject
|
||||
newObject.also { _ ->
|
||||
_newObjectsFlow.emit(newObject)
|
||||
}
|
||||
newObject
|
||||
}
|
||||
}.onEach {
|
||||
_newObjectsFlow.emit(it)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -93,8 +93,9 @@ abstract class WriteMapCRUDRepo<ObjectType, IdType, InputValueType>(
|
||||
|
||||
newValue.also {
|
||||
map[id] = it
|
||||
_updatedObjectsFlow.emit(it)
|
||||
}
|
||||
} ?.also {
|
||||
_updatedObjectsFlow.emit(it)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -104,10 +105,10 @@ abstract class WriteMapCRUDRepo<ObjectType, IdType, InputValueType>(
|
||||
|
||||
override suspend fun deleteById(ids: List<IdType>) {
|
||||
locker.withWriteLock {
|
||||
ids.forEach {
|
||||
map.remove(it) ?.also { _ -> _deletedObjectsIdsFlow.emit(it) }
|
||||
ids.mapNotNull {
|
||||
it.takeIf { map.remove(it) != null }
|
||||
}
|
||||
}
|
||||
}.onEach { _deletedObjectsIdsFlow.emit(it) }
|
||||
}
|
||||
|
||||
}
|
||||
|
@@ -9,6 +9,7 @@ import dev.inmo.micro_utils.pagination.utils.paginate
|
||||
import dev.inmo.micro_utils.pagination.utils.reverse
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.asSharedFlow
|
||||
|
||||
/**
|
||||
* [Map]-based [ReadKeyValueRepo]. All internal operations will be locked with [locker] (mostly with
|
||||
@@ -88,12 +89,14 @@ class WriteMapKeyValueRepo<Key, Value>(
|
||||
private val map: MutableMap<Key, Value>,
|
||||
private val locker: SmartRWLocker
|
||||
) : WriteKeyValueRepo<Key, Value> {
|
||||
private val _onNewValue: MutableSharedFlow<Pair<Key, Value>> = MutableSharedFlow()
|
||||
override val onNewValue: Flow<Pair<Key, Value>>
|
||||
get() = _onNewValue
|
||||
private val _onValueRemoved: MutableSharedFlow<Key> = MutableSharedFlow()
|
||||
override val onValueRemoved: Flow<Key>
|
||||
get() = _onValueRemoved
|
||||
private val _onNewValue: MutableSharedFlow<Pair<Key, Value>> = MapsReposDefaultMutableSharedFlow()
|
||||
override val onNewValue: Flow<Pair<Key, Value>> by lazy {
|
||||
_onNewValue.asSharedFlow()
|
||||
}
|
||||
private val _onValueRemoved: MutableSharedFlow<Key> = MapsReposDefaultMutableSharedFlow()
|
||||
override val onValueRemoved: Flow<Key> by lazy {
|
||||
_onValueRemoved.asSharedFlow()
|
||||
}
|
||||
constructor(map: MutableMap<Key, Value> = mutableMapOf()) : this(map, SmartRWLocker())
|
||||
|
||||
override suspend fun set(toSet: Map<Key, Value>) {
|
||||
@@ -103,10 +106,10 @@ class WriteMapKeyValueRepo<Key, Value>(
|
||||
|
||||
override suspend fun unset(toUnset: List<Key>) {
|
||||
locker.withWriteLock {
|
||||
toUnset.forEach { k ->
|
||||
map.remove(k) ?.also { _ -> _onValueRemoved.emit(k) }
|
||||
toUnset.mapNotNull { k ->
|
||||
map.remove(k) ?.let { _ -> k }
|
||||
}
|
||||
}
|
||||
}.forEach { _onValueRemoved.emit(it) }
|
||||
}
|
||||
|
||||
override suspend fun unsetWithValues(toUnset: List<Value>) {
|
||||
|
@@ -81,68 +81,94 @@ class MapWriteKeyValuesRepo<Key, Value>(
|
||||
private val map: MutableMap<Key, MutableList<Value>> = mutableMapOf(),
|
||||
private val locker: SmartRWLocker = SmartRWLocker()
|
||||
) : WriteKeyValuesRepo<Key, Value> {
|
||||
private val _onNewValue: MutableSharedFlow<Pair<Key, Value>> = MutableSharedFlow()
|
||||
override val onNewValue: Flow<Pair<Key, Value>> = _onNewValue.asSharedFlow()
|
||||
private val _onValueRemoved: MutableSharedFlow<Pair<Key, Value>> = MutableSharedFlow()
|
||||
override val onValueRemoved: Flow<Pair<Key, Value>> = _onValueRemoved.asSharedFlow()
|
||||
private val _onDataCleared: MutableSharedFlow<Key> = MutableSharedFlow()
|
||||
override val onDataCleared: Flow<Key> = _onDataCleared.asSharedFlow()
|
||||
private val _onNewValue: MutableSharedFlow<Pair<Key, Value>> = MapsReposDefaultMutableSharedFlow()
|
||||
override val onNewValue: Flow<Pair<Key, Value>> by lazy {
|
||||
_onNewValue.asSharedFlow()
|
||||
}
|
||||
private val _onValueRemoved: MutableSharedFlow<Pair<Key, Value>> = MapsReposDefaultMutableSharedFlow()
|
||||
override val onValueRemoved: Flow<Pair<Key, Value>> by lazy {
|
||||
_onValueRemoved.asSharedFlow()
|
||||
}
|
||||
private val _onDataCleared: MutableSharedFlow<Key> = MapsReposDefaultMutableSharedFlow()
|
||||
override val onDataCleared: Flow<Key> by lazy {
|
||||
_onDataCleared.asSharedFlow()
|
||||
}
|
||||
|
||||
override suspend fun add(toAdd: Map<Key, List<Value>>) {
|
||||
locker.withWriteLock {
|
||||
toAdd.keys.forEach { k ->
|
||||
if (map.getOrPut(k) { mutableListOf() }.addAll(toAdd[k] ?: return@forEach)) {
|
||||
toAdd[k] ?.forEach { v ->
|
||||
_onNewValue.emit(k to v)
|
||||
}
|
||||
toAdd.keys.mapNotNull { k ->
|
||||
(k to toAdd[k]).takeIf {
|
||||
map.getOrPut(k) { mutableListOf() }.addAll(toAdd[k] ?: return@mapNotNull null)
|
||||
}
|
||||
}
|
||||
}.forEach { (k, vs) ->
|
||||
vs ?.forEach { v ->
|
||||
_onNewValue.emit(k to v)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun remove(toRemove: Map<Key, List<Value>>) {
|
||||
val removed = mutableListOf<Pair<Key, Value>>()
|
||||
val cleared = mutableListOf<Key>()
|
||||
locker.withWriteLock {
|
||||
toRemove.keys.forEach { k ->
|
||||
if (map[k]?.removeAll(toRemove[k] ?: return@forEach) == true) {
|
||||
toRemove[k]?.forEach { v ->
|
||||
_onValueRemoved.emit(k to v)
|
||||
removed.add(k to v)
|
||||
}
|
||||
}
|
||||
if (map[k]?.isEmpty() == true) {
|
||||
map.remove(k)
|
||||
_onDataCleared.emit(k)
|
||||
cleared.add(k)
|
||||
}
|
||||
}
|
||||
}
|
||||
removed.forEach {
|
||||
_onValueRemoved.emit(it)
|
||||
}
|
||||
cleared.forEach {
|
||||
_onDataCleared.emit(it)
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun removeWithValue(v: Value) {
|
||||
locker.withWriteLock {
|
||||
map.forEach { (k, values) ->
|
||||
map.mapNotNull { (k, values) ->
|
||||
if (values.remove(v)) {
|
||||
_onValueRemoved.emit(k to v)
|
||||
k to v
|
||||
} else {
|
||||
null
|
||||
}
|
||||
}
|
||||
}.forEach {
|
||||
_onValueRemoved.emit(it)
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun clear(k: Key) {
|
||||
locker.withWriteLock {
|
||||
map.remove(k) ?.also { _onDataCleared.emit(k) }
|
||||
}
|
||||
map.remove(k)
|
||||
} ?.also { _onDataCleared.emit(k) }
|
||||
}
|
||||
|
||||
override suspend fun clearWithValue(v: Value) {
|
||||
locker.withWriteLock {
|
||||
map.filter { (_, values) ->
|
||||
values.contains(v)
|
||||
}.forEach {
|
||||
map.remove(it.key)?.onEach { v ->
|
||||
_onValueRemoved.emit(it.key to v)
|
||||
}?.also { _ ->
|
||||
_onDataCleared.emit(it.key)
|
||||
}.mapNotNull {
|
||||
if (map.remove(it.key) != null) {
|
||||
it.toPair()
|
||||
} else {
|
||||
null
|
||||
}
|
||||
}
|
||||
}.forEach {
|
||||
it.second.onEach { v ->
|
||||
_onValueRemoved.emit(it.first to v)
|
||||
}.also { _ ->
|
||||
_onDataCleared.emit(it.first)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@@ -0,0 +1,7 @@
|
||||
package dev.inmo.micro_utils.repos
|
||||
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
|
||||
fun <T> MapsReposDefaultMutableSharedFlow() = MutableSharedFlow<T>(
|
||||
extraBufferCapacity = Int.MAX_VALUE
|
||||
)
|
Reference in New Issue
Block a user