[Fix]Reorganize Call state-machine and its transitions (#747)

Author: ipavlidakis

CHANGELOG.md

@@ -4,6 +4,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 # Upcoming
 
+### 🔄 Changed
+- Improved the flow of joining a call [#747](https://github.com/GetStream/stream-video-swift/pull/747)
+
 ### 🐞 Fixed
 - Fix an issue causing audio/video misalignment with the server. [#772](https://github.com/GetStream/stream-video-swift/pull/772)
 

Sources/StreamVideo/Call.swift

@@ -13,7 +13,7 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
     @Injected(\.streamVideo) var streamVideo
     @Injected(\.callCache) var callCache
 
-    private lazy var stateMachine: StreamCallStateMachine = .init(self)
+    private lazy var stateMachine: StateMachine = .init(self)
 
     @MainActor
     public internal(set) var state = CallState()
@@ -53,6 +53,9 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
     internal let coordinatorClient: DefaultAPI
     private var cancellables = DisposableBag()
 
+    /// A serialQueueActor ensuring that call operations (e.g. join) will happen in a serial manner.
+    private let callOperationSerialQueue = SerialActorQueue()
+
     /// This adapter is used to manage closed captions for the
     /// call.
     private lazy var closedCaptionsAdapter = ClosedCaptionsAdapter(self)
@@ -144,50 +147,42 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
         notify: Bool = false,
         callSettings: CallSettings? = nil
     ) async throws -> JoinCallResponse {
-        let currentStage = stateMachine.currentStage
-        switch currentStage.id {
-        case .joining:
-            break
-        case .joined where currentStage is StreamCallStateMachine.Stage.JoinedStage:
-            let stage = currentStage as! StreamCallStateMachine.Stage.JoinedStage
-            return stage.response
-        default:
-            stateMachine.transition(
-                .joining(
-                    self,
-                    actionBlock: { [weak self] in
-                        guard let self else { throw ClientError.Unexpected() }
-                        return try await executeTask(retryPolicy: .fastAndSimple, task: { [weak self] in
-                            guard let self else { throw ClientError.Unexpected() }
-                            let response = try await callController.joinCall(
+        try await callOperationSerialQueue.sync { [weak self] in
+            guard let self else {
+                throw ClientError()
+            }
+            let currentStage = stateMachine.currentStage
+
+            if
+                currentStage.id == .joined,
+                case let .joined(joinResponse) = currentStage.context.output {
+                return joinResponse
+            } else if
+                currentStage.id == .joining,
+                case let .join(input) = currentStage.context.input {
+                return try await input
+                    .deliverySubject
+                    .nextValue(timeout: CallConfiguration.timeout.join)
+            } else {
+                let deliverySubject = PassthroughSubject<JoinCallResponse, Error>()
+                stateMachine.transition(
+                    .joining(
+                        self,
+                        input: .join(
+                            .init(
                                 create: create,
                                 callSettings: callSettings,
                                 options: options,
                                 ring: ring,
-                                notify: notify
+                                notify: notify,
+                                deliverySubject: deliverySubject
                             )
-                            if let callSettings {
-                                await state.update(callSettings: callSettings)
-                            }
-                            await state.update(from: response)
-                            let updated = await state.callSettings
-                            updateCallSettingsManagers(with: updated)
-                            Task { @MainActor [weak self] in
-                                self?.streamVideo.state.activeCall = self
-                            }
-                            return response
-                        })
-                    }
+                        )
+                    )
                 )
-            )
+                return try await deliverySubject.nextValue(timeout: CallConfiguration.timeout.join)
+            }
         }
-
-        return try await stateMachine
-            .nextStageShouldBe(
-                StreamCallStateMachine.Stage.JoinedStage.self,
-                dropFirst: 1
-            )
-            .response
     }
 
     /// Gets the call on the backend with the given parameters.
@@ -350,24 +345,25 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
     @discardableResult
     public func accept() async throws -> AcceptCallResponse {
         let currentStage = stateMachine.currentStage
-        switch currentStage.id {
-        case .accepting:
-            break
-        case .accepted where currentStage is StreamCallStateMachine.Stage.AcceptedStage:
-            let stage = currentStage as! StreamCallStateMachine.Stage.AcceptedStage
-            return stage.response
-        default:
-            stateMachine.transition(.accepting(self, actionBlock: { [coordinatorClient, callType, callId] in
-                try await coordinatorClient.acceptCall(type: callType, id: callId)
-            }))
-        }
 
-        return try await stateMachine
-            .nextStageShouldBe(
-                StreamCallStateMachine.Stage.AcceptedStage.self,
-                dropFirst: 1
+        if
+            currentStage.id == .accepted,
+            case let .accepted(response) = currentStage.context.output {
+            return response
+        } else if
+            currentStage.id == .accepting,
+            case let .accepting(deliverySubject) = currentStage.context.input {
+            return try await deliverySubject.nextValue(timeout: CallConfiguration.timeout.accept)
+        } else {
+            let deliverySubject = PassthroughSubject<AcceptCallResponse, Error>()
+            stateMachine.transition(
+                .accepting(
+                    self,
+                    input: .accepting(deliverySubject: deliverySubject)
+                )
             )
-            .response
+            return try await deliverySubject.nextValue(timeout: CallConfiguration.timeout.accept)
+        }
     }
 
     /// Rejects a call with an optional reason.
@@ -378,34 +374,27 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
     @discardableResult
     public func reject(reason: String? = nil) async throws -> RejectCallResponse {
         let currentStage = stateMachine.currentStage
-        switch currentStage.id {
-        case .rejecting:
-            break
-        case .rejected where currentStage is StreamCallStateMachine.Stage.RejectedStage:
-            let stage = currentStage as! StreamCallStateMachine.Stage.RejectedStage
-            return stage.response
-        default:
-            stateMachine.transition(.rejecting(self, actionBlock: { [coordinatorClient, callType, callId, streamVideo, cId] in
-                let response = try await coordinatorClient.rejectCall(
-                    type: callType,
-                    id: callId,
-                    rejectCallRequest: .init(reason: reason)
-                )
-                if streamVideo.state.ringingCall?.cId == cId {
-                    Task { @MainActor in
-                        streamVideo.state.ringingCall = nil
-                    }
-                }
-                return response
-            }))
-        }
 
-        return try await stateMachine
-            .nextStageShouldBe(
-                StreamCallStateMachine.Stage.RejectedStage.self,
-                dropFirst: 1
+        if
+            currentStage.id == .rejected,
+            case let .rejected(response) = currentStage.context.output {
+            return response
+        } else if
+            currentStage.id == .rejecting,
+            case let .rejecting(input) = currentStage.context.input {
+            return try await input
+                .deliverySubject
+                .nextValue(timeout: CallConfiguration.timeout.reject)
+        } else {
+            let deliverySubject = PassthroughSubject<RejectCallResponse, Error>()
+            stateMachine.transition(
+                .rejecting(
+                    self,
+                    input: .rejecting(.init(deliverySubject: deliverySubject))
+                )
             )
-            .response
+            return try await deliverySubject.nextValue(timeout: CallConfiguration.timeout.reject)
+        }
     }
 
     /// Adds the given user to the list of blocked users for the call.
@@ -521,7 +510,7 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
         cancellables.removeAll()
         callController.leave()
         closedCaptionsAdapter.stop()
-        stateMachine.transition(.idle(self))
+        stateMachine.transition(.idle(.init(call: self)))
         /// Upon `Call.leave` we remove the call from the cache. Any further actions that are required
         /// to happen on the call object (e.g. rejoin) will need to fetch a new instance from `StreamVideo`
         /// client.
@@ -1391,7 +1380,12 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
         if stateMachine.currentStage.id == .joined {
             state.disconnectionError = error
         }
-        stateMachine.transition(.error(self, error: error))
+        stateMachine.transition(
+            .error(
+                .init(call: self),
+                error: error
+            )
+        )
     }
 
     // MARK: - private
@@ -1531,7 +1525,7 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
         }
     }
 
-    private func updateCallSettingsManagers(with callSettings: CallSettings) {
+    func updateCallSettingsManagers(with callSettings: CallSettings) {
         microphone.status = callSettings.audioOn ? .enabled : .disabled
         camera.status = callSettings.videoOn ? .enabled : .disabled
         camera.direction = callSettings.cameraPosition

Sources/StreamVideo/CallConfiguration.swift

@@ -0,0 +1,57 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Foundation
+
+/// Configuration settings for call operations.
+///
+/// This enum provides timeout values and other configuration settings
+/// for various call operations such as joining, accepting, and rejecting calls.
+enum CallConfiguration {
+
+    /// Timeout settings for different call operations.
+    struct Timeout {
+        /// Timeout duration for joining a call in seconds.
+        var join: TimeInterval
+        
+        /// Timeout duration for accepting a call in seconds.
+        var accept: TimeInterval
+        
+        /// Timeout duration for rejecting a call in seconds.
+        var reject: TimeInterval
+
+        /// Timeout values for authentication in production environment.
+        ///
+        /// These values are used when the app is running in production mode.
+        static let production = Timeout(
+            join: 30,
+            accept: 10,
+            reject: 10
+        )
+
+        #if STREAM_TESTS
+        /// Timeout values for authentication in test environment.
+        ///
+        /// These values are used when the app is running in test mode.
+        /// They are shorter than production values to speed up testing.
+        static let testing = Timeout(
+            join: 10,
+            accept: 10,
+            reject: 10
+        )
+        #endif
+    }
+
+    /// Timeout values for various Call operations.
+    ///
+    /// This property returns the appropriate timeout values based on the
+    /// current build configuration (test or production).
+    static var timeout: Timeout {
+        #if STREAM_TESTS
+        return .testing
+        #else
+        return .production
+        #endif
+    }
+}

Sources/StreamVideo/CallStateMachine/CallState+Machine.swift

@@ -0,0 +1,45 @@
+//
+// Copyright © 2025 Stream.io Inc. All rights reserved.
+//
+
+import Combine
+import Foundation
+
+/// A state machine that manages the states of a stream call.
+extension Call {
+    final class StateMachine {
+        /// The underlying state machine that handles state transitions.
+        private let stateMachine: StreamStateMachine<Stage>
+
+        /// The current stage of the state machine.
+        var currentStage: Stage { stateMachine.currentStage }
+
+        /// A publisher that emits the current stage of the state machine.
+        var publisher: AnyPublisher<Stage, Never> { stateMachine.publisher.eraseToAnyPublisher() }
+
+        /// Initializes the state machine with the idle stage for a given call.
+        ///
+        /// - Parameter call: The call to be managed by the state machine.
+        init(_ call: Call) {
+            stateMachine = .init(initialStage: .idle(.init(call: call)))
+        }
+
+        /// Transitions the state machine to the given next stage.
+        ///
+        /// - Parameter nextStage: The next stage to transition to.
+        /// - Throws: An error if the transition is invalid.
+        func transition(
+            _ nextStage: Stage,
+            file: StaticString = #file,
+            function: StaticString = #function,
+            line: UInt = #line
+        ) {
+            stateMachine.transition(
+                to: nextStage,
+                file: file,
+                function: function,
+                line: line
+            )
+        }
+    }
+}