feat: make desktop FFI invocation asynchronous

bind/desktop/invoke_executor.go

@@ -0,0 +1,123 @@
+package main
+
+import (
+	"fmt"
+	"runtime"
+	"sync"
+)
+
+const (
+	minInvokeConcurrency = 4
+	maxInvokeConcurrency = 32
+)
+
+type invokeRequest struct {
+	method string
+	path   string
+	query  string
+	body   string
+}
+
+type invokeTask struct {
+	request  invokeRequest
+	complete func(result string, err error)
+}
+
+// invokeExecutor keeps native API work concurrent without ever rejecting a
+// request because all workers are busy. submit only appends to the waiting
+// queue; callers observe backpressure through the Future completed by the
+// callback rather than by blocking the FFI entry point.
+type invokeExecutor struct {
+	mu      sync.Mutex
+	ready   *sync.Cond
+	queue   []invokeTask
+	closed  bool
+	workers sync.WaitGroup
+	handler func(invokeRequest) string
+}
+
+func invokeConcurrency(cpuCount int) int {
+	concurrency := cpuCount * 2
+	if concurrency < minInvokeConcurrency {
+		return minInvokeConcurrency
+	}
+	if concurrency > maxInvokeConcurrency {
+		return maxInvokeConcurrency
+	}
+	return concurrency
+}
+
+func newInvokeExecutor(workerCount int, handler func(invokeRequest) string) *invokeExecutor {
+	if workerCount < 1 {
+		workerCount = 1
+	}
+	executor := &invokeExecutor{handler: handler}
+	executor.ready = sync.NewCond(&executor.mu)
+	executor.workers.Add(workerCount)
+	for range workerCount {
+		go executor.run()
+	}
+	return executor
+}
+
+func newDefaultInvokeExecutor(handler func(invokeRequest) string) *invokeExecutor {
+	return newInvokeExecutor(invokeConcurrency(runtime.GOMAXPROCS(0)), handler)
+}
+
+func (e *invokeExecutor) submit(task invokeTask) bool {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.closed {
+		return false
+	}
+	e.queue = append(e.queue, task)
+	e.ready.Signal()
+	return true
+}
+
+func (e *invokeExecutor) close() {
+	e.mu.Lock()
+	e.closed = true
+	e.ready.Broadcast()
+	e.mu.Unlock()
+	e.workers.Wait()
+}
+
+func (e *invokeExecutor) run() {
+	defer e.workers.Done()
+	for {
+		task, ok := e.take()
+		if !ok {
+			return
+		}
+		result, err := e.execute(task.request)
+		task.complete(result, err)
+	}
+}
+
+func (e *invokeExecutor) take() (invokeTask, bool) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	for len(e.queue) == 0 && !e.closed {
+		e.ready.Wait()
+	}
+	if len(e.queue) == 0 {
+		return invokeTask{}, false
+	}
+	task := e.queue[0]
+	e.queue[0] = invokeTask{}
+	e.queue = e.queue[1:]
+	if len(e.queue) == 0 {
+		e.queue = nil
+	}
+	return task, true
+}
+
+func (e *invokeExecutor) execute(request invokeRequest) (result string, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			err = fmt.Errorf("invoke panic: %v", recovered)
+		}
+	}()
+	return e.handler(request), nil
+}

bind/desktop/invoke_executor_test.go

@@ -0,0 +1,102 @@
+package main
+
+import (
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestInvokeConcurrency(t *testing.T) {
+	tests := []struct {
+		cpus int
+		want int
+	}{
+		{cpus: 1, want: 4},
+		{cpus: 2, want: 4},
+		{cpus: 4, want: 8},
+		{cpus: 8, want: 16},
+		{cpus: 16, want: 32},
+		{cpus: 64, want: 32},
+	}
+	for _, test := range tests {
+		t.Run(strconv.Itoa(test.cpus), func(t *testing.T) {
+			if got := invokeConcurrency(test.cpus); got != test.want {
+				t.Fatalf("invokeConcurrency(%d) = %d, want %d", test.cpus, got, test.want)
+			}
+		})
+	}
+}
+
+func TestInvokeExecutorWaitsWithoutRejecting(t *testing.T) {
+	const (
+		workerCount = 2
+		taskCount   = 6
+	)
+	release := make(chan struct{})
+	started := make(chan struct{}, taskCount)
+	var active atomic.Int32
+	var maxActive atomic.Int32
+	executor := newInvokeExecutor(workerCount, func(request invokeRequest) string {
+		current := active.Add(1)
+		for {
+			maximum := maxActive.Load()
+			if current <= maximum || maxActive.CompareAndSwap(maximum, current) {
+				break
+			}
+		}
+		started <- struct{}{}
+		<-release
+		active.Add(-1)
+		return request.path
+	})
+
+	var completed sync.WaitGroup
+	completed.Add(taskCount)
+	for i := range taskCount {
+		accepted := executor.submit(invokeTask{
+			request: invokeRequest{path: strconv.Itoa(i)},
+			complete: func(string, error) {
+				completed.Done()
+			},
+		})
+		if !accepted {
+			t.Fatalf("task %d was rejected while all workers were busy", i)
+		}
+	}
+
+	for range workerCount {
+		select {
+		case <-started:
+		case <-time.After(time.Second):
+			t.Fatal("workers did not start queued tasks")
+		}
+	}
+	if got := maxActive.Load(); got != workerCount {
+		t.Fatalf("maximum active tasks = %d, want %d", got, workerCount)
+	}
+
+	close(release)
+	completed.Wait()
+	executor.close()
+}
+
+func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) {
+	executor := newInvokeExecutor(1, func(invokeRequest) string {
+		panic("boom")
+	})
+	completed := make(chan error, 1)
+	executor.submit(invokeTask{complete: func(_ string, err error) {
+		completed <- err
+	}})
+	select {
+	case err := <-completed:
+		if err == nil || err.Error() != "invoke panic: boom" {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("panic result was not completed")
+	}
+	executor.close()
+}

bind/desktop/main.go

@@ -6,14 +6,21 @@ package main
 
 typedef void (*TaskEventCallback)(char* payload);
 
+typedef void (*InvokeResultCallback)(uint64_t request_id, int success, char* payload);
+
 static void callTaskEventCallback(uintptr_t callback, char* payload) {
 	((TaskEventCallback)callback)(payload);
 }
+
+static void callInvokeResultCallback(uintptr_t callback, uint64_t request_id, int success, char* payload) {
+	((InvokeResultCallback)callback)(request_id, success, payload);
+}
 */
 import "C"
 import (
 	"encoding/json"
 	"fmt"
+	"sync"
 	"unsafe"
 
 	"github.com/GopeedLab/gopeed/pkg/rest"
@@ -22,6 +29,20 @@ import (
 
 func main() {}
 
+var (
+	invokeExecutorOnce sync.Once
+	desktopInvoker     *invokeExecutor
+)
+
+func getInvokeExecutor() *invokeExecutor {
+	invokeExecutorOnce.Do(func() {
+		desktopInvoker = newDefaultInvokeExecutor(func(request invokeRequest) string {
+			return rest.Invoke(request.method, request.path, request.query, request.body)
+		})
+	})
+	return desktopInvoker
+}
+
 //export Start
 func Start(cfg *C.char) (int, *C.char) {
 	var config model.StartConfig
@@ -85,6 +106,43 @@ func Invoke(method *C.char, path *C.char, query *C.char, body *C.char) *C.char {
 	))
 }
 
+//export InvokeAsync
+func InvokeAsync(
+	method *C.char,
+	path *C.char,
+	query *C.char,
+	body *C.char,
+	requestID C.ulonglong,
+	callback C.uintptr_t,
+) {
+	if callback == 0 {
+		return
+	}
+	request := invokeRequest{
+		method: C.GoString(method),
+		path:   C.GoString(path),
+		query:  C.GoString(query),
+		body:   C.GoString(body),
+	}
+	complete := func(result string, err error) {
+		success := C.int(1)
+		payload := result
+		if err != nil {
+			success = 0
+			payload = err.Error()
+		}
+		C.callInvokeResultCallback(
+			callback,
+			C.uint64_t(requestID),
+			success,
+			C.CString(payload),
+		)
+	}
+	if !getInvokeExecutor().submit(invokeTask{request: request, complete: complete}) {
+		complete("", fmt.Errorf("invoke executor is closed"))
+	}
+}
+
 //export SubscribeTaskEvents
 func SubscribeTaskEvents(mask C.ulonglong, callback C.uintptr_t) {
 	if mask == 0 || callback == 0 {

pkg/rest/invoke.go

@@ -16,6 +16,8 @@ const (
 // Invoke dispatches a Core API request without using an HTTP listener. It is
 // the shared entry point for Desktop FFI and gomobile bindings.
 func Invoke(method, path, rawQuery, body string) string {
+	runtimeMu.RLock()
+	defer runtimeMu.RUnlock()
 	service := APIService
 	if service == nil {
 		return marshalInvokeResult(model.NewErrorResult("service not started"))
@@ -29,6 +31,8 @@ func Invoke(method, path, rawQuery, body string) string {
 }
 
 func SubscribeTaskEvents(mask uint64, listener func(payload string)) {
+	runtimeMu.RLock()
+	defer runtimeMu.RUnlock()
 	if APIService == nil {
 		return
 	}

pkg/rest/server.go

@@ -29,7 +29,7 @@ import (
 )
 
 var (
-	runtimeMu sync.Mutex
+	runtimeMu sync.RWMutex
 
 	Downloader *download.Downloader
 	APIService *goapi.Service

ui/flutter/docs/backend-transport-architecture.md

@@ -260,6 +260,7 @@ StartAPIServer
 StopAPIServer
 RestartAPIServer
 Invoke
+InvokeAsync
 FreeCString
 SubscribeTaskEvents
 ```
@@ -268,9 +269,12 @@ Requirements:
 
 - Use the cgo C-string ABI in the first phase. Reconsider pointer-and-length parameters only during a future ABI-breaking revision.
 - Memory allocated by Go must be released with `FreeCString`.
-- Dart-owned input memory must be released after the synchronous native call returns.
+- `InvokeAsync` copies every Dart-owned input string before returning, queues the request in a bounded-concurrency Go executor, and completes it through a `NativeCallable.listener` callback.
+- The async executor uses `clamp(GOMAXPROCS * 2, 4, 32)` workers. Requests wait in FIFO order when every worker is busy; pool saturation is backpressure, not an API error.
+- Dart-owned input memory must be released after the synchronous native entry call returns. Go-owned callback payloads remain valid until Dart copies them and calls `FreeCString`.
 - `Invoke` returns the same Result JSON as REST; business errors do not use a separate ABI error model.
-- Run blocking Desktop FFI work outside the Flutter UI isolate. Mobile MethodChannel handlers likewise run blocking Go work on a background task queue.
+- Keep `Invoke` as a compatibility entry point, but route Desktop application requests through `InvokeAsync` in the FFI bridge isolate.
+- Run blocking Desktop FFI work outside the Flutter UI isolate. The bridge isolate owns the dynamic library and native callbacks, decodes callback JSON, and forwards sendable Dart values to the UI isolate. Mobile MethodChannel handlers likewise run blocking Go work on a background task queue.
 - Dart bindings are generated from `include/libgopeed.h` with `ffigen`. Every exported Desktop symbol must be added to the `ffigen.functions.include` allowlist.
 - Never hand-edit `lib/core/ffi/libgopeed_bind.dart`; regenerate it with `flutter pub run ffigen` after changing the C header.
 

ui/flutter/include/libgopeed.h

@@ -12,6 +12,8 @@
 
 #ifndef GO_CGO_GOSTRING_TYPEDEF
 typedef struct { const char *p; ptrdiff_t n; } _GoString_;
+extern size_t _GoStringLen(_GoString_ s);
+extern const char *_GoStringPtr(_GoString_ s);
 #endif
 
 #endif
@@ -19,6 +21,24 @@ typedef struct { const char *p; ptrdiff_t n; } _GoString_;
 /* Start of preamble from import "C" comments.  */
 
 
+#line 3 "main.go"
+
+#include <stdlib.h>
+#include <stdint.h>
+
+typedef void (*TaskEventCallback)(char* payload);
+
+typedef void (*InvokeResultCallback)(uint64_t request_id, int success, char* payload);
+
+static void callTaskEventCallback(uintptr_t callback, char* payload) {
+	((TaskEventCallback)callback)(payload);
+}
+
+static void callInvokeResultCallback(uintptr_t callback, uint64_t request_id, int success, char* payload) {
+	((InvokeResultCallback)callback)(request_id, success, payload);
+}
+
+#line 1 "cgo-generated-wrapper"
 
 
 /* End of preamble from import "C" comments.  */
@@ -44,10 +64,16 @@ typedef size_t GoUintptr;
 typedef float GoFloat32;
 typedef double GoFloat64;
 #ifdef _MSC_VER
+#if !defined(__cplusplus) || _MSVC_LANG <= 201402L
 #include <complex.h>
 typedef _Fcomplex GoComplex64;
 typedef _Dcomplex GoComplex128;
 #else
+#include <complex>
+typedef std::complex<float> GoComplex64;
+typedef std::complex<double> GoComplex128;
+#endif
+#else
 typedef float _Complex GoComplex64;
 typedef double _Complex GoComplex128;
 #endif
@@ -81,13 +107,14 @@ struct Start_return {
 	char* r1;
 };
 extern struct Start_return Start(char* cfg);
-extern void Stop();
-extern char* GetAPIServerState();
-extern char* StartAPIServer();
-extern char* StopAPIServer();
-extern char* RestartAPIServer();
+extern void Stop(void);
+extern char* GetAPIServerState(void);
+extern char* StartAPIServer(void);
+extern char* StopAPIServer(void);
+extern char* RestartAPIServer(void);
 extern char* Invoke(char* method, char* path, char* query, char* body);
-extern void SubscribeTaskEvents(GoUint64 mask, GoUintptr callback);
+extern void InvokeAsync(char* method, char* path, char* query, char* body, long long unsigned int requestID, uintptr_t callback);
+extern void SubscribeTaskEvents(long long unsigned int mask, uintptr_t callback);
 extern void FreeCString(char* value);
 
 #ifdef __cplusplus

ui/flutter/lib/core/common/libgopeed_ffi.dart

@@ -1,10 +1,6 @@
 import 'dart:async';
 import 'dart:convert';
-import 'dart:ffi';
 
-import 'package:ffi/ffi.dart';
-
-import '../ffi/libgopeed_bind.dart';
 import '../ffi/libgopeed_worker.dart';
 import 'api_server_state.dart';
 import 'libgopeed_interface.dart';
@@ -12,51 +8,22 @@ import 'start_config.dart';
 import 'task_event.dart';
 
 class LibgopeedFFi implements LibgopeedInterface {
-  late LibgopeedBind _libgopeed;
+  LibgopeedFFi() {
+    _worker.taskEvents.listen(_onTaskEvent, onError: _taskEvents.addError);
+  }
+
   final _taskEvents = StreamController<TaskEvent>.broadcast();
   final _worker = LibgopeedWorker();
-  late final NativeCallable<Void Function(Pointer<Char>)> _taskEventCallback;
 
-  LibgopeedFFi(LibgopeedBind libgopeed) {
-    _libgopeed = libgopeed;
-    _taskEventCallback = NativeCallable<Void Function(Pointer<Char>)>.listener(_onTaskEvent);
-  }
-
-  void _onTaskEvent(Pointer<Char> payload) {
-    try {
-      final decoded = jsonDecode(payload.cast<Utf8>().toDartString()) as Map<String, dynamic>;
-      _taskEvents.add(TaskEvent.fromJson(decoded));
-    } finally {
-      _libgopeed.FreeCString(payload);
-    }
+  void _onTaskEvent(Map<String, dynamic> payload) {
+    _taskEvents.add(TaskEvent.fromJson(payload));
   }
 
   @override
-  Future<int> start(StartConfig cfg) {
-    var completer = Completer<int>();
-    final cfgPtr = jsonEncode(cfg).toNativeUtf8();
-    try {
-      final result = _libgopeed.Start(cfgPtr.cast());
-      if (result.r1 != nullptr) {
-        final message = result.r1.cast<Utf8>().toDartString();
-        _libgopeed.FreeCString(result.r1);
-        completer.completeError(Exception(message));
-      } else {
-        completer.complete(result.r0);
-      }
-    } finally {
-      malloc.free(cfgPtr);
-    }
-    return completer.future;
-  }
+  Future<int> start(StartConfig cfg) => _worker.start(jsonEncode(cfg));
 
   @override
-  Future<void> stop() {
-    var completer = Completer<void>();
-    _libgopeed.Stop();
-    completer.complete();
-    return completer.future;
-  }
+  Future<void> stop() => _worker.stop();
 
   Future<ApiServerOperationResult> _apiServerOperation(String operation) async {
     final payload = await _worker.apiServer(operation);
@@ -77,33 +44,12 @@ class LibgopeedFFi implements LibgopeedInterface {
 
   @override
   Future<String> invoke(String method, String path, {String query = '', String body = ''}) async {
-    final methodPtr = method.toNativeUtf8();
-    final pathPtr = path.toNativeUtf8();
-    final queryPtr = query.toNativeUtf8();
-    final bodyPtr = body.toNativeUtf8();
-    Pointer<Char>? resultPtr;
-    try {
-      resultPtr = _libgopeed.Invoke(methodPtr.cast(), pathPtr.cast(), queryPtr.cast(), bodyPtr.cast());
-      if (resultPtr == nullptr) {
-        throw StateError('Gopeed Invoke returned a null response');
-      }
-      return resultPtr.cast<Utf8>().toDartString();
-    } finally {
-      malloc.free(methodPtr);
-      malloc.free(pathPtr);
-      malloc.free(queryPtr);
-      malloc.free(bodyPtr);
-      if (resultPtr != null && resultPtr != nullptr) {
-        _libgopeed.FreeCString(resultPtr);
-      }
-    }
+    return jsonEncode(await _worker.invoke(method, path, query, body));
   }
 
   @override
   Stream<TaskEvent> get taskEvents => _taskEvents.stream;
 
   @override
-  Future<void> subscribeTaskEvents(Set<TaskEventType> events) async {
-    _libgopeed.SubscribeTaskEvents(events.mask, events.isEmpty ? 0 : _taskEventCallback.nativeFunction.address);
-  }
+  Future<void> subscribeTaskEvents(Set<TaskEventType> events) => _worker.subscribeTaskEvents(events.mask);
 }

ui/flutter/lib/core/entry/libgopeed_boot_native.dart

@@ -1,6 +1,4 @@
 import 'dart:async';
-import 'dart:ffi';
-import 'dart:io';
 
 import '../../util/util.dart';
 import '../common/libgopeed_channel.dart';
@@ -9,7 +7,6 @@ import '../common/libgopeed_interface.dart';
 import '../common/api_server_state.dart';
 import '../common/start_config.dart';
 import '../common/task_event.dart';
-import '../ffi/libgopeed_bind.dart';
 import '../libgopeed_boot.dart';
 
 LibgopeedBoot create() => LibgopeedBootNative();
@@ -19,17 +16,7 @@ class LibgopeedBootNative implements LibgopeedBoot {
 
   LibgopeedBootNative() {
     if (Util.isDesktop()) {
-      var libName = "libgopeed.";
-      if (Platform.isWindows) {
-        libName += "dll";
-      }
-      if (Platform.isMacOS) {
-        libName += "dylib";
-      }
-      if (Platform.isLinux) {
-        libName += "so";
-      }
-      _libgopeed = LibgopeedFFi(LibgopeedBind(DynamicLibrary.open(libName)));
+      _libgopeed = LibgopeedFFi();
     } else {
       _libgopeed = LibgopeedChannel();
     }

ui/flutter/lib/core/ffi/libgopeed_bind.dart

@@ -30,22 +30,30 @@ class LibgopeedBind {
   late final _StopPtr = _lookup<ffi.NativeFunction<ffi.Void Function()>>('Stop');
   late final _Stop = _StopPtr.asFunction<void Function()>();
 
-  ffi.Pointer<ffi.Char> GetAPIServerState() => _GetAPIServerState();
+  ffi.Pointer<ffi.Char> GetAPIServerState() {
+    return _GetAPIServerState();
+  }
 
   late final _GetAPIServerStatePtr = _lookup<ffi.NativeFunction<ffi.Pointer<ffi.Char> Function()>>('GetAPIServerState');
   late final _GetAPIServerState = _GetAPIServerStatePtr.asFunction<ffi.Pointer<ffi.Char> Function()>();
 
-  ffi.Pointer<ffi.Char> StartAPIServer() => _StartAPIServer();
+  ffi.Pointer<ffi.Char> StartAPIServer() {
+    return _StartAPIServer();
+  }
 
   late final _StartAPIServerPtr = _lookup<ffi.NativeFunction<ffi.Pointer<ffi.Char> Function()>>('StartAPIServer');
   late final _StartAPIServer = _StartAPIServerPtr.asFunction<ffi.Pointer<ffi.Char> Function()>();
 
-  ffi.Pointer<ffi.Char> StopAPIServer() => _StopAPIServer();
+  ffi.Pointer<ffi.Char> StopAPIServer() {
+    return _StopAPIServer();
+  }
 
   late final _StopAPIServerPtr = _lookup<ffi.NativeFunction<ffi.Pointer<ffi.Char> Function()>>('StopAPIServer');
   late final _StopAPIServer = _StopAPIServerPtr.asFunction<ffi.Pointer<ffi.Char> Function()>();
 
-  ffi.Pointer<ffi.Char> RestartAPIServer() => _RestartAPIServer();
+  ffi.Pointer<ffi.Char> RestartAPIServer() {
+    return _RestartAPIServer();
+  }
 
   late final _RestartAPIServerPtr = _lookup<ffi.NativeFunction<ffi.Pointer<ffi.Char> Function()>>('RestartAPIServer');
   late final _RestartAPIServer = _RestartAPIServerPtr.asFunction<ffi.Pointer<ffi.Char> Function()>();
@@ -80,13 +88,48 @@ class LibgopeedBind {
         )
       >();
 
+  void InvokeAsync(
+    ffi.Pointer<ffi.Char> method,
+    ffi.Pointer<ffi.Char> path,
+    ffi.Pointer<ffi.Char> query,
+    ffi.Pointer<ffi.Char> body,
+    int requestID,
+    int callback,
+  ) {
+    return _InvokeAsync(method, path, query, body, requestID, callback);
+  }
+
+  late final _InvokeAsyncPtr =
+      _lookup<
+        ffi.NativeFunction<
+          ffi.Void Function(
+            ffi.Pointer<ffi.Char>,
+            ffi.Pointer<ffi.Char>,
+            ffi.Pointer<ffi.Char>,
+            ffi.Pointer<ffi.Char>,
+            ffi.UnsignedLongLong,
+            ffi.UintPtr,
+          )
+        >
+      >('InvokeAsync');
+  late final _InvokeAsync =
+      _InvokeAsyncPtr.asFunction<
+        void Function(
+          ffi.Pointer<ffi.Char>,
+          ffi.Pointer<ffi.Char>,
+          ffi.Pointer<ffi.Char>,
+          ffi.Pointer<ffi.Char>,
+          int,
+          int,
+        )
+      >();
+
   void SubscribeTaskEvents(int mask, int callback) {
     return _SubscribeTaskEvents(mask, callback);
   }
 
-  late final _SubscribeTaskEventsPtr = _lookup<ffi.NativeFunction<ffi.Void Function(ffi.Uint64, ffi.UintPtr)>>(
-    'SubscribeTaskEvents',
-  );
+  late final _SubscribeTaskEventsPtr =
+      _lookup<ffi.NativeFunction<ffi.Void Function(ffi.UnsignedLongLong, ffi.UintPtr)>>('SubscribeTaskEvents');
   late final _SubscribeTaskEvents = _SubscribeTaskEventsPtr.asFunction<void Function(int, int)>();
 
   void FreeCString(ffi.Pointer<ffi.Char> value) {

ui/flutter/lib/core/ffi/libgopeed_worker.dart

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 import 'dart:ffi';
 import 'dart:io';
 import 'dart:isolate';
@@ -7,38 +8,79 @@ import 'package:ffi/ffi.dart';
 
 import 'libgopeed_bind.dart';
 
+typedef _InvokeResultNative = Void Function(Uint64 requestID, Int32 success, Pointer<Char> payload);
+typedef _TaskEventNative = Void Function(Pointer<Char> payload);
+
 class LibgopeedWorker {
-  LibgopeedWorker() {
+  factory LibgopeedWorker() => _instance;
+
+  LibgopeedWorker._() {
     _receivePort.listen(_handleWorkerMessage);
     _errorPort.listen(_handleWorkerError);
     _isolateReady = Isolate.spawn(_workerMain, _receivePort.sendPort, onError: _errorPort.sendPort);
   }
 
+  static final LibgopeedWorker _instance = LibgopeedWorker._();
+
   final ReceivePort _receivePort = ReceivePort();
   final ReceivePort _errorPort = ReceivePort();
-  final Map<int, Completer<String>> _pending = {};
+  final Map<int, Completer<Object?>> _pending = {};
+  final StreamController<Map<String, dynamic>> _taskEvents = StreamController<Map<String, dynamic>>.broadcast();
+  final Set<int> _invokeRequests = {};
   late final Future<Isolate> _isolateReady;
   final Completer<SendPort> _sendPort = Completer<SendPort>();
+  Completer<void>? _invokeDrain;
+  Future<void>? _stopFuture;
   int _requestID = 0;
+  bool _stopping = false;
+  bool _stopped = false;
 
-  Future<String> invoke(String method, String path, String query, String body) async {
-    await _isolateReady;
-    final sendPort = await _sendPort.future;
-    final requestID = _requestID++;
-    final completer = Completer<String>();
-    _pending[requestID] = completer;
-    sendPort.send(['invoke', requestID, method, path, query, body]);
-    return completer.future;
+  Stream<Map<String, dynamic>> get taskEvents => _taskEvents.stream;
+
+  Future<int> start(String config) => _request<int>('start', [config]);
+
+  Future<void> stop() => _stopFuture ??= _stop();
+
+  Future<void> _stop() async {
+    _stopping = true;
+    if (_invokeRequests.isNotEmpty) {
+      _invokeDrain ??= Completer<void>();
+      await _invokeDrain!.future;
+    }
+    await _request<Object?>('stop', const [], allowWhileStopping: true);
+    _stopped = true;
+    await _taskEvents.close();
+    _receivePort.close();
+    _errorPort.close();
   }
 
-  Future<String> apiServer(String operation) async {
+  Future<Object?> invoke(String method, String path, String query, String body) {
+    return _request<Object?>('invoke', [method, path, query, body], trackInvoke: true);
+  }
+
+  Future<String> apiServer(String operation) => _request<String>('apiServer', [operation]);
+
+  Future<void> subscribeTaskEvents(int mask) async {
+    await _request<Object?>('subscribeTaskEvents', [mask]);
+  }
+
+  Future<T> _request<T>(
+    String type,
+    List<Object?> arguments, {
+    bool trackInvoke = false,
+    bool allowWhileStopping = false,
+  }) async {
+    if (_stopped || (_stopping && !allowWhileStopping)) {
+      throw StateError('libgopeed worker is stopping');
+    }
     await _isolateReady;
     final sendPort = await _sendPort.future;
     final requestID = _requestID++;
-    final completer = Completer<String>();
+    final completer = Completer<Object?>();
     _pending[requestID] = completer;
-    sendPort.send(['apiServer', requestID, operation]);
-    return completer.future;
+    if (trackInvoke) _invokeRequests.add(requestID);
+    sendPort.send([type, requestID, ...arguments]);
+    return (await completer.future) as T;
   }
 
   void _handleWorkerMessage(dynamic message) {
@@ -47,13 +89,22 @@ class LibgopeedWorker {
       return;
     }
     final response = message as List<dynamic>;
-    final requestID = response[0] as int;
-    final completer = _pending.remove(requestID);
-    if (completer == null) return;
-    if (response[1] as bool) {
-      completer.complete(response[2] as String);
-    } else {
-      completer.completeError(StateError(response[2] as String));
+    switch (response[0] as String) {
+      case 'response':
+        final requestID = response[1] as int;
+        final completer = _pending.remove(requestID);
+        if (completer == null) return;
+        if (_invokeRequests.remove(requestID) && _invokeRequests.isEmpty) {
+          _invokeDrain?.complete();
+          _invokeDrain = null;
+        }
+        if (response[2] as bool) {
+          completer.complete(response[3]);
+        } else {
+          completer.completeError(StateError(response[3] as String));
+        }
+      case 'taskEvent':
+        _taskEvents.add((response[1] as Map<Object?, Object?>).cast<String, dynamic>());
     }
   }
 
@@ -65,35 +116,120 @@ class LibgopeedWorker {
       completer.completeError(error);
     }
     _pending.clear();
+    _invokeRequests.clear();
+    _invokeDrain?.completeError(error);
+    _invokeDrain = null;
+    _taskEvents.addError(error);
   }
 
   static void _workerMain(SendPort mainSendPort) {
     final requests = ReceivePort();
     final bindings = LibgopeedBind(DynamicLibrary.open(_libraryName()));
+    late final NativeCallable<_InvokeResultNative> invokeResultCallback;
+    late final NativeCallable<_TaskEventNative> taskEventCallback;
+
+    invokeResultCallback = NativeCallable<_InvokeResultNative>.listener((
+      int requestID,
+      int success,
+      Pointer<Char> payload,
+    ) {
+      if (payload == nullptr) {
+        mainSendPort.send(['response', requestID, false, 'Gopeed InvokeAsync returned a null response']);
+        return;
+      }
+      try {
+        final result = payload.cast<Utf8>().toDartString();
+        mainSendPort.send(['response', requestID, success != 0, success != 0 ? jsonDecode(result) : result]);
+      } catch (error) {
+        mainSendPort.send(['response', requestID, false, error.toString()]);
+      } finally {
+        bindings.FreeCString(payload);
+      }
+    });
+
+    taskEventCallback = NativeCallable<_TaskEventNative>.listener((Pointer<Char> payload) {
+      if (payload == nullptr) return;
+      try {
+        final event = jsonDecode(payload.cast<Utf8>().toDartString()) as Map<String, dynamic>;
+        mainSendPort.send(['taskEvent', event]);
+      } finally {
+        bindings.FreeCString(payload);
+      }
+    });
+
     mainSendPort.send(requests.sendPort);
     requests.listen((dynamic message) {
       final request = message as List<dynamic>;
       final type = request[0] as String;
       final requestID = request[1] as int;
-      try {
-        final response = switch (type) {
-          'invoke' => _invoke(
+      if (type == 'invoke') {
+        try {
+          _invokeAsync(
             bindings,
+            invokeResultCallback.nativeFunction.address,
+            requestID,
             request[2] as String,
             request[3] as String,
             request[4] as String,
             request[5] as String,
-          ),
+          );
+        } catch (error) {
+          mainSendPort.send(['response', requestID, false, error.toString()]);
+        }
+        return;
+      }
+
+      try {
+        final response = switch (type) {
+          'start' => _start(bindings, request[2] as String),
+          'stop' => _stopNative(bindings),
           'apiServer' => _apiServer(bindings, request[2] as String),
+          'subscribeTaskEvents' => _subscribeTaskEvents(
+            bindings,
+            taskEventCallback.nativeFunction.address,
+            request[2] as int,
+          ),
           _ => throw StateError('Unknown libgopeed worker request: $type'),
         };
-        mainSendPort.send([requestID, true, response]);
+        mainSendPort.send(['response', requestID, true, response]);
+        if (type == 'stop') {
+          invokeResultCallback.close();
+          taskEventCallback.close();
+          requests.close();
+        }
       } catch (error) {
-        mainSendPort.send([requestID, false, error.toString()]);
+        mainSendPort.send(['response', requestID, false, error.toString()]);
       }
     });
   }
 
+  static int _start(LibgopeedBind bindings, String config) {
+    final configPtr = config.toNativeUtf8();
+    try {
+      final result = bindings.Start(configPtr.cast());
+      if (result.r1 != nullptr) {
+        try {
+          throw StateError(result.r1.cast<Utf8>().toDartString());
+        } finally {
+          bindings.FreeCString(result.r1);
+        }
+      }
+      return result.r0;
+    } finally {
+      malloc.free(configPtr);
+    }
+  }
+
+  static Object? _stopNative(LibgopeedBind bindings) {
+    bindings.Stop();
+    return null;
+  }
+
+  static Object? _subscribeTaskEvents(LibgopeedBind bindings, int callback, int mask) {
+    bindings.SubscribeTaskEvents(mask, mask == 0 ? 0 : callback);
+    return null;
+  }
+
   static String _apiServer(LibgopeedBind bindings, String operation) {
     final resultPtr = switch (operation) {
       'get' => bindings.GetAPIServerState(),
@@ -112,26 +248,26 @@ class LibgopeedWorker {
     }
   }
 
-  static String _invoke(LibgopeedBind bindings, String method, String path, String query, String body) {
+  static void _invokeAsync(
+    LibgopeedBind bindings,
+    int callback,
+    int requestID,
+    String method,
+    String path,
+    String query,
+    String body,
+  ) {
     final methodPtr = method.toNativeUtf8();
     final pathPtr = path.toNativeUtf8();
     final queryPtr = query.toNativeUtf8();
     final bodyPtr = body.toNativeUtf8();
-    Pointer<Char>? resultPtr;
     try {
-      resultPtr = bindings.Invoke(methodPtr.cast(), pathPtr.cast(), queryPtr.cast(), bodyPtr.cast());
-      if (resultPtr == nullptr) {
-        throw StateError('Gopeed Invoke returned a null response');
-      }
-      return resultPtr.cast<Utf8>().toDartString();
+      bindings.InvokeAsync(methodPtr.cast(), pathPtr.cast(), queryPtr.cast(), bodyPtr.cast(), requestID, callback);
     } finally {
       malloc.free(methodPtr);
       malloc.free(pathPtr);
       malloc.free(queryPtr);
       malloc.free(bodyPtr);
-      if (resultPtr != null && resultPtr != nullptr) {
-        bindings.FreeCString(resultPtr);
-      }
     }
   }
 

ui/flutter/lib/core/network/gopeed/entry/gopeed_transport_native.dart

@@ -2,9 +2,9 @@ import 'dart:convert';
 
 import 'package:dio/dio.dart';
 
-import '../../../libgopeed_boot.dart';
-import '../../../ffi/libgopeed_worker.dart';
 import '../../../../util/util.dart';
+import '../../../ffi/libgopeed_worker.dart';
+import '../../../libgopeed_boot.dart';
 import '../gopeed_transport.dart';
 
 GopeedTransport createTransport(GopeedTransportConfig config) => NativeGopeedTransport();
@@ -32,14 +32,16 @@ class NativeGopeedTransport implements GopeedTransport {
   }) async {
     final request = _RequestParts.from(path, queryParameters);
     final requestBody = data == null ? '' : jsonEncode(data);
-    final payload = _ffiWorker != null
-        ? await _ffiWorker.invoke(method.toUpperCase(), request.path, request.query, requestBody)
-        : await LibgopeedBoot.instance.invoke(
-            method.toUpperCase(),
-            request.path,
-            query: request.query,
-            body: requestBody,
-          );
+    final worker = _ffiWorker;
+    if (worker != null) {
+      return worker.invoke(method.toUpperCase(), request.path, request.query, requestBody);
+    }
+    final payload = await LibgopeedBoot.instance.invoke(
+      method.toUpperCase(),
+      request.path,
+      query: request.query,
+      body: requestBody,
+    );
     return jsonDecode(payload);
   }
 

ui/flutter/pubspec.yaml

@@ -209,6 +209,7 @@ ffigen:
       - "StopAPIServer"
       - "RestartAPIServer"
       - "Invoke"
+      - "InvokeAsync"
       - "SubscribeTaskEvents"
       - "FreeCString"
   structs: