/src/swift-nio/Sources/NIOCore/ChannelPipeline.swift

Source
//===----------------------------------------------------------------------===//
//
// This source file is part of the SwiftNIO open source project
//
// Copyright (c) 2017-2024 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
//
//===----------------------------------------------------------------------===//

/// A list of `ChannelHandler`s that handle or intercept inbound events and outbound operations of a
/// `Channel`. `ChannelPipeline` implements an advanced form of the Intercepting Filter pattern
/// to give a user full control over how an event is handled and how the `ChannelHandler`s in a pipeline
/// interact with each other.
///
/// # Creation of a pipeline
///
/// Each `Channel` has its own `ChannelPipeline` and it is created automatically when a new `Channel` is created.
///
/// # How an event flows in a pipeline
///
/// The following diagram describes how I/O events are typically processed by `ChannelHandler`s in a `ChannelPipeline`.
/// An I/O event is handled by either a `ChannelInboundHandler` or a `ChannelOutboundHandler`
/// and is forwarded to the next handler in the `ChannelPipeline` by calling the event propagation methods defined in
/// `ChannelHandlerContext`, such as `ChannelHandlerContext.fireChannelRead` and
/// `ChannelHandlerContext.write`.
///
/// ```
///                                                    I/O Request
///                                                    via `Channel` or
///                                                    `ChannelHandlerContext`
///                                                      |
///  +---------------------------------------------------+---------------+
///  |                           ChannelPipeline         |               |
///  |                                TAIL              \|/              |
///  |    +---------------------+            +-----------+----------+    |
///  |    | Inbound Handler  N  |            | Outbound Handler  1  |    |
///  |    +----------+----------+            +-----------+----------+    |
///  |              /|\                                  |               |
///  |               |                                  \|/              |
///  |    +----------+----------+            +-----------+----------+    |
///  |    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
///  |    +----------+----------+            +-----------+----------+    |
///  |              /|\                                  .               |
///  |               .                                   .               |
///  | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
///  |        [ method call]                       [method call]         |
///  |               .                                   .               |
///  |               .                                  \|/              |
///  |    +----------+----------+            +-----------+----------+    |
///  |    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
///  |    +----------+----------+            +-----------+----------+    |
///  |              /|\                                  |               |
///  |               |                                  \|/              |
///  |    +----------+----------+            +-----------+----------+    |
///  |    | Inbound Handler  1  |            | Outbound Handler  M  |    |
///  |    +----------+----------+            +-----------+----------+    |
///  |              /|\             HEAD                 |               |
///  +---------------+-----------------------------------+---------------+
///                  |                                  \|/
///  +---------------+-----------------------------------+---------------+
///  |               |                                   |               |
///  |       [ Socket.read ]                    [ Socket.write ]         |
///  |                                                                   |
///  |  SwiftNIO Internal I/O Threads (Transport Implementation)         |
///  +-------------------------------------------------------------------+
/// ```
///
/// An inbound event is handled by the inbound handlers in the head-to-tail direction as shown on the left side of the
/// diagram. An inbound handler usually handles the inbound data generated by the I/O thread on the bottom of the
/// diagram. The inbound data is often read from a remote peer via the actual input operation such as
/// `Socket.read`. If an inbound event goes beyond the tail inbound handler, it is discarded
/// silently, or logged if it needs your attention.
///
/// An outbound event is handled by the outbound handlers in the tail-to-head direction as shown on the right side of the
/// diagram. An outbound handler usually generates or transforms the outbound traffic such as write requests.
/// If an outbound event goes beyond the head outbound handler, it is handled by an I/O thread associated with the
/// `Channel`. The I/O thread often performs the actual output operation such as `Socket.write`.
///
///
/// For example, let us assume that we created the following pipeline:
///
/// ```
/// ChannelPipeline p = ...
/// let future = p.add(name: "1", handler: InboundHandlerA()).flatMap {
///   p.add(name: "2", handler: InboundHandlerB())
/// }.flatMap {
///   p.add(name: "3", handler: OutboundHandlerA())
/// }.flatMap {
///   p.add(name: "4", handler: OutboundHandlerB())
/// }.flatMap {
///   p.add(name: "5", handler: InboundOutboundHandlerX())
/// }
/// // Handle the future as well ....
/// ```
///
/// In the example above, a class whose name starts with `Inbound` is an inbound handler.
/// A class whose name starts with `Outbound` is an outbound handler.
///
/// In the given example configuration, the handler evaluation order is 1, 2, 3, 4, 5 when an event goes inbound.
/// When an event goes outbound, the order is 5, 4, 3, 2, 1.  On top of this principle, `ChannelPipeline` skips
/// the evaluation of certain handlers to shorten the stack depth:
///
/// - 3 and 4 don't implement `ChannelInboundHandler`, and therefore the actual evaluation order of an inbound event will be: 1, 2, and 5.
/// - 1 and 2 don't implement `ChannelOutboundHandler`, and therefore the actual evaluation order of a outbound event will be: 5, 4, and 3.
/// - If 5 implements both `ChannelInboundHandler` and `ChannelOutboundHandler`, the evaluation order of an inbound and a outbound event could be 125 and 543 respectively.
///
/// - Note: Handlers may choose not to propagate messages down the pipeline immediately.  For example a handler may need to wait
/// for additional data before sending a protocol event to the next handler in the pipeline.  Due to this you can't assume that later handlers
/// in the pipeline will receive the same number of events as were sent, or that events of different types will arrive in the same order.
/// For example - a user event could overtake a data event if a handler is aggregating data events before propagating but immediately
/// propagating user events.
///
/// # Forwarding an event to the next handler
///
/// As you might noticed in the diagram above, a handler has to invoke the event propagation methods in
/// `ChannelHandlerContext` to forward an event to its next handler.
/// Those methods include:
///
/// - Inbound event propagation methods defined in `ChannelInboundInvoker`
/// - Outbound event propagation methods defined in `ChannelOutboundInvoker`.
///
/// # Building a pipeline
///
/// A user is supposed to have one or more `ChannelHandler`s in a `ChannelPipeline` to receive I/O events (e.g. read) and
/// to request I/O operations (e.g. write and close).  For example, a typical server will have the following handlers
/// in each channel's pipeline, but your mileage may vary depending on the complexity and characteristics of the
/// protocol and business logic:
///
/// - Protocol Decoder - translates binary data (e.g. `ByteBuffer`) into a struct / class
/// - Protocol Encoder - translates a struct / class into binary data (e.g. `ByteBuffer`)
/// - Business Logic Handler - performs the actual business logic (e.g. database access)
///
/// # Thread safety
///
/// A `ChannelHandler` can be added or removed at any time because a `ChannelPipeline` is thread safe.
public final class ChannelPipeline: ChannelInvoker {
    private var head: Optional<ChannelHandlerContext>
    private var tail: Optional<ChannelHandlerContext>

    private var idx: Int = 0
    internal private(set) var destroyed: Bool = false

    /// The `EventLoop` that is used by the underlying `Channel`.
    public let eventLoop: EventLoop

    /// The `Channel` that this `ChannelPipeline` belongs to.
    ///
    /// - Note: This will be nil after the channel has closed
    private var _channel: Optional<Channel>

    /// The `Channel` that this `ChannelPipeline` belongs to.
    @usableFromInline
    internal var channel: Channel {
        self.eventLoop.assertInEventLoop()
        assert(self._channel != nil || self.destroyed)
        return self._channel ?? DeadChannel(pipeline: self)
    }

    /// Add a `ChannelHandler` to the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - name: the name to use for the `ChannelHandler` when it's added. If none is specified it will generate a name.
    ///   - handler: the `ChannelHandler` to add
    ///   - position: The position in the `ChannelPipeline` to add `handler`. Defaults to `.last`.
    /// - Returns: the `EventLoopFuture` which will be notified once the `ChannelHandler` was added.
    @preconcurrency
    public func addHandler(
        _ handler: ChannelHandler & Sendable,
        name: String? = nil,
        position: ChannelPipeline.Position = .last
    ) -> EventLoopFuture<Void> {
        let future: EventLoopFuture<Void>

        if self.eventLoop.inEventLoop {
            let syncPosition = ChannelPipeline.SynchronousOperations.Position(position)
            future = self.eventLoop.makeCompletedFuture(
                self.addHandlerSync(handler, name: name, position: syncPosition)
            )
        } else {
            future = self.eventLoop.submit {
                let syncPosition = ChannelPipeline.SynchronousOperations.Position(position)
                try self.addHandlerSync(handler, name: name, position: syncPosition).get()
            }
        }

        return future
    }

    /// Synchronously add a `ChannelHandler` to the `ChannelPipeline`.
    ///
    /// May only be called from on the event loop.
    ///
    /// - Parameters:
    ///   - handler: the `ChannelHandler` to add
    ///   - name: the name to use for the `ChannelHandler` when it's added. If none is specified a name will be generated.
    ///   - position: The position in the `ChannelPipeline` to add `handler`. Defaults to `.last`.
    /// - Returns: the result of adding this handler - either success or failure with an error code if this could not be completed.
    fileprivate func addHandlerSync(
        _ handler: ChannelHandler,
        name: String? = nil,
        position: ChannelPipeline.SynchronousOperations.Position = .last
    ) -> Result<Void, Error> {
        self.eventLoop.assertInEventLoop()

        if self.destroyed {
            return .failure(ChannelError._ioOnClosedChannel)
        }

        switch position {
        case .first:
            return self.add0(
                name: name,
                handler: handler,
                relativeContext: head!,
                operation: self.add0(context:after:)
            )
        case .last:
            return self.add0(
                name: name,
                handler: handler,
                relativeContext: tail!,
                operation: self.add0(context:before:)
            )
        case .before(let beforeHandler):
            return self.add0(
                name: name,
                handler: handler,
                relativeHandler: beforeHandler,
                operation: self.add0(context:before:)
            )
        case .after(let afterHandler):
            return self.add0(
                name: name,
                handler: handler,
                relativeHandler: afterHandler,
                operation: self.add0(context:after:)
            )
        }
    }

    /// Synchronously add a `ChannelHandler` to the pipeline, relative to another `ChannelHandler`,
    /// where the insertion is done by a specific operation.
    ///
    /// May only be called from on the event loop.
    ///
    /// This will search the pipeline for `relativeHandler` and, if it cannot find it, will fail
    /// `promise` with `ChannelPipelineError.notFound`.
    ///
    /// - Parameters:
    ///   - name: The name to use for the `ChannelHandler` when its added. If none is specified, a name will be
    ///         automatically generated.
    ///   - handler: The `ChannelHandler` to add.
    ///   - relativeHandler: The `ChannelHandler` already in the `ChannelPipeline` that `handler` will be
    ///         inserted relative to.
    ///   - operation: A callback that will insert `handler` relative to `relativeHandler`.
    /// - Returns: the result of adding this handler - either success or failure with an error code if this could not be completed.
    private func add0(
        name: String?,
        handler: ChannelHandler,
        relativeHandler: ChannelHandler,
        operation: (ChannelHandlerContext, ChannelHandlerContext) -> Void
    ) -> Result<Void, Error> {
        self.eventLoop.assertInEventLoop()
        if self.destroyed {
            return .failure(ChannelError._ioOnClosedChannel)
        }

        guard let context = self.contextForPredicate0({ $0.handler === relativeHandler }) else {
            return .failure(ChannelPipelineError.notFound)
        }

        return self.add0(name: name, handler: handler, relativeContext: context, operation: operation)
    }

    /// Synchronously add a `ChannelHandler` to the pipeline, relative to a `ChannelHandlerContext`,
    /// where the insertion is done by a specific operation.
    ///
    /// May only be called from on the event loop.
    ///
    /// This method is more efficient than the one that takes a `relativeHandler` as it does not need to
    /// search the pipeline for the insertion point. It should be used whenever possible.
    ///
    /// - Parameters:
    ///   - name: The name to use for the `ChannelHandler` when its added. If none is specified, a name will be
    ///         automatically generated.
    ///   - handler: The `ChannelHandler` to add.
    ///   - relativeContext: The `ChannelHandlerContext` already in the `ChannelPipeline` that `handler` will be
    ///         inserted relative to.
    ///   - operation: A callback that will insert `handler` relative to `relativeHandler`.
    /// - Returns: the result of adding this handler - either success or failure with an error code if this could not be completed.
    private func add0(
        name: String?,
        handler: ChannelHandler,
        relativeContext: ChannelHandlerContext,
        operation: (ChannelHandlerContext, ChannelHandlerContext) -> Void
    ) -> Result<Void, Error> {
        self.eventLoop.assertInEventLoop()

        if self.destroyed {
            return .failure(ChannelError._ioOnClosedChannel)
        }

        let context = ChannelHandlerContext(name: name ?? nextName(), handler: handler, pipeline: self)
        operation(context, relativeContext)

        context.invokeHandlerAdded()
        return .success(())
    }

    /// Synchronously add a single new `ChannelHandlerContext` after one that currently exists in the
    /// pipeline.
    ///
    /// Must be called from within the event loop thread, as it synchronously manipulates the
    /// `ChannelHandlerContext`s on the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - new: The `ChannelHandlerContext` to add to the pipeline.
    ///   - existing: The `ChannelHandlerContext` that `new` will be added after.
    private func add0(context new: ChannelHandlerContext, after existing: ChannelHandlerContext) {
        self.eventLoop.assertInEventLoop()

        let next = existing.next
        new.prev = existing
        new.next = next
        existing.next = new
        next?.prev = new
    }

    /// Synchronously add a single new `ChannelHandlerContext` before one that currently exists in the
    /// pipeline.
    ///
    /// Must be called from within the event loop thread, as it synchronously manipulates the
    /// `ChannelHandlerContext`s on the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - new: The `ChannelHandlerContext` to add to the pipeline.
    ///   - existing: The `ChannelHandlerContext` that `new` will be added before.
    private func add0(context new: ChannelHandlerContext, before existing: ChannelHandlerContext) {
        self.eventLoop.assertInEventLoop()

        let prev = existing.prev
        new.prev = prev
        new.next = existing
        existing.prev = new
        prev?.next = new
    }

    /// Remove a `ChannelHandler` from the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - handler: the `ChannelHandler` to remove.
    /// - Returns: the `EventLoopFuture` which will be notified once the `ChannelHandler` was removed.
    @preconcurrency
    public func removeHandler(_ handler: RemovableChannelHandler & Sendable) -> EventLoopFuture<Void> {
        let promise = self.eventLoop.makePromise(of: Void.self)
        self.removeHandler(handler, promise: promise)
        return promise.futureResult
    }

    /// Remove a `ChannelHandler` from the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - name: the name that was used to add the `ChannelHandler` to the `ChannelPipeline` before.
    /// - Returns: the `EventLoopFuture` which will be notified once the `ChannelHandler` was removed.
    public func removeHandler(name: String) -> EventLoopFuture<Void> {
        let promise = self.eventLoop.makePromise(of: Void.self)
        self.removeHandler(name: name, promise: promise)
        return promise.futureResult
    }

    /// Remove a `ChannelHandler` from the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - context: the `ChannelHandlerContext` that belongs to `ChannelHandler` that should be removed.
    /// - Returns: the `EventLoopFuture` which will be notified once the `ChannelHandler` was removed.
    @available(
        *,
        deprecated,
        message: "Use .syncOperations.removeHandler(context:) instead, this method is not Sendable-safe."
    )
    public func removeHandler(context: ChannelHandlerContext) -> EventLoopFuture<Void> {
        let promise = self.eventLoop.makePromise(of: Void.self)
        self.removeHandler(context: context, promise: promise)
        return promise.futureResult
    }

    /// Remove a `ChannelHandler` from the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - handler: the `ChannelHandler` to remove.
    ///   - promise: An `EventLoopPromise` that will complete when the `ChannelHandler` is removed.
    @preconcurrency
    public func removeHandler(_ handler: RemovableChannelHandler & Sendable, promise: EventLoopPromise<Void>?) {
        @Sendable
        func removeHandler0() {
            self.syncOperations.removeHandler(handler, promise: promise)
        }

        if self.eventLoop.inEventLoop {
            removeHandler0()
        } else {
            self.eventLoop.execute {
                removeHandler0()
            }
        }
    }

    /// Remove a `ChannelHandler` from the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - name: the name that was used to add the `ChannelHandler` to the `ChannelPipeline` before.
    ///   - promise: An `EventLoopPromise` that will complete when the `ChannelHandler` is removed.
    public func removeHandler(name: String, promise: EventLoopPromise<Void>?) {
        @Sendable
        func removeHandler0() {
            self.syncOperations.removeHandler(name: name, promise: promise)
        }

        if self.eventLoop.inEventLoop {
            removeHandler0()
        } else {
            self.eventLoop.execute {
                removeHandler0()
            }
        }
    }

    /// Remove a `ChannelHandler` from the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - context: the `ChannelHandlerContext` that belongs to `ChannelHandler` that should be removed.
    ///   - promise: An `EventLoopPromise` that will complete when the `ChannelHandler` is removed.
    @available(
        *,
        deprecated,
        message: "Use .syncOperations.removeHandler(context:) instead, this method is not Sendable-safe."
    )
    public func removeHandler(context: ChannelHandlerContext, promise: EventLoopPromise<Void>?) {
        let sendableView = context.sendableView

        guard sendableView.channelHandlerIsRemovable else {
            promise?.fail(ChannelError._unremovableHandler)
            return
        }

        @Sendable
        func removeHandler0() {
            sendableView.wrappedValue.startUserTriggeredRemoval(promise: promise)
        }

        if self.eventLoop.inEventLoop {
            removeHandler0()
        } else {
            self.eventLoop.execute {
                removeHandler0()
            }
        }
    }

    /// Returns the `ChannelHandlerContext` that belongs to a `ChannelHandler`.
    ///
    /// - Parameters:
    ///   - handler: the `ChannelHandler` for which the `ChannelHandlerContext` should be returned
    /// - Returns: the `EventLoopFuture` which will be notified once the the operation completes.
    @available(
        *,
        deprecated,
        message: "This method is not strict concurrency safe. Prefer .syncOperations.context(handler:)"
    )
    @preconcurrency
    public func context(handler: ChannelHandler & Sendable) -> EventLoopFuture<ChannelHandlerContext> {
        let promise = self.eventLoop.makePromise(of: ChannelHandlerContext.self)

        if self.eventLoop.inEventLoop {
            promise.assumeIsolatedUnsafeUnchecked().completeWith(self.contextSync(handler: handler))
        } else {
            self.eventLoop.execute {
                promise.assumeIsolatedUnsafeUnchecked().completeWith(self.contextSync(handler: handler))
            }
        }

        return promise.futureResult
    }

    /// Synchronously returns the `ChannelHandlerContext` that belongs to a `ChannelHandler`.
    ///
    /// - Important: This must be called on the `EventLoop`.
    /// - Parameters:
    ///   - handler: the `ChannelHandler` for which the `ChannelHandlerContext` should be returned
    /// - Returns: the `ChannelHandlerContext` that belongs to the `ChannelHandler`, if one exists.
    fileprivate func contextSync(handler: ChannelHandler) -> Result<ChannelHandlerContext, Error> {
        self._contextSync({ $0.handler === handler })
    }

    /// Returns the `ChannelHandlerContext` that belongs to a `ChannelHandler`.
    ///
    /// - Parameters:
    ///   - name: the name that was used to add the `ChannelHandler` to the `ChannelPipeline` before.
    /// - Returns: the `EventLoopFuture` which will be notified once the the operation completes.
    public func context(name: String) -> EventLoopFuture<ChannelHandlerContext> {
        let promise = self.eventLoop.makePromise(of: ChannelHandlerContext.self)

        if self.eventLoop.inEventLoop {
            promise.assumeIsolatedUnsafeUnchecked().completeWith(self.contextSync(name: name))
        } else {
            self.eventLoop.execute {
                promise.assumeIsolatedUnsafeUnchecked().completeWith(self.contextSync(name: name))
            }
        }

        return promise.futureResult
    }

    /// Synchronously finds and returns the `ChannelHandlerContext` that belongs to the
    /// `ChannelHandler` with the given name.
    ///
    /// - Important: This must be called on the `EventLoop`.
    /// - Parameter name: The name of the `ChannelHandler` to find.
    /// - Returns: the `ChannelHandlerContext` that belongs to the `ChannelHandler`, if one exists.
    fileprivate func contextSync(name: String) -> Result<ChannelHandlerContext, Error> {
        self._contextSync({ $0.name == name })
    }

    /// Returns the `ChannelHandlerContext` that belongs to a `ChannelHandler` of the given type.
    ///
    /// If multiple channel handlers of the same type are present in the pipeline, returns the context
    /// belonging to the first such handler.
    ///
    /// - Parameters:
    ///   - handlerType: The type of the handler to search for.
    /// - Returns: the `EventLoopFuture` which will be notified once the the operation completes.
    @inlinable
    @preconcurrency
    public func context<Handler: ChannelHandler & _NIOCoreSendableMetatype>(
        handlerType: Handler.Type
    ) -> EventLoopFuture<ChannelHandlerContext> {
        let promise = self.eventLoop.makePromise(of: ChannelHandlerContext.self)

        if self.eventLoop.inEventLoop {
            promise.assumeIsolatedUnsafeUnchecked().completeWith(self._contextSync(handlerType: handlerType))
        } else {
            self.eventLoop.execute {
                promise.assumeIsolatedUnsafeUnchecked().completeWith(self._contextSync(handlerType: handlerType))
            }
        }

        return promise.futureResult
    }

    /// Returns if the ``ChannelHandler`` of the given type is contained in the pipeline.
    ///
    /// - Parameters:
    ///   - type: The type of the handler.
    /// - Returns: An ``EventLoopFuture`` that is succeeded if a handler of the given type is contained in the pipeline. Otherwise
    /// the future will be failed with an error.
    @inlinable
    @preconcurrency
    public func containsHandler<Handler: ChannelHandler & _NIOCoreSendableMetatype>(
        type: Handler.Type
    ) -> EventLoopFuture<Void> {
        self.handler(type: type).map { _ in () }
    }

    /// Returns if the ``ChannelHandler`` of the given type is contained in the pipeline.
    ///
    /// - Parameters:
    ///   - name: The name of the handler.
    /// - Returns: An ``EventLoopFuture`` that is succeeded if a handler of the given type is contained in the pipeline. Otherwise
    /// the future will be failed with an error.
    @inlinable
    public func containsHandler(name: String) -> EventLoopFuture<Void> {
        self.context(name: name).map { _ in () }
    }

    /// Synchronously finds and returns the `ChannelHandlerContext` that belongs to the first
    /// `ChannelHandler` of the given type.
    ///
    /// - Important: This must be called on the `EventLoop`.
    /// - Parameter handlerType: The type of handler to search for.
    /// - Returns: the `ChannelHandlerContext` that belongs to the `ChannelHandler`, if one exists.
    @inlinable  // should be fileprivate
    internal func _contextSync<Handler: ChannelHandler>(
        handlerType: Handler.Type
    ) -> Result<ChannelHandlerContext, Error> {
        self._contextSync({ $0.handler is Handler })
    }

    /// Synchronously finds a `ChannelHandlerContext` in the `ChannelPipeline`.
    /// - Important: This must be called on the `EventLoop`.
    @usableFromInline  // should be fileprivate
    internal func _contextSync(_ body: (ChannelHandlerContext) -> Bool) -> Result<ChannelHandlerContext, Error> {
        self.eventLoop.assertInEventLoop()

        if let context = self.contextForPredicate0(body) {
            return .success(context)
        } else {
            return .failure(ChannelPipelineError.notFound)
        }
    }

    /// Returns a `ChannelHandlerContext` which matches.
    ///
    /// This skips head and tail (as these are internal and should not be accessible by the user).
    ///
    /// - Parameters:
    ///   - body: The predicate to execute per `ChannelHandlerContext` in the `ChannelPipeline`.
    /// - Returns: The first `ChannelHandlerContext` that matches or `nil` if none did.
    private func contextForPredicate0(_ body: (ChannelHandlerContext) -> Bool) -> ChannelHandlerContext? {
        var curCtx: ChannelHandlerContext? = self.head?.next
        while let context = curCtx, context !== self.tail {
            if body(context) {
                return context
            }
            curCtx = context.next
        }

        return nil
    }

    /// Remove a `ChannelHandlerContext` from the `ChannelPipeline` directly without going through the
    /// `RemovableChannelHandler` API. This must only be used to clear the pipeline on `Channel` tear down and
    /// as a result of the `leavePipeline` call in the `RemovableChannelHandler` API.
    internal func removeHandlerFromPipeline(context: ChannelHandlerContext, promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        let nextCtx = context.next
        let prevCtx = context.prev

        if let prevCtx = prevCtx {
            prevCtx.next = nextCtx
        }
        if let nextCtx = nextCtx {
            nextCtx.prev = prevCtx
        }

        context.invokeHandlerRemoved()
        promise?.succeed(())

        // We need to keep the current node alive until after the callout in case the user uses the context.
        context.next = nil
        context.prev = nil
    }

    /// Returns the next name to use for a `ChannelHandler`.
    private func nextName() -> String {
        self.eventLoop.assertInEventLoop()

        let name = "handler\(idx)"
        idx += 1
        return name
    }

    /// Remove all the `ChannelHandler`s from the `ChannelPipeline` and destroy these.
    ///
    /// This method must only be called from within the `EventLoop`. It should only be called from a `ChannelCore`
    /// implementation. Once called, the `ChannelPipeline` is no longer active and cannot be used again.
    func removeHandlers() {
        self.eventLoop.assertInEventLoop()

        if let head = self.head {
            while let context = head.next {
                removeHandlerFromPipeline(context: context, promise: nil)
            }
            removeHandlerFromPipeline(context: self.head!, promise: nil)
        }
        self.head = nil
        self.tail = nil

        self.destroyed = true
        self._channel = nil
    }

    // Just delegate to the head and tail context
    public func fireChannelRegistered() {
        if eventLoop.inEventLoop {
            fireChannelRegistered0()
        } else {
            eventLoop.execute {
                self.fireChannelRegistered0()
            }
        }
    }

    public func fireChannelUnregistered() {
        if eventLoop.inEventLoop {
            fireChannelUnregistered0()
        } else {
            eventLoop.execute {
                self.fireChannelUnregistered0()
            }
        }
    }

    public func fireChannelInactive() {
        if eventLoop.inEventLoop {
            fireChannelInactive0()
        } else {
            eventLoop.execute {
                self.fireChannelInactive0()
            }
        }
    }

    public func fireChannelActive() {
        if eventLoop.inEventLoop {
            fireChannelActive0()
        } else {
            eventLoop.execute {
                self.fireChannelActive0()
            }
        }
    }

    @available(
        *,
        deprecated,
        message: "NIOAny is not Sendable. Avoid wrapping the value in NIOAny to silence this warning."
    )
    public func fireChannelRead(_ data: NIOAny) {
        if eventLoop.inEventLoop {
            _fireChannelRead0(data)
        } else {
            // This is unsafe, but necessary.
            let unsafeTransfer = UnsafeTransfer(data)
            eventLoop.execute {
                self._fireChannelRead0(unsafeTransfer.wrappedValue)
            }
        }
    }

    @inlinable
    public func fireChannelRead<T: Sendable>(_ data: T) {
        if eventLoop.inEventLoop {
            _fireChannelRead0(NIOAny(data))
        } else {
            eventLoop.execute {
                self._fireChannelRead0(NIOAny(data))
            }
        }
    }

    public func fireChannelReadComplete() {
        if eventLoop.inEventLoop {
            fireChannelReadComplete0()
        } else {
            eventLoop.execute {
                self.fireChannelReadComplete0()
            }
        }
    }

    public func fireChannelWritabilityChanged() {
        if eventLoop.inEventLoop {
            fireChannelWritabilityChanged0()
        } else {
            eventLoop.execute {
                self.fireChannelWritabilityChanged0()
            }
        }
    }

    @preconcurrency
    public func fireUserInboundEventTriggered(_ event: Any & Sendable) {
        if eventLoop.inEventLoop {
            fireUserInboundEventTriggered0(event)
        } else {
            eventLoop.execute {
                self.fireUserInboundEventTriggered0(event)
            }
        }
    }

    public func fireErrorCaught(_ error: Error) {
        if eventLoop.inEventLoop {
            fireErrorCaught0(error: error)
        } else {
            eventLoop.execute {
                self.fireErrorCaught0(error: error)
            }
        }
    }

    public func close(mode: CloseMode = .all, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            close0(mode: mode, promise: promise)
        } else {
            eventLoop.execute {
                self.close0(mode: mode, promise: promise)
            }
        }
    }

    public func flush() {
        if eventLoop.inEventLoop {
            flush0()
        } else {
            eventLoop.execute {
                self.flush0()
            }
        }
    }

    public func read() {
        if eventLoop.inEventLoop {
            read0()
        } else {
            eventLoop.execute {
                self.read0()
            }
        }
    }

    @available(
        *,
        deprecated,
        message: "NIOAny is not Sendable. Avoid wrapping the value in NIOAny to silence this warning."
    )
    public func write(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            _write0(data, promise: promise)
        } else {
            // This is unsafe, but unavoidable.
            let unsafeTransfer = UnsafeTransfer(data)
            eventLoop.execute {
                self._write0(unsafeTransfer.wrappedValue, promise: promise)
            }
        }
    }

    @inlinable
    public func write<T: Sendable>(_ data: T, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            _write0(NIOAny(data), promise: promise)
        } else {
            eventLoop.execute {
                self._write0(NIOAny(data), promise: promise)
            }
        }
    }

    @available(
        *,
        deprecated,
        message: "NIOAny is not Sendable. Avoid wrapping the value in NIOAny to silence this warning."
    )
    public func writeAndFlush(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            _writeAndFlush0(data, promise: promise)
        } else {
            // This is unsafe, but unavoidable.
            let unsafeTransfer = UnsafeTransfer(data)
            eventLoop.execute {
                self._writeAndFlush0(unsafeTransfer.wrappedValue, promise: promise)
            }
        }
    }

    @inlinable
    public func writeAndFlush<T: Sendable>(_ data: T, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            _writeAndFlush0(NIOAny(data), promise: promise)
        } else {
            eventLoop.execute {
                self._writeAndFlush0(NIOAny(data), promise: promise)
            }
        }
    }

    public func bind(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            bind0(to: address, promise: promise)
        } else {
            eventLoop.execute {
                self.bind0(to: address, promise: promise)
            }
        }
    }

    public func connect(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            connect0(to: address, promise: promise)
        } else {
            eventLoop.execute {
                self.connect0(to: address, promise: promise)
            }
        }
    }

    public func register(promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            register0(promise: promise)
        } else {
            eventLoop.execute {
                self.register0(promise: promise)
            }
        }
    }

    @preconcurrency
    public func triggerUserOutboundEvent(_ event: Any & Sendable, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            triggerUserOutboundEvent0(event, promise: promise)
        } else {
            eventLoop.execute {
                self.triggerUserOutboundEvent0(event, promise: promise)
            }
        }
    }

    // These methods are expected to only be called from within the EventLoop

    private var firstOutboundCtx: ChannelHandlerContext? {
        self.tail?.prev
    }

    private var firstInboundCtx: ChannelHandlerContext? {
        self.head?.next
    }

    private func close0(mode: CloseMode, promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeClose(mode: mode, promise: promise)
        } else {
            promise?.fail(ChannelError._alreadyClosed)
        }
    }

    private func flush0() {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeFlush()
        }
    }

    private func read0() {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeRead()
        }
    }

    @usableFromInline func _write0(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeWrite(data, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    @usableFromInline func _writeAndFlush0(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeWriteAndFlush(data, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    private func bind0(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeBind(to: address, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    private func connect0(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeConnect(to: address, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    private func register0(promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeRegister(promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    private func triggerUserOutboundEvent0(_ event: Any, promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeTriggerUserOutboundEvent(event, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    private func fireChannelRegistered0() {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeChannelRegistered()
        }
    }

    private func fireChannelUnregistered0() {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeChannelUnregistered()
        }
    }

    private func fireChannelInactive0() {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeChannelInactive()
        }
    }

    private func fireChannelActive0() {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeChannelActive()
        }
    }

    @usableFromInline func _fireChannelRead0(_ data: NIOAny) {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeChannelRead(data)
        }
    }

    private func fireChannelReadComplete0() {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeChannelReadComplete()
        }
    }

    private func fireChannelWritabilityChanged0() {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeChannelWritabilityChanged()
        }
    }

    private func fireUserInboundEventTriggered0(_ event: Any) {
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeUserInboundEventTriggered(event)
        }
    }

    private func fireErrorCaught0(error: Error) {
        assert((error as? ChannelError).map { $0 != .eof } ?? true)
        if let firstInboundCtx = firstInboundCtx {
            firstInboundCtx.invokeErrorCaught(error)
        }
    }

    private var inEventLoop: Bool {
        eventLoop.inEventLoop
    }

    /// Create `ChannelPipeline` for a given `Channel`. This method should never be called by the end-user
    /// directly: it is only intended for use with custom `Channel` implementations. Users should always use
    /// `channel.pipeline` to access the `ChannelPipeline` for a `Channel`.
    ///
    /// - Parameters:
    ///    - channel: The `Channel` this `ChannelPipeline` is created for.
    public init(channel: Channel) {
        self._channel = channel
        self.eventLoop = channel.eventLoop
        self.head = nil  // we need to initialise these to `nil` so we can use `self` in the lines below
        self.tail = nil  // we need to initialise these to `nil` so we can use `self` in the lines below

        self.head = ChannelHandlerContext(
            name: HeadChannelHandler.name,
            handler: HeadChannelHandler.sharedInstance,
            pipeline: self
        )
        self.tail = ChannelHandlerContext(
            name: TailChannelHandler.name,
            handler: TailChannelHandler.sharedInstance,
            pipeline: self
        )
        self.head?.next = self.tail
        self.tail?.prev = self.head
    }
}

extension ChannelPipeline: @unchecked Sendable {}

extension ChannelPipeline {
    /// Adds the provided channel handlers to the pipeline in the order given, taking account
    /// of the behaviour of `ChannelHandler.add(first:)`.
    ///
    /// - Parameters:
    ///   - handlers: The array of `ChannelHandler`s to be added.
    ///   - position: The position in the `ChannelPipeline` to add `handlers`. Defaults to `.last`.
    ///
    /// - Returns: A future that will be completed when all of the supplied `ChannelHandler`s were added.
    @preconcurrency
    public func addHandlers(
        _ handlers: [ChannelHandler & Sendable],
        position: ChannelPipeline.Position = .last
    ) -> EventLoopFuture<Void> {
        let future: EventLoopFuture<Void>

        if self.eventLoop.inEventLoop {
            future = self.eventLoop.makeCompletedFuture(self.addHandlersSync(handlers, position: position))
        } else {
            future = self.eventLoop.submit {
                try self.addHandlersSync(handlers, position: position).get()
            }
        }

        return future
    }

    /// Adds the provided channel handlers to the pipeline in the order given, taking account
    /// of the behaviour of `ChannelHandler.add(first:)`.
    ///
    /// - Parameters:
    ///   - handlers: One or more `ChannelHandler`s to be added.
    ///   - position: The position in the `ChannelPipeline` to add `handlers`. Defaults to `.last`.
    ///
    /// - Returns: A future that will be completed when all of the supplied `ChannelHandler`s were added.
    @preconcurrency
    public func addHandlers(
        _ handlers: (ChannelHandler & Sendable)...,
        position: ChannelPipeline.Position = .last
    ) -> EventLoopFuture<Void> {
        self.addHandlers(handlers, position: position)
    }

    /// Synchronously adds the provided `ChannelHandler`s to the pipeline in the order given, taking
    /// account of the behaviour of `ChannelHandler.add(first:)`.
    ///
    /// - Important: Must be called on the `EventLoop`.
    /// - Parameters:
    ///   - handlers: The array of `ChannelHandler`s to add.
    ///   - position: The position in the `ChannelPipeline` to add the handlers.
    /// - Returns: A result representing whether the handlers were added or not.
    fileprivate func addHandlersSync(
        _ handlers: [ChannelHandler & Sendable],
        position: ChannelPipeline.Position
    ) -> Result<Void, Error> {
        let syncPosition = ChannelPipeline.SynchronousOperations.Position(position)
        switch syncPosition {
        case .first, .after:
            return self._addHandlersSync(handlers.reversed(), position: syncPosition)
        case .last, .before:
            return self._addHandlersSync(handlers, position: syncPosition)
        }
    }

    /// Synchronously adds the provided `ChannelHandler`s to the pipeline in the order given, taking
    /// account of the behaviour of `ChannelHandler.add(first:)`.
    ///
    /// This duplicate of the above method exists to avoid needing to rebox the array of existentials
    /// from any (ChannelHandler & Sendable) to any ChannelHandler.
    ///
    /// - Important: Must be called on the `EventLoop`.
    /// - Parameters:
    ///   - handlers: The array of `ChannelHandler`s to add.
    ///   - position: The position in the `ChannelPipeline` to add the handlers.
    /// - Returns: A result representing whether the handlers were added or not.
    fileprivate func addHandlersSyncNotSendable(
        _ handlers: [ChannelHandler],
        position: ChannelPipeline.SynchronousOperations.Position
    ) -> Result<Void, Error> {
        switch position {
        case .first, .after:
            return self._addHandlersSyncNotSendable(handlers.reversed(), position: position)
        case .last, .before:
            return self._addHandlersSyncNotSendable(handlers, position: position)
        }
    }

    /// Synchronously adds a sequence of `ChannelHandlers` to the pipeline at the given position.
    ///
    /// - Important: Must be called on the `EventLoop`.
    /// - Parameters:
    ///   - handlers: A sequence of handlers to add.
    ///   - position: The position in the `ChannelPipeline` to add the handlers.
    /// - Returns: A result representing whether the handlers were added or not.
    private func _addHandlersSync<Handlers: Sequence>(
        _ handlers: Handlers,
        position: ChannelPipeline.SynchronousOperations.Position
    ) -> Result<Void, Error> where Handlers.Element == ChannelHandler & Sendable {
        self.eventLoop.assertInEventLoop()

        for handler in handlers {
            let result = self.addHandlerSync(handler, position: position)
            switch result {
            case .success:
                ()
            case .failure:
                return result
            }
        }

        return .success(())
    }

    /// Synchronously adds a sequence of `ChannelHandlers` to the pipeline at the given position.
    ///
    /// This duplicate of the above method exists to avoid needing to rebox the array of existentials
    /// from any (ChannelHandler & Sendable) to any ChannelHandler.
    ///
    /// - Important: Must be called on the `EventLoop`.
    /// - Parameters:
    ///   - handlers: A sequence of handlers to add.
    ///   - position: The position in the `ChannelPipeline` to add the handlers.
    /// - Returns: A result representing whether the handlers were added or not.
    private func _addHandlersSyncNotSendable<Handlers: Sequence>(
        _ handlers: Handlers,
        position: ChannelPipeline.SynchronousOperations.Position
    ) -> Result<Void, Error> where Handlers.Element == ChannelHandler {
        self.eventLoop.assertInEventLoop()

        for handler in handlers {
            let result = self.addHandlerSync(handler, position: position)
            switch result {
            case .success:
                ()
            case .failure:
                return result
            }
        }

        return .success(())
    }
}

// MARK: - Synchronous View

extension ChannelPipeline {
    /// A view of a `ChannelPipeline` which may be used to invoke synchronous operations.
    ///
    /// All functions **must** be called from the pipeline's event loop.
    public struct SynchronousOperations {
        @usableFromInline
        internal let _pipeline: ChannelPipeline

        fileprivate init(pipeline: ChannelPipeline) {
            self._pipeline = pipeline
        }

        /// The `EventLoop` of the `Channel` this synchronous operations view corresponds to.
        public var eventLoop: EventLoop {
            self._pipeline.eventLoop
        }

        /// Add a handler to the pipeline.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameters:
        ///   - handler: The handler to add.
        ///   - name: The name to use for the `ChannelHandler` when it's added. If no name is specified the one will be generated.
        ///   - position: The position in the `ChannelPipeline` to add `handler`. Defaults to `.last`.
        public func addHandler(
            _ handler: ChannelHandler,
            name: String? = nil,
            position: ChannelPipeline.SynchronousOperations.Position = .last
        ) throws {
            try self._pipeline.addHandlerSync(handler, name: name, position: position).get()
        }

        /// Add a handler to the pipeline.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameters:
        ///   - handler: The handler to add.
        ///   - name: The name to use for the `ChannelHandler` when it's added. If no name is specified the one will be generated.
        ///   - position: The position in the `ChannelPipeline` to add `handler`. Defaults to `.last`.
        @available(*, deprecated, message: "Use ChannelPipeline.SynchronousOperations.Position instead")
        @_disfavoredOverload
        public func addHandler(
            _ handler: ChannelHandler,
            name: String? = nil,
            position: ChannelPipeline.Position = .last
        ) throws {
            let syncPosition = ChannelPipeline.SynchronousOperations.Position(position)
            try self._pipeline.addHandlerSync(handler, name: name, position: syncPosition).get()
        }

        /// Add an array of handlers to the pipeline.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameters:
        ///   - handlers: The handlers to add.
        ///   - position: The position in the `ChannelPipeline` to add `handlers`. Defaults to `.last`.
        public func addHandlers(
            _ handlers: [ChannelHandler],
            position: ChannelPipeline.SynchronousOperations.Position = .last
        ) throws {
            try self._pipeline.addHandlersSyncNotSendable(handlers, position: position).get()
        }

        /// Add an array of handlers to the pipeline.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameters:
        ///   - handlers: The handlers to add.
        ///   - position: The position in the `ChannelPipeline` to add `handlers`. Defaults to `.last`.
        @available(*, deprecated, message: "Use ChannelPipeline.SynchronousOperations.Position instead")
        @_disfavoredOverload
        public func addHandlers(
            _ handlers: [ChannelHandler],
            position: ChannelPipeline.Position = .last
        ) throws {
            let syncPosition = ChannelPipeline.SynchronousOperations.Position(position)
            try self._pipeline.addHandlersSyncNotSendable(handlers, position: syncPosition).get()
        }

        /// Add one or more handlers to the pipeline.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameters:
        ///   - handlers: The handlers to add.
        ///   - position: The position in the `ChannelPipeline` to add `handlers`. Defaults to `.last`.
        public func addHandlers(
            _ handlers: ChannelHandler...,
            position: ChannelPipeline.SynchronousOperations.Position = .last
        ) throws {
            try self._pipeline.addHandlersSyncNotSendable(handlers, position: position).get()
        }

        /// Add one or more handlers to the pipeline.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameters:
        ///   - handlers: The handlers to add.
        ///   - position: The position in the `ChannelPipeline` to add `handlers`. Defaults to `.last`.
        @available(*, deprecated, message: "Use ChannelPipeline.SynchronousOperations.Position instead")
        @_disfavoredOverload
        public func addHandlers(
            _ handlers: ChannelHandler...,
            position: ChannelPipeline.Position = .last
        ) throws {
            let syncPosition = ChannelPipeline.SynchronousOperations.Position(position)
            try self._pipeline.addHandlersSyncNotSendable(handlers, position: syncPosition).get()
        }

        /// Remove a `ChannelHandler` from the `ChannelPipeline`.
        ///
        /// - Parameters:
        ///   - handler: the `ChannelHandler` to remove.
        /// - Returns: the `EventLoopFuture` which will be notified once the `ChannelHandler` was removed.
        public func removeHandler(_ handler: RemovableChannelHandler) -> EventLoopFuture<Void> {
            let promise = self.eventLoop.makePromise(of: Void.self)
            self.removeHandler(handler, promise: promise)
            return promise.futureResult
        }

        /// Remove a ``ChannelHandler`` from the ``ChannelPipeline``.
        ///
        /// - Parameters:
        ///   - handler: the ``ChannelHandler`` to remove.
        ///   - promise: an ``EventLoopPromise`` to notify when the ``ChannelHandler`` was removed.
        public func removeHandler(_ handler: RemovableChannelHandler, promise: EventLoopPromise<Void>?) {
            switch self._pipeline.contextSync(handler: handler) {
            case .success(let context):
                self.removeHandler(context: context, promise: promise)
            case .failure(let error):
                promise?.fail(error)
            }
        }

        /// Remove a `ChannelHandler` from the `ChannelPipeline`.
        ///
        /// - Parameters:
        ///   - name: the name that was used to add the `ChannelHandler` to the `ChannelPipeline` before.
        /// - Returns: the `EventLoopFuture` which will be notified once the `ChannelHandler` was removed.
        public func removeHandler(name: String) -> EventLoopFuture<Void> {
            let promise = self.eventLoop.makePromise(of: Void.self)
            self.removeHandler(name: name, promise: promise)
            return promise.futureResult
        }

        /// Remove a ``ChannelHandler`` from the ``ChannelPipeline``.
        ///
        /// - Parameters:
        ///   - name: the name that was used to add the `ChannelHandler` to the `ChannelPipeline` before.
        ///   - promise: an ``EventLoopPromise`` to notify when the ``ChannelHandler`` was removed.
        public func removeHandler(name: String, promise: EventLoopPromise<Void>?) {
            switch self._pipeline.contextSync(name: name) {
            case .success(let context):
                self.removeHandler(context: context, promise: promise)
            case .failure(let error):
                promise?.fail(error)
            }
        }

        /// Remove a `ChannelHandler` from the `ChannelPipeline`.
        ///
        /// - Parameters:
        ///   - context: the `ChannelHandlerContext` that belongs to `ChannelHandler` that should be removed.
        /// - Returns: the `EventLoopFuture` which will be notified once the `ChannelHandler` was removed.
        public func removeHandler(context: ChannelHandlerContext) -> EventLoopFuture<Void> {
            let promise = self.eventLoop.makePromise(of: Void.self)
            self.removeHandler(context: context, promise: promise)
            return promise.futureResult
        }

        /// Remove a `ChannelHandler` from the `ChannelPipeline`.
        ///
        /// - Parameters:
        ///   - context: the `ChannelHandlerContext` that belongs to `ChannelHandler` that should be removed.
        ///   - promise: an ``EventLoopPromise`` to notify when the ``ChannelHandler`` was removed.
        public func removeHandler(context: ChannelHandlerContext, promise: EventLoopPromise<Void>?) {
            if context.handler is RemovableChannelHandler {
                context.startUserTriggeredRemoval(promise: promise)
            } else {
                promise?.fail(ChannelError.unremovableHandler)
            }
        }

        /// Returns the `ChannelHandlerContext` for the given handler instance if it is in
        /// the `ChannelPipeline`, if it exists.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameter handler: The handler belonging to the context to fetch.
        /// - Returns: The `ChannelHandlerContext` associated with the handler.
        public func context(handler: ChannelHandler) throws -> ChannelHandlerContext {
            try self._pipeline._contextSync({ $0.handler === handler }).get()
        }

        /// Returns the `ChannelHandlerContext` for the handler with the given name, if one exists.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameter name: The name of the handler whose context is being fetched.
        /// - Returns: The `ChannelHandlerContext` associated with the handler.
        public func context(name: String) throws -> ChannelHandlerContext {
            try self._pipeline.contextSync(name: name).get()
        }

        /// Returns the `ChannelHandlerContext` for the handler of given type, if one exists.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Parameter handlerType: The type of the handler to search for.
        /// - Returns: The `ChannelHandlerContext` associated with the handler.
        @inlinable
        public func context<Handler: ChannelHandler>(handlerType: Handler.Type) throws -> ChannelHandlerContext {
            try self._pipeline._contextSync(handlerType: handlerType).get()
        }

        /// Returns the `ChannelHandler` of the given type from the `ChannelPipeline`, if it exists.
        ///
        /// - Important: This *must* be called on the event loop.
        /// - Returns: A `ChannelHandler` of the given type if one exists in the `ChannelPipeline`.
        @inlinable
        public func handler<Handler: ChannelHandler>(type _: Handler.Type) throws -> Handler {
            try self._pipeline._handlerSync(type: Handler.self).get()
        }

        /// Fires `channelRegistered` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireChannelRegistered() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireChannelRegistered0()
        }

        /// Fires `channelUnregistered` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireChannelUnregistered() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireChannelUnregistered0()
        }

        /// Fires `channelInactive` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireChannelInactive() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireChannelInactive0()
        }

        /// Fires `channelActive` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireChannelActive() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireChannelActive0()
        }

        /// Fires `channelRead` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireChannelRead(_ data: NIOAny) {
            self.eventLoop.assertInEventLoop()
            self._pipeline._fireChannelRead0(data)
        }

        /// Fires `channelReadComplete` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireChannelReadComplete() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireChannelReadComplete0()
        }

        /// Fires `channelWritabilityChanged` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireChannelWritabilityChanged() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireChannelWritabilityChanged0()
        }

        /// Fires `userInboundEventTriggered` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireUserInboundEventTriggered(_ event: Any) {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireUserInboundEventTriggered0(event)
        }

        /// Fires `errorCaught` from the head to the tail.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func fireErrorCaught(_ error: Error) {
            self.eventLoop.assertInEventLoop()
            self._pipeline.fireErrorCaught0(error: error)
        }

        /// Fires `close` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func close(mode: CloseMode = .all, promise: EventLoopPromise<Void>?) {
            self.eventLoop.assertInEventLoop()
            self._pipeline.close0(mode: mode, promise: promise)
        }

        /// Fires `flush` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func flush() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.flush0()
        }

        /// Fires `read` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func read() {
            self.eventLoop.assertInEventLoop()
            self._pipeline.read0()
        }

        /// Fires `write` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func write(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
            self.eventLoop.assertInEventLoop()
            self._pipeline._write0(data, promise: promise)
        }

        /// Fires `write` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func write(_ data: NIOAny) -> EventLoopFuture<Void> {
            self.eventLoop.assertInEventLoop()
            let promise = self.eventLoop.makePromise(of: Void.self)
            self._pipeline._write0(data, promise: promise)
            return promise.futureResult
        }

        /// Fires `writeAndFlush` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func writeAndFlush(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
            self.eventLoop.assertInEventLoop()
            self._pipeline._writeAndFlush0(data, promise: promise)
        }

        /// Fires `writeAndFlush` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func writeAndFlush(_ data: NIOAny) -> EventLoopFuture<Void> {
            self.eventLoop.assertInEventLoop()
            let promise = self.eventLoop.makePromise(of: Void.self)
            self._pipeline._writeAndFlush0(data, promise: promise)
            return promise.futureResult
        }

        /// Fires `bind` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func bind(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
            self.eventLoop.assertInEventLoop()
            self._pipeline.bind0(to: address, promise: promise)
        }

        /// Fires `connect` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func connect(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
            self.eventLoop.assertInEventLoop()
            self._pipeline.connect0(to: address, promise: promise)
        }

        /// Fires `register` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func register(promise: EventLoopPromise<Void>?) {
            self.eventLoop.assertInEventLoop()
            self._pipeline.register0(promise: promise)
        }

        /// Fires `triggerUserOutboundEvent` from the tail to the head.
        ///
        /// This method should typically only be called by `Channel` implementations directly.
        public func triggerUserOutboundEvent(_ event: Any, promise: EventLoopPromise<Void>?) {
            self.eventLoop.assertInEventLoop()
            self._pipeline.triggerUserOutboundEvent0(event, promise: promise)
        }

        /// Provides scoped access to the underlying transport, if the channel supports it.
        ///
        /// This is an advanced API for reading or manipulating the underlying transport that backs a channel. Users
        /// must not close the transport or invalidate any invariants that the channel relies upon for its operation.
        ///
        /// Not all channels support access to the underlying channel. If the channel does not support this API, the
        /// closure is not called and this function immediately returns `nil`.
        ///
        /// Note that you must call this API with an appropriate closure, or otherwise explicitly specify the correct
        /// transport type prarameter, in order for the closure to be run. Calling this function such that the compiler
        /// infers a type for the transport closure parameter that differs from the channel implementation will result
        /// in the closure not being run and this function will return `nil`.
        ///
        /// For example, for socket-based channels, that expose the underlying socket handle:
        ///
        /// ```swift
        /// try channel.pipeline.syncOperations.withUnsafeTransportIfAvailable { transport in
        ///     // This closure is called.
        ///     transport == NIOBSDSocketHandle.invalid
        /// }
        ///
        /// try channel.pipeline.syncOperations.withUnsafeTransportIfAvailable { (_: NIOBSDSocket.Handle) in
        ///     // This closure is called.
        ///     return
        /// }
        ///
        /// try channel.pipeline.syncOperations.withUnsafeTransportIfAvailable(of: NIOBSDSocket.Handle.self) { _ in
        ///     // This closure is called.
        ///     return
        /// }
        ///
        /// try channel.pipeline.syncOperations.withUnsafeTransportIfAvailable {
        ///     // This closure is NOT called.
        ///     return
        /// }
        ///
        /// try channel.pipeline.syncOperations.withUnsafeTransportIfAvailable { (_: Any.self) in
        ///     // This closure is NOT called.
        ///     return
        /// }
        ///
        /// try channel.pipeline.syncOperations.withUnsafeTransportIfAvailable(of: Any.self) { _ in
        ///     // This closure is NOT called.
        ///     return
        /// }
        /// ```
        ///
        /// - Parameters:
        ///   - type: The expected transport type the channel makes available.
        ///   - body: /// A closure that takes the underlying transport, if the channel supports this operation.
        /// - Returns: The value returned by the closure, or `nil` if the channel does not expose its transport.
        /// - Throws: If there was an error accessing the underlying transport, or an error was thrown by the closure.
        @available(macOS 13, iOS 16, tvOS 16, watchOS 9, *)
        @inlinable
        public func withUnsafeTransportIfAvailable<Transport, Result>(
            of type: Transport.Type = Transport.self,
            _ body: (_ transport: Transport) throws -> Result
        ) throws -> Result? {
            self.eventLoop.assertInEventLoop()
            guard let core = self._pipeline.channel._channelCore as? any NIOTransportAccessibleChannelCore<Transport>
            else {
                return nil
            }
            return try core.withUnsafeTransport(body)
        }
    }

    /// Returns a view of operations which can be performed synchronously on this pipeline. All
    /// operations **must** be called on the event loop.
    public var syncOperations: SynchronousOperations {
        SynchronousOperations(pipeline: self)
    }
}

@available(*, unavailable)
extension ChannelPipeline.SynchronousOperations: Sendable {}

extension ChannelPipeline {
    /// A `Position` within the `ChannelPipeline` used to insert handlers into the `ChannelPipeline`.
    @preconcurrency
    public enum Position: Sendable {
        /// The first `ChannelHandler` -- the front of the `ChannelPipeline`.
        case first

        /// The last `ChannelHandler` -- the back of the `ChannelPipeline`.
        case last

        /// Before the given `ChannelHandler`.
        case before(ChannelHandler & Sendable)

        /// After the given `ChannelHandler`.
        case after(ChannelHandler & Sendable)
    }
}

extension ChannelPipeline.SynchronousOperations {
    /// A `Position` within the `ChannelPipeline`'s `SynchronousOperations` used to insert non-sendable handlers
    /// into the `ChannelPipeline` at a certain position.
    public enum Position {
        /// The first `ChannelHandler` -- the front of the `ChannelPipeline`.
        case first

        /// The last `ChannelHandler` -- the back of the `ChannelPipeline`.
        case last

        /// Before the given `ChannelHandler`.
        case before(ChannelHandler)

        /// After the given `ChannelHandler`.
        case after(ChannelHandler)

        public init(_ position: ChannelPipeline.Position) {
            switch position {
            case .first:
                self = .first
            case .last:
                self = .last
            case .before(let handler):
                self = .before(handler)
            case .after(let handler):
                self = .after(handler)
            }
        }
    }
}

@available(macOS 10.15, iOS 13, tvOS 13, watchOS 6, *)
@available(*, unavailable)
extension ChannelPipeline.SynchronousOperations.Position: Sendable {}

/// Special `ChannelHandler` that forwards all events to the `Channel.Unsafe` implementation.
final class HeadChannelHandler: _ChannelOutboundHandler, Sendable {

    static let name = "head"
    static let sharedInstance = HeadChannelHandler()

    private init() {}

    func register(context: ChannelHandlerContext, promise: EventLoopPromise<Void>?) {
        context.channel._channelCore.register0(promise: promise)
    }

    func bind(context: ChannelHandlerContext, to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        context.channel._channelCore.bind0(to: address, promise: promise)
    }

    func connect(context: ChannelHandlerContext, to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        context.channel._channelCore.connect0(to: address, promise: promise)
    }

    func write(context: ChannelHandlerContext, data: NIOAny, promise: EventLoopPromise<Void>?) {
        context.channel._channelCore.write0(data, promise: promise)
    }

    func flush(context: ChannelHandlerContext) {
        context.channel._channelCore.flush0()
    }

    func close(context: ChannelHandlerContext, mode: CloseMode, promise: EventLoopPromise<Void>?) {
        context.channel._channelCore.close0(error: mode.error, mode: mode, promise: promise)
    }

    func read(context: ChannelHandlerContext) {
        context.channel._channelCore.read0()
    }

    func triggerUserOutboundEvent(context: ChannelHandlerContext, event: Any, promise: EventLoopPromise<Void>?) {
        context.channel._channelCore.triggerUserOutboundEvent0(event, promise: promise)
    }

}

extension CloseMode {
    /// Returns the error to fail outstanding operations writes with.
    fileprivate var error: any Error {
        switch self {
        case .all:
            return ChannelError._ioOnClosedChannel
        case .output:
            return ChannelError._outputClosed
        case .input:
            return ChannelError._inputClosed
        }
    }
}

/// Special `ChannelInboundHandler` which will consume all inbound events.
final class TailChannelHandler: _ChannelInboundHandler, Sendable {

    static let name = "tail"
    static let sharedInstance = TailChannelHandler()

    private init() {}

    func channelRegistered(context: ChannelHandlerContext) {
        // Discard
    }

    func channelUnregistered(context: ChannelHandlerContext) {
        // Discard
    }

    func channelActive(context: ChannelHandlerContext) {
        // Discard
    }

    func channelInactive(context: ChannelHandlerContext) {
        // Discard
    }

    func channelReadComplete(context: ChannelHandlerContext) {
        // Discard
    }

    func channelWritabilityChanged(context: ChannelHandlerContext) {
        // Discard
    }

    func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
        // Discard
    }

    func errorCaught(context: ChannelHandlerContext, error: Error) {
        context.channel._channelCore.errorCaught0(error: error)
    }

    func channelRead(context: ChannelHandlerContext, data: NIOAny) {
        context.channel._channelCore.channelRead0(data)
    }
}

/// `Error` that is used by the `ChannelPipeline` to inform the user of an error.
public enum ChannelPipelineError: Error {
    /// `ChannelHandler` was already removed.
    case alreadyRemoved
    /// `ChannelHandler` was not found.
    case notFound
}

/// Every `ChannelHandler` has -- when added to a `ChannelPipeline` -- a corresponding `ChannelHandlerContext` which is
/// the way `ChannelHandler`s can interact with other `ChannelHandler`s in the pipeline.
///
/// Most `ChannelHandler`s need to send events through the `ChannelPipeline` which they do by calling the respective
/// method on their `ChannelHandlerContext`. In fact all the `ChannelHandler` default implementations just forward
/// the event using the `ChannelHandlerContext`.
///
/// Many events are instrumental for a `ChannelHandler`'s life-cycle and it is therefore very important to send them
/// at the right point in time. Often, the right behaviour is to react to an event and then forward it to the next
/// `ChannelHandler`.
public final class ChannelHandlerContext: ChannelInvoker {
    // visible for ChannelPipeline to modify
    fileprivate var next: Optional<ChannelHandlerContext>
    fileprivate var prev: Optional<ChannelHandlerContext>

    public let pipeline: ChannelPipeline

    public var channel: Channel {
        self.pipeline.channel
    }

    public var handler: ChannelHandler {
        self.inboundHandler ?? self.outboundHandler!
    }

    public var remoteAddress: SocketAddress? {
        do {
            // Fast-path access to the remoteAddress.
            return try self.channel._channelCore.remoteAddress0()
        } catch ChannelError.ioOnClosedChannel {
            // Channel was closed already but we may still have the address cached so try to access it via the Channel
            // so we are able to use it in channelInactive(...) / handlerRemoved(...) methods.
            return self.channel.remoteAddress
        } catch {
            return nil
        }
    }

    public var localAddress: SocketAddress? {
        do {
            // Fast-path access to the localAddress.
            return try self.channel._channelCore.localAddress0()
        } catch ChannelError.ioOnClosedChannel {
            // Channel was closed already but we may still have the address cached so try to access it via the Channel
            // so we are able to use it in channelInactive(...) / handlerRemoved(...) methods.
            return self.channel.localAddress
        } catch {
            return nil
        }
    }

    public var eventLoop: EventLoop {
        self.pipeline.eventLoop
    }

    public let name: String
    private let inboundHandler: _ChannelInboundHandler?
    private let outboundHandler: _ChannelOutboundHandler?
    private var removeHandlerInvoked = false
    private var userTriggeredRemovalStarted = false

    // Only created from within ChannelPipeline
    fileprivate init(name: String, handler: ChannelHandler, pipeline: ChannelPipeline) {
        self.name = name
        self.pipeline = pipeline
        self.inboundHandler = handler as? _ChannelInboundHandler
        self.outboundHandler = handler as? _ChannelOutboundHandler
        self.next = nil
        self.prev = nil
        precondition(
            self.inboundHandler != nil || self.outboundHandler != nil,
            "ChannelHandlers need to either be inbound or outbound"
        )
    }

    /// Send a `channelRegistered` event to the next (inbound) `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Note: For correct operation it is very important to forward any `channelRegistered` event using this method at the right point in time, that is usually when received.
    public func fireChannelRegistered() {
        self.next?.invokeChannelRegistered()
    }

    /// Send a `channelUnregistered` event to the next (inbound) `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Note: For correct operation it is very important to forward any `channelUnregistered` event using this method at the right point in time, that is usually when received.
    public func fireChannelUnregistered() {
        self.next?.invokeChannelUnregistered()
    }

    /// Send a `channelActive` event to the next (inbound) `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Note: For correct operation it is very important to forward any `channelActive` event using this method at the right point in time, that is often when received.
    public func fireChannelActive() {
        self.next?.invokeChannelActive()
    }

    /// Send a `channelInactive` event to the next (inbound) `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Note: For correct operation it is very important to forward any `channelInactive` event using this method at the right point in time, that is often when received.
    public func fireChannelInactive() {
        self.next?.invokeChannelInactive()
    }

    /// Send data to the next inbound `ChannelHandler`. The data should be of type `ChannelInboundHandler.InboundOut`.
    public func fireChannelRead(_ data: NIOAny) {
        self.next?.invokeChannelRead(data)
    }

    /// Signal to the next `ChannelHandler` that a read burst has finished.
    public func fireChannelReadComplete() {
        self.next?.invokeChannelReadComplete()
    }

    /// Send a `writabilityChanged` event to the next (inbound) `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Note: For correct operation it is very important to forward any `writabilityChanged` event using this method at the right point in time, that is usually when received.
    public func fireChannelWritabilityChanged() {
        self.next?.invokeChannelWritabilityChanged()
    }

    /// Send an error to the next inbound `ChannelHandler`.
    public func fireErrorCaught(_ error: Error) {
        self.next?.invokeErrorCaught(error)
    }

    /// Send a user event to the next inbound `ChannelHandler`.
    ///
    /// This method exists for compatiblity with ``ChannelInboundInvoker``.
    @available(*, deprecated)
    @_disfavoredOverload
    public func fireUserInboundEventTriggered(_ event: Any & Sendable) {
        self.next?.invokeUserInboundEventTriggered(event)
    }

    /// Send a user event to the next inbound `ChannelHandler` from on the event loop.
    public func fireUserInboundEventTriggered(_ event: Any) {
        self.next?.invokeUserInboundEventTriggered(event)
    }

    /// Send a `register` event to the next (outbound) `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Note: For correct operation it is very important to forward any `register` event using this method at the right point in time, that is usually when received.
    public func register(promise: EventLoopPromise<Void>?) {
        if let outboundNext = self.prev {
            outboundNext.invokeRegister(promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    /// Send a `bind` event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    /// When the `bind` event reaches the `HeadChannelHandler` a `ServerSocketChannel` will be bound.
    ///
    /// - Parameters:
    ///   - address: The address to bind to.
    ///   - promise: The promise fulfilled when the socket is bound or failed if it cannot be bound.
    public func bind(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        if let outboundNext = self.prev {
            outboundNext.invokeBind(to: address, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    /// Send a `connect` event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    /// When the `connect` event reaches the `HeadChannelHandler` a `SocketChannel` will be connected.
    ///
    /// - Parameters:
    ///   - address: The address to connect to.
    ///   - promise: The promise fulfilled when the socket is connected or failed if it cannot be connected.
    public func connect(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        if let outboundNext = self.prev {
            outboundNext.invokeConnect(to: address, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    /// Send a `write` event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    /// When the `write` event reaches the `HeadChannelHandler` the data will be enqueued to be written on the next
    /// `flush` event.
    ///
    /// - Parameters:
    ///   - data: The data to write, should be of type `ChannelOutboundHandler.OutboundOut`.
    ///   - promise: The promise fulfilled when the data has been written or failed if it cannot be written.
    public func write(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if let outboundNext = self.prev {
            outboundNext.invokeWrite(data, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    /// Send a `flush` event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    /// When the `flush` event reaches the `HeadChannelHandler` the data previously enqueued will be attempted to be
    /// written to the socket.
    public func flush() {
        if let outboundNext = self.prev {
            outboundNext.invokeFlush()
        }
    }

    /// Send a `write` event followed by a `flush` event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    /// When the `write` event reaches the `HeadChannelHandler` the data will be enqueued to be written when the `flush`
    /// also reaches the `HeadChannelHandler`.
    ///
    /// - Parameters:
    ///   - data: The data to write, should be of type `ChannelOutboundHandler.OutboundOut`.
    ///   - promise: The promise fulfilled when the previously written data been written and flushed or if that failed.
    public func writeAndFlush(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if let outboundNext = self.prev {
            outboundNext.invokeWriteAndFlush(data, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    /// Send a `read` event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    /// When the `read` event reaches the `HeadChannelHandler` the interest to read data will be signalled to the
    /// `Selector`. This will subsequently -- when data becomes readable -- cause `channelRead` events containing the
    /// data being sent through the `ChannelPipeline`.
    public func read() {
        if let outboundNext = self.prev {
            outboundNext.invokeRead()
        }
    }

    /// Send a `close` event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    /// When the `close` event reaches the `HeadChannelHandler` the socket will be closed.
    ///
    /// - Parameters:
    ///   - mode: The `CloseMode` to use.
    ///   - promise: The promise fulfilled when the `Channel` has been closed or failed if it the closing failed.
    public func close(mode: CloseMode = .all, promise: EventLoopPromise<Void>?) {
        if let outboundNext = self.prev {
            outboundNext.invokeClose(mode: mode, promise: promise)
        } else {
            promise?.fail(ChannelError._alreadyClosed)
        }
    }

    /// Send a user event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - event: The user event to send.
    ///   - promise: The promise fulfilled when the user event has been sent or failed if it couldn't be sent.
    @available(*, deprecated)
    @_disfavoredOverload
    public func triggerUserOutboundEvent(_ event: Any & Sendable, promise: EventLoopPromise<Void>?) {
        self._triggerUserOutboundEvent(event, promise: promise)
    }

    /// Send a user event to the next outbound `ChannelHandler` in the `ChannelPipeline`.
    ///
    /// - Parameters:
    ///   - event: The user event to send.
    ///   - promise: The promise fulfilled when the user event has been sent or failed if it couldn't be sent.
    public func triggerUserOutboundEvent(_ event: Any, promise: EventLoopPromise<Void>?) {
        self._triggerUserOutboundEvent(event, promise: promise)
    }

    private func _triggerUserOutboundEvent(_ event: Any, promise: EventLoopPromise<Void>?) {
        if let outboundNext = self.prev {
            outboundNext.invokeTriggerUserOutboundEvent(event, promise: promise)
        } else {
            promise?.fail(ChannelError._ioOnClosedChannel)
        }
    }

    fileprivate func invokeChannelRegistered() {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.channelRegistered(context: self)
        } else {
            self.next?.invokeChannelRegistered()
        }
    }

    fileprivate func invokeChannelUnregistered() {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.channelUnregistered(context: self)
        } else {
            self.next?.invokeChannelUnregistered()
        }
    }

    fileprivate func invokeChannelActive() {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.channelActive(context: self)
        } else {
            self.next?.invokeChannelActive()
        }
    }

    fileprivate func invokeChannelInactive() {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.channelInactive(context: self)
        } else {
            self.next?.invokeChannelInactive()
        }
    }

    fileprivate func invokeChannelRead(_ data: NIOAny) {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.channelRead(context: self, data: data)
        } else {
            self.next?.invokeChannelRead(data)
        }
    }

    fileprivate func invokeChannelReadComplete() {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.channelReadComplete(context: self)
        } else {
            self.next?.invokeChannelReadComplete()
        }
    }

    fileprivate func invokeChannelWritabilityChanged() {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.channelWritabilityChanged(context: self)
        } else {
            self.next?.invokeChannelWritabilityChanged()
        }
    }

    fileprivate func invokeErrorCaught(_ error: Error) {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.errorCaught(context: self, error: error)
        } else {
            self.next?.invokeErrorCaught(error)
        }
    }

    fileprivate func invokeUserInboundEventTriggered(_ event: Any) {
        self.eventLoop.assertInEventLoop()

        if let inboundHandler = self.inboundHandler {
            inboundHandler.userInboundEventTriggered(context: self, event: event)
        } else {
            self.next?.invokeUserInboundEventTriggered(event)
        }
    }

    fileprivate func invokeRegister(promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.register(context: self, promise: promise)
        } else {
            self.prev?.invokeRegister(promise: promise)
        }
    }

    fileprivate func invokeBind(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.bind(context: self, to: address, promise: promise)
        } else {
            self.prev?.invokeBind(to: address, promise: promise)
        }
    }

    fileprivate func invokeConnect(to address: SocketAddress, promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.connect(context: self, to: address, promise: promise)
        } else {
            self.prev?.invokeConnect(to: address, promise: promise)
        }
    }

    fileprivate func invokeWrite(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.write(context: self, data: data, promise: promise)
        } else {
            self.prev?.invokeWrite(data, promise: promise)
        }
    }

    fileprivate func invokeFlush() {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.flush(context: self)
        } else {
            self.prev?.invokeFlush()
        }
    }

    fileprivate func invokeWriteAndFlush(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.write(context: self, data: data, promise: promise)
            outboundHandler.flush(context: self)
        } else {
            self.prev?.invokeWriteAndFlush(data, promise: promise)
        }
    }

    fileprivate func invokeRead() {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.read(context: self)
        } else {
            self.prev?.invokeRead()
        }
    }

    fileprivate func invokeClose(mode: CloseMode, promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.close(context: self, mode: mode, promise: promise)
        } else {
            self.prev?.invokeClose(mode: mode, promise: promise)
        }
    }

    fileprivate func invokeTriggerUserOutboundEvent(_ event: Any, promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()

        if let outboundHandler = self.outboundHandler {
            outboundHandler.triggerUserOutboundEvent(context: self, event: event, promise: promise)
        } else {
            self.prev?.invokeTriggerUserOutboundEvent(event, promise: promise)
        }
    }

    fileprivate func invokeHandlerAdded() {
        self.eventLoop.assertInEventLoop()

        handler.handlerAdded(context: self)
    }

    fileprivate func invokeHandlerRemoved() {
        self.eventLoop.assertInEventLoop()
        guard !self.removeHandlerInvoked else {
            return
        }
        self.removeHandlerInvoked = true

        handler.handlerRemoved(context: self)
    }
}

// This extension "un-deprecates" some parts of the ChannelInvoker API for
// ChannelHandlerContext specifically. These methods were not sound elsewhere,
// but they're fine here.
extension ChannelHandlerContext {
    /// Write data to the remote peer.
    ///
    /// Be aware that to be sure that data is really written to the remote peer you need to call `flush` or use `writeAndFlush`.
    /// Calling `write` multiple times and then `flush` may allow the `Channel` to `write` multiple data objects to the remote peer with one syscall.
    ///
    /// - Parameters:
    ///   - data: the data to write
    ///   - file: The file this function was called in, for debugging purposes.
    ///   - line: The line this function was called on, for debugging purposes.
    /// - Returns: the future which will be notified once the operation completes.
    public func write(_ data: NIOAny, file: StaticString = #fileID, line: UInt = #line) -> EventLoopFuture<Void> {
        let promise = self.eventLoop.makePromise(of: Void.self, file: file, line: line)
        self.write(data, promise: promise)
        return promise.futureResult
    }

    /// Shortcut for calling `write` and `flush`.
    ///
    /// - Parameters:
    ///   - data: the data to write
    ///   - file: The file this function was called in, for debugging purposes.
    ///   - line: The line this function was called on, for debugging purposes.
    /// - Returns: the future which will be notified once the `write` operation completes.
    public func writeAndFlush(_ data: NIOAny, file: StaticString = #fileID, line: UInt = #line) -> EventLoopFuture<Void>
    {
        let promise = self.eventLoop.makePromise(of: Void.self, file: file, line: line)
        self.writeAndFlush(data, promise: promise)
        return promise.futureResult
    }

    /// Returns this `ChannelHandlerContext` as a `NIOLoopBound`, bound to `self.eventLoop`.
    ///
    /// This is a shorthand for `NIOLoopBound(self, eventLoop: self.eventLoop)`.
    ///
    /// Being able to capture `ChannelHandlerContext`s in `EventLoopFuture` callbacks is important in SwiftNIO programs.
    /// Of course, this is not always safe because the `EventLoopFuture` callbacks may run on other threads. SwiftNIO
    /// programmers therefore always had to manually arrange for those callbacks to run on the correct `EventLoop`
    /// (i.e. `context.eventLoop`) which then made that construction safe.
    ///
    /// Newer Swift versions contain a static feature to automatically detect data races which of course can't detect
    /// the only _dynamically_ ``EventLoop`` a ``EventLoopFuture`` callback is running on. ``NIOLoopBound`` can be used
    /// to prove to the compiler that this is safe and in case it is not, ``NIOLoopBound`` will trap at runtime. This is
    /// therefore dynamically enforce the correct behaviour.
    public var loopBound: NIOLoopBound<ChannelHandlerContext> {
        NIOLoopBound(self, eventLoop: self.eventLoop)
    }
}

@available(*, unavailable)
extension ChannelHandlerContext: Sendable {}

extension ChannelHandlerContext {
    /// A `RemovalToken` is handed to a `RemovableChannelHandler` when its `removeHandler` function is invoked. A
    /// `RemovableChannelHandler` is then required to remove itself from the `ChannelPipeline`. The removal process
    /// is finalized by handing the `RemovalToken` to the `ChannelHandlerContext.leavePipeline` function.
    public struct RemovalToken: Sendable {
        internal let promise: EventLoopPromise<Void>?
    }

    /// Synchronously remove the `ChannelHandler` with the given `ChannelHandlerContext`.
    ///
    /// - Note: This function must only be used from a `RemovableChannelHandler` to remove itself. Calling this method
    ///         on any other `ChannelHandlerContext` leads to undefined behaviour.
    ///
    /// - Parameters:
    ///    - removalToken: The removal token received from `RemovableChannelHandler.removeHandler`
    public func leavePipeline(removalToken: RemovalToken) {
        self.eventLoop.preconditionInEventLoop()
        self.pipeline.removeHandlerFromPipeline(context: self, promise: removalToken.promise)
    }

    internal func startUserTriggeredRemoval(promise: EventLoopPromise<Void>?) {
        self.eventLoop.assertInEventLoop()
        guard !self.userTriggeredRemovalStarted else {
            promise?.fail(NIOAttemptedToRemoveHandlerMultipleTimesError())
            return
        }
        self.userTriggeredRemovalStarted = true
        (self.handler as! RemovableChannelHandler).removeHandler(
            context: self,
            removalToken: .init(promise: promise)
        )
    }
}

extension ChannelHandlerContext {
    var sendableView: SendableView {
        SendableView(wrapping: self)
    }

    /// A wrapper over ``ChannelHandlerContext`` that allows access to the thread-safe API
    /// surface on the type.
    ///
    /// Very little of ``ChannelHandlerContext`` is thread-safe, but in a rare few places
    /// there are things we can access. This type makes those available.
    struct SendableView: @unchecked Sendable {
        private let context: ChannelHandlerContext

        fileprivate init(wrapping context: ChannelHandlerContext) {
            self.context = context
        }

        /// Whether the ``ChannelHandler`` associated with this context conforms to
        /// ``RemovableChannelHandler``.
        var channelHandlerIsRemovable: Bool {
            // `context.handler` is not mutable, and set at construction, so this access is
            // acceptable. The protocol conformance check is also safe.
            self.context.handler is RemovableChannelHandler
        }

        /// Grabs the underlying ``ChannelHandlerContext``. May only be called on the
        /// event loop.
        var wrappedValue: ChannelHandlerContext {
            // The event loop lookup here is also thread-safe, so we can grab the value out
            // and use it.
            self.context.eventLoop.preconditionInEventLoop()
            return self.context
        }
    }
}

extension ChannelPipeline: CustomDebugStringConvertible {
    public var debugDescription: String {
        // This method forms output in the following format:
        //
        // ChannelPipeline[0x0000000000000000]:
        //                      [I] ↓↑ [O]
        //  <incoming handler type> ↓↑                         [<name>]
        //                          ↓↑ <outgoing handler type> [<name>]
        //    <duplex handler type> ↓↑ <duplex handler type>   [<name>]
        //
        var desc = ["ChannelPipeline[\(ObjectIdentifier(self))]:"]
        let debugInfos = self.collectHandlerDebugInfos()
        let maxIncomingTypeNameCount =
            debugInfos.filter { $0.isIncoming }
            .map { $0.typeName.count }
            .max() ?? 0
        let maxOutgoingTypeNameCount =
            debugInfos.filter { $0.isOutgoing }
            .map { $0.typeName.count }
            .max() ?? 0

        func whitespace(count: Int) -> String {
            String(repeating: " ", count: count)
        }

        if debugInfos.isEmpty {
            desc.append(" <no handlers>")
        } else {
            desc.append(whitespace(count: maxIncomingTypeNameCount - 2) + "[I] ↓↑ [O]")
            for debugInfo in debugInfos {
                var line = [String]()
                line.append(" ")
                if debugInfo.isIncoming {
                    line.append(whitespace(count: maxIncomingTypeNameCount - debugInfo.typeName.count))
                    line.append(debugInfo.typeName)
                } else {
                    line.append(whitespace(count: maxIncomingTypeNameCount))
                }
                line.append(" ↓↑ ")
                if debugInfo.isOutgoing {
                    line.append(debugInfo.typeName)
                    line.append(whitespace(count: maxOutgoingTypeNameCount - debugInfo.typeName.count))
                } else {
                    line.append(whitespace(count: maxOutgoingTypeNameCount))
                }
                line.append(" ")
                line.append("[\(debugInfo.name)]")
                desc.append(line.joined())
            }
        }

        return desc.joined(separator: "\n")
    }

    /// Returns the first `ChannelHandler` of the given type.
    ///
    /// - Parameters:
    ///   - type: the type of `ChannelHandler` to return.
    @inlinable
    @preconcurrency
    public func handler<Handler: ChannelHandler & _NIOCoreSendableMetatype>(
        type _: Handler.Type
    ) -> EventLoopFuture<Handler> {
        self.context(handlerType: Handler.self).map { context in
            guard let typedContext = context.handler as? Handler else {
                preconditionFailure(
                    "Expected channel handler of type \(Handler.self), got \(type(of: context.handler)) instead."
                )
            }

            return typedContext
        }
    }

    /// Synchronously finds and returns the first `ChannelHandler` of the given type.
    ///
    /// - Important: This must be called on the `EventLoop`.
    /// - Parameters:
    ///   - type: the type of `ChannelHandler` to return.
    @inlinable  // should be fileprivate
    internal func _handlerSync<Handler: ChannelHandler>(type _: Handler.Type) -> Result<Handler, Error> {
        self._contextSync(handlerType: Handler.self).map { context in
            guard let typedContext = context.handler as? Handler else {
                preconditionFailure(
                    "Expected channel handler of type \(Handler.self), got \(type(of: context.handler)) instead."
                )
            }
            return typedContext
        }
    }

    private struct ChannelHandlerDebugInfo {
        let handler: ChannelHandler
        let name: String
        var isIncoming: Bool {
            self.handler is _ChannelInboundHandler
        }
        var isOutgoing: Bool {
            self.handler is _ChannelOutboundHandler
        }
        var typeName: String {
            "\(type(of: self.handler))"
        }
    }

    private func collectHandlerDebugInfos() -> [ChannelHandlerDebugInfo] {
        var handlers = [ChannelHandlerDebugInfo]()
        var node = self.head?.next
        while let context = node, context !== self.tail {
            handlers.append(.init(handler: context.handler, name: context.name))
            node = context.next
        }
        return handlers
    }
}

extension ChannelPipeline {
    private enum BufferingDirection: Equatable {
        case inbound
        case outbound
    }

    /// Retrieve the total number of bytes buffered for outbound.
    public func outboundBufferedBytes() -> EventLoopFuture<Int> {
        let future: EventLoopFuture<Int>

        if self.eventLoop.inEventLoop {
            future = self.eventLoop.makeSucceededFuture(countAllBufferedBytes(direction: .outbound))
        } else {
            future = self.eventLoop.submit {
                self.countAllBufferedBytes(direction: .outbound)
            }
        }

        return future
    }

    /// Retrieve the total number of bytes buffered for inbound.
    public func inboundBufferedBytes() -> EventLoopFuture<Int> {
        let future: EventLoopFuture<Int>

        if self.eventLoop.inEventLoop {
            future = self.eventLoop.makeSucceededFuture(countAllBufferedBytes(direction: .inbound))
        } else {
            future = self.eventLoop.submit {
                self.countAllBufferedBytes(direction: .inbound)
            }
        }

        return future
    }

    private static func countBufferedBytes(context: ChannelHandlerContext, direction: BufferingDirection) -> Int? {
        switch direction {
        case .inbound:
            guard let handler = context.handler as? NIOInboundByteBufferingChannelHandler else {
                return nil
            }
            return handler.inboundBufferedBytes
        case .outbound:
            guard let handler = context.handler as? NIOOutboundByteBufferingChannelHandler else {
                return nil
            }
            return handler.outboundBufferedBytes
        }

    }

    private func countAllBufferedBytes(direction: BufferingDirection) -> Int {
        self.eventLoop.assertInEventLoop()
        var total = 0
        var current = self.head?.next
        switch direction {
        case .inbound:
            while let c = current, c !== self.tail {
                if let inboundHandler = c.handler as? NIOInboundByteBufferingChannelHandler {
                    total += inboundHandler.inboundBufferedBytes
                }
                current = current?.next
            }
        case .outbound:
            while let c = current, c !== self.tail {
                if let outboundHandler = c.handler as? NIOOutboundByteBufferingChannelHandler {
                    total += outboundHandler.outboundBufferedBytes
                }
                current = current?.next
            }
        }

        return total
    }
}

extension ChannelPipeline.SynchronousOperations {
    /// Retrieve the total number of bytes buffered for outbound.
    ///
    /// - Important: This *must* be called on the event loop.
    public func outboundBufferedBytes() -> Int {
        self.eventLoop.assertInEventLoop()
        return self._pipeline.countAllBufferedBytes(direction: .outbound)
    }

    /// Retrieve the number of outbound bytes buffered in the `ChannelHandler` associated with the given`ChannelHandlerContext`.
    ///
    /// - Parameters:
    ///   - context: the `ChannelHandlerContext` from which the outbound buffered bytes of the `ChannelHandler` will be retrieved.
    /// - Important: This *must* be called on the event loop.
    ///
    /// - Returns: The number of bytes currently buffered in the `ChannelHandler` referenced by the `ChannelHandlerContext` parameter `in`.
    ///            If the `ChannelHandler` in the given `ChannelHandlerContext` does not conform to
    ///            `NIOOutboundByteBufferingChannelHandler`, this method will return `nil`.
    public func outboundBufferedBytes(in context: ChannelHandlerContext) -> Int? {
        self.eventLoop.assertInEventLoop()
        return ChannelPipeline.countBufferedBytes(context: context, direction: .outbound)
    }

    /// Retrieve total number of bytes buffered for inbound.
    ///
    /// - Important: This *must* be called on the event loop.
    public func inboundBufferedBytes() -> Int {
        self.eventLoop.assertInEventLoop()
        return self._pipeline.countAllBufferedBytes(direction: .inbound)
    }

    /// Retrieve the number of inbound bytes buffered in the `ChannelHandler` associated with the given `ChannelHandlerContext`.
    ///
    /// - Parameters:
    ///   - context: the `ChannelHandlerContext` from which the inbound buffered bytes of the `handler` will be retrieved.
    /// - Important: This *must* be called on the event loop.
    ///
    /// - Returns: The number of bytes currently buffered in the `ChannelHandler` referenced by the `ChannelHandlerContext` parameter `in`.
    ///            If the `ChannelHandler` in the given `ChannelHandlerContext` does not conform to
    ///            `NIOInboundByteBufferingChannelHandler`, this method will return `nil`.
    public func inboundBufferedBytes(in context: ChannelHandlerContext) -> Int? {
        self.eventLoop.assertInEventLoop()
        return ChannelPipeline.countBufferedBytes(context: context, direction: .inbound)
    }
}

Coverage Report

Created: 2026-03-12 06:14