From 7d8d06e2e5ba0207c2a01558288302b822f21845 Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Wed, 26 Aug 2026 19:35:10 +0900 Subject: [PATCH] feat(app): parse json-stream reports into structured events MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- lib/data/services/iperf3_service.dart | 56 ++--- lib/domain/models/iperf3_event.dart | 238 ++++++++++++++++++++++ test/domain/models/iperf3_event_test.dart | 107 ++++++++++ 3 files changed, 375 insertions(+), 26 deletions(-) create mode 100644 lib/domain/models/iperf3_event.dart create mode 100644 test/domain/models/iperf3_event_test.dart diff --git a/lib/data/services/iperf3_service.dart b/lib/data/services/iperf3_service.dart index a3e6296..a9b3036 100644 --- a/lib/data/services/iperf3_service.dart +++ b/lib/data/services/iperf3_service.dart @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:flutter/foundation.dart'; import 'package:flutter/services.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; import 'package:iperf3gui/domain/models/iperf3_result.dart'; import 'package:iperf3gui/domain/models/test_configuration.dart'; @@ -10,9 +11,11 @@ import 'package:iperf3gui/domain/models/test_configuration.dart'; /// On Android, iperf3 is compiled into the app as a shared library (.so). /// This avoids SELinux restrictions on executing ELF binaries from assets. /// -/// The service talks to the native layer over a MethodChannel. Native output -/// arrives line by line on [outputLines] while the test runs; [run] resolves -/// with the accumulated transcript once the native layer reports completion. +/// The service talks to the native layer over a MethodChannel. The test runs +/// in `--json-stream` mode, so each native report arrives as one JSON object; +/// they are parsed and published on [events] while the test runs. [run] +/// resolves once the native layer reports completion, carrying the raw JSON +/// lines as the result's output. class Iperf3Service { Iperf3Service() { _channel.setMethodCallHandler(_handleNativeCall); @@ -33,16 +36,15 @@ class Iperf3Service { /// Exit code reported when the native layer never signalled completion. static const int exitCodeTimeout = -3; - /// Lines emitted by the native layer during the current run. - final StreamController _outputController = - StreamController.broadcast(); + /// Events parsed from the native layer during the current run. + final StreamController _eventController = + StreamController.broadcast(); - /// Live output, one line per event. Use this to render progress; [run]'s - /// result carries the same text joined together. - Stream get outputLines => _outputController.stream; + /// Live reports, one event per native report. Use this to render progress. + Stream get events => _eventController.stream; - /// Transcript of the current (or most recent) run. - final List _outputBuffer = []; + /// Raw JSON lines received during the current (or most recent) run. + final List _rawLines = []; /// Completer that resolves when the native iperf3 test finishes. Completer? _runCompleter; @@ -71,7 +73,7 @@ class Iperf3Service { /// Release the stream controller. Call from the owner's dispose(). Future dispose() async { _channel.setMethodCallHandler(null); - await _outputController.close(); + await _eventController.close(); } /// Run iperf3 with the given configuration. @@ -88,7 +90,7 @@ class Iperf3Service { _isRunning = true; _sawError = false; _stopRequested = false; - _outputBuffer.clear(); + _rawLines.clear(); final args = config.toCliArgs(); debugPrint('Iperf3Service: running: ${args.join(" ")}'); @@ -121,9 +123,9 @@ class Iperf3Service { debugPrint('Iperf3Service: error stopping after timeout: $e'); return null; }); - _emit('[ERROR] Timed out waiting for iperf3 to finish.'); + _publish(const Iperf3Error('Timed out waiting for iperf3 to finish.')); return Iperf3Result( - rawOutput: _transcript(), + rawOutput: _rawOutput(), exitCode: exitCodeTimeout, ); }, @@ -145,7 +147,12 @@ class Iperf3Service { case 'onOutput': final output = call.arguments['output'] as String?; if (output != null) { - _emit(output); + _rawLines.add(output); + final event = Iperf3Event.parse(output); + // iperf3 reports a failed test as an `error` event rather than a + // non-zero return, so this is where most failures surface. + if (event is Iperf3Error) _sawError = true; + _publish(event); } case 'onError': final error = call.arguments['error'] as String?; @@ -153,7 +160,7 @@ class Iperf3Service { // Not terminal on its own — iperf3 reports the failure reason and // then completes, so wait for onComplete before resolving. _sawError = true; - _emit('[ERROR] $error'); + _publish(Iperf3Error(error)); } case 'onComplete': debugPrint('Iperf3Service: test completed'); @@ -161,7 +168,7 @@ class Iperf3Service { if (completer != null && !completer.isCompleted) { completer.complete( Iperf3Result( - rawOutput: _transcript(), + rawOutput: _rawOutput(), // A user-requested stop is a normal outcome, not a failure. exitCode: (_sawError && !_stopRequested) ? exitCodeNativeError : 0, @@ -174,17 +181,14 @@ class Iperf3Service { } } - void _emit(String line) { - _outputBuffer.add(line); - if (!_outputController.isClosed) { - _outputController.add(line); + void _publish(Iperf3Event event) { + if (!_eventController.isClosed) { + _eventController.add(event); } - debugPrint('Iperf3Service output: ${line.trimRight()}'); } - /// Native output already carries its own newlines; keep it verbatim except - /// for lines we synthesised ourselves. - String _transcript() => _outputBuffer + /// The raw JSON stream as received, one event per line. + String _rawOutput() => _rawLines .map((line) => line.endsWith('\n') ? line : '$line\n') .join(); diff --git a/lib/domain/models/iperf3_event.dart b/lib/domain/models/iperf3_event.dart new file mode 100644 index 0000000..36147cc --- /dev/null +++ b/lib/domain/models/iperf3_event.dart @@ -0,0 +1,238 @@ +import 'dart:convert'; + +/// One report from iperf3's `--json-stream` output. +/// +/// iperf3 emits `{"event": "", "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) return Iperf3LogLine(trimmed); + + final event = decoded['event']; + final data = decoded['data']; + if (event is! String) return Iperf3LogLine(trimmed); + final map = data is Map ? data : const {}; + + 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) { + 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 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 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 data) { + // Non-bidirectional runs use "sum"; bidir adds "sum_bidir_reverse". + final sum = data['sum'] ?? data['sum_bidir_reverse']; + final sumMap = sum is Map ? sum : const {}; + 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(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 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 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) 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, + }; diff --git a/test/domain/models/iperf3_event_test.dart b/test/domain/models/iperf3_event_test.dart new file mode 100644 index 0000000..a5de174 --- /dev/null +++ b/test/domain/models/iperf3_event_test.dart @@ -0,0 +1,107 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; + +void main() { + group('Iperf3Event.parse', () { + test('parses a start event', () { + const line = '{"event":"start","data":{' + '"connecting_to":{"host":"192.168.2.182","port":5201},' + '"test_start":{"protocol":"TCP","num_streams":4,"duration":10}}}'; + final event = Iperf3Event.parse(line); + expect(event, isA()); + final start = event as Iperf3Start; + expect(start.targetHost, equals('192.168.2.182')); + expect(start.targetPort, equals(5201)); + expect(start.streamCount, equals(4)); + expect(start.durationSeconds, equals(10)); + expect(start.protocol, equals('TCP')); + }); + + test('parses an interval event with per-stream detail', () { + const line = '{"event":"interval","data":{' + '"streams":[{"socket":143,"bytes":49283072,' + '"bits_per_second":394264576.0,"retransmits":0,"snd_cwnd":2023424,' + '"rtt":3521}],' + '"sum":{"start":0.0,"end":1.000151,"seconds":1.000151,' + '"bytes":98566144,"bits_per_second":788410352.0,"retransmits":0,' + '"omitted":false,"sender":true}}}'; + final event = Iperf3Event.parse(line); + expect(event, isA()); + final interval = event as Iperf3Interval; + expect(interval.bytes, equals(98566144)); + expect(interval.bitsPerSecond, closeTo(788410352.0, 0.1)); + expect(interval.retransmits, equals(0)); + expect(interval.omitted, isFalse); + expect(interval.streams, hasLength(1)); + expect(interval.streams.single.socket, equals(143)); + expect(interval.streams.single.rtt, equals(3521)); + }); + + test('marks omitted intervals', () { + const line = '{"event":"interval","data":{"streams":[],' + '"sum":{"start":0.0,"end":1.0,"bytes":1,"bits_per_second":8.0,' + '"omitted":true}}}'; + final interval = Iperf3Event.parse(line) as Iperf3Interval; + expect(interval.omitted, isTrue); + }); + + test('parses an end event with sent and received totals', () { + const line = '{"event":"end","data":{' + '"sum_sent":{"seconds":10.0,"bytes":987654321,' + '"bits_per_second":790123456.0,"retransmits":7},' + '"sum_received":{"seconds":10.01,"bytes":987000000,' + '"bits_per_second":789000000.0},' + '"cpu_utilization_percent":{"host_total":12.5,"remote_total":3.25}}}'; + final end = Iperf3Event.parse(line) as Iperf3End; + expect(end.sent!.bytes, equals(987654321)); + expect(end.sent!.retransmits, equals(7)); + expect(end.received!.bitsPerSecond, closeTo(789000000.0, 0.1)); + expect(end.hostCpuPercent, closeTo(12.5, 0.01)); + expect(end.remoteCpuPercent, closeTo(3.25, 0.01)); + }); + + test('falls back to bidir-reverse keys when present', () { + const line = '{"event":"interval","data":{"streams":[],' + '"sum_bidir_reverse":{"start":0.0,"end":1.0,"bytes":42,' + '"bits_per_second":336.0}}}'; + final interval = Iperf3Event.parse(line) as Iperf3Interval; + expect(interval.bytes, equals(42)); + }); + + test('parses an error event', () { + const line = '{"event":"error","data":"unable to connect to server"}'; + final event = Iperf3Event.parse(line); + expect(event, isA()); + expect((event as Iperf3Error).message, + equals('unable to connect to server')); + }); + + test('parses the bridge stop marker', () { + final event = Iperf3Event.parse('{"event":"jni_stopped","data":{}}'); + expect(event, isA()); + }); + + test('surfaces non-JSON output as a log line rather than dropping it', () { + final event = Iperf3Event.parse('iperf3: something unexpected'); + expect(event, isA()); + expect((event as Iperf3LogLine).text, + equals('iperf3: something unexpected')); + }); + + test('surfaces malformed JSON as a log line', () { + final event = Iperf3Event.parse('{"event":"interval",'); + expect(event, isA()); + }); + + test('treats an unknown event name as a log line', () { + final event = Iperf3Event.parse('{"event":"future_thing","data":{}}'); + expect(event, isA()); + }); + + test('tolerates a missing data object', () { + final event = Iperf3Event.parse('{"event":"interval"}'); + expect(event, isA()); + expect((event as Iperf3Interval).bytes, equals(0)); + }); + }); +}