Add room's update publisher (#952)

* Add room update publisher

* Cleanup

* Update room’s title

* Add refresh logic in RoomDetailsScreenViewModel

* Fix UTs

* Add UT

* Update timeline listener in RoomProxy

* Fix caching in LoadableAvatarImage

* Cleanup

* Update RoomProxy api

* Inject room proxy in RoomScreenViewModel

* Cleanup

* Fix UTs

ElementX/Sources/Services/Room/RoomProxy.swift

@@ -41,7 +41,29 @@ class RoomProxy: RoomProxyProtocol {
     var membersPublisher: AnyPublisher<[RoomMemberProxyProtocol], Never> {
         membersSubject.eraseToAnyPublisher()
     }
-        
+    
+    private var timelineListener: RoomTimelineListener?
+    private let updatesSubject = PassthroughSubject<TimelineDiff, Never>()
+    var updatesPublisher: AnyPublisher<TimelineDiff, Never> {
+        updatesSubject.eraseToAnyPublisher()
+    }
+    
+    deinit {
+        Task { @MainActor [roomTimelineObservationToken, slidingSyncRoom] in
+            roomTimelineObservationToken?.cancel()
+            
+            let task = ExpiringTaskRunner {
+                let unsubscribeTask = slidingSyncRoom.unsubscribeFromRoom()
+                
+                while !unsubscribeTask.isFinished() {
+                    try await Task.sleep(for: .seconds(2))
+                }
+            }
+            
+            try? await task.run(timeout: .seconds(30))
+        }
+    }
+
     init(slidingSyncRoom: SlidingSyncRoomProtocol,
          room: RoomProtocol,
          backgroundTaskService: BackgroundTaskServiceProtocol) {
@@ -139,15 +161,27 @@ class RoomProxy: RoomProxyProtocol {
         }
     }
         
-    func addTimelineListener(listener: TimelineListener) -> Result<[TimelineItem], RoomProxyError> {
-        let settings = RoomSubscription(requiredState: [RequiredState(key: "m.room.topic", value: ""),
+    func registerTimelineListenerIfNeeded() -> Result<[TimelineItem], RoomProxyError> {
+        guard timelineListener == nil else {
+            return .failure(.roomListenerAlreadyRegistered)
+        }
+        
+        let settings = RoomSubscription(requiredState: [RequiredState(key: "m.room.name", value: ""),
+                                                        RequiredState(key: "m.room.topic", value: ""),
+                                                        RequiredState(key: "m.room.avatar", value: ""),
                                                         RequiredState(key: "m.room.canonical_alias", value: ""),
                                                         RequiredState(key: "m.room.join_rules", value: "")],
                                         timelineLimit: UInt32(SlidingSyncConstants.timelinePrecachingTimelineLimit))
         roomSubscriptionObservationToken = slidingSyncRoom.subscribeToRoom(settings: settings)
         
+        let listener = RoomTimelineListener { [weak self] timelineDiff in
+            self?.updatesSubject.send(timelineDiff)
+        }
+        
         if let result = try? slidingSyncRoom.addTimelineListener(listener: listener) {
             roomTimelineObservationToken = result.taskHandle
+            timelineListener = listener
+            
             Task {
                 await fetchMembers()
                 await updateMembers()
@@ -158,15 +192,6 @@ class RoomProxy: RoomProxyProtocol {
         }
     }
     
-    func removeTimelineListener() {
-        roomTimelineObservationToken?.cancel()
-        roomTimelineObservationToken = nil
-        
-        roomSubscriptionObservationToken = nil
-        
-        roomUnsubscriptionObservationToken = slidingSyncRoom.unsubscribeFromRoom()
-    }
-    
     func paginateBackwards(requestSize: UInt, untilNumberOfItems: UInt) async -> Result<Void, RoomProxyError> {
         do {
             try await Task.dispatch(on: .global()) {
@@ -487,3 +512,15 @@ class RoomProxy: RoomProxyProtocol {
         self.displayName = displayName
     }
 }
+
+private class RoomTimelineListener: TimelineListener {
+    private let onUpdateClosure: (TimelineDiff) -> Void
+   
+    init(_ onUpdateClosure: @escaping (TimelineDiff) -> Void) {
+        self.onUpdateClosure = onUpdateClosure
+    }
+    
+    func onUpdate(update: TimelineDiff) {
+        onUpdateClosure(update)
+    }
+}

ElementX/Sources/Services/Room/RoomProxyProtocol.swift

@@ -20,6 +20,7 @@ import MatrixRustSDK
 
 enum RoomProxyError: Error {
     case noMoreMessagesToBackPaginate
+    case roomListenerAlreadyRegistered
     case failedPaginatingBackwards
     case failedRetrievingMemberAvatarURL
     case failedRetrievingMemberDisplayName
@@ -59,14 +60,19 @@ protocol RoomProxyProtocol {
     var avatarURL: URL? { get }
 
     var membersPublisher: AnyPublisher<[RoomMemberProxyProtocol], Never> { get }
+    
+    /// Publishes the room's updates.
+    /// The publisher starts publishing after the first call to `registerTimelineListenerIfNeeded()`
+    /// The thread on which this publisher sends the output isn't defined.
+    var updatesPublisher: AnyPublisher<TimelineDiff, Never> { get }
 
     func loadAvatarURLForUserId(_ userId: String) async -> Result<URL?, RoomProxyError>
     
     func loadDisplayNameForUserId(_ userId: String) async -> Result<String?, RoomProxyError>
     
-    func addTimelineListener(listener: TimelineListener) -> Result<[TimelineItem], RoomProxyError>
-    
-    func removeTimelineListener()
+    /// Registers a timeline listener if not registered already.
+    /// Updates for this object will be published on the `updatesPublisher` publisher.
+    func registerTimelineListenerIfNeeded() -> Result<[TimelineItem], RoomProxyError>
     
     func paginateBackwards(requestSize: UInt, untilNumberOfItems: UInt) async -> Result<Void, RoomProxyError>
     
@@ -130,4 +136,10 @@ extension RoomProxyProtocol {
     func sendMessage(_ message: String) async -> Result<Void, RoomProxyError> {
         await sendMessage(message, inReplyTo: nil)
     }
+    
+    // Avoids to duplicate the same logic around in the app
+    // Probably this should be done in rust.
+    var roomTitle: String {
+        displayName ?? name ?? "Unknown room 💥"
+    }
 }