[web] Development server proxy buffers SSE and NDJSON responses #194354

import 'dart:async';
import 'dart:io';

Future<void> main() async {
  final server = await HttpServer.bind(InternetAddress.loopbackIPv4, 51880);
  print('Backend: http://127.0.0.1:51880/api/events');
  await for (final request in server) {
    unawaited(handle(request));
  }
}

Future<void> handle(HttpRequest request) async {
  final response = request.response;
  response.headers.set('Access-Control-Allow-Origin', '*');
  if (request.uri.path != '/api/events') {
    response.statusCode = HttpStatus.notFound;
    await response.close();
    return;
  }
  response.bufferOutput = false;
  response.headers.set('Content-Type', 'application/x-ndjson');
  response.headers.set('Cache-Control', 'no-store');
  response.headers.set('X-Accel-Buffering', 'no');
  try {
    response.write('{"type":"connected"}\n');
    await response.flush();
    // Keep the response open without filling any downstream proxy buffer.
    await Future<void>.delayed(const Duration(seconds: 30));
    await response.close();
  } catch (_) {
    // The browser aborts after receiving its first line or timing out.
  }
}

import 'dart:async';
import 'dart:convert';

import 'package:flutter/material.dart';
import 'package:http/browser_client.dart';
import 'package:http/http.dart' as http;

void main() => runApp(const MaterialApp(home: StreamDemo()));

class StreamDemo extends StatefulWidget {
  const StreamDemo({super.key});

  @override
  State<StreamDemo> createState() => _StreamDemoState();
}

class _StreamDemoState extends State<StreamDemo> {
  final _messages = <String>[];

  @override
  void initState() {
    super.initState();
    WidgetsBinding.instance.addPostFrameCallback((_) async {
      await _probe('http://127.0.0.1:51880/api/events');
      await _probe('/api/events');
    });
  }

  void _log(String message) {
    if (mounted) setState(() => _messages.add(message));
  }

  Future<void> _probe(String path) async {
    final client = BrowserClient();
    final abort = Completer<void>();
    final uri = Uri.base.resolve(path);
    final watch = Stopwatch()..start();
    _log('GET $uri');
    try {
      final firstLine = await (() async {
        final response = await client.send(
          http.AbortableRequest('GET', uri, abortTrigger: abort.future),
        );
        _log('HTTP ${response.statusCode} at ${watch.elapsedMilliseconds} ms');
        return response.stream
            .transform(utf8.decoder)
            .transform(const LineSplitter())
            .first;
      })().timeout(const Duration(seconds: 3));
      _log('First line at ${watch.elapsedMilliseconds} ms: $firstLine');
    } catch (error) {
      _log('After ${watch.elapsedMilliseconds} ms: $error');
    } finally {
      abort.complete();
      client.close();
    }
  }

  @override
  Widget build(BuildContext context) => Scaffold(
    appBar: AppBar(title: const Text('Flutter proxy streaming reproduction')),
    body: Padding(
      padding: const EdgeInsets.all(16),
      child: SingleChildScrollView(
        child: SelectableText(_messages.join('\n')),
      ),
    ),
  );
}