fix superClient

This commit is contained in:
anlicheng 2026-05-27 19:47:44 +08:00
parent 1c0ed04fa2
commit e7f58dce03
2 changed files with 97 additions and 60 deletions

View File

@ -0,0 +1,53 @@
//
// AsyncPromise.swift
// punchnet
//
// Created by on 2026/5/27.
//
import Foundation
public actor AsyncPromise<Value: Sendable> {
private enum State {
case pending([CheckedContinuation<Value, Error>])
case completed(Result<Value, Error>)
}
private var state: State = .pending([])
public init() {
}
public func value() async throws -> Value {
try await withCheckedThrowingContinuation { continuation in
switch state {
case .pending(var continuations):
continuations.append(continuation)
state = .pending(continuations)
case .completed(let result):
continuation.resume(with: result)
}
}
}
public func succeed(_ value: Value) {
complete(.success(value))
}
public func fail(_ error: Error) {
complete(.failure(error))
}
private func complete(_ result: Result<Value, Error>) {
guard case .pending(let continuations) = state else {
return
}
state = .completed(result)
for continuation in continuations {
continuation.resume(with: result)
}
}
}

View File

@ -13,6 +13,7 @@ final class SDLSuperClient: @unchecked Sendable {
// //
public let messageStream: AsyncThrowingStream<SDLQUICInboundMessage, Error> public let messageStream: AsyncThrowingStream<SDLQUICInboundMessage, Error>
private let messageCont: AsyncThrowingStream<SDLQUICInboundMessage, Error>.Continuation private let messageCont: AsyncThrowingStream<SDLQUICInboundMessage, Error>.Continuation
private let stateLock = NSLock() private let stateLock = NSLock()
private var isStarted = false private var isStarted = false
private var isStopped = false private var isStopped = false
@ -60,17 +61,43 @@ final class SDLSuperClient: @unchecked Sendable {
guard self.markStarted() else { guard self.markStarted() else {
return return
} }
let stateStream = Self.makeStateStream(for: self.connection)
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 {
self.connection.start(queue: self.queue) try await withThrowingTaskGroup { group in
try await self.runStateLoop(stateStream) group.addTask {
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 group.next()
}
} onCancel: { } onCancel: {
self.connection.cancel() self.stop()
} }
} }
@ -90,79 +117,36 @@ final class SDLSuperClient: @unchecked Sendable {
return AsyncThrowingStream(bufferingPolicy: .bufferingNewest(16)) { continuation in return AsyncThrowingStream(bufferingPolicy: .bufferingNewest(16)) { continuation in
connection.stateUpdateHandler = { state in connection.stateUpdateHandler = { state in
SDLLogger.log("[SDLSuperClient] new state: \(state)", for: .debug) SDLLogger.log("[SDLSuperClient] new state: \(state)", for: .debug)
continuation.yield(state)
switch state { switch state {
case .failed(let error): case .failed(let error):
continuation.finish(throwing: SDLSuperError.connectionFailed(error)) continuation.finish(throwing: error)
case .cancelled: case .cancelled:
continuation.finish(throwing: SDLSuperError.connectionCancelled) continuation.finish(throwing: SDLSuperError.connectionCancelled)
default: default:
break continuation.yield(state)
} }
} }
continuation.onTermination = { _ in continuation.onTermination = { _ in
connection.stateUpdateHandler = nil connection.stateUpdateHandler = nil
} }
} }
} }
private func runStateLoop(_ stateStream: AsyncThrowingStream<NWConnection.State, Error>) async throws {
do {
try await self.waitUntilReady(stateStream)
self.connection.stateUpdateHandler = nil
try await self.readLoop()
} catch is CancellationError {
throw CancellationError()
} catch let error as SDLSuperError {
self.finishMessageStream(throwing: error)
throw error
} catch {
let wrappedError = SDLSuperError.internalError(error)
self.finishMessageStream(throwing: wrappedError)
throw wrappedError
}
}
private func waitUntilReady(_ stateStream: AsyncThrowingStream<NWConnection.State, Error>) async throws {
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 {
let frameParser = SDLSuperFrameParser(maxBufferSize: self.maxBufferSize) let frameParser = SDLSuperFrameParser(maxBufferSize: self.maxBufferSize)
do { while true {
while true { try Task.checkCancellation()
let data = try await Self.readOnce(connection: self.connection)
let frames = try frameParser.parseFrames(data: data)
for frame in frames {
try Task.checkCancellation() try Task.checkCancellation()
let data = try await Self.readOnce(connection: self.connection) if let message = SDLSuperCodec.decode(frame: frame) {
let frames = try frameParser.parseFrames(data: data) self.messageCont.yield(message)
for frame in frames { } else {
try Task.checkCancellation() throw SDLSuperError.decodeError("invalid message")
if let message = SDLSuperCodec.decode(frame: frame) {
self.messageCont.yield(message)
} else {
throw SDLSuperError.decodeError("invalid message")
}
} }
} }
} catch is CancellationError {
throw CancellationError()
} catch let error as SDLSuperError {
self.finishMessageStream(throwing: error)
throw error
} catch {
let wrappedError = SDLSuperError.internalError(error)
self.finishMessageStream(throwing: wrappedError)
throw wrappedError
} }
} }