Ayush yadav001/invalid header listener (#71)

Author: ayush-yadav001

Sources/RSocketCore/Client/TransportChannelHandler.swift

@@ -31,6 +31,6 @@ public protocol TransportChannelHandler {
         channel: Channel,
         maximumIncomingFragmentSize: Int,
         endpoint: Endpoint,
-        upgradeComplete: @escaping () -> EventLoopFuture<Void>
-    ) -> EventLoopFuture<Void>
+        upgradeComplete: @escaping () -> EventLoopFuture<Void>,
+        resultHandler: @escaping (Result<Void, Swift.Error>) -> EventLoopFuture<Void>)-> EventLoopFuture<Void>
 }

Sources/RSocketNIOChannel/ClientBootstrap.swift

@@ -65,15 +65,26 @@ extension ClientBootstrap: RSocketCore.ClientBootstrap {
                     transport.addChannelHandler(
                         channel: channel,
                         maximumIncomingFragmentSize: config.fragmentation.maximumIncomingFragmentSize,
-                        endpoint: endpoint
-                    ) {
+                        endpoint: endpoint,
+                        upgradeComplete:{
                         channel.pipeline.addRSocketClientHandlers(
                             config: config,
                             setupPayload: payload,
                             responder: responder,
                             connectedPromise: requesterPromise
                         )
-                    }
+                    },resultHandler: { result in
+                        if case .failure(let error) = result {
+                            requesterPromise.fail(error)
+                            return requesterPromise.futureResult.eventLoop.makeFailedFuture(error)
+                        }
+                       return channel.pipeline.addRSocketClientHandlers(
+                            config: config,
+                            setupPayload: payload,
+                            responder: responder,
+                            connectedPromise: requesterPromise
+                        )
+                    })
                 }
                 if sslContext != nil || endpoint.requiresTLS {
                     do {

Sources/RSocketTCPTransport/TCPTransport.swift

@@ -32,12 +32,7 @@ public struct TCPTransport {
 }
 
 extension TCPTransport: TransportChannelHandler {
-    public func addChannelHandler(
-        channel: Channel,
-        maximumIncomingFragmentSize: Int,
-        endpoint: Endpoint,
-        upgradeComplete: @escaping () -> EventLoopFuture<Void>
-    ) -> EventLoopFuture<Void> {
+    public func addChannelHandler(channel: Channel, maximumIncomingFragmentSize: Int, endpoint: Endpoint, upgradeComplete: @escaping () -> EventLoopFuture<Void>, resultHandler: @escaping (Result<Void, Swift.Error>) -> EventLoopFuture<Void>) -> EventLoopFuture<Void> {
         channel.pipeline.addHandlers([
             ByteToMessageHandler(
                 LengthFieldBasedFrameDecoder(lengthFieldBitLength: .threeBytes),

Sources/RSocketTSChannel/ClientBootstrap.swift

@@ -71,15 +71,26 @@ extension ClientBootstrap: RSocketCore.ClientBootstrap {
                 transport.addChannelHandler(
                     channel: channel,
                     maximumIncomingFragmentSize: config.fragmentation.maximumIncomingFragmentSize,
-                    endpoint: endpoint
-                ) {
+                    endpoint: endpoint,
+                    upgradeComplete:{
                     channel.pipeline.addRSocketClientHandlers(
                         config: config,
                         setupPayload: payload,
                         responder: responder,
                         connectedPromise: requesterPromise
                     )
-                }
+                    },resultHandler: { result in
+                        if case .failure(let error) = result {
+                            requesterPromise.fail(error)
+                            return requesterPromise.futureResult.eventLoop.makeFailedFuture(error)
+                        }
+                       return channel.pipeline.addRSocketClientHandlers(
+                            config: config,
+                            setupPayload: payload,
+                            responder: responder,
+                            connectedPromise: requesterPromise
+                        )
+                    })
             }
             .connect(host: endpoint.host, port: endpoint.port)
         connectFuture.cascadeFailure(to: requesterPromise)

Sources/RSocketWSTransport/HTTPInitialRequestHandler.swift

@@ -26,12 +26,14 @@ internal final class HTTPInitialRequestHandler: ChannelInboundHandler, Removable
     private let port: Int
     private let uri: String
     private let additionalHTTPHeader: [String: String]
-
-    internal init(host: String, port: Int, uri: String, additionalHTTPHeader: [String: String]) {
+    private let completionhandler :(Result<Void, Error>) -> EventLoopFuture<Void>
+    
+    internal init(host: String, port: Int, uri: String, additionalHTTPHeader: [String: String],completionhandler : @escaping (Result<Void,Error>)-> EventLoopFuture<Void>){
         self.host = host
         self.port = port
         self.uri = uri
         self.additionalHTTPHeader = additionalHTTPHeader
+        self.completionhandler = completionhandler
     }
 
     internal func channelActive(context: ChannelHandlerContext) {
@@ -63,6 +65,7 @@ internal final class HTTPInitialRequestHandler: ChannelInboundHandler, Removable
     internal func errorCaught(context: ChannelHandlerContext, error: Error) {
         // As we are not really interested getting notified on success or failure
         // we just pass nil as promise to reduce allocations.
+        let _ = completionhandler(.failure(error))
         context.close(promise: nil)
     }
 }

Sources/RSocketWSTransport/WSTransport.swift

@@ -72,17 +72,13 @@ extension WSTransport.Endpoint: Endpoint {
 }
 
 extension WSTransport: TransportChannelHandler {
-    public func addChannelHandler(
-        channel: Channel,
-        maximumIncomingFragmentSize: Int,
-        endpoint: Endpoint,
-        upgradeComplete: @escaping () -> EventLoopFuture<Void>
-    ) -> EventLoopFuture<Void> {
+    public func addChannelHandler(channel: Channel, maximumIncomingFragmentSize: Int, endpoint: Endpoint, upgradeComplete: @escaping () -> EventLoopFuture<Void>, resultHandler : @escaping (Result<Void, Swift.Error>) -> EventLoopFuture<Void>) -> EventLoopFuture<Void> {
         let httpHandler = HTTPInitialRequestHandler(
             host: endpoint.host,
             port: endpoint.port,
             uri: endpoint.uri,
-            additionalHTTPHeader: endpoint.additionalHTTPHeader
+            additionalHTTPHeader: endpoint.additionalHTTPHeader,
+            completionhandler: resultHandler
         )
         let websocketUpgrader = NIOWebSocketClientUpgrader(
             maxFrameSize: maximumIncomingFragmentSize,