Start of a HTTP/2 Server implementation in Dart - by Simon Lightfoot

server.dart

// MIT License
//
// Copyright (c) 2024 Simon Lightfoot
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
import 'dart:async';
import 'dart:convert';
import 'dart:io';

import 'package:http2/http2.dart' as http2;
import 'package:http2/src/artificial_server_socket.dart' as http2;
import 'package:shelf/shelf.dart';
import 'package:shelf/shelf_io.dart';
import 'package:shelf_router/shelf_router.dart';

String localFile(path) => Platform.script.resolve(path).toFilePath();

final _router = Router()
  ..get('/', _rootHandler)
  ..get('/echo/<message>', _echoHandler);

Response _rootHandler(Request req) {
  return Response.ok('Hello, World!\n');
}

Response _echoHandler(Request request) {
  final message = request.params['message'];
  return Response.ok('$message\n');
}

void main(List<String> args) async {
  final address = InternetAddress.anyIPv4;
  final port = 443;
  final handler = Pipeline() //
      .addMiddleware(logRequests())
      .addHandler(_router.call);

  // https://httpwg.org/specs/rfc7540.html

  // https://stackoverflow.com/questions/10175812/how-to-generate-a-self-signed-ssl-certificate-using-openssl
  // https://stackoverflow.com/a/10176685/1383790
  // openssl req -x509 -newkey rsa:4096 -sha256 -days 3650 -nodes -keyout localhost.key -out localhost.crt -subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,DNS:localhost.localdomain,IP:127.0.0.1,IP:::1"
  final serverContext = SecurityContext()
    ..setTrustedCertificates(localFile('localhost.crt'))
    ..useCertificateChain(localFile('localhost.crt'))
    ..usePrivateKey(localFile('localhost.key'))
    ..setAlpnProtocols(
      ['h2', 'h2-14', 'http/1.1'],
      true,
    );
  final secureServer = await SecureServerSocket.bind(
    address,
    port,
    serverContext,
    supportedProtocols: ['h2', 'h2-14', 'http/1.1'],
  );
  final http11Controller = _ServerSocketController(secureServer.address, secureServer.port);
  final http11Server = HttpServer.listenOn(http11Controller.stream);
  final http2Controller = StreamController<http2.ServerTransportStream>();

  secureServer.listen(
    (SecureSocket socket) {
      var protocol = socket.selectedProtocol;
      if (protocol == null || protocol == 'http/1.1') {
        http11Controller.addHttp11Socket(socket);
      } else if (protocol == 'h2' || protocol == 'h2-14') {
        print('connection started on $socket');
        final connection = http2.ServerTransportConnection.viaSocket(socket);
        connection.onPingReceived.listen((int ping) {
          print('ping $ping');
          connection.ping();
        });
        connection.incomingStreams.listen(
          http2Controller.add,
          onError: (error, stackTrace) {
            print('connection error: $error\n$stackTrace');
          },
          onDone: () {
            print('connection done');
          },
        );
      } else {
        socket.destroy();
        throw Exception('Unexpected negotiated ALPN protocol: $protocol.');
      }
    },
    onError: (error, stackTrace) {
      print('socket error: $error\n$stackTrace');
    },
  );

  http11Server.listen((HttpRequest request) {
    print('HTTP1/1');
    handleRequest(request, handler);
  });
  http2Controller.stream.listen((http2.ServerTransportStream stream) {
    print('HTTP2');
    Http2Server.handleStream(stream, handler);
  });

  print('Server listening on port $port');
}

/// An internal helper class.
class _ServerSocketController {
  final InternetAddress address;
  final int port;
  final StreamController<Socket> _controller = StreamController();

  _ServerSocketController(this.address, this.port);

  http2.ArtificialServerSocket get stream {
    return http2.ArtificialServerSocket(address, port, _controller.stream);
  }

  void addHttp11Socket(Socket socket) {
    _controller.add(socket);
  }

  Future close() => _controller.close();
}

class Http2Server {
  static Future<void> serve(
    Object address,
    int port,
    Handler handler,
  ) async {
    final socket = await ServerSocket.bind(address, port);
    socket.listen((socket) {
      print('http2 connection established.');
      final connection = http2.ServerTransportConnection.viaSocket(socket);
      connection.incomingStreams.listen(
        (http2.ServerTransportStream stream) {
          handleStream(stream, handler);
        },
        onError: (error, stackTrace) {
          print('http2 error: $error\n$stackTrace');
        },
        onDone: () {
          print('http2 done');
        },
      );
      connection.onPingReceived.listen((event) {
        print('http2 ping');
        connection.ping();
      });
    });
  }

  static Future<void> handleStream(http2.ServerTransportStream stream, Handler handler) async {
    print('http2 id:${stream.id} push:${stream.canPush}');
    final requestHeaders = <String, String>{};
    final requestData = <int>[];
    late StreamSubscription<http2.StreamMessage> incomingSub;
    incomingSub = stream.incomingMessages.listen((http2.StreamMessage message) async {
      print('http2 id:${stream.id} message: $message ${message.endStream}');
      if (message is http2.HeadersStreamMessage) {
        for (final header in message.headers) {
          requestHeaders[ascii.decode(header.name)] = ascii.decode(header.value);
        }
      } else if (message is http2.DataStreamMessage) {
        requestData.addAll(message.bytes);
      }
      if (message.endStream) {
        final uri = Uri(
          scheme: requestHeaders[':scheme']!,
          host: requestHeaders[':authority']!,
          path: requestHeaders[':path']!,
        );
        final method = requestHeaders[':method']!.toUpperCase();
        print('$method $uri');
        print('headers:');
        for (final MapEntry(:key, :value) in requestHeaders.entries) {
          print('\t$key: $value');
        }
        final request = Request(
          method,
          uri,
          protocolVersion: '2.0',
          headers: Map.fromEntries(requestHeaders.entries.where(
            (entry) => !entry.key.startsWith(':'),
          )),
          body: requestData,
        );
        final response = await handler(request);
        stream.sendHeaders([
          http2.Header.ascii(':status', response.statusCode.toString()),
          for (final MapEntry(:key, :value) in response.headers.entries) //
            http2.Header.ascii(key, value),
        ]);
        await for (final (buffer, :isLast) in response.read().withLast()) {
          stream.sendData(buffer, endStream: isLast);
        }
        await stream.outgoingMessages.close();
        await incomingSub.cancel();
      }
    });
  }
}

extension StreamWithLast<T> on Stream<T> {
  Stream<(T buffer, {bool isLast})> withLast() async* {
    T? buffer;
    await for (final bytes in this) {
      if (buffer != null) {
        yield (buffer, isLast: false);
      }
      buffer = bytes;
    }
    if (buffer != null) {
      yield (buffer, isLast: true);
    }
  }
}