Fixes #2842, fixes #2841 - Adopt new message sending queue API

- automatically retry failed requests as soon as the network is availble again
- remove manual retry options
This commit is contained in:
Stefan Ceriu
2024-05-28 10:17:13 +03:00
committed by Stefan Ceriu
parent 53fc5df55b
commit 6c1647b851
28 changed files with 827 additions and 606 deletions
@@ -33,11 +33,6 @@ final class TimelineProxy: TimelineProxyProtocol {
private let forwardPaginationStatusSubject = CurrentValueSubject<PaginationStatus, Never>(.timelineEndReached)
private let timelineUpdatesSubject = PassthroughSubject<[TimelineDiff], Never>()
private let actionsSubject = PassthroughSubject<TimelineProxyAction, Never>()
var actions: AnyPublisher<TimelineProxyAction, Never> {
actionsSubject.eraseToAnyPublisher()
}
let isLive: Bool
private var innerTimelineProvider: RoomTimelineProviderProtocol!
@@ -83,38 +78,6 @@ final class TimelineProxy: TimelineProxyProtocol {
paginationStatePublisher: paginationStatePublisher)
}
func cancelSend(transactionID: String) async {
MXLog.info("Cancelling sending for transaction ID: \(transactionID)")
return await Task.dispatch(on: messageSendingDispatchQueue) {
self.timeline.cancelSend(txnId: transactionID)
MXLog.info("Finished cancelling sending for transaction ID: \(transactionID)")
}
}
func editMessage(_ message: String,
html: String?,
original eventID: String,
intentionalMentions: IntentionalMentions) async -> Result<Void, TimelineProxyError> {
MXLog.info("Editing message with original event ID: \(eventID)")
let messageContent = buildMessageContentFor(message,
html: html,
intentionalMentions: intentionalMentions.toRustMentions())
do {
let originalEvent = try await timeline.getEventTimelineItemByEventId(eventId: eventID)
try await timeline.edit(newContent: messageContent, editItem: originalEvent)
MXLog.info("Finished editing message with original event ID: \(eventID)")
return .success(())
} catch {
MXLog.error("Failed editing message with original event ID: \(eventID) with error: \(error)")
return .failure(.failedEditingMessage)
}
}
func fetchDetails(for eventID: String) {
Task {
do {
@@ -127,17 +90,8 @@ final class TimelineProxy: TimelineProxyProtocol {
}
}
func messageEventContent(for eventID: String) async -> RoomMessageEventContentWithoutRelation? {
MXLog.info("Fetching event content for \(eventID)")
do {
let result = try await timeline.getTimelineEventContentByEventId(eventId: eventID)
MXLog.info("Finished fetching event content for eventID: \(eventID)")
return result
} catch {
MXLog.error("Failed fetching event content for eventID: \(eventID) with error: \(error)")
return nil
}
func messageEventContent(for timelineItemID: TimelineItemIdentifier) async -> RoomMessageEventContentWithoutRelation? {
await timelineProvider.itemProxies.firstEventTimelineItemUsingID(timelineItemID)?.content().asMessage()?.content()
}
func paginateBackwards(requestSize: UInt16) async -> Result<Void, TimelineProxyError> {
@@ -150,7 +104,7 @@ final class TimelineProxy: TimelineProxyProtocol {
return .success(())
} catch {
MXLog.error("Failed paginating backwards with error: \(error)")
return .failure(.failedPaginatingBackwards)
return .failure(.sdkError(error))
}
}
@@ -159,7 +113,7 @@ final class TimelineProxy: TimelineProxyProtocol {
// We need it to make sure we send a valid status after a failure.
guard forwardPaginationStatusSubject.value == .idle else {
MXLog.error("Attempting to paginate forwards when already at the end.")
return .failure(.failedPaginatingBackwards)
return .failure(.failedPaginatingEndReached)
}
MXLog.info("Paginating forwards")
@@ -174,7 +128,7 @@ final class TimelineProxy: TimelineProxyProtocol {
} catch {
MXLog.error("Failed paginating forwards with error: \(error)")
forwardPaginationStatusSubject.send(.idle)
return .failure(.failedPaginatingBackwards)
return .failure(.sdkError(error))
}
}
@@ -187,15 +141,59 @@ final class TimelineProxy: TimelineProxyProtocol {
}
}
func retrySend(transactionID: String) async {
MXLog.info("Retrying sending for transactionID: \(transactionID)")
func edit(_ timelineItemID: TimelineItemIdentifier,
message: String, html: String?,
intentionalMentions: IntentionalMentions) async -> Result<Void, TimelineProxyError> {
MXLog.info("Editing timeline item: \(timelineItemID)")
return await Task.dispatch(on: messageSendingDispatchQueue) {
self.timeline.retrySend(txnId: transactionID)
MXLog.info("Finished retrying sending for transactionID: \(transactionID)")
guard let eventTimelineItem = await timelineProvider.itemProxies.firstEventTimelineItemUsingID(timelineItemID) else {
MXLog.error("Unknown timeline item: \(timelineItemID)")
return .failure(.failedEditing)
}
let messageContent = buildMessageContentFor(message,
html: html,
intentionalMentions: intentionalMentions.toRustMentions())
do {
try await timeline.edit(newContent: messageContent, editItem: eventTimelineItem)
MXLog.info("Finished editing timeline item: \(timelineItemID)")
return .success(())
} catch {
MXLog.error("Failed editing timeline item: \(timelineItemID) with error: \(error)")
return .failure(.sdkError(error))
}
}
func redact(_ timelineItemID: TimelineItemIdentifier, reason: String?) async -> Result<Void, TimelineProxyError> {
MXLog.info("Redacting timeline item: \(timelineItemID)")
guard let eventTimelineItem = await timelineProvider.itemProxies.firstEventTimelineItemUsingID(timelineItemID) else {
MXLog.error("Unknown timeline item: \(timelineItemID)")
return .failure(.failedRedacting)
}
do {
let success = try await timeline.redactEvent(item: eventTimelineItem, reason: reason)
guard success else {
MXLog.error("Failed redacting timeline item: \(timelineItemID)")
return .failure(.failedRedacting)
}
MXLog.info("Redacted timeline item: \(timelineItemID)")
return .success(())
} catch {
MXLog.error("Failed redacting timeline item: \(timelineItemID) with error: \(error)")
return .failure(.sdkError(error))
}
}
// MARK: - Sending
func sendAudio(url: URL,
audioInfo: AudioInfo,
progressSubject: CurrentValueSubject<Double, Never>?,
@@ -213,11 +211,9 @@ final class TimelineProxy: TimelineProxyProtocol {
MXLog.info("Finished sending audio")
} catch {
MXLog.error("Failed sending audio with error: \(error)")
return .failure(.failedSendingMedia)
return .failure(.sdkError(error))
}
actionsSubject.send(.sentMessage)
return .success(())
}
@@ -238,11 +234,9 @@ final class TimelineProxy: TimelineProxyProtocol {
MXLog.info("Finished sending file")
} catch {
MXLog.error("Failed sending file with error: \(error)")
return .failure(.failedSendingMedia)
return .failure(.sdkError(error))
}
actionsSubject.send(.sentMessage)
return .success(())
}
@@ -264,11 +258,9 @@ final class TimelineProxy: TimelineProxyProtocol {
MXLog.info("Finished sending image")
} catch {
MXLog.error("Failed sending image with error: \(error)")
return .failure(.failedSendingMedia)
return .failure(.sdkError(error))
}
actionsSubject.send(.sentMessage)
return .success(())
}
@@ -279,19 +271,15 @@ final class TimelineProxy: TimelineProxyProtocol {
assetType: AssetType?) async -> Result<Void, TimelineProxyError> {
MXLog.info("Sending location")
return await Task.dispatch(on: messageSendingDispatchQueue) {
self.timeline.sendLocation(body: body,
geoUri: geoURI.string,
description: description,
zoomLevel: zoomLevel,
assetType: assetType)
self.actionsSubject.send(.sentMessage)
MXLog.info("Finished sending location")
return .success(())
}
await timeline.sendLocation(body: body,
geoUri: geoURI.string,
description: description,
zoomLevel: zoomLevel,
assetType: assetType)
MXLog.info("Finished sending location")
return .success(())
}
func sendVideo(url: URL,
@@ -312,11 +300,9 @@ final class TimelineProxy: TimelineProxyProtocol {
MXLog.info("Finished sending video")
} catch {
MXLog.error("Failed sending video with error: \(error)")
return .failure(.failedSendingMedia)
return .failure(.sdkError(error))
}
actionsSubject.send(.sentMessage)
return .success(())
}
@@ -338,11 +324,9 @@ final class TimelineProxy: TimelineProxyProtocol {
MXLog.info("Finished sending voice message")
} catch {
MXLog.error("Failed sending vocie message with error: \(error)")
return .failure(.failedSendingMedia)
return .failure(.sdkError(error))
}
actionsSubject.send(.sentMessage)
return .success(())
}
@@ -366,36 +350,34 @@ final class TimelineProxy: TimelineProxyProtocol {
try await timeline.sendReply(msg: messageContent, replyItem: replyItem)
MXLog.info("Finished sending reply to eventID: \(eventID)")
} else {
timeline.send(msg: messageContent)
_ = try await timeline.send(msg: messageContent)
MXLog.info("Finished sending message")
}
} catch {
if let eventID {
MXLog.error("Failed sending reply to eventID: \(eventID)")
MXLog.error("Failed sending reply to eventID: \(eventID) with error: \(error)")
} else {
MXLog.error("Failed sending message")
MXLog.error("Failed sending message with error: \(error)")
}
return .failure(.failedSendingMessage)
return .failure(.sdkError(error))
}
actionsSubject.send(.sentMessage)
return .success(())
}
func sendMessageEventContent(_ messageContent: RoomMessageEventContentWithoutRelation) async -> Result<Void, TimelineProxyError> {
MXLog.info("Sending message content")
return await Task.dispatch(on: messageSendingDispatchQueue) {
self.timeline.send(msg: messageContent)
self.actionsSubject.send(.sentMessage)
MXLog.info("Finished sending message content")
return .success(())
do {
_ = try await timeline.send(msg: messageContent)
} catch {
MXLog.error("Failed sending message with error: \(error)")
}
MXLog.info("Finished sending message content")
return .success(())
}
func sendReadReceipt(for eventID: String, type: ReceiptType) async -> Result<Void, TimelineProxyError> {
@@ -407,7 +389,7 @@ final class TimelineProxy: TimelineProxyProtocol {
return .success(())
} catch {
MXLog.error("Failed sending read receipt for eventID: \(eventID) with error: \(error)")
return .failure(.failedSendingReadReceipt)
return .failure(.sdkError(error))
}
}
@@ -420,7 +402,7 @@ final class TimelineProxy: TimelineProxyProtocol {
return .success(())
} catch {
MXLog.error("Failed toggling reaction for eventID: \(eventID)")
return .failure(.failedSendingReaction)
return .failure(.sdkError(error))
}
}
@@ -429,19 +411,15 @@ final class TimelineProxy: TimelineProxyProtocol {
func createPoll(question: String, answers: [String], pollKind: Poll.Kind) async -> Result<Void, TimelineProxyError> {
MXLog.info("Creating poll")
return await Task.dispatch(on: .global()) {
do {
try self.timeline.createPoll(question: question, answers: answers, maxSelections: 1, pollKind: .init(pollKind: pollKind))
self.actionsSubject.send(.sentMessage)
MXLog.info("Finished creating poll")
return .success(())
} catch {
MXLog.error("Failed creating poll with error: \(error)")
return .failure(.failedCreatingPoll)
}
do {
try await timeline.createPoll(question: question, answers: answers, maxSelections: 1, pollKind: .init(pollKind: pollKind))
MXLog.info("Finished creating poll")
return .success(())
} catch {
MXLog.error("Failed creating poll with error: \(error)")
return .failure(.sdkError(error))
}
}
@@ -461,7 +439,7 @@ final class TimelineProxy: TimelineProxyProtocol {
return .success(())
} catch {
MXLog.error("Failed editing poll with eventID: \(eventID) with error: \(error)")
return .failure(.failedEditingPoll)
return .failure(.sdkError(error))
}
}
@@ -477,7 +455,7 @@ final class TimelineProxy: TimelineProxyProtocol {
return .success(())
} catch {
MXLog.error("Failed ending poll with eventID: \(pollStartID) with error: \(error)")
return .failure(.failedEndingPoll)
return .failure(.sdkError(error))
}
}
}
@@ -485,17 +463,15 @@ final class TimelineProxy: TimelineProxyProtocol {
func sendPollResponse(pollStartID: String, answers: [String]) async -> Result<Void, TimelineProxyError> {
MXLog.info("Sending response for poll with eventID: \(pollStartID)")
return await Task.dispatch(on: .global()) {
do {
try self.timeline.sendPollResponse(pollStartId: pollStartID, answers: answers)
MXLog.info("Finished sending response for poll with eventID: \(pollStartID)")
return .success(())
} catch {
MXLog.error("Failed sending response for poll with eventID: \(pollStartID) with error: \(error)")
return .failure(.failedSendingPollResponse)
}
do {
try await timeline.sendPollResponse(pollStartId: pollStartID, answers: answers)
MXLog.info("Finished sending response for poll with eventID: \(pollStartID)")
return .success(())
} catch {
MXLog.error("Failed sending response for poll with eventID: \(pollStartID) with error: \(error)")
return .failure(.sdkError(error))
}
}
@@ -607,3 +583,20 @@ private extension MatrixRustSDK.PollKind {
}
}
}
extension Array where Element == TimelineItemProxy {
func firstEventTimelineItemUsingID(_ id: TimelineItemIdentifier) -> EventTimelineItem? {
var eventTimelineItemProxy: EventTimelineItemProxy?
for item in self {
if case let .event(eventTimelineItem) = item {
if eventTimelineItem.id == id {
eventTimelineItemProxy = eventTimelineItem
break
}
}
}
return eventTimelineItemProxy?.item
}
}