fix superClient

This commit is contained in:
anlicheng 2026-05-27 20:16:26 +08:00
parent e7f58dce03
commit 2618ac3d05

View File

@ -62,42 +62,17 @@ final class SDLSuperClient: @unchecked Sendable {
return return
} }
let stateStream = self.makeStateStream()
defer { defer {
self.stop() self.stop()
} }
let stateStream = Self.makeStateStream(for: self.connection)
let promise = AsyncPromise<Bool>()
self.connection.start(queue: self.queue)
try await withTaskCancellationHandler { try await withTaskCancellationHandler {
try await withThrowingTaskGroup { group in self.connection.start(queue: self.queue)
group.addTask { try await self.waitUntilReady(stateStream)
do {
for try await state in stateStream {
switch state {
case .ready:
await promise.succeed(true)
default:
()
}
}
} catch let err {
await promise.fail(err)
throw err
}
}
group.addTask {
_ = try await promise.value()
try Task.checkCancellation()
try await self.readLoop() try await self.readLoop()
}
try await group.next()
}
} onCancel: { } onCancel: {
self.stop() self.connection.cancel()
} }
} }
@ -113,24 +88,36 @@ final class SDLSuperClient: @unchecked Sendable {
preconditionFailure("invalid super server IP: \(ip)") preconditionFailure("invalid super server IP: \(ip)")
} }
private static func makeStateStream(for connection: NWConnection) -> AsyncThrowingStream<NWConnection.State, Error> { private func makeStateStream() -> AsyncThrowingStream<NWConnection.State, Error> {
let connection = self.connection
return AsyncThrowingStream(bufferingPolicy: .bufferingNewest(16)) { continuation in return AsyncThrowingStream(bufferingPolicy: .bufferingNewest(16)) { continuation in
connection.stateUpdateHandler = { state in connection.stateUpdateHandler = { [weak self] state in
SDLLogger.log("[SDLSuperClient] new state: \(state)", for: .debug) SDLLogger.log("[SDLSuperClient] new state: \(state)", for: .debug)
switch state { switch state {
case .failed(let error): case .failed(let error):
continuation.finish(throwing: error) let wrappedError = SDLSuperError.connectionFailed(error)
self?.finishMessageStream(throwing: wrappedError)
continuation.finish(throwing: wrappedError)
case .cancelled: case .cancelled:
self?.finishMessageStream(throwing: SDLSuperError.connectionCancelled)
continuation.finish(throwing: SDLSuperError.connectionCancelled) continuation.finish(throwing: SDLSuperError.connectionCancelled)
default: default:
continuation.yield(state) continuation.yield(state)
} }
} }
}
}
continuation.onTermination = { _ in private func waitUntilReady(_ stateStream: AsyncThrowingStream<NWConnection.State, Error>) async throws {
connection.stateUpdateHandler = nil for try await state in stateStream {
try Task.checkCancellation()
if case .ready = state {
return
} }
} }
throw SDLSuperError.connectionCancelled
} }
private func readLoop() async throws { private func readLoop() async throws {