From 3b01b0c4a6ac4f02c290ca98f0b89c6d3107e7ca Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Wed, 26 Aug 2026 19:03:22 +0900 Subject: [PATCH] fix(app): stream native output and release run state on every exit path _isRunning was never cleared on the timeout path, so a single timed-out run locked the app out of starting another until it was restarted. It is now cleared in a finally block covering all three exits. Output was buffered inside the service and only handed over once the run finished, so the UI showed nothing until the end despite the native layer streaming line by line. The service republishes each line on outputLines and the view model appends to a transcript as they arrive. The channel handler was re-registered on every run and dereferenced _runCompleter with `!`, which threw once a late callback arrived after the field had been nulled. It is registered once in the constructor and reads the completer through a local. Also drops the duplicated argument list in favour of TestConfiguration.toCliArgs(), so the flags the UI sends have one definition. Co-Authored-By: Claude Opus 5 --- lib/data/services/iperf3_service.dart | 179 +++++++++++------- .../test_run/test_run_view_model.dart | 59 +++++- .../test_run/test_run_view_model_test.dart | 146 ++++++++++++-- 3 files changed, 300 insertions(+), 84 deletions(-) diff --git a/lib/data/services/iperf3_service.dart b/lib/data/services/iperf3_service.dart index 2c22025..24e9e43 100644 --- a/lib/data/services/iperf3_service.dart +++ b/lib/data/services/iperf3_service.dart @@ -1,5 +1,4 @@ import 'dart:async'; -import 'dart:io'; import 'package:flutter/foundation.dart'; import 'package:flutter/services.dart'; @@ -11,13 +10,38 @@ 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 communicates with the native layer via a MethodChannel. -/// Output is streamed live to the Flutter UI through callbacks. +/// 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. class Iperf3Service { + Iperf3Service() { + _channel.setMethodCallHandler(_handleNativeCall); + } + static const MethodChannel _channel = MethodChannel('com.novoyuuparosk.iperf3gui/iperf'); - /// Collect output lines during a test run. + /// Exit code reported when the native layer signalled an error. + static const int exitCodeNativeError = 1; + + /// Exit code reported when another test is already in flight. + static const int exitCodeBusy = -2; + + /// Exit code reported when the test could not be started at all. + static const int exitCodeStartFailed = -1; + + /// 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(); + + /// 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; + + /// Transcript of the current (or most recent) run. final List _outputBuffer = []; /// Completer that resolves when the native iperf3 test finishes. @@ -25,10 +49,11 @@ class Iperf3Service { /// Whether iperf3 is currently running. bool _isRunning = false; - - /// Whether iperf3 is currently running. bool get isRunning => _isRunning; + /// Whether the native layer reported an error during this run. + bool _sawError = false; + /// Cancel any running test. Future stop() async { if (!_isRunning) return; @@ -39,104 +64,124 @@ class Iperf3Service { } } + /// Release the stream controller. Call from the owner's dispose(). + Future dispose() async { + _channel.setMethodCallHandler(null); + await _outputController.close(); + } + /// Run iperf3 with the given configuration. /// - /// Returns an [Iperf3Result] with stdout/stderr and exit code. + /// Returns an [Iperf3Result] with the full output and an exit code. Future run(TestConfiguration config) async { if (_isRunning) { - return Iperf3Result( + return const Iperf3Result( rawOutput: 'Another test is already running.', - exitCode: -2, + exitCode: exitCodeBusy, ); } _isRunning = true; + _sawError = false; _outputBuffer.clear(); - final args = [ - '-c', config.server, - '-p', config.port.toString(), - '-t', config.duration.inSeconds.toString(), - '-i', config.interval.toString(), - '-P', config.parallelStreams.toString(), - ]; - + final args = config.toCliArgs(); debugPrint('Iperf3Service: running: ${args.join(" ")}'); - // Set up the completer to await completion - _runCompleter = Completer(); + final completer = Completer(); + _runCompleter = completer; - // Listen for events from native layer - _channel.setMethodCallHandler((call) async { - if (!_runCompleter!.isCompleted) { - switch (call.method) { - case 'onOutput': - final output = call.arguments['output'] as String?; - if (output != null) { - _outputBuffer.add(output); - debugPrint('Iperf3Service output: ${output.trim()}'); - } - break; - case 'onError': - final error = call.arguments['error'] as String?; - if (error != null) { - debugPrint('Iperf3Service error: $error'); - _outputBuffer.add('[ERROR] $error'); - // Don't complete yet — the test might still be running - } - break; - case 'onComplete': - debugPrint('Iperf3Service: test completed'); - _isRunning = false; - _runCompleter!.complete( - Iperf3Result( - rawOutput: _outputBuffer.join(''), - exitCode: 0, - ), - ); - break; - } - } - }); - - // Start the test try { await _channel.invokeMethod('runIperf', {'args': args}); } catch (e, stackTrace) { debugPrint('Iperf3Service exception: $e'); debugPrint('Iperf3Service stack: $stackTrace'); - _isRunning = false; - if (!_runCompleter!.isCompleted) { - _runCompleter!.completeError(e); - } + _finishRun(); return Iperf3Result( rawOutput: 'Error starting iperf3: $e', - exitCode: -1, + exitCode: exitCodeStartFailed, ); } - // Wait for the native layer to signal completion (with a timeout) + // Wait for the native layer to signal completion. The grace period covers + // DNS resolution and the 30s native connect timeout. try { - final result = await _runCompleter!.future.timeout( - config.duration + const Duration(seconds: 30), + return await completer.future.timeout( + config.duration + const Duration(seconds: 45), onTimeout: () { debugPrint('Iperf3Service: timeout, stopping test'); - stop(); + // Ask the native layer to wind down, then release local state so a + // stuck run can never lock the UI out of starting another test. + _channel.invokeMethod('stopIperf').catchError((Object e) { + debugPrint('Iperf3Service: error stopping after timeout: $e'); + return null; + }); + _emit('[ERROR] Timed out waiting for iperf3 to finish.'); return Iperf3Result( - rawOutput: 'Test timed out after ${config.duration.inSeconds}s', - exitCode: -3, + rawOutput: _transcript(), + exitCode: exitCodeTimeout, ); }, ); - return result; } catch (e) { debugPrint('Iperf3Service: error waiting for completion: $e'); return Iperf3Result( rawOutput: 'Test failed: $e', - exitCode: -1, + exitCode: exitCodeStartFailed, ); } finally { - _runCompleter = null; + _finishRun(); } } + + /// Handle a callback pushed up from the Kotlin/JNI layer. + Future _handleNativeCall(MethodCall call) async { + switch (call.method) { + case 'onOutput': + final output = call.arguments['output'] as String?; + if (output != null) { + _emit(output); + } + case 'onError': + final error = call.arguments['error'] as String?; + if (error != null) { + // Not terminal on its own — iperf3 reports the failure reason and + // then completes, so wait for onComplete before resolving. + _sawError = true; + _emit('[ERROR] $error'); + } + case 'onComplete': + debugPrint('Iperf3Service: test completed'); + final completer = _runCompleter; + if (completer != null && !completer.isCompleted) { + completer.complete( + Iperf3Result( + rawOutput: _transcript(), + exitCode: _sawError ? exitCodeNativeError : 0, + ), + ); + } + default: + debugPrint('Iperf3Service: unhandled native call ${call.method}'); + } + } + + void _emit(String line) { + _outputBuffer.add(line); + if (!_outputController.isClosed) { + _outputController.add(line); + } + debugPrint('Iperf3Service output: ${line.trimRight()}'); + } + + /// Native output already carries its own newlines; keep it verbatim except + /// for lines we synthesised ourselves. + String _transcript() => _outputBuffer + .map((line) => line.endsWith('\n') ? line : '$line\n') + .join(); + + void _finishRun() { + _isRunning = false; + _runCompleter = null; + } } diff --git a/lib/ui/features/test_run/test_run_view_model.dart b/lib/ui/features/test_run/test_run_view_model.dart index 04f2aa0..d247b10 100644 --- a/lib/ui/features/test_run/test_run_view_model.dart +++ b/lib/ui/features/test_run/test_run_view_model.dart @@ -1,15 +1,26 @@ +import 'dart:async'; + import 'package:flutter/foundation.dart'; import 'package:iperf3gui/data/services/iperf3_service.dart'; -import 'package:iperf3gui/domain/models/test_configuration.dart'; import 'package:iperf3gui/domain/models/iperf3_result.dart'; +import 'package:iperf3gui/domain/models/test_configuration.dart'; /// State of the test run. enum TestRunState { idle, running, completed, error } /// ViewModel for managing test run state. +/// +/// Subscribes to [Iperf3Service.outputLines] so the transcript grows while the +/// test runs rather than appearing all at once at the end. class TestRunViewModel extends ChangeNotifier { final Iperf3Service iperf3Service; + TestRunViewModel({required this.iperf3Service}) { + _outputSubscription = iperf3Service.outputLines.listen(_onOutputLine); + } + + StreamSubscription? _outputSubscription; + TestRunState _state = TestRunState.idle; TestRunState get state => _state; @@ -19,12 +30,25 @@ class TestRunViewModel extends ChangeNotifier { TestConfiguration? _currentConfig; TestConfiguration? get currentConfig => _currentConfig; - TestRunViewModel({required this.iperf3Service}); + final StringBuffer _liveOutput = StringBuffer(); + + /// Output accumulated so far. Grows line by line during a run, and holds the + /// finished transcript once the test completes. + String get output => _liveOutput.toString(); + + /// Whether a test is currently in flight. + bool get isRunning => _state == TestRunState.running; /// Start a test with the given configuration. Future runTest(TestConfiguration config) async { + if (_state == TestRunState.running) return; + if (!config.isValid) { + _liveOutput + ..clear() + ..writeln('Invalid configuration: check the server address and port.'); _state = TestRunState.error; + _result = null; notifyListeners(); return; } @@ -32,26 +56,51 @@ class TestRunViewModel extends ChangeNotifier { _state = TestRunState.running; _currentConfig = config; _result = null; + _liveOutput.clear(); notifyListeners(); try { final result = await iperf3Service.run(config); _result = result; - _state = result.exitCode == 0 - ? TestRunState.completed - : TestRunState.error; + // Prefer the streamed transcript — it is what the user watched build up. + // Fall back to the service's copy if nothing streamed through. + if (_liveOutput.isEmpty && result.rawOutput.isNotEmpty) { + _liveOutput.write(result.rawOutput); + } + _state = + result.exitCode == 0 ? TestRunState.completed : TestRunState.error; } catch (e) { + _liveOutput.writeln('Test failed: $e'); _state = TestRunState.error; } finally { notifyListeners(); } } + /// Request that a running test stop early. + Future stopTest() async { + if (_state != TestRunState.running) return; + await iperf3Service.stop(); + } + /// Reset to idle state. void reset() { _state = TestRunState.idle; _result = null; _currentConfig = null; + _liveOutput.clear(); notifyListeners(); } + + void _onOutputLine(String line) { + _liveOutput.write(line.endsWith('\n') ? line : '$line\n'); + notifyListeners(); + } + + @override + void dispose() { + _outputSubscription?.cancel(); + iperf3Service.dispose(); + super.dispose(); + } } diff --git a/test/ui/features/test_run/test_run_view_model_test.dart b/test/ui/features/test_run/test_run_view_model_test.dart index 46b7532..77088c3 100644 --- a/test/ui/features/test_run/test_run_view_model_test.dart +++ b/test/ui/features/test_run/test_run_view_model_test.dart @@ -1,3 +1,5 @@ +import 'dart:async'; + import 'package:flutter_test/flutter_test.dart'; import 'package:iperf3gui/data/services/iperf3_service.dart'; import 'package:iperf3gui/domain/models/iperf3_result.dart'; @@ -7,29 +9,26 @@ import 'package:iperf3gui/ui/features/test_run/test_run_view_model.dart'; void main() { group('TestRunViewModel', () { test('starts in idle state', () { - final vm = TestRunViewModel( - iperf3Service: _MockIperf3Service(), - ); + final vm = TestRunViewModel(iperf3Service: _MockIperf3Service()); expect(vm.state, equals(TestRunState.idle)); expect(vm.result, isNull); expect(vm.currentConfig, isNull); + expect(vm.output, isEmpty); }); test('can be reset', () { - final vm = TestRunViewModel( - iperf3Service: _MockIperf3Service(), - ); + final vm = TestRunViewModel(iperf3Service: _MockIperf3Service()); vm.reset(); expect(vm.state, equals(TestRunState.idle)); }); test('invalid config sets error state without running', () async { - final vm = TestRunViewModel( - iperf3Service: _MockIperf3Service(), - ); + final mock = _MockIperf3Service(); + final vm = TestRunViewModel(iperf3Service: mock); const config = TestConfiguration(server: ''); await vm.runTest(config); expect(vm.state, equals(TestRunState.error)); + expect(mock.runCallCount, equals(0)); }); test('runTest with valid config calls service', () async { @@ -45,17 +44,140 @@ void main() { expect(vm.state, equals(TestRunState.completed)); expect(vm.result, isNotNull); }); + + test('non-zero exit code lands in error state', () async { + final mock = _MockIperf3Service( + onRun: (_) async => + const Iperf3Result(rawOutput: 'boom', exitCode: 1), + ); + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.state, equals(TestRunState.error)); + expect(vm.result!.exitCode, equals(1)); + }); + + test('streamed lines accumulate into output during the run', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit('Connecting to host 192.168.1.1\n'); + mock.emit('[ 5] 0.00-1.00 sec 11.2 MBytes 94.1 Mbits/sec\n'); + // Let the broadcast stream deliver before the run resolves. + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: 'ignored', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + + expect(vm.output, contains('Connecting to host')); + expect(vm.output, contains('94.1 Mbits/sec')); + // The streamed transcript wins over the service's own copy. + expect(vm.output, isNot(contains('ignored'))); + }); + + test('falls back to result output when nothing streamed', () async { + final mock = _MockIperf3Service( + onRun: (_) async => + const Iperf3Result(rawOutput: 'summary only', exitCode: 0), + ); + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.output, contains('summary only')); + }); + + test('output is cleared when a new run starts', () async { + final mock = _MockIperf3Service( + onRun: (_) async => + const Iperf3Result(rawOutput: 'first', exitCode: 0), + ); + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.output, contains('first')); + + mock.onRun = (_) async => + const Iperf3Result(rawOutput: 'second', exitCode: 0); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.output, isNot(contains('first'))); + expect(vm.output, contains('second')); + }); + + test('stopTest forwards to the service only while running', () async { + final mock = _MockIperf3Service(); + final vm = TestRunViewModel(iperf3Service: mock); + + await vm.stopTest(); + expect(mock.stopCallCount, equals(0), reason: 'idle: nothing to stop'); + + final gate = Completer(); + mock.onRun = (_) => gate.future; + final run = vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.isRunning, isTrue); + + await vm.stopTest(); + expect(mock.stopCallCount, equals(1)); + + gate.complete(const Iperf3Result(rawOutput: 'stopped', exitCode: 0)); + await run; + }); + + test('a second runTest while running is ignored', () async { + final mock = _MockIperf3Service(); + final gate = Completer(); + mock.onRun = (_) => gate.future; + + final vm = TestRunViewModel(iperf3Service: mock); + final run = vm.runTest(const TestConfiguration(server: '192.168.1.1')); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(mock.runCallCount, equals(1)); + + gate.complete(const Iperf3Result(rawOutput: 'done', exitCode: 0)); + await run; + }); }); } class _MockIperf3Service implements Iperf3Service { _MockIperf3Service({this.onRun}); - final Future Function(TestConfiguration)? onRun; + Future Function(TestConfiguration)? onRun; + + final StreamController _controller = + StreamController.broadcast(); + + int runCallCount = 0; + int stopCallCount = 0; + bool disposed = false; + bool _isRunning = false; + + /// Push a line as if the native layer had emitted it. + void emit(String line) => _controller.add(line); + + @override + Stream get outputLines => _controller.stream; + + @override + bool get isRunning => _isRunning; @override Future run(TestConfiguration config) async { - if (onRun != null) return onRun!(config); - return const Iperf3Result(rawOutput: '', exitCode: 0); + runCallCount++; + _isRunning = true; + try { + if (onRun != null) return await onRun!(config); + return const Iperf3Result(rawOutput: '', exitCode: 0); + } finally { + _isRunning = false; + } + } + + @override + Future stop() async { + stopCallCount++; + } + + @override + Future dispose() async { + disposed = true; + await _controller.close(); } }