anlicheng/swiftlib_sdlan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加事件逻辑处理

Browse Source
This commit is contained in:
anlicheng 2025-08-01 14:39:33 +08:00
parent 3ce4e05613
commit f273da3b11
3 changed files with 21 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Sources/Punchnet/SDLContext.swift
View File

@ -124,7 +124,6 @@ public class SDLContext: @unchecked Sendable {
try await self.handleUDPEvent(event: event)
}
}
}
group.addTask {

11
Sources/Punchnet/SDLSuperClient.swift
View File

@ -58,12 +58,15 @@ actor SDLSuperClient {
func start() async throws {
try await self.asyncChannel.executeThenClose { inbound, outbound in
self.inboundContinuation.yield(.ready)
try await withThrowingTaskGroup(of: Void.self) { group in
group.addTask {
defer {
self.inboundContinuation.finish()
}
try await self.asyncChannel.channel.closeFuture.get()
self.inboundContinuation.yield(.closed)
self.inboundContinuation.finish()
}
group.addTask {
for try await var packet in inbound {
if let message = SDLSuperClientDecoder.decode(buffer: &packet) {
SDLLogger.log("[SDLSuperTransport] read message: \(message)", level: .warning)

30
Sources/Punchnet/SDLUDPHole.swift
View File

@ -19,13 +19,13 @@ struct UDPMessage {
actor SDLUDPHole {
private let group = MultiThreadedEventLoopGroup(numberOfThreads: 1)
private let asyncChannel: NIOAsyncChannel<AddressedEnvelope<ByteBuffer>, AddressedEnvelope<ByteBuffer>>
private let (writeStream, continuation) = AsyncStream.makeStream(of: UDPMessage.self, bufferingPolicy: .unbounded)
private let (writeStream, writeContinuation) = AsyncStream.makeStream(of: UDPMessage.self, bufferingPolicy: .unbounded)
private var cookieGenerator = SDLIdGenerator(seed: 1)
private var promises: [UInt32:EventLoopPromise<SDLStunProbeReply>] = [:]
public var localAddress: SocketAddress?
public let (eventFlow, inboundContinuation) = AsyncStream.makeStream(of: UDPEvent.self, bufferingPolicy: .unbounded)
public let (eventFlow, eventContinuation) = AsyncStream.makeStream(of: UDPEvent.self, bufferingPolicy: .unbounded)
// 定义事件类型
enum UDPEvent {
@ -55,12 +55,16 @@ actor SDLUDPHole {
func start() async throws {
try await self.asyncChannel.executeThenClose { inbound, outbound in
self.eventContinuation.yield(.ready)
try await withThrowingTaskGroup(of: Void.self) { group in
group.addTask {
defer {
self.inboundContinuation.finish()
}
try await self.asyncChannel.channel.closeFuture.get()
self.eventContinuation.yield(.closed)
self.eventContinuation.finish()
}
group.addTask {
for try await envelope in inbound {
var buffer = envelope.data
let remoteAddress = envelope.remoteAddress
@ -69,12 +73,12 @@ actor SDLUDPHole {
switch message {
case .data(let data):
SDLLogger.log("[SDLUDPHole] read data: \(data.format()), from: \(remoteAddress)", level: .debug)
self.inboundContinuation.yield(.data(data))
self.eventContinuation.yield(.data(data))
case .stunProbeReply(let probeReply):
// 执行并移除回调
await self.trigger(probeReply: probeReply)
default:
self.inboundContinuation.yield(.message(remoteAddress, message))
self.eventContinuation.yield(.message(remoteAddress, message))
}
} else {
SDLLogger.log("[SDLUDPHole] decode message, get null", level: .warning)
@ -87,7 +91,7 @@ actor SDLUDPHole {
group.addTask {
defer {
self.continuation.finish()
self.writeContinuation.finish()
}
for try await message in self.writeStream {
@ -99,11 +103,6 @@ actor SDLUDPHole {
try await outbound.write(envelope)
}
}
//eventFlow.send(.ready)
try await group.waitForAll()
}
}
@ -224,7 +223,7 @@ actor SDLUDPHole {
// 处理写入逻辑
private func send(remoteAddress: SocketAddress, type: SDLPacketType, data: Data) {
let message = UDPMessage(remoteAddress: remoteAddress, type: type, data: data)
self.continuation.yield(message)
self.writeContinuation.yield(message)
}
//--MARK: 编解码器
@ -260,5 +259,4 @@ actor SDLUDPHole {
deinit {
try? self.group.syncShutdownGracefully()
}
}