If connectSocket() completes directly we need to mark channel as acti…

…ve (#205)

Motivation:

If connectSocket() completes directly we failed to mark the Channel active by calling becomeActive0(...).

Modifications:

Correctly call becomeActive0(...)

Result:

Mark Channel active if connect succeed directly.

Sources/NIO/SocketChannel.swift

@@ -691,7 +691,7 @@ class BaseSocketChannel<T: BaseSocket>: SelectableChannel, ChannelCore {
                 registerForWritable()
             } else {
                 self.updateCachedAddressesFromSocket()
-                promise?.succeed(result: ())
+                becomeActive0(promise: promise)
             }
         } catch let error {
             promise?.fail(error: error)
@@ -854,7 +854,7 @@ final class SocketChannel: BaseSocketChannel<Socket> {
         return .socketChannel(self, interested)
     }
 
-    fileprivate init(socket: Socket, parent: Channel? = nil, eventLoop: SelectableEventLoop) throws {
+    init(socket: Socket, parent: Channel? = nil, eventLoop: SelectableEventLoop) throws {
         self.pendingWrites = PendingStreamWritesManager(iovecs: eventLoop.iovecs, storageRefs: eventLoop.storageRefs)
         try super.init(socket: socket, parent: parent, eventLoop: eventLoop, recvAllocator: AdaptiveRecvByteBufferAllocator())
     }

Tests/NIOTests/SocketChannelTest+XCTest.swift

@@ -35,6 +35,7 @@ extension SocketChannelTest {
                 ("testAcceptFailsWithENOMEM", testAcceptFailsWithENOMEM),
                 ("testAcceptFailsWithEFAULT", testAcceptFailsWithEFAULT),
                 ("testSetGetOptionClosedServerSocketChannel", testSetGetOptionClosedServerSocketChannel),
+                ("testConnect", testConnect),
            ]
    }
 }

Tests/NIOTests/SocketChannelTest.swift

@@ -176,4 +176,47 @@ public class SocketChannelTest : XCTestCase {
         try assertSetGetOptionOnOpenAndClosed(channel: clientChannel, option: ChannelOptions.allowRemoteHalfClosure, value: true)
         try assertSetGetOptionOnOpenAndClosed(channel: serverChannel, option: ChannelOptions.backlog, value: 100)
     }
+
+    public func testConnect() throws {
+        final class ActiveVerificationHandler: ChannelInboundHandler {
+            typealias InboundIn = ByteBuffer
+            typealias InboundOut = ByteBuffer
+
+            private let promise: EventLoopPromise<Void>
+
+            init(_ promise: EventLoopPromise<Void>) {
+                self.promise = promise
+            }
+
+            func channelActive(ctx: ChannelHandlerContext) {
+                promise.succeed(result: ())
+            }
+        }
+
+        class ConnectSocket: Socket {
+            init() throws {
+                try super.init(protocolFamily: PF_INET, type: Posix.SOCK_STREAM)
+            }
+
+            override func connect(to address: SocketAddress) throws -> Bool {
+                return true
+            }
+        }
+
+        let group = MultiThreadedEventLoopGroup(numThreads: 1)
+        defer {
+            XCTAssertNoThrow(try group.syncShutdownGracefully())
+        }
+        let socket = try ConnectSocket()
+        let channel = try SocketChannel(socket: socket, eventLoop: group.next() as! SelectableEventLoop)
+        let promise: EventLoopPromise<Void> = channel.eventLoop.newPromise()
+
+        XCTAssertNoThrow(try channel.register().wait())
+        XCTAssertNoThrow(try channel.pipeline.add(handler: ActiveVerificationHandler(promise)).wait())
+        XCTAssertNoThrow(try channel.connect(to: SocketAddress.init(ipAddress: "127.0.0.1", port: 0)).wait())
+
+        try channel.close().wait()
+        try channel.closeFuture.wait()
+        try promise.futureResult.wait()
+    }
 }