Skip to content

Commit 3216825

Browse files
committed
Use real semaphore in flatten/flatMapMerge
1 parent d90eb26 commit 3216825

File tree

1 file changed

+4
-4
lines changed
  • kotlinx-coroutines-core/common/src/flow/operators

1 file changed

+4
-4
lines changed

kotlinx-coroutines-core/common/src/flow/operators/Merge.kt

+4-4
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import kotlinx.coroutines.channels.*
1313
import kotlinx.coroutines.channels.Channel.Factory.OPTIONAL_CHANNEL
1414
import kotlinx.coroutines.flow.internal.*
1515
import kotlinx.coroutines.internal.*
16+
import kotlinx.coroutines.sync.*
1617
import kotlin.coroutines.*
1718
import kotlin.jvm.*
1819
import kotlinx.coroutines.flow.unsafeFlow as flow
@@ -149,16 +150,15 @@ private class ChannelFlowMerge<T>(
149150

150151
// The actual merge implementation with concurrency limit
151152
private suspend fun mergeImpl(scope: CoroutineScope, collector: ConcurrentFlowCollector<T>) {
152-
val semaphore = Channel<Unit>(concurrency)
153+
val semaphore = Semaphore(concurrency)
153154
@Suppress("UNCHECKED_CAST")
154155
flow.collect { inner ->
155-
// TODO real semaphore (#94)
156-
semaphore.send(Unit) // Acquire concurrency permit
156+
semaphore.acquire() // Acquire concurrency permit
157157
scope.launch {
158158
try {
159159
inner.collect(collector)
160160
} finally {
161-
semaphore.receive() // Release concurrency permit
161+
semaphore.release() // Release concurrency permit
162162
}
163163
}
164164
}

0 commit comments

Comments
 (0)