/src/grpc-swift/Sources/GRPC/GRPCIdleHandler.swift

Source
/*
 * Copyright 2020, gRPC Authors All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
import Logging
import NIOCore
import NIOHTTP2
import NIOTLS
import NIOTransportServices

internal final class GRPCIdleHandler: ChannelInboundHandler {
  typealias InboundIn = HTTP2Frame
  typealias OutboundOut = HTTP2Frame

  /// The amount of time to wait before closing the channel when there are no active streams.
  /// If nil, then we shouldn't schedule idle tasks.
  private let idleTimeout: TimeAmount?

  /// The maximum amount of time the connection is allowed to live before quiescing.
  private let maxAge: TimeAmount?

  /// The ping handler.
  private var pingHandler: PingHandler

  /// The scheduled task which will close the connection gently after the max connection age
  /// has been reached.
  private var scheduledMaxAgeClose: Scheduled<Void>?

  /// The scheduled task which will close the connection after the keep-alive timeout has expired.
  private var scheduledKeepAliveClose: Scheduled<Void>?

  /// The scheduled task which will ping.
  private var scheduledPing: RepeatedTask?

  /// The mode we're operating in.
  private let mode: Mode

  /// The time the handler was created.
  private let creationTime: NIODeadline

  /// Returns the age of the connection in seconds.
  private var connectionAgeInSeconds: UInt64 {
    let now = NIODeadline.now()
    let nanoseconds = now.uptimeNanoseconds - self.creationTime.uptimeNanoseconds
    let seconds = nanoseconds / 1_000_000_000
    return seconds
  }

  private var context: ChannelHandlerContext?

  /// The mode of operation: the client tracks additional connection state in the connection
  /// manager.
  internal enum Mode {
    case client(ConnectionManager, HTTP2StreamMultiplexer)
    case server

    var connectionManager: ConnectionManager? {
      switch self {
      case let .client(manager, _):
        return manager
      case .server:
        return nil
      }
    }
  }

  /// The current state.
  private var stateMachine: GRPCIdleHandlerStateMachine

  init(
    connectionManager: ConnectionManager,
    multiplexer: HTTP2StreamMultiplexer,
    idleTimeout: TimeAmount,
    maxAge: TimeAmount?,
    keepalive configuration: ClientConnectionKeepalive,
    logger: Logger
  ) {
    self.mode = .client(connectionManager, multiplexer)
    switch connectionManager.idleBehavior {
    case .neverGoIdle:
      self.idleTimeout = nil
    case .closeWhenIdleTimeout:
      self.idleTimeout = idleTimeout
    }
    self.stateMachine = .init(role: .client, logger: logger)
    self.pingHandler = PingHandler(
      pingCode: 5,
      interval: configuration.interval,
      timeout: configuration.timeout,
      permitWithoutCalls: configuration.permitWithoutCalls,
      maximumPingsWithoutData: configuration.maximumPingsWithoutData,
      minimumSentPingIntervalWithoutData: configuration.minimumSentPingIntervalWithoutData
    )
    self.creationTime = .now()
    self.maxAge = maxAge
  }

  init(
    idleTimeout: TimeAmount,
    keepalive configuration: ServerConnectionKeepalive,
    logger: Logger
  ) {
    self.mode = .server
    self.stateMachine = .init(role: .server, logger: logger)
    self.idleTimeout = idleTimeout
    self.pingHandler = PingHandler(
      pingCode: 10,
      interval: configuration.interval,
      timeout: configuration.timeout,
      permitWithoutCalls: configuration.permitWithoutCalls,
      maximumPingsWithoutData: configuration.maximumPingsWithoutData,
      minimumSentPingIntervalWithoutData: configuration.minimumSentPingIntervalWithoutData,
      minimumReceivedPingIntervalWithoutData: configuration.minimumReceivedPingIntervalWithoutData,
      maximumPingStrikes: configuration.maximumPingStrikes
    )
    self.creationTime = .now()
    self.maxAge = nil
  }

  private func perform(operations: GRPCIdleHandlerStateMachine.Operations) {
    // Prod the connection manager.
    if let event = operations.connectionManagerEvent, let manager = self.mode.connectionManager {
      switch event {
      case .idle:
        manager.idle()
      case .inactive:
        manager.channelInactive()
      case .ready:
        manager.ready()
      case .quiescing:
        manager.beginQuiescing()
      }
    }

    // Max concurrent streams changed.
    if let manager = self.mode.connectionManager,
      let maxConcurrentStreams = operations.maxConcurrentStreamsChange
    {
      manager.maxConcurrentStreamsChanged(maxConcurrentStreams)
    }

    // Handle idle timeout creation/cancellation.
    if let idleTimeout = self.idleTimeout, let idleTask = operations.idleTask {
      switch idleTask {
      case let .cancel(task):
        self.stateMachine.logger.debug("idle timeout task cancelled")
        task.cancel()

      case .schedule:
        if self.idleTimeout != .nanoseconds(.max), let context = self.context {
          self.stateMachine.logger.debug(
            "scheduling idle timeout task",
            metadata: [MetadataKey.delayMs: "\(idleTimeout.milliseconds)"]
          )
          let task = context.eventLoop.scheduleTask(in: idleTimeout) {
            self.stateMachine.logger.debug("idle timeout task fired")
            self.idleTimeoutFired()
          }
          self.perform(operations: self.stateMachine.scheduledIdleTimeoutTask(task))
        }
      }
    }

    // Send a GOAWAY frame.
    if let streamID = operations.sendGoAwayWithLastPeerInitiatedStreamID {
      self.stateMachine.logger.debug(
        "sending GOAWAY frame",
        metadata: [
          MetadataKey.h2GoAwayLastStreamID: "\(Int(streamID))"
        ]
      )

      let goAwayFrame = HTTP2Frame(
        streamID: .rootStream,
        payload: .goAway(lastStreamID: streamID, errorCode: .noError, opaqueData: nil)
      )

      self.context?.write(self.wrapOutboundOut(goAwayFrame), promise: nil)

      // We emit a ping after some GOAWAY frames.
      if operations.shouldPingAfterGoAway {
        let pingFrame = HTTP2Frame(
          streamID: .rootStream,
          payload: .ping(self.pingHandler.pingDataGoAway, ack: false)
        )
        self.context?.write(self.wrapOutboundOut(pingFrame), promise: nil)
      }

      self.context?.flush()
    }

    // Close the channel, if necessary.
    if operations.shouldCloseChannel, let context = self.context {
      // Close on the next event-loop tick so we don't drop any events which are
      // currently being processed.
      context.eventLoop.execute {
        self.stateMachine.logger.debug(
          "closing connection",
          metadata: ["connection_age_secs": .stringConvertible(self.connectionAgeInSeconds)]
        )
        context.close(mode: .all, promise: nil)
      }
    }
  }

  private func handlePingAction(_ action: PingHandler.Action) {
    switch action {
    case .none:
      ()

    case .ack:
      // NIO's HTTP2 handler acks for us so this is a no-op. Log so it doesn't appear that we are
      // ignoring pings.
      self.stateMachine.logger.debug(
        "sending PING frame",
        metadata: [MetadataKey.h2PingAck: "true"]
      )

    case .cancelScheduledTimeout:
      self.scheduledKeepAliveClose?.cancel()
      self.scheduledKeepAliveClose = nil

    case let .schedulePing(delay, timeout):
      self.schedulePing(in: delay, timeout: timeout)

    case let .reply(framePayload):
      switch framePayload {
      case .ping(_, let ack):
        self.stateMachine.logger.debug(
          "sending PING frame",
          metadata: [MetadataKey.h2PingAck: "\(ack)"]
        )
      default:
        ()
      }
      let frame = HTTP2Frame(streamID: .rootStream, payload: framePayload)
      self.context?.writeAndFlush(self.wrapOutboundOut(frame), promise: nil)

    case .ratchetDownLastSeenStreamID:
      self.perform(operations: self.stateMachine.ratchetDownGoAwayStreamID())
    }
  }

  private func schedulePing(in delay: TimeAmount, timeout: TimeAmount) {
    guard delay != .nanoseconds(.max) else {
      return
    }

    self.stateMachine.logger.debug(
      "scheduled keepalive pings",
      metadata: [MetadataKey.intervalMs: "\(delay.milliseconds)"]
    )

    self.scheduledPing = self.context?.eventLoop.scheduleRepeatedTask(
      initialDelay: delay,
      delay: delay
    ) { _ in
      let action = self.pingHandler.pingFired()
      if case .none = action { return }
      self.handlePingAction(action)
      // `timeout` is less than `interval`, guaranteeing that the close task
      // will be fired before a new ping is triggered.
      assert(timeout < delay, "`timeout` must be less than `interval`")
      self.scheduleClose(in: timeout)
    }
  }

  private func scheduleClose(in timeout: TimeAmount) {
    self.scheduledKeepAliveClose = self.context?.eventLoop.scheduleTask(in: timeout) {
      self.stateMachine.logger.debug("keepalive timer expired")
      self.perform(operations: self.stateMachine.shutdownNow())
    }
  }

  private func idleTimeoutFired() {
    self.perform(operations: self.stateMachine.idleTimeoutTaskFired())
  }

  func handlerAdded(context: ChannelHandlerContext) {
    self.context = context
  }

  func handlerRemoved(context: ChannelHandlerContext) {
    self.context = nil
  }

  func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
    if let created = event as? NIOHTTP2StreamCreatedEvent {
      self.perform(operations: self.stateMachine.streamCreated(withID: created.streamID))
      self.handlePingAction(self.pingHandler.streamCreated())
      self.mode.connectionManager?.streamOpened()
      context.fireUserInboundEventTriggered(event)
    } else if let closed = event as? StreamClosedEvent {
      self.perform(operations: self.stateMachine.streamClosed(withID: closed.streamID))
      self.handlePingAction(self.pingHandler.streamClosed())
      self.mode.connectionManager?.streamClosed()
      context.fireUserInboundEventTriggered(event)
    } else if event is ChannelShouldQuiesceEvent {
      self.perform(operations: self.stateMachine.initiateGracefulShutdown())
      // Swallow this event.
    } else if case let .handshakeCompleted(negotiatedProtocol) = event as? TLSUserEvent {
      let tlsVersion = try? context.channel.getTLSVersionSync()
      self.stateMachine.logger.debug(
        "TLS handshake completed",
        metadata: [
          "alpn": "\(negotiatedProtocol ?? "nil")",
          "tls_version": "\(tlsVersion.map(String.init(describing:)) ?? "nil")",
        ]
      )
      context.fireUserInboundEventTriggered(event)
    } else {
      #if canImport(Network)
      if #available(macOS 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *) {
        if let waitsForConnectivity = event as? NIOTSNetworkEvents.WaitingForConnectivity {
          self.mode.connectionManager?.channelError(waitsForConnectivity.transientError)
        }
      }
      #endif

      context.fireUserInboundEventTriggered(event)
    }
  }

  func errorCaught(context: ChannelHandlerContext, error: Error) {
    // No state machine action here.
    self.mode.connectionManager?.channelError(error)
    context.fireErrorCaught(error)
  }

  func channelActive(context: ChannelHandlerContext) {
    self.stateMachine.logger.addIPAddressMetadata(
      local: context.localAddress,
      remote: context.remoteAddress
    )

    // If a max age has been set then start a timer. This will only be cancelled when it fires or when
    // the channel eventually becomes inactive.
    if let maxAge = self.maxAge {
      assert(self.scheduledMaxAgeClose == nil)
      self.scheduledMaxAgeClose = context.eventLoop.scheduleTask(in: maxAge) {
        let operations = self.stateMachine.reachedMaxAge()
        self.perform(operations: operations)
      }
    }

    // No state machine action here.
    switch self.mode {
    case let .client(connectionManager, multiplexer):
      connectionManager.channelActive(channel: context.channel, multiplexer: multiplexer)
    case .server:
      ()
    }

    context.fireChannelActive()
  }

  func channelInactive(context: ChannelHandlerContext) {
    self.perform(operations: self.stateMachine.channelInactive())
    self.scheduledPing?.cancel()
    self.scheduledKeepAliveClose?.cancel()
    self.scheduledMaxAgeClose?.cancel()
    self.scheduledPing = nil
    self.scheduledKeepAliveClose = nil
    self.scheduledMaxAgeClose = nil
    context.fireChannelInactive()
  }

  func channelRead(context: ChannelHandlerContext, data: NIOAny) {
    let frame = self.unwrapInboundIn(data)

    switch frame.payload {
    case let .goAway(lastStreamID, errorCode, _):
      self.stateMachine.logger.debug(
        "received GOAWAY frame",
        metadata: [
          MetadataKey.h2GoAwayLastStreamID: "\(Int(lastStreamID))",
          MetadataKey.h2GoAwayError: "\(errorCode.networkCode)",
        ]
      )
      self.perform(operations: self.stateMachine.receiveGoAway())
    case let .settings(.settings(settings)):
      self.perform(operations: self.stateMachine.receiveSettings(settings))
    case let .ping(data, ack):
      self.stateMachine.logger.debug(
        "received PING frame",
        metadata: [MetadataKey.h2PingAck: "\(ack)"]
      )
      self.handlePingAction(self.pingHandler.read(pingData: data, ack: ack))
    default:
      // We're not interested in other events.
      ()
    }

    context.fireChannelRead(data)
  }
}

extension HTTP2SettingsParameter {
  internal var loggingMetadataKey: String {
    switch self {
    case .headerTableSize:
      return "h2_settings_header_table_size"
    case .enablePush:
      return "h2_settings_enable_push"
    case .maxConcurrentStreams:
      return "h2_settings_max_concurrent_streams"
    case .initialWindowSize:
      return "h2_settings_initial_window_size"
    case .maxFrameSize:
      return "h2_settings_max_frame_size"
    case .maxHeaderListSize:
      return "h2_settings_max_header_list_size"
    case .enableConnectProtocol:
      return "h2_settings_enable_connect_protocol"
    default:
      return String(describing: self)
    }
  }
}

extension TimeAmount {
  fileprivate var milliseconds: Int64 {
    self.nanoseconds / 1_000_000
  }
}

Coverage Report

Created: 2026-10-23 06:54

Line	Count	Source
1		/*
2		* Copyright 2020, gRPC Authors All rights reserved.
3		*
4		* Licensed under the Apache License, Version 2.0 (the "License");
5		* you may not use this file except in compliance with the License.
6		* You may obtain a copy of the License at
7		*
8		* http://www.apache.org/licenses/LICENSE-2.0
9		*
10		* Unless required by applicable law or agreed to in writing, software
11		* distributed under the License is distributed on an "AS IS" BASIS,
12		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13		* See the License for the specific language governing permissions and
14		* limitations under the License.
15		*/
16		import Logging
17		import NIOCore
18		import NIOHTTP2
19		import NIOTLS
20		import NIOTransportServices
21
22		internal final class GRPCIdleHandler: ChannelInboundHandler {
23		typealias InboundIn = HTTP2Frame
24		typealias OutboundOut = HTTP2Frame
25
26		/// The amount of time to wait before closing the channel when there are no active streams.
27		/// If nil, then we shouldn't schedule idle tasks.
28		private let idleTimeout: TimeAmount?
29
30		/// The maximum amount of time the connection is allowed to live before quiescing.
31		private let maxAge: TimeAmount?
32
33		/// The ping handler.
34		private var pingHandler: PingHandler
35
36		/// The scheduled task which will close the connection gently after the max connection age
37		/// has been reached.
38		private var scheduledMaxAgeClose: Scheduled<Void>?
39
40		/// The scheduled task which will close the connection after the keep-alive timeout has expired.
41		private var scheduledKeepAliveClose: Scheduled<Void>?
42
43		/// The scheduled task which will ping.
44		private var scheduledPing: RepeatedTask?
45
46		/// The mode we're operating in.
47		private let mode: Mode
48
49		/// The time the handler was created.
50		private let creationTime: NIODeadline
51
52		/// Returns the age of the connection in seconds.
53	0	private var connectionAgeInSeconds: UInt64 {
54	0	let now = NIODeadline.now()
55	0	let nanoseconds = now.uptimeNanoseconds - self.creationTime.uptimeNanoseconds
56	0	let seconds = nanoseconds / 1_000_000_000
57	0	return seconds
58	0	}
59
60		private var context: ChannelHandlerContext?
61
62		/// The mode of operation: the client tracks additional connection state in the connection
63		/// manager.
64		internal enum Mode {
65		case client(ConnectionManager, HTTP2StreamMultiplexer)
66		case server
67
68	141k	var connectionManager: ConnectionManager? {
69	141k	switch self {
70	141k	case let .client(manager, _):
71	0	return manager
72	141k	case .server:
73	141k	return nil
74	141k	}
75	141k	}
76		}
77
78		/// The current state.
79		private var stateMachine: GRPCIdleHandlerStateMachine
80
81		init(
82		connectionManager: ConnectionManager,
83		multiplexer: HTTP2StreamMultiplexer,
84		idleTimeout: TimeAmount,
85		maxAge: TimeAmount?,
86		keepalive configuration: ClientConnectionKeepalive,
87		logger: Logger
88	0	) {
89	0	self.mode = .client(connectionManager, multiplexer)
90	0	switch connectionManager.idleBehavior {
91	0	case .neverGoIdle:
92	0	self.idleTimeout = nil
93	0	case .closeWhenIdleTimeout:
94	0	self.idleTimeout = idleTimeout
95	0	}
96	0	self.stateMachine = .init(role: .client, logger: logger)
97	0	self.pingHandler = PingHandler(
98	0	pingCode: 5,
99	0	interval: configuration.interval,
100	0	timeout: configuration.timeout,
101	0	permitWithoutCalls: configuration.permitWithoutCalls,
102	0	maximumPingsWithoutData: configuration.maximumPingsWithoutData,
103	0	minimumSentPingIntervalWithoutData: configuration.minimumSentPingIntervalWithoutData
104	0	)
105	0	self.creationTime = .now()
106	0	self.maxAge = maxAge
107	0	}
108
109		init(
110		idleTimeout: TimeAmount,
111		keepalive configuration: ServerConnectionKeepalive,
112		logger: Logger
113	7.93k	) {
114	7.93k	self.mode = .server
115	7.93k	self.stateMachine = .init(role: .server, logger: logger)
116	7.93k	self.idleTimeout = idleTimeout
117	7.93k	self.pingHandler = PingHandler(
118	7.93k	pingCode: 10,
119	7.93k	interval: configuration.interval,
120	7.93k	timeout: configuration.timeout,
121	7.93k	permitWithoutCalls: configuration.permitWithoutCalls,
122	7.93k	maximumPingsWithoutData: configuration.maximumPingsWithoutData,
123	7.93k	minimumSentPingIntervalWithoutData: configuration.minimumSentPingIntervalWithoutData,
124	7.93k	minimumReceivedPingIntervalWithoutData: configuration.minimumReceivedPingIntervalWithoutData,
125	7.93k	maximumPingStrikes: configuration.maximumPingStrikes
126	7.93k	)
127	7.93k	self.creationTime = .now()
128	7.93k	self.maxAge = nil
129	7.93k	}
130
131	131k	private func perform(operations: GRPCIdleHandlerStateMachine.Operations) {
132	131k	// Prod the connection manager.
133	131k	if let event = operations.connectionManagerEvent, let manager = self.mode.connectionManager {
134	0	switch event {
135	0	case .idle:
136	0	manager.idle()
137	0	case .inactive:
138	0	manager.channelInactive()
139	0	case .ready:
140	0	manager.ready()
141	0	case .quiescing:
142	0	manager.beginQuiescing()
143	0	}
144	131k	}
145	131k
146	131k	// Max concurrent streams changed.
147	131k	if let manager = self.mode.connectionManager,
148	131k	let maxConcurrentStreams = operations.maxConcurrentStreamsChange
149	131k	{
150	0	manager.maxConcurrentStreamsChanged(maxConcurrentStreams)
151	0	}
152	131k
153	131k	// Handle idle timeout creation/cancellation.
154	131k	if let idleTimeout = self.idleTimeout, let idleTask = operations.idleTask {
155	11.9k	switch idleTask {
156	11.9k	case let .cancel(task):
157	0	self.stateMachine.logger.debug("idle timeout task cancelled")
158	0	task.cancel()
159	11.9k
160	11.9k	case .schedule:
161	11.9k	if self.idleTimeout != .nanoseconds(.max), let context = self.context {
162	0	self.stateMachine.logger.debug(
163	0	"scheduling idle timeout task",
164	0	metadata: [MetadataKey.delayMs: "\(idleTimeout.milliseconds)"]
165	0	)
166	0	let task = context.eventLoop.scheduleTask(in: idleTimeout) {
167	0	self.stateMachine.logger.debug("idle timeout task fired")
168	0	self.idleTimeoutFired()
169	0	}
170	0	self.perform(operations: self.stateMachine.scheduledIdleTimeoutTask(task))
171	0	}
172	11.9k	}
173	131k	}
174	131k
175	131k	// Send a GOAWAY frame.
176	131k	if let streamID = operations.sendGoAwayWithLastPeerInitiatedStreamID {
177	10.6k	self.stateMachine.logger.debug(
178	10.6k	"sending GOAWAY frame",
179	10.6k	metadata: [
180	0	MetadataKey.h2GoAwayLastStreamID: "\(Int(streamID))"
181	0	]
182	10.6k	)
183	10.6k
184	10.6k	let goAwayFrame = HTTP2Frame(
185	10.6k	streamID: .rootStream,
186	10.6k	payload: .goAway(lastStreamID: streamID, errorCode: .noError, opaqueData: nil)
187	10.6k	)
188	10.6k
189	10.6k	self.context?.write(self.wrapOutboundOut(goAwayFrame), promise: nil)
190	10.6k
191	10.6k	// We emit a ping after some GOAWAY frames.
192	10.6k	if operations.shouldPingAfterGoAway {
193	1.49k	let pingFrame = HTTP2Frame(
194	1.49k	streamID: .rootStream,
195	1.49k	payload: .ping(self.pingHandler.pingDataGoAway, ack: false)
196	1.49k	)
197	1.49k	self.context?.write(self.wrapOutboundOut(pingFrame), promise: nil)
198	1.49k	}
199	10.6k
200	10.6k	self.context?.flush()
201	10.6k	}
202	131k
203	131k	// Close the channel, if necessary.
204	131k	if operations.shouldCloseChannel, let context = self.context {
205	1.30k	// Close on the next event-loop tick so we don't drop any events which are
206	1.30k	// currently being processed.
207	1.30k	context.eventLoop.execute {
208	1.30k	self.stateMachine.logger.debug(
209	1.30k	"closing connection",
210	1.30k	metadata: ["connection_age_secs": .stringConvertible(self.connectionAgeInSeconds)]
211	1.30k	)
212	1.30k	context.close(mode: .all, promise: nil)
213	1.30k	}
214	1.30k	}
215	131k	}
216
217	60.0k	private func handlePingAction(_ action: PingHandler.Action) {
218	60.0k	switch action {
219	60.0k	case .none:
220	43.8k	()
221	60.0k
222	60.0k	case .ack:
223	2.32k	// NIO's HTTP2 handler acks for us so this is a no-op. Log so it doesn't appear that we are
224	2.32k	// ignoring pings.
225	2.32k	self.stateMachine.logger.debug(
226	2.32k	"sending PING frame",
227	2.32k	metadata: [MetadataKey.h2PingAck: "true"]
228	2.32k	)
229	60.0k
230	60.0k	case .cancelScheduledTimeout:
231	627	self.scheduledKeepAliveClose?.cancel()
232	627	self.scheduledKeepAliveClose = nil
233	60.0k
234	60.0k	case let .schedulePing(delay, timeout):
235	4.55k	self.schedulePing(in: delay, timeout: timeout)
236	60.0k
237	60.0k	case let .reply(framePayload):
238	0	switch framePayload {
239	0	case .ping(_, let ack):
240	0	self.stateMachine.logger.debug(
241	0	"sending PING frame",
242	0	metadata: [MetadataKey.h2PingAck: "\(ack)"]
243	0	)
244	0	default:
245	0	()
246	0	}
247	0	let frame = HTTP2Frame(streamID: .rootStream, payload: framePayload)
248	0	self.context?.writeAndFlush(self.wrapOutboundOut(frame), promise: nil)
249	60.0k
250	60.0k	case .ratchetDownLastSeenStreamID:
251	8.62k	self.perform(operations: self.stateMachine.ratchetDownGoAwayStreamID())
252	60.0k	}
253	60.0k	}
254
255	4.55k	private func schedulePing(in delay: TimeAmount, timeout: TimeAmount) {
256	4.55k	guard delay != .nanoseconds(.max) else {
257	0	return
258	4.55k	}
259	4.55k
260	4.55k	self.stateMachine.logger.debug(
261	4.55k	"scheduled keepalive pings",
262	4.55k	metadata: [MetadataKey.intervalMs: "\(delay.milliseconds)"]
263	4.55k	)
264	4.55k
265	4.55k	self.scheduledPing = self.context?.eventLoop.scheduleRepeatedTask(
266	4.55k	initialDelay: delay,
267	4.55k	delay: delay
268	4.55k	) { _ in
269	0	let action = self.pingHandler.pingFired()
270	0	if case .none = action { return }
271	0	self.handlePingAction(action)
272	0	// `timeout` is less than `interval`, guaranteeing that the close task
273	0	// will be fired before a new ping is triggered.
274	0	assert(timeout < delay, "`timeout` must be less than `interval`")
275	0	self.scheduleClose(in: timeout)
276	0	}
277	4.55k	}
278
279	0	private func scheduleClose(in timeout: TimeAmount) {
280	0	self.scheduledKeepAliveClose = self.context?.eventLoop.scheduleTask(in: timeout) {
281	0	self.stateMachine.logger.debug("keepalive timer expired")
282	0	self.perform(operations: self.stateMachine.shutdownNow())
283	0	}
284	0	}
285
286	0	private func idleTimeoutFired() {
287	0	self.perform(operations: self.stateMachine.idleTimeoutTaskFired())
288	0	}
289
290	7.93k	func handlerAdded(context: ChannelHandlerContext) {
291	7.93k	self.context = context
292	7.93k	}
293
294	7.93k	func handlerRemoved(context: ChannelHandlerContext) {
295	7.93k	self.context = nil
296	7.93k	}
297
298	451k	func userInboundEventTriggered(context: ChannelHandlerContext, event: Any) {
299	451k	if let created = event as? NIOHTTP2StreamCreatedEvent {
300	16.6k	self.perform(operations: self.stateMachine.streamCreated(withID: created.streamID))
301	16.6k	self.handlePingAction(self.pingHandler.streamCreated())
302	16.6k	self.mode.connectionManager?.streamOpened()
303	16.6k	context.fireUserInboundEventTriggered(event)
304	16.6k	} else if let closed = event as? StreamClosedEvent {
305	11.5k	self.perform(operations: self.stateMachine.streamClosed(withID: closed.streamID))
306	11.5k	self.handlePingAction(self.pingHandler.streamClosed())
307	11.5k	self.mode.connectionManager?.streamClosed()
308	11.5k	context.fireUserInboundEventTriggered(event)
309	423k	} else if event is ChannelShouldQuiesceEvent {
310	0	self.perform(operations: self.stateMachine.initiateGracefulShutdown())
311	0	// Swallow this event.
312	0	} else if case let .handshakeCompleted(negotiatedProtocol) = event as? TLSUserEvent {
313	0	let tlsVersion = try? context.channel.getTLSVersionSync()
314	0	self.stateMachine.logger.debug(
315	0	"TLS handshake completed",
316	0	metadata: [
317	0	"alpn": "\(negotiatedProtocol ?? "nil")",
318	0	"tls_version": "\(tlsVersion.map(String.init(describing:)) ?? "nil")",
319	0	]
320	0	)
321	0	context.fireUserInboundEventTriggered(event)
322	423k	} else {
323		#if canImport(Network)
324		if #available(macOS 10.14, iOS 12.0, tvOS 12.0, watchOS 6.0, *) {
325		if let waitsForConnectivity = event as? NIOTSNetworkEvents.WaitingForConnectivity {
326		self.mode.connectionManager?.channelError(waitsForConnectivity.transientError)
327		}
328		}
329		#endif
330	423k
331	423k	context.fireUserInboundEventTriggered(event)
332	423k	}
333	451k	}
334
335	24.5k	func errorCaught(context: ChannelHandlerContext, error: Error) {
336	24.5k	// No state machine action here.
337	24.5k	self.mode.connectionManager?.channelError(error)
338	24.5k	context.fireErrorCaught(error)
339	24.5k	}
340
341	0	func channelActive(context: ChannelHandlerContext) {
342	0	self.stateMachine.logger.addIPAddressMetadata(
343	0	local: context.localAddress,
344	0	remote: context.remoteAddress
345	0	)
346	0
347	0	// If a max age has been set then start a timer. This will only be cancelled when it fires or when
348	0	// the channel eventually becomes inactive.
349	0	if let maxAge = self.maxAge {
350	0	assert(self.scheduledMaxAgeClose == nil)
351	0	self.scheduledMaxAgeClose = context.eventLoop.scheduleTask(in: maxAge) {
352	0	let operations = self.stateMachine.reachedMaxAge()
353	0	self.perform(operations: operations)
354	0	}
355	0	}
356	0
357	0	// No state machine action here.
358	0	switch self.mode {
359	0	case let .client(connectionManager, multiplexer):
360	0	connectionManager.channelActive(channel: context.channel, multiplexer: multiplexer)
361	0	case .server:
362	0	()
363	0	}
364	0
365	0	context.fireChannelActive()
366	0	}
367
368	7.93k	func channelInactive(context: ChannelHandlerContext) {
369	7.93k	self.perform(operations: self.stateMachine.channelInactive())
370	7.93k	self.scheduledPing?.cancel()
371	7.93k	self.scheduledKeepAliveClose?.cancel()
372	7.93k	self.scheduledMaxAgeClose?.cancel()
373	7.93k	self.scheduledPing = nil
374	7.93k	self.scheduledKeepAliveClose = nil
375	7.93k	self.scheduledMaxAgeClose = nil
376	7.93k	context.fireChannelInactive()
377	7.93k	}
378
379	262k	func channelRead(context: ChannelHandlerContext, data: NIOAny) {
380	262k	let frame = self.unwrapInboundIn(data)
381	262k
382	262k	switch frame.payload {
383	262k	case let .goAway(lastStreamID, errorCode, _):
384	20.6k	self.stateMachine.logger.debug(
385	20.6k	"received GOAWAY frame",
386	20.6k	metadata: [
387	0	MetadataKey.h2GoAwayLastStreamID: "\(Int(lastStreamID))",
388	0	MetadataKey.h2GoAwayError: "\(errorCode.networkCode)",
389	0	]
390	20.6k	)
391	20.6k	self.perform(operations: self.stateMachine.receiveGoAway())
392	262k	case let .settings(.settings(settings)):
393	13.1k	self.perform(operations: self.stateMachine.receiveSettings(settings))
394	262k	case let .ping(data, ack):
395	5.03k	self.stateMachine.logger.debug(
396	5.03k	"received PING frame",
397	5.03k	metadata: [MetadataKey.h2PingAck: "\(ack)"]
398	5.03k	)
399	5.03k	self.handlePingAction(self.pingHandler.read(pingData: data, ack: ack))
400	262k	default:
401	223k	// We're not interested in other events.
402	223k	()
403	262k	}
404	262k
405	262k	context.fireChannelRead(data)
406	262k	}
407		}
408
409		extension HTTP2SettingsParameter {
410	0	internal var loggingMetadataKey: String {
411	0	switch self {
412	0	case .headerTableSize:
413	0	return "h2_settings_header_table_size"
414	0	case .enablePush:
415	0	return "h2_settings_enable_push"
416	0	case .maxConcurrentStreams:
417	0	return "h2_settings_max_concurrent_streams"
418	0	case .initialWindowSize:
419	0	return "h2_settings_initial_window_size"
420	0	case .maxFrameSize:
421	0	return "h2_settings_max_frame_size"
422	0	case .maxHeaderListSize:
423	0	return "h2_settings_max_header_list_size"
424	0	case .enableConnectProtocol:
425	0	return "h2_settings_enable_connect_protocol"
426	0	default:
427	0	return String(describing: self)
428	0	}
429	0	}
430		}
431
432		extension TimeAmount {
433	0	fileprivate var milliseconds: Int64 {
434	0	self.nanoseconds / 1_000_000
435	0	}
436		}