diff --git a/Sources/StreamAttachments/CDNClient/CDNClient.swift b/Sources/StreamAttachments/CDNClient/CDNClient.swift index 59b5e3a..b715831 100644 --- a/Sources/StreamAttachments/CDNClient/CDNClient.swift +++ b/Sources/StreamAttachments/CDNClient/CDNClient.swift @@ -123,8 +123,8 @@ public class StreamCDNClient: CDNClient, @unchecked Sendable { return } - let data = multipartFormData.getMultipartFormData() - urlRequest.addValue("multipart/form-data; boundary=\(MultipartFormData.boundary)", forHTTPHeaderField: "Content-Type") + let data = multipartFormData.encode() + urlRequest.addValue("multipart/form-data; boundary=\(multipartFormData.boundary)", forHTTPHeaderField: "Content-Type") urlRequest.addValue("stream-feeds-swift-v0.0.1", forHTTPHeaderField: "X-Stream-Client") // TODO: fix this. urlRequest.httpBody = data diff --git a/Sources/StreamAttachments/CDNClient/MultipartFormData.swift b/Sources/StreamAttachments/CDNClient/MultipartFormData.swift deleted file mode 100644 index d61eaae..0000000 --- a/Sources/StreamAttachments/CDNClient/MultipartFormData.swift +++ /dev/null @@ -1,45 +0,0 @@ -// -// Copyright © 2026 Stream.io Inc. All rights reserved. -// - -import Foundation - -struct MultipartFormData { - private static let crlf = "\r\n" - static let boundary: String = String( - format: "chat-%08x%08x", - UInt32.random(in: 0...UInt32.max), - UInt32.random(in: 0...UInt32.max) - ) - - let data: Data - let fileName: String - let mimeType: String? - - init(_ data: Data, fileName: String, mimeType: String? = nil) { - self.data = data - self.fileName = fileName - self.mimeType = mimeType - } - - func getMultipartFormData() -> Data { - var data = "--\(Self.boundary)\(MultipartFormData.crlf)".data(using: .utf8, allowLossyConversion: false)! - data.append("Content-Disposition: form-data; name=\"file\"; filename=\"\(fileName)\"\(MultipartFormData.crlf)") - - if let mimeType { - data.append("Content-Type: \(mimeType)\(MultipartFormData.crlf)") - } - - data.append(MultipartFormData.crlf) - data.append(self.data) - data.append("\(MultipartFormData.crlf)--\(Self.boundary)--\(MultipartFormData.crlf)") - - return data - } -} - -private extension Data { - mutating func append(_ string: String, encoding: String.Encoding = .utf8) { - append(string.data(using: encoding, allowLossyConversion: false)!) - } -} diff --git a/Sources/StreamAttachments/Info.plist b/Sources/StreamAttachments/Info.plist index 4f01845..1e85a1a 100644 --- a/Sources/StreamAttachments/Info.plist +++ b/Sources/StreamAttachments/Info.plist @@ -15,7 +15,7 @@ CFBundlePackageType $(PRODUCT_BUNDLE_PACKAGE_TYPE) CFBundleShortVersionString - 0.9.0 + 0.10.0 CFBundleVersion $(CURRENT_PROJECT_VERSION) diff --git a/Sources/StreamCore/Info.plist b/Sources/StreamCore/Info.plist index 4f01845..1e85a1a 100644 --- a/Sources/StreamCore/Info.plist +++ b/Sources/StreamCore/Info.plist @@ -15,7 +15,7 @@ CFBundlePackageType $(PRODUCT_BUNDLE_PACKAGE_TYPE) CFBundleShortVersionString - 0.9.0 + 0.10.0 CFBundleVersion $(CURRENT_PROJECT_VERSION) diff --git a/Sources/StreamCore/Logger/Logger.swift b/Sources/StreamCore/Logger/Logger.swift index 1ca8f40..45193ee 100644 --- a/Sources/StreamCore/Logger/Logger.swift +++ b/Sources/StreamCore/Logger/Logger.swift @@ -2,6 +2,7 @@ // Copyright © 2026 Stream.io Inc. All rights reserved. // +import Combine import Foundation public var log: Logger { @@ -251,9 +252,17 @@ public enum LogConfig { $0.level = newValue $0.invalidateLogger() } + Self.levelSubject.send(newValue) } } - + + private nonisolated(unsafe) static let levelSubject = CurrentValueSubject(_state.value.level) + + /// Publishes the current log level and all subsequent changes. + public static var levelPublisher: AnyPublisher { + levelSubject.eraseToAnyPublisher() + } + /// Date formatter for the logger. Defaults to ISO8601 public static var dateFormatter: DateFormatter { get { @@ -432,11 +441,12 @@ public enum LogConfig { static func reset() { _state.withLock { $0 = State() } + levelSubject.send(level) } } /// Entity used for logging messages. -open class Logger { +open class Logger: @unchecked Sendable { /// Identifier of the Logger. Will be visible if a destination has `showIdentifiers` enabled. public let identifier: String diff --git a/Sources/StreamCore/Utils/DisposableBag.swift b/Sources/StreamCore/Utils/DisposableBag.swift index de38fc3..23f0269 100644 --- a/Sources/StreamCore/Utils/DisposableBag.swift +++ b/Sources/StreamCore/Utils/DisposableBag.swift @@ -24,9 +24,11 @@ public final class DisposableBag: @unchecked Sendable { } } - func remove(_ key: String) { + func remove(_ key: String, cancel: Bool) { queue.sync { - storage[key]?.cancel() + if cancel { + storage[key]?.cancel() + } storage[key] = nil } } @@ -53,7 +55,7 @@ public final class DisposableBag: @unchecked Sendable { } public func remove(_ key: String) { - storage.remove(key) + storage.remove(key, cancel: true) } public func removeAll() { @@ -61,6 +63,10 @@ public final class DisposableBag: @unchecked Sendable { } public var isEmpty: Bool { storage.isEmpty } + + public func completed(_ key: String) { + storage.remove(key, cancel: false) + } } extension AnyCancellable { diff --git a/Sources/StreamCore/Utils/MultipartFormData.swift b/Sources/StreamCore/Utils/MultipartFormData.swift new file mode 100644 index 0000000..61af17d --- /dev/null +++ b/Sources/StreamCore/Utils/MultipartFormData.swift @@ -0,0 +1,53 @@ +// +// Copyright © 2026 Stream.io Inc. All rights reserved. +// + +import Foundation + +/// Encodes a file as multipart form data. +public struct MultipartFormData: Sendable { + private static let crlf = "\r\n" + + @usableFromInline static let defaultBoundary: String = String( + format: "chat-%08x%08x", + UInt32.random(in: 0...UInt32.max), + UInt32.random(in: 0...UInt32.max) + ) + + private let data: Data + private let fileName: String + private let mimeType: String? + + /// The boundary used to separate multipart form data parts. + public let boundary: String + + /// Creates multipart form data for a file. + public init(_ data: Data, fileName: String, mimeType: String? = nil, boundary: String = Self.defaultBoundary) { + self.data = data + self.fileName = fileName + self.mimeType = mimeType + self.boundary = boundary + } + + /// Encodes the file as a multipart form data body. + public func encode() -> Data { + var data = "--\(boundary)\(Self.crlf)".data(using: .utf8, allowLossyConversion: false)! + data.append("Content-Disposition: form-data; name=\"file\"; filename=\"\(fileName)\"\(Self.crlf)") + + if let mimeType { + data.append("Content-Type: \(mimeType)\(Self.crlf)") + } + + data.append(Self.crlf) + data.append(self.data) + data.append("\(Self.crlf)--\(boundary)--\(Self.crlf)") + + return data + } +} + +private extension Data { + mutating func append(_ string: String, encoding: String.Encoding = .utf8) { + append(string.data(using: encoding, allowLossyConversion: false)!) + } +} diff --git a/Sources/StreamCore/Utils/SystemEnvironment+Version.swift b/Sources/StreamCore/Utils/SystemEnvironment+Version.swift index 6d4dcd1..6367507 100644 --- a/Sources/StreamCore/Utils/SystemEnvironment+Version.swift +++ b/Sources/StreamCore/Utils/SystemEnvironment+Version.swift @@ -6,5 +6,5 @@ import Foundation enum SystemEnvironment { /// A Stream Core version. - public static let version: String = "0.9.0" + public static let version: String = "0.10.0" } diff --git a/Sources/StreamCore/WebSocket/Client/WebSocketClient.swift b/Sources/StreamCore/WebSocket/Client/WebSocketClient.swift index f7d9267..c67017f 100644 --- a/Sources/StreamCore/WebSocket/Client/WebSocketClient.swift +++ b/Sources/StreamCore/WebSocket/Client/WebSocketClient.swift @@ -5,6 +5,62 @@ import Combine import Foundation +/// Describes why a client-initiated WebSocket close is requested. +public enum WebSocketCloseContext: Equatable, Sendable { + /// Closes the current socket because a disconnection was requested or + /// detected. + /// + /// The source lets a provider distinguish cases such as a user request, + /// automatic recovery, or a failed health check and select the appropriate + /// close code. + case disconnection(source: WebSocketConnectionState.DisconnectionSource) + + /// Closes the current socket so it can be replaced with new connection + /// settings while the higher-level operation continues. + /// + /// For example, Stream Video uses this context when replacing its SFU + /// signaling socket after receiving an updated WebSocket configuration. + case reconfiguration + + /// Closes the current socket with a protocol-specific code requested by the + /// caller. + /// + /// This supports integrations that already know the required close code. + /// The provider receives that code and may preserve or override it. + case explicit( + code: URLSessionWebSocketTask.CloseCode, + source: WebSocketConnectionState.DisconnectionSource + ) + + var disconnectionSource: WebSocketConnectionState.DisconnectionSource { + switch self { + case let .disconnection(source), let .explicit(_, source): + return source + case .reconfiguration: + return .userInitiated + } + } +} + +/// Provides the close code for a client-initiated WebSocket close. +public protocol WebSocketCloseCodeProviding: Sendable { + func closeCode(for context: WebSocketCloseContext) -> URLSessionWebSocketTask.CloseCode +} + +/// Preserves the default WebSocket close-code behavior. +public struct DefaultWebSocketCloseCodeProvider: WebSocketCloseCodeProviding { + public init() {} + + public func closeCode(for context: WebSocketCloseContext) -> URLSessionWebSocketTask.CloseCode { + switch context { + case let .explicit(code, _): + return code + case .disconnection, .reconfiguration: + return .normalClosure + } + } +} + public class WebSocketClient: @unchecked Sendable { /// The notification center `WebSocketClient` uses to send notifications about incoming events. public let eventNotificationCenter: EventNotificationCenter @@ -30,9 +86,16 @@ public class WebSocketClient: @unchecked Sendable { } } - let connectionSubject = PassthroughSubject() + private let connectionSubject = CurrentValueSubject< + WebSocketConnectionState, + Never + >(.initialized) public let eventSubject = PassthroughSubject() + public var connectionStatePublisher: AnyPublisher { + connectionSubject.eraseToAnyPublisher() + } + public weak var connectionStateDelegate: ConnectionStateDelegate? public var connectRequest: URLRequest? { @@ -69,6 +132,7 @@ public class WebSocketClient: @unchecked Sendable { private let environment: Environment private let webSocketClientType: WebSocketClientType + private let closeCodeProvider: any WebSocketCloseCodeProviding let pingController: WebSocketPingController @@ -85,6 +149,9 @@ public class WebSocketClient: @unchecked Sendable { public var onWSConnectionEstablished: (() -> Void)? public var onConnected: (() -> Void)? + /// Creates a WebSocket client. + /// - Parameter pingInterval: The interval between WebSocket keep-alive pings. + /// Defaults to 25 seconds. public init( sessionConfiguration: URLSessionConfiguration, eventDecoder: AnyEventDecoder, @@ -94,6 +161,8 @@ public class WebSocketClient: @unchecked Sendable { connectRequest: URLRequest?, healthCheckBeforeConnected: Bool = false, requiresAuth: Bool = true, + pingInterval: TimeInterval = 25, + closeCodeProvider: any WebSocketCloseCodeProviding = DefaultWebSocketCloseCodeProvider(), pingRequestBuilder: (() -> any SendableEvent)? = nil ) { self.environment = environment @@ -104,10 +173,12 @@ public class WebSocketClient: @unchecked Sendable { self.eventNotificationCenter = eventNotificationCenter self.healthCheckBeforeConnected = healthCheckBeforeConnected self.requiresAuth = requiresAuth + self.closeCodeProvider = closeCodeProvider pingController = environment.createPingController( environment.timerType, engineQueue, - webSocketClientType + webSocketClientType, + pingInterval ) pingController.pingRequestBuilder = pingRequestBuilder @@ -143,16 +214,46 @@ public class WebSocketClient: @unchecked Sendable { } } - /// Disconnects the web socket. + /// Disconnects using the existing close-code-based API. /// - /// Calling this function has no effect, if the connection is in an inactive state. - /// - Parameter source: Additional information about the source of the disconnection. Default value is `.userInitiated`. + /// The close-code provider receives an + /// ``WebSocketCloseContext/explicit(code:source:)`` context and chooses the + /// final close code. Prefer ``disconnect(context:completion:)`` for new + /// code that can describe why the socket is closing without selecting a + /// close code. + /// + /// - Parameters: + /// - code: The requested close code. The provider may return a different + /// code. + /// - source: The source recorded in the connection state. + /// - completion: Called after pending batched events are processed. public func disconnect( code: URLSessionWebSocketTask.CloseCode = .normalClosure, source: WebSocketConnectionState.DisconnectionSource = .userInitiated, completion: @Sendable @escaping () -> Void ) { - connectionState = .disconnecting(source: source) + disconnect( + context: .explicit(code: code, source: source), + completion: completion + ) + } + + /// Disconnects by describing why the socket is closing. + /// + /// Use this API for new integrations. The close-code provider receives the + /// context and chooses the final close code. For example, use + /// ``WebSocketCloseContext/reconfiguration`` when replacing a socket so the + /// provider can select a product-specific reconfiguration code. + /// + /// - Parameters: + /// - context: The reason for closing the socket. + /// - completion: Called after pending batched events are processed. + public func disconnect( + context: WebSocketCloseContext, + completion: @Sendable @escaping () -> Void + ) { + connectionState = .disconnecting(source: context.disconnectionSource) + let code = closeCodeProvider.closeCode(for: context) engineQueue.async { [engine, eventsBatcher] in engine?.disconnect(with: code) @@ -160,13 +261,21 @@ public class WebSocketClient: @unchecked Sendable { } } + /// Asynchronously disconnects with the given disconnection source. + /// + /// The close-code provider receives a + /// ``WebSocketCloseContext/disconnection(source:)`` context and chooses the + /// final close code. The call returns after pending batched events are + /// processed, not after the remote peer acknowledges the closure. + /// + /// - Parameter source: The source recorded in the connection state. public func disconnect(source: WebSocketConnectionState.DisconnectionSource = .userInitiated) async { await withCheckedContinuation { [weak self] continuation in guard let self else { continuation.resume() return } - disconnect { + disconnect(context: .disconnection(source: source)) { continuation.resume() } } @@ -192,7 +301,8 @@ public extension WebSocketClient { typealias CreatePingController = ( _ timerType: TimerScheduling.Type, _ timerQueue: DispatchQueue, - _ webSocketClientType: WebSocketClientType + _ webSocketClientType: WebSocketClientType, + _ pingInterval: TimeInterval ) -> WebSocketPingController typealias CreateEngine = ( @@ -280,6 +390,8 @@ extension WebSocketClient: WebSocketEngineDelegate { if let error = event.error() { log.error("Received an error webSocket event.", subsystems: .webSocket, error: error) + // Publish before disconnecting so observers receive the terminal event. + eventSubject.send(event) connectionState = .disconnecting(source: .serverInitiated(error: ClientError(with: error))) return } else { @@ -354,7 +466,7 @@ extension WebSocketClient: WebSocketPingControllerDelegate { func disconnectOnNoPongReceived() { log.debug("disconnecting from \(String(describing: connectRequest?.url))", subsystems: .webSocket) - disconnect(source: .noPongReceived) { + disconnect(context: .disconnection(source: .noPongReceived)) { log.debug("Websocket is disconnected because of no pong received", subsystems: .webSocket) } } diff --git a/Sources/StreamCore/WebSocket/Client/WebSocketPingController.swift b/Sources/StreamCore/WebSocket/Client/WebSocketPingController.swift index c85238f..a6cdbb5 100644 --- a/Sources/StreamCore/WebSocket/Client/WebSocketPingController.swift +++ b/Sources/StreamCore/WebSocket/Client/WebSocketPingController.swift @@ -21,13 +21,12 @@ public protocol HealthCheck: Event, Equatable {} /// The controller manages ping and pong timers. It sends ping periodically to keep a web socket connection alive. /// After ping is sent, a pong waiting timer is started, and if pong does not come, a forced disconnect is called. class WebSocketPingController { - /// The time interval to ping connection to keep it alive. - static let pingTimeInterval: TimeInterval = 25 /// The time interval for pong timeout. static let pongTimeoutTimeInterval: TimeInterval = 3 private let timerType: TimerScheduling.Type private let timerQueue: DispatchQueue + private let pingInterval: TimeInterval /// The timer used for scheduling `ping` calls private var pingTimerControl: RepeatingTimerControl? @@ -53,11 +52,13 @@ class WebSocketPingController { init( timerType: TimerScheduling.Type, timerQueue: DispatchQueue, - webSocketClientType: WebSocketClientType + webSocketClientType: WebSocketClientType, + pingInterval: TimeInterval ) { self.timerType = timerType self.timerQueue = timerQueue self.webSocketClientType = webSocketClientType + self.pingInterval = pingInterval } /// `WebSocketClient` should call this when the connection state did change. @@ -98,7 +99,10 @@ class WebSocketPingController { private func schedulePingTimerIfNeeded() { guard pingTimerControl == nil else { return } - pingTimerControl = timerType.scheduleRepeating(timeInterval: Self.pingTimeInterval, queue: timerQueue) { [weak self] in + pingTimerControl = timerType.scheduleRepeating( + timeInterval: pingInterval, + queue: timerQueue + ) { [weak self] in self?.sendPing() } } diff --git a/Sources/StreamCoreUI/Info.plist b/Sources/StreamCoreUI/Info.plist index 4f01845..1e85a1a 100644 --- a/Sources/StreamCoreUI/Info.plist +++ b/Sources/StreamCoreUI/Info.plist @@ -15,7 +15,7 @@ CFBundlePackageType $(PRODUCT_BUNDLE_PACKAGE_TYPE) CFBundleShortVersionString - 0.9.0 + 0.10.0 CFBundleVersion $(CURRENT_PROJECT_VERSION) diff --git a/Tests/StreamCoreTests/Logger/Logger_Tests.swift b/Tests/StreamCoreTests/Logger/Logger_Tests.swift index da06be1..4db25dc 100644 --- a/Tests/StreamCoreTests/Logger/Logger_Tests.swift +++ b/Tests/StreamCoreTests/Logger/Logger_Tests.swift @@ -2,10 +2,55 @@ // Copyright © 2026 Stream.io Inc. All rights reserved. // +import Combine +import Foundation @testable import StreamCore import Testing +@Suite(.serialized) struct Logger_Tests { + @Test func concurrentLoggingProcessesAllMessages() async throws { + let iterations = 200 + let destination = CapturingDestination() + let logger = Logger(identifier: "test", destinations: [destination]) + + await withTaskGroup(of: Void.self) { group in + for index in 0..([]) + let cancellable = LogConfig.levelPublisher.sink { level in + #expect(LogConfig.level == level) + receivedLevels.withLock { $0.append(level) } + } + + withExtendedLifetime(cancellable) { + LogConfig.level = .debug + } + + #expect(receivedLevels.value == [.error, .debug]) + } + // MARK: - Concurrent Logger Invalidation Tests @Test func concurrentLoggerInvalidation() async throws { @@ -141,3 +186,79 @@ struct Logger_Tests { LogConfig.reset() } } + +private final class CapturingDestination: BaseLogDestination, @unchecked Sendable { + private var logDetails: [LogDetails] = [] + private let lock = NSLock() + + init() { + let formatter = DateFormatter() + formatter.dateFormat = "yyyy-MM-dd HH:mm:ss.SSS" + super.init( + identifier: UUID().uuidString, + level: .debug, + subsystems: .all, + showDate: false, + dateFormatter: formatter, + formatters: [], + showLevel: false, + showIdentifier: false, + showThreadName: true, + showFileName: false, + showLineNumber: false, + showFunctionName: false + ) + } + + @available(*, unavailable) + required init( + identifier: String, + level: LogLevel, + subsystems: LogSubsystem, + showDate: Bool, + dateFormatter: DateFormatter, + formatters: [LogFormatter], + showLevel: Bool, + showIdentifier: Bool, + showThreadName: Bool, + showFileName: Bool, + showLineNumber: Bool, + showFunctionName: Bool + ) { + super.init( + identifier: identifier, + level: level, + subsystems: subsystems, + showDate: showDate, + dateFormatter: dateFormatter, + formatters: formatters, + showLevel: showLevel, + showIdentifier: showIdentifier, + showThreadName: showThreadName, + showFileName: showFileName, + showLineNumber: showLineNumber, + showFunctionName: showFunctionName + ) + fatalError("Unsupported initializer") + } + + override func process(logDetails: LogDetails) { + lock.lock() + self.logDetails.append(logDetails) + lock.unlock() + } + + override func write(message: String) {} + + var processedCount: Int { + lock.lock() + defer { lock.unlock() } + return logDetails.count + } + + var recordedThreadNames: [String] { + lock.lock() + defer { lock.unlock() } + return logDetails.map(\.threadName) + } +} diff --git a/Tests/StreamCoreTests/Mocks/WebSocketEngine_Mock.swift b/Tests/StreamCoreTests/Mocks/WebSocketEngine_Mock.swift index c84b7b2..6a350d2 100644 --- a/Tests/StreamCoreTests/Mocks/WebSocketEngine_Mock.swift +++ b/Tests/StreamCoreTests/Mocks/WebSocketEngine_Mock.swift @@ -17,6 +17,7 @@ final class WebSocketEngine_Mock: WebSocketEngine, @unchecked Sendable { /// How many times was `disconnect()` called @Atomic var disconnect_calledCount = 0 + @Atomic var disconnect_closeCode: URLSessionWebSocketTask.CloseCode? /// How many times was `sendPing()` called @Atomic var sendPing_calledCount = 0 @@ -40,6 +41,7 @@ final class WebSocketEngine_Mock: WebSocketEngine, @unchecked Sendable { } func disconnect(with code: URLSessionWebSocketTask.CloseCode) { + disconnect_closeCode = code disconnect_calledCount += 1 } diff --git a/Tests/StreamCoreTests/Utils/DisposableBag_Tests.swift b/Tests/StreamCoreTests/Utils/DisposableBag_Tests.swift new file mode 100644 index 0000000..24c8746 --- /dev/null +++ b/Tests/StreamCoreTests/Utils/DisposableBag_Tests.swift @@ -0,0 +1,35 @@ +// +// Copyright © 2026 Stream.io Inc. All rights reserved. +// + +import Combine +@testable import StreamCore +import XCTest + +final class DisposableBag_Tests: XCTestCase, @unchecked Sendable { + func test_completed_removesCancellableWithoutCancelling() { + let subject = DisposableBag() + var wasCancelled = false + let cancellable = AnyCancellable { wasCancelled = true } + + subject.insert(cancellable, with: "task") + subject.completed("task") + + XCTAssertTrue(subject.isEmpty) + XCTAssertFalse(wasCancelled) + withExtendedLifetime(cancellable) {} + } + + func test_remove_removesAndCancelsCancellable() { + let subject = DisposableBag() + var wasCancelled = false + let cancellable = AnyCancellable { wasCancelled = true } + + subject.insert(cancellable, with: "task") + subject.remove("task") + + XCTAssertTrue(subject.isEmpty) + XCTAssertTrue(wasCancelled) + withExtendedLifetime(cancellable) {} + } +} diff --git a/Tests/StreamCoreTests/Utils/MultipartFormData_Tests.swift b/Tests/StreamCoreTests/Utils/MultipartFormData_Tests.swift new file mode 100644 index 0000000..280fe96 --- /dev/null +++ b/Tests/StreamCoreTests/Utils/MultipartFormData_Tests.swift @@ -0,0 +1,80 @@ +// +// Copyright © 2026 Stream.io Inc. All rights reserved. +// + +import Foundation +import StreamCore +import Testing + +struct MultipartFormData_Tests { + @Test func `Encode uses the generated boundary by default`() { + let encodedData = MultipartFormData(Data(), fileName: "attachment.bin").encode() + let encodedString = String(decoding: encodedData, as: UTF8.self) + let boundary = String(encodedString.dropFirst(2).prefix(21)) + + #expect(boundary.hasPrefix("chat-")) + #expect(boundary.count == 21) + #expect(boundary.dropFirst(5).allSatisfy { $0.isHexDigit }) + #expect(encodedString.hasSuffix("\r\n--\(boundary)--\r\n")) + } + + @Test func `Encode uses the provided boundary`() { + let payload = Data("content".utf8) + let multipartFormData = MultipartFormData( + payload, + fileName: "attachment.bin", + boundary: "test-boundary" + ) + + #expect(multipartFormData.boundary == "test-boundary") + #expect(multipartFormData.encode() == expectedData( + payload: payload, + fileName: "attachment.bin", + mimeType: nil, + boundary: "test-boundary" + )) + } + + @Test func `Encode includes MIME type and preserves binary payload`() { + let payload = Data([0x00, 0xff, 0x01]) + let multipartFormData = MultipartFormData( + payload, + fileName: "attachment.bin", + mimeType: "application/octet-stream", + boundary: "test-boundary" + ) + + #expect(multipartFormData.encode() == expectedData( + payload: payload, + fileName: "attachment.bin", + mimeType: "application/octet-stream", + boundary: "test-boundary" + )) + } + + @Test func `Encode omits MIME type header when no MIME type is provided`() { + let payload = Data("content".utf8) + let multipartFormData = MultipartFormData(payload, fileName: "attachment.bin", boundary: "test-boundary") + + #expect(multipartFormData.encode() == expectedData( + payload: payload, + fileName: "attachment.bin", + mimeType: nil, + boundary: "test-boundary" + )) + } + + private func expectedData(payload: Data, fileName: String, mimeType: String?, boundary: String) -> Data { + var expected = Data("--\(boundary)\r\n".utf8) + expected.append(Data("Content-Disposition: form-data; name=\"file\"; filename=\"\(fileName)\"\r\n".utf8)) + + if let mimeType { + expected.append(Data("Content-Type: \(mimeType)\r\n".utf8)) + } + + expected.append(Data("\r\n".utf8)) + expected.append(payload) + expected.append(Data("\r\n--\(boundary)--\r\n".utf8)) + return expected + } +} diff --git a/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketClient_Tests.swift b/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketClient_Tests.swift index 5a11a75..25000f0 100644 --- a/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketClient_Tests.swift +++ b/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketClient_Tests.swift @@ -23,6 +23,7 @@ final class WebSocketClient_Tests: XCTestCase, @unchecked Sendable { private var eventNotificationCenterMiddleware: EventMiddleware_Mock! private let connectURL = URL(string: "http://example.com/ws")! + private let pingInterval: TimeInterval = 5 private let createdAt = Date() private lazy var healthCheckInfo = HealthCheckInfo( @@ -50,7 +51,8 @@ final class WebSocketClient_Tests: XCTestCase, @unchecked Sendable { eventNotificationCenter: eventNotificationCenter, webSocketClientType: .coordinator, environment: environment, - connectRequest: URLRequest(url: connectURL) + connectRequest: URLRequest(url: connectURL), + pingInterval: pingInterval ) connectionId = UUID().uuidString @@ -148,6 +150,54 @@ final class WebSocketClient_Tests: XCTestCase, @unchecked Sendable { // Assert disconnect is called AssertAsync.willBeEqual(engine!.disconnect_calledCount, 1) + XCTAssertEqual(engine!.disconnect_closeCode, .normalClosure) + } + + func test_disconnect_withExplicitCode_usesCloseCodeProviderDecision() { + let providerCloseCode = URLSessionWebSocketTask.CloseCode(rawValue: 4001)! + let requestedCloseCode = URLSessionWebSocketTask.CloseCode(rawValue: 4002)! + let closeCodeProvider = setUpWebSocketClient(closeCode: providerCloseCode) + + webSocketClient.disconnect( + code: requestedCloseCode, + source: .userInitiated + ) {} + + AssertAsync.willBeEqual(engine!.disconnect_calledCount, 1) + XCTAssertEqual(engine!.disconnect_closeCode, providerCloseCode) + XCTAssertEqual( + closeCodeProvider.receivedContext, + .explicit(code: requestedCloseCode, source: .userInitiated) + ) + } + + func test_disconnect_withExplicitCode_defaultProviderUsesRequestedCode() { + let requestedCloseCode = URLSessionWebSocketTask.CloseCode(rawValue: 4002)! + webSocketClient.connect() + AssertAsync.willBeEqual(engine!.connect_calledCount, 1) + + webSocketClient.disconnect( + code: requestedCloseCode, + source: .userInitiated + ) {} + + AssertAsync.willBeEqual(engine!.disconnect_calledCount, 1) + XCTAssertEqual(engine!.disconnect_closeCode, requestedCloseCode) + } + + func test_disconnect_withReconfigurationContext_usesCloseCodeProviderDecision() { + let providerCloseCode = URLSessionWebSocketTask.CloseCode(rawValue: 4002)! + let closeCodeProvider = setUpWebSocketClient(closeCode: providerCloseCode) + + webSocketClient.disconnect(context: .reconfiguration) {} + + AssertAsync.willBeEqual(engine!.disconnect_calledCount, 1) + XCTAssertEqual( + webSocketClient.connectionState, + .disconnecting(source: .userInitiated) + ) + XCTAssertEqual(engine!.disconnect_closeCode, providerCloseCode) + XCTAssertEqual(closeCodeProvider.receivedContext, .reconfiguration) } func test_whenConnectedAndEngineDisconnectsWithServerError_itIsTreatedAsServerInitiatedDisconnect() { @@ -242,6 +292,15 @@ final class WebSocketClient_Tests: XCTestCase, @unchecked Sendable { AssertAsync.willBeEqual(engine!.sendPing_calledCount, 1) } + func test_webSocketPingController_usesConfiguredPingInterval() { + test_connectionFlow() + engine!.sendPing_calledCount = 0 + + time.run(numberOfSeconds: pingInterval + 1) + + XCTAssertEqual(engine!.sendPing_calledCount, 1) + } + func test_pongReceived_callsPingController_pongReceived() { // Simulate connection to make sure web socket engine exists test_connectionFlow() @@ -266,6 +325,24 @@ final class WebSocketClient_Tests: XCTestCase, @unchecked Sendable { Assert.willBeEqual(self.webSocketClient.connectionState, .disconnecting(source: .noPongReceived)) Assert.willBeEqual(self.engine!.disconnect_calledCount, 1) } + XCTAssertEqual(engine!.disconnect_closeCode, .normalClosure) + } + + func test_webSocketPingController_disconnectOnNoPongReceived_usesCloseCodeProvider() { + let providerCloseCode = URLSessionWebSocketTask.CloseCode(rawValue: 4001)! + let closeCodeProvider = setUpWebSocketClient(closeCode: providerCloseCode) + + pingController.delegate?.disconnectOnNoPongReceived() + + AssertAsync { + Assert.willBeEqual(self.webSocketClient.connectionState, .disconnecting(source: .noPongReceived)) + Assert.willBeEqual(self.engine!.disconnect_calledCount, 1) + } + XCTAssertEqual(engine!.disconnect_closeCode, providerCloseCode) + XCTAssertEqual( + closeCodeProvider.receivedContext, + .disconnection(source: .noPongReceived) + ) } // MARK: - Event handling tests @@ -360,6 +437,41 @@ final class WebSocketClient_Tests: XCTestCase, @unchecked Sendable { self.webSocketClient.connect() }) } + + private func setUpWebSocketClient( + closeCode: URLSessionWebSocketTask.CloseCode + ) -> WebSocketCloseCodeProvider_Mock { + let closeCodeProvider = WebSocketCloseCodeProvider_Mock(closeCode: closeCode) + var environment = WebSocketClient.Environment.mock + environment.timerType = VirtualTimeTimer.self + webSocketClient = WebSocketClient( + sessionConfiguration: .ephemeral, + eventDecoder: decoder, + eventNotificationCenter: eventNotificationCenter, + webSocketClientType: .coordinator, + environment: environment, + connectRequest: URLRequest(url: connectURL), + pingInterval: pingInterval, + closeCodeProvider: closeCodeProvider + ) + webSocketClient.connect() + AssertAsync.willBeEqual(engine!.connect_calledCount, 1) + return closeCodeProvider + } +} + +private final class WebSocketCloseCodeProvider_Mock: WebSocketCloseCodeProviding, @unchecked Sendable { + private let closeCode: URLSessionWebSocketTask.CloseCode + private(set) var receivedContext: WebSocketCloseContext? + + init(closeCode: URLSessionWebSocketTask.CloseCode) { + self.closeCode = closeCode + } + + func closeCode(for context: WebSocketCloseContext) -> URLSessionWebSocketTask.CloseCode { + receivedContext = context + return closeCode + } } private final class HealthCheckEvent: @unchecked Sendable, Event, Codable, Hashable { diff --git a/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketPingController_Tests.swift b/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketPingController_Tests.swift index 6868a3f..38def9e 100644 --- a/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketPingController_Tests.swift +++ b/Tests/StreamCoreTests/WebSockets/WebSocketClient/WebSocketPingController_Tests.swift @@ -6,6 +6,7 @@ import XCTest final class WebSocketPingController_Tests: XCTestCase, @unchecked Sendable { + private let pingInterval: TimeInterval = 5 var time: VirtualTime! var pingController: WebSocketPingController! private var delegate: WebSocketPingController_Delegate! @@ -17,7 +18,8 @@ final class WebSocketPingController_Tests: XCTestCase, @unchecked Sendable { pingController = .init( timerType: VirtualTimeTimer.self, timerQueue: .main, - webSocketClientType: .coordinator + webSocketClientType: .coordinator, + pingInterval: pingInterval ) delegate = WebSocketPingController_Delegate() @@ -36,21 +38,21 @@ final class WebSocketPingController_Tests: XCTestCase, @unchecked Sendable { assert(delegate.sendPing_calledCount == 0) // Check `sendPing` is not called when the connection is not connected - time.run(numberOfSeconds: WebSocketPingController.pingTimeInterval + 1) + time.run(numberOfSeconds: pingInterval + 1) XCTAssertEqual(delegate.sendPing_calledCount, 0) // Set the connection state as connected pingController.connectionStateDidChange(.connected(healthCheckInfo: HealthCheckInfo())) // Simulate time passed 3x pingTimeInterval (+1 for margin errors) - time.run(numberOfSeconds: 3 * (WebSocketPingController.pingTimeInterval + 1)) + time.run(numberOfSeconds: 3 * (pingInterval + 1)) XCTAssertEqual(delegate.sendPing_calledCount, 3) let oldPingCount = delegate.sendPing_calledCount // Set the connection state to not connected and check `sendPing` is no longer called pingController.connectionStateDidChange(.authenticating) - time.run(numberOfSeconds: 3 * (WebSocketPingController.pingTimeInterval + 1)) + time.run(numberOfSeconds: 3 * (pingInterval + 1)) XCTAssertEqual(delegate.sendPing_calledCount, oldPingCount) }