Prepare connection / region pinning

Sources/LiveKit/Core/Room+Region.swift

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2024 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Foundation
+
+// MARK: - Room+Region
+
+extension Room {
+    static let defaultCacheInterval: TimeInterval = 3000
+
+    func resolveBestRegion() async throws -> RegionInfo {
+        try await requestRegionSettings()
+
+        let sortedByDistance = _regionState.remaining.sorted { $0.distance < $1.distance }
+        log("[Region] Remaining regions: \(String(describing: sortedByDistance))")
+
+        guard let selectedRegion = sortedByDistance.first else {
+            throw LiveKitError(.regionUrlProvider, message: "No more remaining regions.")
+        }
+
+        log("[Region] Resolved region: \(String(describing: selectedRegion))")
+
+        return selectedRegion
+    }
+
+    func add(failedRegion region: RegionInfo) {
+        _regionState.mutate {
+            $0.remaining.remove(region)
+        }
+    }
+
+    // MARK: - Private
+
+    private func requestRegionSettings() async throws {
+        let (serverUrl, token) = _state.read { ($0.url, $0.token) }
+
+        guard let serverUrl, let token else {
+            throw LiveKitError(.invalidState)
+        }
+
+        let shouldRequestRegionSettings = _regionState.read {
+            guard serverUrl == $0.url, let regionSettingsUpdated = $0.lastRequested else { return true }
+            let interval = Date().timeIntervalSince(regionSettingsUpdated)
+            log("[Region] Interval: \(String(describing: interval))")
+            return interval > Self.defaultCacheInterval
+        }
+
+        guard shouldRequestRegionSettings else { return }
+
+        // Ensure url is for cloud.
+        guard serverUrl.isCloud() else {
+            throw LiveKitError(.onlyForCloud)
+        }
+
+        // Make a request which ignores cache.
+        var request = URLRequest(url: serverUrl.regionSettingsUrl(),
+                                 cachePolicy: .reloadIgnoringLocalAndRemoteCacheData)
+
+        request.addValue("Bearer \(token)", forHTTPHeaderField: "authorization")
+
+        log("[Region] Requesting region settings...")
+
+        let (data, response) = try await URLSession.shared.data(for: request)
+        // Response must be a HTTPURLResponse.
+        guard let httpResponse = response as? HTTPURLResponse else {
+            throw LiveKitError(.regionUrlProvider, message: "Failed to fetch region settings")
+        }
+
+        // Check the status code.
+        guard httpResponse.isStatusCodeOK else {
+            log("[Region] Failed to fetch region settings, error: \(String(describing: httpResponse))", .error)
+            throw LiveKitError(.regionUrlProvider, message: "Failed to fetch region settings with status code: \(httpResponse.statusCode)")
+        }
+
+        do {
+            // Try to parse the JSON data.
+            let regionSettings = try Livekit_RegionSettings(jsonUTF8Data: data)
+            let allRegions = regionSettings.regions.compactMap { $0.toLKType() }
+
+            if allRegions.isEmpty {
+                throw LiveKitError(.regionUrlProvider, message: "Fetched region data is empty.")
+            }
+
+            log("[Region] all regions: \(String(describing: allRegions))")
+
+            _regionState.mutate {
+                $0.url = serverUrl
+                $0.all = Set(allRegions)
+                $0.remaining = Set(allRegions)
+                $0.lastRequested = Date()
+            }
+        } catch {
+            throw LiveKitError(.regionUrlProvider, message: "Failed to parse region settings with error: \(error)")
+        }
+    }
+}

Sources/LiveKit/Core/Room.swift

@@ -168,7 +168,16 @@ public class Room: NSObject, ObservableObject, Loggable {
         }
     }
 
+    struct RegionState {
+        // Region
+        var url: URL?
+        var lastRequested: Date?
+        var all: Set<RegionInfo> = []
+        var remaining: Set<RegionInfo> = []
+    }
+
     let _state: StateSync<State>
+    let _regionState = StateSync(RegionState())
 
     private let _sidCompleter = AsyncCompleter<Sid>(label: "sid", defaultTimeout: .resolveSid)
 
@@ -315,28 +324,35 @@ public class Room: NSObject, ObservableObject, Loggable {
 
         try Task.checkCancellation()
 
-        _state.mutate { $0.connectionState = .connecting }
+        _state.mutate {
+            $0.url = url
+            $0.token = token
+            $0.connectionState = .connecting
+        }
 
         do {
-            try await fullConnectSequence(url, token)
-
-            // Connect sequence successful
-            log("Connect sequence completed")
+            while true {
+                let region = try await resolveBestRegion()
 
-            // Final check if cancelled, don't fire connected events
-            try Task.checkCancellation()
-
-            // update internal vars (only if connect succeeded)
-            _state.mutate {
-                $0.url = url
-                $0.token = token
-                $0.connectionState = .connected
+                do {
+                    try await fullConnectSequence(region.url, token)
+                    // Connect sequence successful
+                    log("Connect sequence completed")
+                    // Final check if cancelled, don't fire connected events
+                    try Task.checkCancellation()
+                    _state.mutate { $0.connectionState = .connected }
+                    break // Exit loop on successful connection
+                } catch {
+                    log("Connect failed with region: \(region)")
+                    add(failedRegion: region)
+                    // Prepare for next connect attempt.
+                    await cleanUp(isFullReconnect: true)
+                }
             }
-
         } catch {
+            log("Failed to resolve a region or connect: \(error)")
             await cleanUp(withError: error)
-            // Re-throw error
-            throw error
+            throw error // Re-throw the original error
         }
 
         log("Connected to \(String(describing: self))", .info)

Sources/LiveKit/Errors.swift

@@ -51,6 +51,10 @@ public enum LiveKitErrorType: Int {
     case unableToResolveFPSRange = 703
     case capturerDimensionsNotResolved = 704
     case deviceAccessDenied = 705
+
+    // LiveKit Cloud
+    case onlyForCloud = 901
+    case regionUrlProvider = 902
 }
 
 extension LiveKitErrorType: CustomStringConvertible {

Sources/LiveKit/Extensions/URL.swift

@@ -28,4 +28,33 @@ extension URL {
     var isSecure: Bool {
         scheme == "https" || scheme == "wss"
     }
+
+    /// Checks whether the URL is a LiveKit Cloud URL.
+    func isCloud() -> Bool {
+        guard let host else { return false }
+        return host.hasSuffix(".livekit.cloud") || host.hasSuffix(".livekit.run")
+    }
+
+    func cloudConfigUrl() -> URL {
+        var components = URLComponents(url: self, resolvingAgainstBaseURL: false)!
+        components.scheme = scheme?.replacingOccurrences(of: "ws", with: "http")
+        components.path = "/settings"
+        return components.url!
+    }
+
+    func regionSettingsUrl() -> URL {
+        cloudConfigUrl().appendingPathComponent("/regions")
+    }
+
+    func toSocketUrl() -> URL {
+        var components = URLComponents(url: self, resolvingAgainstBaseURL: false)!
+        components.scheme = scheme?.replacingOccurrences(of: "http", with: "ws")
+        return components.url!
+    }
+
+    func toHTTPUrl() -> URL {
+        var components = URLComponents(url: self, resolvingAgainstBaseURL: false)!
+        components.scheme = scheme?.replacingOccurrences(of: "ws", with: "http")
+        return components.url!
+    }
 }

Sources/LiveKit/Support/Utils.swift

@@ -268,6 +268,12 @@ extension MutableCollection {
     }
 }
 
+extension HTTPURLResponse {
+    var isStatusCodeOK: Bool {
+        (200 ... 299).contains(statusCode)
+    }
+}
+
 func computeAttributesDiff(oldValues: [String: String], newValues: [String: String]) -> [String: String] {
     let allKeys = Set(oldValues.keys).union(newValues.keys)
     var diff = [String: String]()

Sources/LiveKit/Types/RegionInfo.swift

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2024 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Foundation
+
+@objc
+public class RegionInfo: NSObject {
+    let regionId: String
+    let url: URL
+    let distance: Int64
+
+    init?(region: String, url: String, distance: Int64) {
+        guard let url = URL(string: url) else { return nil }
+        regionId = region
+        self.url = url
+        self.distance = distance
+    }
+
+    // MARK: - Equal
+
+    override public func isEqual(_ object: Any?) -> Bool {
+        guard let other = object as? Self else { return false }
+        return regionId == other.regionId
+    }
+
+    override public var hash: Int {
+        var hasher = Hasher()
+        hasher.combine(regionId)
+        return hasher.finalize()
+    }
+
+    //
+
+    override public var description: String {
+        "RegionInfo(id: \(regionId), url: \(url), distance: \(distance))"
+    }
+}
+
+extension Livekit_RegionInfo {
+    func toLKType() -> RegionInfo? {
+        RegionInfo(region: region,
+                   url: url,
+                   distance: distance)
+    }
+}

Tests/LiveKitTests/RegionUrlProviderTests.swift

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2024 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+@testable import LiveKit
+import XCTest
+
+class RegionUrlProviderTests: XCTestCase {
+    func testResolveUrl() async throws {
+        let testCacheInterval: TimeInterval = 3
+        // Test data.
+        let testRegionSettings = Livekit_RegionSettings.with {
+            $0.regions.append(Livekit_RegionInfo.with {
+                $0.region = "otokyo1a"
+                $0.url = "https://example.otokyo1a.production.livekit.cloud"
+                $0.distance = 32838
+            })
+            $0.regions.append(Livekit_RegionInfo.with {
+                $0.region = "dblr1a"
+                $0.url = "https://example.dblr1a.production.livekit.cloud"
+                $0.distance = 6_660_301
+            })
+            $0.regions.append(Livekit_RegionInfo.with {
+                $0.region = "dsyd1a"
+                $0.url = "https://example.dsyd1a.production.livekit.cloud"
+                $0.distance = 7_823_582
+            })
+        }
+
+        let provider = RegionUrlProvider(url: "wss://test.livekit.cloud", token: "", cacheInterval: testCacheInterval)
+
+        // See if request should be initiated.
+        XCTAssert(provider.shouldRequestRegionSettings(), "Should require to request region settings")
+
+        // Set test data.
+        provider.set(regionSettings: testRegionSettings)
+
+        // See if request is not required to be initiated.
+        XCTAssert(!provider.shouldRequestRegionSettings(), "Should require to request region settings")
+
+        let attempt1 = try await provider.nextBestRegionUrl()
+        print("Next url: \(String(describing: attempt1))")
+        XCTAssert(attempt1 == URL(string: testRegionSettings.regions[0].url)?.toSocketUrl())
+
+        let attempt2 = try await provider.nextBestRegionUrl()
+        print("Next url: \(String(describing: attempt2))")
+        XCTAssert(attempt2 == URL(string: testRegionSettings.regions[1].url)?.toSocketUrl())
+
+        let attempt3 = try await provider.nextBestRegionUrl()
+        print("Next url: \(String(describing: attempt3))")
+        XCTAssert(attempt3 == URL(string: testRegionSettings.regions[2].url)?.toSocketUrl())
+
+        let attempt4 = try await provider.nextBestRegionUrl()
+        print("Next url: \(String(describing: attempt4))")
+        XCTAssert(attempt4 == nil)
+
+        // Simulate cache time elapse.
+        await asyncSleep(for: testCacheInterval)
+
+        // After cache time elapsed, should require to request region settings again.
+        XCTAssert(provider.shouldRequestRegionSettings(), "Should require to request region settings")
+    }
+}

Tests/LiveKitTests/Support/Utils.swift

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2024 LiveKit
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import Foundation
+
+func asyncSleep(for duration: TimeInterval) async {
+    let nanoseconds = UInt64(duration * Double(NSEC_PER_SEC))
+    try? await Task.sleep(nanoseconds: nanoseconds)
+}