webSocketHandler function

Future<Response> Function(Request) webSocketHandler(ConnectionCallback onConnection, { Iterable<String>? protocols, Iterable<String>? allowedOrigins, Duration? pingInterval, int maxMessageSize = 1024 * 1024, })

Shelf-style callback and options, with bounded native message queues. Use sink.addStream (or NativeWebSocketChannel.send) for backpressure. sink.add buffers in Dart and applies backpressure to the network, matching dart:io WebSocket semantics without throwing on a full queue. Frames added in one turn of the event loop are written together, from the isolate's own thread whenever the socket can take them at once. Messages default to at most 1 MiB.

Implementation

Future<Response> Function(Request) webSocketHandler(
  ConnectionCallback onConnection, {
  Iterable<String>? protocols,
  Iterable<String>? allowedOrigins,
  Duration? pingInterval,
  int maxMessageSize = 1024 * 1024,
}) {
  if (maxMessageSize <= 0 || maxMessageSize > 64 * 1024 * 1024) {
    throw ArgumentError.value(maxMessageSize, 'maxMessageSize');
  }
  if (pingInterval != null && pingInterval <= Duration.zero) {
    throw ArgumentError.value(pingInterval, 'pingInterval');
  }
  final origins = allowedOrigins?.map((o) => o.toLowerCase()).toSet();
  final supported = protocols?.toSet() ?? <String>{};
  return (request) async {
    if (request.method != 'GET' ||
        request.headers.get('upgrade')?.toLowerCase() != 'websocket') {
      return Response.text('WebSocket upgrade required', status: 404);
    }
    final origin = request.headers.get('origin');
    if (origin != null &&
        origins != null &&
        !origins.contains(origin.toLowerCase())) {
      return Response.text('Origin refused', status: 403);
    }
    final tokens = request.headers
        .get('connection')
        ?.toLowerCase()
        .split(',')
        .map((v) => v.trim());
    final key = request.headers.get('sec-websocket-key');
    var validKey = false;
    try {
      validKey = key != null && base64.decode(key).length == 16;
    } on FormatException {
      /* refused below */
    }
    if (tokens?.contains('upgrade') != true ||
        !validKey ||
        request.headers.get('sec-websocket-version') != '13') {
      return Response.text('Invalid WebSocket handshake', status: 400);
    }
    String? protocol;
    for (final value
        in (request.headers.get('sec-websocket-protocol') ?? '').split(',')) {
      if (supported.contains(value.trim())) {
        protocol = value.trim();
        break;
      }
    }
    return WebSocketResponse(
      onConnection,
      protocol,
      maxMessageSize,
      pingInterval,
    );
  };
}