List the messages of a WebSocket and of an event stream

CHANGELOG.md

@@ -11,6 +11,7 @@ board can be cross-checked against what actually shipped.
 
 ### Added
 
+- WebSockets and server-sent event streams. A WebSocket is one call that stays open, and it lists its messages with the ping, pong and close frames, which way each went, how big it was and when it crossed, compressed messages included. An event stream lists its events as they complete, with their names and ids, and keeps the raw stream as the body of the response. Both get a Messages or Events tab, a badge in the list, a line in the overview, entries in the HAR and text exports, and their messages are kept on disk with the call. The latest 1000 messages are kept, with a megabyte of payload at most, and the tab says how many came before (M16)
 - Any call can be opened in an editor, changed and sent again. The editor starts from the method, address, header fields and body that were recorded. A value the redaction masked was never kept, so it is flagged and not sent unless it is typed again. What goes out is recorded as a new call that names the one it repeats, marked in the list and linked from its overview. It needs `dart:io`, so the web does not offer it (M15)
 - A line above the list that says when Flutwire may be missing calls, with the reason and what to change: a client or adapter that does not use `dart:io`, an adapter that ran in another isolate, or something that replaced the capture's `HttpOverrides`. It stays out of sight until Flutwire has seen one of them happen, and it names the client or the isolate. `Flutwire.coverage` lists the notes, and an adapter that stands down for the capture reports through `Flutwire.watchStandDown`. The example app has a probe for each case (M24)
 - An example for each adapter, so the package page on pub.dev has one to show (M34)

README.md

@@ -13,7 +13,7 @@
 
 When an app misbehaves on a phone, the call that explains it is usually out of reach. The console shows only what a logger was told to print. A proxy needs a certificate on the device and a changed Wi-Fi setting. Every HTTP client in the app has to be wired by hand before it says anything, and when the app is killed the evidence goes with it.
 
-Flutwire records the calls an app makes, keeps them on the device and lists the latest ones in the notification shade. Tapping it opens the inspector, and each call opens in full: URL, timings, headers, cookies, and the request and response bodies as a JSON tree, indented text, form fields, an image or a hex dump. A call can be edited and sent again from the same screen. A debug build needs one line in `main`; a release build records nothing.
+Flutwire records the calls an app makes, keeps them on the device and lists the latest ones in the notification shade. Tapping it opens the inspector, and each call opens in full: URL, timings, headers, cookies, and the request and response bodies as a JSON tree, indented text, form fields, an image or a hex dump. A call can be edited and sent again from the same screen, and a WebSocket or an event stream lists its messages as they arrive. A debug build needs one line in `main`; a release build records nothing.
 
 <p align="center">
   <a href="docs/media/flutwire-demo.mp4"><img src="docs/media/flutwire-demo.gif" alt="The notification in the shade lists the latest calls, and tapping it opens the inspector, where a call opens as a JSON tree" width="300"></a><br>

packages/flutwire/README.md

@@ -6,7 +6,7 @@
 
 **A network inspector for Flutter that opens from the notification shade.**
 
-Flutwire records the HTTP calls an app makes, keeps them on the device and lists the latest ones in the notification shade. Tapping it opens the inspector, and each call opens in full: URL, timings, headers, cookies, and the request and response bodies as a JSON tree, indented text, form fields, an image or a hex dump. A call can be edited and sent again from the same screen.
+Flutwire records the HTTP calls an app makes, keeps them on the device and lists the latest ones in the notification shade. Tapping it opens the inspector, and each call opens in full: URL, timings, headers, cookies, and the request and response bodies as a JSON tree, indented text, form fields, an image or a hex dump. A call can be edited and sent again from the same screen, and a WebSocket or an event stream lists its messages as they arrive.
 
 <p align="center">
   <a href="https://github.com/NaCode-Studios/Flutwire/blob/main/docs/media/flutwire-demo.mp4"><img src="https://raw.githubusercontent.com/NaCode-Studios/Flutwire/main/docs/media/flutwire-demo.gif" alt="The notification in the shade lists the latest calls, and tapping it opens the inspector, where a call opens as a JSON tree" width="300"></a><br>
@@ -49,6 +49,12 @@ MaterialApp(
 );
 ```
 
+## WebSockets and event streams
+
+A WebSocket opened through `dart:io` is one call that stays open. Its Messages tab lists every message and control frame in the order it crossed, which way it went, how big it was and when, and a compressed message is shown after it is decompressed. An event stream (`text/event-stream`) lists its events as they complete, with their names and ids, and the raw stream stays the body of the response. Only the latest 1000 messages are kept, and the tab says how many came before them.
+
+`dart:io` makes one client for all WebSockets the first time one is opened, so a WebSocket opened before `Flutwire.init` is not seen: call `init` first thing in `main`. On the web the browser opens WebSockets itself, and none is recorded there.
+
 ## Sending a call again
 
 Open a call and tap the replay button in the top bar. The method, the address, the header fields and the body can all be changed, and Send makes the request. What goes out is recorded as a new call that names the one it repeats.

packages/flutwire/example/lib/main.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
 import 'dart:isolate';
+import 'dart:typed_data';
 
 import 'package:dio/dio.dart';
 import 'package:flutter/foundation.dart' show kIsWeb;
@@ -177,6 +178,10 @@ final List<(String, List<Probe>)> groups = [
       ),
     ],
   ),
+  (
+    'Streams',
+    [Probe('WebSocket', _webSocket), Probe('Event stream', _eventStream)],
+  ),
   (
     'Trouble',
     [
@@ -210,6 +215,55 @@ final List<(String, List<Probe>)> groups = [
   ),
 ];
 
+/// Opens a WebSocket to an echo server, sends a text and a binary message, and
+/// closes it, so the inspector has a conversation to list. `dart:io` makes one
+/// client for every WebSocket the first time it opens one, so this works
+/// because `Flutwire.init` ran before that.
+Future<void> _webSocket() async {
+  if (kIsWeb) return;
+  try {
+    final socket = await WebSocket.connect('wss://echo.websocket.org');
+    final closed = Completer<void>();
+    socket.listen(
+      (_) {},
+      onDone: closed.complete,
+      onError: (Object _) => closed.complete(),
+    );
+    socket
+      ..add('hello')
+      ..add(jsonEncode({'subscribe': 'prices'}))
+      ..add(Uint8List.fromList([1, 2, 3]));
+    await Future<void>.delayed(const Duration(seconds: 2));
+    await socket.close(1000, 'done');
+    await closed.future.timeout(const Duration(seconds: 5));
+  } on Object {
+    // The inspector shows what went wrong.
+  }
+}
+
+/// Reads a few chunks of a public event stream and walks away, so the
+/// inspector lists the events that arrived and the call ends as cancelled.
+Future<void> _eventStream() async {
+  if (kIsWeb) return;
+  final io = HttpClient();
+  try {
+    final request = await io.getUrl(
+      Uri.parse('https://stream.wikimedia.org/v2/stream/recentchange'),
+    );
+    request.headers.set('accept', 'text/event-stream');
+    final response = await request.close();
+    await response
+        .transform(utf8.decoder)
+        .take(6)
+        .drain<void>()
+        .timeout(const Duration(seconds: 10));
+  } on Object {
+    // The inspector shows what went wrong.
+  } finally {
+    io.close(force: true);
+  }
+}
+
 /// What Flutwire cannot see, so that the inspector has something to say about
 /// it. These are left out of "Run everything": replacing the capture stops the
 /// recording of everything after it.

packages/flutwire/lib/flutwire.dart

@@ -14,6 +14,7 @@ export 'src/flutwire.dart' show Flutwire;
 export 'src/model/body.dart';
 export 'src/model/call.dart';
 export 'src/model/headers.dart';
+export 'src/model/message.dart';
 export 'src/model/multipart.dart';
 export 'src/store/preferences.dart';
 export 'src/store/store.dart';

packages/flutwire/lib/src/capture/messages/message_log.dart

@@ -0,0 +1,48 @@
+import 'dart:collection';
+
+import '../../model/message.dart';
+
+/// The messages of one call, kept within bounds.
+///
+/// A conversation can run for hours, so the log keeps the latest messages and
+/// lets go of the oldest, and it counts all of them.
+final class MessageLog {
+  /// Creates a log that keeps at most [maxMessages] messages, and at most
+  /// [maxBytes] of payload between them.
+  MessageLog({this.maxMessages = 1000, this.maxBytes = 1 << 20});
+
+  /// The most messages the log keeps.
+  final int maxMessages;
+
+  /// The most payload bytes the messages of the log keep together. A single
+  /// message that is bigger than that is kept all the same, on its own.
+  final int maxBytes;
+
+  final Queue<FlutwireMessage> _items = Queue<FlutwireMessage>();
+  int _total = 0;
+  int _bytes = 0;
+
+  /// How many messages were added, kept or not. It is also the index the next
+  /// one gets.
+  int get total => _total;
+
+  /// Adds [message] and lets go of the oldest ones while the log is over a
+  /// limit.
+  void add(FlutwireMessage message) {
+    _items.add(message);
+    _total++;
+    _bytes += message.bytes.length;
+    while (_items.length > maxMessages ||
+        (_bytes > maxBytes && _items.length > 1)) {
+      _bytes -= _items.removeFirst().bytes.length;
+    }
+  }
+
+  /// What the log holds now, as the messages of a call of the given [kind].
+  FlutwireMessages snapshot(FlutwireConversationKind kind) => FlutwireMessages(
+    kind: kind,
+    items: List.unmodifiable(_items),
+    total: _total,
+    keptBytes: _bytes,
+  );
+}

packages/flutwire/lib/src/capture/messages/sse_parser.dart

@@ -0,0 +1,178 @@
+import 'dart:convert';
+
+/// A server-sent event, as the stream wrote it.
+final class SseEvent {
+  /// Creates an event.
+  const SseEvent({
+    required this.data,
+    required this.totalBytes,
+    this.event,
+    this.id,
+  });
+
+  /// The `data:` lines joined by a line feed, cut at the limit the parser was
+  /// given.
+  final String data;
+
+  /// The size of the whole of [data] in UTF-8 bytes.
+  final int totalBytes;
+
+  /// The `event:` field, when the event had one.
+  final String? event;
+
+  /// The last `id:` the stream sent, which stays in force for the events after
+  /// it.
+  final String? id;
+}
+
+/// Reads a `text/event-stream` as it arrives and reports each event once it is
+/// complete.
+///
+/// It follows the rules of the HTML standard: lines end with a line feed, a
+/// carriage return or both, a line that starts with a colon is a comment, an
+/// event ends at an empty line, and a leading byte order mark is ignored.
+final class SseParser {
+  /// Creates a parser that calls [onEvent] for each event. A `data` field
+  /// longer than [maxChars] is cut there, and its real size is reported.
+  SseParser(this._onEvent, {this.maxChars = 250000}) {
+    _decoder = const Utf8Decoder(
+      allowMalformed: true,
+    ).startChunkedConversion(_TextSink(_text));
+  }
+
+  final void Function(SseEvent event) _onEvent;
+
+  /// How many characters of an event's data are kept.
+  final int maxChars;
+
+  late final Sink<List<int>> _decoder;
+  final StringBuffer _line = StringBuffer();
+  final StringBuffer _data = StringBuffer();
+  bool _first = true;
+  bool _afterCarriageReturn = false;
+  bool _hasData = false;
+  int _dataBytes = 0;
+  int _kept = 0;
+  String? _eventType;
+  String? _lastId;
+
+  /// Feeds the next [chunk] of the stream.
+  void add(List<int> chunk) => _decoder.add(chunk);
+
+  /// Says the stream ended. An event that was not completed by an empty line
+  /// is dropped, as the standard asks.
+  void close() => _decoder.close();
+
+  void _text(String chunk) {
+    var text = chunk;
+    if (_first) {
+      _first = false;
+      if (text.startsWith('')) text = text.substring(1);
+    }
+    for (var i = 0; i < text.length; i++) {
+      final unit = text.codeUnitAt(i);
+      if (_afterCarriageReturn) {
+        _afterCarriageReturn = false;
+        if (unit == 0x0A) continue;
+      }
+      if (unit == 0x0D) {
+        _endOfLine();
+        _afterCarriageReturn = true;
+      } else if (unit == 0x0A) {
+        _endOfLine();
+      } else {
+        _line.writeCharCode(unit);
+      }
+    }
+  }
+
+  void _endOfLine() {
+    final line = _line.toString();
+    _line.clear();
+    if (line.isEmpty) {
+      _dispatch();
+      return;
+    }
+    if (line.startsWith(':')) return;
+    final colon = line.indexOf(':');
+    final name = colon < 0 ? line : line.substring(0, colon);
+    var value = colon < 0 ? '' : line.substring(colon + 1);
+    if (value.startsWith(' ')) value = value.substring(1);
+    switch (name) {
+      case 'event':
+        _eventType = value;
+      case 'data':
+        _addData(value);
+      case 'id':
+        if (!value.contains('\u0000')) _lastId = value;
+    }
+  }
+
+  void _addData(String value) {
+    if (_hasData) {
+      _dataBytes += 1;
+      _write('\n');
+    }
+    _hasData = true;
+    _dataBytes += _utf8Length(value);
+    _write(value);
+  }
+
+  void _write(String text) {
+    final room = maxChars - _kept;
+    if (room <= 0) return;
+    final piece = text.length <= room ? text : text.substring(0, room);
+    _data.write(piece);
+    _kept += piece.length;
+  }
+
+  void _dispatch() {
+    final had = _hasData;
+    final type = _eventType;
+    final data = _data.toString();
+    final bytes = _dataBytes;
+    _data.clear();
+    _hasData = false;
+    _dataBytes = 0;
+    _kept = 0;
+    _eventType = null;
+    if (!had) return;
+    _onEvent(
+      SseEvent(
+        data: data,
+        totalBytes: bytes,
+        event: type == null || type.isEmpty ? null : type,
+        id: _lastId == null || _lastId!.isEmpty ? null : _lastId,
+      ),
+    );
+  }
+
+  static int _utf8Length(String text) {
+    var bytes = 0;
+    for (final unit in text.codeUnits) {
+      if (unit < 0x80) {
+        bytes += 1;
+      } else if (unit < 0x800) {
+        bytes += 2;
+      } else if (unit >= 0xD800 && unit <= 0xDFFF) {
+        // Half of a surrogate pair, which is four bytes in all.
+        bytes += 2;
+      } else {
+        bytes += 3;
+      }
+    }
+    return bytes;
+  }
+}
+
+class _TextSink implements Sink<String> {
+  _TextSink(this._onText);
+
+  final void Function(String text) _onText;
+
+  @override
+  void add(String data) => _onText(data);
+
+  @override
+  void close() {}
+}