mirror of
https://github.com/apple/swift-nio.git
synced 2026-05-20 20:30:36 +00:00
b0e024792a
## Motivation NIO channels abstract over an underlying transport mechanisms (sockets, pipes, etc.), which users typically need not interact with. However, there are scenarios where users need direct access to the underlying transport for low-level operations that work outside NIOs abstraction. One example is performing out out-of-band operations on the underlying file descriptor for a socket-based channel. This PR adds a structured way to access the underlying transport of a channel, for channels that choose to implement it. ## Modifications - Add a new `public protocol NIOTransportAccessibleChannelCore<Transport>` which provides a scoped `withUnsafeTransport(_:)`, used for channel implementations to opt-in. - Add conformance to `NIOTransportAccessibleChannel<NIOBSDSocket.Handle>` for `BaseSocketChannel`, to make this API available for all socket-based channels, including channels returned from the socket-based bootstrap public APIs. - Add public API `ChannelPipeline.SynchronousOperations.withUnsafeTransportIfAvailable(_:)` Note that not all channels need to or should their transport, which is why this was added as an additional protocol that refines `ChannelCore`, vs. extending `Channel` or `ChannelCore` with a default implementation. The protocol uses a primary associated type allowing channels to provide typed access to the transport. E.g. this could be used in NIO Transport Services to expose the underlying `NWConnection`, if desired. The method itself is spelled with "unsafe", uses scoped access, and has clear documentation that users must not violated any of NIOs assumptions about the state of the underlying transport. It's very much not intended for every day use. It being on `ChannelCore` should defer users of the low-level API, since `Channel._channelCore` is marked as for NIO internal use, and the public `withUnsafeTransportIfAvailable(_:)` will take care of the runtime checks for channels that have opted into this API. ## Result - New opt-in API for channel implementations to expose their underlying transport - All socket-based channels from NIOPosix now expose the underlying socket file descriptor - New API `ChannelPipeline.SynchronousOperations.withUnsafeTransportIfAvailable(_:)` for users --------- Co-authored-by: Agam Dua <agam_dua@apple.com>
2679 lines
107 KiB
Swift
2679 lines
107 KiB
Swift
//===----------------------------------------------------------------------===//
|
|
//
|
|
// 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)
|
|
}
|
|
}
|