[Fix]Convert AsyncStream to Publishers > AsyncStream to break memory leaks (#759)

Author: ipavlidakis

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ### ✅ Added
 - Countdown timer and waiting participants info to the livestream player [#754](https://github.com/GetStream/stream-video-swift/pull/754)
+- EventPublisher for `Call` objects. [#759](https://github.com/GetStream/stream-video-swift/pull/759)
+
+### 🐞 Fixed
+- `CallViewModel.callingState` transition to `.idle` just before moving to `.inCall` after the user has accepted the call. [#759](https://github.com/GetStream/stream-video-swift/pull/759)
 
 ### 🔄 Changed
 - `CallSettings` won't be propagated between calls when using the `CallViewModel`. [#751](https://github.com/GetStream/stream-video-swift/pull/751)

Sources/StreamVideo/Call.swift

@@ -28,6 +28,20 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
         callCid(from: callId, callType: callType)
     }
 
+    private let eventSubject: PassthroughSubject<WrappedEvent, Never> = .init()
+    public var eventPublisher: AnyPublisher<VideoEvent, Never> {
+        eventSubject
+            .compactMap {
+                switch $0 {
+                case let .coordinatorEvent(event):
+                    return event
+                default:
+                    return nil
+                }
+            }
+            .eraseToAnyPublisher()
+    }
+
     /// Provides access to the microphone.
     public let microphone: MicrophoneManager
     /// Provides access to the camera.
@@ -37,7 +51,6 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
 
     internal let callController: CallController
     internal let coordinatorClient: DefaultAPI
-    private var eventHandlers = [EventHandler]()
     private var cancellables = DisposableBag()
 
     /// This adapter is used to manage closed captions for the
@@ -90,6 +103,7 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
     }
 
     deinit {
+        log.debug("Call cID:\(cId) is deallocating...")
         cancellables.removeAll()
     }
 
@@ -481,45 +495,30 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
         try await callController.stopScreensharing()
     }
 
+    public func eventPublisher<WSEvent: Event>(for event: WSEvent.Type) -> AnyPublisher<WSEvent, Never> {
+        eventPublisher
+            .compactMap { $0.rawValue as? WSEvent }
+            .eraseToAnyPublisher()
+    }
+
     /// Subscribes to video events.
     /// - Returns: `AsyncStream` of `VideoEvent`s.
     public func subscribe() -> AsyncStream<VideoEvent> {
-        AsyncStream(VideoEvent.self) { [weak self] continuation in
-            let eventHandler = EventHandler(handler: { event in
-                guard case let .coordinatorEvent(event) = event else {
-                    return
-                }
-                continuation.yield(event)
-            }, cancel: { continuation.finish() })
-            self?.eventHandlers.append(eventHandler)
-        }
+        eventPublisher.eraseAsAsyncStream()
     }
 
     /// Subscribes to a particular web socket event.
     /// - Parameter event: the type of the event you are subscribing to.
     /// - Returns: `AsyncStream` of web socket events from the provided type.
     public func subscribe<WSEvent: Event>(for event: WSEvent.Type) -> AsyncStream<WSEvent> {
-        AsyncStream(event) { [weak self] continuation in
-            let eventHandler = EventHandler(handler: { event in
-                guard case let .coordinatorEvent(event) = event else {
-                    return
-                }
-                if let event = event.rawValue as? WSEvent {
-                    continuation.yield(event)
-                }
-            }, cancel: { continuation.finish() })
-
-            self?.eventHandlers.append(eventHandler)
-        }
+        eventPublisher(for: event).eraseAsAsyncStream()
     }
 
     /// Leave the current call.
     public func leave() {
         postNotification(with: CallNotification.callEnded, object: self)
-        eventHandlers.forEach { $0.cancel() }
 
         cancellables.removeAll()
-        eventHandlers.removeAll()
         callController.leave()
         closedCaptionsAdapter.stop()
         stateMachine.transition(.idle(self))
@@ -1384,12 +1383,7 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
             self.state.updateState(from: videoEvent)
         }
 
-        // Get a copy of eventHandlers to avoid crashes when `leave` call is being
-        // triggered, during event processing.
-        let eventHandlers = self.eventHandlers
-        for eventHandler in eventHandlers {
-            eventHandler.handler(event)
-        }
+        eventSubject.send(event)
     }
 
     @MainActor

Sources/StreamVideo/Controllers/CallController.swift

@@ -8,6 +8,10 @@ import StreamWebRTC
 
 /// Class that handles a particular call.
 class CallController: @unchecked Sendable {
+    private enum DisposableKey: String {
+        case participantsCountUpdatesEvent
+        case currentUserBlocked
+    }
 
     private lazy var webRTCCoordinator = webRTCCoordinatorFactory.buildCoordinator(
         user: user,
@@ -42,13 +46,8 @@ class CallController: @unchecked Sendable {
     private let apiKey: String
     private let defaultAPI: DefaultAPI
     private let videoConfig: VideoConfig
-    private let sfuReconnectionTime: CGFloat
     private let webRTCCoordinatorFactory: WebRTCCoordinatorProviding
-    private var reconnectionDate: Date?
     private var cachedLocation: String?
-    private var currentSFU: String?
-    private var participantsCountUpdatesTask: Task<Void, Never>?
-    private var currentUserBlockedTask: Task<Void, Never>?
 
     private let joinCallResponseSubject = CurrentValueSubject<JoinCallResponse?, Never>(nil)
     private var joinCallResponseFetchObserver: AnyCancellable?
@@ -74,7 +73,6 @@ class CallController: @unchecked Sendable {
         self.callType = callType
         self.apiKey = apiKey
         self.videoConfig = videoConfig
-        sfuReconnectionTime = 30
         self.defaultAPI = defaultAPI
         self.cachedLocation = cachedLocation
         self.webRTCCoordinatorFactory = webRTCCoordinatorFactory
@@ -562,7 +560,6 @@ class CallController: @unchecked Sendable {
 
     private func didFetch(_ response: JoinCallResponse) async {
         let sessionId = await webRTCCoordinator.stateAdapter.sessionID
-        currentSFU = response.credentials.server.edgeName
         Task { @MainActor [weak self] in
             self?.call?.state.sessionId = sessionId
             self?.call?.update(recordingState: response.call.recording ? .recording : .noRecording)
@@ -584,8 +581,7 @@ class CallController: @unchecked Sendable {
         case .joined:
             /// Once connected we should stop listening for CallSessionParticipantCountsUpdatedEvent
             /// updates and only rely on the healthCheck event.
-            participantsCountUpdatesTask?.cancel()
-            participantsCountUpdatesTask = nil
+            disposableBag.remove(DisposableKey.participantsCountUpdatesEvent.rawValue)
 
             call?.update(reconnectionStatus: .connected)
         case .error:
@@ -601,37 +597,36 @@ class CallController: @unchecked Sendable {
     }
 
     private func subscribeToParticipantsCountUpdatesEvent(_ call: Call?) {
-        participantsCountUpdatesTask?.cancel()
-        participantsCountUpdatesTask = nil
+        disposableBag.remove(DisposableKey.participantsCountUpdatesEvent.rawValue)
 
         guard let call else { return }
 
-        participantsCountUpdatesTask = Task {
-            let anonymousUserRoleKey = "anonymous"
-            for await event in call.subscribe(for: CallSessionParticipantCountsUpdatedEvent.self) {
-                Task { @MainActor in
-                    call.state.participantCount = event
-                        .participantsCountByRole
-                        .filter { $0.key != anonymousUserRoleKey } // TODO: Workaround. To be removed
-                        .values
-                        .map(UInt32.init)
-                        .reduce(0) { $0 + $1 }
-
-                    // TODO: Workaround. To be removed
-                    if event.anonymousParticipantCount > 0 {
-                        call.state.anonymousParticipantCount = UInt32(event.anonymousParticipantCount)
-                    } else if let anonymousCount = event.participantsCountByRole[anonymousUserRoleKey] {
-                        call.state.anonymousParticipantCount = UInt32(anonymousCount)
-                    } else {
-                        call.state.anonymousParticipantCount = 0
-                    }
+        let anonymousUserRoleKey = "anonymous"
+
+        call
+            .eventPublisher(for: CallSessionParticipantCountsUpdatedEvent.self)
+            .sinkTask { @MainActor [weak call] event in
+                call?.state.participantCount = event
+                    .participantsCountByRole
+                    .filter { $0.key != anonymousUserRoleKey } // TODO: Workaround. To be removed
+                    .values
+                    .map(UInt32.init)
+                    .reduce(0) { $0 + $1 }
+
+                // TODO: Workaround. To be removed
+                if event.anonymousParticipantCount > 0 {
+                    call?.state.anonymousParticipantCount = UInt32(event.anonymousParticipantCount)
+                } else if let anonymousCount = event.participantsCountByRole[anonymousUserRoleKey] {
+                    call?.state.anonymousParticipantCount = UInt32(anonymousCount)
+                } else {
+                    call?.state.anonymousParticipantCount = 0
                 }
             }
-        }
+            .store(in: disposableBag, key: DisposableKey.participantsCountUpdatesEvent.rawValue)
     }
 
     private func subscribeToCurrentUserBlockedState(_ call: Call?) {
-        disposableBag.remove("current-user-blocked")
+        disposableBag.remove(DisposableKey.currentUserBlocked.rawValue)
         guard let call else { return }
         let currentUser = user
         Task { @MainActor [weak self] in
@@ -651,7 +646,7 @@ class CallController: @unchecked Sendable {
                         .stateMachine
                         .transition(.blocked(self.webRTCCoordinator.stateMachine.currentStage.context))
                 }
-                .store(in: disposableBag, key: "current-user-blocked")
+                .store(in: disposableBag, key: DisposableKey.currentUserBlocked.rawValue)
         }
     }
 

Sources/StreamVideo/Utils/CallCache/CallCache.swift

@@ -29,6 +29,7 @@ final class CallCache {
                 log.debug("Will create and cache call:\(cId)")
                 let call = factory()
                 storage[cId] = call
+                log.debug("CallCache count:\(storage.count).")
                 return call
             }
         }
@@ -42,13 +43,15 @@ final class CallCache {
         log.debug("Will remove call:\(cId)")
         queue.sync {
             storage[cId] = nil
+            log.debug("CallCache count:\(storage.count).")
         }
     }
 
     func removeAll() {
         queue.sync {
             log.debug("Will remove \(storage.count) calls.")
             storage.removeAll()
+            log.debug("CallCache count:\(storage.count).")
         }
     }
 }

Sources/StreamVideo/Utils/Extensions/Combine/Publisher+AsyncStream.swift

@@ -0,0 +1,31 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Combine
+
+public extension Publisher where Output: Sendable {
+
+    /// Converts the current publisher into an `AsyncStream` of its output.
+    ///
+    /// This allows you to consume any Combine publisher using Swift's `for await`
+    /// syntax, providing a bridge between Combine and Swift Concurrency.
+    ///
+    /// - Returns: An `AsyncStream` emitting values of the publisher as they arrive.
+    func eraseAsAsyncStream() -> AsyncStream<Output> {
+        AsyncStream { continuation in
+            let cancellable = self.sink(
+                receiveCompletion: { _ in
+                    continuation.finish()
+                },
+                receiveValue: {
+                    continuation.yield($0)
+                }
+            )
+
+            continuation.onTermination = { @Sendable _ in
+                cancellable.cancel()
+            }
+        }
+    }
+}