From 7a18352b5edfe543ebd0e6fec9c0a1b2d9a878b6 Mon Sep 17 00:00:00 2001
From: Joannis Orlandos <joannis@orlandos.nl>
Date: Sat, 8 Jul 2023 13:41:46 +0200
Subject: [PATCH] Extract the NWConnection code into
 StateManagedNWConnectionChannel (#174)

* Extract the NWConnection code into StateManagedNWConnectionChannel

* Process cory's feedback

* Remove dead code whose functionality is already handled by the helper protocol

* Make the address cache non-locked, because its lock has been moved to the accessors

* Move getting the Multipath option into the TCP channel, re-introduce the underscores to the addressCache properties

* Drop the umbrella import

---------

Co-authored-by: Cory Benfield <lukasa@apple.com>
---
 .../NIOTSConnectionBootstrap.swift            |   2 +-
 .../NIOTSConnectionChannel.swift              | 550 ++-------------
 .../NIOTSListenerChannel.swift                |  15 +-
 .../StateManagedChannel.swift                 |   4 +-
 .../StateManagedNWConnectionChannel.swift     | 625 ++++++++++++++++++
 .../TCPOptions+SocketChannelOption.swift      |   2 +-
 6 files changed, 691 insertions(+), 507 deletions(-)
 create mode 100644 Sources/NIOTransportServices/StateManagedNWConnectionChannel.swift

diff --git a/Sources/NIOTransportServices/NIOTSConnectionBootstrap.swift b/Sources/NIOTransportServices/NIOTSConnectionBootstrap.swift
index c653635..c2a159c 100644
--- a/Sources/NIOTransportServices/NIOTSConnectionBootstrap.swift
+++ b/Sources/NIOTransportServices/NIOTSConnectionBootstrap.swift
@@ -199,7 +199,7 @@ public final class NIOTSConnectionBootstrap {
 
     /// Specify the `endpoint` to connect to for the TCP `Channel` that will be established.
     public func connect(endpoint: NWEndpoint) -> EventLoopFuture<Channel> {
-        return self.connect(shouldRegister: true) {channel, promise in
+        return self.connect(shouldRegister: true) { channel, promise in
             channel.triggerUserOutboundEvent(NIOTSNetworkEvents.ConnectToNWEndpoint(endpoint: endpoint),
                                              promise: promise)
         }
diff --git a/Sources/NIOTransportServices/NIOTSConnectionChannel.swift b/Sources/NIOTransportServices/NIOTSConnectionChannel.swift
index 9e0a988..e3f0b55 100644
--- a/Sources/NIOTransportServices/NIOTSConnectionChannel.swift
+++ b/Sources/NIOTransportServices/NIOTSConnectionChannel.swift
@@ -25,7 +25,7 @@ import Atomics
 
 /// Channel options for the connection channel.
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
-private struct ConnectionChannelOptions {
+struct TransportServicesChannelOptions {
     /// Whether autoRead is enabled for this channel.
     internal var autoRead: Bool = true
 
@@ -37,10 +37,6 @@ private struct ConnectionChannelOptions {
     internal var waitForActivity: Bool = true
 }
 
-
-private typealias PendingWrite = (data: ByteBuffer, promise: EventLoopPromise<Void>?)
-
-
 internal struct AddressCache {
     // deliberately lets because they must always be updated together (so forcing `init` is useful).
     let local: Optional<SocketAddress>
@@ -55,7 +51,7 @@ internal struct AddressCache {
 
 /// A structure that manages backpressure signaling on this channel.
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
-private struct BackpressureManager {
+internal struct BackpressureManager {
     /// Whether the channel is writable, given the current watermark state.
     ///
     /// This is an atomic only because the channel writability flag needs to be safe to access from multiple
@@ -131,7 +127,7 @@ private struct BackpressureManager {
 
 
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
-internal final class NIOTSConnectionChannel {
+internal final class NIOTSConnectionChannel: StateManagedNWConnectionChannel {
     /// The `ByteBufferAllocator` for this `Channel`.
     public let allocator = ByteBufferAllocator()
 
@@ -146,23 +142,29 @@ internal final class NIOTSConnectionChannel {
     /// The `EventLoop` this `Channel` belongs to.
     internal let tsEventLoop: NIOTSEventLoop
 
-    private var _pipeline: ChannelPipeline! = nil  // this is really a constant (set in .init) but needs `self` to be constructed and therefore a `var`. Do not change as this needs to accessed from arbitrary threads.
+    private(set) var _pipeline: ChannelPipeline! = nil  // this is really a constant (set in .init) but needs `self` to be constructed and therefore a `var`. Do not change as this needs to accessed from arbitrary threads.
 
     internal let closePromise: EventLoopPromise<Void>
 
     /// The underlying `NWConnection` that this `Channel` wraps. This is only non-nil
     /// after the initial connection attempt has been made.
-    private var nwConnection: NWConnection?
+    internal var connection: NWConnection?
 
     /// The `DispatchQueue` that socket events for this connection will be dispatched onto.
-    private let connectionQueue: DispatchQueue
+    internal let connectionQueue: DispatchQueue
 
     /// An `EventLoopPromise` that will be succeeded or failed when a connection attempt succeeds or fails.
-    private var connectPromise: EventLoopPromise<Void>?
+    internal var connectPromise: EventLoopPromise<Void>?
+
+    internal var parameters: NWParameters {
+        NWParameters(tls: self.tlsOptions, tcp: self.tcpOptions)
+    }
 
     /// The TCP options for this connection.
     private var tcpOptions: NWProtocolTCP.Options
 
+    internal var nwOptions: NWProtocolTCP.Options { self.tcpOptions }
+
     /// The TLS options for this connection, if any.
     private var tlsOptions: NWProtocolTLS.Options?
 
@@ -177,37 +179,37 @@ internal final class NIOTSConnectionChannel {
 
     /// Whether a call to NWConnection.receive has been made, but the completion
     /// handler has not yet been invoked.
-    private var outstandingRead: Bool = false
+    internal var outstandingRead: Bool = false
 
     /// The options for this channel.
-    private var options: ConnectionChannelOptions = ConnectionChannelOptions()
+    internal var options = TransportServicesChannelOptions()
 
     /// Any pending writes that have yet to be delivered to the network stack.
-    private var pendingWrites = CircularBuffer<PendingWrite>(initialCapacity: 8)
+    internal var pendingWrites = CircularBuffer<PendingWrite>(initialCapacity: 8)
 
     /// An object to keep track of pending writes and manage our backpressure signaling.
-    private var backpressureManager = BackpressureManager()
+    internal var _backpressureManager = BackpressureManager()
 
     /// The value of SO_REUSEADDR.
-    private var reuseAddress = false
+    internal var reuseAddress = false
 
     /// The value of SO_REUSEPORT.
-    private var reusePort = false
+    internal var reusePort = false
     
     /// The value of the allowLocalEndpointReuse option.
-    private var allowLocalEndpointReuse = false
+    internal var allowLocalEndpointReuse = false
 
     /// Whether to use peer-to-peer connectivity when connecting to Bonjour services.
-    private var enablePeerToPeer = false
+    internal var enablePeerToPeer = false
 
     /// The default multipath service type.
-    private var multipathServiceType = NWParameters.MultipathServiceType.disabled
+    internal var multipathServiceType = NWParameters.MultipathServiceType.disabled
 
     /// The cache of the local and remote socket addresses. Must be accessed using _addressCacheLock.
-    private var _addressCache = AddressCache(local: nil, remote: nil)
+    internal var _addressCache = AddressCache(local: nil, remote: nil)
 
     /// A lock that guards the _addressCache.
-    private let _addressCacheLock = NIOLock()
+    internal let _addressCacheLock = NIOLock()
 
     /// Create a `NIOTSConnectionChannel` on a given `NIOTSEventLoop`.
     ///
@@ -240,7 +242,7 @@ internal final class NIOTSConnectionChannel {
                   qos: qos,
                   tcpOptions: tcpOptions,
                   tlsOptions: tlsOptions)
-        self.nwConnection = connection
+        self.connection = connection
     }
 }
 
@@ -248,99 +250,16 @@ internal final class NIOTSConnectionChannel {
 // MARK:- NIOTSConnectionChannel implementation of Channel
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
 extension NIOTSConnectionChannel: Channel {
-    /// The `ChannelPipeline` for this `Channel`.
-    public var pipeline: ChannelPipeline {
-        return self._pipeline
-    }
-
-    /// The local address for this channel.
-    public var localAddress: SocketAddress? {
-        return self._addressCacheLock.withLock {
-            return self._addressCache.local
-        }
-    }
-
-    /// The remote address for this channel.
-    public var remoteAddress: SocketAddress? {
-        return self._addressCacheLock.withLock {
-            return self._addressCache.remote
-        }
-    }
-
-    /// Whether this channel is currently writable.
-    public var isWritable: Bool {
-        return self.backpressureManager.writable.load(ordering: .relaxed)
-    }
-
-    public var _channelCore: ChannelCore {
-        return self
-    }
-
-    public func setOption<Option: ChannelOption>(_ option: Option, value: Option.Value) -> EventLoopFuture<Void> {
-        if self.eventLoop.inEventLoop {
-            return self.eventLoop.makeCompletedFuture(Result { try setOption0(option: option, value: value) })
-        } else {
-            return self.eventLoop.submit { try self.setOption0(option: option, value: value) }
-        }
-    }
-
-    private func setOption0<Option: ChannelOption>(option: Option, value: Option.Value) throws {
-        self.eventLoop.preconditionInEventLoop()
-
-        guard !self.closed else {
-            throw ChannelError.ioOnClosedChannel
-        }
-
+    func getChannelSpecificOption0<Option>(option: Option) throws -> Option.Value where Option : ChannelOption {
         switch option {
-        case _ as ChannelOptions.Types.AutoReadOption:
-            self.options.autoRead = value as! Bool
-            self.readIfNeeded0()
-        case _ as ChannelOptions.Types.AllowRemoteHalfClosureOption:
-            self.options.supportRemoteHalfClosure = value as! Bool
-        case _ as ChannelOptions.Types.SocketOption:
-            let optionValue = option as! ChannelOptions.Types.SocketOption
-
-            // SO_REUSEADDR and SO_REUSEPORT are handled here.
-            switch (optionValue.level, optionValue.name) {
-            case (SOL_SOCKET, SO_REUSEADDR):
-                self.reuseAddress = (value as! SocketOptionValue) != Int32(0)
-            case (SOL_SOCKET, SO_REUSEPORT):
-                self.reusePort = (value as! SocketOptionValue) != Int32(0)
-            default:
-                try self.tcpOptions.applyChannelOption(option: optionValue, value: value as! SocketOptionValue)
-            }
-        case _ as ChannelOptions.Types.WriteBufferWaterMarkOption:
-            if self.backpressureManager.writabilityChanges(whenUpdatingWaterMarks: value as! ChannelOptions.Types.WriteBufferWaterMark) {
-                self.pipeline.fireChannelWritabilityChanged()
-            }
-        case _ as NIOTSChannelOptions.Types.NIOTSWaitForActivityOption:
-            let newValue = value as! Bool
-            self.options.waitForActivity = newValue
-
-            if let state = self.nwConnection?.state, case .waiting(let err) = state, !newValue {
-                // We're in waiting now, so we should drop the connection.
-                self.close0(error: err, mode: .all, promise: nil)
-            }
-        case is NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption:
-            self.enablePeerToPeer = value as! NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption.Value
-        case is NIOTSChannelOptions.Types.NIOTSAllowLocalEndpointReuse:
-            self.allowLocalEndpointReuse = value as! NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption.Value
         case is NIOTSChannelOptions.Types.NIOTSMultipathOption:
-            self.multipathServiceType = value as! NIOTSChannelOptions.Types.NIOTSMultipathOption.Value
+            return self.multipathServiceType as! Option.Value
         default:
             fatalError("option \(type(of: option)).\(option) not supported")
         }
     }
 
-    public func getOption<Option: ChannelOption>(_ option: Option) -> EventLoopFuture<Option.Value> {
-        if self.eventLoop.inEventLoop {
-            return self.eventLoop.makeCompletedFuture(Result { try getOption0(option: option) })
-        } else {
-            return eventLoop.submit { try self.getOption0(option: option) }
-        }
-    }
-
-    func getOption0<Option: ChannelOption>(option: Option) throws -> Option.Value {
+    func setChannelSpecificOption0<Option: ChannelOption>(option: Option, value: Option.Value) throws {
         self.eventLoop.preconditionInEventLoop()
 
         guard !self.closed else {
@@ -348,66 +267,9 @@ extension NIOTSConnectionChannel: Channel {
         }
 
         switch option {
-        case _ as ChannelOptions.Types.AutoReadOption:
-            return self.options.autoRead as! Option.Value
-        case _ as ChannelOptions.Types.AllowRemoteHalfClosureOption:
-            return self.options.supportRemoteHalfClosure as! Option.Value
-        case _ as ChannelOptions.Types.SocketOption:
-            let optionValue = option as! ChannelOptions.Types.SocketOption
-
-            // SO_REUSEADDR and SO_REUSEPORT are handled here.
-            switch (optionValue.level, optionValue.name) {
-            case (SOL_SOCKET, SO_REUSEADDR):
-                return Int32(self.reuseAddress ? 1 : 0) as! Option.Value
-            case (SOL_SOCKET, SO_REUSEPORT):
-                return Int32(self.reusePort ? 1 : 0) as! Option.Value
-            default:
-                return try self.tcpOptions.valueFor(socketOption: optionValue) as! Option.Value
-            }
-        case _ as ChannelOptions.Types.WriteBufferWaterMarkOption:
-            return self.backpressureManager.waterMarks as! Option.Value
-        case _ as NIOTSChannelOptions.Types.NIOTSWaitForActivityOption:
-            return self.options.waitForActivity as! Option.Value
-        case is NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption:
-            return self.enablePeerToPeer as! Option.Value
-        case is NIOTSChannelOptions.Types.NIOTSAllowLocalEndpointReuse:
-            return self.allowLocalEndpointReuse as! Option.Value
-        case is NIOTSChannelOptions.Types.NIOTSCurrentPathOption:
-            guard let currentPath = self.nwConnection?.currentPath else {
-                throw NIOTSErrors.NoCurrentPath()
-            }
-            return currentPath as! Option.Value
-        case is NIOTSChannelOptions.Types.NIOTSMetadataOption:
-            let optionValue = option as! NIOTSChannelOptions.Types.NIOTSMetadataOption
-            guard let nwConnection = self.nwConnection else {
-                throw NIOTSErrors.NoCurrentConnection()
-            }
-            return nwConnection.metadata(definition: optionValue.definition) as! Option.Value
         case is NIOTSChannelOptions.Types.NIOTSMultipathOption:
-            return self.multipathServiceType as! Option.Value
+            self.multipathServiceType = value as! NIOTSChannelOptions.Types.NIOTSMultipathOption.Value
         default:
-            // watchOS 6.0 availability is covered by the @available on this extension.
-            if #available(OSX 10.15, iOS 13.0, tvOS 13.0, *) {
-                switch option {
-                case is NIOTSChannelOptions.Types.NIOTSEstablishmentReportOption:
-                    guard let nwConnection = self.nwConnection else {
-                        throw NIOTSErrors.NoCurrentConnection()
-                    }
-                    let promise: EventLoopPromise<NWConnection.EstablishmentReport?> = eventLoop.makePromise()
-                    nwConnection.requestEstablishmentReport(queue: connectionQueue) { report in
-                        promise.succeed(report)
-                    }
-                    return promise.futureResult as! Option.Value
-                case is NIOTSChannelOptions.Types.NIOTSDataTransferReportOption:
-                    guard let nwConnection = self.nwConnection else {
-                        throw NIOTSErrors.NoCurrentConnection()
-                    }
-                    return nwConnection.startDataTransferReport() as! Option.Value
-                default:
-                    break
-                }
-            }
-            
             fatalError("option \(type(of: option)).\(option) not supported")
         }
     }
@@ -424,7 +286,7 @@ extension NIOTSConnectionChannel: StateManagedChannel {
     /// data.
     ///
     /// We keep track of this to manage the half-closure state of the TCP connection.
-    enum TCPSubstate: ActiveChannelSubstate {
+    enum TCPSubstate: NWConnectionSubstate {
         /// Both peers may send.
         case open
 
@@ -442,280 +304,39 @@ extension NIOTSConnectionChannel: StateManagedChannel {
         init() {
             self = .open
         }
-    }
 
-    public func localAddress0() throws -> SocketAddress {
-        guard let localEndpoint = self.nwConnection?.currentPath?.localEndpoint else {
-            throw NIOTSErrors.NoCurrentPath()
-        }
-        // TODO: Support wider range of address types.
-        return try SocketAddress(fromNWEndpoint: localEndpoint)
-    }
-
-    public func remoteAddress0() throws -> SocketAddress {
-        guard let remoteEndpoint = self.nwConnection?.currentPath?.remoteEndpoint else {
-            throw NIOTSErrors.NoCurrentPath()
-        }
-        // TODO: Support wider range of address types.
-        return try SocketAddress(fromNWEndpoint: remoteEndpoint)
-    }
-
-    internal func alreadyConfigured0(promise: EventLoopPromise<Void>?) {
-        guard let connection = nwConnection else {
-            promise?.fail(NIOTSErrors.NotPreConfigured())
-            return
-        }
-
-        guard case .setup = connection.state else {
-            promise?.fail(NIOTSErrors.NotPreConfigured())
-            return
-        }
-        self.connectPromise = promise
-        connection.stateUpdateHandler = self.stateUpdateHandler(newState:)
-        connection.betterPathUpdateHandler = self.betterPathHandler
-        connection.pathUpdateHandler = self.pathChangedHandler(newPath:)
-        connection.start(queue: self.connectionQueue)
-    }
-
-    internal func beginActivating0(to target: NWEndpoint, promise: EventLoopPromise<Void>?) {
-        assert(self.nwConnection == nil)
-        assert(self.connectPromise == nil)
-
-        // Before we start, we validate that the target won't cause a crash: see
-        // https://github.com/apple/swift-nio/issues/1617.
-        if case .hostPort(host: let host, port: _) = target, host == "" {
-            // We don't pass the promise in here because we'll actually not complete it. We complete it manually ourselves.
-            self.close0(error: NIOTSErrors.InvalidHostname(), mode: .all, promise: nil)
-            promise?.fail(NIOTSErrors.InvalidHostname())
-            return
-        }
-
-        self.connectPromise = promise
-
-        let parameters = NWParameters(tls: self.tlsOptions, tcp: self.tcpOptions)
-
-        // Network.framework munges REUSEADDR and REUSEPORT together, so we turn this on if we need
-        // either or it's been explicitly set.
-        parameters.allowLocalEndpointReuse = self.reuseAddress || self.reusePort || self.allowLocalEndpointReuse
-
-        parameters.includePeerToPeer = self.enablePeerToPeer
-
-        parameters.multipathServiceType = self.multipathServiceType
-
-        let connection = NWConnection(to: target, using: parameters)
-        connection.stateUpdateHandler = self.stateUpdateHandler(newState:)
-        connection.betterPathUpdateHandler = self.betterPathHandler
-        connection.pathUpdateHandler = self.pathChangedHandler(newPath:)
-
-        // Ok, state is ready. Let's go!
-        self.nwConnection = connection
-        connection.start(queue: self.connectionQueue)
-    }
-
-    public func write0(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
-        guard self.isActive else {
-            promise?.fail(ChannelError.ioOnClosedChannel)
-            return
-        }
-
-        // TODO: We would ideally support all of IOData here, gotta work out how to do that without HOL blocking
-        // all writes terribly.
-        // My best guess at this time is that Data(contentsOf:) may mmap the file in question, which would let us
-        // at least only block the network stack itself rather than our thread. I'm not certain though, especially
-        // on Linux. Should investigate.
-        let data = self.unwrapData(data, as: ByteBuffer.self)
-        self.pendingWrites.append((data, promise))
-
-
-        /// This may cause our writability state to change.
-        if self.backpressureManager.writabilityChanges(whenQueueingBytes: data.readableBytes) {
-            self.pipeline.fireChannelWritabilityChanged()
-        }
-    }
-
-    public func flush0() {
-        guard self.isActive else {
-            return
-        }
-
-        guard let conn = self.nwConnection else {
-            preconditionFailure("nwconnection cannot be nil while channel is active")
-        }
-
-        func completionCallback(promise: EventLoopPromise<Void>?, sentBytes: Int) -> ((NWError?) -> Void) {
-            return { error in
-                if let error = error {
-                    promise?.fail(error)
-                } else {
-                    promise?.succeed(())
-                }
-
-                if self.backpressureManager.writabilityChanges(whenBytesSent: sentBytes) {
-                    self.pipeline.fireChannelWritabilityChanged()
-                }
+        static func closeInput(state: inout ChannelState<NIOTSConnectionChannel.TCPSubstate>) throws {
+            switch state {
+            case .active(.open):
+                state = .active(.halfClosedRemote)
+            case .active(.halfClosedLocal):
+                state = .active(.closed)
+            case .idle, .registered, .activating, .active(.halfClosedRemote), .active(.closed), .inactive:
+                throw NIOTSErrors.InvalidChannelStateTransition()
             }
         }
 
-        conn.batch {
-            while self.pendingWrites.count > 0 {
-                let write = self.pendingWrites.removeFirst()
-                let buffer = write.data
-                let content = buffer.getData(at: buffer.readerIndex, length: buffer.readableBytes)
-                conn.send(content: content, completion: .contentProcessed(completionCallback(promise: write.promise, sentBytes: buffer.readableBytes)))
+        static func closeOutput(state: inout ChannelState<NIOTSConnectionChannel.TCPSubstate>) throws {
+            switch state {
+            case .active(.open):
+                state = .active(.halfClosedLocal)
+            case .active(.halfClosedRemote):
+                state = .active(.closed)
+            case .active(.halfClosedLocal), .active(.closed):
+                // This is a special case for closing the output, as it's user-controlled. If they already
+                // closed it, we want to throw a special error to tell them.
+                throw ChannelError.outputClosed
+            case .idle, .registered, .activating, .inactive:
+                throw NIOTSErrors.InvalidChannelStateTransition()
             }
         }
     }
-
-    /// Perform a read from the network.
-    ///
-    /// This method has a slightly strange semantic, because we do not allow multiple reads at once. As a result, this
-    /// is a *request* to read, and if there is a read already being processed then this method will do nothing.
-    public func read0() {
-        guard self.inboundStreamOpen && !self.outstandingRead else {
-            return
-        }
-
-        guard let conn = self.nwConnection else {
-            preconditionFailure("Connection should not be nil")
-        }
-
-        // TODO: Can we do something sensible with these numbers?
-        self.outstandingRead = true
-        conn.receive(minimumIncompleteLength: 1, maximumLength: 8192, completion: self.dataReceivedHandler(content:context:isComplete:error:))
-    }
-
-    public func doClose0(error: Error) {
-        guard let conn = self.nwConnection else {
-            // We don't have a connection to close here, so we're actually done. Our old state
-            // was idle.
-            assert(self.pendingWrites.count == 0)
-            return
-        }
-
-        // Step 1 is to tell the network stack we're done.
-        // TODO: Does this drop the connection fully, or can we keep receiving data? Must investigate.
-        conn.cancel()
-
-        // Step 2 is to fail all outstanding writes.
-        self.dropOutstandingWrites(error: error)
-
-        // Step 3 is to cancel a pending connect promise, if any.
-        if let pendingConnect = self.connectPromise {
-            self.connectPromise = nil
-            pendingConnect.fail(error)
-        }
-    }
-
-    public func doHalfClose0(error: Error, promise: EventLoopPromise<Void>?) {
-        guard let conn = self.nwConnection else {
-            // We don't have a connection to half close, so fail the promise.
-            promise?.fail(ChannelError.ioOnClosedChannel)
-            return
-        }
-
-
-        do {
-            try self.state.closeOutput()
-        } catch ChannelError.outputClosed {
-            // Here we *only* fail the promise, no need to blow up the connection.
-            promise?.fail(ChannelError.outputClosed)
-            return
-        } catch {
-            // For any other error, this is fatal.
-            self.close0(error: error, mode: .all, promise: promise)
-            return
-        }
-
-        func completionCallback(for promise: EventLoopPromise<Void>?) -> ((NWError?) -> Void) {
-            return { error in
-                if let error = error {
-                    promise?.fail(error)
-                } else {
-                    promise?.succeed(())
-                }
-            }
-        }
-
-        // It should not be possible to have a pending connect promise while we're doing half-closure.
-        assert(self.connectPromise == nil)
-
-        // Step 1 is to tell the network stack we're done.
-        conn.send(content: nil, contentContext: .finalMessage, completion: .contentProcessed(completionCallback(for: promise)))
-
-        // Step 2 is to fail all outstanding writes.
-        self.dropOutstandingWrites(error: error)
-    }
-
-    public func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
-        switch event {
-        case let x as NIOTSNetworkEvents.ConnectToNWEndpoint:
-            self.connect0(to: x.endpoint, promise: promise)
-        default:
-            promise?.fail(ChannelError.operationUnsupported)
-        }
-    }
-
-    public func channelRead0(_ data: NIOAny) {
-        // drop the data, do nothing
-        return
-    }
-
-    public func errorCaught0(error: Error) {
-        // Currently we don't do anything with errors that pass through the pipeline
-        return
-    }
-
-    /// A function that will trigger a socket read if necessary.
-    internal func readIfNeeded0() {
-        if self.options.autoRead {
-            self.pipeline.read()
-        }
-    }
 }
 
 
 // MARK:- Implementations of the callbacks passed to NWConnection.
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
 extension NIOTSConnectionChannel {
-    /// Called by the underlying `NWConnection` when its internal state has changed.
-    private func stateUpdateHandler(newState: NWConnection.State) {
-        switch newState {
-        case .setup:
-            preconditionFailure("Should not be told about this state.")
-        case .waiting(let err):
-            if case .activating = self.state, self.options.waitForActivity {
-                // This means the connection cannot currently be completed. We should notify the pipeline
-                // here, or support this with a channel option or something, but for now for the sake of
-                // demos we will just allow ourselves into this stage.
-                self.pipeline.fireUserInboundEventTriggered(NIOTSNetworkEvents.WaitingForConnectivity(transientError: err))
-                break
-            }
-
-            // In this state we've transitioned into waiting, presumably from active or closing. In this
-            // version of NIO this is an error, but we should aim to support this at some stage.
-            self.close0(error: err, mode: .all, promise: nil)
-        case .preparing:
-            // This just means connections are being actively established. We have no specific action
-            // here.
-            break
-        case .ready:
-            // Transitioning to ready means the connection was succeeded. Hooray!
-            self.connectionComplete0()
-        case .cancelled:
-            // This is the network telling us we're closed. We don't need to actually do anything here
-            // other than check our state is ok.
-            assert(self.closed)
-            self.nwConnection = nil
-        case .failed(let err):
-            // The connection has failed for some reason.
-            self.close0(error: err, mode: .all, promise: nil)
-        default:
-            // This clause is here to help the compiler out: it's otherwise not able to
-            // actually validate that the switch is exhaustive. Trust me, it is.
-            fatalError("Unreachable")
-        }
-    }
-
     /// Called by the underlying `NWConnection` when a network receive has completed.
     ///
     /// The state matrix here is large. If `content` is non-nil, some data was received: we need to send it down the pipeline
@@ -779,7 +400,7 @@ extension NIOTSConnectionChannel {
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
 extension NIOTSConnectionChannel {
     /// Whether the inbound side of the connection is still open.
-    private var inboundStreamOpen: Bool {
+    internal var _inboundStreamOpen: Bool {
         switch self.state {
         case .active(.open), .active(.halfClosedLocal):
             return true
@@ -788,37 +409,6 @@ extension NIOTSConnectionChannel {
         }
     }
 
-    /// Make the channel active.
-    private func connectionComplete0() {
-        let promise = self.connectPromise
-        self.connectPromise = nil
-
-        // Before becoming active, update the cached addresses.
-        let localAddress = try? self.localAddress0()
-        let remoteAddress = try? self.remoteAddress0()
-
-        self._addressCacheLock.withLock {
-            self._addressCache = AddressCache(local: localAddress, remote: remoteAddress)
-        }
-
-        self.becomeActive0(promise: promise)
-
-        if let metadata = self.nwConnection?.metadata(definition: NWProtocolTLS.definition) as? NWProtocolTLS.Metadata {
-            // This is a TLS connection, we may need to fire some other events.
-            let securityMetadata = metadata.securityProtocolMetadata
-
-            // The pointer returned by `sec_protocol_metadata_get_negotiated_protocol` is presumably owned by it, so we need
-            // to confirm it's still alive while we copy the data out.
-            let negotiatedProtocol = withExtendedLifetime(securityMetadata) {
-                sec_protocol_metadata_get_negotiated_protocol(metadata.securityProtocolMetadata).map {
-                    String(cString: $0)
-                }
-            }
-
-            self.pipeline.fireUserInboundEventTriggered(TLSUserEvent.handshakeCompleted(negotiatedProtocol: negotiatedProtocol))
-        }
-    }
-
     /// Drop all outstanding writes. Must only be called in the inactive
     /// state.
     private func dropOutstandingWrites(error: Error) {
@@ -835,7 +425,7 @@ extension NIOTSConnectionChannel {
         if self.options.supportRemoteHalfClosure {
             // This is a half-closure, but the connection is still valid.
             do {
-                try self.state.closeInput()
+                try ActiveSubstate.closeInput(state: &self.state)
             } catch {
                 return self.close0(error: error, mode: .all, promise: nil)
             }
@@ -848,38 +438,6 @@ extension NIOTSConnectionChannel {
 }
 
 
-// MARK:- Managing TCP substate.
-@available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
-fileprivate extension ChannelState where ActiveSubstate == NIOTSConnectionChannel.TCPSubstate {
-    /// Close the input side of the TCP state machine.
-    mutating func closeInput() throws {
-        switch self {
-        case .active(.open):
-            self = .active(.halfClosedRemote)
-        case .active(.halfClosedLocal):
-            self = .active(.closed)
-        case .idle, .registered, .activating, .active(.halfClosedRemote), .active(.closed), .inactive:
-            throw NIOTSErrors.InvalidChannelStateTransition()
-        }
-    }
-
-    /// Close the output side of the TCP state machine.
-    mutating func closeOutput() throws {
-        switch self {
-        case .active(.open):
-            self = .active(.halfClosedLocal)
-        case .active(.halfClosedRemote):
-            self = .active(.closed)
-        case .active(.halfClosedLocal), .active(.closed):
-            // This is a special case for closing the output, as it's user-controlled. If they already
-            // closed it, we want to throw a special error to tell them.
-            throw ChannelError.outputClosed
-        case .idle, .registered, .activating, .inactive:
-            throw NIOTSErrors.InvalidChannelStateTransition()
-        }
-    }
-}
-
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
 extension NIOTSConnectionChannel {
     internal struct SynchronousOptions: NIOSynchronousChannelOptions {
@@ -915,10 +473,10 @@ public struct NIOTSChannelIsNotANIOTSConnectionChannel: Error, Hashable {
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
 extension NIOTSConnectionChannel {
     fileprivate func metadata(definition: NWProtocolDefinition) throws -> NWProtocolMetadata? {
-        guard let nwConnection = self.nwConnection else {
+        guard let connection = self.connection else {
             throw NIOTSConnectionNotInitialized()
         }
-        return nwConnection.metadata(definition: definition)
+        return connection.metadata(definition: definition)
     }
 }
 
diff --git a/Sources/NIOTransportServices/NIOTSListenerChannel.swift b/Sources/NIOTransportServices/NIOTSListenerChannel.swift
index 0e131e4..f8595e3 100644
--- a/Sources/NIOTransportServices/NIOTSListenerChannel.swift
+++ b/Sources/NIOTransportServices/NIOTSListenerChannel.swift
@@ -101,7 +101,7 @@ internal final class NIOTSListenerChannel {
     private let childTLSOptions: NWProtocolTLS.Options?
 
     /// The cache of the local and remote socket addresses. Must be accessed using _addressCacheLock.
-    private var _addressCache = AddressCache(local: nil, remote: nil)
+    internal var _addressCache = AddressCache(local: nil, remote: nil)
 
     /// A lock that guards the _addressCache.
     private let _addressCacheLock = NIOLock()
@@ -480,12 +480,13 @@ extension NIOTSListenerChannel {
             return
         }
 
-        let newChannel = NIOTSConnectionChannel(wrapping: connection,
-                                                on: self.childLoopGroup.next() as! NIOTSEventLoop,
-                                                parent: self,
-                                                qos: self.childChannelQoS,
-                                                tcpOptions: self.childTCPOptions,
-                                                tlsOptions: self.childTLSOptions)
+        let newChannel = NIOTSConnectionChannel(
+            wrapping: connection,
+            on: self.childLoopGroup.next() as! NIOTSEventLoop,
+            parent: self,
+            qos: self.childChannelQoS,
+            tcpOptions: self.childTCPOptions,
+            tlsOptions: self.childTLSOptions)
 
         self.pipeline.fireChannelRead(NIOAny(newChannel))
         self.pipeline.fireChannelReadComplete()
diff --git a/Sources/NIOTransportServices/StateManagedChannel.swift b/Sources/NIOTransportServices/StateManagedChannel.swift
index 72a29c2..aa86890 100644
--- a/Sources/NIOTransportServices/StateManagedChannel.swift
+++ b/Sources/NIOTransportServices/StateManagedChannel.swift
@@ -45,7 +45,7 @@ internal enum ChannelState<ActiveSubstate: ActiveChannelSubstate> {
     /// Unlike every other one of these methods, this one has a side-effect. This is because
     /// it's impossible to correctly be in the reigstered state without verifying that
     /// registration has occurred.
-    fileprivate mutating func register(eventLoop: NIOTSEventLoop, channel: Channel) throws {
+    mutating func register(eventLoop: NIOTSEventLoop, channel: Channel) throws {
         guard case .idle = self else {
             throw NIOTSErrors.InvalidChannelStateTransition()
         }
@@ -53,7 +53,7 @@ internal enum ChannelState<ActiveSubstate: ActiveChannelSubstate> {
         self = .registered
     }
 
-    fileprivate mutating func beginActivating() throws {
+    mutating func beginActivating() throws {
         guard case .registered = self else {
             throw NIOTSErrors.InvalidChannelStateTransition()
         }
diff --git a/Sources/NIOTransportServices/StateManagedNWConnectionChannel.swift b/Sources/NIOTransportServices/StateManagedNWConnectionChannel.swift
new file mode 100644
index 0000000..30026d5
--- /dev/null
+++ b/Sources/NIOTransportServices/StateManagedNWConnectionChannel.swift
@@ -0,0 +1,625 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the SwiftNIO open source project
+//
+// Copyright (c) 2020-2023 Apple Inc. and the SwiftNIO project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of SwiftNIO project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+#if canImport(Network)
+import Foundation
+import NIOCore
+import NIOConcurrencyHelpers
+import NIOFoundationCompat
+import NIOTLS
+import Dispatch
+import Network
+import Security
+
+@available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
+protocol NWConnectionSubstate: ActiveChannelSubstate {
+    static func closeInput(state: inout ChannelState<Self>) throws
+    static func closeOutput(state: inout ChannelState<Self>) throws
+}
+
+internal typealias PendingWrite = (data: ByteBuffer, promise: EventLoopPromise<Void>?)
+
+protocol NWOptionsProtocol {
+    /// Apply a given channel `SocketOption` to this protocol options state.
+    func applyChannelOption(option: ChannelOptions.Types.SocketOption, value: SocketOptionValue) throws
+
+    /// Obtain the given `SocketOption` value for this protocol options state.
+    func valueFor(socketOption option: ChannelOptions.Types.SocketOption) throws -> SocketOptionValue
+}
+
+@available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
+internal protocol StateManagedNWConnectionChannel: StateManagedChannel where ActiveSubstate: NWConnectionSubstate {
+    associatedtype NWOptions: NWOptionsProtocol
+
+    var parameters: NWParameters { get }
+
+    var nwOptions: NWOptions { get }
+    
+    var connection: NWConnection? { get set }
+
+    var connectionQueue: DispatchQueue { get }
+
+    var connectPromise: EventLoopPromise<Void>? { get set }
+
+    var outstandingRead: Bool { get set }
+
+    var options: TransportServicesChannelOptions { get set }
+
+    var pendingWrites: CircularBuffer<PendingWrite> { get set }
+
+    var _backpressureManager: BackpressureManager { get set }
+
+    var reuseAddress: Bool { get set }
+
+    var reusePort: Bool { get set }
+
+    var enablePeerToPeer: Bool { get set }
+    
+    var _inboundStreamOpen: Bool { get }
+
+    var _pipeline: ChannelPipeline! { get }
+
+    var _addressCache: AddressCache { get set }
+
+    var _addressCacheLock: NIOLock { get }
+
+    var allowLocalEndpointReuse: Bool { get set }
+
+    var multipathServiceType: NWParameters.MultipathServiceType { get }
+
+    func setChannelSpecificOption0<Option: ChannelOption>(option: Option, value: Option.Value) throws
+
+    func getChannelSpecificOption0<Option: ChannelOption>(option: Option) throws -> Option.Value
+}
+
+@available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
+extension StateManagedNWConnectionChannel {
+    public var pipeline: ChannelPipeline {
+        self._pipeline
+    }
+
+    public var _channelCore: ChannelCore {
+        return self
+    }
+
+    /// The local address for this channel.
+    public var localAddress: SocketAddress? {
+        return self._addressCacheLock.withLock {
+            return self._addressCache.local
+        }
+    }
+
+    /// The remote address for this channel.
+    public var remoteAddress: SocketAddress? {
+        return self._addressCacheLock.withLock {
+            return self._addressCache.remote
+        }
+    }
+
+    /// Whether this channel is currently writable.
+    public var isWritable: Bool {
+        return self._backpressureManager.writable.load(ordering: .relaxed)
+    }
+
+    internal func beginActivating0(to target: NWEndpoint, promise: EventLoopPromise<Void>?) {
+        assert(self.connection == nil)
+        assert(self.connectPromise == nil)
+
+        // Before we start, we validate that the target won't cause a crash: see
+        // https://github.com/apple/swift-nio/issues/1617.
+        if case .hostPort(host: let host, port: _) = target, host == "" {
+            // We don't pass the promise in here because we'll actually not complete it. We complete it manually ourselves.
+            self.close0(error: NIOTSErrors.InvalidHostname(), mode: .all, promise: nil)
+            promise?.fail(NIOTSErrors.InvalidHostname())
+            return
+        }
+
+        self.connectPromise = promise
+
+        let parameters = parameters
+
+        // Network.framework munges REUSEADDR and REUSEPORT together, so we turn this on if we need
+        // either or it's been explicitly set.
+        parameters.allowLocalEndpointReuse = self.reuseAddress || self.reusePort || self.allowLocalEndpointReuse
+
+        parameters.includePeerToPeer = self.enablePeerToPeer
+
+        parameters.multipathServiceType = self.multipathServiceType
+
+        let connection = NWConnection(to: target, using: parameters)
+        connection.stateUpdateHandler = self.stateUpdateHandler(newState:)
+        connection.betterPathUpdateHandler = self.betterPathHandler
+        connection.pathUpdateHandler = self.pathChangedHandler(newPath:)
+
+        // Ok, state is ready. Let's go!
+        self.connection = connection
+        connection.start(queue: self.connectionQueue)
+    }
+
+    public func write0(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
+        guard self.isActive else {
+            promise?.fail(ChannelError.ioOnClosedChannel)
+            return
+        }
+
+        // TODO: We would ideally support all of IOData here, gotta work out how to do that without HOL blocking
+        // all writes terribly.
+        // My best guess at this time is that Data(contentsOf:) may mmap the file in question, which would let us
+        // at least only block the network stack itself rather than our thread. I'm not certain though, especially
+        // on Linux. Should investigate.
+        let data = self.unwrapData(data, as: ByteBuffer.self)
+        self.pendingWrites.append((data, promise))
+
+
+        /// This may cause our writability state to change.
+        if self._backpressureManager.writabilityChanges(whenQueueingBytes: data.readableBytes) {
+            self.pipeline.fireChannelWritabilityChanged()
+        }
+    }
+
+    public func flush0() {
+        guard self.isActive else {
+            return
+        }
+
+        guard let conn = self.connection else {
+            preconditionFailure("nwconnection cannot be nil while channel is active")
+        }
+
+        func completionCallback(promise: EventLoopPromise<Void>?, sentBytes: Int) -> ((NWError?) -> Void) {
+            return { error in
+                if let error = error {
+                    promise?.fail(error)
+                } else {
+                    promise?.succeed(())
+                }
+
+                if self._backpressureManager.writabilityChanges(whenBytesSent: sentBytes) {
+                    self.pipeline.fireChannelWritabilityChanged()
+                }
+            }
+        }
+
+        conn.batch {
+            while self.pendingWrites.count > 0 {
+                let write = self.pendingWrites.removeFirst()
+                let buffer = write.data
+                let content = buffer.getData(at: buffer.readerIndex, length: buffer.readableBytes)
+                conn.send(content: content, completion: .contentProcessed(completionCallback(promise: write.promise, sentBytes: buffer.readableBytes)))
+            }
+        }
+    }
+    
+    public func localAddress0() throws -> SocketAddress {
+        guard let localEndpoint = self.connection?.currentPath?.localEndpoint else {
+            throw NIOTSErrors.NoCurrentPath()
+        }
+        // TODO: Support wider range of address types.
+        return try SocketAddress(fromNWEndpoint: localEndpoint)
+    }
+
+    public func remoteAddress0() throws -> SocketAddress {
+        guard let remoteEndpoint = self.connection?.currentPath?.remoteEndpoint else {
+            throw NIOTSErrors.NoCurrentPath()
+        }
+        // TODO: Support wider range of address types.
+        return try SocketAddress(fromNWEndpoint: remoteEndpoint)
+    }
+
+    internal func alreadyConfigured0(promise: EventLoopPromise<Void>?) {
+        guard let connection = connection else {
+            promise?.fail(NIOTSErrors.NotPreConfigured())
+            return
+        }
+
+        guard case .setup = connection.state else {
+            promise?.fail(NIOTSErrors.NotPreConfigured())
+            return
+        }
+        self.connectPromise = promise
+        connection.stateUpdateHandler = self.stateUpdateHandler(newState:)
+        connection.betterPathUpdateHandler = self.betterPathHandler
+        connection.pathUpdateHandler = self.pathChangedHandler(newPath:)
+        connection.start(queue: self.connectionQueue)
+    }
+
+    /// Perform a read from the network.
+    ///
+    /// This method has a slightly strange semantic, because we do not allow multiple reads at once. As a result, this
+    /// is a *request* to read, and if there is a read already being processed then this method will do nothing.
+    public func read0() {
+        guard self._inboundStreamOpen && !self.outstandingRead else {
+            return
+        }
+
+        guard let conn = self.connection else {
+            preconditionFailure("Connection should not be nil")
+        }
+
+        // TODO: Can we do something sensible with these numbers?
+        self.outstandingRead = true
+        conn.receive(minimumIncompleteLength: 1, maximumLength: 8192, completion: self.dataReceivedHandler(content:context:isComplete:error:))
+    }
+
+    public func doClose0(error: Error) {
+        guard let conn = self.connection else {
+            // We don't have a connection to close here, so we're actually done. Our old state
+            // was idle.
+            assert(self.pendingWrites.count == 0)
+            return
+        }
+
+        // Step 1 is to tell the network stack we're done.
+        // TODO: Does this drop the connection fully, or can we keep receiving data? Must investigate.
+        conn.cancel()
+
+        // Step 2 is to fail all outstanding writes.
+        self.dropOutstandingWrites(error: error)
+
+        // Step 3 is to cancel a pending connect promise, if any.
+        if let pendingConnect = self.connectPromise {
+            self.connectPromise = nil
+            pendingConnect.fail(error)
+        }
+    }
+
+    public func doHalfClose0(error: Error, promise: EventLoopPromise<Void>?) {
+        guard let conn = self.connection else {
+            // We don't have a connection to half close, so fail the promise.
+            promise?.fail(ChannelError.ioOnClosedChannel)
+            return
+        }
+
+        do {
+            try ActiveSubstate.closeOutput(state: &self.state)
+        } catch ChannelError.outputClosed {
+            // Here we *only* fail the promise, no need to blow up the connection.
+            promise?.fail(ChannelError.outputClosed)
+            return
+        } catch {
+            // For any other error, this is fatal.
+            self.close0(error: error, mode: .all, promise: promise)
+            return
+        }
+
+        func completionCallback(for promise: EventLoopPromise<Void>?) -> ((NWError?) -> Void) {
+            return { error in
+                if let error = error {
+                    promise?.fail(error)
+                } else {
+                    promise?.succeed(())
+                }
+            }
+        }
+
+        // It should not be possible to have a pending connect promise while we're doing half-closure.
+        assert(self.connectPromise == nil)
+
+        // Step 1 is to tell the network stack we're done.
+        conn.send(content: nil, contentContext: .finalMessage, completion: .contentProcessed(completionCallback(for: promise)))
+
+        // Step 2 is to fail all outstanding writes.
+        self.dropOutstandingWrites(error: error)
+    }
+
+    public func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
+        switch event {
+        case let x as NIOTSNetworkEvents.ConnectToNWEndpoint:
+            self.connect0(to: x.endpoint, promise: promise)
+        default:
+            promise?.fail(ChannelError.operationUnsupported)
+        }
+    }
+
+    public func channelRead0(_ data: NIOAny) {
+        // drop the data, do nothing
+        return
+    }
+
+    public func errorCaught0(error: Error) {
+        // Currently we don't do anything with errors that pass through the pipeline
+        return
+    }
+
+    /// A function that will trigger a socket read if necessary.
+    internal func readIfNeeded0() {
+        if self.options.autoRead {
+            self.pipeline.read()
+        }
+    }
+    
+    /// Called by the underlying `NWConnection` when its internal state has changed.
+    private func stateUpdateHandler(newState: NWConnection.State) {
+        switch newState {
+        case .setup:
+            preconditionFailure("Should not be told about this state.")
+        case .waiting(let err):
+            if case .activating = self.state, self.options.waitForActivity {
+                // This means the connection cannot currently be completed. We should notify the pipeline
+                // here, or support this with a channel option or something, but for now for the sake of
+                // demos we will just allow ourselves into this stage.tage.
+                self.pipeline.fireUserInboundEventTriggered(NIOTSNetworkEvents.WaitingForConnectivity(transientError: err))
+                break
+            }
+
+            // In this state we've transitioned into waiting, presumably from active or closing. In this
+            // version of NIO this is an error, but we should aim to support this at some stage.
+            self.close0(error: err, mode: .all, promise: nil)
+        case .preparing:
+            // This just means connections are being actively established. We have no specific action
+            // here.
+            break
+        case .ready:
+            // Transitioning to ready means the connection was succeeded. Hooray!
+            self.connectionComplete0()
+        case .cancelled:
+            // This is the network telling us we're closed. We don't need to actually do anything here
+            // other than check our state is ok.
+            assert(self.closed)
+            self.connection = nil
+        case .failed(let err):
+            // The connection has failed for some reason.
+            self.close0(error: err, mode: .all, promise: nil)
+        default:
+            // This clause is here to help the compiler out: it's otherwise not able to
+            // actually validate that the switch is exhaustive. Trust me, it is.
+            fatalError("Unreachable")
+        }
+    }
+
+    /// Called by the underlying `NWConnection` when a network receive has completed.
+    ///
+    /// The state matrix here is large. If `content` is non-nil, some data was received: we need to send it down the pipeline
+    /// and call channelReadComplete. This may be nil, in which case we expect either `isComplete` to be `true` or `error`
+    /// to be non-nil. `isComplete` indicates half-closure on the read side of a connection. `error` is set if the receive
+    /// did not complete due to an error, though there may still be some data.
+    private func dataReceivedHandler(content: Data?, context: NWConnection.ContentContext?, isComplete: Bool, error: NWError?) {
+        precondition(self.outstandingRead)
+        self.outstandingRead = false
+
+        guard self.isActive else {
+            // If we're already not active, we aren't going to process any of this: it's likely the result of an extra
+            // read somewhere along the line.
+            return
+        }
+
+        // First things first, if there's data we need to deliver it.
+        if let content = content {
+            // It would be nice if we didn't have to do this copy, but I'm not sure how to avoid it with the current Data
+            // APIs.
+            var buffer = self.allocator.buffer(capacity: content.count)
+            buffer.writeBytes(content)
+            self.pipeline.fireChannelRead(NIOAny(buffer))
+            self.pipeline.fireChannelReadComplete()
+        }
+
+        // Next, we want to check if there's an error. If there is, we're going to deliver it, and then close the connection with
+        // it. Otherwise, we're going to check if we read EOF, and if we did we'll close with that instead.
+        if let error = error {
+            self.pipeline.fireErrorCaught(error)
+            self.close0(error: error, mode: .all, promise: nil)
+        } else if isComplete {
+            self.didReadEOF()
+        }
+
+        // Last, issue a new read automatically if we need to.
+        self.readIfNeeded0()
+    }
+
+    /// Called by the underlying `NWConnection` when a better path for this connection is available.
+    ///
+    /// Notifies the channel pipeline of the new option.
+    private func betterPathHandler(available: Bool) {
+        if available {
+            self.pipeline.fireUserInboundEventTriggered(NIOTSNetworkEvents.BetterPathAvailable())
+        } else {
+            self.pipeline.fireUserInboundEventTriggered(NIOTSNetworkEvents.BetterPathUnavailable())
+        }
+    }
+
+    /// Called by the underlying `NWConnection` when this connection changes its network path.
+    ///
+    /// Notifies the channel pipeline of the new path.
+    private func pathChangedHandler(newPath path: NWPath) {
+        self.pipeline.fireUserInboundEventTriggered(NIOTSNetworkEvents.PathChanged(newPath: path))
+    }
+
+    /// Handle a read EOF.
+    ///
+    /// If the user has indicated they support half-closure, we will emit the standard half-closure
+    /// event. If they have not, we upgrade this to regular closure.
+    private func didReadEOF() {
+        if self.options.supportRemoteHalfClosure {
+            // This is a half-closure, but the connection is still valid.
+            do {
+                try ActiveSubstate.closeInput(state: &self.state)
+            } catch {
+                return self.close0(error: error, mode: .all, promise: nil)
+            }
+
+            self.pipeline.fireUserInboundEventTriggered(ChannelEvent.inputClosed)
+        } else {
+            self.close0(error: ChannelError.eof, mode: .all, promise: nil)
+        }
+    }
+    /// Make the channel active.
+    private func connectionComplete0() {
+        let promise = self.connectPromise
+        self.connectPromise = nil
+
+        // Before becoming active, update the cached addresses.
+        let localAddress = try? self.localAddress0()
+        let remoteAddress = try? self.remoteAddress0()
+
+        self._addressCache = AddressCache(local: localAddress, remote: remoteAddress)
+
+        self.becomeActive0(promise: promise)
+
+        if let metadata = self.connection?.metadata(definition: NWProtocolTLS.definition) as? NWProtocolTLS.Metadata {
+            // This is a TLS connection, we may need to fire some other events.
+            let securityMetadata = metadata.securityProtocolMetadata
+
+            // The pointer returned by `sec_protocol_metadata_get_negotiated_protocol` is presumably owned by it, so we need
+            // to confirm it's still alive while we copy the data out.
+            let negotiatedProtocol = withExtendedLifetime(securityMetadata) {
+                sec_protocol_metadata_get_negotiated_protocol(metadata.securityProtocolMetadata).map {
+                    String(cString: $0)
+                }
+            }
+
+            self.pipeline.fireUserInboundEventTriggered(TLSUserEvent.handshakeCompleted(negotiatedProtocol: negotiatedProtocol))
+        }
+    }
+
+    /// Drop all outstanding writes. Must only be called in the inactive
+    /// state.
+    private func dropOutstandingWrites(error: Error) {
+        while self.pendingWrites.count > 0 {
+            self.pendingWrites.removeFirst().promise?.fail(error)
+        }
+    }
+
+    public func setOption<Option: ChannelOption>(_ option: Option, value: Option.Value) -> EventLoopFuture<Void> {
+        if self.eventLoop.inEventLoop {
+            return self.eventLoop.makeCompletedFuture(Result { try setOption0(option: option, value: value) })
+        } else {
+            return self.eventLoop.submit { try self.setOption0(option: option, value: value) }
+        }
+    }
+
+    func setOption0<Option: ChannelOption>(option: Option, value: Option.Value) throws {
+        self.eventLoop.assertInEventLoop()
+
+        guard !self.closed else {
+            throw ChannelError.ioOnClosedChannel
+        }
+
+        switch option {
+        case _ as ChannelOptions.Types.AutoReadOption:
+            self.options.autoRead = value as! Bool
+            self.readIfNeeded0()
+        case _ as ChannelOptions.Types.SocketOption:
+            let optionValue = option as! ChannelOptions.Types.SocketOption
+
+            // SO_REUSEADDR and SO_REUSEPORT are handled here.
+            switch (optionValue.level, optionValue.name) {
+            case (SOL_SOCKET, SO_REUSEADDR):
+                self.reuseAddress = (value as! SocketOptionValue) != Int32(0)
+            case (SOL_SOCKET, SO_REUSEPORT):
+                self.reusePort = (value as! SocketOptionValue) != Int32(0)
+            default:
+                try self.nwOptions.applyChannelOption(option: optionValue, value: value as! SocketOptionValue)
+            }
+        case _ as ChannelOptions.Types.WriteBufferWaterMarkOption:
+            if self._backpressureManager.writabilityChanges(whenUpdatingWaterMarks: value as! ChannelOptions.Types.WriteBufferWaterMark) {
+                self.pipeline.fireChannelWritabilityChanged()
+            }
+        case is NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption:
+            self.enablePeerToPeer = value as! NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption.Value
+        case _ as NIOTSChannelOptions.Types.NIOTSWaitForActivityOption:
+            let newValue = value as! Bool
+            self.options.waitForActivity = newValue
+
+            if let state = self.connection?.state, case .waiting(let err) = state, !newValue {
+                // We're in waiting now, so we should drop the connection.
+                self.close0(error: err, mode: .all, promise: nil)
+            }
+        case _ as ChannelOptions.Types.AllowRemoteHalfClosureOption:
+            self.options.supportRemoteHalfClosure = value as! Bool
+        case is NIOTSChannelOptions.Types.NIOTSAllowLocalEndpointReuse:
+            self.allowLocalEndpointReuse = value as! NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption.Value
+        default:
+            try self.setChannelSpecificOption0(option: option, value: value)
+        }
+    }
+
+    public func getOption<Option: ChannelOption>(_ option: Option) -> EventLoopFuture<Option.Value> {
+        if self.eventLoop.inEventLoop {
+            return self.eventLoop.makeCompletedFuture(Result { try getOption0(option: option) })
+        } else {
+            return eventLoop.submit { try self.getOption0(option: option) }
+        }
+    }
+
+    func getOption0<Option: ChannelOption>(option: Option) throws -> Option.Value {
+        self.eventLoop.assertInEventLoop()
+
+        guard !self.closed else {
+            throw ChannelError.ioOnClosedChannel
+        }
+
+        switch option {
+        case _ as ChannelOptions.Types.AutoReadOption:
+            return self.options.autoRead as! Option.Value
+        case _ as ChannelOptions.Types.SocketOption:
+            let optionValue = option as! ChannelOptions.Types.SocketOption
+
+            // SO_REUSEADDR and SO_REUSEPORT are handled here.
+            switch (optionValue.level, optionValue.name) {
+            case (SOL_SOCKET, SO_REUSEADDR):
+                return Int32(self.reuseAddress ? 1 : 0) as! Option.Value
+            case (SOL_SOCKET, SO_REUSEPORT):
+                return Int32(self.reusePort ? 1 : 0) as! Option.Value
+            default:
+                return try self.nwOptions.valueFor(socketOption: optionValue) as! Option.Value
+            }
+        case _ as ChannelOptions.Types.WriteBufferWaterMarkOption:
+            return self._backpressureManager.waterMarks as! Option.Value
+        case is NIOTSChannelOptions.Types.NIOTSEnablePeerToPeerOption:
+            return self.enablePeerToPeer as! Option.Value
+        case is NIOTSChannelOptions.Types.NIOTSAllowLocalEndpointReuse:
+            return self.allowLocalEndpointReuse as! Option.Value
+        case _ as ChannelOptions.Types.AllowRemoteHalfClosureOption:
+            return self.options.supportRemoteHalfClosure as! Option.Value
+        case _ as NIOTSChannelOptions.Types.NIOTSWaitForActivityOption:
+            return self.options.waitForActivity as! Option.Value
+        case is NIOTSChannelOptions.Types.NIOTSCurrentPathOption:
+            guard let currentPath = self.connection?.currentPath else {
+                throw NIOTSErrors.NoCurrentPath()
+            }
+            return currentPath as! Option.Value
+        case is NIOTSChannelOptions.Types.NIOTSMetadataOption:
+            let optionValue = option as! NIOTSChannelOptions.Types.NIOTSMetadataOption
+            guard let connection = self.connection else {
+                throw NIOTSErrors.NoCurrentConnection()
+            }
+            return connection.metadata(definition: optionValue.definition) as! Option.Value
+        default:
+            // watchOS 6.0 availability is covered by the @available on this extension.
+            if #available(OSX 10.15, iOS 13.0, tvOS 13.0, *) {
+                switch option {
+                case is NIOTSChannelOptions.Types.NIOTSEstablishmentReportOption:
+                    guard let connection = self.connection else {
+                        throw NIOTSErrors.NoCurrentConnection()
+                    }
+                    let promise: EventLoopPromise<NWConnection.EstablishmentReport?> = eventLoop.makePromise()
+                    connection.requestEstablishmentReport(queue: connectionQueue) { report in
+                        promise.succeed(report)
+                    }
+                    return promise.futureResult as! Option.Value
+                case is NIOTSChannelOptions.Types.NIOTSDataTransferReportOption:
+                    guard let connection = self.connection else {
+                        throw NIOTSErrors.NoCurrentConnection()
+                    }
+                    return connection.startDataTransferReport() as! Option.Value
+                default:
+                    break
+                }
+            }
+
+            return try getChannelSpecificOption0(option: option)
+        }
+    }
+}
+#endif
diff --git a/Sources/NIOTransportServices/TCPOptions+SocketChannelOption.swift b/Sources/NIOTransportServices/TCPOptions+SocketChannelOption.swift
index 666b7cf..e00894e 100644
--- a/Sources/NIOTransportServices/TCPOptions+SocketChannelOption.swift
+++ b/Sources/NIOTransportServices/TCPOptions+SocketChannelOption.swift
@@ -18,7 +18,7 @@ import NIOCore
 import Network
 
 @available(OSX 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *)
-internal extension NWProtocolTCP.Options {
+extension NWProtocolTCP.Options: NWOptionsProtocol {
     /// Apply a given channel `SocketOption` to this protocol options state.
     func applyChannelOption(option: ChannelOptions.Types.SocketOption, value: SocketOptionValue) throws {
         switch (option.level, option.name) {