Matrix/element-android-Github
1
0
Fork 0
mirror of https://github.com/vector-im/element-android.git synced 2024-11-16 02:05:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

Coroutine sequencer: use semaphore

Browse Source
This commit is contained in:
Ganard 2020-02-11 12:08:47 +01:00 committed by ganfra
parent b04ee7153a
commit a305ce302e
3 changed files with 16 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
matrix-sdk-android/src/main/java/im/vector/matrix/android/internal/session/sync/SyncTaskSequencer.kt
View File

@ -17,8 +17,8 @@
package im.vector.matrix.android.internal.session.sync package im.vector.matrix.android.internal.session.sync
import im.vector.matrix.android.internal.session.SessionScope import im.vector.matrix.android.internal.session.SessionScope
import im.vector.matrix.android.internal.task.ChannelCoroutineSequencer import im.vector.matrix.android.internal.task.SemaphoreCoroutineSequencer
import javax.inject.Inject import javax.inject.Inject
@SessionScope @SessionScope
internal class SyncTaskSequencer @Inject constructor() : ChannelCoroutineSequencer<Unit>() internal class SyncTaskSequencer @Inject constructor() : SemaphoreCoroutineSequencer()

63
matrix-sdk-android/src/main/java/im/vector/matrix/android/internal/task/CoroutineSequencer.kt
View File

@ -16,72 +16,27 @@
package im.vector.matrix.android.internal.task package im.vector.matrix.android.internal.task
import kotlinx.coroutines.* import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.sync.withPermit
import java.util.concurrent.Executors
/** /**
* This class intends to be used for ensure suspendable methods are played sequentially all the way long. * This class intends to be used for ensure suspendable methods are played sequentially all the way long.
*/ */
internal interface CoroutineSequencer<T> { internal interface CoroutineSequencer {
/** /**
* @param block the suspendable block to execute * @param block the suspendable block to execute
* @return the result of the block * @return the result of the block
*/ */
suspend fun post(block: suspend () -> T): T suspend fun <T> post(block: suspend () -> T): T
/**
* Cancel all and close, so you won't be able to post anything else after
*/
fun close()
} }
internal open class ChannelCoroutineSequencer<T> : CoroutineSequencer<T> { internal open class SemaphoreCoroutineSequencer : CoroutineSequencer {
private data class Message<T>( private val semaphore = Semaphore(1) // Permits 1 suspend function at a time.
val block: suspend () -> T,
val deferred: CompletableDeferred<T>
)
private var messageChannel: Channel<Message<T>> = Channel() override suspend fun <T> post(block: suspend () -> T): T {
private val coroutineScope = CoroutineScope(SupervisorJob()) return semaphore.withPermit {
// This will ensure block()
private val singleDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
init {
launchCoroutine()
}
private fun launchCoroutine() {
coroutineScope.launch(singleDispatcher) {
for (message in messageChannel) {
try {
val result = message.block()
message.deferred.complete(result)
} catch (exception: Throwable) {
message.deferred.completeExceptionally(exception)
}
}
}
}
override fun close() {
coroutineScope.coroutineContext.cancelChildren()
messageChannel.close()
}
override suspend fun post(block: suspend () -> T): T {
val deferred = CompletableDeferred<T>()
val message = Message(block, deferred)
messageChannel.send(message)
return try {
deferred.await()
} catch (cancellation: CancellationException) {
// In case of cancellation, we stop the current coroutine context
// and relaunch one to consume next messages
coroutineScope.coroutineContext.cancelChildren()
launchCoroutine()
throw cancellation
} }
} }
} }

10
matrix-sdk-android/src/test/java/im/vector/matrix/android/internal/task/CoroutineSequencersTest.kt
View File

@ -32,7 +32,7 @@ class CoroutineSequencersTest {
@Test @Test
fun sequencer_should_run_sequential() { fun sequencer_should_run_sequential() {
val sequencer = ChannelCoroutineSequencer<String>() val sequencer = SemaphoreCoroutineSequencer()
val results = ArrayList<String>() val results = ArrayList<String>()
val jobs = listOf( val jobs = listOf(
@ -63,9 +63,9 @@ class CoroutineSequencersTest {
@Test @Test
fun sequencer_should_run_parallel() { fun sequencer_should_run_parallel() {
val sequencer1 = ChannelCoroutineSequencer<String>() val sequencer1 = SemaphoreCoroutineSequencer()
val sequencer2 = ChannelCoroutineSequencer<String>() val sequencer2 = SemaphoreCoroutineSequencer()
val sequencer3 = ChannelCoroutineSequencer<String>() val sequencer3 = SemaphoreCoroutineSequencer()
val results = ArrayList<String>() val results = ArrayList<String>()
val jobs = listOf( val jobs = listOf(
GlobalScope.launch(dispatcher) { GlobalScope.launch(dispatcher) {
@ -92,7 +92,7 @@ class CoroutineSequencersTest {
@Test @Test
fun sequencer_should_jump_to_next_when_current_job_canceled() { fun sequencer_should_jump_to_next_when_current_job_canceled() {
val sequencer = ChannelCoroutineSequencer<String>() val sequencer = SemaphoreCoroutineSequencer()
val results = ArrayList<String>() val results = ArrayList<String>()
val jobs = listOf( val jobs = listOf(
GlobalScope.launch(dispatcher) { GlobalScope.launch(dispatcher) {