pkgs/http2/lib/src/connection.dart - http.git - Git at Google

 // Copyright (c) 2015, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.

 import 'dart:async';
 import 'dart:convert' show utf8;

 import '../transport.dart';
 import 'connection_preface.dart';
 import 'flowcontrol/connection_queues.dart';
 import 'flowcontrol/queue_messages.dart';
 import 'flowcontrol/window.dart';
 import 'flowcontrol/window_handler.dart';
 import 'frames/frame_defragmenter.dart';
 import 'frames/frames.dart';
 import 'hpack/hpack.dart';
 import 'ping/ping_handler.dart';
 import 'settings/settings.dart';
 import 'streams/stream_handler.dart';
 import 'sync_errors.dart';

 class ConnectionState {
   /// The connection has been established, we're waiting for the settings frame
   /// of the remote end.
   static const int Initialized = 1;

   /// The connection has been established and is fully operational.
   static const int Operational = 2;

   /// The connection is no longer accepting new streams or creating new streams.
   static const int Finishing = 3;

   /// The connection has been terminated and cannot be used anymore.
   static const int Terminated = 4;

   /// Whether we actively were finishing the connection.
   static const int FinishingActive = 1;

   /// Whether we passively were finishing the connection.
   static const int FinishingPassive = 2;

   int state = Initialized;
   int finishingState = 0;

   ConnectionState();

   bool get isInitialized => state == ConnectionState.Initialized;

   bool get isOperational => state == ConnectionState.Operational;

   bool get isFinishing => state == ConnectionState.Finishing;

   bool get isTerminated => state == ConnectionState.Terminated;

   bool get activeFinishing =>
       state == Finishing && (finishingState & FinishingActive) != 0;

   bool get passiveFinishing =>
       state == Finishing && (finishingState & FinishingPassive) != 0;

   @override
   String toString() {
     var message = '';

     void add(bool condition, String flag) {
       if (condition) {
         if (message.isEmpty) {
           message = flag;
         } else {
           message = '$message/$flag';
         }
       }
     }

     add(isInitialized, 'Initialized');
     add(isOperational, 'IsOperational');
     add(isFinishing, 'IsFinishing');
     add(isTerminated, 'IsTerminated');
     add(activeFinishing, 'ActiveFinishing');
     add(passiveFinishing, 'PassiveFinishing');

     return message;
   }
 }

 abstract class Connection {
   /// The settings the other end has acknowledged to use when communicating with
   /// us.
   final ActiveSettings acknowledgedSettings = ActiveSettings();

   /// The settings we have to obey communicating with the other side.
   final ActiveSettings peerSettings = ActiveSettings();

   /// Whether this connection is a client connection.
   final bool isClientConnection;

   /// Active state handler for this connection.
   ActiveStateHandler? onActiveStateChanged;

   final Completer<void> _onInitialPeerSettingsReceived = Completer<void>();

   final StreamController<int> _pingReceived = StreamController<int>();

   final StreamController<void> _frameReceived = StreamController<void>();

   /// Future which completes when the first SETTINGS frame is received from
   /// the peer.
   Future<void> get onInitialPeerSettingsReceived =>
       _onInitialPeerSettingsReceived.future;

   /// The HPack context for this connection.
   final HPackContext _hpackContext = HPackContext();

   /// The flow window for this connection of the peer.
   final Window _peerWindow = Window();

   /// The flow window for this connection of this end.
   final Window _localWindow = Window();

   /// Used for defragmenting PushPromise/Header frames.
   final FrameDefragmenter _defragmenter = FrameDefragmenter();

   /// The outgoing frames of this connection;
   late FrameWriter _frameWriter;

   /// A subscription of incoming [Frame]s.
   late StreamSubscription<Frame> _frameReaderSubscription;

   /// The incoming connection-level message queue.
   late ConnectionMessageQueueIn _incomingQueue;

   /// The outgoing connection-level message queue.
   late ConnectionMessageQueueOut _outgoingQueue;

   /// The ping handler used for making pings & handling remote pings.
   late PingHandler _pingHandler;

   /// The settings handler used for changing settings & for handling remote
   /// setting changes.
   late SettingsHandler _settingsHandler;

   /// The set of active streams this connection has.
   late StreamHandler _streams;

   /// The connection-level flow control window handler for outgoing messages.
   late OutgoingConnectionWindowHandler _connectionWindowHandler;

   /// The state of this connection.
   late ConnectionState _state;

   Connection(Stream<List<int>> incoming, StreamSink<List<int>> outgoing,
       Settings settings,
       {this.isClientConnection = true}) {
     _setupConnection(incoming, outgoing, settings);
   }

   /// Runs all setup necessary before new streams can be created with the remote
   /// peer.
   void _setupConnection(Stream<List<int>> incoming,
       StreamSink<List<int>> outgoing, Settings settingsObject) {
     // Setup frame reading.
     var incomingFrames =
         FrameReader(incoming, acknowledgedSettings).startDecoding();
     _frameReaderSubscription = incomingFrames.listen((Frame frame) {
       _catchProtocolErrors(() => _handleFrameImpl(frame));
     }, onError: (error, stack) {
       _terminate(ErrorCode.CONNECT_ERROR, causedByTransportError: true);
     }, onDone: () {
       // Ensure existing messages from lower levels are sent to the upper
       // levels before we terminate everything.
       _incomingQueue.forceDispatchIncomingMessages();
       _streams.forceDispatchIncomingMessages();

       _terminate(ErrorCode.CONNECT_ERROR, causedByTransportError: true);
     });

     // Setup frame writing.
     _frameWriter = FrameWriter(_hpackContext.encoder, outgoing, peerSettings);
     _frameWriter.doneFuture.whenComplete(() {
       _terminate(ErrorCode.CONNECT_ERROR, causedByTransportError: true);
     });

     // Setup handlers.
     _settingsHandler = SettingsHandler(_hpackContext.encoder, _frameWriter,
         acknowledgedSettings, peerSettings);
     _pingHandler = PingHandler(_frameWriter, _pingReceived);

     var settings = _decodeSettings(settingsObject);

     // Do the initial settings handshake (possibly with pushes disabled).
     _settingsHandler.changeSettings(settings).catchError((Object error) {
       // TODO: The [error] can contain sensitive information we now expose via
       // a [Goaway] frame. We should somehow ensure we're only sending useful
       // but non-sensitive information.
       _terminate(ErrorCode.PROTOCOL_ERROR,
           message: 'Failed to set initial settings (error: $error).');
     });

     _settingsHandler.onInitialWindowSizeChange.listen((int difference) {
       _catchProtocolErrors(() {
         _streams.processInitialWindowSizeSettingChange(difference);
       });
     });

     // Setup the connection window handler, which keeps track of the
     // size of the outgoing connection window.
     _connectionWindowHandler = OutgoingConnectionWindowHandler(_peerWindow);

     var connectionWindowUpdater =
         IncomingWindowHandler.connection(_frameWriter, _localWindow);

     // Setup queues for outgoing/incoming messages on the connection level.
     _outgoingQueue =
         ConnectionMessageQueueOut(_connectionWindowHandler, _frameWriter);
     _incomingQueue =
         ConnectionMessageQueueIn(connectionWindowUpdater, _catchProtocolErrors);

     if (isClientConnection) {
       _streams = StreamHandler.client(
           _frameWriter,
           _incomingQueue,
           _outgoingQueue,
           _settingsHandler.peerSettings,
           _settingsHandler.acknowledgedSettings,
           _activeStateHandler);
     } else {
       _streams = StreamHandler.server(
           _frameWriter,
           _incomingQueue,
           _outgoingQueue,
           _settingsHandler.peerSettings,
           _settingsHandler.acknowledgedSettings,
           _activeStateHandler);
     }

     // NOTE: We're not waiting until initial settings have been exchanged
     // before we start using the connection (i.e. we don't wait for half a
     // round-trip-time).
     _state = ConnectionState();
   }

   List<Setting> _decodeSettings(Settings settings) {
     var settingsList = <Setting>[];

     // By default a endpoint can make an unlimited number of concurrent streams.
     var concurrentStreamLimit = settings.concurrentStreamLimit;
     if (concurrentStreamLimit != null) {
       settingsList.add(Setting(
           Setting.SETTINGS_MAX_CONCURRENT_STREAMS, concurrentStreamLimit));
     }

     // By default the stream level flow control window is 64 KiB.
     var streamWindowSize = settings.streamWindowSize;
     if (streamWindowSize != null) {
       settingsList
           .add(Setting(Setting.SETTINGS_INITIAL_WINDOW_SIZE, streamWindowSize));
     }

     if (settings is ClientSettings) {
       // By default the server is allowed to do server pushes.
       if (!settings.allowServerPushes) {
         settingsList.add(Setting(Setting.SETTINGS_ENABLE_PUSH, 0));
       }
     } else if (settings is ServerSettings) {
       // No special server settings at the moment.
     } else {
       assert(false);
     }

     return settingsList;
   }

   /// Pings the remote peer (can e.g. be used for measuring latency).
   Future<void> ping() {
     return _pingHandler.ping().catchError((e, s) {
       return Future<void>.error(
           TransportException('The connection has been terminated.'));
     }, test: (e) => e is TerminatedException);
   }

   /// Finishes this connection.
   Future<void> finish() {
     _finishing(active: true);

     // TODO: There is probably more we need to wait for.
     return _streams.done.whenComplete(() =>
         Future.wait([_frameWriter.close(), _frameReaderSubscription.cancel()]));
   }

   /// Terminates this connection forcefully.
   Future<void> terminate([int? errorCode]) {
     return _terminate(errorCode ?? ErrorCode.NO_ERROR);
   }

   void _activeStateHandler(bool isActive) =>
       onActiveStateChanged?.call(isActive);

   /// Invokes the passed in closure and catches any exceptions.
   void _catchProtocolErrors(void Function() fn) {
     try {
       fn();
     } on ProtocolException catch (error) {
       _terminate(ErrorCode.PROTOCOL_ERROR, message: '$error');
     } on FlowControlException catch (error) {
       _terminate(ErrorCode.FLOW_CONTROL_ERROR, message: '$error');
     } on FrameSizeException catch (error) {
       _terminate(ErrorCode.FRAME_SIZE_ERROR, message: '$error');
     } on HPackDecodingException catch (error) {
       _terminate(ErrorCode.PROTOCOL_ERROR, message: '$error');
     } on TerminatedException {
       // We tried to perform an action even though the connection was already
       // terminated.
       // TODO: Can this even happen and if so, how should we propagate this
       // error?
     } catch (error) {
       _terminate(ErrorCode.INTERNAL_ERROR, message: '$error');
     }
   }

   void _handleFrameImpl(Frame? frame) {
     // The first frame from the other side must be a [SettingsFrame], otherwise
     // we terminate the connection.
     if (_state.isInitialized) {
       if (frame is! SettingsFrame) {
         _terminate(ErrorCode.PROTOCOL_ERROR,
             message: 'Expected to first receive a settings frame.');
         return;
       }
       _state.state = ConnectionState.Operational;
       _onInitialPeerSettingsReceived.complete();
     }

     // Try to defragment [frame] if it is a Headers/PushPromise frame.
     frame = _defragmenter.tryDefragmentFrame(frame);
     if (frame == null) return;

     // Try to decode headers if it's a Headers/PushPromise frame.
     // [This needs to be done even if the frames get ignored, since the entire
     //  connection shares one HPack compression context.]
     if (frame is HeadersFrame) {
       frame.decodedHeaders =
           _hpackContext.decoder.decode(frame.headerBlockFragment);
     } else if (frame is PushPromiseFrame) {
       frame.decodedHeaders =
           _hpackContext.decoder.decode(frame.headerBlockFragment);
     }
     if (_frameReceived.hasListener) {
       _frameReceived.add(null);
     }

     // Handle the frame as either a connection or a stream frame.
     if (frame.header.streamId == 0) {
       if (frame is SettingsFrame) {
         _settingsHandler.handleSettingsFrame(frame);
       } else if (frame is PingFrame) {
         _pingHandler.processPingFrame(frame);
       } else if (frame is WindowUpdateFrame) {
         _connectionWindowHandler.processWindowUpdate(frame);
       } else if (frame is GoawayFrame) {
         _streams.processGoawayFrame(frame);
         _finishing(active: false);
       } else if (frame is UnknownFrame) {
         // We can safely ignore these.
       } else {
         throw ProtocolException(
             'Cannot handle frame type ${frame.runtimeType} with stream-id 0.');
       }
     } else {
       _streams.processStreamFrame(_state, frame);
     }
   }

   void _finishing({bool active = true, String? message}) {
     // If this connection is already dead, we return.
     if (_state.isTerminated) return;

     // If this connection is already finishing, we make sure to store the
     // passive bit, since this information is used by [StreamHandler].
     //
     // Vice versa should not matter: If we started passively finishing, an
     // active finish should be a NOP.
     if (_state.isFinishing) {
       if (!active) _state.finishingState |= ConnectionState.FinishingPassive;
       return;
     }

     assert(_state.isInitialized || _state.isOperational);

     // If we are actively finishing this connection, we'll send a
     // GoawayFrame otherwise we'll just propagate the message.
     if (active) {
       _state.state = ConnectionState.Finishing;
       _state.finishingState |= ConnectionState.FinishingActive;

       _outgoingQueue.enqueueMessage(GoawayMessage(
           _streams.highestPeerInitiatedStream,
           ErrorCode.NO_ERROR,
           message != null ? utf8.encode(message) : []));
     } else {
       _state.state = ConnectionState.Finishing;
       _state.finishingState |= ConnectionState.FinishingPassive;
     }

     _streams.startClosing();
   }

   /// Terminates this connection (if it is not already terminated).
   ///
   /// The returned future will never complete with an error.
   Future _terminate(int errorCode,
       {bool causedByTransportError = false, String? message}) {
     // TODO: When do we complete here?
     if (_state.state != ConnectionState.Terminated) {
       _state.state = ConnectionState.Terminated;

       var cancelFuture = Future.sync(_frameReaderSubscription.cancel);
       if (!causedByTransportError) {
         _outgoingQueue.enqueueMessage(GoawayMessage(
             _streams.highestPeerInitiatedStream,
             errorCode,
             message != null ? utf8.encode(message) : []));
       }
       var closeFuture = _frameWriter.close().catchError((e, s) {
         // We ignore any errors after writing to [GoawayFrame]
       });

       // Close all lower level handlers with an error message.
       // (e.g. if there is a pending connection.ping(), it's returned
       //  Future will complete with this error).
       var exception = TransportConnectionException(
           errorCode, 'Connection is being forcefully terminated.');

       // Close all streams & stream queues
       _streams.terminate(exception);

       // Close the connection queues
       _incomingQueue.terminate(exception);
       _outgoingQueue.terminate(exception);

       _pingHandler.terminate(exception);
       _settingsHandler.terminate(exception);

       return Future.wait([cancelFuture, closeFuture])
           .catchError((_) => const <void>[]);
     }
     return Future<void>.value();
   }
 }

 class ClientConnection extends Connection implements ClientTransportConnection {
   ClientConnection._(super.incoming, super.outgoing, super.settings)
       : super(isClientConnection: true);

   factory ClientConnection(Stream<List<int>> incoming,
       StreamSink<List<int>> outgoing, ClientSettings clientSettings) {
     outgoing.add(CONNECTION_PREFACE);
     return ClientConnection._(incoming, outgoing, clientSettings);
   }

   @override
   bool get isOpen =>
       !_state.isFinishing && !_state.isTerminated && _streams.canOpenStream;

   @override
   ClientTransportStream makeRequest(List<Header> headers,
       {bool endStream = false}) {
     if (_state.isFinishing) {
       throw StateError(
           'The http/2 connection is finishing and can therefore not be used to '
           'make new streams.');
     } else if (_state.isTerminated) {
       throw StateError(
           'The http/2 connection is no longer active and can therefore not be '
           'used to make new streams.');
     }
     var hStream = _streams.newStream(headers, endStream: endStream);
     if (_streams.ranOutOfStreamIds) {
       _finishing(active: true, message: 'Ran out of stream ids');
     }
     return hStream;
   }

   @override
   Stream<int> get onPingReceived => _pingReceived.stream;

   @override
   Stream<void> get onFrameReceived => _frameReceived.stream;
 }

 class ServerConnection extends Connection implements ServerTransportConnection {
   ServerConnection._(super.incoming, super.outgoing, super.settings)
       : super(isClientConnection: false);

   factory ServerConnection(Stream<List<int>> incoming,
       StreamSink<List<int>> outgoing, ServerSettings serverSettings) {
     var frameBytes = readConnectionPreface(incoming);
     return ServerConnection._(frameBytes, outgoing, serverSettings);
   }

   @override
   Stream<ServerTransportStream> get incomingStreams =>
       _streams.incomingStreams.cast<ServerTransportStream>();

   @override
   Stream<int> get onPingReceived => _pingReceived.stream;

   @override
   Stream<void> get onFrameReceived => _frameReceived.stream;
 }
	// Copyright (c) 2015, the Dart project authors. Please see the AUTHORS file
	// for details. All rights reserved. Use of this source code is governed by a
	// BSD-style license that can be found in the LICENSE file.

	import 'dart:async';
	import 'dart:convert' show utf8;

	import '../transport.dart';
	import 'connection_preface.dart';
	import 'flowcontrol/connection_queues.dart';
	import 'flowcontrol/queue_messages.dart';
	import 'flowcontrol/window.dart';
	import 'flowcontrol/window_handler.dart';
	import 'frames/frame_defragmenter.dart';
	import 'frames/frames.dart';
	import 'hpack/hpack.dart';
	import 'ping/ping_handler.dart';
	import 'settings/settings.dart';
	import 'streams/stream_handler.dart';
	import 'sync_errors.dart';

	class ConnectionState {
	/// The connection has been established, we're waiting for the settings frame
	/// of the remote end.
	static const int Initialized = 1;

	/// The connection has been established and is fully operational.
	static const int Operational = 2;

	/// The connection is no longer accepting new streams or creating new streams.
	static const int Finishing = 3;

	/// The connection has been terminated and cannot be used anymore.
	static const int Terminated = 4;

	/// Whether we actively were finishing the connection.
	static const int FinishingActive = 1;

	/// Whether we passively were finishing the connection.
	static const int FinishingPassive = 2;

	int state = Initialized;
	int finishingState = 0;

	ConnectionState();

	bool get isInitialized => state == ConnectionState.Initialized;

	bool get isOperational => state == ConnectionState.Operational;

	bool get isFinishing => state == ConnectionState.Finishing;

	bool get isTerminated => state == ConnectionState.Terminated;

	bool get activeFinishing =>
	state == Finishing && (finishingState & FinishingActive) != 0;

	bool get passiveFinishing =>
	state == Finishing && (finishingState & FinishingPassive) != 0;

	@override
	String toString() {
	var message = '';

	void add(bool condition, String flag) {
	if (condition) {
	if (message.isEmpty) {
	message = flag;
	} else {
	message = '$message/$flag';
	}
	}
	}

	add(isInitialized, 'Initialized');
	add(isOperational, 'IsOperational');
	add(isFinishing, 'IsFinishing');
	add(isTerminated, 'IsTerminated');
	add(activeFinishing, 'ActiveFinishing');
	add(passiveFinishing, 'PassiveFinishing');

	return message;
	}
	}

	abstract class Connection {
	/// The settings the other end has acknowledged to use when communicating with
	/// us.
	final ActiveSettings acknowledgedSettings = ActiveSettings();

	/// The settings we have to obey communicating with the other side.
	final ActiveSettings peerSettings = ActiveSettings();

	/// Whether this connection is a client connection.
	final bool isClientConnection;

	/// Active state handler for this connection.
	ActiveStateHandler? onActiveStateChanged;

	final Completer<void> _onInitialPeerSettingsReceived = Completer<void>();

	final StreamController<int> _pingReceived = StreamController<int>();

	final StreamController<void> _frameReceived = StreamController<void>();

	/// Future which completes when the first SETTINGS frame is received from
	/// the peer.
	Future<void> get onInitialPeerSettingsReceived =>
	_onInitialPeerSettingsReceived.future;

	/// The HPack context for this connection.
	final HPackContext _hpackContext = HPackContext();

	/// The flow window for this connection of the peer.
	final Window _peerWindow = Window();

	/// The flow window for this connection of this end.
	final Window _localWindow = Window();

	/// Used for defragmenting PushPromise/Header frames.
	final FrameDefragmenter _defragmenter = FrameDefragmenter();

	/// The outgoing frames of this connection;
	late FrameWriter _frameWriter;

	/// A subscription of incoming [Frame]s.
	late StreamSubscription<Frame> _frameReaderSubscription;

	/// The incoming connection-level message queue.
	late ConnectionMessageQueueIn _incomingQueue;

	/// The outgoing connection-level message queue.
	late ConnectionMessageQueueOut _outgoingQueue;

	/// The ping handler used for making pings & handling remote pings.
	late PingHandler _pingHandler;

	/// The settings handler used for changing settings & for handling remote
	/// setting changes.
	late SettingsHandler _settingsHandler;

	/// The set of active streams this connection has.
	late StreamHandler _streams;

	/// The connection-level flow control window handler for outgoing messages.
	late OutgoingConnectionWindowHandler _connectionWindowHandler;

	/// The state of this connection.
	late ConnectionState _state;

	Connection(Stream<List<int>> incoming, StreamSink<List<int>> outgoing,
	Settings settings,
	{this.isClientConnection = true}) {
	_setupConnection(incoming, outgoing, settings);
	}

	/// Runs all setup necessary before new streams can be created with the remote
	/// peer.
	void _setupConnection(Stream<List<int>> incoming,
	StreamSink<List<int>> outgoing, Settings settingsObject) {
	// Setup frame reading.
	var incomingFrames =
	FrameReader(incoming, acknowledgedSettings).startDecoding();
	_frameReaderSubscription = incomingFrames.listen((Frame frame) {
	_catchProtocolErrors(() => _handleFrameImpl(frame));
	}, onError: (error, stack) {
	_terminate(ErrorCode.CONNECT_ERROR, causedByTransportError: true);
	}, onDone: () {
	// Ensure existing messages from lower levels are sent to the upper
	// levels before we terminate everything.
	_incomingQueue.forceDispatchIncomingMessages();
	_streams.forceDispatchIncomingMessages();

	_terminate(ErrorCode.CONNECT_ERROR, causedByTransportError: true);
	});

	// Setup frame writing.
	_frameWriter = FrameWriter(_hpackContext.encoder, outgoing, peerSettings);
	_frameWriter.doneFuture.whenComplete(() {
	_terminate(ErrorCode.CONNECT_ERROR, causedByTransportError: true);
	});

	// Setup handlers.
	_settingsHandler = SettingsHandler(_hpackContext.encoder, _frameWriter,
	acknowledgedSettings, peerSettings);
	_pingHandler = PingHandler(_frameWriter, _pingReceived);

	var settings = _decodeSettings(settingsObject);

	// Do the initial settings handshake (possibly with pushes disabled).
	_settingsHandler.changeSettings(settings).catchError((Object error) {
	// TODO: The [error] can contain sensitive information we now expose via
	// a [Goaway] frame. We should somehow ensure we're only sending useful
	// but non-sensitive information.
	_terminate(ErrorCode.PROTOCOL_ERROR,
	message: 'Failed to set initial settings (error: $error).');
	});

	_settingsHandler.onInitialWindowSizeChange.listen((int difference) {
	_catchProtocolErrors(() {
	_streams.processInitialWindowSizeSettingChange(difference);
	});
	});

	// Setup the connection window handler, which keeps track of the
	// size of the outgoing connection window.
	_connectionWindowHandler = OutgoingConnectionWindowHandler(_peerWindow);

	var connectionWindowUpdater =
	IncomingWindowHandler.connection(_frameWriter, _localWindow);

	// Setup queues for outgoing/incoming messages on the connection level.
	_outgoingQueue =
	ConnectionMessageQueueOut(_connectionWindowHandler, _frameWriter);
	_incomingQueue =
	ConnectionMessageQueueIn(connectionWindowUpdater, _catchProtocolErrors);

	if (isClientConnection) {
	_streams = StreamHandler.client(
	_frameWriter,
	_incomingQueue,
	_outgoingQueue,
	_settingsHandler.peerSettings,
	_settingsHandler.acknowledgedSettings,
	_activeStateHandler);
	} else {
	_streams = StreamHandler.server(
	_frameWriter,
	_incomingQueue,
	_outgoingQueue,
	_settingsHandler.peerSettings,
	_settingsHandler.acknowledgedSettings,
	_activeStateHandler);
	}

	// NOTE: We're not waiting until initial settings have been exchanged
	// before we start using the connection (i.e. we don't wait for half a
	// round-trip-time).
	_state = ConnectionState();
	}

	List<Setting> _decodeSettings(Settings settings) {
	var settingsList = <Setting>[];

	// By default a endpoint can make an unlimited number of concurrent streams.
	var concurrentStreamLimit = settings.concurrentStreamLimit;
	if (concurrentStreamLimit != null) {
	settingsList.add(Setting(
	Setting.SETTINGS_MAX_CONCURRENT_STREAMS, concurrentStreamLimit));
	}

	// By default the stream level flow control window is 64 KiB.
	var streamWindowSize = settings.streamWindowSize;
	if (streamWindowSize != null) {
	settingsList
	.add(Setting(Setting.SETTINGS_INITIAL_WINDOW_SIZE, streamWindowSize));
	}

	if (settings is ClientSettings) {
	// By default the server is allowed to do server pushes.
	if (!settings.allowServerPushes) {
	settingsList.add(Setting(Setting.SETTINGS_ENABLE_PUSH, 0));
	}
	} else if (settings is ServerSettings) {
	// No special server settings at the moment.
	} else {
	assert(false);
	}

	return settingsList;
	}

	/// Pings the remote peer (can e.g. be used for measuring latency).
	Future<void> ping() {
	return _pingHandler.ping().catchError((e, s) {
	return Future<void>.error(
	TransportException('The connection has been terminated.'));
	}, test: (e) => e is TerminatedException);
	}

	/// Finishes this connection.
	Future<void> finish() {
	_finishing(active: true);

	// TODO: There is probably more we need to wait for.
	return _streams.done.whenComplete(() =>
	Future.wait([_frameWriter.close(), _frameReaderSubscription.cancel()]));
	}

	/// Terminates this connection forcefully.
	Future<void> terminate([int? errorCode]) {
	return _terminate(errorCode ?? ErrorCode.NO_ERROR);
	}

	void _activeStateHandler(bool isActive) =>
	onActiveStateChanged?.call(isActive);

	/// Invokes the passed in closure and catches any exceptions.
	void _catchProtocolErrors(void Function() fn) {
	try {
	fn();
	} on ProtocolException catch (error) {
	_terminate(ErrorCode.PROTOCOL_ERROR, message: '$error');
	} on FlowControlException catch (error) {
	_terminate(ErrorCode.FLOW_CONTROL_ERROR, message: '$error');
	} on FrameSizeException catch (error) {
	_terminate(ErrorCode.FRAME_SIZE_ERROR, message: '$error');
	} on HPackDecodingException catch (error) {
	_terminate(ErrorCode.PROTOCOL_ERROR, message: '$error');
	} on TerminatedException {
	// We tried to perform an action even though the connection was already
	// terminated.
	// TODO: Can this even happen and if so, how should we propagate this
	// error?
	} catch (error) {
	_terminate(ErrorCode.INTERNAL_ERROR, message: '$error');
	}
	}

	void _handleFrameImpl(Frame? frame) {
	// The first frame from the other side must be a [SettingsFrame], otherwise
	// we terminate the connection.
	if (_state.isInitialized) {
	if (frame is! SettingsFrame) {
	_terminate(ErrorCode.PROTOCOL_ERROR,
	message: 'Expected to first receive a settings frame.');
	return;
	}
	_state.state = ConnectionState.Operational;
	_onInitialPeerSettingsReceived.complete();
	}

	// Try to defragment [frame] if it is a Headers/PushPromise frame.
	frame = _defragmenter.tryDefragmentFrame(frame);
	if (frame == null) return;

	// Try to decode headers if it's a Headers/PushPromise frame.
	// [This needs to be done even if the frames get ignored, since the entire
	// connection shares one HPack compression context.]
	if (frame is HeadersFrame) {
	frame.decodedHeaders =
	_hpackContext.decoder.decode(frame.headerBlockFragment);
	} else if (frame is PushPromiseFrame) {
	frame.decodedHeaders =
	_hpackContext.decoder.decode(frame.headerBlockFragment);
	}
	if (_frameReceived.hasListener) {
	_frameReceived.add(null);
	}

	// Handle the frame as either a connection or a stream frame.
	if (frame.header.streamId == 0) {
	if (frame is SettingsFrame) {
	_settingsHandler.handleSettingsFrame(frame);
	} else if (frame is PingFrame) {
	_pingHandler.processPingFrame(frame);
	} else if (frame is WindowUpdateFrame) {
	_connectionWindowHandler.processWindowUpdate(frame);
	} else if (frame is GoawayFrame) {
	_streams.processGoawayFrame(frame);
	_finishing(active: false);
	} else if (frame is UnknownFrame) {
	// We can safely ignore these.
	} else {
	throw ProtocolException(
	'Cannot handle frame type ${frame.runtimeType} with stream-id 0.');
	}
	} else {
	_streams.processStreamFrame(_state, frame);
	}
	}

	void _finishing({bool active = true, String? message}) {
	// If this connection is already dead, we return.
	if (_state.isTerminated) return;

	// If this connection is already finishing, we make sure to store the
	// passive bit, since this information is used by [StreamHandler].
	//
	// Vice versa should not matter: If we started passively finishing, an
	// active finish should be a NOP.
	if (_state.isFinishing) {
	if (!active) _state.finishingState \|= ConnectionState.FinishingPassive;
	return;
	}

	assert(_state.isInitialized \|\| _state.isOperational);

	// If we are actively finishing this connection, we'll send a
	// GoawayFrame otherwise we'll just propagate the message.
	if (active) {
	_state.state = ConnectionState.Finishing;
	_state.finishingState \|= ConnectionState.FinishingActive;

	_outgoingQueue.enqueueMessage(GoawayMessage(
	_streams.highestPeerInitiatedStream,
	ErrorCode.NO_ERROR,
	message != null ? utf8.encode(message) : []));
	} else {
	_state.state = ConnectionState.Finishing;
	_state.finishingState \|= ConnectionState.FinishingPassive;
	}

	_streams.startClosing();
	}

	/// Terminates this connection (if it is not already terminated).
	///
	/// The returned future will never complete with an error.
	Future _terminate(int errorCode,
	{bool causedByTransportError = false, String? message}) {
	// TODO: When do we complete here?
	if (_state.state != ConnectionState.Terminated) {
	_state.state = ConnectionState.Terminated;

	var cancelFuture = Future.sync(_frameReaderSubscription.cancel);
	if (!causedByTransportError) {
	_outgoingQueue.enqueueMessage(GoawayMessage(
	_streams.highestPeerInitiatedStream,
	errorCode,
	message != null ? utf8.encode(message) : []));
	}
	var closeFuture = _frameWriter.close().catchError((e, s) {
	// We ignore any errors after writing to [GoawayFrame]
	});

	// Close all lower level handlers with an error message.
	// (e.g. if there is a pending connection.ping(), it's returned
	// Future will complete with this error).
	var exception = TransportConnectionException(
	errorCode, 'Connection is being forcefully terminated.');

	// Close all streams & stream queues
	_streams.terminate(exception);

	// Close the connection queues
	_incomingQueue.terminate(exception);
	_outgoingQueue.terminate(exception);

	_pingHandler.terminate(exception);
	_settingsHandler.terminate(exception);

	return Future.wait([cancelFuture, closeFuture])
	.catchError((_) => const <void>[]);
	}
	return Future<void>.value();
	}
	}

	class ClientConnection extends Connection implements ClientTransportConnection {
	ClientConnection._(super.incoming, super.outgoing, super.settings)
	: super(isClientConnection: true);

	factory ClientConnection(Stream<List<int>> incoming,
	StreamSink<List<int>> outgoing, ClientSettings clientSettings) {
	outgoing.add(CONNECTION_PREFACE);
	return ClientConnection._(incoming, outgoing, clientSettings);
	}

	@override
	bool get isOpen =>
	!_state.isFinishing && !_state.isTerminated && _streams.canOpenStream;

	@override
	ClientTransportStream makeRequest(List<Header> headers,
	{bool endStream = false}) {
	if (_state.isFinishing) {
	throw StateError(
	'The http/2 connection is finishing and can therefore not be used to '
	'make new streams.');
	} else if (_state.isTerminated) {
	throw StateError(
	'The http/2 connection is no longer active and can therefore not be '
	'used to make new streams.');
	}
	var hStream = _streams.newStream(headers, endStream: endStream);
	if (_streams.ranOutOfStreamIds) {
	_finishing(active: true, message: 'Ran out of stream ids');
	}
	return hStream;
	}

	@override
	Stream<int> get onPingReceived => _pingReceived.stream;

	@override
	Stream<void> get onFrameReceived => _frameReceived.stream;
	}

	class ServerConnection extends Connection implements ServerTransportConnection {
	ServerConnection._(super.incoming, super.outgoing, super.settings)
	: super(isClientConnection: false);

	factory ServerConnection(Stream<List<int>> incoming,
	StreamSink<List<int>> outgoing, ServerSettings serverSettings) {
	var frameBytes = readConnectionPreface(incoming);
	return ServerConnection._(frameBytes, outgoing, serverSettings);
	}

	@override
	Stream<ServerTransportStream> get incomingStreams =>
	_streams.incomingStreams.cast<ServerTransportStream>();

	@override
	Stream<int> get onPingReceived => _pingReceived.stream;

	@override
	Stream<void> get onFrameReceived => _frameReceived.stream;
	}