Bump the SDK and handle breaking changes on timeline diffing

- avoid race conditions between when a timeline listener starts publishing to the `updatePublisher` and when they are consumed (and in which order)
- subscribe to the timeline directly in the TimelineProvider and correctly process updates serially
- update the AutoUpdatingRoomTimelineProviderMock to use an SDK TimlineMock and listener to publish updates

ElementX/Sources/Services/Timeline/RoomTimelineProvider.swift

@@ -21,6 +21,8 @@ import MatrixRustSDK
 class RoomTimelineProvider: RoomTimelineProviderProtocol {
     private var cancellables = Set<AnyCancellable>()
     private let serialDispatchQueue: DispatchQueue
+    
+    private var roomTimelineObservationToken: TaskHandle?
 
     private let paginationStateSubject = CurrentValueSubject<PaginationState, Never>(.default)
     var paginationState: PaginationState {
@@ -28,8 +30,10 @@ class RoomTimelineProvider: RoomTimelineProviderProtocol {
     }
 
     private let itemProxiesSubject: CurrentValueSubject<[TimelineItemProxy], Never>
-    var itemProxies: [TimelineItemProxy] {
-        itemProxiesSubject.value
+    private(set) var itemProxies: [TimelineItemProxy] = [] {
+        didSet {
+            itemProxiesSubject.send(itemProxies)
+        }
     }
 
     var updatePublisher: AnyPublisher<([TimelineItemProxy], PaginationState), Never> {
@@ -45,28 +49,29 @@ class RoomTimelineProvider: RoomTimelineProviderProtocol {
         membershipChangeSubject
             .eraseToAnyPublisher()
     }
+    
+    deinit {
+        roomTimelineObservationToken?.cancel()
+    }
 
-    init(currentItems: [TimelineItem],
-         isLive: Bool,
-         updatePublisher: AnyPublisher<[TimelineDiff], Never>,
-         paginationStatePublisher: AnyPublisher<PaginationState, Never>) {
+    init(timeline: Timeline, isLive: Bool, paginationStatePublisher: AnyPublisher<PaginationState, Never>) {
         serialDispatchQueue = DispatchQueue(label: "io.element.elementx.roomtimelineprovider", qos: .utility)
-        itemProxiesSubject = CurrentValueSubject<[TimelineItemProxy], Never>(currentItems.map(TimelineItemProxy.init))
+        itemProxiesSubject = CurrentValueSubject<[TimelineItemProxy], Never>([])
         self.isLive = isLive
         
-        // Manually call it here as the didSet doesn't work from constructors
-        itemProxiesSubject.send(itemProxies)
-
-        updatePublisher
-            .receive(on: serialDispatchQueue)
-            .sink { [weak self] in self?.updateItemsWithDiffs($0) }
-            .store(in: &cancellables)
-
         paginationStatePublisher
             .sink { [weak self] in
                 self?.paginationStateSubject.send($0)
             }
             .store(in: &cancellables)
+        
+        Task {
+            roomTimelineObservationToken = await timeline.addListener(listener: RoomTimelineListener { [weak self] timelineDiffs in
+                self?.serialDispatchQueue.sync {
+                    self?.updateItemsWithDiffs(timelineDiffs)
+                }
+            })
+        }
     }
     
     // MARK: - Private
@@ -80,22 +85,19 @@ class RoomTimelineProvider: RoomTimelineProviderProtocol {
         
         MXLog.verbose("Received timeline diff")
         
-        let items = diffs
-            .reduce(itemProxies) { currentItems, diff in
-                guard let collectionDiff = buildDiff(from: diff, on: currentItems) else {
-                    MXLog.error("Failed building CollectionDifference from \(diff)")
-                    return currentItems
-                }
-                
-                guard let updatedItems = currentItems.applying(collectionDiff) else {
-                    MXLog.error("Failed applying diff: \(collectionDiff)")
-                    return currentItems
-                }
-                
-                return updatedItems
+        itemProxies = diffs.reduce(itemProxies) { currentItems, diff in
+            guard let collectionDiff = buildDiff(from: diff, on: currentItems) else {
+                MXLog.error("Failed building CollectionDifference from \(diff)")
+                return currentItems
             }
-
-        itemProxiesSubject.send(items)
+            
+            guard let updatedItems = currentItems.applying(collectionDiff) else {
+                MXLog.error("Failed applying diff: \(collectionDiff)")
+                return currentItems
+            }
+            
+            return updatedItems
+        }
         
         MXLog.verbose("Finished applying diffs, current items (\(itemProxies.count)) : \(itemProxies.map(\.debugIdentifier))")
     }
@@ -251,3 +253,15 @@ enum DebugIdentifier {
     case virtual(timelineID: String, dscription: String)
     case unknown(timelineID: String)
 }
+
+private final class RoomTimelineListener: TimelineListener {
+    private let onUpdateClosure: ([TimelineDiff]) -> Void
+   
+    init(_ onUpdateClosure: @escaping ([TimelineDiff]) -> Void) {
+        self.onUpdateClosure = onUpdateClosure
+    }
+    
+    func onUpdate(diff: [TimelineDiff]) {
+        onUpdateClosure(diff)
+    }
+}

ElementX/Sources/Services/Timeline/TimelineProxy.swift

@@ -22,14 +22,9 @@ final class TimelineProxy: TimelineProxyProtocol {
     private let timeline: Timeline
     
     private var backPaginationStatusObservationToken: TaskHandle?
-    private var roomTimelineObservationToken: TaskHandle?
-    
-    // periphery:ignore - retaining purpose
-    private var timelineListener: RoomTimelineListener?
     
     private let backPaginationSubscriptionSubject = CurrentValueSubject<PaginationStatus, Never>(.idle)
     private let forwardPaginationStatusSubject = CurrentValueSubject<PaginationStatus, Never>(.timelineEndReached)
-    private let timelineUpdatesSubject = PassthroughSubject<[TimelineDiff], Never>()
     
     let isLive: Bool
    
@@ -40,7 +35,6 @@ final class TimelineProxy: TimelineProxyProtocol {
     
     deinit {
         backPaginationStatusObservationToken?.cancel()
-        roomTimelineObservationToken?.cancel()
     }
     
     init(timeline: Timeline, isLive: Bool) {
@@ -54,15 +48,6 @@ final class TimelineProxy: TimelineProxyProtocol {
             return
         }
         
-        let timelineListener = RoomTimelineListener { [weak self] timelineDiffs in
-            self?.timelineUpdatesSubject.send(timelineDiffs)
-        }
-        
-        self.timelineListener = timelineListener
-        
-        let result = await timeline.addListener(listener: timelineListener)
-        roomTimelineObservationToken = result.itemsStream
-        
         let paginationStatePublisher = backPaginationSubscriptionSubject
             .combineLatest(forwardPaginationStatusSubject)
             .map { PaginationState(backward: $0.0, forward: $0.1) }
@@ -70,10 +55,7 @@ final class TimelineProxy: TimelineProxyProtocol {
         
         await subscribeToPagination()
         
-        innerTimelineProvider = await RoomTimelineProvider(currentItems: result.items,
-                                                           isLive: isLive,
-                                                           updatePublisher: timelineUpdatesSubject.eraseToAnyPublisher(),
-                                                           paginationStatePublisher: paginationStatePublisher)
+        innerTimelineProvider = await RoomTimelineProvider(timeline: timeline, isLive: isLive, paginationStatePublisher: paginationStatePublisher)
     }
     
     func fetchDetails(for eventID: String) {
@@ -533,18 +515,6 @@ final class TimelineProxy: TimelineProxyProtocol {
     }
 }
 
-private final class RoomTimelineListener: TimelineListener {
-    private let onUpdateClosure: ([TimelineDiff]) -> Void
-   
-    init(_ onUpdateClosure: @escaping ([TimelineDiff]) -> Void) {
-        self.onUpdateClosure = onUpdateClosure
-    }
-    
-    func onUpdate(diff: [TimelineDiff]) {
-        onUpdateClosure(diff)
-    }
-}
-
 private final class RoomPaginationStatusListener: PaginationStatusListener {
     private let onUpdateClosure: (LiveBackPaginationStatus) -> Void