diff --git a/android/app/src/main/cpp/iperf_jni.c b/android/app/src/main/cpp/iperf_jni.c index 5b4cfac..ec6dca4 100644 --- a/android/app/src/main/cpp/iperf_jni.c +++ b/android/app/src/main/cpp/iperf_jni.c @@ -5,16 +5,25 @@ * the Flutter app to run iperf3 tests natively inside the app process * (no Process.start needed, no SELinux restrictions). * + * OUTPUT NOTE: + * The test runs in --json-stream mode with a json_callback installed, so + * iperf3 hands each event ("start", "interval", "end", "error") straight to + * us as a JSON string and never touches test->outfile. That is what the hook + * exists for -- iperf.h calls it "allow user apps to receive the JSON + * strings, instead of writing them to the output file". + * + * An earlier version piped test->outfile to a reader thread instead. That + * required closing the pipe's write end before joining the reader (which + * blocks in fgets() until EOF) and single ownership of the shared state + * struct. Both were gotten wrong and deadlocked every test. Reading events + * from the callback removes the pipe, the reader thread, and that whole class + * of ordering bug. + * * THREADING NOTE: * runIperfLive() BLOCKS for the whole duration of the test. Kotlin must * call it from a worker thread, never from the Flutter platform thread. - * - * PIPE / SHUTDOWN NOTE: - * iperf's output goes to a pipe drained by a reader thread that blocks in - * fgets(). The reader only sees EOF once the write end is closed, so the - * write end MUST be closed before pthread_join() — joining first deadlocks. - * The state struct is owned solely by the JNI caller, which frees it after - * joining the reader; the reader must never free it. + * json_callback fires on whichever thread iperf3 reports from, so it resolves + * a JNIEnv for the calling thread rather than reusing a stored one. * * SIGNAL HANDLING NOTE: * iperf3's run() function (called from iperf_run_client()) uses @@ -61,21 +70,22 @@ static pthread_mutex_t g_test_mutex = PTHREAD_MUTEX_INITIALIZER; static volatile int g_stop = 0; static char g_tmp_template[1024] = "/data/local/tmp/iperf3.XXXXXX"; -/* Per-call state shared between the JNI caller and the reader thread. - * We store the callback object (global ref) and method IDs. +/* Per-call state, reachable from the json_callback. * Callbacks are invoked via CallVoidMethod, NOT as function pointers. */ struct iperf_run_state { jobject callback_ref; /* Global ref to the callback object */ jmethodID mid_output; /* Method ID for onOutput(String) */ jmethodID mid_error; /* Method ID for onError(String) */ jmethodID mid_complete; /* Method ID for onComplete() */ - FILE *outfile; /* Pipe write end; NULL once closed */ - int pipe_fd[2]; JavaVM *jvm; - pthread_t reader_thread; - int reader_started; /* Only join if pthread_create succeeded */ }; +/* iperf3's json_callback signature carries no user-data pointer, so the + * current run's state has to be reachable from a global. Only one test runs + * at a time (enforced in runIperfLive), and this is published before + * iperf_run_client() and cleared after it returns. */ +static struct iperf_run_state *g_current_state = NULL; + /* Helper: invoke the output callback via CallVoidMethod */ static void signal_output(JNIEnv *env, struct iperf_run_state *state, const char *msg) { @@ -108,50 +118,36 @@ static void signal_complete(JNIEnv *env, struct iperf_run_state *state) { } /* ------------------------------------------------------------------ */ -/* Output reader thread — reads from iperf's output pipe and forwards */ -/* each line to the Java callback via JNI. */ +/* json_callback — receives one JSON event string per report from */ +/* iperf3 and forwards it to Java verbatim. Parsing happens in Dart. */ /* */ -/* Exits when the write end of the pipe is closed (EOF). Does NOT own */ -/* `state`: the JNI caller frees it after joining this thread. */ +/* Runs on iperf3's reporting thread. In the client that is the thread */ +/* that called iperf_run_client() (already attached, since Kotlin's */ +/* executor thread is a Java thread), but GetEnv is checked rather */ +/* than assumed, with an attach as fallback. */ /* ------------------------------------------------------------------ */ -static void *reader_thread_func(void *arg) { - struct iperf_run_state *state = (struct iperf_run_state *) arg; - JNIEnv *env; - int ret; +static void json_event_callback(struct iperf_test *test, char *json) { + struct iperf_run_state *state = g_current_state; + if (!state || !json) return; - /* Attach this native thread to the Java VM */ - ret = (*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL); - if (ret != JNI_OK) { - LOGE("AttachCurrentThread failed: %d", ret); - /* Still drain the pipe so writers never block on a full pipe. */ - char drain[1024]; - while (read(state->pipe_fd[0], drain, sizeof(drain)) > 0) { } - close(state->pipe_fd[0]); - return NULL; - } + JNIEnv *env = NULL; + int attached = 0; - char buf[1024]; - FILE *fp = fdopen(state->pipe_fd[0], "r"); - if (fp) { - while (fgets(buf, sizeof(buf), fp) != NULL) { - /* Skip empty lines */ - if (buf[0] == '\n' || buf[0] == '\0') - continue; - jstring line = (*env)->NewStringUTF(env, buf); - if (line != NULL) { - (*env)->CallVoidMethod(env, state->callback_ref, - state->mid_output, line); - (*env)->DeleteLocalRef(env, line); - } + if ((*state->jvm)->GetEnv(state->jvm, (void **) &env, JNI_VERSION_1_6) + != JNI_OK) { + if ((*state->jvm)->AttachCurrentThread(state->jvm, &env, NULL) + != JNI_OK) { + LOGE("json_event_callback: could not reach a JNIEnv"); + return; } - fclose(fp); /* also closes pipe_fd[0] */ - } else { - close(state->pipe_fd[0]); + attached = 1; } - /* Detach this native thread */ - (*state->jvm)->DetachCurrentThread(state->jvm); - return NULL; + signal_output(env, state, json); + + if (attached) { + (*state->jvm)->DetachCurrentThread(state->jvm); + } } /* ------------------------------------------------------------------ */ @@ -198,30 +194,12 @@ static void free_argv(char **argv, int argc) { free(argv); } -/* ------------------------------------------------------------------ */ -/* Close the pipe write end. Idempotent — both the normal and the */ -/* longjmp path call it, and the second call must be a no-op. */ -/* */ -/* Closing this is what makes the reader thread see EOF and exit, so */ -/* it must happen before any pthread_join(). */ -/* ------------------------------------------------------------------ */ -static void close_outfile(struct iperf_run_state *state) { - if (state && state->outfile) { - fflush(state->outfile); - fclose(state->outfile); - state->outfile = NULL; - } -} - /* Retire the global test pointer and free it, under the mutex. */ static void retire_test(void) { pthread_mutex_lock(&g_test_mutex); struct iperf_test *test = g_test; g_test = NULL; if (test) { - /* outfile is a closed FILE* by now — point it somewhere harmless so - * nothing in the teardown path can write to a dead stream. */ - test->outfile = stdout; iperf_free_test(test); } pthread_mutex_unlock(&g_test_mutex); @@ -236,14 +214,8 @@ static void iperf_cleanup_and_signal(JNIEnv *env, struct iperf_run_state *state, char **argv, int argc, int ret, int stopped) { - /* Close the write end FIRST: the reader thread is blocked in fgets() - * and only unblocks on EOF. Joining before this would deadlock. */ - close_outfile(state); - - if (state->reader_started) { - pthread_join(state->reader_thread, NULL); - state->reader_started = 0; - } + /* No more events can be delivered once the run is over. */ + g_current_state = NULL; /* Restore default signal handlers before g_sigend_jmp_buf goes stale. */ iperf_catch_sigend(SIG_DFL); @@ -259,15 +231,11 @@ static void iperf_cleanup_and_signal(JNIEnv *env, /* Signal result to Java */ if (user_stopped) { - signal_output(env, state, "[iPerf JNI] Test was stopped by user.\n"); + signal_output(env, state, "{\"event\":\"jni_stopped\",\"data\":{}}"); } else if (ret < 0) { char errbuf[256]; - snprintf(errbuf, sizeof(errbuf), - "[iPerf JNI] Test failed: %s\n", - iperf_strerror(i_errno)); + snprintf(errbuf, sizeof(errbuf), "%s", iperf_strerror(i_errno)); signal_error(env, state, errbuf); - } else { - signal_output(env, state, "[iPerf JNI] Test completed successfully.\n"); } /* Signal completion to Java */ @@ -325,20 +293,20 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( if (mid_complete) (*env)->CallVoidMethod(env, callback, mid_complete); \ } while (0) - /* Refuse to start a second concurrent test — the globals below hold + /* Refuse to start a second concurrent test — the globals above hold * exactly one test's worth of state. */ pthread_mutex_lock(&g_test_mutex); int busy = (g_test != NULL); pthread_mutex_unlock(&g_test_mutex); if (busy) { - FAIL_EARLY("[iPerf JNI] A test is already running."); + FAIL_EARLY("A test is already running."); return; } /* Create the iperf test object */ struct iperf_test *test = iperf_new_test(); if (!test) { - FAIL_EARLY("[iPerf JNI] Failed to create iperf test"); + FAIL_EARLY("Failed to create iperf test"); return; } @@ -352,7 +320,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( char **argv = build_argv(env, arguments, &total_argc); if (!argv) { retire_test(); - FAIL_EARLY("[iPerf JNI] Failed to allocate argv"); + FAIL_EARLY("Failed to allocate argv"); return; } @@ -365,19 +333,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( if (iperf_parse_arguments(test, total_argc, argv) < 0) { char errbuf[256]; snprintf(errbuf, sizeof(errbuf), - "[iPerf JNI] Argument parse error: %s", iperf_strerror(i_errno)); - free_argv(argv, total_argc); - retire_test(); - FAIL_EARLY(errbuf); - return; - } - - /* Create a pipe to capture iperf's output */ - int pipefd[2]; - if (pipe(pipefd) != 0) { - char errbuf[256]; - snprintf(errbuf, sizeof(errbuf), - "[iPerf JNI] Failed to create output pipe: %s", strerror(errno)); + "Argument parse error: %s", iperf_strerror(i_errno)); free_argv(argv, total_argc); retire_test(); FAIL_EARLY(errbuf); @@ -389,25 +345,11 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( calloc(1, sizeof(struct iperf_run_state)); if (!state) { free_argv(argv, total_argc); - close(pipefd[0]); - close(pipefd[1]); retire_test(); - FAIL_EARLY("[iPerf JNI] Failed to allocate run state"); + FAIL_EARLY("Failed to allocate run state"); return; } - FILE *outfile = fdopen(pipefd[1], "w"); - if (!outfile) { - free(state); - free_argv(argv, total_argc); - close(pipefd[0]); - close(pipefd[1]); - retire_test(); - FAIL_EARLY("[iPerf JNI] Failed to open output stream"); - return; - } - - /* Get JavaVM for reader-thread attachment */ JavaVM *jvm; (*env)->GetJavaVM(env, &jvm); @@ -415,14 +357,16 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( state->mid_error = mid_error; state->mid_complete = mid_complete; state->callback_ref = (*env)->NewGlobalRef(env, callback); - state->outfile = outfile; - state->pipe_fd[0] = pipefd[0]; - state->pipe_fd[1] = pipefd[1]; state->jvm = jvm; - /* Redirect iperf's output to our pipe */ - test->outfile = state->outfile; - setvbuf(test->outfile, NULL, _IOLBF, 0); /* Line-buffered */ + /* Deliver results as a stream of JSON events straight to + * json_event_callback, rather than as text through test->outfile. + * --json-stream sets both of these flags, so setting json_stream alone + * would leave iperf3 printing human-readable text as well. */ + iperf_set_test_json_output(test, 1); + iperf_set_test_json_stream(test, 1); + iperf_set_test_json_callback(test, json_event_callback); + g_current_state = state; /* Set temp directory for stream buffers (mkstemp in iperf_new_stream). * On Android, /data/local/tmp is NOT writable from the app sandbox — @@ -430,22 +374,6 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( * the strdup internally. */ iperf_set_test_template(test, g_tmp_template); - /* Spawn reader thread to forward iperf output to Java. Started before - * the test so early output is not lost, and before setjmp so the - * longjmp path can rely on reader_started being accurate. */ - if (pthread_create(&state->reader_thread, NULL, - reader_thread_func, state) != 0) { - close_outfile(state); - close(pipefd[0]); - free_argv(argv, total_argc); - retire_test(); - (*env)->DeleteGlobalRef(env, state->callback_ref); - free(state); - FAIL_EARLY("[iPerf JNI] Failed to start output reader thread"); - return; - } - state->reader_started = 1; - #undef FAIL_EARLY /* @@ -462,8 +390,6 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( signal(SIGPIPE, SIG_IGN); /* Ignore SIGPIPE for socket errors */ iperf_catch_sigend(sigend_handler); - signal_output(env, state, "[iPerf JNI] Starting iPerf3 client...\n"); - g_stop = 0; int ret = iperf_run_client(test); LOGI("iperf_run_client returned: %d i_errno=%d strerror=%s", @@ -478,8 +404,8 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_runIperfLive( } else { /* * We got here via longjmp() from sigend_handler. - * iperf_cleanup_and_signal() closes the pipe, joins the reader and - * frees everything, so there is nothing to unwind by hand here. + * iperf_cleanup_and_signal() frees everything, so there is nothing + * to unwind by hand here. */ LOGI("Test interrupted by signal %d", g_signed_sig); g_stop = 1; @@ -515,19 +441,7 @@ Java_com_novoyuuparosk_iperf3gui_IperfRunner_forceStopIperfTest( pthread_mutex_unlock(&g_test_mutex); if (!requested) return; - - /* Notify Java */ - jclass callback_cls = (*env)->GetObjectClass(env, callback); - jmethodID mid_output = (*env)->GetMethodID( - env, callback_cls, "onOutput", "(Ljava/lang/String;)V"); - if (mid_output) { - jstring msg = (*env)->NewStringUTF(env, - "[iPerf JNI] Requested graceful stop.\n"); - if (msg) { - (*env)->CallVoidMethod(env, callback, mid_output, msg); - (*env)->DeleteLocalRef(env, msg); - } - } + LOGI("Graceful stop requested"); } /* ------------------------------------------------------------------ */ diff --git a/docs/README.md b/docs/README.md index 321cfbc..9728d37 100644 --- a/docs/README.md +++ b/docs/README.md @@ -42,8 +42,9 @@ User taps GO → Kotlin IperfRunner: submits to the iperf3-run worker thread, returns at once → JNI C: build_argv() (prepends argv[0]) → iperf_parse_arguments() → iperf_run_client(test) [blocks on the worker thread] - → Reader thread pipes output → CallVoidMethod → main-thread hop → Dart onOutput - → Dart streams each line to the UI via Iperf3Service.outputLines + → iperf3 json_callback → one JSON event per report → CallVoidMethod + → main-thread hop → Dart onOutput → Iperf3Event.parse() + → Iperf3Service.events → view model: transcript + interval series + summary → On completion: Dart Completer resolves with Iperf3Result ``` @@ -61,16 +62,51 @@ to the main looper — stalled every callback until the test was already over. ### Callback Mechanism The JNI bridge uses `CallVoidMethod` on stored `jmethodID`s to invoke the Kotlin `IperfCallback` interface, NOT function pointer casts. Direct function pointer casts caused SIGBUS crashes on ARM due to JNINativeInterface method ID layout differences. -### Output Streaming -iperf3 output is redirected via pipe to a background reader thread that attaches -to the Java VM via `AttachCurrentThread` and invokes callbacks via -`CallVoidMethod`. Dart re-publishes each line on `Iperf3Service.outputLines`, so -the transcript grows while the test runs. +### Output via json_callback, not a pipe +The test runs with `json_output` and `json_stream` set and a `json_callback` +installed, so iperf3 hands each report to the bridge as a JSON string and never +writes to `test->outfile`. `iperf.h` describes the hook as "allow user apps to +receive the JSON strings, instead of writing them to the output file" — this is +its intended use. -The reader blocks in `fgets()` and only unblocks on EOF, so the pipe's write end -must be closed **before** `pthread_join()` — joining first deadlocks the worker -thread. The JNI caller solely owns the run-state struct and frees it after the -join; the reader must never free it. +An earlier design piped `test->outfile` to a reader thread. That imposed two +constraints, both of which were gotten wrong and deadlocked every test: the +pipe's write end had to be closed *before* joining a reader blocked in +`fgets()`, and exactly one side could own the shared state struct. Reading from +the callback removes the pipe, the reader thread and both constraints. Don't +reintroduce it. + +`json_callback` carries no user-data pointer, so the run state is reached +through a global published before `iperf_run_client()` and cleared afterwards; +only one test may run at a time, which `runIperfLive` enforces. The callback +fires on iperf3's reporting thread, so it resolves a `JNIEnv` for the calling +thread rather than reusing a stored one. + +Dart parses each event into a sealed `Iperf3Event`; unrecognised output becomes +an `Iperf3LogLine` rather than being discarded. The UI rebuilds iperf3's text +layout from the events and keeps the interval series as data. + +### Embedding a CLI in a long-lived process +iperf3's `main()` runs one test and exits, so anything it leaves behind is +reclaimed by the OS. Linked into an app process that runs tests repeatedly for +hours, every such leftover becomes this app's problem. What that costs in +practice: + +- **Signal handlers are process-global.** `iperf_catch_sigend()` and + `signal(SIGPIPE, SIG_IGN)` change behaviour for the whole process, Flutter and + ART included. Handlers are restored to `SIG_DFL` when a test ends. +- **`g_sigend_jmp_buf` outlives its frame.** Harmless in a process about to + exit; a jump into a dead stack frame otherwise. Same fix as above. +- **Globals mean one test at a time.** `i_errno` and the bridge's own statics + hold a single test's worth of state, so concurrent runs are refused. + +What iperf3 itself handles correctly, verified rather than assumed: per-stream +threads are joined (`iperf_client_api.c`), stream buffer fds are closed, and the +`mkstemp` temp file is `unlink`ed immediately after creation +(`iperf_api.c:4652`), so nothing accumulates in the cache dir. Measured across +four consecutive 4-stream runs on device, thread count returned to baseline +(33 → 40 mid-run → 34) and fd count returned exactly to baseline (175 → 183 → +175). ### Stop Handling A stop shuts down the control socket, so `iperf_run_client()` returns -1 with 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/lib/ui/features/test_run/iperf3_event_formatter.dart b/lib/ui/features/test_run/iperf3_event_formatter.dart new file mode 100644 index 0000000..898533f --- /dev/null +++ b/lib/ui/features/test_run/iperf3_event_formatter.dart @@ -0,0 +1,95 @@ +import 'package:iperf3gui/domain/models/iperf3_event.dart'; + +/// Renders parsed iperf3 events as the kind of line iperf3 itself would print. +/// +/// The app consumes `--json-stream`, so the familiar text layout has to be +/// rebuilt here. Keeping it in one pure function makes it testable and leaves +/// the structured events untouched for anything else that wants them. +String? formatIperf3Event(Iperf3Event event) { + return switch (event) { + Iperf3Start e => _formatStart(e), + Iperf3Interval e => _formatInterval(e), + Iperf3End e => _formatEnd(e), + Iperf3Error e => 'iperf3 error: ${e.message}', + Iperf3Stopped _ => 'Stopped by user.', + Iperf3LogLine e => e.text.isEmpty ? null : e.text, + }; +} + +String _formatStart(Iperf3Start e) { + final parts = []; + if (e.targetHost != null) { + final port = e.targetPort != null ? ', port ${e.targetPort}' : ''; + parts.add('Connecting to host ${e.targetHost}$port'); + } + final detail = [ + if (e.protocol != null) e.protocol!, + if (e.streamCount != null) + '${e.streamCount} stream${e.streamCount == 1 ? '' : 's'}', + if (e.durationSeconds != null) '${e.durationSeconds}s', + ]; + if (detail.isNotEmpty) parts.add(detail.join(' · ')); + return parts.isEmpty ? '' : parts.join('\n'); +} + +String _formatInterval(Iperf3Interval e) { + final window = '${_seconds(e.start)}-${_seconds(e.end)} sec'; + final line = StringBuffer('[SUM] $window ' + '${formatBytes(e.bytes)} ${formatBitrate(e.bitsPerSecond)}'); + if (e.retransmits != null) { + line.write(' ${e.retransmits} retr'); + } + if (e.omitted) line.write(' (omitted)'); + return line.toString(); +} + +String _formatEnd(Iperf3End e) { + final lines = ['- ' * 20]; + final sent = e.sent; + final received = e.received; + if (sent != null) { + lines.add('[SUM] 0.00-${_seconds(sent.seconds)} sec ' + '${formatBytes(sent.bytes)} ${formatBitrate(sent.bitsPerSecond)}' + '${sent.retransmits != null ? ' ${sent.retransmits} retr' : ''}' + ' sender'); + } + if (received != null) { + lines.add('[SUM] 0.00-${_seconds(received.seconds)} sec ' + '${formatBytes(received.bytes)} ' + '${formatBitrate(received.bitsPerSecond)} receiver'); + } + if (e.hostCpuPercent != null) { + lines.add('CPU: local ${e.hostCpuPercent!.toStringAsFixed(1)}%' + '${e.remoteCpuPercent != null ? ', remote ${e.remoteCpuPercent!.toStringAsFixed(1)}%' : ''}'); + } + return lines.join('\n'); +} + +String _seconds(double value) => value.toStringAsFixed(2); + +/// Byte totals the way iperf3 prints them — `93.8 MBytes`. +String formatBytes(int bytes) { + const units = ['Bytes', 'KBytes', 'MBytes', 'GBytes', 'TBytes']; + var value = bytes.toDouble(); + var unit = 0; + while (value >= 1024 && unit < units.length - 1) { + value /= 1024; + unit++; + } + final precision = value >= 100 ? 0 : (value >= 10 ? 1 : 2); + return '${value.toStringAsFixed(precision)} ${units[unit]}'; +} + +/// Bitrates the way iperf3 prints them — `786 Mbits/sec`. Decimal units, since +/// network rates are powers of ten while transfer totals are powers of two. +String formatBitrate(double bitsPerSecond) { + const units = ['bits/sec', 'Kbits/sec', 'Mbits/sec', 'Gbits/sec']; + var value = bitsPerSecond; + var unit = 0; + while (value >= 1000 && unit < units.length - 1) { + value /= 1000; + unit++; + } + final precision = value >= 100 ? 0 : (value >= 10 ? 1 : 2); + return '${value.toStringAsFixed(precision)} ${units[unit]}'; +} diff --git a/lib/ui/features/test_run/test_run_screen.dart b/lib/ui/features/test_run/test_run_screen.dart index fb2a85d..16aa122 100644 --- a/lib/ui/features/test_run/test_run_screen.dart +++ b/lib/ui/features/test_run/test_run_screen.dart @@ -3,6 +3,7 @@ import 'package:flutter/services.dart'; import 'package:provider/provider.dart'; import 'package:iperf3gui/data/services/iperf3_service.dart'; import 'package:iperf3gui/domain/models/test_configuration.dart'; +import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart'; import 'package:iperf3gui/ui/features/test_run/test_run_view_model.dart'; /// Main test run screen. @@ -193,6 +194,35 @@ class _TestRunContentState extends State<_TestRunContent> { ), const SizedBox(height: 12), + // Live throughput figures, derived from the parsed interval series + Consumer( + builder: (context, vm, _) { + final intervals = vm.intervals; + if (intervals.isEmpty) return const SizedBox.shrink(); + final rates = + intervals.map((i) => i.bitsPerSecond).toList(growable: false); + final latest = rates.last; + final peak = rates.reduce((a, b) => a > b ? a : b); + final mean = rates.reduce((a, b) => a + b) / rates.length; + final summary = vm.summary?.sent; + return Padding( + padding: const EdgeInsets.only(bottom: 12), + child: Row( + mainAxisAlignment: MainAxisAlignment.spaceBetween, + children: [ + _Stat( + label: summary != null ? 'total' : 'now', + value: formatBitrate(summary?.bitsPerSecond ?? latest), + ), + _Stat(label: 'avg', value: formatBitrate(mean)), + _Stat(label: 'peak', value: formatBitrate(peak)), + _Stat(label: 'samples', value: '${rates.length}'), + ], + ), + ); + }, + ), + // Live output Expanded( child: Container( @@ -229,3 +259,32 @@ class _TestRunContentState extends State<_TestRunContent> { ); } } + + +/// One figure in the throughput row. +class _Stat extends StatelessWidget { + const _Stat({required this.label, required this.value}); + + final String label; + final String value; + + @override + Widget build(BuildContext context) { + final theme = Theme.of(context); + return Column( + crossAxisAlignment: CrossAxisAlignment.start, + children: [ + Text( + label, + style: theme.textTheme.labelSmall + ?.copyWith(color: theme.colorScheme.onSurfaceVariant), + ), + Text( + value, + style: theme.textTheme.titleSmall + ?.copyWith(fontFeatures: const [FontFeature.tabularFigures()]), + ), + ], + ); + } +} 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 ec267ec..965cb14 100644 --- a/lib/ui/features/test_run/test_run_view_model.dart +++ b/lib/ui/features/test_run/test_run_view_model.dart @@ -2,24 +2,33 @@ import 'dart:async'; import 'package:flutter/foundation.dart'; import 'package:iperf3gui/data/services/iperf3_service.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'; +import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.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. +/// Subscribes to [Iperf3Service.events] so the transcript and the interval +/// series both grow while the test runs, rather than appearing at the end. class TestRunViewModel extends ChangeNotifier { final Iperf3Service iperf3Service; + /// How often to notify listeners while output is arriving. iperf3 can report + /// several times a second across many streams, and each rebuild re-renders + /// the whole transcript — so repaints are coalesced instead of one per event. + static const Duration displayInterval = Duration(milliseconds: 250); + TestRunViewModel({required this.iperf3Service}) { - _outputSubscription = iperf3Service.outputLines.listen(_onOutputLine); + _eventSubscription = iperf3Service.events.listen(_onEvent); } - StreamSubscription? _outputSubscription; + StreamSubscription? _eventSubscription; + Timer? _repaintTimer; + bool _dirty = false; TestRunState _state = TestRunState.idle; TestRunState get state => _state; @@ -30,11 +39,26 @@ class TestRunViewModel extends ChangeNotifier { TestConfiguration? _currentConfig; TestConfiguration? get currentConfig => _currentConfig; - final StringBuffer _liveOutput = StringBuffer(); + final StringBuffer _transcript = 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(); + /// Human-readable output accumulated so far. + String get output => _transcript.toString(); + + final List _intervals = []; + + /// Per-interval throughput samples, in arrival order. Structured, so a chart + /// can be driven from this without re-parsing text. + List get intervals => List.unmodifiable(_intervals); + + Iperf3Start? _start; + + /// Connection details reported at the start of the run. + Iperf3Start? get start => _start; + + Iperf3End? _summary; + + /// Totals reported at the end of the run. + Iperf3End? get summary => _summary; /// Whether a test is currently in flight. bool get isRunning => _state == TestRunState.running; @@ -47,35 +71,32 @@ class TestRunViewModel extends ChangeNotifier { if (_state == TestRunState.running) return; if (!config.isValid) { - _liveOutput - ..clear() - ..writeln('Invalid configuration: check the server address and port.'); + _resetRunData(); + _transcript + .writeln('Invalid configuration: check the server address and port.'); _state = TestRunState.error; - _result = null; notifyListeners(); return; } + _resetRunData(); _state = TestRunState.running; _currentConfig = config; - _result = null; - _liveOutput.clear(); notifyListeners(); try { final result = await iperf3Service.run(config); _result = result; - // 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); + if (_transcript.isEmpty && result.rawOutput.isNotEmpty) { + _transcript.write(result.rawOutput); } _state = result.exitCode == 0 ? TestRunState.completed : TestRunState.error; } catch (e) { - _liveOutput.writeln('Test failed: $e'); + _transcript.writeln('Test failed: $e'); _state = TestRunState.error; } finally { + _stopRepaintTimer(); notifyListeners(); } } @@ -88,21 +109,63 @@ class TestRunViewModel extends ChangeNotifier { /// Reset to idle state. void reset() { + _resetRunData(); _state = TestRunState.idle; - _result = null; _currentConfig = null; - _liveOutput.clear(); notifyListeners(); } - void _onOutputLine(String line) { - _liveOutput.write(line.endsWith('\n') ? line : '$line\n'); - notifyListeners(); + void _resetRunData() { + _result = null; + _transcript.clear(); + _intervals.clear(); + _start = null; + _summary = null; + _dirty = false; + } + + void _onEvent(Iperf3Event event) { + switch (event) { + case Iperf3Start e: + _start = e; + case Iperf3Interval e: + // Omitted intervals are warm-up (-O) and are not part of the series. + if (!e.omitted) _intervals.add(e); + case Iperf3End e: + _summary = e; + case Iperf3Error _: + case Iperf3Stopped _: + case Iperf3LogLine _: + break; + } + + final line = formatIperf3Event(event); + if (line != null && line.isNotEmpty) { + _transcript.writeln(line); + } + _scheduleRepaint(); + } + + /// Coalesce repaints: mark dirty and let a timer flush at [displayInterval]. + void _scheduleRepaint() { + _dirty = true; + _repaintTimer ??= Timer.periodic(displayInterval, (_) { + if (!_dirty) return; + _dirty = false; + notifyListeners(); + }); + } + + void _stopRepaintTimer() { + _repaintTimer?.cancel(); + _repaintTimer = null; + _dirty = false; } @override void dispose() { - _outputSubscription?.cancel(); + _stopRepaintTimer(); + _eventSubscription?.cancel(); iperf3Service.dispose(); super.dispose(); } 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)); + }); + }); +} diff --git a/test/ui/features/test_run/iperf3_event_formatter_test.dart b/test/ui/features/test_run/iperf3_event_formatter_test.dart new file mode 100644 index 0000000..f5abc90 --- /dev/null +++ b/test/ui/features/test_run/iperf3_event_formatter_test.dart @@ -0,0 +1,87 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:iperf3gui/domain/models/iperf3_event.dart'; +import 'package:iperf3gui/ui/features/test_run/iperf3_event_formatter.dart'; + +void main() { + group('formatBytes', () { + test('scales in binary units the way iperf3 prints totals', () { + expect(formatBytes(0), equals('0.00 Bytes')); + expect(formatBytes(1024), equals('1.00 KBytes')); + expect(formatBytes(98566144), equals('94.0 MBytes')); + expect(formatBytes(1073741824), equals('1.00 GBytes')); + }); + }); + + group('formatBitrate', () { + test('scales in decimal units, as network rates are quoted', () { + expect(formatBitrate(0), equals('0.00 bits/sec')); + expect(formatBitrate(8000), equals('8.00 Kbits/sec')); + expect(formatBitrate(790000000), equals('790 Mbits/sec')); + expect(formatBitrate(2500000000), equals('2.50 Gbits/sec')); + }); + }); + + group('formatIperf3Event', () { + test('renders a start event', () { + const event = Iperf3Start( + targetHost: '192.168.2.182', + targetPort: 5201, + protocol: 'TCP', + streamCount: 1, + durationSeconds: 10, + ); + final text = formatIperf3Event(event)!; + expect(text, contains('Connecting to host 192.168.2.182, port 5201')); + expect(text, contains('TCP')); + expect(text, contains('1 stream')); + }); + + test('renders an interval with retransmits', () { + const event = Iperf3Interval( + start: 0, + end: 1, + bytes: 98566144, + bitsPerSecond: 788410352, + retransmits: 3, + ); + final text = formatIperf3Event(event)!; + expect(text, contains('0.00-1.00 sec')); + expect(text, contains('94.0 MBytes')); + expect(text, contains('788 Mbits/sec')); + expect(text, contains('3 retr')); + }); + + test('marks an omitted interval', () { + const event = Iperf3Interval( + start: 0, end: 1, bytes: 1, bitsPerSecond: 8, omitted: true); + expect(formatIperf3Event(event), contains('omitted')); + }); + + test('renders sender and receiver lines from an end event', () { + const event = Iperf3End( + sent: Iperf3Total( + seconds: 10, bytes: 987654321, bitsPerSecond: 790000000, + retransmits: 7), + received: Iperf3Total( + seconds: 10.01, bytes: 987000000, bitsPerSecond: 789000000), + hostCpuPercent: 12.5, + ); + final text = formatIperf3Event(event)!; + expect(text, contains('sender')); + expect(text, contains('receiver')); + expect(text, contains('7 retr')); + expect(text, contains('CPU: local 12.5%')); + }); + + test('renders errors and stops', () { + expect(formatIperf3Event(const Iperf3Error('boom')), + equals('iperf3 error: boom')); + expect(formatIperf3Event(const Iperf3Stopped()), + equals('Stopped by user.')); + }); + + test('drops an empty log line', () { + expect(formatIperf3Event(const Iperf3LogLine('')), isNull); + }); + }); +} 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 6f83d36..025019c 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 @@ -2,6 +2,7 @@ import 'dart:async'; import 'package:flutter_test/flutter_test.dart'; import 'package:iperf3gui/data/services/iperf3_service.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'; import 'package:iperf3gui/ui/features/test_run/test_run_view_model.dart'; @@ -56,11 +57,16 @@ void main() { expect(vm.result!.exitCode, equals(1)); }); - test('streamed lines accumulate into output during the run', () async { + test('streamed events 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'); + mock.emit(const Iperf3Start(targetHost: '192.168.1.1', targetPort: 5201)); + mock.emit(const Iperf3Interval( + start: 0, + end: 1, + bytes: 11744051, + bitsPerSecond: 94100000, + )); // Let the broadcast stream deliver before the run resolves. await Future.delayed(Duration.zero); return const Iperf3Result(rawOutput: 'ignored', exitCode: 0); @@ -69,12 +75,83 @@ void main() { 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('Connecting to host 192.168.1.1')); 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('intervals are collected as structured samples', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3Interval( + start: 0, end: 1, bytes: 1000, bitsPerSecond: 8000)); + mock.emit(const Iperf3Interval( + start: 1, end: 2, bytes: 2000, bitsPerSecond: 16000)); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + + expect(vm.intervals, hasLength(2)); + expect(vm.intervals[1].bitsPerSecond, equals(16000)); + }); + + test('omitted warm-up intervals are excluded from the series', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3Interval( + start: 0, end: 1, bytes: 1, bitsPerSecond: 1, omitted: true)); + mock.emit(const Iperf3Interval( + start: 1, end: 2, bytes: 2000, bitsPerSecond: 16000)); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + + expect(vm.intervals, hasLength(1)); + expect(vm.intervals.single.bitsPerSecond, equals(16000)); + }); + + test('end event is retained as the summary', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3End( + sent: Iperf3Total( + seconds: 10, bytes: 987654321, bitsPerSecond: 790000000), + )); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + + expect(vm.summary?.sent?.bitsPerSecond, equals(790000000)); + expect(vm.output, contains('790 Mbits/sec')); + }); + + test('a new run clears the previous interval series', () async { + final mock = _MockIperf3Service(); + mock.onRun = (_) async { + mock.emit(const Iperf3Interval( + start: 0, end: 1, bytes: 1000, bitsPerSecond: 8000)); + await Future.delayed(Duration.zero); + return const Iperf3Result(rawOutput: '', exitCode: 0); + }; + + final vm = TestRunViewModel(iperf3Service: mock); + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.intervals, hasLength(1)); + + await vm.runTest(const TestConfiguration(server: '192.168.1.1')); + expect(vm.intervals, hasLength(1), reason: 'not appended to the old run'); + }); + test('falls back to result output when nothing streamed', () async { final mock = _MockIperf3Service( onRun: (_) async => @@ -166,19 +243,19 @@ class _MockIperf3Service implements Iperf3Service { Future Function(TestConfiguration)? onRun; - final StreamController _controller = - StreamController.broadcast(); + 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); + /// Push an event as if the native layer had reported it. + void emit(Iperf3Event event) => _controller.add(event); @override - Stream get outputLines => _controller.stream; + Stream get events => _controller.stream; @override bool get isRunning => _isRunning;