Fix cancellation of sending event (#2438)

This commit is contained in:
Benoit Marty 2020-11-30 10:08:31 +01:00
parent f6cc05634f
commit cd983de058
6 changed files with 26 additions and 4 deletions

View File

@ -8,7 +8,7 @@ Improvements 🙌:
- -
Bugfix 🐛: Bugfix 🐛:
- - Fix cancellation of sending event (#2438)
Translations 🗣: Translations 🗣:
- -

View File

@ -210,6 +210,8 @@ internal class DefaultSendService @AssistedInject constructor(
override fun cancelSend(eventId: String) { override fun cancelSend(eventId: String) {
cancelSendTracker.markLocalEchoForCancel(eventId, roomId) cancelSendTracker.markLocalEchoForCancel(eventId, roomId)
// This is maybe the current task, so cancel it too
eventSenderProcessor.cancel(eventId, roomId)
taskExecutor.executorScope.launch { taskExecutor.executorScope.launch {
localEchoRepository.deleteFailedEcho(roomId, eventId) localEchoRepository.deleteFailedEcho(roomId, eventId)
} }

View File

@ -16,6 +16,7 @@
package org.matrix.android.sdk.internal.session.room.send.queue package org.matrix.android.sdk.internal.session.room.send.queue
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.launch import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import org.matrix.android.sdk.api.auth.data.SessionParams import org.matrix.android.sdk.api.auth.data.SessionParams
@ -109,10 +110,18 @@ internal class EventSenderProcessor @Inject constructor(
return task return task
} }
fun cancel(eventId: String, roomId: String) {
(currentTask as? SendEventQueuedTask)
?.takeIf { it -> it.event.eventId == eventId && it.event.roomId == roomId }
?.cancel()
}
companion object { companion object {
private const val RETRY_WAIT_TIME_MS = 10_000L private const val RETRY_WAIT_TIME_MS = 10_000L
} }
private var currentTask: QueuedTask? = null
private var sendingQueue = LinkedBlockingQueue<QueuedTask>() private var sendingQueue = LinkedBlockingQueue<QueuedTask>()
private var networkAvailableLock = Object() private var networkAvailableLock = Object()
@ -125,6 +134,7 @@ internal class EventSenderProcessor @Inject constructor(
while (!isInterrupted) { while (!isInterrupted) {
Timber.v("## SendThread wait for task to process") Timber.v("## SendThread wait for task to process")
val task = sendingQueue.take() val task = sendingQueue.take()
.also { currentTask = it }
Timber.v("## SendThread Found task to process $task") Timber.v("## SendThread Found task to process $task")
if (task.isCancelled()) { if (task.isCancelled()) {
@ -179,6 +189,10 @@ internal class EventSenderProcessor @Inject constructor(
task.onTaskFailed() task.onTaskFailed()
throw InterruptedException() throw InterruptedException()
} }
exception is CancellationException -> {
Timber.v("## SendThread task has been cancelled")
break@retryLoop
}
else -> { else -> {
Timber.v("## SendThread retryLoop Un-Retryable error, try next task") Timber.v("## SendThread retryLoop Un-Retryable error, try next task")
// this task is in error, check next one? // this task is in error, check next one?

View File

@ -23,7 +23,13 @@ abstract class QueuedTask : Cancelable {
private var hasBeenCancelled: Boolean = false private var hasBeenCancelled: Boolean = false
abstract suspend fun execute() suspend fun execute() {
if (!isCancelled()) {
doExecute()
}
}
abstract suspend fun doExecute()
abstract fun onTaskFailed() abstract fun onTaskFailed()

View File

@ -33,7 +33,7 @@ internal class RedactQueuedTask(
override fun toString() = "[RedactQueuedTask $redactionLocalEchoId]" override fun toString() = "[RedactQueuedTask $redactionLocalEchoId]"
override suspend fun execute() { override suspend fun doExecute() {
redactEventTask.execute(RedactEventTask.Params(redactionLocalEchoId, roomId, toRedactEventId, reason)) redactEventTask.execute(RedactEventTask.Params(redactionLocalEchoId, roomId, toRedactEventId, reason))
} }

View File

@ -35,7 +35,7 @@ internal class SendEventQueuedTask(
override fun toString() = "[SendEventQueuedTask ${event.eventId}]" override fun toString() = "[SendEventQueuedTask ${event.eventId}]"
override suspend fun execute() { override suspend fun doExecute() {
sendEventTask.execute(SendEventTask.Params(event, encrypt)) sendEventTask.execute(SendEventTask.Params(event, encrypt))
} }