feat(app): parse json-stream reports into structured events

The native layer now emits one JSON object per report, so the service parses each into a sealed Iperf3Event — start, interval, end, error — and publishes those instead of opaque lines. Interval samples carry bytes, bits per second, retransmits and per-stream socket/cwnd/rtt detail; the end event carries sent and received totals and CPU utilisation. None of that was available while the output was scraped text. Anything unrecognised — non-JSON, malformed JSON, an event name a future iperf3 adds — becomes an Iperf3LogLine rather than being dropped, so unexpected output still reaches the user. Failures arrive as an `error` event rather than a non-zero return from iperf_run_client, so that event is what now marks a run as failed. The result's rawOutput keeps the JSON lines verbatim, since that is what the native layer actually produced. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-26 19:35:10 +09:00
parent 598a2d99c9
commit 7d8d06e2e5
3 changed files with 375 additions and 26 deletions
@@ -0,0 +1,238 @@
+import 'dart:convert';
+
+/// One report from iperf3's `--json-stream` output.
+///
+/// iperf3 emits `{"event": "<name>", "data": {...}}` per report; the events a
+/// client run produces are `start`, `interval`, `end` and `error`. The bridge
+/// adds `jni_stopped` when a run ended because the user stopped it.
+sealed class Iperf3Event {
+  const Iperf3Event();
+
+  /// Parse one line of `--json-stream` output.
+  ///
+  /// Returns an [Iperf3LogLine] for anything that is not a recognised JSON
+  /// event, so unexpected output is surfaced rather than dropped.
+  static Iperf3Event parse(String line) {
+    final trimmed = line.trim();
+    if (trimmed.isEmpty) return const Iperf3LogLine('');
+    if (!trimmed.startsWith('{')) return Iperf3LogLine(trimmed);
+
+    final Object? decoded;
+    try {
+      decoded = jsonDecode(trimmed);
+    } on FormatException {
+      return Iperf3LogLine(trimmed);
+    }
+    if (decoded is! Map<String, dynamic>) return Iperf3LogLine(trimmed);
+
+    final event = decoded['event'];
+    final data = decoded['data'];
+    if (event is! String) return Iperf3LogLine(trimmed);
+    final map = data is Map<String, dynamic> ? data : const <String, dynamic>{};
+
+    return switch (event) {
+      'start' => Iperf3Start.fromJson(map),
+      'interval' => Iperf3Interval.fromJson(map),
+      'end' => Iperf3End.fromJson(map),
+      'error' => Iperf3Error(_stringOf(data) ?? 'unknown error'),
+      'jni_stopped' => const Iperf3Stopped(),
+      _ => Iperf3LogLine(trimmed),
+    };
+  }
+
+  static String? _stringOf(Object? value) {
+    if (value is String) return value;
+    if (value is Map<String, dynamic>) {
+      final error = value['error'];
+      if (error is String) return error;
+    }
+    return null;
+  }
+}
+
+/// Connection details, emitted once at the start of a test.
+class Iperf3Start extends Iperf3Event {
+  final String? targetHost;
+  final int? targetPort;
+  final int? streamCount;
+  final int? durationSeconds;
+  final String? protocol;
+
+  const Iperf3Start({
+    this.targetHost,
+    this.targetPort,
+    this.streamCount,
+    this.durationSeconds,
+    this.protocol,
+  });
+
+  factory Iperf3Start.fromJson(Map<String, dynamic> data) {
+    final connecting = data['connecting_to'];
+    final testStart = data['test_start'];
+    return Iperf3Start(
+      targetHost: connecting is Map ? connecting['host'] as String? : null,
+      targetPort: connecting is Map ? _asInt(connecting['port']) : null,
+      streamCount: testStart is Map ? _asInt(testStart['num_streams']) : null,
+      durationSeconds: testStart is Map ? _asInt(testStart['duration']) : null,
+      protocol: testStart is Map ? testStart['protocol'] as String? : null,
+    );
+  }
+}
+
+/// One reporting interval, summed across all streams.
+class Iperf3Interval extends Iperf3Event {
+  final double start;
+  final double end;
+  final int bytes;
+  final double bitsPerSecond;
+  final int? retransmits;
+  final bool omitted;
+
+  /// Per-stream figures, when more than one stream is running.
+  final List<Iperf3StreamInterval> streams;
+
+  const Iperf3Interval({
+    required this.start,
+    required this.end,
+    required this.bytes,
+    required this.bitsPerSecond,
+    this.retransmits,
+    this.omitted = false,
+    this.streams = const [],
+  });
+
+  factory Iperf3Interval.fromJson(Map<String, dynamic> data) {
+    // Non-bidirectional runs use "sum"; bidir adds "sum_bidir_reverse".
+    final sum = data['sum'] ?? data['sum_bidir_reverse'];
+    final sumMap = sum is Map<String, dynamic> ? sum : const <String, dynamic>{};
+    final rawStreams = data['streams'];
+
+    return Iperf3Interval(
+      start: _asDouble(sumMap['start']) ?? 0,
+      end: _asDouble(sumMap['end']) ?? 0,
+      bytes: _asInt(sumMap['bytes']) ?? 0,
+      bitsPerSecond: _asDouble(sumMap['bits_per_second']) ?? 0,
+      retransmits: _asInt(sumMap['retransmits']),
+      omitted: sumMap['omitted'] == true,
+      streams: rawStreams is List
+          ? rawStreams
+              .whereType<Map<String, dynamic>>()
+              .map(Iperf3StreamInterval.fromJson)
+              .toList(growable: false)
+          : const [],
+    );
+  }
+}
+
+/// One stream's contribution to an interval.
+class Iperf3StreamInterval {
+  final int? socket;
+  final int bytes;
+  final double bitsPerSecond;
+  final int? retransmits;
+  final int? sendCwnd;
+  final int? rtt;
+
+  const Iperf3StreamInterval({
+    this.socket,
+    required this.bytes,
+    required this.bitsPerSecond,
+    this.retransmits,
+    this.sendCwnd,
+    this.rtt,
+  });
+
+  factory Iperf3StreamInterval.fromJson(Map<String, dynamic> data) {
+    return Iperf3StreamInterval(
+      socket: _asInt(data['socket']),
+      bytes: _asInt(data['bytes']) ?? 0,
+      bitsPerSecond: _asDouble(data['bits_per_second']) ?? 0,
+      retransmits: _asInt(data['retransmits']),
+      sendCwnd: _asInt(data['snd_cwnd']),
+      rtt: _asInt(data['rtt']),
+    );
+  }
+}
+
+/// Totals for the whole test, emitted once at the end.
+class Iperf3End extends Iperf3Event {
+  final Iperf3Total? sent;
+  final Iperf3Total? received;
+  final double? hostCpuPercent;
+  final double? remoteCpuPercent;
+
+  const Iperf3End({
+    this.sent,
+    this.received,
+    this.hostCpuPercent,
+    this.remoteCpuPercent,
+  });
+
+  factory Iperf3End.fromJson(Map<String, dynamic> data) {
+    final cpu = data['cpu_utilization_percent'];
+    return Iperf3End(
+      sent: Iperf3Total.fromJson(
+          data['sum_sent'] ?? data['sum_sent_bidir_reverse']),
+      received: Iperf3Total.fromJson(
+          data['sum_received'] ?? data['sum_received_bidir_reverse']),
+      hostCpuPercent: cpu is Map ? _asDouble(cpu['host_total']) : null,
+      remoteCpuPercent: cpu is Map ? _asDouble(cpu['remote_total']) : null,
+    );
+  }
+}
+
+/// A sent or received total from the `end` event.
+class Iperf3Total {
+  final double seconds;
+  final int bytes;
+  final double bitsPerSecond;
+  final int? retransmits;
+
+  const Iperf3Total({
+    required this.seconds,
+    required this.bytes,
+    required this.bitsPerSecond,
+    this.retransmits,
+  });
+
+  static Iperf3Total? fromJson(Object? data) {
+    if (data is! Map<String, dynamic>) return null;
+    return Iperf3Total(
+      seconds: _asDouble(data['seconds']) ?? 0,
+      bytes: _asInt(data['bytes']) ?? 0,
+      bitsPerSecond: _asDouble(data['bits_per_second']) ?? 0,
+      retransmits: _asInt(data['retransmits']),
+    );
+  }
+}
+
+/// iperf3 reported an error for this test.
+class Iperf3Error extends Iperf3Event {
+  final String message;
+  const Iperf3Error(this.message);
+}
+
+/// The run ended because the user stopped it.
+class Iperf3Stopped extends Iperf3Event {
+  const Iperf3Stopped();
+}
+
+/// Output that was not a recognised JSON event.
+class Iperf3LogLine extends Iperf3Event {
+  final String text;
+  const Iperf3LogLine(this.text);
+}
+
+int? _asInt(Object? value) => switch (value) {
+      int v => v,
+      double v => v.round(),
+      String v => int.tryParse(v),
+      _ => null,
+    };
+
+double? _asDouble(Object? value) => switch (value) {
+      double v => v,
+      int v => v.toDouble(),
+      String v => double.tryParse(v),
+      _ => null,
+    };