Allow application event to influence batch sending

Sources/SwiftNetwork/Protocols/NetworkEvents.swift

@@ -111,6 +111,7 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible {
     enum InternalEvent: Equatable {
         case dataStall
         case connectionIdle(idle: Bool)
+        case outboundDataPending(pending: Bool)
         case quic(event: QUICApplicationEvent)
         #if !NETWORK_EMBEDDED
         case custom(event: any DomainSpecificApplicationEvent)
@@ -120,6 +121,7 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible {
             switch (lhs, rhs) {
             case (.dataStall, .dataStall): return true
             case (.connectionIdle(let lState), .connectionIdle(let rState)): return lState == rState
+            case (.outboundDataPending(let lState), .outboundDataPending(let rState)): return lState == rState
             default:
                 return false
             }
@@ -131,6 +133,7 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible {
         switch internalEvent {
         case .dataStall: return nil
         case .connectionIdle: return nil
+        case .outboundDataPending: return nil
         case .quic(let event): return event.domain
         #if !NETWORK_EMBEDDED
         case .custom(let event): return event.domain
@@ -154,6 +157,14 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible {
         .init(internalEvent: .connectionIdle(idle: false))
     }
 
+    static public var outboundDataBatchStart: ApplicationEvent {
+        .init(internalEvent: .outboundDataPending(pending: true))
+    }
+
+    static public var outboundDataBatchEnd: ApplicationEvent {
+        .init(internalEvent: .outboundDataPending(pending: false))
+    }
+
     public init(quicEvent: QUICApplicationEvent) {
         self.internalEvent = .quic(event: quicEvent)
     }
@@ -173,6 +184,12 @@ public struct ApplicationEvent: Sendable, Equatable, CustomStringConvertible {
             } else {
                 return "Connection Reused"
             }
+        case .outboundDataPending(let pending):
+            if pending {
+                return "Outbound Data Pending"
+            } else {
+                return "Outbound Data No Longer Pending"
+            }
         case .quic(let event): return event.description
         #if !NETWORK_EMBEDDED
         case .custom(let event): return event.description

Sources/SwiftNetwork/QUIC/QUICConnection.swift

@@ -359,6 +359,8 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol,
     private(set) var testSendingShortPackets = false
     private(set) var migrationSupported = false
 
+    private var pendOutboundData = false  // Don't immediately process application sends
+
     // false == IPv6, true == IPv4
     private(set) var initialAddressIsIPv4 = false
 
@@ -2460,6 +2462,11 @@ public final class QUICConnection: ManyToManyApplicationStreamProtocol,
         }
         // Note: trigger sending of any frames based on this external event
         checkConnectionIdle()
+
+        guard !pendOutboundData else {
+            log.datapath("Outbound data pended, ignore send frames")
+            return
+        }
         sendFrames()
     }
 
@@ -5490,6 +5497,11 @@ extension QUICConnection {
 
         // Note: trigger sendFrames() based on this external event
         checkConnectionIdle()
+
+        guard !pendOutboundData else {
+            log.datapath("Outbound data pended, ignore send frames")
+            return
+        }
         if !sendFrames() {
             log.datapath("failed to send DATAGRAM frames")
         }
@@ -5889,6 +5901,19 @@ extension QUICConnection {
     }
 
     public func handleApplicationEvent(_ event: ApplicationEvent) -> HandleNetworkEventResult {
+        if event == .outboundDataBatchStart {
+            // Start pending processing
+            pendOutboundData = true
+            return .consumed
+        }
+
+        if event == .outboundDataBatchEnd {
+            // Stop pending processing, resume sending
+            pendOutboundData = false
+            sendFrames()
+            return .consumed
+        }
+
         guard let quicEvent = event.quicEvent else {
             return .unconsumed
         }

Tests/SwiftNetworkTests/QUICTestHarness.swift

@@ -550,7 +550,8 @@ final class QUICTestHarness {
         dataGenerator: TestDataGenerator,
         streamIndex: Int,
         readChunkSize: Int = .max,
-        timeout: TimeInterval = 5.0
+        timeout: TimeInterval = 5.0,
+        shouldBatchSends: Bool = false
     ) {
         guard let state else {
             XCTFail("State must be non-nil")
@@ -585,6 +586,9 @@ final class QUICTestHarness {
         // Write on the client stream
         context.async {
             var chunkCount = 1
+            if shouldBatchSends {
+                state.clientHarness.invokeApplicationEvent(.outboundDataBatchStart)
+            }
             for dataChunk in dataGenerator {
                 var writeResult = false
                 if dataGenerator.numberOfBlocks == chunkCount && dataGenerator.sendFIN {
@@ -595,6 +599,9 @@ final class QUICTestHarness {
                 XCTAssertTrue(writeResult)
                 chunkCount += 1
             }
+            if shouldBatchSends {
+                state.clientHarness.invokeApplicationEvent(.outboundDataBatchEnd)
+            }
         }
 
         // Wait for the server stream
@@ -910,6 +917,7 @@ final class QUICTestHarness {
         sendStreamStopSendingError: Bool = false,
         verifyResetStreamHalfClosure: Bool = false,
         shouldMarkIdle: Bool = false,
+        shouldBatchSends: Bool = false,
         clientOptions: ProtocolOptions<QUICProtocol> = QUICProtocol.options(),
         serverOptions: ProtocolOptions<QUICProtocol> = QUICProtocol.options(),
         sendMaxStreamUpdate: Bool = false,
@@ -1010,7 +1018,8 @@ final class QUICTestHarness {
                         dataGenerator: generator,
                         streamIndex: index,
                         readChunkSize: clientReadChunkSize,
-                        timeout: timeout
+                        timeout: timeout,
+                        shouldBatchSends: shouldBatchSends
                     )
                 } else {
                     XCTAssertTrue(dataBlock == nil && blockSize == 0 && blockCount == 0)

Tests/SwiftNetworkTests/SwiftNetworkQUICHarnessTests.swift

@@ -389,6 +389,10 @@ final class SwiftNetworkQUICHarnessTests: NetTestCase {
         QUICTestHarness().runQUICTest(blockSize: 10240, blockCount: 4)
     }
 
+    func testQUICEcho40KiBBatched() {
+        QUICTestHarness().runQUICTest(blockSize: 10240, blockCount: 4, shouldBatchSends: true)
+    }
+
     func testQUICEcho40KiBSmallReads() {
         QUICTestHarness().runQUICTest(blockSize: 10240, blockCount: 4, clientReadChunkSize: 1000)
     }