From 76c7590f887f69cead6943a17f61b8a950fa6743 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 22 Sep 2025 12:12:34 +0200 Subject: [PATCH 01/14] Add structured concurrency aware ServerBootstrap bind --- Sources/NIOPosix/Bootstrap.swift | 221 +++++++++++++++++++++++++++++-- 1 file changed, 210 insertions(+), 11 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index e5dc82dfa5..2ba78d9eed 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -521,6 +521,134 @@ public final class ServerBootstrap { // MARK: Async bind methods extension ServerBootstrap { + + /// Represents a target address or socket for binding a server socket. + /// + /// `BindTarget` provides a type-safe way to specify different types of binding targets + /// for server bootstraps. It supports various address types including network addresses, + /// Unix domain sockets, VSOCK addresses, and existing socket handles. + public struct BindTarget: Sendable { + + enum Base { + case hostAndPort(host: String, port: Int) + case address(SocketAddress) + case unixDomainSocketPath(String) + case vsock(VsockAddress) + case socket(NIOBSDSocket.Handle) + } + + var base: Base + + /// Creates a binding target for a hostname and port. + /// + /// This method creates a target that will resolve the hostname and bind to the + /// specified port. The hostname resolution follows standard system behavior + /// and may resolve to both IPv4 and IPv6 addresses depending on system configuration. + /// + /// - Parameters: + /// - host: The hostname or IP address to bind to. Can be a domain name like + /// "localhost" or "example.com", or an IP address like "127.0.0.1" or "::1" + /// - port: The port number to bind to (0-65535). Use 0 to let the system + /// choose an available port + public static func hostAndPort(_ host: String, _ port: Int) -> BindTarget { + BindTarget(base: .hostAndPort(host: host, port: port)) + } + + /// Creates a binding target for a specific socket address. + /// + /// Use this method when you have a pre-constructed ``SocketAddress`` that + /// specifies the exact binding location, including IPv4, IPv6, or Unix domain addresses. + /// + /// - Parameter address: The socket address to bind to + public static func address(_ address: SocketAddress) -> BindTarget { + BindTarget(base: .address(address)) + } + + /// Creates a binding target for a Unix domain socket. + /// + /// Unix domain sockets provide high-performance inter-process communication + /// on the same machine using filesystem paths. The socket file will be created + /// at the specified path when binding occurs. + /// + /// - Parameter path: The filesystem path for the Unix domain socket. + /// Must be a valid filesystem path and should not exist + /// unless cleanup is enabled in the binding operation + /// - Warning: The path must not exist. + public static func unixDomainSocketPath(_ path: String) -> BindTarget { + BindTarget(base: .unixDomainSocketPath(path)) + } + + /// Creates a binding target for a VSOCK address. + /// + /// VSOCK (Virtual Socket) provides communication between virtual machines and their hosts, + /// or between different virtual machines on the same host. This is commonly used + /// in virtualized environments for guest-host communication. + /// + /// - Parameter vsockAddress: The VSOCK address to bind to, containing both + /// context ID (CID) and port number + /// - Note: VSOCK support depends on the underlying platform and virtualization technology + public static func vsock(_ vsockAddress: VsockAddress) -> BindTarget { + BindTarget(base: .vsock(vsockAddress)) + } + + /// Creates a binding target for an existing socket handle. + /// + /// This method allows you to use a pre-existing socket that has already been + /// created and optionally configured. This is useful for advanced scenarios where you + /// need custom socket setup before binding, or when integrating with external libraries. + /// + /// - Parameters: + /// - handle: The existing socket handle to use. Must be a valid, open socket + /// that is compatible with the intended server bootstrap type + /// - Note: The bootstrap will take ownership of the socket handle and will close + /// it when the server shuts down + public static func socket(_ handle: NIOBSDSocket.Handle) -> BindTarget { + BindTarget(base: .socket(handle)) + } + } + + /// Bind the `ServerSocketChannel` to the ``BindTarget``. This method will returns once all connections that + /// were spawned have been closed. + /// + /// - Parameters: + /// - target: The ``BindTarget`` to use. + /// - serverBackPressureStrategy: The back pressure strategy used by the server socket channel. + /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is used in the `onConnection` + /// closure. + /// - onConnection: A closure to handle the connection. Use `inbound` to read from the connection and `outbound` to write + /// to the connection. + /// - Note: This method must be cancelled to return. + @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) + public func bind( + target: BindTarget, + serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, + childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, + _ onConnection: @escaping @Sendable ( + _ inbound: NIOAsyncChannelInboundStream, + _ outbound: NIOAsyncChannelOutboundWriter + ) async throws -> () + ) async throws { + let channel = try await self.makeConnectedChannel( + target: target, + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + + try await channel.executeThenClose { inbound, outbound in + try await withThrowingDiscardingTaskGroup { group in + try await channel.executeThenClose { inbound in + for try await connectionChannel in inbound { + group.addTask { + try await connectionChannel.executeThenClose { inbound, outbound in + try await onConnection(inbound, outbound) + } + } + } + } + } + } + } + /// Bind the `ServerSocketChannel` to the `host` and `port` parameters. /// /// - Parameters: @@ -622,6 +750,87 @@ extension ServerBootstrap { to vsockAddress: VsockAddress, serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture + ) async throws -> NIOAsyncChannel { + try await self._bind( + to: vsockAddress, + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + } + + /// Use the existing bound socket file descriptor. + /// + /// - Parameters: + /// - socket: The _Unix file descriptor_ representing the bound stream socket. + /// - cleanupExistingSocketFile: Unused. + /// - serverBackPressureStrategy: The back pressure strategy used by the server socket channel. + /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is returned from the `connect` + /// method. + /// - Returns: The result of the channel initializer. + @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) + public func bind( + _ socket: NIOBSDSocket.Handle, + cleanupExistingSocketFile: Bool = false, + serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, + childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture + ) async throws -> NIOAsyncChannel { + try await self._bind( + socket, + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + } + + @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) + private func makeConnectedChannel( + target: BindTarget, + serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, + childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture> + ) async throws -> NIOAsyncChannel, Never> { + switch target.base { + case .hostAndPort(let host, let port): + try await self.bind( + to: try SocketAddress.makeAddressResolvingHost(host, port: port), + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + + case .unixDomainSocketPath(let unixDomainSocketPath): + try await self.bind( + to: try SocketAddress(unixDomainSocketPath: unixDomainSocketPath), + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + + case .address(let address): + try await self.bind( + to: address, + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + + case .vsock(let vsockAddress): + try await self._bind( + to: vsockAddress, + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + + case .socket(let handle): + try await self._bind( + handle, + serverBackPressureStrategy: serverBackPressureStrategy, + childChannelInitializer: childChannelInitializer + ) + } + } + + + @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) + private func _bind( + to vsockAddress: VsockAddress, + serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, + childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture ) async throws -> NIOAsyncChannel { func makeChannel( _ eventLoop: SelectableEventLoop, @@ -652,19 +861,9 @@ extension ServerBootstrap { }.get() } - /// Use the existing bound socket file descriptor. - /// - /// - Parameters: - /// - socket: The _Unix file descriptor_ representing the bound stream socket. - /// - cleanupExistingSocketFile: Unused. - /// - serverBackPressureStrategy: The back pressure strategy used by the server socket channel. - /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is returned from the `connect` - /// method. - /// - Returns: The result of the channel initializer. @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) - public func bind( + private func _bind( _ socket: NIOBSDSocket.Handle, - cleanupExistingSocketFile: Bool = false, serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture ) async throws -> NIOAsyncChannel { From 70b0aa12fdb7a967e0aa66a19555240c513d23ec Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 22 Sep 2025 15:39:21 +0200 Subject: [PATCH 02/14] Undeprecate NIOAsyncChannel properties --- Sources/NIOCore/AsyncChannel/AsyncChannel.swift | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/Sources/NIOCore/AsyncChannel/AsyncChannel.swift b/Sources/NIOCore/AsyncChannel/AsyncChannel.swift index ee761680a6..38797233eb 100644 --- a/Sources/NIOCore/AsyncChannel/AsyncChannel.swift +++ b/Sources/NIOCore/AsyncChannel/AsyncChannel.swift @@ -79,12 +79,11 @@ public struct NIOAsyncChannel: Sendable { /// The stream of inbound messages. /// /// - Important: The `inbound` stream is a unicast `AsyncSequence` and only one iterator can be created. - @available(*, deprecated, message: "Use the executeThenClose scoped method instead.") public var inbound: NIOAsyncChannelInboundStream { self._inbound } + /// The writer for writing outbound messages. - @available(*, deprecated, message: "Use the executeThenClose scoped method instead.") public var outbound: NIOAsyncChannelOutboundWriter { self._outbound } From afa3d9db146e27fbd379b7d721fdfe5fdbbfb197 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 22 Sep 2025 15:45:59 +0200 Subject: [PATCH 03/14] Doc how to graceful shutdown --- Sources/NIOPosix/Bootstrap.swift | 78 +++++++++++++++++++++++++++----- 1 file changed, 66 insertions(+), 12 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 2ba78d9eed..0d6cb52671 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -610,22 +610,72 @@ extension ServerBootstrap { /// Bind the `ServerSocketChannel` to the ``BindTarget``. This method will returns once all connections that /// were spawned have been closed. /// + /// # Supporting graceful shutdown + /// + /// To support a graceful server shutdown we recommend using the `ServerQuiescingHelper` from the + /// SwiftNIO extras package. The `ServerQuiescingHelper` can be installed using the + /// ``ServerBootstrap/serverChannelInitializer`` callback. + /// + /// Below you can find the code to setup a simple TCP echo server that supports graceful server closure. + /// + /// ```swift + /// let quiesce = ServerQuiescingHelper(group: group) + /// let signalSource = DispatchSource.makeSignalSource(signal: SIGINT, queue: signalQueue) + /// signalSource.setEventHandler { + /// signalSource.cancel() + /// print("\nreceived signal, initiating shutdown which should complete after the last request finished.") + /// + /// quiesce.initiateShutdown(promise: fullyShutdownPromise) + /// } + /// try await ServerBootstrap(group: self.eventLoopGroup) + /// .serverChannelInitializer { channel in + /// channel.eventLoop.makeCompletedFuture { + /// try channel.pipeline.syncOperations.addHandler(quiesce.makeServerChannelHandler(channel: channel)) + /// } + /// } + /// .serverChannelOption(.socketOption(.so_reuseaddr), value: 1) + /// .bind( + /// target: .hostAndPort(self.host, self.port), + /// childChannelInitializer: { channel in + /// channel.eventLoop.makeCompletedFuture { + /// try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(NewlineDelimiterCoder())) + /// try channel.pipeline.syncOperations.addHandler(MessageToByteHandler(NewlineDelimiterCoder())) + /// + /// return try NIOAsyncChannel( + /// wrappingChannelSynchronously: channel, + /// configuration: NIOAsyncChannel.Configuration( + /// inboundType: String.self, + /// outboundType: String.self + /// ) + /// ) + /// } + /// } + /// ) { channel in + /// print("Handling new connection") + /// await self.handleConnection(channel: channel) + /// print("Done handling connection") + /// } + /// + /// ``` + /// /// - Parameters: /// - target: The ``BindTarget`` to use. /// - serverBackPressureStrategy: The back pressure strategy used by the server socket channel. /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is used in the `onConnection` /// closure. - /// - onConnection: A closure to handle the connection. Use `inbound` to read from the connection and `outbound` to write - /// to the connection. - /// - Note: This method must be cancelled to return. + /// - onConnection: A closure to handle the connection. Use the channel's `inbound` property to read from + /// the connection and channel's `outbound` to write to the connection. + /// + /// - Note: If the server is not closed using a closure mechanism like above, the bind method must be cancelled using task + /// cancellation for the method to return. Task cancellation will force close the server and wait for all remaining + /// sub task (connection callbacks) to finish. @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) public func bind( target: BindTarget, serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, _ onConnection: @escaping @Sendable ( - _ inbound: NIOAsyncChannelInboundStream, - _ outbound: NIOAsyncChannelOutboundWriter + _ channel: NIOAsyncChannel ) async throws -> () ) async throws { let channel = try await self.makeConnectedChannel( @@ -634,18 +684,22 @@ extension ServerBootstrap { childChannelInitializer: childChannelInitializer ) - try await channel.executeThenClose { inbound, outbound in - try await withThrowingDiscardingTaskGroup { group in - try await channel.executeThenClose { inbound in - for try await connectionChannel in inbound { - group.addTask { - try await connectionChannel.executeThenClose { inbound, outbound in - try await onConnection(inbound, outbound) + try await withTaskCancellationHandler { + try await channel.executeThenClose { inbound, outbound in + try await withThrowingDiscardingTaskGroup { group in + try await channel.executeThenClose { inbound in + for try await connectionChannel in inbound { + group.addTask { + try await connectionChannel.executeThenClose { _, _ in + try await onConnection(connectionChannel) + } } } } } } + } onCancel: { + channel.channel.close(promise: nil) } } From 8dbb9617686d0027e5b0071547c496caceadc5a8 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 22 Sep 2025 15:46:23 +0200 Subject: [PATCH 04/14] Update NIOTCPEchoServer example --- Sources/NIOTCPEchoServer/Server.swift | 55 +++++++++------------------ 1 file changed, 19 insertions(+), 36 deletions(-) diff --git a/Sources/NIOTCPEchoServer/Server.swift b/Sources/NIOTCPEchoServer/Server.swift index 6f0d98c1b8..a5521985b2 100644 --- a/Sources/NIOTCPEchoServer/Server.swift +++ b/Sources/NIOTCPEchoServer/Server.swift @@ -36,44 +36,29 @@ struct Server { /// This method starts the server and handles incoming connections. func run() async throws { - let channel = try await ServerBootstrap(group: self.eventLoopGroup) + try await ServerBootstrap(group: self.eventLoopGroup) .serverChannelOption(.socketOption(.so_reuseaddr), value: 1) .bind( - host: self.host, - port: self.port - ) { channel in - channel.eventLoop.makeCompletedFuture { - // We are using two simple handlers here to frame our messages with "\n" - try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(NewlineDelimiterCoder())) - try channel.pipeline.syncOperations.addHandler(MessageToByteHandler(NewlineDelimiterCoder())) + target: .hostAndPort(self.host, self.port), + childChannelInitializer: { channel in + channel.eventLoop.makeCompletedFuture { + try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(NewlineDelimiterCoder())) + try channel.pipeline.syncOperations.addHandler(MessageToByteHandler(NewlineDelimiterCoder())) - return try NIOAsyncChannel( - wrappingChannelSynchronously: channel, - configuration: NIOAsyncChannel.Configuration( - inboundType: String.self, - outboundType: String.self + return try NIOAsyncChannel( + wrappingChannelSynchronously: channel, + configuration: NIOAsyncChannel.Configuration( + inboundType: String.self, + outboundType: String.self + ) ) - ) - } - } - - // We are handling each incoming connection in a separate child task. It is important - // to use a discarding task group here which automatically discards finished child tasks. - // A normal task group retains all child tasks and their outputs in memory until they are - // consumed by iterating the group or by exiting the group. Since, we are never consuming - // the results of the group we need the group to automatically discard them; otherwise, this - // would result in a memory leak over time. - try await withThrowingDiscardingTaskGroup { group in - try await channel.executeThenClose { inbound in - for try await connectionChannel in inbound { - group.addTask { - print("Handling new connection") - await self.handleConnection(channel: connectionChannel) - print("Done handling connection") } } + ) { channel in + print("Handling new connection") + await self.handleConnection(channel: channel) + print("Done handling connection") } - } } /// This method handles a single connection by echoing back all inbound data. @@ -82,11 +67,9 @@ struct Server { // We do this since we don't want to tear down the whole server when a single connection // encounters an error. do { - try await channel.executeThenClose { inbound, outbound in - for try await inboundData in inbound { - print("Received request (\(inboundData))") - try await outbound.write(inboundData) - } + for try await inboundData in channel.inbound { + print("Received request (\(inboundData))") + try await channel.outbound.write(inboundData) } } catch { print("Hit error: \(error)") From 227beb8e2cfafc26de079654bb2344387981e54c Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 22 Sep 2025 15:50:47 +0200 Subject: [PATCH 05/14] PR review: Use non throwing task group. --- Sources/NIOPosix/Bootstrap.swift | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 0d6cb52671..5c4a55e49a 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -686,17 +686,29 @@ extension ServerBootstrap { try await withTaskCancellationHandler { try await channel.executeThenClose { inbound, outbound in - try await withThrowingDiscardingTaskGroup { group in - try await channel.executeThenClose { inbound in - for try await connectionChannel in inbound { - group.addTask { - try await connectionChannel.executeThenClose { _, _ in - try await onConnection(connectionChannel) + // we need to dance the result dance here, since we can't throw from the + // withDiscardingTaskGroup closure. + let result = await withDiscardingTaskGroup { group -> Result in + do { + try await channel.executeThenClose { inbound in + for try await connectionChannel in inbound { + group.addTask { + do { + try await connectionChannel.executeThenClose { _, _ in + try await onConnection(connectionChannel) + } + } catch { + // ignore single connection failures? + } } } } + return .success(()) + } catch { + return .failure(error) } } + try result.get() } } onCancel: { channel.channel.close(promise: nil) From 21cd4962dbad3b2d798aa2b2f488ac1685da6e7c Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 22 Sep 2025 16:39:49 +0200 Subject: [PATCH 06/14] PR review --- Sources/NIOPosix/Bootstrap.swift | 29 ++++++++++++++--------------- 1 file changed, 14 insertions(+), 15 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 5c4a55e49a..a5903a3e33 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -531,9 +531,9 @@ extension ServerBootstrap { enum Base { case hostAndPort(host: String, port: Int) - case address(SocketAddress) + case socketAddress(SocketAddress) case unixDomainSocketPath(String) - case vsock(VsockAddress) + case vsockAddress(VsockAddress) case socket(NIOBSDSocket.Handle) } @@ -560,8 +560,8 @@ extension ServerBootstrap { /// specifies the exact binding location, including IPv4, IPv6, or Unix domain addresses. /// /// - Parameter address: The socket address to bind to - public static func address(_ address: SocketAddress) -> BindTarget { - BindTarget(base: .address(address)) + public static func socketAddress(_ address: SocketAddress) -> BindTarget { + BindTarget(base: .socketAddress(address)) } /// Creates a binding target for a Unix domain socket. @@ -587,8 +587,8 @@ extension ServerBootstrap { /// - Parameter vsockAddress: The VSOCK address to bind to, containing both /// context ID (CID) and port number /// - Note: VSOCK support depends on the underlying platform and virtualization technology - public static func vsock(_ vsockAddress: VsockAddress) -> BindTarget { - BindTarget(base: .vsock(vsockAddress)) + public static func vsockAddress(_ vsockAddress: VsockAddress) -> BindTarget { + BindTarget(base: .vsockAddress(vsockAddress)) } /// Creates a binding target for an existing socket handle. @@ -623,7 +623,7 @@ extension ServerBootstrap { /// let signalSource = DispatchSource.makeSignalSource(signal: SIGINT, queue: signalQueue) /// signalSource.setEventHandler { /// signalSource.cancel() - /// print("\nreceived signal, initiating shutdown which should complete after the last request finished.") + /// print("received signal, initiating shutdown which should complete after the last request finished.") /// /// quiesce.initiateShutdown(promise: fullyShutdownPromise) /// } @@ -666,9 +666,8 @@ extension ServerBootstrap { /// - onConnection: A closure to handle the connection. Use the channel's `inbound` property to read from /// the connection and channel's `outbound` to write to the connection. /// - /// - Note: If the server is not closed using a closure mechanism like above, the bind method must be cancelled using task - /// cancellation for the method to return. Task cancellation will force close the server and wait for all remaining - /// sub task (connection callbacks) to finish. + /// - Note: The bind method respects task cancellation which will force close the server. If you want to gracefully + /// shut-down use the quiescing helper approach as outlined above. @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) public func bind( target: BindTarget, @@ -676,7 +675,7 @@ extension ServerBootstrap { childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, _ onConnection: @escaping @Sendable ( _ channel: NIOAsyncChannel - ) async throws -> () + ) async -> () ) async throws { let channel = try await self.makeConnectedChannel( target: target, @@ -695,10 +694,10 @@ extension ServerBootstrap { group.addTask { do { try await connectionChannel.executeThenClose { _, _ in - try await onConnection(connectionChannel) + await onConnection(connectionChannel) } } catch { - // ignore single connection failures? + // ignore single connection failures } } } @@ -868,14 +867,14 @@ extension ServerBootstrap { childChannelInitializer: childChannelInitializer ) - case .address(let address): + case .socketAddress(let address): try await self.bind( to: address, serverBackPressureStrategy: serverBackPressureStrategy, childChannelInitializer: childChannelInitializer ) - case .vsock(let vsockAddress): + case .vsockAddress(let vsockAddress): try await self._bind( to: vsockAddress, serverBackPressureStrategy: serverBackPressureStrategy, From c05579c8fefbe61ab93f8db603306ce561805fb5 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Tue, 23 Sep 2025 11:59:48 +0200 Subject: [PATCH 07/14] Add once startup closure --- Sources/NIOPosix/Bootstrap.swift | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index a5903a3e33..af8f42dc8d 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -663,6 +663,8 @@ extension ServerBootstrap { /// - serverBackPressureStrategy: The back pressure strategy used by the server socket channel. /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is used in the `onConnection` /// closure. + /// - onceStartup: A closure that will be called once the server has been started. Use this to get access to + /// the port number, if you used port `0` in the ``BindTarget``. /// - onConnection: A closure to handle the connection. Use the channel's `inbound` property to read from /// the connection and channel's `outbound` to write to the connection. /// @@ -673,7 +675,8 @@ extension ServerBootstrap { target: BindTarget, serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, - _ onConnection: @escaping @Sendable ( + onceStartup: (Channel) -> () = { _ in }, + onConnection: @escaping @Sendable ( _ channel: NIOAsyncChannel ) async -> () ) async throws { @@ -683,6 +686,8 @@ extension ServerBootstrap { childChannelInitializer: childChannelInitializer ) + onceStartup(channel.channel) + try await withTaskCancellationHandler { try await channel.executeThenClose { inbound, outbound in // we need to dance the result dance here, since we can't throw from the From 378dca53f4f0fa37f6f1fa4bedd86635e0aa5ad1 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Thu, 25 Sep 2025 11:46:09 +0200 Subject: [PATCH 08/14] PR reviews --- Sources/NIOPosix/Bootstrap.swift | 33 +++++++++++++++++++-------- Sources/NIOTCPEchoServer/Server.swift | 3 +++ 2 files changed, 27 insertions(+), 9 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index af8f42dc8d..5093b010db 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -527,6 +527,7 @@ extension ServerBootstrap { /// `BindTarget` provides a type-safe way to specify different types of binding targets /// for server bootstraps. It supports various address types including network addresses, /// Unix domain sockets, VSOCK addresses, and existing socket handles. + @_spi(StructuredConcurrencyNIOAsyncChannel) public struct BindTarget: Sendable { enum Base { @@ -665,20 +666,26 @@ extension ServerBootstrap { /// closure. /// - onceStartup: A closure that will be called once the server has been started. Use this to get access to /// the port number, if you used port `0` in the ``BindTarget``. - /// - onConnection: A closure to handle the connection. Use the channel's `inbound` property to read from - /// the connection and channel's `outbound` to write to the connection. - /// + /// - handleConnection: A closure to handle the connection. Use the channel's `inbound` property to read from + /// the connection and channel's `outbound` to write to the connection. + /// - onListeningChannel: A closure that will be called once the server has been started. Use this to get access to + /// the serverChannel, if you used port `0` in the ``BindTarget``. You can also use it to + /// send events on the server channel pipeline. You must not call the channels `inbound` or + /// `outbound` properties. /// - Note: The bind method respects task cancellation which will force close the server. If you want to gracefully /// shut-down use the quiescing helper approach as outlined above. @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) + @_spi(StructuredConcurrencyNIOAsyncChannel) public func bind( target: BindTarget, serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, - onceStartup: (Channel) -> () = { _ in }, - onConnection: @escaping @Sendable ( + handleConnection: @escaping @Sendable ( _ channel: NIOAsyncChannel - ) async -> () + ) async -> (), + onListeningChannel: @Sendable @escaping ( + NIOAsyncChannel, Never> + ) async -> () = { _ in }, ) async throws { let channel = try await self.makeConnectedChannel( target: target, @@ -686,20 +693,23 @@ extension ServerBootstrap { childChannelInitializer: childChannelInitializer ) - onceStartup(channel.channel) - try await withTaskCancellationHandler { try await channel.executeThenClose { inbound, outbound in // we need to dance the result dance here, since we can't throw from the // withDiscardingTaskGroup closure. let result = await withDiscardingTaskGroup { group -> Result in + + group.addTask { + await onListeningChannel(channel) + } + do { try await channel.executeThenClose { inbound in for try await connectionChannel in inbound { group.addTask { do { try await connectionChannel.executeThenClose { _, _ in - await onConnection(connectionChannel) + await handleConnection(connectionChannel) } } catch { // ignore single connection failures @@ -1539,6 +1549,11 @@ public final class ClientBootstrap: NIOClientTCPBootstrapProtocol { // MARK: Async connect methods extension ClientBootstrap { + + struct Endpoint { + + } + /// Specify the `host` and `port` to connect to for the TCP `Channel` that will be established. /// /// - Parameters: diff --git a/Sources/NIOTCPEchoServer/Server.swift b/Sources/NIOTCPEchoServer/Server.swift index a5521985b2..170c3a6472 100644 --- a/Sources/NIOTCPEchoServer/Server.swift +++ b/Sources/NIOTCPEchoServer/Server.swift @@ -58,6 +58,9 @@ struct Server { print("Handling new connection") await self.handleConnection(channel: channel) print("Done handling connection") + } onListeningChannel: { serverChannel in + // you can access the server channel here. You must not use call + // `inbound` or `outbound` on it. } } From 1b3dde727efb68350e211109d46c871d302ab35a Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Thu, 25 Sep 2025 11:47:46 +0200 Subject: [PATCH 09/14] Use spi --- Sources/NIOTCPEchoServer/Server.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Sources/NIOTCPEchoServer/Server.swift b/Sources/NIOTCPEchoServer/Server.swift index 170c3a6472..e95c2f58cc 100644 --- a/Sources/NIOTCPEchoServer/Server.swift +++ b/Sources/NIOTCPEchoServer/Server.swift @@ -13,7 +13,7 @@ //===----------------------------------------------------------------------===// import NIOCore -import NIOPosix +@_spi(StructuredConcurrencyNIOAsyncChannel) import NIOPosix @available(macOS 14, iOS 17, tvOS 17, watchOS 10, *) @main From cf261628d9c7e0c06ee59efd083ed60768ae4476 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Thu, 25 Sep 2025 15:33:30 +0200 Subject: [PATCH 10/14] PR review --- Sources/NIOPosix/Bootstrap.swift | 27 +++++++++------------------ Sources/NIOTCPEchoServer/Server.swift | 27 +++++++++++++-------------- 2 files changed, 22 insertions(+), 32 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 5093b010db..55fceb4a44 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -664,14 +664,11 @@ extension ServerBootstrap { /// - serverBackPressureStrategy: The back pressure strategy used by the server socket channel. /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is used in the `onConnection` /// closure. - /// - onceStartup: A closure that will be called once the server has been started. Use this to get access to - /// the port number, if you used port `0` in the ``BindTarget``. - /// - handleConnection: A closure to handle the connection. Use the channel's `inbound` property to read from - /// the connection and channel's `outbound` to write to the connection. - /// - onListeningChannel: A closure that will be called once the server has been started. Use this to get access to - /// the serverChannel, if you used port `0` in the ``BindTarget``. You can also use it to - /// send events on the server channel pipeline. You must not call the channels `inbound` or - /// `outbound` properties. + /// - handleChildChannel: A closure to handle the connection. Use the channel's `inbound` property to read from + /// the connection and channel's `outbound` to write to the connection. + /// - handleServerChannel: A closure that will be called once the server has been started. Use this to get access to + /// the serverChannel, if you used port `0` in the ``BindTarget``. You can also use it to + /// send events on the server channel pipeline. /// - Note: The bind method respects task cancellation which will force close the server. If you want to gracefully /// shut-down use the quiescing helper approach as outlined above. @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) @@ -680,12 +677,10 @@ extension ServerBootstrap { target: BindTarget, serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, - handleConnection: @escaping @Sendable ( + handleChildChannel: @escaping @Sendable ( _ channel: NIOAsyncChannel ) async -> (), - onListeningChannel: @Sendable @escaping ( - NIOAsyncChannel, Never> - ) async -> () = { _ in }, + handleServerChannel: @Sendable @escaping (Channel) async -> () = { _ in }, ) async throws { let channel = try await self.makeConnectedChannel( target: target, @@ -700,7 +695,7 @@ extension ServerBootstrap { let result = await withDiscardingTaskGroup { group -> Result in group.addTask { - await onListeningChannel(channel) + await handleServerChannel(channel.channel) } do { @@ -709,7 +704,7 @@ extension ServerBootstrap { group.addTask { do { try await connectionChannel.executeThenClose { _, _ in - await handleConnection(connectionChannel) + await handleChildChannel(connectionChannel) } } catch { // ignore single connection failures @@ -1550,10 +1545,6 @@ public final class ClientBootstrap: NIOClientTCPBootstrapProtocol { extension ClientBootstrap { - struct Endpoint { - - } - /// Specify the `host` and `port` to connect to for the TCP `Channel` that will be established. /// /// - Parameters: diff --git a/Sources/NIOTCPEchoServer/Server.swift b/Sources/NIOTCPEchoServer/Server.swift index e95c2f58cc..2d3ea8da6d 100644 --- a/Sources/NIOTCPEchoServer/Server.swift +++ b/Sources/NIOTCPEchoServer/Server.swift @@ -39,26 +39,25 @@ struct Server { try await ServerBootstrap(group: self.eventLoopGroup) .serverChannelOption(.socketOption(.so_reuseaddr), value: 1) .bind( - target: .hostAndPort(self.host, self.port), - childChannelInitializer: { channel in - channel.eventLoop.makeCompletedFuture { - try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(NewlineDelimiterCoder())) - try channel.pipeline.syncOperations.addHandler(MessageToByteHandler(NewlineDelimiterCoder())) + target: .hostAndPort(self.host, self.port) + ) { channel in + channel.eventLoop.makeCompletedFuture { + try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(NewlineDelimiterCoder())) + try channel.pipeline.syncOperations.addHandler(MessageToByteHandler(NewlineDelimiterCoder())) - return try NIOAsyncChannel( - wrappingChannelSynchronously: channel, - configuration: NIOAsyncChannel.Configuration( - inboundType: String.self, - outboundType: String.self - ) + return try NIOAsyncChannel( + wrappingChannelSynchronously: channel, + configuration: NIOAsyncChannel.Configuration( + inboundType: String.self, + outboundType: String.self ) - } + ) } - ) { channel in + } handleChildChannel: { channel in print("Handling new connection") await self.handleConnection(channel: channel) print("Done handling connection") - } onListeningChannel: { serverChannel in + } handleServerChannel: { serverChannel in // you can access the server channel here. You must not use call // `inbound` or `outbound` on it. } From 6b26145491e1a0e4b797391e4c32ea4ac79ecb89 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Thu, 25 Sep 2025 15:37:52 +0200 Subject: [PATCH 11/14] Swift format --- Sources/NIOPosix/Bootstrap.swift | 50 ++++++++++++++++++++++++++------ 1 file changed, 41 insertions(+), 9 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 55fceb4a44..400bef820a 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -567,12 +567,12 @@ extension ServerBootstrap { /// Creates a binding target for a Unix domain socket. /// - /// Unix domain sockets provide high-performance inter-process communication - /// on the same machine using filesystem paths. The socket file will be created + /// Unix domain sockets provide high-performance inter-process communication + /// on the same machine using filesystem paths. The socket file will be created /// at the specified path when binding occurs. /// - /// - Parameter path: The filesystem path for the Unix domain socket. - /// Must be a valid filesystem path and should not exist + /// - Parameter path: The filesystem path for the Unix domain socket. + /// Must be a valid filesystem path and should not exist /// unless cleanup is enabled in the binding operation /// - Warning: The path must not exist. public static func unixDomainSocketPath(_ path: String) -> BindTarget { @@ -677,10 +677,11 @@ extension ServerBootstrap { target: BindTarget, serverBackPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, childChannelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, - handleChildChannel: @escaping @Sendable ( - _ channel: NIOAsyncChannel - ) async -> (), - handleServerChannel: @Sendable @escaping (Channel) async -> () = { _ in }, + handleChildChannel: + @escaping @Sendable ( + _ channel: NIOAsyncChannel + ) async -> (), + handleServerChannel: @Sendable @escaping (Channel) async -> () = { _ in } ) async throws { let channel = try await self.makeConnectedChannel( target: target, @@ -900,7 +901,6 @@ extension ServerBootstrap { } } - @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) private func _bind( to vsockAddress: VsockAddress, @@ -1571,6 +1571,38 @@ extension ClientBootstrap { ) } + /// Specify the `host` and `port` to connect to for the TCP `Channel` that will be established. + /// + /// - Parameters: + /// - host: The host to connect to. + /// - port: The port to connect to. + /// - channelInitializer: A closure to initialize the channel. The return value of this closure is returned from the `connect` + /// - channelHandler: A closure that provides scoped access to the `NIOAsyncChannel`. Use the `inbound` + /// property to read from the channel and the `outbound` property to write to the channel. + /// - Returns: The result of the `channelHandler` closure. + @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) + public func connect( + host: String, + port: Int, + channelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, + channelHandler: (NIOAsyncChannel) async throws -> sending Result + ) async throws -> sending Result { + let eventLoop = self.group.next() + let channel = try await self.connect( + host: host, + port: port, + eventLoop: eventLoop, + channelInitializer: channelInitializer, + postRegisterTransformation: { output, eventLoop in + eventLoop.makeSucceededFuture(output) + } + ) + + return try await channel.executeThenClose { _, _ in + try await channelHandler(channel) + } + } + /// Specify the `address` to connect to for the TCP `Channel` that will be established. /// /// - Parameters: From a4fde79f6415ac7b2b7541ffec00e38ac18c70f2 Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Thu, 25 Sep 2025 16:18:01 +0200 Subject: [PATCH 12/14] swift-format 2 --- Sources/NIOPosix/Bootstrap.swift | 44 +++++--------------------------- 1 file changed, 6 insertions(+), 38 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 400bef820a..e6165a8b8a 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -539,7 +539,7 @@ extension ServerBootstrap { } var base: Base - + /// Creates a binding target for a hostname and port. /// /// This method creates a target that will resolve the hostname and bind to the @@ -547,9 +547,9 @@ extension ServerBootstrap { /// and may resolve to both IPv4 and IPv6 addresses depending on system configuration. /// /// - Parameters: - /// - host: The hostname or IP address to bind to. Can be a domain name like + /// - host: The hostname or IP address to bind to. Can be a domain name like /// "localhost" or "example.com", or an IP address like "127.0.0.1" or "::1" - /// - port: The port number to bind to (0-65535). Use 0 to let the system + /// - port: The port number to bind to (0-65535). Use 0 to let the system /// choose an available port public static func hostAndPort(_ host: String, _ port: Int) -> BindTarget { BindTarget(base: .hostAndPort(host: host, port: port)) @@ -585,7 +585,7 @@ extension ServerBootstrap { /// or between different virtual machines on the same host. This is commonly used /// in virtualized environments for guest-host communication. /// - /// - Parameter vsockAddress: The VSOCK address to bind to, containing both + /// - Parameter vsockAddress: The VSOCK address to bind to, containing both /// context ID (CID) and port number /// - Note: VSOCK support depends on the underlying platform and virtualization technology public static func vsockAddress(_ vsockAddress: VsockAddress) -> BindTarget { @@ -680,8 +680,8 @@ extension ServerBootstrap { handleChildChannel: @escaping @Sendable ( _ channel: NIOAsyncChannel - ) async -> (), - handleServerChannel: @Sendable @escaping (Channel) async -> () = { _ in } + ) async -> Void, + handleServerChannel: @Sendable @escaping (Channel) async -> Void = { _ in } ) async throws { let channel = try await self.makeConnectedChannel( target: target, @@ -1571,38 +1571,6 @@ extension ClientBootstrap { ) } - /// Specify the `host` and `port` to connect to for the TCP `Channel` that will be established. - /// - /// - Parameters: - /// - host: The host to connect to. - /// - port: The port to connect to. - /// - channelInitializer: A closure to initialize the channel. The return value of this closure is returned from the `connect` - /// - channelHandler: A closure that provides scoped access to the `NIOAsyncChannel`. Use the `inbound` - /// property to read from the channel and the `outbound` property to write to the channel. - /// - Returns: The result of the `channelHandler` closure. - @available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *) - public func connect( - host: String, - port: Int, - channelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, - channelHandler: (NIOAsyncChannel) async throws -> sending Result - ) async throws -> sending Result { - let eventLoop = self.group.next() - let channel = try await self.connect( - host: host, - port: port, - eventLoop: eventLoop, - channelInitializer: channelInitializer, - postRegisterTransformation: { output, eventLoop in - eventLoop.makeSucceededFuture(output) - } - ) - - return try await channel.executeThenClose { _, _ in - try await channelHandler(channel) - } - } - /// Specify the `address` to connect to for the TCP `Channel` that will be established. /// /// - Parameters: From b6563f5bc092d72da8617213bdcce98ef0c211df Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 29 Sep 2025 16:10:59 +0200 Subject: [PATCH 13/14] Add ClientBootstrap methods. --- Sources/NIOPosix/Bootstrap.swift | 135 ++++++++++++++++++++++++++++++- 1 file changed, 131 insertions(+), 4 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 0bc4c106c5..2d83ee1087 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -572,8 +572,7 @@ extension ServerBootstrap { /// at the specified path when binding occurs. /// /// - Parameter path: The filesystem path for the Unix domain socket. - /// Must be a valid filesystem path and should not exist - /// unless cleanup is enabled in the binding operation + /// Must be a valid filesystem path and should not exist. /// - Warning: The path must not exist. public static func unixDomainSocketPath(_ path: String) -> BindTarget { BindTarget(base: .unixDomainSocketPath(path)) @@ -662,8 +661,8 @@ extension ServerBootstrap { /// - Parameters: /// - target: The ``BindTarget`` to use. /// - serverBackPressureStrategy: The back pressure strategy used by the server socket channel. - /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is used in the `onConnection` - /// closure. + /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is used in the + /// `handleChildChannel` closure. /// - handleChildChannel: A closure to handle the connection. Use the channel's `inbound` property to read from /// the connection and channel's `outbound` to write to the connection. /// - handleServerChannel: A closure that will be called once the server has been started. Use this to get access to @@ -1545,6 +1544,134 @@ public final class ClientBootstrap: NIOClientTCPBootstrapProtocol { extension ClientBootstrap { + /// Represents a target address or socket for creating a client socket. + /// + /// `ConnectTarget` provides a type-safe way to specify different types of connecting targets + /// for client bootstraps. It supports various address types including network addresses, + /// Unix domain sockets, VSOCK addresses, and existing socket handles. + @_spi(StructuredConcurrencyNIOAsyncChannel) + public struct ConnectTarget: Sendable { + + enum Base { + case hostAndPort(host: String, port: Int) + case socketAddress(SocketAddress) + case unixDomainSocketPath(String) + case vsockAddress(VsockAddress) + case socket(NIOBSDSocket.Handle) + } + + var base: Base + + /// Creates a connect target for a hostname and port. + /// + /// This method creates a target that will resolve the hostname and connect to the + /// specified port. The hostname resolution follows standard system behavior + /// and may resolve to both IPv4 and IPv6 addresses depending on system configuration. + /// + /// - Parameters: + /// - host: The hostname or IP address to bind to. Can be a domain name like + /// "localhost" or "example.com", or an IP address like "189.201.14.13" or "::1" + /// - port: The port number to connect to (0-65535). + public static func hostAndPort(_ host: String, _ port: Int) -> ConnectTarget { + ConnectTarget(base: .hostAndPort(host: host, port: port)) + } + + /// Creates a connect target for a specific socket address. + /// + /// Use this method when you have a pre-constructed ``SocketAddress`` that + /// specifies the exact connect location, including IPv4, IPv6, or Unix domain addresses. + /// + /// - Parameter address: The socket address to connect to + public static func socketAddress(_ address: SocketAddress) -> ConnectTarget { + ConnectTarget(base: .socketAddress(address)) + } + + /// Creates a connect target for a Unix domain socket. + /// + /// Unix domain sockets provide high-performance inter-process communication + /// on the same machine using filesystem paths. The socket file needs to exist in + /// order to connect to it. + /// + /// - Parameter path: The filesystem path for the Unix domain socket. + /// Must be a valid filesystem path and should exist. + /// - Warning: The path must exist. + public static func unixDomainSocketPath(_ path: String) -> ConnectTarget { + ConnectTarget(base: .unixDomainSocketPath(path)) + } + + /// Creates a connect target for a VSOCK address. + /// + /// VSOCK (Virtual Socket) provides communication between virtual machines and their hosts, + /// or between different virtual machines on the same host. This is commonly used + /// in virtualized environments for guest-host communication. + /// + /// - Parameter vsockAddress: The VSOCK address to connect to, containing both + /// context ID (CID) and port number + /// - Note: VSOCK support depends on the underlying platform and virtualization technology + public static func vsockAddress(_ vsockAddress: VsockAddress) -> ConnectTarget { + ConnectTarget(base: .vsockAddress(vsockAddress)) + } + + /// Creates a connect target for an existing socket handle. + /// + /// This method allows you to use a pre-existing socket that has already been + /// created and optionally configured. This is useful for advanced scenarios where you + /// need custom socket setup before binding, or when integrating with external libraries. + /// + /// - Parameters: + /// - handle: The existing socket handle to use. Must be a valid, open socket + /// that is compatible with the intended server bootstrap type + /// - Note: The bootstrap will take ownership of the socket handle and will close + /// it when the server shuts down + public static func socket(_ handle: NIOBSDSocket.Handle) -> ConnectTarget { + ConnectTarget(base: .socket(handle)) + } + } + + /// Create a client connection to the ``ConnectTarget``. The connection will be closed once the scope of the + /// `handleChannel` closure is exited. + /// + /// - Parameters: + /// - target: The ``ConnectTarget`` to use. + /// - backPressureStrategy: The back pressure strategy used by the channel. + /// - childChannelInitializer: A closure to initialize the channel. The return value of this closure is used in handleChannel + /// closure. + /// - handleChannel: A closure to handle the client connection. Use the channel's `inbound` property to read from + /// the connection and channel's `outbound` to write to the connection. + @_spi(StructuredConcurrencyNIOAsyncChannel) + @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) + public func connect( + target: ConnectTarget, + backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, + channelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, + handleChannel: + @escaping @Sendable ( + _ channel: NIOAsyncChannel + ) async -> sending Result + ) async throws -> sending Result { + let channel: NIOAsyncChannel + switch target.base { + case .socketAddress(let socketAddress): + channel = try await self.connect(to: socketAddress, channelInitializer: channelInitializer) + + case .hostAndPort(let host, let port): + channel = try await self.connect(host: host, port: port, channelInitializer: channelInitializer) + + case .unixDomainSocketPath(let path): + channel = try await self.connect(unixDomainSocketPath: path, channelInitializer: channelInitializer) + + case .vsockAddress(let vsockAddress): + channel = try await self.connect(to: vsockAddress, channelInitializer: channelInitializer) + + case .socket(let handle): + channel = try await self.withConnectedSocket(handle, channelInitializer: channelInitializer) + } + + return try await channel.executeThenClose { _, _ in + await handleChannel(channel) + } + } + /// Specify the `host` and `port` to connect to for the TCP `Channel` that will be established. /// /// - Parameters: From 91c1ae32574fb3b7ce4b8b8962b01a1cd3e7895d Mon Sep 17 00:00:00 2001 From: Fabian Fett Date: Mon, 29 Sep 2025 17:51:58 +0200 Subject: [PATCH 14/14] ClientBootstrap structured as well --- Sources/NIOPosix/Bootstrap.swift | 7 ++--- Sources/NIOTCPEchoClient/Client.swift | 39 ++++++++++++++------------- 2 files changed, 22 insertions(+), 24 deletions(-) diff --git a/Sources/NIOPosix/Bootstrap.swift b/Sources/NIOPosix/Bootstrap.swift index 2d83ee1087..4acbdc8a9f 100644 --- a/Sources/NIOPosix/Bootstrap.swift +++ b/Sources/NIOPosix/Bootstrap.swift @@ -1638,16 +1638,13 @@ extension ClientBootstrap { /// closure. /// - handleChannel: A closure to handle the client connection. Use the channel's `inbound` property to read from /// the connection and channel's `outbound` to write to the connection. - @_spi(StructuredConcurrencyNIOAsyncChannel) @available(macOS 14.0, iOS 17.0, watchOS 10.0, tvOS 17.0, *) + @_spi(StructuredConcurrencyNIOAsyncChannel) public func connect( target: ConnectTarget, backPressureStrategy: NIOAsyncSequenceProducerBackPressureStrategies.HighLowWatermark? = nil, channelInitializer: @escaping @Sendable (Channel) -> EventLoopFuture>, - handleChannel: - @escaping @Sendable ( - _ channel: NIOAsyncChannel - ) async -> sending Result + handleChannel: (_ channel: NIOAsyncChannel) async -> sending Result ) async throws -> sending Result { let channel: NIOAsyncChannel switch target.base { diff --git a/Sources/NIOTCPEchoClient/Client.swift b/Sources/NIOTCPEchoClient/Client.swift index 67f917e9da..4e04624dfd 100644 --- a/Sources/NIOTCPEchoClient/Client.swift +++ b/Sources/NIOTCPEchoClient/Client.swift @@ -13,7 +13,7 @@ //===----------------------------------------------------------------------===// import NIOCore -import NIOPosix +@_spi(StructuredConcurrencyNIOAsyncChannel) import NIOPosix @available(macOS 14, iOS 17, tvOS 17, watchOS 10, *) @main @@ -48,11 +48,10 @@ struct Client { } private func sendRequest(number: Int) async throws { - let channel = try await ClientBootstrap(group: self.eventLoopGroup) + try await ClientBootstrap(group: self.eventLoopGroup) .connect( - host: self.host, - port: self.port - ) { channel in + target: ClientBootstrap.ConnectTarget.hostAndPort(self.host, self.port) + ) { channel -> EventLoopFuture> in channel.eventLoop.makeCompletedFuture { // We are using two simple handlers here to frame our messages with "\n" try channel.pipeline.syncOperations.addHandler(ByteToMessageHandler(NewlineDelimiterCoder())) @@ -66,21 +65,23 @@ struct Client { ) ) } + } handleChannel: { channel in + print("Connection(\(number)): Writing request") + do { + try await channel.outbound.write("Hello on connection \(number)") + + for try await inboundData in channel.inbound { + print("Connection(\(number)): Received response (\(inboundData))") + + // We only expect a single response so we can exit here. + // Once, we exit out of this loop and the references to the `NIOAsyncChannel` are dropped + // the connection is going to close itself. + break + } + } catch { + print("Unexpected error occured: \(error)") + } } - - try await channel.executeThenClose { inbound, outbound in - print("Connection(\(number)): Writing request") - try await outbound.write("Hello on connection \(number)") - - for try await inboundData in inbound { - print("Connection(\(number)): Received response (\(inboundData))") - - // We only expect a single response so we can exit here. - // Once, we exit out of this loop and the references to the `NIOAsyncChannel` are dropped - // the connection is going to close itself. - break - } - } } }