mirrors/SchildiChat-android
1
0
Fork 0
mirror of https://github.com/SchildiChat/SchildiChat-android.git synced 2024-11-26 11:26:01 +03:00
Code Issues Projects Releases Packages Wiki Activity

Add event sender in app thread

Browse source
This commit is contained in:
Valere 2020-10-08 12:08:41 +02:00 committed by Benoit Marty
parent c05147a019
commit 47746d6997
31 changed files with 842 additions and 494 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
matrix-sdk-android/src/main/java/org/matrix/android/sdk/api/session/room/send/SendService.kt
View file

@ -123,11 +123,6 @@ interface SendService {
*/
fun deleteFailedEcho(localEcho: TimelineEvent)
/**
* Delete all the events in one of the sending states
*/
fun clearSendingQueue()
/**
* Cancel sending a specific event. It has to be in one of the sending states
*/

43
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/tasks/EncryptEventTask.kt
View file

@ -17,8 +17,13 @@ package org.matrix.android.sdk.internal.crypto.tasks
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.events.model.Event
import org.matrix.android.sdk.api.session.events.model.EventType
import org.matrix.android.sdk.api.session.events.model.toContent
import org.matrix.android.sdk.api.session.room.send.SendState
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.model.MXEncryptEventContentResult
import org.matrix.android.sdk.internal.database.mapper.ContentMapper
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
import org.matrix.android.sdk.internal.task.Task
import org.matrix.android.sdk.internal.util.awaitCallback
@ -38,13 +43,14 @@ internal class DefaultEncryptEventTask @Inject constructor(
private val localEchoRepository: LocalEchoRepository
) : EncryptEventTask {
override suspend fun execute(params: EncryptEventTask.Params): Event {
if (!params.crypto.isRoomEncrypted(params.roomId)) return params.event
// don't want to wait for any query
//if (!params.crypto.isRoomEncrypted(params.roomId)) return params.event
val localEvent = params.event
if (localEvent.eventId == null) {
throw IllegalArgumentException()
}
localEchoRepository.updateSendState(localEvent.eventId, SendState.ENCRYPTING)
localEchoRepository.updateSendState(localEvent.eventId, localEvent.roomId, SendState.ENCRYPTING)
val localMutableContent = localEvent.content?.toMutableMap() ?: mutableMapOf()
params.keepKeys?.forEach {
@ -52,6 +58,7 @@ internal class DefaultEncryptEventTask @Inject constructor(
}
// try {
// let it throws
awaitCallback<MXEncryptEventContentResult> {
params.crypto.encryptEventContent(localMutableContent, localEvent.type, params.roomId, it)
}.let { result ->
@ -63,18 +70,34 @@ internal class DefaultEncryptEventTask @Inject constructor(
}
}
val safeResult = result.copy(eventContent = modifiedContent)
// Better handling of local echo, to avoid decrypting transition on remote echo
// Should I only do it for text messages?
val decryptionLocalEcho = if (result.eventContent["algorithm"] == MXCRYPTO_ALGORITHM_MEGOLM) {
MXEventDecryptionResult(
clearEvent = Event(
type = localEvent.type,
content = localEvent.content,
roomId = localEvent.roomId
).toContent(),
forwardingCurve25519KeyChain = emptyList(),
senderCurve25519Key = result.eventContent["sender_key"] as? String,
claimedEd25519Key = params.crypto.getMyDevice().fingerprint()
)
} else {
null
}
localEchoRepository.updateEcho(localEvent.eventId) { _, localEcho ->
localEcho.type = EventType.ENCRYPTED
localEcho.content = ContentMapper.map(modifiedContent)
decryptionLocalEcho?.also {
localEcho.setDecryptionResult(it)
}
}
return localEvent.copy(
type = safeResult.eventType,
content = safeResult.eventContent
)
}
// } catch (throwable: Throwable) {
// val sendState = when (throwable) {
// is Failure.CryptoError -> SendState.FAILED_UNKNOWN_DEVICES
// else -> SendState.UNDELIVERED
// }
// localEchoUpdater.updateSendState(localEvent.eventId, sendState)
// throw throwable
// }
}
}

50
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/tasks/RedactEventTask.kt Normal file
View file

@ -0,0 +1,50 @@
/*
* Copyright (c) 2020 New Vector Ltd
* Copyright 2020 The Matrix.org Foundation C.I.C.
*
* 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.crypto.tasks
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.internal.network.executeRequest
import org.matrix.android.sdk.internal.session.room.RoomAPI
import org.matrix.android.sdk.internal.session.room.send.SendResponse
import org.matrix.android.sdk.internal.task.Task
import javax.inject.Inject
internal interface RedactEventTask : Task<RedactEventTask.Params, String> {
data class Params(
val txID: String,
val roomId: String,
val eventId: String,
val reason: String?
)
}
internal class DefaultRedactEventTask @Inject constructor(
private val roomAPI: RoomAPI,
private val eventBus: EventBus) : RedactEventTask {
override suspend fun execute(params: RedactEventTask.Params): String {
val executeRequest = executeRequest<SendResponse>(eventBus) {
apiCall = roomAPI.redactEvent(
txId = params.txID,
roomId = params.roomId,
eventId = params.eventId,
reason = if (params.reason == null) emptyMap() else mapOf("reason" to params.reason)
)
}
return executeRequest.eventId
}
}

34
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/tasks/SendEventTask.kt
View file

@ -15,6 +15,7 @@
*/
package org.matrix.android.sdk.internal.crypto.tasks
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.events.model.Event
import org.matrix.android.sdk.api.session.room.send.SendState
@ -23,12 +24,12 @@ 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.SendResponse
import org.matrix.android.sdk.internal.task.Task
import org.greenrobot.eventbus.EventBus
import javax.inject.Inject
internal interface SendEventTask : Task<SendEventTask.Params, String> {
data class Params(
val event: Event,
val encrypt: Boolean,
val cryptoService: CryptoService?
)
}
@ -40,11 +41,11 @@ internal class DefaultSendEventTask @Inject constructor(
private val eventBus: EventBus) : SendEventTask {
override suspend fun execute(params: SendEventTask.Params): String {
val event = handleEncryption(params)
val localId = event.eventId!!
try {
localEchoRepository.updateSendState(localId, SendState.SENDING)
val event = handleEncryption(params)
val localId = event.eventId!!
localEchoRepository.updateSendState(localId, params.event.roomId, SendState.SENDING)
val executeRequest = executeRequest<SendResponse>(eventBus) {
apiCall = roomAPI.send(
localId,
@ -53,26 +54,23 @@ internal class DefaultSendEventTask @Inject constructor(
eventType = event.type
)
}
localEchoRepository.updateSendState(localId, SendState.SENT)
localEchoRepository.updateSendState(localId, params.event.roomId, SendState.SENT)
return executeRequest.eventId
} catch (e: Throwable) {
localEchoRepository.updateSendState(localId, SendState.UNDELIVERED)
// localEchoRepository.updateSendState(params.event.eventId!!, SendState.UNDELIVERED)
throw e
}
}
@Throws
private suspend fun handleEncryption(params: SendEventTask.Params): Event {
if (params.cryptoService?.isRoomEncrypted(params.event.roomId ?: "") == true) {
try {
return encryptEventTask.execute(EncryptEventTask.Params(
params.event.roomId ?: "",
params.event,
listOf("m.relates_to"),
params.cryptoService
))
} catch (throwable: Throwable) {
// We said it's ok to send verification request in clear
}
if (params.encrypt && !params.event.isEncrypted()) {
return encryptEventTask.execute(EncryptEventTask.Params(
params.event.roomId ?: "",
params.event,
listOf("m.relates_to"),
params.cryptoService!!
))
}
return params.event
}

6
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/crypto/tasks/SendVerificationMessageTask.kt
View file

@ -44,7 +44,7 @@ internal class DefaultSendVerificationMessageTask @Inject constructor(
val localId = event.eventId!!
try {
localEchoRepository.updateSendState(localId, SendState.SENDING)
localEchoRepository.updateSendState(localId, event.roomId, SendState.SENDING)
val executeRequest = executeRequest<SendResponse>(eventBus) {
apiCall = roomAPI.send(
localId,
@ -53,10 +53,10 @@ internal class DefaultSendVerificationMessageTask @Inject constructor(
eventType = event.type
)
}
localEchoRepository.updateSendState(localId, SendState.SENT)
localEchoRepository.updateSendState(localId, event.roomId, SendState.SENT)
return executeRequest.eventId
} catch (e: Throwable) {
localEchoRepository.updateSendState(localId, SendState.UNDELIVERED)
localEchoRepository.updateSendState(localId, event.roomId, SendState.UNDELIVERED)
throw e
}
}

5
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/DefaultSession.kt
View file

@ -64,6 +64,7 @@ import org.matrix.android.sdk.internal.di.SessionId
import org.matrix.android.sdk.internal.di.UnauthenticatedWithCertificate
import org.matrix.android.sdk.internal.di.WorkManagerProvider
import org.matrix.android.sdk.internal.session.identity.DefaultIdentityService
import org.matrix.android.sdk.internal.session.room.send.queue.EventSenderProcessor
import org.matrix.android.sdk.internal.session.room.timeline.TimelineEventDecryptor
import org.matrix.android.sdk.internal.session.sync.SyncTokenStore
import org.matrix.android.sdk.internal.session.sync.job.SyncThread
@ -121,7 +122,8 @@ internal class DefaultSession @Inject constructor(
private val taskExecutor: TaskExecutor,
private val callSignalingService: Lazy<CallSignalingService>,
@UnauthenticatedWithCertificate
private val unauthenticatedWithCertificateOkHttpClient: Lazy<OkHttpClient>
private val unauthenticatedWithCertificateOkHttpClient: Lazy<OkHttpClient>,
private val eventSenderProcessor: EventSenderProcessor
) : Session,
RoomService by roomService.get(),
RoomDirectoryService by roomDirectoryService.get(),
@ -161,6 +163,7 @@ internal class DefaultSession @Inject constructor(
}
eventBus.register(this)
timelineEventDecryptor.start()
eventSenderProcessor.start()
}
override fun requireBackgroundSync() {

3
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/SessionComponent.kt
View file

@ -45,7 +45,6 @@ import org.matrix.android.sdk.internal.session.pushers.AddHttpPusherWorker
import org.matrix.android.sdk.internal.session.pushers.PushersModule
import org.matrix.android.sdk.internal.session.room.RoomModule
import org.matrix.android.sdk.internal.session.room.relation.SendRelationWorker
import org.matrix.android.sdk.internal.session.room.send.EncryptEventWorker
import org.matrix.android.sdk.internal.session.room.send.MultipleEventSendingDispatcherWorker
import org.matrix.android.sdk.internal.session.room.send.RedactEventWorker
import org.matrix.android.sdk.internal.session.room.send.SendEventWorker
@ -109,8 +108,6 @@ internal interface SessionComponent {
fun inject(worker: SendRelationWorker)
fun inject(worker: EncryptEventWorker)
fun inject(worker: MultipleEventSendingDispatcherWorker)
fun inject(worker: RedactEventWorker)

5
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/SessionModule.kt
View file

@ -43,6 +43,8 @@ import org.matrix.android.sdk.api.session.securestorage.SharedSecretStorageServi
import org.matrix.android.sdk.api.session.typing.TypingUsersTracker
import org.matrix.android.sdk.internal.crypto.crosssigning.ShieldTrustUpdater
import org.matrix.android.sdk.internal.crypto.secrets.DefaultSharedSecretStorageService
import org.matrix.android.sdk.internal.crypto.tasks.DefaultRedactEventTask
import org.matrix.android.sdk.internal.crypto.tasks.RedactEventTask
import org.matrix.android.sdk.internal.crypto.verification.VerificationMessageProcessor
import org.matrix.android.sdk.internal.database.DatabaseCleaner
import org.matrix.android.sdk.internal.database.EventInsertLiveObserver
@ -367,4 +369,7 @@ internal abstract class SessionModule {
@Binds
abstract fun bindTypingUsersTracker(tracker: DefaultTypingUsersTracker): TypingUsersTracker
@Binds
abstract fun bindRedactEventTask(task: DefaultRedactEventTask): RedactEventTask
}

8
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/call/DefaultCallSignalingService.kt
View file

@ -36,8 +36,8 @@ import org.matrix.android.sdk.api.util.NoOpCancellable
import org.matrix.android.sdk.internal.di.UserId
import org.matrix.android.sdk.internal.session.SessionScope
import org.matrix.android.sdk.internal.session.call.model.MxCallImpl
import org.matrix.android.sdk.internal.session.room.send.queue.EventSenderProcessor
import org.matrix.android.sdk.internal.session.room.send.LocalEchoEventFactory
import org.matrix.android.sdk.internal.session.room.send.RoomEventSender
import org.matrix.android.sdk.internal.task.TaskExecutor
import org.matrix.android.sdk.internal.task.configureWith
import timber.log.Timber
@ -50,7 +50,7 @@ internal class DefaultCallSignalingService @Inject constructor(
private val userId: String,
private val activeCallHandler: ActiveCallHandler,
private val localEchoEventFactory: LocalEchoEventFactory,
private val roomEventSender: RoomEventSender,
private val eventSenderProcessor: EventSenderProcessor,
private val taskExecutor: TaskExecutor,
private val turnServerTask: GetTurnServerTask
) : CallSignalingService {
@ -103,7 +103,7 @@ internal class DefaultCallSignalingService @Inject constructor(
otherUserId = otherUserId,
isVideoCall = isVideoCall,
localEchoEventFactory = localEchoEventFactory,
roomEventSender = roomEventSender
eventSenderProcessor = eventSenderProcessor
)
activeCallHandler.addCall(call).also {
return call
@ -165,7 +165,7 @@ internal class DefaultCallSignalingService @Inject constructor(
otherUserId = event.senderId ?: return@let,
isVideoCall = content.isVideo(),
localEchoEventFactory = localEchoEventFactory,
roomEventSender = roomEventSender
eventSenderProcessor = eventSenderProcessor
)
activeCallHandler.addCall(incomingCall)
onCallInvite(incomingCall, content)

12
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/call/model/MxCallImpl.kt
View file

@ -29,8 +29,8 @@ import org.matrix.android.sdk.api.session.room.model.call.CallCandidatesContent
import org.matrix.android.sdk.api.session.room.model.call.CallHangupContent
import org.matrix.android.sdk.api.session.room.model.call.CallInviteContent
import org.matrix.android.sdk.internal.session.call.DefaultCallSignalingService
import org.matrix.android.sdk.internal.session.room.send.queue.EventSenderProcessor
import org.matrix.android.sdk.internal.session.room.send.LocalEchoEventFactory
import org.matrix.android.sdk.internal.session.room.send.RoomEventSender
import org.webrtc.IceCandidate
import org.webrtc.SessionDescription
import timber.log.Timber
@ -43,7 +43,7 @@ internal class MxCallImpl(
override val otherUserId: String,
override val isVideoCall: Boolean,
private val localEchoEventFactory: LocalEchoEventFactory,
private val roomEventSender: RoomEventSender
private val eventSenderProcessor: EventSenderProcessor
) : MxCall {
override var state: CallState = CallState.Idle
@ -91,7 +91,7 @@ internal class MxCallImpl(
offer = CallInviteContent.Offer(sdp = sdp.description)
)
.let { createEventAndLocalEcho(type = EventType.CALL_INVITE, roomId = roomId, content = it.toContent()) }
.also { roomEventSender.sendEvent(it) }
.also { eventSenderProcessor.postEvent(it) }
}
override fun sendLocalIceCandidates(candidates: List<IceCandidate>) {
@ -106,7 +106,7 @@ internal class MxCallImpl(
}
)
.let { createEventAndLocalEcho(type = EventType.CALL_CANDIDATES, roomId = roomId, content = it.toContent()) }
.also { roomEventSender.sendEvent(it) }
.also { eventSenderProcessor.postEvent(it) }
}
override fun sendLocalIceCandidateRemovals(candidates: List<IceCandidate>) {
@ -119,7 +119,7 @@ internal class MxCallImpl(
callId = callId
)
.let { createEventAndLocalEcho(type = EventType.CALL_HANGUP, roomId = roomId, content = it.toContent()) }
.also { roomEventSender.sendEvent(it) }
.also { eventSenderProcessor.postEvent(it) }
state = CallState.Terminated
}
@ -132,7 +132,7 @@ internal class MxCallImpl(
answer = CallAnswerContent.Answer(sdp = sdp.description)
)
.let { createEventAndLocalEcho(type = EventType.CALL_ANSWER, roomId = roomId, content = it.toContent()) }
.also { roomEventSender.sendEvent(it) }
.also { eventSenderProcessor.postEvent(it) }
}
private fun createEventAndLocalEcho(localId: String = LocalEcho.createLocalEchoId(), type: String, roomId: String, content: Content): Event {

69
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/relation/DefaultRelationService.kt
View file

@ -17,7 +17,6 @@ package org.matrix.android.sdk.internal.session.room.relation
import androidx.lifecycle.LiveData
import androidx.lifecycle.Transformations
import androidx.work.OneTimeWorkRequest
import com.squareup.inject.assisted.Assisted
import com.squareup.inject.assisted.AssistedInject
import com.zhuinden.monarchy.Monarchy
@ -39,21 +38,18 @@ import org.matrix.android.sdk.internal.database.model.TimelineEventEntity
import org.matrix.android.sdk.internal.database.query.where
import org.matrix.android.sdk.internal.di.SessionDatabase
import org.matrix.android.sdk.internal.di.SessionId
import org.matrix.android.sdk.internal.session.room.send.EncryptEventWorker
import org.matrix.android.sdk.internal.session.room.send.LocalEchoEventFactory
import org.matrix.android.sdk.internal.session.room.send.RedactEventWorker
import org.matrix.android.sdk.internal.session.room.send.SendEventWorker
import org.matrix.android.sdk.internal.session.room.timeline.TimelineSendEventWorkCommon
import org.matrix.android.sdk.internal.session.room.send.queue.EventSenderProcessor
import org.matrix.android.sdk.internal.task.TaskExecutor
import org.matrix.android.sdk.internal.task.configureWith
import org.matrix.android.sdk.internal.util.fetchCopyMap
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import timber.log.Timber
internal class DefaultRelationService @AssistedInject constructor(
@Assisted private val roomId: String,
@SessionId private val sessionId: String,
private val timeLineSendEventWorkCommon: TimelineSendEventWorkCommon,
// private val timeLineSendEventWorkCommon: TimelineSendEventWorkCommon,
private val eventSenderProcessor: EventSenderProcessor,
private val eventFactory: LocalEchoEventFactory,
private val cryptoService: CryptoService,
private val findReactionEventForUndoTask: FindReactionEventForUndoTask,
@ -83,8 +79,7 @@ internal class DefaultRelationService @AssistedInject constructor(
.none { it.addedByMe && it.key == reaction }) {
val event = eventFactory.createReactionEvent(roomId, targetEventId, reaction)
.also { saveLocalEcho(it) }
val sendRelationWork = createSendEventWork(event, true)
timeLineSendEventWorkCommon.postWork(roomId, sendRelationWork)
return eventSenderProcessor.postEvent(event, false /* reaction are not encrypted*/)
} else {
Timber.w("Reaction already added")
NoOpCancellable
@ -107,9 +102,7 @@ internal class DefaultRelationService @AssistedInject constructor(
data.redactEventId?.let { toRedact ->
val redactEvent = eventFactory.createRedactEvent(roomId, toRedact, null)
.also { saveLocalEcho(it) }
val redactWork = createRedactEventWork(redactEvent, toRedact, null)
timeLineSendEventWorkCommon.postWork(roomId, redactWork)
eventSenderProcessor.postRedaction(redactEvent, null)
}
}
}
@ -121,18 +114,6 @@ internal class DefaultRelationService @AssistedInject constructor(
.executeBy(taskExecutor)
}
// TODO duplicate with send service?
private fun createRedactEventWork(localEvent: Event, eventId: String, reason: String?): OneTimeWorkRequest {
val sendContentWorkerParams = RedactEventWorker.Params(
sessionId,
localEvent.eventId!!,
roomId,
eventId,
reason)
val redactWorkData = WorkerParamsFactory.toData(sendContentWorkerParams)
return timeLineSendEventWorkCommon.createWork<RedactEventWorker>(redactWorkData, true)
}
override fun editTextMessage(targetEventId: String,
msgType: String,
newBodyText: CharSequence,
@ -141,14 +122,7 @@ internal class DefaultRelationService @AssistedInject constructor(
val event = eventFactory
.createReplaceTextEvent(roomId, targetEventId, newBodyText, newBodyAutoMarkdown, msgType, compatibilityBodyText)
.also { saveLocalEcho(it) }
return if (cryptoService.isRoomEncrypted(roomId)) {
val encryptWork = createEncryptEventWork(event, listOf("m.relates_to"))
val workRequest = createSendEventWork(event, false)
timeLineSendEventWorkCommon.postSequentialWorks(roomId, encryptWork, workRequest)
} else {
val workRequest = createSendEventWork(event, true)
timeLineSendEventWorkCommon.postWork(roomId, workRequest)
}
return eventSenderProcessor.postEvent(event, cryptoService.isRoomEncrypted(roomId))
}
override fun editReply(replyToEdit: TimelineEvent,
@ -165,14 +139,7 @@ internal class DefaultRelationService @AssistedInject constructor(
compatibilityBodyText
)
.also { saveLocalEcho(it) }
return if (cryptoService.isRoomEncrypted(roomId)) {
val encryptWork = createEncryptEventWork(event, listOf("m.relates_to"))
val workRequest = createSendEventWork(event, false)
timeLineSendEventWorkCommon.postSequentialWorks(roomId, encryptWork, workRequest)
} else {
val workRequest = createSendEventWork(event, true)
timeLineSendEventWorkCommon.postWork(roomId, workRequest)
}
return eventSenderProcessor.postEvent(event, cryptoService.isRoomEncrypted(roomId))
}
override fun fetchEditHistory(eventId: String, callback: MatrixCallback<List<Event>>) {
@ -189,27 +156,7 @@ internal class DefaultRelationService @AssistedInject constructor(
?.also { saveLocalEcho(it) }
?: return null
return if (cryptoService.isRoomEncrypted(roomId)) {
val encryptWork = createEncryptEventWork(event, listOf("m.relates_to"))
val workRequest = createSendEventWork(event, false)
timeLineSendEventWorkCommon.postSequentialWorks(roomId, encryptWork, workRequest)
} else {
val workRequest = createSendEventWork(event, true)
timeLineSendEventWorkCommon.postWork(roomId, workRequest)
}
}
private fun createEncryptEventWork(event: Event, keepKeys: List<String>?): OneTimeWorkRequest {
// Same parameter
val params = EncryptEventWorker.Params(sessionId, event.eventId!!, keepKeys)
val sendWorkData = WorkerParamsFactory.toData(params)
return timeLineSendEventWorkCommon.createWork<EncryptEventWorker>(sendWorkData, true)
}
private fun createSendEventWork(event: Event, startChain: Boolean): OneTimeWorkRequest {
val sendContentWorkerParams = SendEventWorker.Params(sessionId = sessionId, eventId = event.eventId!!)
val sendWorkData = WorkerParamsFactory.toData(sendContentWorkerParams)
return timeLineSendEventWorkCommon.createWork<SendEventWorker>(sendWorkData, startChain)
return eventSenderProcessor.postEvent(event, cryptoService.isRoomEncrypted(roomId))
}
override fun getEventAnnotationsSummary(eventId: String): EventAnnotationsSummary? {

86
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/DefaultSendService.kt
View file

@ -48,10 +48,9 @@ import org.matrix.android.sdk.api.util.NoOpCancellable
import org.matrix.android.sdk.internal.di.SessionId
import org.matrix.android.sdk.internal.di.WorkManagerProvider
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.send.queue.EventSenderProcessor
import org.matrix.android.sdk.internal.task.TaskExecutor
import org.matrix.android.sdk.internal.util.CancelableWork
import org.matrix.android.sdk.internal.worker.AlwaysSuccessfulWorker
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.startChain
import timber.log.Timber
@ -63,13 +62,12 @@ private const val UPLOAD_WORK = "UPLOAD_WORK"
internal class DefaultSendService @AssistedInject constructor(
@Assisted private val roomId: String,
private val workManagerProvider: WorkManagerProvider,
private val timelineSendEventWorkCommon: TimelineSendEventWorkCommon,
@SessionId private val sessionId: String,
private val localEchoEventFactory: LocalEchoEventFactory,
private val cryptoService: CryptoService,
private val taskExecutor: TaskExecutor,
private val localEchoRepository: LocalEchoRepository,
private val roomEventSender: RoomEventSender,
private val eventSenderProcessor: EventSenderProcessor,
private val cancelSendTracker: CancelSendTracker
) : SendService {
@ -93,17 +91,6 @@ internal class DefaultSendService @AssistedInject constructor(
}
// For test only
private fun sendTextMessages(text: CharSequence, msgType: String, autoMarkdown: Boolean, times: Int): Cancelable {
return CancelableBag().apply {
// Send the event several times
repeat(times) { i ->
localEchoEventFactory.createTextEvent(roomId, msgType, "$text - $i", autoMarkdown)
.also { createLocalEcho(it) }
.let { sendEvent(it) }
.also { add(it) }
}
}
}
override fun sendFormattedTextMessage(text: String, formattedText: String, msgType: String): Cancelable {
return localEchoEventFactory.createFormattedTextEvent(roomId, TextContent(text, formattedText), msgType)
@ -133,13 +120,14 @@ internal class DefaultSendService @AssistedInject constructor(
override fun redactEvent(event: Event, reason: String?): Cancelable {
// TODO manage media/attachements?
return createRedactEventWork(event, reason)
.let { timelineSendEventWorkCommon.postWork(roomId, it) }
val redactionEcho = localEchoEventFactory.createRedactEvent(roomId, event.eventId!!, reason)
.also { createLocalEcho(it) }
return eventSenderProcessor.postRedaction(redactionEcho, reason)
}
override fun resendTextMessage(localEcho: TimelineEvent): Cancelable {
if (localEcho.root.isTextMessage() && localEcho.root.sendState.hasFailed()) {
localEchoRepository.updateSendState(localEcho.eventId, SendState.UNSENT)
localEchoRepository.updateSendState(localEcho.eventId, roomId, SendState.UNSENT)
return sendEvent(localEcho.root)
}
return NoOpCancellable
@ -153,7 +141,7 @@ internal class DefaultSendService @AssistedInject constructor(
val url = messageContent.getFileUrl() ?: return NoOpCancellable
if (url.startsWith("mxc://")) {
// We need to resend only the message as the attachment is ok
localEchoRepository.updateSendState(localEcho.eventId, SendState.UNSENT)
localEchoRepository.updateSendState(localEcho.eventId, roomId, SendState.UNSENT)
return sendEvent(localEcho.root)
}
@ -170,7 +158,7 @@ internal class DefaultSendService @AssistedInject constructor(
queryUri = Uri.parse(messageContent.url),
type = ContentAttachmentData.Type.IMAGE
)
localEchoRepository.updateSendState(localEcho.eventId, SendState.UNSENT)
localEchoRepository.updateSendState(localEcho.eventId, roomId, SendState.UNSENT)
internalSendMedia(listOf(localEcho.root), attachmentData, true)
}
is MessageVideoContent -> {
@ -184,7 +172,7 @@ internal class DefaultSendService @AssistedInject constructor(
queryUri = Uri.parse(messageContent.url),
type = ContentAttachmentData.Type.VIDEO
)
localEchoRepository.updateSendState(localEcho.eventId, SendState.UNSENT)
localEchoRepository.updateSendState(localEcho.eventId, roomId, SendState.UNSENT)
internalSendMedia(listOf(localEcho.root), attachmentData, true)
}
is MessageFileContent -> {
@ -195,7 +183,7 @@ internal class DefaultSendService @AssistedInject constructor(
queryUri = Uri.parse(messageContent.url),
type = ContentAttachmentData.Type.FILE
)
localEchoRepository.updateSendState(localEcho.eventId, SendState.UNSENT)
localEchoRepository.updateSendState(localEcho.eventId, roomId, SendState.UNSENT)
internalSendMedia(listOf(localEcho.root), attachmentData, true)
}
is MessageAudioContent -> {
@ -207,7 +195,7 @@ internal class DefaultSendService @AssistedInject constructor(
queryUri = Uri.parse(messageContent.url),
type = ContentAttachmentData.Type.AUDIO
)
localEchoRepository.updateSendState(localEcho.eventId, SendState.UNSENT)
localEchoRepository.updateSendState(localEcho.eventId, roomId, SendState.UNSENT)
internalSendMedia(listOf(localEcho.root), attachmentData, true)
}
else -> NoOpCancellable
@ -222,25 +210,6 @@ internal class DefaultSendService @AssistedInject constructor(
}
}
override fun clearSendingQueue() {
timelineSendEventWorkCommon.cancelAllWorks(roomId)
workManagerProvider.workManager.cancelUniqueWork(buildWorkName(UPLOAD_WORK))
// Replace the worker chains with a AlwaysSuccessfulWorker, to ensure the queues are well emptied
workManagerProvider.matrixOneTimeWorkRequestBuilder<AlwaysSuccessfulWorker>()
.build().let {
timelineSendEventWorkCommon.postWork(roomId, it, ExistingWorkPolicy.REPLACE)
// need to clear also image sending queue
workManagerProvider.workManager
.beginUniqueWork(buildWorkName(UPLOAD_WORK), ExistingWorkPolicy.REPLACE, it)
.enqueue()
}
taskExecutor.executorScope.launch {
localEchoRepository.clearSendingQueue(roomId)
}
}
override fun cancelSend(eventId: String) {
cancelSendTracker.markLocalEchoForCancel(eventId, roomId)
taskExecutor.executorScope.launch {
@ -262,13 +231,6 @@ internal class DefaultSendService @AssistedInject constructor(
}
}
// override fun failAllPendingMessages() {
// taskExecutor.executorScope.launch {
// val eventsToResend = localEchoRepository.getAllEventsWithStates(roomId, SendState.PENDING_STATES)
// localEchoRepository.updateSendState(roomId, eventsToResend.map { it.eventId }, SendState.UNDELIVERED)
// }
// }
override fun sendMedia(attachment: ContentAttachmentData,
compressBeforeSending: Boolean,
roomIds: Set<String>): Cancelable {
@ -301,7 +263,7 @@ internal class DefaultSendService @AssistedInject constructor(
val dispatcherWork = createMultipleEventDispatcherWork(isRoomEncrypted)
workManagerProvider.workManager
.beginUniqueWork(buildWorkName(UPLOAD_WORK), ExistingWorkPolicy.APPEND, uploadWork)
.beginUniqueWork(buildWorkName(UPLOAD_WORK), ExistingWorkPolicy.APPEND_OR_REPLACE, uploadWork)
.then(dispatcherWork)
.enqueue()
.also { operation ->
@ -322,7 +284,7 @@ internal class DefaultSendService @AssistedInject constructor(
}
private fun sendEvent(event: Event): Cancelable {
return roomEventSender.sendEvent(event)
return eventSenderProcessor.postEvent(event, cryptoService.isRoomEncrypted(event.roomId!!))
}
private fun createLocalEcho(event: Event) {
@ -333,28 +295,6 @@ internal class DefaultSendService @AssistedInject constructor(
return "${roomId}_$identifier"
}
private fun createEncryptEventWork(event: Event, startChain: Boolean): OneTimeWorkRequest {
// Same parameter
return EncryptEventWorker.Params(sessionId, event.eventId ?: "")
.let { WorkerParamsFactory.toData(it) }
.let {
workManagerProvider.matrixOneTimeWorkRequestBuilder<EncryptEventWorker>()
.setConstraints(WorkManagerProvider.workConstraints)
.setInputData(it)
.startChain(startChain)
.setBackoffCriteria(BackoffPolicy.LINEAR, WorkManagerProvider.BACKOFF_DELAY, TimeUnit.MILLISECONDS)
.build()
}
}
private fun createRedactEventWork(event: Event, reason: String?): OneTimeWorkRequest {
return localEchoEventFactory.createRedactEvent(roomId, event.eventId!!, reason)
.also { createLocalEcho(it) }
.let { RedactEventWorker.Params(sessionId, it.eventId!!, roomId, event.eventId, reason) }
.let { WorkerParamsFactory.toData(it) }
.let { timelineSendEventWorkCommon.createWork<RedactEventWorker>(it, true) }
}
private fun createUploadMediaWork(allLocalEchos: List<Event>,
attachment: ContentAttachmentData,
isRoomEncrypted: Boolean,

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

@ -1,146 +0,0 @@
/*
* Copyright 2020 The Matrix.org Foundation C.I.C.
*
* 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.session.room.send
import android.content.Context
import androidx.work.WorkerParameters
import com.squareup.moshi.JsonClass
import org.matrix.android.sdk.api.failure.Failure
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.events.model.Event
import org.matrix.android.sdk.api.session.events.model.EventType
import org.matrix.android.sdk.api.session.events.model.toContent
import org.matrix.android.sdk.api.session.room.send.SendState
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.model.MXEncryptEventContentResult
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.worker.SessionSafeCoroutineWorker
import org.matrix.android.sdk.internal.worker.SessionWorkerParams
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import timber.log.Timber
import javax.inject.Inject
/**
* Possible previous worker: None
* Possible next worker : Always [SendEventWorker]
*/
internal class EncryptEventWorker(context: Context, params: WorkerParameters)
: SessionSafeCoroutineWorker<EncryptEventWorker.Params>(context, params, Params::class.java) {
@JsonClass(generateAdapter = true)
internal data class Params(
override val sessionId: String,
val eventId: String,
/** Do not encrypt these keys, keep them as is in encrypted content (e.g. m.relates_to) */
val keepKeys: List<String>? = null,
override val lastFailureMessage: String? = null
) : SessionWorkerParams
@Inject lateinit var crypto: CryptoService
@Inject lateinit var localEchoRepository: LocalEchoRepository
@Inject lateinit var cancelSendTracker: CancelSendTracker
override fun injectWith(injector: SessionComponent) {
injector.inject(this)
}
override suspend fun doSafeWork(params: Params): Result {
Timber.v("## SendEvent: Start Encrypt work for event ${params.eventId}")
val localEvent = localEchoRepository.getUpToDateEcho(params.eventId)
if (localEvent?.eventId == null) {
return Result.success()
}
if (cancelSendTracker.isCancelRequestedFor(localEvent.eventId, localEvent.roomId)) {
return Result.success()
.also { Timber.e("## SendEvent: Event sending has been cancelled ${localEvent.eventId}") }
}
localEchoRepository.updateSendState(localEvent.eventId, SendState.ENCRYPTING)
val localMutableContent = localEvent.content?.toMutableMap() ?: mutableMapOf()
params.keepKeys?.forEach {
localMutableContent.remove(it)
}
var error: Throwable? = null
var result: MXEncryptEventContentResult? = null
try {
result = awaitCallback {
crypto.encryptEventContent(localMutableContent, localEvent.type, localEvent.roomId!!, it)
}
} catch (throwable: Throwable) {
error = throwable
}
if (result != null) {
val modifiedContent = HashMap(result.eventContent)
params.keepKeys?.forEach { toKeep ->
localEvent.content?.get(toKeep)?.let {
// put it back in the encrypted thing
modifiedContent[toKeep] = it
}
}
// Better handling of local echo, to avoid decrypting transition on remote echo
// Should I only do it for text messages?
val decryptionLocalEcho = if (result.eventContent["algorithm"] == MXCRYPTO_ALGORITHM_MEGOLM) {
MXEventDecryptionResult(
clearEvent = Event(
type = localEvent.type,
content = localEvent.content,
roomId = localEvent.roomId
).toContent(),
forwardingCurve25519KeyChain = emptyList(),
senderCurve25519Key = result.eventContent["sender_key"] as? String,
claimedEd25519Key = crypto.getMyDevice().fingerprint()
)
} else {
null
}
localEchoRepository.updateEcho(localEvent.eventId) { _, localEcho ->
localEcho.type = EventType.ENCRYPTED
localEcho.content = ContentMapper.map(modifiedContent)
decryptionLocalEcho?.also {
localEcho.setDecryptionResult(it)
}
}
val nextWorkerParams = SendEventWorker.Params(sessionId = params.sessionId, eventId = params.eventId)
return Result.success(WorkerParamsFactory.toData(nextWorkerParams))
} else {
val sendState = when (error) {
is Failure.CryptoError -> SendState.FAILED_UNKNOWN_DEVICES
else -> SendState.UNDELIVERED
}
localEchoRepository.updateSendState(localEvent.eventId, sendState)
// always return success, or the chain will be stuck for ever!
val nextWorkerParams = SendEventWorker.Params(
sessionId = params.sessionId,
eventId = localEvent.eventId,
lastFailureMessage = error?.localizedMessage ?: "Error"
)
return Result.success(WorkerParamsFactory.toData(nextWorkerParams))
}
}
override fun buildErrorParams(params: Params, message: String): Params {
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
}

11
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/LocalEchoRepository.kt
View file

@ -88,8 +88,9 @@ internal class LocalEchoRepository @Inject constructor(@SessionDatabase private
}
}
fun updateSendState(eventId: String, sendState: SendState) {
fun updateSendState(eventId: String, roomId: String?, sendState: SendState) {
Timber.v("## SendEvent: [${System.currentTimeMillis()}] Update local state of $eventId to ${sendState.name}")
eventBus.post(DefaultTimeline.OnLocalEchoUpdated(roomId ?: "", eventId, sendState))
updateEchoAsync(eventId) { realm, sendingEventEntity ->
if (sendState == SendState.SENT && sendingEventEntity.sendState == SendState.SYNCED) {
// If already synced, do not put as sent
@ -137,6 +138,14 @@ internal class LocalEchoRepository @Inject constructor(@SessionDatabase private
}
}
fun deleteFailedEchoAsync(roomId: String, eventId: String?) {
monarchy.runTransactionSync { realm ->
TimelineEventEntity.where(realm, roomId = roomId, eventId = eventId ?: "").findFirst()?.deleteFromRealm()
EventEntity.where(realm, eventId = eventId ?: "").findFirst()?.deleteFromRealm()
roomSummaryUpdater.updateSendingInformation(realm, roomId)
}
}
suspend fun clearSendingQueue(roomId: String) {
monarchy.awaitTransaction { realm ->
TimelineEventEntity

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

@ -57,7 +57,7 @@ internal class MultipleEventSendingDispatcherWorker(context: Context, params: Wo
override fun doOnError(params: Params): Result {
params.localEchoIds.forEach { localEchoIds ->
localEchoRepository.updateSendState(localEchoIds.eventId, SendState.UNDELIVERED)
localEchoRepository.updateSendState(localEchoIds.eventId, localEchoIds.roomId, SendState.UNDELIVERED)
}
return super.doOnError(params)
@ -73,19 +73,10 @@ internal class MultipleEventSendingDispatcherWorker(context: Context, params: Wo
params.localEchoIds.forEach { localEchoIds ->
val roomId = localEchoIds.roomId
val eventId = localEchoIds.eventId
if (params.isEncrypted) {
localEchoRepository.updateSendState(eventId, SendState.ENCRYPTING)
Timber.v("## SendEvent: [${System.currentTimeMillis()}] Schedule encrypt and send event $eventId")
val encryptWork = createEncryptEventWork(params.sessionId, eventId, true)
// Note that event will be replaced by the result of the previous work
val sendWork = createSendEventWork(params.sessionId, eventId, false)
timelineSendEventWorkCommon.postSequentialWorks(roomId, encryptWork, sendWork)
} else {
localEchoRepository.updateSendState(eventId, SendState.SENDING)
localEchoRepository.updateSendState(eventId, roomId, SendState.SENDING)
Timber.v("## SendEvent: [${System.currentTimeMillis()}] Schedule send event $eventId")
val sendWork = createSendEventWork(params.sessionId, eventId, true)
timelineSendEventWorkCommon.postWork(roomId, sendWork)
}
}
return Result.success()
@ -95,18 +86,6 @@ internal class MultipleEventSendingDispatcherWorker(context: Context, params: Wo
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
private fun createEncryptEventWork(sessionId: String, eventId: String, startChain: Boolean): OneTimeWorkRequest {
val params = EncryptEventWorker.Params(sessionId, eventId)
val sendWorkData = WorkerParamsFactory.toData(params)
return workManagerProvider.matrixOneTimeWorkRequestBuilder<EncryptEventWorker>()
.setConstraints(WorkManagerProvider.workConstraints)
.setInputData(sendWorkData)
.startChain(startChain)
.setBackoffCriteria(BackoffPolicy.LINEAR, WorkManagerProvider.BACKOFF_DELAY, TimeUnit.MILLISECONDS)
.build()
}
private fun createSendEventWork(sessionId: String, eventId: String, startChain: Boolean): OneTimeWorkRequest {
val sendContentWorkerParams = SendEventWorker.Params(sessionId = sessionId, eventId = eventId)
val sendWorkData = WorkerParamsFactory.toData(sendContentWorkerParams)

75
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/RoomEventSender.kt
View file

@ -1,75 +0,0 @@
/*
* Copyright 2020 The Matrix.org Foundation C.I.C.
*
* 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.session.room.send
import androidx.work.BackoffPolicy
import androidx.work.OneTimeWorkRequest
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.events.model.Event
import org.matrix.android.sdk.api.util.Cancelable
import org.matrix.android.sdk.internal.di.SessionId
import org.matrix.android.sdk.internal.di.WorkManagerProvider
import org.matrix.android.sdk.internal.session.room.timeline.TimelineSendEventWorkCommon
import org.matrix.android.sdk.internal.worker.WorkerParamsFactory
import org.matrix.android.sdk.internal.worker.startChain
import timber.log.Timber
import java.util.concurrent.TimeUnit
import javax.inject.Inject
internal class RoomEventSender @Inject constructor(
private val workManagerProvider: WorkManagerProvider,
private val timelineSendEventWorkCommon: TimelineSendEventWorkCommon,
@SessionId private val sessionId: String,
private val cryptoService: CryptoService
) {
fun sendEvent(event: Event): Cancelable {
// Encrypted room handling
return if (cryptoService.isRoomEncrypted(event.roomId ?: "")
&& !event.isEncrypted() // In case of resend where it's already encrypted so skip to send
) {
Timber.v("## SendEvent: [${System.currentTimeMillis()}] Schedule encrypt and send event ${event.eventId}")
val encryptWork = createEncryptEventWork(event, true)
// Note that event will be replaced by the result of the previous work
val sendWork = createSendEventWork(event, false)
timelineSendEventWorkCommon.postSequentialWorks(event.roomId ?: "", encryptWork, sendWork)
} else {
Timber.v("## SendEvent: [${System.currentTimeMillis()}] Schedule send event ${event.eventId}")
val sendWork = createSendEventWork(event, true)
timelineSendEventWorkCommon.postWork(event.roomId ?: "", sendWork)
}
}
private fun createEncryptEventWork(event: Event, startChain: Boolean): OneTimeWorkRequest {
// Same parameter
val params = EncryptEventWorker.Params(sessionId, event.eventId!!)
val sendWorkData = WorkerParamsFactory.toData(params)
return workManagerProvider.matrixOneTimeWorkRequestBuilder<EncryptEventWorker>()
.setConstraints(WorkManagerProvider.workConstraints)
.setInputData(sendWorkData)
.startChain(startChain)
.setBackoffCriteria(BackoffPolicy.LINEAR, WorkManagerProvider.BACKOFF_DELAY, TimeUnit.MILLISECONDS)
.build()
}
private fun createSendEventWork(event: Event, startChain: Boolean): OneTimeWorkRequest {
val sendContentWorkerParams = SendEventWorker.Params(sessionId = sessionId, eventId = event.eventId!!)
val sendWorkData = WorkerParamsFactory.toData(sendContentWorkerParams)
return timelineSendEventWorkCommon.createWork<SendEventWorker>(sendWorkData, startChain)
}
}

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

@ -22,8 +22,10 @@ import com.squareup.moshi.JsonClass
import io.realm.RealmConfiguration
import org.greenrobot.eventbus.EventBus
import org.matrix.android.sdk.api.failure.shouldBeRetried
import org.matrix.android.sdk.api.session.crypto.CryptoService
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.internal.crypto.tasks.SendEventTask
import org.matrix.android.sdk.internal.di.SessionDatabase
import org.matrix.android.sdk.internal.network.executeRequest
import org.matrix.android.sdk.internal.session.SessionComponent
@ -47,11 +49,14 @@ internal class SendEventWorker(context: Context,
internal data class Params(
override val sessionId: String,
override val lastFailureMessage: String? = null,
val eventId: String
val eventId: String,
// use this as an override if you want to send in clear in encrypted room
val isEncrypted: Boolean? = null
) : SessionWorkerParams
@Inject lateinit var localEchoRepository: LocalEchoRepository
@Inject lateinit var roomAPI: RoomAPI
@Inject lateinit var sendEventTask: SendEventTask
@Inject lateinit var cryptoService: CryptoService
@Inject lateinit var eventBus: EventBus
@Inject lateinit var cancelSendTracker: CancelSendTracker
@SessionDatabase @Inject lateinit var realmConfiguration: RealmConfiguration
@ -63,7 +68,7 @@ internal class SendEventWorker(context: Context,
override suspend fun doSafeWork(params: Params): Result {
val event = localEchoRepository.getUpToDateEcho(params.eventId)
if (event?.eventId == null || event.roomId == null) {
localEchoRepository.updateSendState(params.eventId, SendState.UNDELIVERED)
localEchoRepository.updateSendState(params.eventId, event?.roomId, SendState.UNDELIVERED)
return Result.success()
.also { Timber.e("Work cancelled due to bad input data") }
}
@ -77,7 +82,7 @@ internal class SendEventWorker(context: Context,
}
if (params.lastFailureMessage != null) {
localEchoRepository.updateSendState(event.eventId, SendState.UNDELIVERED)
localEchoRepository.updateSendState(event.eventId, event.roomId, SendState.UNDELIVERED)
// Transmit the error
return Result.success(inputData)
.also { Timber.e("Work cancelled due to input error from parent") }
@ -85,12 +90,12 @@ internal class SendEventWorker(context: Context,
Timber.v("## SendEvent: [${System.currentTimeMillis()}] Send event ${params.eventId}")
return try {
sendEvent(event.eventId, event.roomId, event.type, event.content)
sendEventTask.execute(SendEventTask.Params(event, params.isEncrypted ?: cryptoService.isRoomEncrypted(event.roomId) , cryptoService))
Result.success()
} catch (exception: Throwable) {
if (/*currentAttemptCount >= MAX_NUMBER_OF_RETRY_BEFORE_FAILING ||**/ !exception.shouldBeRetried()) {
Timber.e("## SendEvent: [${System.currentTimeMillis()}] Send event Failed cannot retry ${params.eventId} > ${exception.localizedMessage}")
localEchoRepository.updateSendState(event.eventId, SendState.UNDELIVERED)
localEchoRepository.updateSendState(event.eventId, event.roomId, SendState.UNDELIVERED)
return Result.success()
} else {
Timber.e("## SendEvent: [${System.currentTimeMillis()}] Send event Failed schedule retry ${params.eventId} > ${exception.localizedMessage}")
@ -103,11 +108,4 @@ internal class SendEventWorker(context: Context,
return params.copy(lastFailureMessage = params.lastFailureMessage ?: message)
}
private suspend fun sendEvent(eventId: String, roomId: String, type: String, content: Content?) {
localEchoRepository.updateSendState(eventId, SendState.SENDING)
executeRequest<SendResponse>(eventBus) {
apiCall = roomAPI.send(eventId, roomId, type, content)
}
localEchoRepository.updateSendState(eventId, SendState.SENT)
}
}

242
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/queue/EventSenderProcessor.kt Normal file
View file

@ -0,0 +1,242 @@
/*
* 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.session.room.send.queue
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.matrix.android.sdk.api.auth.data.SessionParams
import org.matrix.android.sdk.api.auth.data.sessionId
import org.matrix.android.sdk.api.extensions.tryOrNull
import org.matrix.android.sdk.api.failure.Failure
import org.matrix.android.sdk.api.failure.MatrixError
import org.matrix.android.sdk.api.failure.isTokenError
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.events.model.Event
import org.matrix.android.sdk.api.session.sync.SyncState
import org.matrix.android.sdk.api.util.Cancelable
import org.matrix.android.sdk.internal.session.SessionScope
import org.matrix.android.sdk.internal.task.TaskExecutor
import timber.log.Timber
import java.io.IOException
import java.net.InetAddress
import java.net.InetSocketAddress
import java.net.Socket
import java.util.Timer
import java.util.TimerTask
import java.util.concurrent.LinkedBlockingQueue
import javax.inject.Inject
import kotlin.concurrent.schedule
/**
* A simple ever running thread unique for that session responsible of sending events in order.
* Each send is retried 3 times, if there is no network (e.g if cannot ping home server) it will wait and
* periodically test reachability before resume (does not count as a retry)
*
* If the app is killed before all event were sent, on next wakeup the sheduled events will be re posted
*/
@SessionScope
internal class EventSenderProcessor @Inject constructor(
private val cryptoService: CryptoService,
private val sessionParams: SessionParams,
private val queuedTaskFactory: QueuedTaskFactory,
private val taskExecutor: TaskExecutor,
private val memento: QueueMemento
) : Thread("SENDER_THREAD_SID_${sessionParams.credentials.sessionId()}") {
private fun markAsManaged(task: QueuedTask) {
memento.track(task)
}
private fun markAsFinished(task: QueuedTask) {
memento.unTrack(task)
}
// API
fun postEvent(event: Event): Cancelable {
return postEvent(event, event.roomId?.let { cryptoService.isRoomEncrypted(it) } ?: false)
}
override fun start() {
super.start()
// We should check for sending events not handled because app was killed
// But we should be careful of only took those that was submitted to us, because if it's
// for example it's a media event it is handled by some worker and he will handle it
// This is a bit fragile :/
// also some events cannot be retried manually by users, e.g reactions
// they were previously relying on workers to do the work :/ and was expected to always finally succeed
// Also some echos are not to be resent like redaction echos (fake event created for aggregation)
tryOrNull {
taskExecutor.executorScope.launch {
Timber.d("## Send relaunched pending events on restart")
memento.restoreTasks(this@EventSenderProcessor)
}
}
}
fun postEvent(event: Event, encrypt: Boolean): Cancelable {
val task = queuedTaskFactory.createSendTask(event, encrypt)
return postTask(task)
}
fun postRedaction(redactionLocalEcho: Event, reason: String?): Cancelable {
return postRedaction(redactionLocalEcho.eventId!!, redactionLocalEcho.redacts!!, redactionLocalEcho.roomId!!, reason)
}
fun postRedaction(redactionLocalEchoId: String, eventToRedactId: String, roomId: String, reason: String?): Cancelable {
val task = queuedTaskFactory.createRedactTask(redactionLocalEchoId, eventToRedactId, roomId, reason)
return postTask(task)
}
fun postTask(task: QueuedTask): Cancelable {
// non blocking add to queue
sendingQueue.add(task)
markAsManaged(task)
return object : Cancelable {
override fun cancel() {
task.cancel()
}
}
}
companion object {
private const val RETRY_WAIT_TIME_MS = 10_000L
}
private var sendingQueue = LinkedBlockingQueue<QueuedTask>()
private var networkAvailableLock = Object()
private var canReachServer = true
private var retryNoNetworkTask: TimerTask? = null
override fun run() {
Timber.v("## SendThread started ts:${System.currentTimeMillis()}")
try {
while (!isInterrupted) {
Timber.v("## SendThread wait for task to process")
val task = sendingQueue.take()
Timber.v("## SendThread Found task to process $task")
if (task.isCancelled()) {
Timber.v("## SendThread send cancelled for $task")
// we do not execute this one
continue
}
// we check for network connectivity
while (!canReachServer) {
Timber.v("## SendThread cannot reach server, wait ts:${System.currentTimeMillis()}")
// schedule to retry
waitForNetwork()
// if thread as been killed meanwhile
// if (state == State.KILLING) break
}
Timber.v("## Server is Reachable")
// so network is available
runBlocking {
retryLoop@ while (task.retryCount < 3) {
try {
// SendPerformanceProfiler.startStage(task.event.eventId!!, SendPerformanceProfiler.Stages.SEND_WORKER)
Timber.v("## SendThread retryLoop for $task retryCount ${task.retryCount}")
task.execute()
//sendEventTask.execute(SendEventTask.Params(task.event, task.encrypt, cryptoService))
// SendPerformanceProfiler.stopStage(task.event.eventId, SendPerformanceProfiler.Stages.SEND_WORKER)
break@retryLoop
} catch (exception: Throwable) {
when {
exception is IOException || exception is Failure.NetworkConnection -> {
canReachServer = false
task.retryCount++
if (task.retryCount >= 3) task.onTaskFailed() //localEchoRepository.updateSendState(task.event.eventId!!, task.event.roomId, SendState.UNDELIVERED)
while (!canReachServer) {
Timber.v("## SendThread retryLoop cannot reach server, wait ts:${System.currentTimeMillis()}")
// schedule to retry
waitForNetwork()
// if thread as been killed meanwhile
// if (state == State.KILLING) break
}
}
(exception is Failure.ServerError && exception.error.code == MatrixError.M_LIMIT_EXCEEDED) -> {
task.retryCount++
if (task.retryCount >= 3) task.onTaskFailed() //localEchoRepository.updateSendState(task.event.eventId!!, task.event.roomId, SendState.UNDELIVERED)
Timber.v("## SendThread retryLoop retryable error for ${task} reason: ${exception.localizedMessage}")
// wait a bit
// Todo if its a quota exception can we get timout?
sleep(3_000)
continue@retryLoop
}
exception.isTokenError() -> {
Timber.v("## SendThread retryLoop retryable TOKEN error, interrupt")
// we can exit the loop
task.onTaskFailed()
// localEchoRepository.updateSendState(task.event.eventId!!, task.event.roomId, SendState.UNDELIVERED)
throw InterruptedException()
}
else -> {
Timber.v("## SendThread retryLoop Un-Retryable error, try next task")
// this task is in error, check next one?
break@retryLoop
}
}
}
}
}
markAsFinished(task)
}
} catch (interruptionException: InterruptedException) {
// will be thrown is thread is interrupted while seeping
interrupt();
Timber.v("## InterruptedException!! ${interruptionException.localizedMessage}")
}
// state = State.KILLED
// is this needed?
retryNoNetworkTask?.cancel()
Timber.w("## SendThread finished ${System.currentTimeMillis()}")
}
private fun waitForNetwork() {
retryNoNetworkTask = Timer(SyncState.NoNetwork.toString(), false).schedule(RETRY_WAIT_TIME_MS) {
synchronized(networkAvailableLock) {
canReachServer = checkHostAvailable().also {
Timber.v("## SendThread checkHostAvailable ${it}")
}
networkAvailableLock.notify()
}
}
synchronized(networkAvailableLock) { networkAvailableLock.wait() }
}
/**
* Check if homeserver is reachable.
*/
private fun checkHostAvailable(): Boolean {
val host = sessionParams.homeServerConnectionConfig.homeServerUri.host ?: return false
val port = sessionParams.homeServerConnectionConfig.homeServerUri.port.takeIf { it != -1 } ?: 80
val timeout = 30_000
try {
Socket().use { socket ->
val inetAddress: InetAddress = InetAddress.getByName(host)
val inetSocketAddress = InetSocketAddress(inetAddress, port)
socket.connect(inetSocketAddress, timeout)
return true
}
} catch (e: IOException) {
Timber.v("## EventSender isHostAvailable failure ${e.localizedMessage}")
return false
}
}
}

121
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/queue/QueueMemento.kt Normal file
View file

@ -0,0 +1,121 @@
/*
* 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.session.room.send.queue
import android.content.Context
import org.matrix.android.sdk.api.auth.data.SessionParams
import org.matrix.android.sdk.api.auth.data.sessionId
import org.matrix.android.sdk.api.extensions.tryOrNull
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.room.send.SendState
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
import timber.log.Timber
import javax.inject.Inject
/**
* Simple lightweight persistence
* Don't want to go in DB due to current issues
* Will never manage lots of events, it simply uses sharedPreferences.
* It is just used to remember what events/localEchos was managed by the event sender in order to
* reschedule them (and only them) on next restart
*/
internal class QueueMemento @Inject constructor(context: Context,
sessionParams: SessionParams,
private val queuedTaskFactory: QueuedTaskFactory,
private val localEchoRepository: LocalEchoRepository,
private val cryptoService: CryptoService) {
private val storage = context.getSharedPreferences("QueueMemento_${sessionParams.credentials.sessionId()}", Context.MODE_PRIVATE)
private val managedTaskInfos = mutableListOf<QueuedTask>()
fun track(task: QueuedTask) {
synchronized(managedTaskInfos) {
managedTaskInfos.add(task)
persist()
}
}
fun unTrack(task: QueuedTask) {
managedTaskInfos.remove(task)
persist()
}
private fun persist() {
managedTaskInfos.mapIndexedNotNull { index, queuedTask ->
toTaskInfo(queuedTask, index)?.let { TaskInfo.map(it) }
}.toSet().let { set ->
storage.edit()
.putStringSet("ManagedBySender", set)
.apply()
}
}
private fun toTaskInfo(task: QueuedTask, order: Int): TaskInfo? {
synchronized(managedTaskInfos) {
return when (task) {
is SendEventQueuedTask -> SendEventTaskInfo(
localEchoId = task.event.eventId ?: "",
encrypt = task.encrypt,
order = order
)
is RedactQueuedTask -> RedactEventTaskInfo(
redactionLocalEcho = task.redactionLocalEchoId,
order = order
)
else -> null
}
}
}
suspend fun restoreTasks(eventProcessor: EventSenderProcessor) {
// events should be restarted in correct order
storage.getStringSet("ManagedBySender", null)?.let { pending ->
Timber.d("## Send - Recovering unsent events ${pending}")
pending.mapNotNull { tryOrNull { TaskInfo.map(it) } }
}
?.sortedBy { it.order }
?.forEach { info ->
try {
when (info) {
is SendEventTaskInfo -> {
localEchoRepository.getUpToDateEcho(info.localEchoId)?.let {
if (it.sendState.isSending() && it.eventId != null && it.roomId != null) {
localEchoRepository.updateSendState(it.eventId, it.roomId, SendState.UNSENT)
Timber.d("## Send -Reschedule send ${info}")
eventProcessor.postTask(queuedTaskFactory.createSendTask(it, info.encrypt ?: cryptoService.isRoomEncrypted(it.roomId)))
}
}
}
is RedactEventTaskInfo -> {
info.redactionLocalEcho?.let { localEchoRepository.getUpToDateEcho(it) }?.let {
localEchoRepository.updateSendState(it.eventId!!, it.roomId, SendState.UNSENT)
// try to get reason
val reason = it.content?.get("reason") as? String
if (it.redacts != null && it.roomId != null) {
Timber.d("## Send -Reschedule redact ${info}")
eventProcessor.postTask(queuedTaskFactory.createRedactTask(it.eventId, it.redacts, it.roomId, reason))
}
}
//postTask(queuedTaskFactory.createRedactTask(info.eventToRedactId, info.)
}
}
} catch (failure: Throwable) {
Timber.e("failed to restore task ${info}")
}
}
}
}

29
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/queue/QueuedTask.kt Normal file
View file

@ -0,0 +1,29 @@
/*
* 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.session.room.send.queue
abstract class QueuedTask {
var retryCount = 0
abstract suspend fun execute()
abstract fun onTaskFailed()
abstract fun isCancelled() : Boolean
abstract fun cancel()
}

57
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/queue/QueuedTaskFactory.kt Normal file
View file

@ -0,0 +1,57 @@
/*
* 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.session.room.send.queue
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.events.model.Event
import org.matrix.android.sdk.internal.crypto.tasks.RedactEventTask
import org.matrix.android.sdk.internal.crypto.tasks.SendEventTask
import org.matrix.android.sdk.internal.session.room.send.CancelSendTracker
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
import javax.inject.Inject
internal class QueuedTaskFactory @Inject constructor(
private val sendEventTask: SendEventTask,
private val cryptoService: CryptoService,
private val localEchoRepository: LocalEchoRepository,
private val redactEventTask: RedactEventTask,
private val cancelSendTracker: CancelSendTracker
) {
fun createSendTask(event: Event, encrypt: Boolean): QueuedTask {
return SendEventQueuedTask(
event = event,
encrypt = encrypt,
cryptoService = cryptoService,
localEchoRepository = localEchoRepository,
sendEventTask = sendEventTask,
cancelSendTracker = cancelSendTracker
)
}
fun createRedactTask(redactionLocalEcho: String, eventId: String, roomId: String, reason: String?): QueuedTask {
return RedactQueuedTask(
redactionLocalEchoId = redactionLocalEcho,
toRedactEventId = eventId,
roomId = roomId,
reason = reason,
redactEventTask = redactEventTask,
localEchoRepository = localEchoRepository,
cancelSendTracker = cancelSendTracker
)
}
}

54
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/queue/RedactQueuedTask.kt Normal file
View file

@ -0,0 +1,54 @@
/*
* 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.session.room.send.queue
import org.matrix.android.sdk.api.session.room.send.SendState
import org.matrix.android.sdk.internal.crypto.tasks.RedactEventTask
import org.matrix.android.sdk.internal.session.room.send.CancelSendTracker
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
internal class RedactQueuedTask(
val toRedactEventId: String,
val redactionLocalEchoId: String,
val roomId: String,
val reason: String?,
val redactEventTask: RedactEventTask,
val localEchoRepository: LocalEchoRepository,
val cancelSendTracker: CancelSendTracker
) : QueuedTask() {
private var _isCancelled: Boolean = false
override fun toString() = "[RedactEventRunnableTask $redactionLocalEchoId]"
override suspend fun execute() {
redactEventTask.execute(RedactEventTask.Params(redactionLocalEchoId, roomId, toRedactEventId, reason))
}
override fun onTaskFailed() {
localEchoRepository.updateSendState(redactionLocalEchoId, roomId, SendState.UNDELIVERED)
}
override fun isCancelled(): Boolean {
return _isCancelled || cancelSendTracker.isCancelRequestedFor(redactionLocalEchoId, roomId)
}
override fun cancel() {
_isCancelled = true
}
}

65
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/queue/SendEventQueuedTask.kt Normal file
View file

@ -0,0 +1,65 @@
/*
* 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.session.room.send.queue
import org.matrix.android.sdk.api.session.crypto.CryptoService
import org.matrix.android.sdk.api.session.events.model.Event
import org.matrix.android.sdk.api.session.events.model.EventType
import org.matrix.android.sdk.api.session.room.send.SendState
import org.matrix.android.sdk.internal.crypto.tasks.SendEventTask
import org.matrix.android.sdk.internal.session.room.send.CancelSendTracker
import org.matrix.android.sdk.internal.session.room.send.LocalEchoRepository
internal class SendEventQueuedTask(
val event: Event,
val encrypt: Boolean,
val sendEventTask: SendEventTask,
val cryptoService: CryptoService,
val localEchoRepository: LocalEchoRepository,
val cancelSendTracker: CancelSendTracker
) : QueuedTask() {
private var _isCancelled: Boolean = false
override fun toString() = "[SendEventRunnableTask ${event.eventId}]"
override suspend fun execute() {
sendEventTask.execute(SendEventTask.Params(event, encrypt, cryptoService))
}
override fun onTaskFailed() {
when (event.getClearType()) {
EventType.REDACTION,
EventType.REACTION -> {
// we just delete? it will not be present on timeline and no ux to retry
localEchoRepository.deleteFailedEchoAsync(eventId = event.eventId, roomId = event.roomId ?: "")
// TODO update aggregation :/ or it will stay locally
}
else -> {
localEchoRepository.updateSendState(event.eventId!!, event.roomId, SendState.UNDELIVERED)
}
}
}
override fun isCancelled(): Boolean {
return _isCancelled || cancelSendTracker.isCancelRequestedFor(event.eventId, event.roomId)
}
override fun cancel() {
_isCancelled = true
}
}

76
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/send/queue/TaskInfo.kt Normal file
View file

@ -0,0 +1,76 @@
/*
* 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.session.room.send.queue
import com.squareup.moshi.Json
import com.squareup.moshi.JsonClass
import com.squareup.moshi.Moshi
import org.matrix.android.sdk.api.extensions.tryOrNull
import org.matrix.android.sdk.internal.di.SerializeNulls
import org.matrix.android.sdk.internal.network.parsing.RuntimeJsonAdapterFactory
/**
* Info that need to be persisted by the sender thread
* With polymorphic moshi parsing
*/
interface TaskInfo {
val type: String
val order: Int
companion object {
const val TYPE_UNKNOWN = "TYPE_UNKNOWN"
const val TYPE_SEND = "TYPE_SEND"
const val TYPE_REDACT = "TYPE_REDACT"
val moshi = Moshi.Builder()
.add(RuntimeJsonAdapterFactory.of(TaskInfo::class.java, "type", FallbackTaskInfo::class.java)
.registerSubtype(SendEventTaskInfo::class.java, TYPE_SEND)
.registerSubtype(RedactEventTaskInfo::class.java, TYPE_REDACT)
)
.add(SerializeNulls.JSON_ADAPTER_FACTORY)
.build()
fun map(info: TaskInfo): String {
return moshi.adapter(TaskInfo::class.java).toJson(info)
}
fun map(string: String): TaskInfo? {
return tryOrNull { moshi.adapter(TaskInfo::class.java).fromJson(string) }
}
}
}
@JsonClass(generateAdapter = true)
data class SendEventTaskInfo(
@Json(name = "type") override val type: String = TaskInfo.TYPE_SEND,
@Json(name = "localEchoId") val localEchoId: String,
@Json(name = "encrypt") val encrypt: Boolean?,
@Json(name = "order") override val order: Int
) : TaskInfo
@JsonClass(generateAdapter = true)
data class RedactEventTaskInfo(
@Json(name = "type") override val type: String = TaskInfo.TYPE_REDACT,
@Json(name = "redactionLocalEcho") val redactionLocalEcho: String?,
@Json(name = "order") override val order: Int
) : TaskInfo
@JsonClass(generateAdapter = true)
data class FallbackTaskInfo(
@Json(name = "type") override val type: String = TaskInfo.TYPE_REDACT,
@Json(name = "order") override val order: Int
) : TaskInfo

24
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/timeline/DefaultTimeline.kt
View file

@ -83,6 +83,7 @@ internal class DefaultTimeline(
data class OnNewTimelineEvents(val roomId: String, val eventIds: List<String>)
data class OnLocalEchoCreated(val roomId: String, val timelineEvent: TimelineEvent)
data class OnLocalEchoUpdated(val roomId: String, val eventId: String, val sendState: SendState)
companion object {
val BACKGROUND_HANDLER = createBackgroundHandler("TIMELINE_DB_THREAD")
@ -103,6 +104,7 @@ internal class DefaultTimeline(
private var prevDisplayIndex: Int? = null
private var nextDisplayIndex: Int? = null
private val inMemorySendingEvents = Collections.synchronizedList<TimelineEvent>(ArrayList())
private val inMemorySendingStates = Collections.synchronizedMap<String, SendState>(HashMap())
private val builtEvents = Collections.synchronizedList<TimelineEvent>(ArrayList())
private val builtEventsIdMap = Collections.synchronizedMap(HashMap<String, Int>())
private val backwardsState = AtomicReference(State())
@ -167,6 +169,9 @@ internal class DefaultTimeline(
events.forEach { te ->
inMemorySendingEvents.removeAll { te.eventId == it.eventId }
}
inMemorySendingStates.keys.removeAll { key ->
events.find { it.eventId == key } == null
}
postSnapshot()
}
nonFilteredEvents = buildEventQuery(realm).sort(TimelineEventEntityFields.DISPLAY_INDEX, Sort.DESCENDING).findAll()
@ -331,6 +336,18 @@ internal class DefaultTimeline(
}
}
@Subscribe(threadMode = ThreadMode.MAIN)
fun onLocalEchoUpdated(onLocalEchoUpdated: OnLocalEchoUpdated) {
if (isLive && onLocalEchoUpdated.roomId == roomId) {
val existingState = inMemorySendingStates[onLocalEchoUpdated.eventId]
inMemorySendingStates[onLocalEchoUpdated.eventId] = onLocalEchoUpdated.sendState
if (existingState != onLocalEchoUpdated.sendState) {
// Timber.v("## SendEvent: onLocalEchoUpdated $onLocalEchoUpdated in mem size = ${inMemorySendingStates.size}")
postSnapshot()
}
}
}
// Private methods *****************************************************************************
private fun rebuildEvent(eventId: String, builder: (TimelineEvent) -> TimelineEvent?): Boolean {
@ -410,7 +427,12 @@ internal class DefaultTimeline(
builtSendingEvents.addAll(inMemorySendingEvents.filterEventsWithSettings())
sendingEvents.forEach { timelineEventEntity ->
if (builtSendingEvents.find { it.eventId == timelineEventEntity.eventId } == null) {
builtSendingEvents.add(timelineEventMapper.map(timelineEventEntity))
val element = timelineEventMapper.map(timelineEventEntity)
inMemorySendingStates[element.eventId]?.let {
// Timber.v("## ${System.currentTimeMillis()} Send event refresh echo with live state ${it} from state ${element.root.sendState}")
element.root.sendState = element.root.sendState.takeIf { it == SendState.SENT } ?: it
}
builtSendingEvents.add(element)
}
}
}

23
matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/timeline/TimelineSendEventWorkCommon.kt
View file

@ -38,24 +38,6 @@ internal class TimelineSendEventWorkCommon @Inject constructor(
private val workManagerProvider: WorkManagerProvider
) {
fun postSequentialWorks(roomId: String, vararg workRequests: OneTimeWorkRequest): Cancelable {
return when {
workRequests.isEmpty() -> NoOpCancellable
workRequests.size == 1 -> postWork(roomId, workRequests.first())
else -> {
val firstWork = workRequests.first()
var continuation = workManagerProvider.workManager
.beginUniqueWork(buildWorkName(roomId), ExistingWorkPolicy.APPEND, firstWork)
for (i in 1 until workRequests.size) {
val workRequest = workRequests[i]
continuation = continuation.then(workRequest)
}
continuation.enqueue()
CancelableWork(workManagerProvider.workManager, firstWork.id)
}
}
}
fun postWork(roomId: String, workRequest: OneTimeWorkRequest, policy: ExistingWorkPolicy = ExistingWorkPolicy.APPEND_OR_REPLACE): Cancelable {
workManagerProvider.workManager
.beginUniqueWork(buildWorkName(roomId), policy, workRequest)
@ -77,11 +59,6 @@ internal class TimelineSendEventWorkCommon @Inject constructor(
return "${roomId}_$SEND_WORK"
}
fun cancelAllWorks(roomId: String) {
workManagerProvider.workManager
.cancelUniqueWork(buildWorkName(roomId))
}
companion object {
private const val SEND_WORK = "SEND_WORK"
private const val BACKOFF_DELAY = 10_000L

1
vector/src/main/java/im/vector/app/features/home/room/detail/RoomDetailAction.kt
View file

@ -68,7 +68,6 @@ sealed class RoomDetailAction : VectorViewModelAction {
data class IgnoreUser(val userId: String?) : RoomDetailAction()
object ClearSendQueue : RoomDetailAction()
object ResendAll : RoomDetailAction()
data class StartCall(val isVideo: Boolean) : RoomDetailAction()
object EndCall : RoomDetailAction()

7
vector/src/main/java/im/vector/app/features/home/room/detail/RoomDetailFragment.kt
View file

@ -645,13 +645,6 @@ class RoomDetailFragment @Inject constructor(
override fun onOptionsItemSelected(item: MenuItem): Boolean {
return when (item.itemId) {
R.id.clear_message_queue -> {
// This a temporary option during dev as it is not super stable
// Cancel all pending actions in room queue and post a dummy
// Then mark all sending events as undelivered
roomDetailViewModel.handle(RoomDetailAction.ClearSendQueue)
true
}
R.id.invite -> {
navigator.openInviteUsersToRoom(requireActivity(), roomDetailArgs.roomId)
true

7
vector/src/main/java/im/vector/app/features/home/room/detail/RoomDetailViewModel.kt
View file

@ -251,7 +251,6 @@ class RoomDetailViewModel @AssistedInject constructor(
is RoomDetailAction.HandleTombstoneEvent -> handleTombstoneEvent(action)
is RoomDetailAction.ResendMessage -> handleResendEvent(action)
is RoomDetailAction.RemoveFailedEcho -> handleRemove(action)
is RoomDetailAction.ClearSendQueue -> handleClearSendQueue()
is RoomDetailAction.ResendAll -> handleResendAll()
is RoomDetailAction.MarkAllAsRead -> handleMarkAllAsRead()
is RoomDetailAction.ReportContent -> handleReportContent(action)
@ -542,9 +541,6 @@ class RoomDetailViewModel @AssistedInject constructor(
return@withState false
}
when (itemId) {
R.id.clear_message_queue ->
// For now always disable when not in developer mode, worker cancellation is not working properly
timeline.pendingEventCount() > 0 && vectorPreferences.developerMode()
R.id.resend_all -> state.asyncRoomSummary()?.hasFailedSending == true
R.id.timeline_setting -> true
R.id.invite -> state.canInvite
@ -1065,9 +1061,6 @@ class RoomDetailViewModel @AssistedInject constructor(
}
}
private fun handleClearSendQueue() {
room.clearSendingQueue()
}
private fun handleResendAll() {
room.resendAllFailedMessages()

16
vector/src/main/java/im/vector/app/features/home/room/detail/timeline/item/MessageTextItem.kt
View file

@ -41,23 +41,27 @@ abstract class MessageTextItem : AbsMessageItem<MessageTextItem.Holder>() {
var movementMethod: MovementMethod? = null
override fun bind(holder: Holder) {
super.bind(holder)
holder.messageView.movementMethod = movementMethod
if (useBigFont) {
holder.messageView.textSize = 44F
} else {
holder.messageView.textSize = 14F
}
renderSendState(holder.messageView, holder.messageView)
holder.messageView.setOnClickListener(attributes.itemClickListener)
holder.messageView.setOnLongClickListener(attributes.itemLongClickListener)
if (searchForPills) {
message?.findPillsAndProcess(coroutineScope) { it.bind(holder.messageView) }
message?.findPillsAndProcess(coroutineScope) {
// mmm.. not sure this is so safe in regards to cell reuse
it.bind(holder.messageView)
}
}
val textFuture = PrecomputedTextCompat.getTextFuture(
message ?: "",
TextViewCompat.getTextMetricsParams(holder.messageView),
null)
super.bind(holder)
holder.messageView.movementMethod = movementMethod
renderSendState(holder.messageView, holder.messageView)
holder.messageView.setOnClickListener(attributes.itemClickListener)
holder.messageView.setOnLongClickListener(attributes.itemLongClickListener)
holder.messageView.setTextFuture(textFuture)
}

7
vector/src/main/res/menu/menu_timeline.xml
View file

@ -68,11 +68,4 @@
app:showAsAction="never"
tools:visible="true" />
<item
android:id="@+id/clear_message_queue"
android:title="@string/clear_timeline_send_queue"
android:visible="@bool/debug_mode"
app:showAsAction="never"
tools:visible="true" />
</menu>