diff --git a/ios/Runner.xcodeproj/project.pbxproj b/ios/Runner.xcodeproj/project.pbxproj index ba956dd5..1eeee458 100644 --- a/ios/Runner.xcodeproj/project.pbxproj +++ b/ios/Runner.xcodeproj/project.pbxproj @@ -7,6 +7,7 @@ objects = { /* Begin PBXBuildFile section */ + D08A594730CF000100000001 /* DownloadProgressSubscription.swift in Sources */ = {isa = PBXBuildFile; fileRef = D08A594730CF000100000002 /* DownloadProgressSubscription.swift */; }; 1498D2341E8E89220040F4C2 /* GeneratedPluginRegistrant.m in Sources */ = {isa = PBXBuildFile; fileRef = 1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */; }; 331C808B294A63AB00263BE5 /* RunnerTests.swift in Sources */ = {isa = PBXBuildFile; fileRef = 331C807B294A618700263BE5 /* RunnerTests.swift */; }; 3B3967161E833CAA004F5970 /* AppFrameworkInfo.plist in Resources */ = {isa = PBXBuildFile; fileRef = 3B3967151E833CAA004F5970 /* AppFrameworkInfo.plist */; }; @@ -47,6 +48,7 @@ /* End PBXCopyFilesBuildPhase section */ /* Begin PBXFileReference section */ + D08A594730CF000100000002 /* DownloadProgressSubscription.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = DownloadProgressSubscription.swift; sourceTree = ""; }; 0448E6A8F05C4FAF8461F7C9 /* libPods-RunnerTests.a */ = {isa = PBXFileReference; explicitFileType = archive.ar; includeInIndex = 0; path = "libPods-RunnerTests.a"; sourceTree = BUILT_PRODUCTS_DIR; }; 1498D2321E8E86230040F4C2 /* GeneratedPluginRegistrant.h */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.c.h; path = GeneratedPluginRegistrant.h; sourceTree = ""; }; 1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.objc; path = GeneratedPluginRegistrant.m; sourceTree = ""; }; @@ -162,6 +164,7 @@ 1498D2321E8E86230040F4C2 /* GeneratedPluginRegistrant.h */, 1498D2331E8E89220040F4C2 /* GeneratedPluginRegistrant.m */, 74858FAE1ED2DC5600515810 /* AppDelegate.swift */, + D08A594730CF000100000002 /* DownloadProgressSubscription.swift */, A11CE0022F00000000000001 /* ExtensionCallbackParser.swift */, 74858FAD1ED2DC5600515810 /* Runner-Bridging-Header.h */, ); @@ -397,6 +400,7 @@ buildActionMask = 2147483647; files = ( 74858FAF1ED2DC5600515810 /* AppDelegate.swift in Sources */, + D08A594730CF000100000001 /* DownloadProgressSubscription.swift in Sources */, A11CE0012F00000000000001 /* ExtensionCallbackParser.swift in Sources */, 1498D2341E8E89220040F4C2 /* GeneratedPluginRegistrant.m in Sources */, ); diff --git a/ios/Runner/AppDelegate.swift b/ios/Runner/AppDelegate.swift index 7d6fafc8..15c9ed99 100644 --- a/ios/Runner/AppDelegate.swift +++ b/ios/Runner/AppDelegate.swift @@ -12,15 +12,9 @@ import Gobackend private let LARGE_JSON_RESULT_FILE_KEY = "__json_file" private let LARGE_JSON_RESULT_FILE_THRESHOLD_BYTES = 256 * 1024 private let streamQueue = DispatchQueue(label: "com.zarz.spotiflac.progress_stream", qos: .utility) - private let downloadProgressQueue = DispatchQueue( - label: "com.zarz.spotiflac.download_progress_stream", - qos: .utility - ) - private var downloadProgressTimer: DispatchSourceTimer? - private var downloadProgressEventSink: FlutterEventSink? - private var lastDownloadProgressPayload: String? - private var lastDownloadProgressSeq: Int64 = 0 - private var downloadProgressGeneration: UInt64 = 0 + private let downloadProgressSubscription = DownloadProgressSubscription { sequence, timeout in + GobackendWaitForAllDownloadProgressDelta(sequence, timeout) as String? ?? "" + } private var libraryScanProgressTimer: DispatchSourceTimer? private var libraryScanProgressEventSink: FlutterEventSink? private var lastLibraryScanProgressPayload: String? @@ -236,43 +230,11 @@ import Gobackend } private func startDownloadProgressStream(_ eventSink: @escaping FlutterEventSink) { - stopDownloadProgressStream() - downloadProgressGeneration &+= 1 - let generation = downloadProgressGeneration - downloadProgressEventSink = eventSink - lastDownloadProgressPayload = nil - lastDownloadProgressSeq = 0 - - let timer = DispatchSource.makeTimerSource(queue: downloadProgressQueue) - timer.schedule(deadline: .now(), repeating: .milliseconds(250)) - timer.setEventHandler { [weak self] in - guard let self, self.downloadProgressGeneration == generation else { return } - let payload = GobackendWaitForAllDownloadProgressDelta( - self.lastDownloadProgressSeq, - 15_000 - ) as String? ?? "" - if payload.isEmpty || payload == self.lastDownloadProgressPayload { - return - } - self.updateDownloadProgressSeq(payload) - self.lastDownloadProgressPayload = payload - DispatchQueue.main.async { [weak self] in - guard let self, self.downloadProgressGeneration == generation else { return } - eventSink(self.parseJsonPayload(payload)) - } - } - downloadProgressTimer = timer - timer.resume() + downloadProgressSubscription.start(eventSink) } private func stopDownloadProgressStream() { - downloadProgressGeneration &+= 1 - downloadProgressTimer?.setEventHandler {} - downloadProgressTimer?.cancel() - downloadProgressTimer = nil - downloadProgressEventSink = nil - lastDownloadProgressPayload = nil - lastDownloadProgressSeq = 0 + downloadProgressSubscription.stop() } private func startLibraryScanProgressStream(_ eventSink: @escaping FlutterEventSink) { @@ -320,18 +282,6 @@ import Gobackend } } - private func updateDownloadProgressSeq(_ payload: String) { - guard let data = payload.data(using: .utf8) else { return } - do { - if let obj = try JSONSerialization.jsonObject(with: data, options: [.fragmentsAllowed]) as? [String: Any], - let seq = obj["seq"] as? NSNumber, - seq.int64Value > lastDownloadProgressSeq { - lastDownloadProgressSeq = seq.int64Value - } - } catch { - } - } - private func bridgeJsonResult(_ payload: String) -> Any { if payload.utf8.count < LARGE_JSON_RESULT_FILE_THRESHOLD_BYTES { return payload diff --git a/ios/Runner/DownloadProgressSubscription.swift b/ios/Runner/DownloadProgressSubscription.swift new file mode 100644 index 00000000..fa433473 --- /dev/null +++ b/ios/Runner/DownloadProgressSubscription.swift @@ -0,0 +1,82 @@ +import Foundation + +/// Each listener owns its cursor and pending waiter. Lifecycle and delivery run +/// on the main queue; a cancelled wait may finish, but can only touch its own state. +final class DownloadProgressSubscription { + typealias Waiter = (Int64, Int64) -> String + private final class Session { + let queue = DispatchQueue(label: "com.zarz.spotiflac.download_progress_subscription", qos: .utility) + private let lock = NSLock() + private var cancelled = false + // Accessed only on this session's queue. + var sequence: Int64 = 0 + var lastPayload: String? + + func cancel() { + lock.lock() + cancelled = true + lock.unlock() + } + + var isCancelled: Bool { + lock.lock() + defer { lock.unlock() } + return cancelled + } + } + + private let waiter: Waiter + private let interval: TimeInterval + private var current: Session? + + init(interval: TimeInterval = 0.25, waiter: @escaping Waiter) { + self.interval = interval + self.waiter = waiter + } + + func start(_ receive: @escaping (Any) -> Void) { + dispatchPrecondition(condition: .onQueue(.main)) + stop() + let session = Session() + current = session + poll(session, receive: receive) + } + + func stop() { + dispatchPrecondition(condition: .onQueue(.main)) + current?.cancel() + current = nil + } + + deinit { current?.cancel() } + + private func poll(_ session: Session, receive: @escaping (Any) -> Void) { + let waiter = self.waiter + session.queue.async { [weak self] in + guard !session.isCancelled else { return } + let payload = waiter(session.sequence, 15_000) + guard !session.isCancelled else { return } + if !payload.isEmpty && payload != session.lastPayload, + let data = payload.data(using: .utf8), + let object = try? JSONSerialization.jsonObject(with: data, options: [.fragmentsAllowed]), + let delta = object as? [String: Any], + let sequence = delta["seq"] as? NSNumber { + session.sequence = max(session.sequence, sequence.int64Value) + session.lastPayload = payload + DispatchQueue.main.async { [weak self] in + guard let self, self.current === session, !session.isCancelled else { return } + receive(object) + } + } + // Schedule from main so subscription identity is never accessed + // concurrently, and stop/relisten does not wait for an old Go call. + DispatchQueue.main.async { [weak self] in + guard let self, self.current === session, !session.isCancelled else { return } + DispatchQueue.main.asyncAfter(deadline: .now() + self.interval) { [weak self] in + guard let self, self.current === session, !session.isCancelled else { return } + self.poll(session, receive: receive) + } + } + } + } +} diff --git a/ios/RunnerTests/RunnerTests.swift b/ios/RunnerTests/RunnerTests.swift index c60759e3..caebfebe 100644 --- a/ios/RunnerTests/RunnerTests.swift +++ b/ios/RunnerTests/RunnerTests.swift @@ -3,6 +3,48 @@ import XCTest @testable import Runner class RunnerTests: XCTestCase { + func testProgressRestartDoesNotAcceptCancelledWaiterState() { + let oldStarted = expectation(description: "old waiter started") + let replacementDelivered = expectation(description: "replacement snapshot") + let replacementAdvanced = expectation(description: "replacement owns cursor") + let oldReleased = expectation(description: "old waiter returned") + let releaseOld = DispatchSemaphore(value: 0) + let lock = NSLock() + var calls = 0 + var invalidCursors = [Int64]() + let stream = DownloadProgressSubscription(interval: 0.01) { sequence, _ in + lock.lock() + calls += 1 + let call = calls + if (call <= 2 && sequence != 0) || (call > 2 && sequence != 2) { + invalidCursors.append(sequence) + } + lock.unlock() + if call == 1 { + oldStarted.fulfill() + _ = releaseOld.wait(timeout: .now() + 3) + oldReleased.fulfill() + return "{\"seq\":99,\"items\":{}}" + } + if call == 2 { return "{\"seq\":2,\"reset\":true,\"items\":{}}" } + if call == 3 { replacementAdvanced.fulfill() } + return "" + } + stream.start { _ in XCTFail("Cancelled listener received an event") } + wait(for: [oldStarted], timeout: 2) + stream.stop() + stream.start { event in + XCTAssertEqual((event as? [String: Any])?["seq"] as? Int, 2) + replacementDelivered.fulfill() + releaseOld.signal() + } + wait(for: [replacementDelivered, oldReleased, replacementAdvanced], timeout: 2) + stream.stop() + lock.lock() + XCTAssertTrue(invalidCursors.isEmpty, "Unexpected cursors: \(invalidCursors)") + lock.unlock() + } + func testParsesOAuthCallback() { let route = ExtensionCallbackParser.parse( URL(string: "spotiflac://callback?code=auth-code&state=spotify-web")!