[Fix]Remove unnecessary operation during tests and fix flakiness (#784)

Author: ipavlidakis

Sources/StreamVideo/Call.swift

@@ -158,10 +158,34 @@ public class Call: @unchecked Sendable, WSEventsSubscriber {
                 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
+                currentStage.id == .joining {
+                return try await stateMachine
+                    .publisher
+                    .tryCompactMap {
+                        switch $0.id {
+                        case .joined:
+                            guard
+                                let stage = $0 as? Call.StateMachine.Stage.JoinedStage
+                            else {
+                                throw ClientError()
+                            }
+                            switch stage.context.output {
+                            case let .joined(response):
+                                return response
+                            default:
+                                throw ClientError()
+                            }
+                        case .error:
+                            guard
+                                let stage = $0 as? Call.StateMachine.Stage.ErrorStage
+                            else {
+                                throw ClientError()
+                            }
+                            throw stage.error
+                        default:
+                            return nil
+                        }
+                    }
                     .nextValue(timeout: CallConfiguration.timeout.join)
             } else {
                 let deliverySubject = PassthroughSubject<JoinCallResponse, Error>()

Sources/StreamVideo/StreamVideo.swift

@@ -126,18 +126,41 @@ public class StreamVideo: ObservableObject, @unchecked Sendable {
             videoConfig: videoConfig,
             tokenProvider: tokenProvider ?? { _ in },
             pushNotificationsConfig: pushNotificationsConfig,
-            environment: Environment()
+            environment: Environment(),
+            autoConnectOnInit: true
         )
     }
-        
+
+    convenience init(
+        apiKey: String,
+        user: User,
+        token: UserToken,
+        videoConfig: VideoConfig = VideoConfig(),
+        pushNotificationsConfig: PushNotificationsConfig = .default,
+        tokenProvider: UserTokenProvider? = nil,
+        autoConnectOnInit: Bool
+    ) {
+        self.init(
+            apiKey: apiKey,
+            user: user,
+            token: token,
+            videoConfig: videoConfig,
+            tokenProvider: tokenProvider ?? { _ in },
+            pushNotificationsConfig: pushNotificationsConfig,
+            environment: Environment(),
+            autoConnectOnInit: autoConnectOnInit
+        )
+    }
+
     init(
         apiKey: String,
         user: User,
         token: UserToken,
         videoConfig: VideoConfig = VideoConfig(),
         tokenProvider: @escaping UserTokenProvider,
         pushNotificationsConfig: PushNotificationsConfig,
-        environment: Environment
+        environment: Environment,
+        autoConnectOnInit: Bool
     ) {
         self.apiKey = APIKey(apiKey)
         state = State(user: user)
@@ -188,32 +211,12 @@ public class StreamVideo: ObservableObject, @unchecked Sendable {
             coordinatorClient.middlewares.append(anonymousAuth)
         }
         prefetchLocation()
-        connectTask = Task {
-            if user.type == .guest {
-                do {
-                    try Task.checkCancellation()
-                    let guestInfo = try await loadGuestUserInfo(for: user, apiKey: apiKey)
-
-                    self.state.user = guestInfo.user
-                    self.token = guestInfo.token
-                    self.tokenProvider = guestInfo.tokenProvider
 
-                    try Task.checkCancellation()
-                    try await self.connectUser(isInitial: true)
-                } catch {
-                    log.error("Error connecting as guest", error: error)
-                }
-            } else if user.type != .anonymous {
-                do {
-                    try Task.checkCancellation()
-                    try await self.connectUser(isInitial: true)
-                } catch {
-                    log.error(error)
-                }
-            }
+        if autoConnectOnInit {
+            initialConnectIfRequired(apiKey: apiKey)
         }
     }
-    
+
     deinit {
         connectTask?.cancel()
         connectTask = nil
@@ -418,7 +421,42 @@ public class StreamVideo: ObservableObject, @unchecked Sendable {
     }
 
     // MARK: - private
-    
+
+    /// When initializing we perform an automatic connection attempt.
+    ///
+    /// - Important: This behaviour is only enabled for non-test environments. This is to reduce the
+    /// noise in logs and avoid unnecessary network operations with the backend.
+    private func initialConnectIfRequired(apiKey: String) {
+        guard connectTask == nil else {
+            return
+        }
+
+        connectTask = Task {
+            if user.type == .guest {
+                do {
+                    try Task.checkCancellation()
+                    let guestInfo = try await loadGuestUserInfo(for: user, apiKey: apiKey)
+
+                    self.state.user = guestInfo.user
+                    self.token = guestInfo.token
+                    self.tokenProvider = guestInfo.tokenProvider
+
+                    try Task.checkCancellation()
+                    try await self.connectUser(isInitial: true)
+                } catch {
+                    log.error("Error connecting as guest", error: error)
+                }
+            } else if user.type != .anonymous {
+                do {
+                    try Task.checkCancellation()
+                    try await self.connectUser(isInitial: true)
+                } catch {
+                    log.error(error)
+                }
+            }
+        }
+    }
+
     /// Creates a call controller, used for establishing and managing a call.
     /// - Parameters:
     ///    - callType: the type of the call.

Sources/StreamVideo/Utils/StateMachine/Publisher+NextValue.swift

@@ -54,7 +54,11 @@ extension Publisher where Output: Sendable {
                 receiveValue: { value in
                     timeoutWorkItem?.cancel()
                     if !receivedValue {
-                        continuation.resume(returning: value) // Resume only if value hasn't been received
+                        if let error = value as? Error {
+                            continuation.resume(throwing: error)
+                        } else {
+                            continuation.resume(returning: value) // Resume only if value hasn't been received
+                        }
                         receivedValue = true
                     }
                     cancellable?.cancel()