ChannelPipeline.swift

//===----------------------------------------------------------------------===//
//
// This source file is part of the SwiftNIO open source project
//
// Copyright (c) 2017-2018 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.
    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.
    public func addHandler(_ handler: ChannelHandler,
                           name: String? = nil,
                           position: ChannelPipeline.Position = .last) -> EventLoopFuture<Void> {
        let future: EventLoopFuture<Void>

        if self.eventLoop.inEventLoop {
            future = self.eventLoop.makeCompletedFuture(self.addHandlerSync(handler, name: name, position: position))
        } else {
            future = self.eventLoop.submit {
                try self.addHandlerSync(handler, name: name, position: position).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.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.
    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:
    ///     - 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.
    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.
    public func removeHandler(_ handler: RemovableChannelHandler, promise: EventLoopPromise<Void>?) {
        func removeHandler0() {
            switch self.contextSync(handler: handler) {
            case .success(let context):
                self.removeHandler(context: context, promise: promise)
            case .failure(let error):
                promise?.fail(error)
            }
        }

        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>?) {
        func removeHandler0() {
            switch self.contextSync(name: name) {
            case .success(let context):
                self.removeHandler(context: context, promise: promise)
            case .failure(let error):
                promise?.fail(error)
            }
        }

        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.
    public func removeHandler(context: ChannelHandlerContext, promise: EventLoopPromise<Void>?) {
        guard context.handler is RemovableChannelHandler else {
            promise?.fail(ChannelError.unremovableHandler)
            return
        }
        func removeHandler0() {
            context.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.
    public func context(handler: ChannelHandler) -> EventLoopFuture<ChannelHandlerContext> {
        let promise = self.eventLoop.makePromise(of: ChannelHandlerContext.self)

        if self.eventLoop.inEventLoop {
            promise.completeWith(self.contextSync(handler: handler))
        } else {
            self.eventLoop.execute {
                promise.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> {
        return 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.completeWith(self.contextSync(name: name))
        } else {
            self.eventLoop.execute {
                promise.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> {
        return 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
    public func context<Handler: ChannelHandler>(handlerType: Handler.Type) -> EventLoopFuture<ChannelHandlerContext> {
        let promise = self.eventLoop.makePromise(of: ChannelHandlerContext.self)

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

        return promise.futureResult
    }

    /// 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> {
        return 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()
            }
        }
    }

    public func fireChannelRead(_ data: NIOAny) {
        if eventLoop.inEventLoop {
            fireChannelRead0(data)
        } else {
            eventLoop.execute {
                self.fireChannelRead0(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()
            }
        }
    }

    public func fireUserInboundEventTriggered(_ event: Any) {
        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()
            }
        }
    }

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

    public func writeAndFlush(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if eventLoop.inEventLoop {
            writeAndFlush0(data, promise: promise)
        } else {
            eventLoop.execute {
                self.writeAndFlush0(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)
            }
        }
    }

    public func triggerUserOutboundEvent(_ event: Any, 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? {
        return self.tail?.prev
    }

    private var firstInboundCtx: ChannelHandlerContext? {
        return 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()
        }
    }

    private func write0(_ data: NIOAny, promise: EventLoopPromise<Void>?) {
        if let firstOutboundCtx = firstOutboundCtx {
            firstOutboundCtx.invokeWrite(data, promise: promise)
        } else {
            promise?.fail(ChannelError.ioOnClosedChannel)
        }
    }

    private 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()
        }
    }

    private 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 {
        return 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
    }
}

#if swift(>=5.7)
extension ChannelPipeline: @unchecked Sendable {}
#endif

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.
    public func addHandlers(_ handlers: [ChannelHandler],
                            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.
    public func addHandlers(_ handlers: ChannelHandler...,
                            position: ChannelPipeline.Position = .last) -> EventLoopFuture<Void> {
        return 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],
                                     position: ChannelPipeline.Position) -> Result<Void, Error> {
        switch position {
        case .first, .after:
            return self._addHandlersSync(handlers.reversed(), position: position)
        case .last, .before:
            return self._addHandlersSync(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.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 {
            return 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.Position = .last) throws {
            try self._pipeline.addHandlerSync(handler, name: name, 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`.
        public func addHandlers(_ handlers: [ChannelHandler],
                                position: ChannelPipeline.Position = .last) throws {
            try self._pipeline.addHandlersSync(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`.
        public func addHandlers(_ handlers: ChannelHandler...,
                                position: ChannelPipeline.Position = .last) throws {
            try self._pipeline.addHandlersSync(handlers, position: position).get()
        }

        /// 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 {
            return 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 {
            return 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 name: The name of the handler whose context is being fetched.
        /// - Returns: The `ChannelHandlerContext` associated with the handler.
        @inlinable
        public func context<Handler: ChannelHandler>(handlerType: Handler.Type) throws -> ChannelHandlerContext {
            return 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 {
            return 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 `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 `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)
        }
    }

    /// 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 {
        return SynchronousOperations(pipeline: self)
    }
}

#if swift(>=5.7)
@available(*, unavailable)
extension ChannelPipeline.SynchronousOperations: Sendable {}
#endif

extension ChannelPipeline {
    /// A `Position` within the `ChannelPipeline` used to insert handlers into the `ChannelPipeline`.
    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)
    }
}

#if swift(>=5.7)
@available(*, unavailable)
extension ChannelPipeline.Position: Sendable {}
#endif

/// Special `ChannelHandler` that forwards all events to the `Channel.Unsafe` implementation.
/* private but tests */ final class HeadChannelHandler: _ChannelOutboundHandler {

    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)
    }

}

private extension CloseMode {
    /// Returns the error to fail outstanding operations writes with.
    var error: ChannelError {
        switch self {
        case .all:
            return .ioOnClosedChannel
        case .output:
            return .outputClosed
        case .input:
            return .inputClosed
        }
    }
}

/// Special `ChannelInboundHandler` which will consume all inbound events.
/* private but tests */ final class TailChannelHandler: _ChannelInboundHandler {

    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 {
        return self.pipeline.channel
    }

    public var handler: ChannelHandler {
        return 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 {
        return 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`.
    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.
    ///
    /// - parameters:
    ///     - promise: The promise fulfilled when the previously written data been flushed or failed if it cannot be flushed.
    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.
    public 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)
    }
}

#if swift(>=5.7)
@available(*, unavailable)
extension ChannelHandlerContext: Sendable {}
#endif

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 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 {
            return 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
    public func handler<Handler: ChannelHandler>(type _: Handler.Type) -> EventLoopFuture<Handler> {
        return 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> {
        return 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 {
            return self.handler is _ChannelInboundHandler
        }
        var isOutgoing: Bool {
            return self.handler is _ChannelOutboundHandler
        }
        var typeName: String {
            return "\(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
    }
}