mirror of
https://github.com/InsanusMokrassar/MicroUtils.git
synced 2025-09-18 14:59:24 +00:00
Compare commits
13 Commits
Author | SHA1 | Date | |
---|---|---|---|
7f6c02ffdf | |||
f368616e6f | |||
dd632f4203 | |||
a8f3ae501b | |||
ea76963ac2 | |||
99dfa97958 | |||
f68270a5b3 | |||
542ed81034 | |||
404a11f5e7 | |||
411221070e | |||
25d35d0c76 | |||
c72904d61c | |||
b94c9acd26 |
17
CHANGELOG.md
17
CHANGELOG.md
@@ -1,5 +1,22 @@
|
|||||||
# Changelog
|
# Changelog
|
||||||
|
|
||||||
|
## 0.10.4
|
||||||
|
|
||||||
|
* `Versions`:
|
||||||
|
* `Serialization`: `1.3.2` -> `1.3.3`
|
||||||
|
|
||||||
|
## 0.10.3
|
||||||
|
|
||||||
|
* `Versions`:
|
||||||
|
* `Compose`: `1.2.0-alpha01-dev682` -> `1.2.0-alpha01-dev683`
|
||||||
|
* `Coroutines`:
|
||||||
|
* Fixes in `AccumulatorFlow`
|
||||||
|
|
||||||
|
## 0.10.2
|
||||||
|
|
||||||
|
* `Versions`:
|
||||||
|
* `Compose`: `1.2.0-alpha01-dev675` -> `1.2.0-alpha01-dev682`
|
||||||
|
|
||||||
## 0.10.1
|
## 0.10.1
|
||||||
|
|
||||||
* `Versions`:
|
* `Versions`:
|
||||||
|
@@ -6,11 +6,12 @@ import kotlinx.coroutines.channels.Channel
|
|||||||
import kotlinx.coroutines.flow.*
|
import kotlinx.coroutines.flow.*
|
||||||
import kotlinx.coroutines.sync.Mutex
|
import kotlinx.coroutines.sync.Mutex
|
||||||
import kotlinx.coroutines.sync.withLock
|
import kotlinx.coroutines.sync.withLock
|
||||||
|
import kotlin.coroutines.cancellation.CancellationException
|
||||||
|
|
||||||
private sealed interface AccumulatorFlowStep
|
private sealed interface AccumulatorFlowStep<T>
|
||||||
private data class DataRetrievedAccumulatorFlowStep(val data: Any) : AccumulatorFlowStep
|
private data class DataRetrievedAccumulatorFlowStep<T>(val data: T) : AccumulatorFlowStep<T>
|
||||||
private data class SubscribeAccumulatorFlowStep(val channel: Channel<Any>) : AccumulatorFlowStep
|
private data class SubscribeAccumulatorFlowStep<T>(val channel: Channel<T>) : AccumulatorFlowStep<T>
|
||||||
private data class UnsubscribeAccumulatorFlowStep(val channel: Channel<Any>) : AccumulatorFlowStep
|
private data class UnsubscribeAccumulatorFlowStep<T>(val channel: Channel<T>) : AccumulatorFlowStep<T>
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* This [Flow] will have behaviour very similar to [SharedFlow], but there are several differences:
|
* This [Flow] will have behaviour very similar to [SharedFlow], but there are several differences:
|
||||||
@@ -26,12 +27,12 @@ class AccumulatorFlow<T>(
|
|||||||
private val subscope = scope.LinkedSupervisorScope()
|
private val subscope = scope.LinkedSupervisorScope()
|
||||||
private val activeData = ArrayDeque<T>()
|
private val activeData = ArrayDeque<T>()
|
||||||
private val dataMutex = Mutex()
|
private val dataMutex = Mutex()
|
||||||
private val channelsForBroadcast = mutableListOf<Channel<Any>>()
|
private val channelsForBroadcast = mutableListOf<Channel<T>>()
|
||||||
private val channelsMutex = Mutex()
|
private val channelsMutex = Mutex()
|
||||||
private val steps = subscope.actor<AccumulatorFlowStep> { step ->
|
private val steps = subscope.actor<AccumulatorFlowStep<T>> { step ->
|
||||||
when (step) {
|
when (step) {
|
||||||
is DataRetrievedAccumulatorFlowStep -> {
|
is DataRetrievedAccumulatorFlowStep -> {
|
||||||
if (activeData.first() === step.data) {
|
if (activeData.firstOrNull() === step.data) {
|
||||||
dataMutex.withLock {
|
dataMutex.withLock {
|
||||||
activeData.removeFirst()
|
activeData.removeFirst()
|
||||||
}
|
}
|
||||||
@@ -42,7 +43,7 @@ class AccumulatorFlow<T>(
|
|||||||
dataMutex.withLock {
|
dataMutex.withLock {
|
||||||
val dataToSend = activeData.toList()
|
val dataToSend = activeData.toList()
|
||||||
safelyWithoutExceptions {
|
safelyWithoutExceptions {
|
||||||
dataToSend.forEach { step.channel.send(it as Any) }
|
dataToSend.forEach { step.channel.send(it) }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -58,24 +59,29 @@ class AccumulatorFlow<T>(
|
|||||||
channelsMutex.withLock {
|
channelsMutex.withLock {
|
||||||
channelsForBroadcast.forEach { channel ->
|
channelsForBroadcast.forEach { channel ->
|
||||||
safelyWithResult {
|
safelyWithResult {
|
||||||
channel.send(it as Any)
|
channel.send(it)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
override suspend fun collectSafely(collector: FlowCollector<T>) {
|
override suspend fun collectSafely(collector: FlowCollector<T>) {
|
||||||
val channel = Channel<Any>(Channel.UNLIMITED, BufferOverflow.SUSPEND)
|
val channel = Channel<T>(Channel.UNLIMITED, BufferOverflow.SUSPEND)
|
||||||
steps.send(SubscribeAccumulatorFlowStep(channel))
|
steps.send(SubscribeAccumulatorFlowStep(channel))
|
||||||
for (data in channel) {
|
val result = runCatchingSafely {
|
||||||
try {
|
for (data in channel) {
|
||||||
collector.emit(data as T)
|
val emitResult = runCatchingSafely {
|
||||||
steps.send(DataRetrievedAccumulatorFlowStep(data))
|
collector.emit(data)
|
||||||
} finally {
|
}
|
||||||
channel.cancel()
|
if (emitResult.isSuccess || emitResult.exceptionOrNull() is CancellationException) {
|
||||||
steps.send(UnsubscribeAccumulatorFlowStep(channel))
|
steps.send(DataRetrievedAccumulatorFlowStep(data))
|
||||||
|
}
|
||||||
|
emitResult.getOrThrow()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
channel.cancel()
|
||||||
|
steps.send(UnsubscribeAccumulatorFlowStep(channel))
|
||||||
|
result.getOrThrow()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@@ -14,5 +14,5 @@ crypto_js_version=4.1.1
|
|||||||
# Project data
|
# Project data
|
||||||
|
|
||||||
group=dev.inmo
|
group=dev.inmo
|
||||||
version=0.10.1
|
version=0.10.4
|
||||||
android_code_version=116
|
android_code_version=119
|
||||||
|
@@ -1,10 +1,10 @@
|
|||||||
[versions]
|
[versions]
|
||||||
|
|
||||||
kt = "1.6.21"
|
kt = "1.6.21"
|
||||||
kt-serialization = "1.3.2"
|
kt-serialization = "1.3.3"
|
||||||
kt-coroutines = "1.6.1"
|
kt-coroutines = "1.6.1"
|
||||||
|
|
||||||
jb-compose = "1.2.0-alpha01-dev675"
|
jb-compose = "1.2.0-alpha01-dev683"
|
||||||
jb-exposed = "0.38.2"
|
jb-exposed = "0.38.2"
|
||||||
jb-dokka = "1.6.21"
|
jb-dokka = "1.6.21"
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user