Adopt batched timeline diffs and other RustSDK changes

ElementX/Sources/Services/Notification/Proxy/NotificationItemProxy.swift

@@ -88,7 +88,12 @@ struct NotificationItemProxy: NotificationItemProxyProtocol {
     }
 
     var senderID: String {
-        notificationItem.senderInfo.userId
+        switch notificationItem.event {
+        case .timeline(let event):
+            return event.senderId()
+        case .invite(let senderID):
+            return senderID
+        }
     }
 
     var roomDisplayName: String {

ElementX/Sources/Services/Room/RoomProxy.swift

@@ -45,8 +45,8 @@ class RoomProxy: RoomProxyProtocol {
     }
     
     private var timelineListener: RoomTimelineListener?
-    private let updatesSubject = PassthroughSubject<TimelineDiff, Never>()
-    var updatesPublisher: AnyPublisher<TimelineDiff, Never> {
+    private let updatesSubject = PassthroughSubject<[TimelineDiff], Never>()
+    var updatesPublisher: AnyPublisher<[TimelineDiff], Never> {
         updatesSubject.eraseToAnyPublisher()
     }
     
@@ -80,8 +80,8 @@ class RoomProxy: RoomProxyProtocol {
                                         timelineLimit: UInt32(SlidingSyncConstants.defaultTimelineLimit))
         roomListItem.subscribe(settings: settings)
 
-        let timelineListener = RoomTimelineListener { [weak self] timelineDiff in
-            self?.updatesSubject.send(timelineDiff)
+        let timelineListener = RoomTimelineListener { [weak self] timelineDiffs in
+            self?.updatesSubject.send(timelineDiffs)
         }
 
         self.timelineListener = timelineListener
@@ -691,13 +691,13 @@ class RoomProxy: RoomProxyProtocol {
 }
 
 private final class RoomTimelineListener: TimelineListener {
-    private let onUpdateClosure: (TimelineDiff) -> Void
+    private let onUpdateClosure: ([TimelineDiff]) -> Void
    
-    init(_ onUpdateClosure: @escaping (TimelineDiff) -> Void) {
+    init(_ onUpdateClosure: @escaping ([TimelineDiff]) -> Void) {
         self.onUpdateClosure = onUpdateClosure
     }
     
-    func onUpdate(diff: TimelineDiff) {
+    func onUpdate(diff: [TimelineDiff]) {
         onUpdateClosure(diff)
     }
 }

ElementX/Sources/Services/Room/RoomProxyProtocol.swift

@@ -74,7 +74,7 @@ protocol RoomProxyProtocol {
     
     /// Publishes the room's updates.
     /// The thread on which this publisher sends the output isn't defined.
-    var updatesPublisher: AnyPublisher<TimelineDiff, Never> { get }
+    var updatesPublisher: AnyPublisher<[TimelineDiff], Never> { get }
     
     var timelineProvider: RoomTimelineProviderProtocol { get }
 

ElementX/Sources/Services/Timeline/RoomTimelineProvider.swift

@@ -40,7 +40,7 @@ class RoomTimelineProvider: RoomTimelineProviderProtocol {
     }
 
     init(currentItems: [TimelineItem],
-         updatePublisher: AnyPublisher<TimelineDiff, Never>,
+         updatePublisher: AnyPublisher<[TimelineDiff], Never>,
          backPaginationStatePublisher: AnyPublisher<BackPaginationStatus, Never>) {
         serialDispatchQueue = DispatchQueue(label: "io.element.elementx.roomtimelineprovider", qos: .utility)
         itemProxiesSubject = CurrentValueSubject<[TimelineItemProxy], Never>(currentItems.map(TimelineItemProxy.init))
@@ -49,7 +49,7 @@ class RoomTimelineProvider: RoomTimelineProviderProtocol {
         itemProxiesSubject.send(itemProxies)
 
         updatePublisher
-            .collect(.byTime(serialDispatchQueue, 0.1))
+            .receive(on: serialDispatchQueue)
             .sink { [weak self] in self?.updateItemsWithDiffs($0) }
             .store(in: &cancellables)