filippodb
/
SchildiChat-android
mirror of https://github.com/SchildiChat/SchildiChat-android.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2135 from vector-im/feature/safe_workers 

Create parent class for all MatrixWorker
This commit is contained in:
Benoit Marty 2020-09-22 14:35:15 +02:00 committed by GitHub
parent 48b10e6d08 d868ef0168
commit 23d911cc2c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 315 additions and 252 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/CancelGossipRequestWorker.kt
View File

@ -18,10 +18,9 @@
package org.matrix.android.sdk.internal.crypto package org.matrix.android.sdk.internal.crypto
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.Data
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.api.auth.data.Credentials import org.matrix.android.sdk.api.auth.data.Credentials
import org.matrix.android.sdk.api.failure.shouldBeRetried import org.matrix.android.sdk.api.failure.shouldBeRetried
import org.matrix.android.sdk.api.session.events.model.Event import org.matrix.android.sdk.api.session.events.model.Event
@ -32,28 +31,29 @@ import org.matrix.android.sdk.internal.crypto.model.MXUsersDevicesMap
import org.matrix.android.sdk.internal.crypto.model.rest.ShareRequestCancellation import org.matrix.android.sdk.internal.crypto.model.rest.ShareRequestCancellation
import org.matrix.android.sdk.internal.crypto.store.IMXCryptoStore import org.matrix.android.sdk.internal.crypto.store.IMXCryptoStore
import org.matrix.android.sdk.internal.crypto.tasks.SendToDeviceTask import org.matrix.android.sdk.internal.crypto.tasks.SendToDeviceTask
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.worker.getSessionComponent import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.greenrobot.eventbus.EventBus import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
internal class CancelGossipRequestWorker(context: Context, internal class CancelGossipRequestWorker(context: Context,
params: WorkerParameters) params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<CancelGossipRequestWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
val sessionId: String, override val sessionId: String,
val requestId: String, val requestId: String,
val recipients: Map<String, List<String>> val recipients: Map<String, List<String>>,
) { override val lastFailureMessage: String? = null
) : SessionWorkerParams {
companion object { companion object {
fun fromRequest(sessionId: String, request: OutgoingGossipingRequest): Params { fun fromRequest(sessionId: String, request: OutgoingGossipingRequest): Params {
return Params( return Params(
sessionId = sessionId, sessionId = sessionId,
requestId = request.requestId, requestId = request.requestId,
recipients = request.recipients recipients = request.recipients,
lastFailureMessage = null
) )
} }
} }
@ -64,18 +64,11 @@ internal class CancelGossipRequestWorker(context: Context,
@Inject lateinit var eventBus: EventBus @Inject lateinit var eventBus: EventBus
@Inject lateinit var credentials: Credentials @Inject lateinit var credentials: Credentials
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val errorOutputData = Data.Builder().putBoolean("failed", true).build() injector.inject(this)
val params = WorkerParamsFactory.fromData<Params>(inputData)
?: return Result.success(errorOutputData)
val sessionComponent = getSessionComponent(params.sessionId)
?: return Result.success(errorOutputData).also {
// TODO, can this happen? should I update local echo?
Timber.e("Unknown Session, cannot send message, sessionId: ${params.sessionId}")
} }
sessionComponent.inject(this)
override suspend fun doSafeWork(params: Params): Result {
val localId = LocalEcho.createLocalEchoId() val localId = LocalEcho.createLocalEchoId()
val contentMap = MXUsersDevicesMap<Any>() val contentMap = MXUsersDevicesMap<Any>()
val toDeviceContent = ShareRequestCancellation( val toDeviceContent = ShareRequestCancellation(
@ -107,13 +100,17 @@ internal class CancelGossipRequestWorker(context: Context,
) )
cryptoStore.updateOutgoingGossipingRequestState(params.requestId, OutgoingGossipingRequestState.CANCELLED) cryptoStore.updateOutgoingGossipingRequestState(params.requestId, OutgoingGossipingRequestState.CANCELLED)
return Result.success() return Result.success()
} catch (exception: Throwable) { } catch (throwable: Throwable) {
return if (exception.shouldBeRetried()) { return if (throwable.shouldBeRetried()) {
Result.retry() Result.retry()
} else { } else {
cryptoStore.updateOutgoingGossipingRequestState(params.requestId, OutgoingGossipingRequestState.FAILED_TO_CANCEL) cryptoStore.updateOutgoingGossipingRequestState(params.requestId, OutgoingGossipingRequestState.FAILED_TO_CANCEL)
Result.success(errorOutputData) buildErrorResult(params, throwable.localizedMessage ?: "error")
} }
} }
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
} }

43
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/SendGossipRequestWorker.kt
View File

@ -18,10 +18,9 @@
package org.matrix.android.sdk.internal.crypto package org.matrix.android.sdk.internal.crypto
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.Data
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.api.auth.data.Credentials import org.matrix.android.sdk.api.auth.data.Credentials
import org.matrix.android.sdk.api.failure.shouldBeRetried import org.matrix.android.sdk.api.failure.shouldBeRetried
import org.matrix.android.sdk.api.session.events.model.Event import org.matrix.android.sdk.api.session.events.model.Event
@ -34,40 +33,34 @@ import org.matrix.android.sdk.internal.crypto.model.rest.RoomKeyShareRequest
import org.matrix.android.sdk.internal.crypto.model.rest.SecretShareRequest import org.matrix.android.sdk.internal.crypto.model.rest.SecretShareRequest
import org.matrix.android.sdk.internal.crypto.store.IMXCryptoStore import org.matrix.android.sdk.internal.crypto.store.IMXCryptoStore
import org.matrix.android.sdk.internal.crypto.tasks.SendToDeviceTask import org.matrix.android.sdk.internal.crypto.tasks.SendToDeviceTask
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.worker.getSessionComponent import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.greenrobot.eventbus.EventBus import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import timber.log.Timber import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
internal class SendGossipRequestWorker(context: Context, internal class SendGossipRequestWorker(context: Context,
params: WorkerParameters) params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<SendGossipRequestWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
val sessionId: String, override val sessionId: String,
val keyShareRequest: OutgoingRoomKeyRequest? = null, val keyShareRequest: OutgoingRoomKeyRequest? = null,
val secretShareRequest: OutgoingSecretRequest? = null val secretShareRequest: OutgoingSecretRequest? = null,
) override val lastFailureMessage: String? = null
) : SessionWorkerParams
@Inject lateinit var sendToDeviceTask: SendToDeviceTask @Inject lateinit var sendToDeviceTask: SendToDeviceTask
@Inject lateinit var cryptoStore: IMXCryptoStore @Inject lateinit var cryptoStore: IMXCryptoStore
@Inject lateinit var eventBus: EventBus @Inject lateinit var eventBus: EventBus
@Inject lateinit var credentials: Credentials @Inject lateinit var credentials: Credentials
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val errorOutputData = Data.Builder().putBoolean("failed", true).build() injector.inject(this)
val params = WorkerParamsFactory.fromData<Params>(inputData)
?: return Result.success(errorOutputData)
val sessionComponent = getSessionComponent(params.sessionId)
?: return Result.success(errorOutputData).also {
// TODO, can this happen? should I update local echo?
Timber.e("Unknown Session, cannot send message, sessionId: ${params.sessionId}")
} }
sessionComponent.inject(this)
override suspend fun doSafeWork(params: Params): Result {
val localId = LocalEcho.createLocalEchoId() val localId = LocalEcho.createLocalEchoId()
val contentMap = MXUsersDevicesMap<Any>() val contentMap = MXUsersDevicesMap<Any>()
val eventType: String val eventType: String
@ -121,7 +114,7 @@ internal class SendGossipRequestWorker(context: Context,
} }
} }
else -> { else -> {
return Result.success(errorOutputData).also { return buildErrorResult(params, "Unknown empty gossiping request").also {
Timber.e("Unknown empty gossiping request: $params") Timber.e("Unknown empty gossiping request: $params")
} }
} }
@ -137,13 +130,17 @@ internal class SendGossipRequestWorker(context: Context,
) )
cryptoStore.updateOutgoingGossipingRequestState(requestId, OutgoingGossipingRequestState.SENT) cryptoStore.updateOutgoingGossipingRequestState(requestId, OutgoingGossipingRequestState.SENT)
return Result.success() return Result.success()
} catch (exception: Throwable) { } catch (throwable: Throwable) {
return if (exception.shouldBeRetried()) { return if (throwable.shouldBeRetried()) {
Result.retry() Result.retry()
} else { } else {
cryptoStore.updateOutgoingGossipingRequestState(requestId, OutgoingGossipingRequestState.FAILED_TO_SEND) cryptoStore.updateOutgoingGossipingRequestState(requestId, OutgoingGossipingRequestState.FAILED_TO_SEND)
Result.success(errorOutputData) buildErrorResult(params, throwable.localizedMessage ?: "error")
} }
} }
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
} }

45
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/SendGossipWorker.kt
View File

@ -18,10 +18,9 @@
package org.matrix.android.sdk.internal.crypto package org.matrix.android.sdk.internal.crypto
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.Data
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.api.auth.data.Credentials import org.matrix.android.sdk.api.auth.data.Credentials
import org.matrix.android.sdk.api.failure.shouldBeRetried import org.matrix.android.sdk.api.failure.shouldBeRetried
import org.matrix.android.sdk.api.session.events.model.Event import org.matrix.android.sdk.api.session.events.model.Event
@ -34,22 +33,23 @@ import org.matrix.android.sdk.internal.crypto.model.MXUsersDevicesMap
import org.matrix.android.sdk.internal.crypto.model.event.SecretSendEventContent import org.matrix.android.sdk.internal.crypto.model.event.SecretSendEventContent
import org.matrix.android.sdk.internal.crypto.store.IMXCryptoStore import org.matrix.android.sdk.internal.crypto.store.IMXCryptoStore
import org.matrix.android.sdk.internal.crypto.tasks.SendToDeviceTask import org.matrix.android.sdk.internal.crypto.tasks.SendToDeviceTask
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.worker.getSessionComponent import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.greenrobot.eventbus.EventBus import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import timber.log.Timber import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
internal class SendGossipWorker(context: Context, internal class SendGossipWorker(context: Context,
params: WorkerParameters) params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<SendGossipWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
val sessionId: String, override val sessionId: String,
val secretValue: String, val secretValue: String,
val request: IncomingSecretShareRequest val request: IncomingSecretShareRequest,
) override val lastFailureMessage: String? = null
) : SessionWorkerParams
@Inject lateinit var sendToDeviceTask: SendToDeviceTask @Inject lateinit var sendToDeviceTask: SendToDeviceTask
@Inject lateinit var cryptoStore: IMXCryptoStore @Inject lateinit var cryptoStore: IMXCryptoStore
@ -58,18 +58,11 @@ internal class SendGossipWorker(context: Context,
@Inject lateinit var messageEncrypter: MessageEncrypter @Inject lateinit var messageEncrypter: MessageEncrypter
@Inject lateinit var ensureOlmSessionsForDevicesAction: EnsureOlmSessionsForDevicesAction @Inject lateinit var ensureOlmSessionsForDevicesAction: EnsureOlmSessionsForDevicesAction
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val errorOutputData = Data.Builder().putBoolean("failed", true).build() injector.inject(this)
val params = WorkerParamsFactory.fromData<Params>(inputData)
?: return Result.success(errorOutputData)
val sessionComponent = getSessionComponent(params.sessionId)
?: return Result.success(errorOutputData).also {
// TODO, can this happen? should I update local echo?
Timber.e("Unknown Session, cannot send message, sessionId: ${params.sessionId}")
} }
sessionComponent.inject(this)
override suspend fun doSafeWork(params: Params): Result {
val localId = LocalEcho.createLocalEchoId() val localId = LocalEcho.createLocalEchoId()
val eventType: String = EventType.SEND_SECRET val eventType: String = EventType.SEND_SECRET
@ -81,7 +74,7 @@ internal class SendGossipWorker(context: Context,
val requestingUserId = params.request.userId ?: "" val requestingUserId = params.request.userId ?: ""
val requestingDeviceId = params.request.deviceId ?: "" val requestingDeviceId = params.request.deviceId ?: ""
val deviceInfo = cryptoStore.getUserDevice(requestingUserId, requestingDeviceId) val deviceInfo = cryptoStore.getUserDevice(requestingUserId, requestingDeviceId)
?: return Result.success(errorOutputData).also { ?: return buildErrorResult(params, "Unknown deviceInfo, cannot send message").also {
cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.FAILED_TO_ACCEPTED) cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.FAILED_TO_ACCEPTED)
Timber.e("Unknown deviceInfo, cannot send message, sessionId: ${params.request.deviceId}") Timber.e("Unknown deviceInfo, cannot send message, sessionId: ${params.request.deviceId}")
} }
@ -94,7 +87,7 @@ internal class SendGossipWorker(context: Context,
if (olmSessionResult?.sessionId == null) { if (olmSessionResult?.sessionId == null) {
// no session with this device, probably because there // no session with this device, probably because there
// were no one-time keys. // were no one-time keys.
return Result.success(errorOutputData).also { return buildErrorResult(params, "no session with this device").also {
cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.FAILED_TO_ACCEPTED) cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.FAILED_TO_ACCEPTED)
Timber.e("no session with this device, probably because there were no one-time keys.") Timber.e("no session with this device, probably because there were no one-time keys.")
} }
@ -130,13 +123,17 @@ internal class SendGossipWorker(context: Context,
) )
cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.ACCEPTED) cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.ACCEPTED)
return Result.success() return Result.success()
} catch (exception: Throwable) { } catch (throwable: Throwable) {
return if (exception.shouldBeRetried()) { return if (throwable.shouldBeRetried()) {
Result.retry() Result.retry()
} else { } else {
cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.FAILED_TO_ACCEPTED) cryptoStore.updateGossipingRequestState(params.request, GossipingRequestState.FAILED_TO_ACCEPTED)
Result.success(errorOutputData) buildErrorResult(params, throwable.localizedMessage ?: "error")
} }
} }
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
} }

48
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/verification/SendVerificationMessageWorker.kt
View File

@ -17,18 +17,17 @@
package org.matrix.android.sdk.internal.crypto.verification package org.matrix.android.sdk.internal.crypto.verification
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.Data import androidx.work.Data
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.api.failure.shouldBeRetried import org.matrix.android.sdk.api.failure.shouldBeRetried
import org.matrix.android.sdk.api.session.crypto.CryptoService import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.internal.crypto.tasks.SendVerificationMessageTask import org.matrix.android.sdk.internal.crypto.tasks.SendVerificationMessageTask
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.session.room.send.CancelSendTracker import org.matrix.android.sdk.internal.session.room.send.CancelSendTracker
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import timber.log.Timber import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
@ -38,7 +37,7 @@ import javax.inject.Inject
*/ */
internal class SendVerificationMessageWorker(context: Context, internal class SendVerificationMessageWorker(context: Context,
params: WorkerParameters) params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<SendVerificationMessageWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -47,30 +46,17 @@ internal class SendVerificationMessageWorker(context: Context,
override val lastFailureMessage: String? = null override val lastFailureMessage: String? = null
) : SessionWorkerParams ) : SessionWorkerParams
@Inject @Inject lateinit var sendVerificationMessageTask: SendVerificationMessageTask
lateinit var sendVerificationMessageTask: SendVerificationMessageTask @Inject lateinit var localEchoRepository: LocalEchoRepository
@Inject lateinit var cryptoService: CryptoService
@Inject
lateinit var localEchoRepository: LocalEchoRepository
@Inject
lateinit var cryptoService: CryptoService
@Inject lateinit var cancelSendTracker: CancelSendTracker @Inject lateinit var cancelSendTracker: CancelSendTracker
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val errorOutputData = Data.Builder().putBoolean(OUTPUT_KEY_FAILED, true).build() injector.inject(this)
val params = WorkerParamsFactory.fromData<Params>(inputData)
?: return Result.success(errorOutputData)
val sessionComponent = getSessionComponent(params.sessionId)
?: return Result.success(errorOutputData).also {
// TODO, can this happen? should I update local echo?
Timber.e("Unknown Session, cannot send message, sessionId: ${params.sessionId}")
} }
sessionComponent.inject(this)
val localEvent = localEchoRepository.getUpToDateEcho(params.eventId) ?: return Result.success(errorOutputData) override suspend fun doSafeWork(params: Params): Result {
val localEvent = localEchoRepository.getUpToDateEcho(params.eventId) ?: return buildErrorResult(params, "Event not found")
val localEventId = localEvent.eventId ?: "" val localEventId = localEvent.eventId ?: ""
val roomId = localEvent.roomId ?: "" val roomId = localEvent.roomId ?: ""
@ -91,20 +77,16 @@ internal class SendVerificationMessageWorker(context: Context,
) )
Result.success(Data.Builder().putString(localEventId, resultEventId).build()) Result.success(Data.Builder().putString(localEventId, resultEventId).build())
} catch (exception: Throwable) { } catch (throwable: Throwable) {
if (exception.shouldBeRetried()) { if (throwable.shouldBeRetried()) {
Result.retry() Result.retry()
} else { } else {
Result.success(errorOutputData) buildErrorResult(params, throwable.localizedMessage ?: "error")
} }
} }
} }
companion object { override fun buildErrorParams(params: Params, message: String): Params {
private const val OUTPUT_KEY_FAILED = "failed" return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
fun hasFailed(outputData: Data): Boolean {
return outputData.getBoolean(OUTPUT_KEY_FAILED, false)
}
} }
} }

6
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/verification/VerificationTransportRoomMessage.kt
View File

@ -54,6 +54,7 @@ import org.matrix.android.sdk.internal.di.WorkManagerProvider
import org.matrix.android.sdk.internal.session.room.send.LocalEchoEventFactory import org.matrix.android.sdk.internal.session.room.send.LocalEchoEventFactory
import org.matrix.android.sdk.internal.task.TaskExecutor import org.matrix.android.sdk.internal.task.TaskExecutor
import org.matrix.android.sdk.internal.util.StringProvider import org.matrix.android.sdk.internal.util.StringProvider
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import timber.log.Timber import timber.log.Timber
import java.util.UUID import java.util.UUID
@ -117,14 +118,13 @@ internal class VerificationTransportRoomMessage(
workInfoList workInfoList
?.firstOrNull { it.id == enqueueInfo.second } ?.firstOrNull { it.id == enqueueInfo.second }
?.let { wInfo -> ?.let { wInfo ->
when (wInfo.state) { when (wInfo.state) {
WorkInfo.State.FAILED -> { WorkInfo.State.FAILED -> {
tx?.cancel(onErrorReason) tx?.cancel(onErrorReason)
workLiveData.removeObserver(this) workLiveData.removeObserver(this)
} }
WorkInfo.State.SUCCEEDED -> { WorkInfo.State.SUCCEEDED -> {
if (SendVerificationMessageWorker.hasFailed(wInfo.outputData)) { if (SessionSafeCoroutineWorker.hasFailed(wInfo.outputData)) {
Timber.e("## SAS verification [${tx?.transactionId}] failed to send verification message in state : ${tx?.state}") Timber.e("## SAS verification [${tx?.transactionId}] failed to send verification message in state : ${tx?.state}")
tx?.cancel(onErrorReason) tx?.cancel(onErrorReason)
} else { } else {
@ -210,7 +210,7 @@ internal class VerificationTransportRoomMessage(
?.filter { it.state == WorkInfo.State.SUCCEEDED } ?.filter { it.state == WorkInfo.State.SUCCEEDED }
?.firstOrNull { it.id == workRequest.id } ?.firstOrNull { it.id == workRequest.id }
?.let { wInfo -> ?.let { wInfo ->
if (SendVerificationMessageWorker.hasFailed(wInfo.outputData)) { if (SessionSafeCoroutineWorker.hasFailed(wInfo.outputData)) {
callback(null, null) callback(null, null)
} else { } else {
val eventId = wInfo.outputData.getString(localId) val eventId = wInfo.outputData.getString(localId)

38
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/content/UploadContentWorker.kt
View File

@ -19,7 +19,6 @@ package org.matrix.android.sdk.internal.session.content
import android.content.Context import android.content.Context
import android.graphics.BitmapFactory import android.graphics.BitmapFactory
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.api.extensions.tryOrNull import org.matrix.android.sdk.api.extensions.tryOrNull
@ -37,13 +36,14 @@ import org.matrix.android.sdk.internal.database.mapper.ContentMapper
import org.matrix.android.sdk.internal.database.mapper.asDomain import org.matrix.android.sdk.internal.database.mapper.asDomain
import org.matrix.android.sdk.internal.network.ProgressRequestBody import org.matrix.android.sdk.internal.network.ProgressRequestBody
import org.matrix.android.sdk.internal.session.DefaultFileService import org.matrix.android.sdk.internal.session.DefaultFileService
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.session.room.send.CancelSendTracker import org.matrix.android.sdk.internal.session.room.send.CancelSendTracker
import org.matrix.android.sdk.internal.session.room.send.LocalEchoIdentifiers import org.matrix.android.sdk.internal.session.room.send.LocalEchoIdentifiers
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
import org.matrix.android.sdk.internal.session.room.send.MultipleEventSendingDispatcherWorker import org.matrix.android.sdk.internal.session.room.send.MultipleEventSendingDispatcherWorker
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import timber.log.Timber import timber.log.Timber
import java.io.File import java.io.File
import java.util.UUID import java.util.UUID
@ -59,7 +59,8 @@ private data class NewImageAttributes(
* Possible previous worker: None * Possible previous worker: None
* Possible next worker : Always [MultipleEventSendingDispatcherWorker] * Possible next worker : Always [MultipleEventSendingDispatcherWorker]
*/ */
internal class UploadContentWorker(val context: Context, params: WorkerParameters) : CoroutineWorker(context, params) { internal class UploadContentWorker(val context: Context, params: WorkerParameters)
: SessionSafeCoroutineWorker<UploadContentWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -78,19 +79,12 @@ internal class UploadContentWorker(val context: Context, params: WorkerParameter
@Inject lateinit var imageCompressor: ImageCompressor @Inject lateinit var imageCompressor: ImageCompressor
@Inject lateinit var localEchoRepository: LocalEchoRepository @Inject lateinit var localEchoRepository: LocalEchoRepository
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val params = WorkerParamsFactory.fromData<Params>(inputData) injector.inject(this)
?: return Result.success()
.also { Timber.e("Unable to parse work parameters") }
Timber.v("Starting upload media work with params $params")
if (params.lastFailureMessage != null) {
// Transmit the error
return Result.success(inputData)
.also { Timber.e("Work cancelled due to input error from parent") }
} }
override suspend fun doSafeWork(params: Params): Result {
Timber.v("Starting upload media work with params $params")
// Just defensive code to ensure that we never have an uncaught exception that could break the queue // Just defensive code to ensure that we never have an uncaught exception that could break the queue
return try { return try {
internalDoWork(params) internalDoWork(params)
@ -100,10 +94,11 @@ internal class UploadContentWorker(val context: Context, params: WorkerParameter
} }
} }
private suspend fun internalDoWork(params: Params): Result { override fun buildErrorParams(params: Params, message: String): Params {
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success() return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
sessionComponent.inject(this) }
private suspend fun internalDoWork(params: Params): Result {
val allCancelled = params.localEchoIds.all { cancelSendTracker.isCancelRequestedFor(it.eventId, it.roomId) } val allCancelled = params.localEchoIds.all { cancelSendTracker.isCancelRequestedFor(it.eventId, it.roomId) }
if (allCancelled) { if (allCancelled) {
// there is no point in uploading the image! // there is no point in uploading the image!
@ -218,14 +213,7 @@ internal class UploadContentWorker(val context: Context, params: WorkerParameter
} }
} catch (e: Exception) { } catch (e: Exception) {
Timber.e(e, "## FileService: ERROR") Timber.e(e, "## FileService: ERROR")
notifyTracker(params) { contentUploadStateTracker.setFailure(it, e) } return handleFailure(params, e)
return Result.success(
WorkerParamsFactory.toData(
params.copy(
lastFailureMessage = e.localizedMessage
)
)
)
} finally { } finally {
// Delete all temporary files // Delete all temporary files
filesToDelete.forEach { filesToDelete.forEach {

21
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/group/GetGroupDataWorker.kt
View File

@ -18,19 +18,19 @@
package org.matrix.android.sdk.internal.session.group package org.matrix.android.sdk.internal.session.group
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import javax.inject.Inject import javax.inject.Inject
/** /**
* Possible previous worker: None * Possible previous worker: None
* Possible next worker : None * Possible next worker : None
*/ */
internal class GetGroupDataWorker(context: Context, params: WorkerParameters) : CoroutineWorker(context, params) { internal class GetGroupDataWorker(context: Context, params: WorkerParameters)
: SessionSafeCoroutineWorker<GetGroupDataWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -40,12 +40,11 @@ internal class GetGroupDataWorker(context: Context, params: WorkerParameters) :
@Inject lateinit var getGroupDataTask: GetGroupDataTask @Inject lateinit var getGroupDataTask: GetGroupDataTask
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val params = WorkerParamsFactory.fromData<Params>(inputData) injector.inject(this)
?: return Result.failure() }
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success() override suspend fun doSafeWork(params: Params): Result {
sessionComponent.inject(this)
return runCatching { return runCatching {
getGroupDataTask.execute(GetGroupDataTask.Params.FetchAllActive) getGroupDataTask.execute(GetGroupDataTask.Params.FetchAllActive)
}.fold( }.fold(
@ -53,4 +52,8 @@ internal class GetGroupDataWorker(context: Context, params: WorkerParameters) :
{ Result.retry() } { Result.retry() }
) )
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
} }

25
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/pushers/AddHttpPusherWorker.kt
View File

@ -17,10 +17,10 @@
package org.matrix.android.sdk.internal.session.pushers package org.matrix.android.sdk.internal.session.pushers
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import com.zhuinden.monarchy.Monarchy import com.zhuinden.monarchy.Monarchy
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.api.failure.Failure import org.matrix.android.sdk.api.failure.Failure
import org.matrix.android.sdk.api.session.pushers.PusherState import org.matrix.android.sdk.api.session.pushers.PusherState
import org.matrix.android.sdk.internal.database.mapper.toEntity import org.matrix.android.sdk.internal.database.mapper.toEntity
@ -28,16 +28,14 @@ import org.matrix.android.sdk.internal.database.model.PusherEntity
import org.matrix.android.sdk.internal.database.query.where import org.matrix.android.sdk.internal.database.query.where
import org.matrix.android.sdk.internal.di.SessionDatabase import org.matrix.android.sdk.internal.di.SessionDatabase
import org.matrix.android.sdk.internal.network.executeRequest import org.matrix.android.sdk.internal.network.executeRequest
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.util.awaitTransaction import org.matrix.android.sdk.internal.util.awaitTransaction
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import org.greenrobot.eventbus.EventBus
import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
internal class AddHttpPusherWorker(context: Context, params: WorkerParameters) internal class AddHttpPusherWorker(context: Context, params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<AddHttpPusherWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -50,14 +48,11 @@ internal class AddHttpPusherWorker(context: Context, params: WorkerParameters)
@Inject @SessionDatabase lateinit var monarchy: Monarchy @Inject @SessionDatabase lateinit var monarchy: Monarchy
@Inject lateinit var eventBus: EventBus @Inject lateinit var eventBus: EventBus
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val params = WorkerParamsFactory.fromData<Params>(inputData) injector.inject(this)
?: return Result.failure() }
.also { Timber.e("Unable to parse work parameters") }
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success()
sessionComponent.inject(this)
override suspend fun doSafeWork(params: Params): Result {
val pusher = params.pusher val pusher = params.pusher
if (pusher.pushKey.isBlank()) { if (pusher.pushKey.isBlank()) {
@ -82,6 +77,10 @@ internal class AddHttpPusherWorker(context: Context, params: WorkerParameters)
} }
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
private suspend fun setPusher(pusher: JsonPusher) { private suspend fun setPusher(pusher: JsonPusher) {
executeRequest<Unit>(eventBus) { executeRequest<Unit>(eventBus) {
apiCall = pushersAPI.setPusher(pusher) apiCall = pushersAPI.setPusher(pusher)

27
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/relation/SendRelationWorker.kt
View File

@ -17,7 +17,6 @@
package org.matrix.android.sdk.internal.session.room.relation package org.matrix.android.sdk.internal.session.room.relation
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.greenrobot.eventbus.EventBus import org.greenrobot.eventbus.EventBus
@ -27,17 +26,17 @@ import org.matrix.android.sdk.api.session.events.model.toModel
import org.matrix.android.sdk.api.session.room.model.relation.ReactionContent import org.matrix.android.sdk.api.session.room.model.relation.ReactionContent
import org.matrix.android.sdk.api.session.room.model.relation.ReactionInfo import org.matrix.android.sdk.api.session.room.model.relation.ReactionInfo
import org.matrix.android.sdk.internal.network.executeRequest import org.matrix.android.sdk.internal.network.executeRequest
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.session.room.RoomAPI import org.matrix.android.sdk.internal.session.room.RoomAPI
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
import org.matrix.android.sdk.internal.session.room.send.SendResponse import org.matrix.android.sdk.internal.session.room.send.SendResponse
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
// TODO This is not used. Delete? // TODO This is not used. Delete?
internal class SendRelationWorker(context: Context, params: WorkerParameters) : CoroutineWorker(context, params) { internal class SendRelationWorker(context: Context, params: WorkerParameters)
: SessionSafeCoroutineWorker<SendRelationWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -52,19 +51,11 @@ internal class SendRelationWorker(context: Context, params: WorkerParameters) :
@Inject lateinit var eventBus: EventBus @Inject lateinit var eventBus: EventBus
@Inject lateinit var localEchoRepository: LocalEchoRepository @Inject lateinit var localEchoRepository: LocalEchoRepository
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val params = WorkerParamsFactory.fromData<Params>(inputData) injector.inject(this)
?: return Result.failure()
if (params.lastFailureMessage != null) {
// Transmit the error
return Result.success(inputData)
.also { Timber.e("Work cancelled due to input error from parent") }
} }
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success() override suspend fun doSafeWork(params: Params): Result {
sessionComponent.inject(this)
val localEvent = localEchoRepository.getUpToDateEcho(params.eventId) val localEvent = localEchoRepository.getUpToDateEcho(params.eventId)
if (localEvent?.eventId == null) { if (localEvent?.eventId == null) {
return Result.failure() return Result.failure()
@ -89,6 +80,10 @@ internal class SendRelationWorker(context: Context, params: WorkerParameters) :
} }
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
private suspend fun sendRelation(roomId: String, relationType: String, relatedEventId: String, localEvent: Event) { private suspend fun sendRelation(roomId: String, relationType: String, relatedEventId: String, localEvent: Event) {
executeRequest<SendResponse>(eventBus) { executeRequest<SendResponse>(eventBus) {
apiCall = roomAPI.sendRelation( apiCall = roomAPI.sendRelation(

24
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/EncryptEventWorker.kt
View File

@ -18,7 +18,6 @@
package org.matrix.android.sdk.internal.session.room.send package org.matrix.android.sdk.internal.session.room.send
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.api.failure.Failure import org.matrix.android.sdk.api.failure.Failure
@ -31,10 +30,11 @@ import org.matrix.android.sdk.internal.crypto.MXCRYPTO_ALGORITHM_MEGOLM
import org.matrix.android.sdk.internal.crypto.MXEventDecryptionResult import org.matrix.android.sdk.internal.crypto.MXEventDecryptionResult
import org.matrix.android.sdk.internal.crypto.model.MXEncryptEventContentResult import org.matrix.android.sdk.internal.crypto.model.MXEncryptEventContentResult
import org.matrix.android.sdk.internal.database.mapper.ContentMapper import org.matrix.android.sdk.internal.database.mapper.ContentMapper
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.util.awaitCallback import org.matrix.android.sdk.internal.util.awaitCallback
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import timber.log.Timber import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
@ -43,7 +43,7 @@ import javax.inject.Inject
* Possible next worker : Always [SendEventWorker] * Possible next worker : Always [SendEventWorker]
*/ */
internal class EncryptEventWorker(context: Context, params: WorkerParameters) internal class EncryptEventWorker(context: Context, params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<EncryptEventWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -58,20 +58,12 @@ internal class EncryptEventWorker(context: Context, params: WorkerParameters)
@Inject lateinit var localEchoRepository: LocalEchoRepository @Inject lateinit var localEchoRepository: LocalEchoRepository
@Inject lateinit var cancelSendTracker: CancelSendTracker @Inject lateinit var cancelSendTracker: CancelSendTracker
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
Timber.v("Start Encrypt work") injector.inject(this)
val params = WorkerParamsFactory.fromData<Params>(inputData)
?: return Result.success()
if (params.lastFailureMessage != null) {
// Transmit the error
return Result.success(inputData)
.also { Timber.e("Work cancelled due to input error from parent") }
} }
override suspend fun doSafeWork(params: Params): Result {
Timber.v("## SendEvent: Start Encrypt work for event ${params.eventId}") Timber.v("## SendEvent: Start Encrypt work for event ${params.eventId}")
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success()
sessionComponent.inject(this)
val localEvent = localEchoRepository.getUpToDateEcho(params.eventId) val localEvent = localEchoRepository.getUpToDateEcho(params.eventId)
if (localEvent?.eventId == null) { if (localEvent?.eventId == null) {
@ -148,4 +140,8 @@ internal class EncryptEventWorker(context: Context, params: WorkerParameters)
return Result.success(WorkerParamsFactory.toData(nextWorkerParams)) return Result.success(WorkerParamsFactory.toData(nextWorkerParams))
} }
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
} }

32
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/MultipleEventSendingDispatcherWorker.kt
View File

@ -19,17 +19,17 @@ package org.matrix.android.sdk.internal.session.room.send
import android.content.Context import android.content.Context
import androidx.work.BackoffPolicy import androidx.work.BackoffPolicy
import androidx.work.CoroutineWorker
import androidx.work.OneTimeWorkRequest import androidx.work.OneTimeWorkRequest
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.api.session.room.send.SendState import org.matrix.android.sdk.api.session.room.send.SendState
import org.matrix.android.sdk.internal.di.WorkManagerProvider import org.matrix.android.sdk.internal.di.WorkManagerProvider
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.session.content.UploadContentWorker import org.matrix.android.sdk.internal.session.content.UploadContentWorker
import org.matrix.android.sdk.internal.session.room.timeline.TimelineSendEventWorkCommon import org.matrix.android.sdk.internal.session.room.timeline.TimelineSendEventWorkCommon
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import org.matrix.android.sdk.internal.worker.startChain import org.matrix.android.sdk.internal.worker.startChain
import timber.log.Timber import timber.log.Timber
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
@ -42,7 +42,7 @@ import javax.inject.Inject
* Possible next worker : None, but it will post new work to send events, encrypted or not * Possible next worker : None, but it will post new work to send events, encrypted or not
*/ */
internal class MultipleEventSendingDispatcherWorker(context: Context, params: WorkerParameters) internal class MultipleEventSendingDispatcherWorker(context: Context, params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<MultipleEventSendingDispatcherWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -56,22 +56,20 @@ internal class MultipleEventSendingDispatcherWorker(context: Context, params: Wo
@Inject lateinit var timelineSendEventWorkCommon: TimelineSendEventWorkCommon @Inject lateinit var timelineSendEventWorkCommon: TimelineSendEventWorkCommon
@Inject lateinit var localEchoRepository: LocalEchoRepository @Inject lateinit var localEchoRepository: LocalEchoRepository
override suspend fun doWork(): Result { override fun doOnError(params: Params): Result {
Timber.v("## SendEvent: Start dispatch sending multiple event work")
val params = WorkerParamsFactory.fromData<Params>(inputData)
?: return Result.success()
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success()
sessionComponent.inject(this)
if (params.lastFailureMessage != null) {
params.localEchoIds.forEach { localEchoIds -> params.localEchoIds.forEach { localEchoIds ->
localEchoRepository.updateSendState(localEchoIds.eventId, SendState.UNDELIVERED) localEchoRepository.updateSendState(localEchoIds.eventId, SendState.UNDELIVERED)
} }
// Transmit the error if needed?
return Result.success(inputData) return super.doOnError(params)
.also { Timber.e("## SendEvent: Work cancelled due to input error from parent ${params.lastFailureMessage}") }
} }
override fun injectWith(injector: SessionComponent) {
injector.inject(this)
}
override suspend fun doSafeWork(params: Params): Result {
Timber.v("## SendEvent: Start dispatch sending multiple event work")
// Create a work for every event // Create a work for every event
params.localEchoIds.forEach { localEchoIds -> params.localEchoIds.forEach { localEchoIds ->
val roomId = localEchoIds.roomId val roomId = localEchoIds.roomId
@ -94,6 +92,10 @@ internal class MultipleEventSendingDispatcherWorker(context: Context, params: Wo
return Result.success() return Result.success()
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
private fun createEncryptEventWork(sessionId: String, eventId: String, startChain: Boolean): OneTimeWorkRequest { private fun createEncryptEventWork(sessionId: String, eventId: String, startChain: Boolean): OneTimeWorkRequest {
val params = EncryptEventWorker.Params(sessionId, eventId) val params = EncryptEventWorker.Params(sessionId, eventId)
val sendWorkData = WorkerParamsFactory.toData(params) val sendWorkData = WorkerParamsFactory.toData(params)

28
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/RedactEventWorker.kt
View File

@ -17,24 +17,24 @@
package org.matrix.android.sdk.internal.session.room.send package org.matrix.android.sdk.internal.session.room.send
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.api.failure.Failure import org.matrix.android.sdk.api.failure.Failure
import org.matrix.android.sdk.internal.network.executeRequest import org.matrix.android.sdk.internal.network.executeRequest
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.session.room.RoomAPI import org.matrix.android.sdk.internal.session.room.RoomAPI
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import org.greenrobot.eventbus.EventBus
import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
/** /**
* Possible previous worker: None * Possible previous worker: None
* Possible next worker : None * Possible next worker : None
*/ */
internal class RedactEventWorker(context: Context, params: WorkerParameters) : CoroutineWorker(context, params) { internal class RedactEventWorker(context: Context, params: WorkerParameters)
: SessionSafeCoroutineWorker<RedactEventWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -49,19 +49,11 @@ internal class RedactEventWorker(context: Context, params: WorkerParameters) : C
@Inject lateinit var roomAPI: RoomAPI @Inject lateinit var roomAPI: RoomAPI
@Inject lateinit var eventBus: EventBus @Inject lateinit var eventBus: EventBus
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val params = WorkerParamsFactory.fromData<Params>(inputData) injector.inject(this)
?: return Result.failure()
if (params.lastFailureMessage != null) {
// Transmit the error
return Result.success(inputData)
.also { Timber.e("Work cancelled due to input error from parent") }
} }
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success() override suspend fun doSafeWork(params: Params): Result {
sessionComponent.inject(this)
val eventId = params.eventId val eventId = params.eventId
return runCatching { return runCatching {
executeRequest<SendResponse>(eventBus) { executeRequest<SendResponse>(eventBus) {
@ -90,4 +82,8 @@ internal class RedactEventWorker(context: Context, params: WorkerParameters) : C
} }
) )
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
} }

21
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/SendEventWorker.kt
View File

@ -18,7 +18,6 @@
package org.matrix.android.sdk.internal.session.room.send package org.matrix.android.sdk.internal.session.room.send
import android.content.Context import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import io.realm.RealmConfiguration import io.realm.RealmConfiguration
@ -28,10 +27,10 @@ import org.matrix.android.sdk.api.session.events.model.Content
import org.matrix.android.sdk.api.session.room.send.SendState import org.matrix.android.sdk.api.session.room.send.SendState
import org.matrix.android.sdk.internal.di.SessionDatabase import org.matrix.android.sdk.internal.di.SessionDatabase
import org.matrix.android.sdk.internal.network.executeRequest import org.matrix.android.sdk.internal.network.executeRequest
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.session.room.RoomAPI import org.matrix.android.sdk.internal.session.room.RoomAPI
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import timber.log.Timber import timber.log.Timber
import javax.inject.Inject import javax.inject.Inject
@ -43,7 +42,7 @@ import javax.inject.Inject
*/ */
internal class SendEventWorker(context: Context, internal class SendEventWorker(context: Context,
params: WorkerParameters) params: WorkerParameters)
: CoroutineWorker(context, params) { : SessionSafeCoroutineWorker<SendEventWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -58,13 +57,11 @@ internal class SendEventWorker(context: Context,
@Inject lateinit var cancelSendTracker: CancelSendTracker @Inject lateinit var cancelSendTracker: CancelSendTracker
@SessionDatabase @Inject lateinit var realmConfiguration: RealmConfiguration @SessionDatabase @Inject lateinit var realmConfiguration: RealmConfiguration
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
val params = WorkerParamsFactory.fromData<Params>(inputData) injector.inject(this)
?: return Result.success() }
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success()
sessionComponent.inject(this)
override suspend fun doSafeWork(params: Params): Result {
val event = localEchoRepository.getUpToDateEcho(params.eventId) val event = localEchoRepository.getUpToDateEcho(params.eventId)
if (event?.eventId == null || event.roomId == null) { if (event?.eventId == null || event.roomId == null) {
localEchoRepository.updateSendState(params.eventId, SendState.UNDELIVERED) localEchoRepository.updateSendState(params.eventId, SendState.UNDELIVERED)
@ -103,6 +100,10 @@ internal class SendEventWorker(context: Context,
} }
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
private suspend fun sendEvent(eventId: String, roomId: String, type: String, content: Content?) { private suspend fun sendEvent(eventId: String, roomId: String, type: String, content: Content?) {
localEchoRepository.updateSendState(eventId, SendState.SENDING) localEchoRepository.updateSendState(eventId, SendState.SENDING)
executeRequest<SendResponse>(eventBus) { executeRequest<SendResponse>(eventBus) {

2
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/SyncResponseHandler.kt
View File

@ -131,7 +131,7 @@ internal class SyncResponseHandler @Inject constructor(@SessionDatabase private
/** /**
* At the moment we don't get any group data through the sync, so we poll where every hour. * At the moment we don't get any group data through the sync, so we poll where every hour.
You can also force to refetch group data using [Group] API. * You can also force to refetch group data using [Group] API.
*/ */
private fun scheduleGroupDataFetchingIfNeeded(groupsSyncResponse: GroupsSyncResponse) { private fun scheduleGroupDataFetchingIfNeeded(groupsSyncResponse: GroupsSyncResponse) {
val groupIds = ArrayList<String>() val groupIds = ArrayList<String>()

23
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/sync/job/SyncWorker.kt
View File

@ -18,18 +18,18 @@ package org.matrix.android.sdk.internal.session.sync.job
import android.content.Context import android.content.Context
import androidx.work.BackoffPolicy import androidx.work.BackoffPolicy
import androidx.work.CoroutineWorker
import androidx.work.ExistingWorkPolicy import androidx.work.ExistingWorkPolicy
import androidx.work.WorkerParameters import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.api.failure.isTokenError import org.matrix.android.sdk.api.failure.isTokenError
import org.matrix.android.sdk.internal.di.WorkManagerProvider import org.matrix.android.sdk.internal.di.WorkManagerProvider
import org.matrix.android.sdk.internal.network.NetworkConnectivityChecker import org.matrix.android.sdk.internal.network.NetworkConnectivityChecker
import org.matrix.android.sdk.internal.session.SessionComponent
import org.matrix.android.sdk.internal.session.sync.SyncTask import org.matrix.android.sdk.internal.session.sync.SyncTask
import org.matrix.android.sdk.internal.task.TaskExecutor import org.matrix.android.sdk.internal.task.TaskExecutor
import org.matrix.android.sdk.internal.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.getSessionComponent
import timber.log.Timber import timber.log.Timber
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
import javax.inject.Inject import javax.inject.Inject
@ -43,7 +43,7 @@ private const val DEFAULT_DELAY_TIMEOUT = 30_000L
*/ */
internal class SyncWorker(context: Context, internal class SyncWorker(context: Context,
workerParameters: WorkerParameters workerParameters: WorkerParameters
) : CoroutineWorker(context, workerParameters) { ) : SessionSafeCoroutineWorker<SyncWorker.Params>(context, workerParameters, Params::class.java) {
@JsonClass(generateAdapter = true) @JsonClass(generateAdapter = true)
internal data class Params( internal data class Params(
@ -59,14 +59,13 @@ internal class SyncWorker(context: Context,
@Inject lateinit var networkConnectivityChecker: NetworkConnectivityChecker @Inject lateinit var networkConnectivityChecker: NetworkConnectivityChecker
@Inject lateinit var workManagerProvider: WorkManagerProvider @Inject lateinit var workManagerProvider: WorkManagerProvider
override suspend fun doWork(): Result { override fun injectWith(injector: SessionComponent) {
Timber.i("Sync work starting") injector.inject(this)
val params = WorkerParamsFactory.fromData<Params>(inputData) }
?: return Result.success()
.also { Timber.e("Unable to parse work parameters") } override suspend fun doSafeWork(params: Params): Result {
Timber.i("Sync work starting")
val sessionComponent = getSessionComponent(params.sessionId) ?: return Result.success()
sessionComponent.inject(this)
return runCatching { return runCatching {
doSync(params.timeout) doSync(params.timeout)
}.fold( }.fold(
@ -91,6 +90,10 @@ internal class SyncWorker(context: Context,
) )
} }
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
private suspend fun doSync(timeout: Long) { private suspend fun doSync(timeout: Long) {
val taskParams = SyncTask.Params(timeout * 1000) val taskParams = SyncTask.Params(timeout * 1000)
syncTask.execute(taskParams) syncTask.execute(taskParams)

102
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/worker/SessionSafeCoroutineWorker.kt Normal file
View File

@ -0,0 +1,102 @@
/*
* Copyright (c) 2020 New Vector Ltd
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.matrix.android.sdk.internal.worker
import android.content.Context
import androidx.annotation.CallSuper
import androidx.work.CoroutineWorker
import androidx.work.Data
import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.internal.session.SessionComponent
import timber.log.Timber
/**
* This worker should only sends Result.Success when added to a unique queue to avoid breaking the unique queue.
* This abstract class handle the cases of problem when parsing parameter, and forward the error if any to
* the next workers.
*/
internal abstract class SessionSafeCoroutineWorker<PARAM : SessionWorkerParams>(
context: Context,
workerParameters: WorkerParameters,
private val paramClass: Class<PARAM>
) : CoroutineWorker(context, workerParameters) {
@JsonClass(generateAdapter = true)
internal data class ErrorData(
override val sessionId: String,
override val lastFailureMessage: String? = null
) : SessionWorkerParams
final override suspend fun doWork(): Result {
val params = WorkerParamsFactory.fromData(paramClass, inputData)
?: return buildErrorResult(null, "Unable to parse work parameters")
.also { Timber.e("Unable to parse work parameters") }
return try {
val sessionComponent = getSessionComponent(params.sessionId)
?: return buildErrorResult(params, "No session")
// Make sure to inject before handling error as you may need some dependencies to process them.
injectWith(sessionComponent)
if (params.lastFailureMessage != null) {
// Forward error to the next workers
doOnError(params)
} else {
doSafeWork(params)
}
} catch (throwable: Throwable) {
buildErrorResult(params, throwable.localizedMessage ?: "error")
}
}
abstract fun injectWith(injector: SessionComponent)
/**
* Should only return Result.Success for workers added to a unique queue
*/
abstract suspend fun doSafeWork(params: PARAM): Result
protected fun buildErrorResult(params: PARAM?, message: String): Result {
return Result.success(
if (params != null) {
WorkerParamsFactory.toData(paramClass, buildErrorParams(params, message))
} else {
WorkerParamsFactory.toData(ErrorData::class.java, ErrorData(sessionId = "", lastFailureMessage = message))
}
)
}
abstract fun buildErrorParams(params: PARAM, message: String): PARAM
/**
* This is called when the input parameters are correct, but contain an error from the previous worker.
*/
@CallSuper
open fun doOnError(params: PARAM): Result {
// Forward the error
return Result.success(inputData)
.also { Timber.e("Work cancelled due to input error from parent") }
}
companion object {
fun hasFailed(outputData: Data): Boolean {
return WorkerParamsFactory.fromData(ErrorData::class.java, outputData)
.let { it?.lastFailureMessage != null }
}
}
}

17
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/worker/WorkerParamsFactory.kt
View File

@ -18,13 +18,14 @@
package org.matrix.android.sdk.internal.worker package org.matrix.android.sdk.internal.worker
import androidx.work.Data import androidx.work.Data
import com.squareup.moshi.Moshi
import org.matrix.android.sdk.api.extensions.tryOrNull import org.matrix.android.sdk.api.extensions.tryOrNull
import org.matrix.android.sdk.internal.di.MoshiProvider import org.matrix.android.sdk.internal.di.MoshiProvider
import org.matrix.android.sdk.internal.network.parsing.CheckNumberType import org.matrix.android.sdk.internal.network.parsing.CheckNumberType
internal object WorkerParamsFactory { internal object WorkerParamsFactory {
val moshi by lazy { private val moshi: Moshi by lazy {
// We are adding the CheckNumberType as we are serializing/deserializing multiple time in a row // We are adding the CheckNumberType as we are serializing/deserializing multiple time in a row
// and we lost typing information doing so. // and we lost typing information doing so.
// We don't want this check to be done on all adapters, so we just add it here. // We don't want this check to be done on all adapters, so we just add it here.
@ -34,20 +35,24 @@ internal object WorkerParamsFactory {
.build() .build()
} }
const val KEY = "WORKER_PARAMS_JSON" private const val KEY = "WORKER_PARAMS_JSON"
inline fun <reified T> toData(params: T): Data { inline fun <reified T> toData(params: T) = toData(T::class.java, params)
val adapter = moshi.adapter(T::class.java)
fun <T> toData(clazz: Class<T>, params: T): Data {
val adapter = moshi.adapter(clazz)
val json = adapter.toJson(params) val json = adapter.toJson(params)
return Data.Builder().putString(KEY, json).build() return Data.Builder().putString(KEY, json).build()
} }
inline fun <reified T> fromData(data: Data): T? = tryOrNull("Unable to parse work parameters") { inline fun <reified T> fromData(data: Data) = fromData(T::class.java, data)
fun <T> fromData(clazz: Class<T>, data: Data): T? = tryOrNull("Unable to parse work parameters") {
val json = data.getString(KEY) val json = data.getString(KEY)
return if (json == null) { return if (json == null) {
null null
} else { } else {
val adapter = moshi.adapter(T::class.java) val adapter = moshi.adapter(clazz)
adapter.fromJson(json) adapter.fromJson(json)
} }
} }