Support attaching application-layer stream protocols to existing flows

Sources/SwiftNetwork/EndpointFlow/EndpointFlowProtocols.swift

@@ -446,6 +446,13 @@ final class StreamEndpointFlowProtocol: EndpointFlowProtocol<InboundStreamLinkag
         throw NetworkError.posix(EINVAL)
     }
 
+    func attachLowerStreamProtocolToExistingFlow(
+        listener: StreamListenerLinkage,
+        flowReference: ProtocolInstanceReference
+    ) throws(NetworkError) {
+        throw NetworkError.posix(EINVAL)
+    }
+
     convenience init(
         identifier: String = "",
         local: Endpoint?,

Sources/SwiftNetwork/Protocols/OneToOneProtocol.swift

@@ -702,6 +702,22 @@ extension OneToOneProtocolHandler where Self: ~Copyable, LowerProtocol == Outbou
         )
     }
 
+    public mutating func attachLowerStreamProtocolToExistingFlow(
+        listener: StreamListenerLinkage,
+        flowReference: ProtocolInstanceReference
+    ) throws(NetworkError) {
+        guard lower.isDetached else {
+            throw NetworkError.posix(EALREADY)
+        }
+        if upper.isDetached {
+            passthroughEvents = false
+        }
+        self.lower = try listener.invokeAttachUpperStreamProtocolToExistingFlow(
+            effectiveSelfReference,
+            flowReference: flowReference
+        )
+    }
+
     public mutating func invokeReceiveStreamData(
         minimumBytes: Int,
         maximumBytes: Int

Sources/SwiftNetwork/Protocols/ProtocolControlHandlers.swift

@@ -761,6 +761,31 @@ extension ProtocolInstanceReference {
         }
     }
 
+    public func attachLowerStreamProtocolToExistingFlow(
+        listener: StreamListenerLinkage,
+        flowReference: ProtocolInstanceReference
+    ) throws(NetworkError) {
+        try self.fromExternal { () throws(NetworkError) in
+            switch self.reference {
+            case .none: fatalError("Cannot attach to empty protocol")
+            case .tls(var instance):
+                try instance.attachLowerStreamProtocolToExistingFlow(listener: listener, flowReference: flowReference)
+            case .streamEndpointFlow(let instance):
+                try instance.attachLowerStreamProtocolToExistingFlow(listener: listener, flowReference: flowReference)
+            #if !NETWORK_EMBEDDED
+            case .custom(let container, let index):
+                try container.accessInboundStreamHandler(at: index) { instance throws(NetworkError) in
+                    try instance.attachLowerStreamProtocolToExistingFlow(
+                        listener: listener,
+                        flowReference: flowReference
+                    )
+                }
+            #endif
+            default: fatalError("Protocol cannot accept attachLowerStreamProtocolToExistingFlow call")
+            }
+        }
+    }
+
     #if !NETWORK_EMBEDDED
     public func attachLowerProtocolForNewPath(
         _ lowerProtocol: ProtocolInstanceReference,

Sources/SwiftNetwork/Protocols/ProtocolListenerHandlers.swift

@@ -383,4 +383,19 @@ extension ProtocolInstanceReference {
             }
         }
     }
+
+    func deliverEnqueuedInboundStreamData(flowReference: ProtocolInstanceReference) throws(NetworkError) {
+        try self.fromExternal { () throws(NetworkError) in
+            switch self.reference {
+            #if !NETWORK_NO_SWIFT_QUIC
+            case .quic(let instance):
+                try instance.deliverEnqueuedInboundStreamData(
+                    flow: MultiplexedFlowIdentifier(inboundReference: flowReference)
+                )
+            #endif
+            default:
+                return
+            }
+        }
+    }
 }

Sources/SwiftNetwork/Protocols/ProtocolOptions.swift

@@ -154,6 +154,10 @@ public class AbstractProtocolOptions: PerProtocolOptions, Hashable {
         self.protocolInstance = reference
     }
 
+    public func newProtocolInstance(context: NetworkContext) -> ProtocolInstanceReference? {
+        nil
+    }
+
     public var identifier: ProtocolIdentifier
 
     public var topID: Int? = nil
@@ -232,6 +236,10 @@ public final class ProtocolOptions<P: NetworkProtocol>: AbstractProtocolOptions
         perProtocolOptions?.serialize() ?? nil
     }
 
+    public override func newProtocolInstance(context: NetworkContext) -> ProtocolInstanceReference? {
+        P().newProtocolInstance(context: context)
+    }
+
     public init(protocolIdentifier: ProtocolIdentifier, perProtocolOptions: P.Options?) {
         self.perProtocolOptions = perProtocolOptions
         super.init(identifier: protocolIdentifier)

Sources/SwiftNetwork/Protocols/ProtocolStreamHandlers.swift

@@ -150,6 +150,11 @@ public protocol InboundStreamHandler: ~Copyable, InboundDataHandler where LowerP
         path: PathProperties?
     ) throws(NetworkError)
 
+    mutating func attachLowerStreamProtocolToExistingFlow(
+        listener: StreamListenerLinkage,
+        flowReference: ProtocolInstanceReference
+    ) throws(NetworkError)
+
     mutating func handleInboundAbortedEvent(_ from: ProtocolInstanceReference, error: NetworkError?)
     mutating func handleOutboundAbortedEvent(_ from: ProtocolInstanceReference, error: NetworkError?)
 }

Sources/SwiftNetwork/Protocols/TopProtocol.swift

@@ -324,6 +324,19 @@ extension TopProtocolHandler where Self: ~Copyable, LowerProtocol == OutboundStr
             path: path
         )
     }
+
+    public mutating func attachLowerStreamProtocolToExistingFlow(
+        listener: StreamListenerLinkage,
+        flowReference: ProtocolInstanceReference
+    ) throws(NetworkError) {
+        guard lower.isDetached else {
+            throw NetworkError.posix(EALREADY)
+        }
+        self.lower = try listener.invokeAttachUpperStreamProtocolToExistingFlow(
+            reference,
+            flowReference: flowReference
+        )
+    }
 }
 
 @available(Network 0.1.0, *)

Sources/SwiftNetworkBenchmarks/StreamPerfTestHandler.swift

@@ -284,6 +284,14 @@ extension StreamPerfTestHandler: UpperProtocolHandler {
         throw NetworkError.posix(ENOTSUP)
     }
 
+    // InboundStreamHandler conformance
+    public func attachLowerStreamProtocolToExistingFlow(
+        listener: StreamListenerLinkage,
+        flowReference: ProtocolInstanceReference
+    ) throws(NetworkError) {
+        throw NetworkError.posix(ENOTSUP)
+    }
+
     // UpperProtocolHandler conformance
     public func attachLowerProtocol(
         _ lowerProtocol: ProtocolInstanceReference,