fix superClient
This commit is contained in:
parent
4b005cb75d
commit
3cdf1953bf
@ -22,9 +22,7 @@ actor SDLSuperClient {
|
||||
// 数据流
|
||||
public let messageStream: AsyncThrowingStream<SDLQUICInboundMessage, Error>
|
||||
private let messageCont: AsyncThrowingStream<SDLQUICInboundMessage, Error>.Continuation
|
||||
|
||||
private var pendingConnectionError: Error?
|
||||
|
||||
|
||||
private let connection: NWConnection
|
||||
private let maxBufferSize: Int
|
||||
|
||||
@ -68,21 +66,14 @@ actor SDLSuperClient {
|
||||
return
|
||||
}
|
||||
|
||||
self.connection.stateUpdateHandler = { [weak self] state in
|
||||
SDLLogger.log("[SDLSuperClient] new state: \(state)", for: .debug)
|
||||
Task {
|
||||
await self?.handleConnectionState(state: state)
|
||||
}
|
||||
}
|
||||
self.connection.start(queue: queue)
|
||||
|
||||
let stateStream = Self.makeStateStream(for: self.connection)
|
||||
defer {
|
||||
self.stop()
|
||||
}
|
||||
|
||||
try await withTaskCancellationHandler {
|
||||
try await self.waitUntilReady()
|
||||
try await self.readLoop()
|
||||
self.connection.start(queue: self.queue)
|
||||
try await self.runStateLoop(stateStream)
|
||||
} onCancel: {
|
||||
self.connection.cancel()
|
||||
}
|
||||
@ -99,35 +90,48 @@ actor SDLSuperClient {
|
||||
|
||||
preconditionFailure("invalid super server IP: \(ip)")
|
||||
}
|
||||
|
||||
private func handleConnectionState(state: NWConnection.State) {
|
||||
switch state {
|
||||
case .ready:
|
||||
self.state = .running
|
||||
case .failed(let error):
|
||||
self.pendingConnectionError = SDLSuperError.connectionFailed(error)
|
||||
self.messageCont.finish(throwing: SDLSuperError.connectionFailed(error))
|
||||
case .cancelled:
|
||||
self.pendingConnectionError = SDLSuperError.connectionCancelled
|
||||
self.messageCont.finish(throwing: SDLSuperError.connectionCancelled)
|
||||
default:
|
||||
()
|
||||
|
||||
private static func makeStateStream(for connection: NWConnection) -> AsyncThrowingStream<NWConnection.State, Error> {
|
||||
return AsyncThrowingStream(bufferingPolicy: .bufferingNewest(16)) { continuation in
|
||||
connection.stateUpdateHandler = { state in
|
||||
SDLLogger.log("[SDLSuperClient] new state: \(state)", for: .debug)
|
||||
continuation.yield(state)
|
||||
|
||||
switch state {
|
||||
case .failed(let error):
|
||||
continuation.finish(throwing: SDLSuperError.connectionFailed(error))
|
||||
case .cancelled:
|
||||
continuation.finish(throwing: SDLSuperError.connectionCancelled)
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
continuation.onTermination = { _ in
|
||||
connection.stateUpdateHandler = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func waitUntilReady() async throws {
|
||||
while true {
|
||||
|
||||
private func runStateLoop(_ stateStream: AsyncThrowingStream<NWConnection.State, Error>) async throws {
|
||||
for try await state in stateStream {
|
||||
try Task.checkCancellation()
|
||||
|
||||
if case .running = self.state {
|
||||
switch state {
|
||||
case .ready:
|
||||
self.state = .running
|
||||
try await self.readLoop()
|
||||
return
|
||||
case .failed(let error):
|
||||
let wrappedError = SDLSuperError.connectionFailed(error)
|
||||
self.messageCont.finish(throwing: wrappedError)
|
||||
throw wrappedError
|
||||
case .cancelled:
|
||||
self.messageCont.finish(throwing: SDLSuperError.connectionCancelled)
|
||||
throw SDLSuperError.connectionCancelled
|
||||
default:
|
||||
break
|
||||
}
|
||||
|
||||
if let pendingConnectionError {
|
||||
throw pendingConnectionError
|
||||
}
|
||||
|
||||
try await Task.sleep(for: .milliseconds(100))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user