Prepare connection / region pinning

Sources/LiveKit/Errors.swift

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

Sources/LiveKit/Support/RegionUrlProvider.swift

@@ -0,0 +1,131 @@
+/*
+ * 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
+
+class RegionUrlProvider: Loggable {
+    // Default
+    static let defaultCacheInterval: TimeInterval = 3000
+
+    private struct State {
+        var serverUrl: URL
+        var token: String
+        var regionSettings: Livekit_RegionSettings?
+        var regionSettingsUpdated: Date?
+        var attemptedRegions: [Livekit_RegionInfo] = []
+    }
+
+    private let _state: StateSync<State>
+    public let cacheInterval: TimeInterval
+
+    public var serverUrl: URL {
+        _state.mutate { $0.serverUrl }
+    }
+
+    init(url: String, token: String, cacheInterval: TimeInterval = defaultCacheInterval) {
+        let serverUrl = URL(string: url)!
+        self.cacheInterval = cacheInterval
+        _state = StateSync(State(serverUrl: serverUrl, token: token))
+    }
+
+    func set(regionSettings: Livekit_RegionSettings) {
+        _state.mutate {
+            $0.regionSettings = regionSettings
+            $0.regionSettingsUpdated = Date()
+        }
+    }
+
+    func set(token: String) {
+        _state.mutate { $0.token = token }
+    }
+
+    func resetAttempts() {
+        _state.mutate {
+            $0.attemptedRegions = []
+        }
+    }
+
+    func shouldRequestRegionSettings() -> Bool {
+        _state.read {
+            guard $0.regionSettings != nil, let regionSettingsUpdated = $0.regionSettingsUpdated else {
+                return true
+            }
+
+            let interval = Date().timeIntervalSince(regionSettingsUpdated)
+            return interval > cacheInterval
+        }
+    }
+
+    func nextBestRegionUrl() async throws -> URL? {
+        if shouldRequestRegionSettings() {
+            try await requestRegionSettings()
+        }
+
+        let (regionSettings, attemptedRegions) = _state.read { ($0.regionSettings, $0.attemptedRegions) }
+
+        guard let regionSettings else {
+            return nil
+        }
+
+        let remainingRegions = regionSettings.regions.filter { region in
+            !attemptedRegions.contains { $0.url == region.url }
+        }
+
+        guard let selectedRegion = remainingRegions.first else {
+            return nil
+        }
+
+        _state.mutate {
+            $0.attemptedRegions.append(selectedRegion)
+        }
+
+        return URL(string: selectedRegion.url)?.toSocketUrl()
+    }
+
+    func requestRegionSettings() async throws {
+        let (serverUrl, token) = _state.read { ($0.serverUrl, $0.token) }
+
+        // Ensure url is for cloud.
+        guard serverUrl.isCloud() else {
+            throw LiveKitError(.onlyForCloud)
+        }
+
+        var request = URLRequest(url: serverUrl.regionSettingsUrl())
+        request.addValue("Bearer \(token)", forHTTPHeaderField: "authorization")
+
+        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 {
+            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)
+            _state.mutate {
+                $0.regionSettings = regionSettings
+                $0.regionSettingsUpdated = Date()
+            }
+        } catch {
+            throw LiveKitError(.regionUrlProvider, message: "Failed to parse region settings with error: \(error)")
+        }
+    }
+}

Sources/LiveKit/Support/Utils.swift

@@ -263,3 +263,40 @@ extension MutableCollection {
         }
     }
 }
+
+extension URL {
+    /// 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!
+    }
+}
+
+extension HTTPURLResponse {
+    var isStatusCodeOK: Bool {
+        (200 ... 299).contains(statusCode)
+    }
+}

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)
+}