From 607d1729e6329813d90220e7923da752ef9e76f5 Mon Sep 17 00:00:00 2001 From: Levi Date: Sun, 6 Sep 2026 11:59:11 +0800 Subject: [PATCH] refactor: share async invoke bridge across native platforms --- bind/desktop/invoke_executor.go | 123 ----------- bind/desktop/main.go | 44 +--- bind/mobile/main.go | 25 ++- bind/native/invoke.go | 202 ++++++++++++++++++ .../invoke_test.go} | 96 ++++++++- pkg/rest/invoke.go | 7 +- pkg/rest/invoke_test.go | 8 +- .../kotlin/com/gopeed/gopeed/MainActivity.kt | 25 ++- .../docs/backend-transport-architecture.md | 25 ++- ui/flutter/include/libgopeed.h | 1 - ui/flutter/ios/Runner/AppDelegate.swift | 8 +- .../ios/Runner/GopeedTaskEventForwarder.h | 8 + .../ios/Runner/GopeedTaskEventForwarder.m | 51 +++++ .../lib/core/common/libgopeed_channel.dart | 3 + ui/flutter/lib/core/ffi/libgopeed_bind.dart | 30 --- ui/flutter/pubspec.yaml | 1 - .../core/common/libgopeed_channel_test.dart | 29 +++ 17 files changed, 464 insertions(+), 222 deletions(-) delete mode 100644 bind/desktop/invoke_executor.go create mode 100644 bind/native/invoke.go rename bind/{desktop/invoke_executor_test.go => native/invoke_test.go} (51%) create mode 100644 ui/flutter/test/core/common/libgopeed_channel_test.dart diff --git a/bind/desktop/invoke_executor.go b/bind/desktop/invoke_executor.go deleted file mode 100644 index 7120f489..00000000 --- a/bind/desktop/invoke_executor.go +++ /dev/null @@ -1,123 +0,0 @@ -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 -} diff --git a/bind/desktop/main.go b/bind/desktop/main.go index 45cae19d..7054093b 100644 --- a/bind/desktop/main.go +++ b/bind/desktop/main.go @@ -20,29 +20,15 @@ import "C" import ( "encoding/json" "fmt" - "sync" "unsafe" + nativebridge "github.com/GopeedLab/gopeed/bind/native" "github.com/GopeedLab/gopeed/pkg/rest" "github.com/GopeedLab/gopeed/pkg/rest/model" ) 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 @@ -56,11 +42,13 @@ func Start(cfg *C.char) (int, *C.char) { if err != nil { return 0, C.CString(err.Error()) } + nativebridge.ResumeInvokes() return realPort, nil } //export Stop func Stop() { + nativebridge.PauseInvokesAndWait() rest.Stop() } @@ -96,16 +84,6 @@ func apiServerResult(state *model.APIServerState, err error) string { return string(data) } -//export Invoke -func Invoke(method *C.char, path *C.char, query *C.char, body *C.char) *C.char { - return C.CString(rest.Invoke( - C.GoString(method), - C.GoString(path), - C.GoString(query), - C.GoString(body), - )) -} - //export InvokeAsync func InvokeAsync( method *C.char, @@ -118,12 +96,6 @@ func InvokeAsync( 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 @@ -138,9 +110,13 @@ func InvokeAsync( C.CString(payload), ) } - if !getInvokeExecutor().submit(invokeTask{request: request, complete: complete}) { - complete("", fmt.Errorf("invoke executor is closed")) - } + nativebridge.InvokeAsync( + C.GoString(method), + C.GoString(path), + C.GoString(query), + C.GoString(body), + complete, + ) } //export SubscribeTaskEvents diff --git a/bind/mobile/main.go b/bind/mobile/main.go index ab45ae32..2d4b8dae 100644 --- a/bind/mobile/main.go +++ b/bind/mobile/main.go @@ -5,6 +5,7 @@ import "C" import ( "encoding/json" + nativebridge "github.com/GopeedLab/gopeed/bind/native" "github.com/GopeedLab/gopeed/internal/webview/rpcprovider" "github.com/GopeedLab/gopeed/pkg/rest" "github.com/GopeedLab/gopeed/pkg/rest/model" @@ -18,10 +19,15 @@ func Start(cfg string) (int, error) { config.ProductionMode = true config.NativeMode = true applyWebViewProvider(&config) - return rest.Start(&config) + port, err := rest.Start(&config) + if err == nil { + nativebridge.ResumeInvokes() + } + return port, err } func Stop() { + nativebridge.PauseInvokesAndWait() rest.Stop() } @@ -50,8 +56,21 @@ func apiServerResult(state *model.APIServerState, err error) string { return string(data) } -func Invoke(method, path, query, body string) string { - return rest.Invoke(method, path, query, body) +type InvokeResultListener interface { + OnResult(requestID int64, success bool, payload string) +} + +func InvokeAsync(method, path, query, body string, requestID int64, listener InvokeResultListener) { + if listener == nil { + return + } + nativebridge.InvokeAsync(method, path, query, body, func(result string, err error) { + if err != nil { + listener.OnResult(requestID, false, err.Error()) + return + } + listener.OnResult(requestID, true, result) + }) } type TaskEventListener interface { diff --git a/bind/native/invoke.go b/bind/native/invoke.go new file mode 100644 index 00000000..27056551 --- /dev/null +++ b/bind/native/invoke.go @@ -0,0 +1,202 @@ +// Package nativebridge contains transport-neutral behavior shared by the +// Desktop C FFI and gomobile bridge adapters. +package nativebridge + +import ( + "fmt" + "runtime" + "sync" + + "github.com/GopeedLab/gopeed/pkg/rest" +) + +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) +} + +// InvokeAsync queues an in-process API request and completes it through the +// supplied callback. Desktop and mobile adapters share this executor. +func InvokeAsync(method, path, query, body string, callback func(result string, err error)) { + if callback == nil { + return + } + task := invokeTask{ + request: invokeRequest{ + method: method, + path: path, + query: query, + body: body, + }, + complete: callback, + } + if !getInvokeExecutor().submit(task) { + completeCallback(callback, "", fmt.Errorf("native bridge is stopping")) + } +} + +// ResumeInvokes enables request submission after the native runtime starts. +func ResumeInvokes() { + getInvokeExecutor().resume() +} + +// PauseInvokesAndWait prevents new submissions and waits for every queued or +// running request to finish before a native runtime is stopped. +func PauseInvokesAndWait() { + getInvokeExecutor().pauseAndWait() +} + +var ( + invokeExecutorOnce sync.Once + sharedInvoker *invokeExecutor +) + +func getInvokeExecutor() *invokeExecutor { + invokeExecutorOnce.Do(func() { + sharedInvoker = newDefaultInvokeExecutor(func(request invokeRequest) string { + return rest.Dispatch(request.method, request.path, request.query, request.body) + }) + }) + return sharedInvoker +} + +// invokeExecutor keeps native API work concurrent without rejecting requests +// when every worker is busy. Additional requests wait in FIFO order and are +// completed through their callback. +type invokeExecutor struct { + mu sync.Mutex + ready *sync.Cond + queue []invokeTask + closed bool + accepting bool + pending int + 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, accepting: true} + 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 || !e.accepting { + return false + } + e.queue = append(e.queue, task) + e.pending++ + e.ready.Signal() + return true +} + +func (e *invokeExecutor) resume() { + e.mu.Lock() + e.accepting = true + e.mu.Unlock() +} + +func (e *invokeExecutor) pauseAndWait() { + e.mu.Lock() + e.accepting = false + for e.pending > 0 { + e.ready.Wait() + } + e.mu.Unlock() +} + +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) + completeCallback(task.complete, result, err) + e.mu.Lock() + e.pending-- + if e.pending == 0 { + e.ready.Broadcast() + } + e.mu.Unlock() + } +} + +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 +} + +func completeCallback(callback func(string, error), result string, err error) { + defer func() { + _ = recover() + }() + callback(result, err) +} diff --git a/bind/desktop/invoke_executor_test.go b/bind/native/invoke_test.go similarity index 51% rename from bind/desktop/invoke_executor_test.go rename to bind/native/invoke_test.go index 1bdd1d43..dc55bdb4 100644 --- a/bind/desktop/invoke_executor_test.go +++ b/bind/native/invoke_test.go @@ -1,6 +1,7 @@ -package main +package nativebridge import ( + "runtime" "strconv" "sync" "sync/atomic" @@ -100,3 +101,96 @@ func TestInvokeExecutorConvertsPanicsToErrors(t *testing.T) { } executor.close() } + +func TestInvokeExecutorSurvivesCallbackPanic(t *testing.T) { + executor := newInvokeExecutor(1, func(request invokeRequest) string { + return request.path + }) + executor.submit(invokeTask{complete: func(string, error) { + panic("callback failed") + }}) + + completed := make(chan string, 1) + executor.submit(invokeTask{ + request: invokeRequest{path: "next"}, + complete: func(result string, _ error) { + completed <- result + }, + }) + select { + case result := <-completed: + if result != "next" { + t.Fatalf("unexpected result: %q", result) + } + case <-time.After(time.Second): + t.Fatal("worker stopped after callback panic") + } + executor.close() +} + +func TestInvokeExecutorPausesAndDrains(t *testing.T) { + release := make(chan struct{}) + started := make(chan struct{}) + executor := newInvokeExecutor(1, func(request invokeRequest) string { + if request.path == "running" { + close(started) + <-release + } + return request.path + }) + completed := make(chan struct{}) + if !executor.submit(invokeTask{ + request: invokeRequest{path: "running"}, + complete: func(string, error) { + close(completed) + }, + }) { + t.Fatal("initial task was rejected") + } + <-started + + drained := make(chan struct{}) + go func() { + executor.pauseAndWait() + close(drained) + }() + deadline := time.Now().Add(time.Second) + for { + executor.mu.Lock() + paused := !executor.accepting + executor.mu.Unlock() + if paused { + break + } + if time.Now().After(deadline) { + t.Fatal("executor did not pause") + } + runtime.Gosched() + } + select { + case <-drained: + t.Fatal("pause returned before the running request completed") + default: + } + if executor.submit(invokeTask{}) { + t.Fatal("paused executor accepted a new request") + } + + close(release) + <-completed + select { + case <-drained: + case <-time.After(time.Second): + t.Fatal("pause did not return after pending requests drained") + } + + executor.resume() + resumed := make(chan struct{}) + if !executor.submit(invokeTask{complete: func(string, error) { + close(resumed) + }}) { + t.Fatal("resumed executor rejected a request") + } + <-resumed + executor.close() +} diff --git a/pkg/rest/invoke.go b/pkg/rest/invoke.go index be801896..78c25291 100644 --- a/pkg/rest/invoke.go +++ b/pkg/rest/invoke.go @@ -13,9 +13,10 @@ const ( TaskEventError = uint64(goapi.TaskEventError) ) -// 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 { +// Dispatch executes one in-process API request. Native scheduling and callback +// behavior belong to bind/native; this function only bridges into the shared +// API Service without opening an HTTP listener. +func Dispatch(method, path, rawQuery, body string) string { runtimeMu.RLock() defer runtimeMu.RUnlock() service := APIService diff --git a/pkg/rest/invoke_test.go b/pkg/rest/invoke_test.go index 1d9dfe2e..d1121802 100644 --- a/pkg/rest/invoke_test.go +++ b/pkg/rest/invoke_test.go @@ -27,7 +27,7 @@ func TestNativeInvokeWithoutRESTListener(t *testing.T) { } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { @@ -106,7 +106,7 @@ func TestAPIServerLifecycleUsesPersistedConfigWithoutRestartingCore(t *testing.T } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { @@ -153,7 +153,7 @@ func TestRestartAPIServerFailsFastAndLeavesListenerStopped(t *testing.T) { } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { @@ -229,7 +229,7 @@ func TestNativeAPIServerAutoStartFailureKeepsCoreAvailable(t *testing.T) { } var result model.Result[json.RawMessage] - if err := json.Unmarshal([]byte(Invoke(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { + if err := json.Unmarshal([]byte(Dispatch(http.MethodGet, "/api/v1/info", "", "")), &result); err != nil { t.Fatal(err) } if result.Code != model.CodeOk { diff --git a/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt b/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt index 165070da..47b0e3b9 100644 --- a/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt +++ b/ui/flutter/android/app/src/main/kotlin/com/gopeed/gopeed/MainActivity.kt @@ -1,6 +1,7 @@ package com.gopeed.gopeed import com.gopeed.libgopeed.Libgopeed +import com.gopeed.libgopeed.InvokeResultListener import com.gopeed.libgopeed.TaskEventListener import io.flutter.embedding.android.FlutterActivity import io.flutter.embedding.engine.FlutterEngine @@ -38,13 +39,23 @@ class MainActivity : FlutterActivity() { "restartApiServer" -> result.success(Libgopeed.restartAPIServer()) "invoke" -> { try { - result.success( - Libgopeed.invoke( - call.argument("method") ?: "", - call.argument("path") ?: "", - call.argument("query") ?: "", - call.argument("body") ?: "", - ), + Libgopeed.invokeAsync( + call.argument("method") ?: "", + call.argument("path") ?: "", + call.argument("query") ?: "", + call.argument("body") ?: "", + call.argument("requestID")?.toLong() ?: 0L, + object : InvokeResultListener { + override fun onResult(requestID: Long, success: Boolean, payload: String?) { + runOnUiThread { + if (success) { + result.success(payload ?: "") + } else { + result.error("ERROR", payload ?: "InvokeAsync failed", null) + } + } + } + }, ) } catch (error: Exception) { result.error("ERROR", error.message, null) diff --git a/ui/flutter/docs/backend-transport-architecture.md b/ui/flutter/docs/backend-transport-architecture.md index 1d3879ee..a00985c1 100644 --- a/ui/flutter/docs/backend-transport-architecture.md +++ b/ui/flutter/docs/backend-transport-architecture.md @@ -259,7 +259,6 @@ GetAPIServerState StartAPIServer StopAPIServer RestartAPIServer -Invoke InvokeAsync FreeCString SubscribeTaskEvents @@ -269,12 +268,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`. -- `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. -- 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. +- `InvokeAsync` is the only native invocation entry point. Desktop and gomobile adapters both submit to the shared bounded-concurrency executor in `bind/native`. +- The shared 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. +- The Desktop adapter copies every Dart-owned input string before returning and completes through a `NativeCallable.listener` callback. +- Dart-owned input memory must be released after the non-blocking `InvokeAsync` native entry returns. Go-owned callback payloads remain valid until Dart copies them and calls `FreeCString`. +- `InvokeAsync` returns the same Result JSON as REST through its callback; business errors do not use a separate ABI error model. +- The Desktop bridge isolate owns the dynamic library and native callbacks, decodes callback JSON, and forwards sendable Dart values to the UI isolate. Mobile MethodChannel handlers run on background task queues, while invoke completion is delivered through the shared Go callback flow. - 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. @@ -289,14 +288,14 @@ GetAPIServerState() StartAPIServer() StopAPIServer() RestartAPIServer() -Invoke(method, path, query, body) +InvokeAsync(method, path, query, body, requestID, listener) SubscribeTaskEvents(mask, listener) ``` Flutter bridge rules: -- Normal calls use MethodChannel. -- Android performs normal calls on a background task queue. +- Normal calls use MethodChannel, while `InvokeAsync` completes the pending MethodChannel result through a gomobile listener. +- Android and iOS run the Libgopeed MethodChannel handler on a serial background task queue; callback adapters marshal completion back to the platform UI thread. - Events reuse the same MethodChannel through the reverse `taskEvent` message. - Kotlin and Swift contain no business logic. They only forward arguments, dispatch callbacks to the UI thread, and translate errors. @@ -593,8 +592,8 @@ Flutter converts all of them into `GopeedException`. Components do not handle Di ### 11.3 Platform smoke tests -- Windows, macOS, Linux: startup, Invoke, events, shutdown, and repeated startup. -- Android, iOS: gomobile Invoke, reverse MethodChannel events, and background/foreground transitions. +- Windows, macOS, Linux: startup, InvokeAsync, events, shutdown, and repeated startup. +- Android, iOS: gomobile InvokeAsync callbacks, reverse MethodChannel events, and background/foreground transitions. - Web: REST, login, proxy, and task-file downloads. - Native: enabling and disabling REST does not interrupt download tasks. @@ -609,7 +608,7 @@ Flutter converts all of them into `GopeedException`. Components do not handle Di ### Phase 2: Native in-process calls — implemented on this branch - Integrate the FFI Transport on Desktop. -- Extend gomobile Invoke and MethodChannel on Mobile. +- Extend gomobile InvokeAsync callbacks and MethodChannel on Mobile. - Move Flutter API calls to the unified Transport so each typed method is maintained once. - Do not create a socket listener by default on native clients. diff --git a/ui/flutter/include/libgopeed.h b/ui/flutter/include/libgopeed.h index 945955be..9eb93ac8 100644 --- a/ui/flutter/include/libgopeed.h +++ b/ui/flutter/include/libgopeed.h @@ -112,7 +112,6 @@ 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 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); diff --git a/ui/flutter/ios/Runner/AppDelegate.swift b/ui/flutter/ios/Runner/AppDelegate.swift index aac051ad..e99ee6de 100644 --- a/ui/flutter/ios/Runner/AppDelegate.swift +++ b/ui/flutter/ios/Runner/AppDelegate.swift @@ -14,10 +14,13 @@ import UIKit func didInitializeImplicitFlutterEngine(_ engineBridge: FlutterImplicitEngineBridge) { GeneratedPluginRegistrant.register(with: engineBridge.pluginRegistry) let messenger = engineBridge.applicationRegistrar.messenger() + let libgopeedTaskQueue = messenger.makeBackgroundTaskQueue?() let libgopeedChannel = FlutterMethodChannel( name: "gopeed.com/libgopeed", - binaryMessenger: messenger + binaryMessenger: messenger, + codec: FlutterStandardMethodCodec.sharedInstance(), + taskQueue: libgopeedTaskQueue ) let taskEventForwarder = GopeedTaskEventForwarder(channel: libgopeedChannel) libgopeedChannel.setMethodCallHandler { call, result in @@ -56,7 +59,8 @@ import UIKit let path = arguments?["path"] as? String ?? "" let query = arguments?["query"] as? String ?? "" let body = arguments?["body"] as? String ?? "" - result(LibgopeedInvoke(method, path, query, body)) + let requestID = (arguments?["requestID"] as? NSNumber)?.int64Value ?? 0 + GopeedInvokeAsyncWithResult(method, path, query, body, requestID, result) case "subscribeTaskEvents": let arguments = call.arguments as? [String: Any] let mask = (arguments?["mask"] as? NSNumber)?.int64Value ?? 0 diff --git a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h index e12dd1fd..6f9869e6 100644 --- a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h +++ b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.h @@ -13,4 +13,12 @@ FOUNDATION_EXPORT void GopeedSubscribeTaskEventsWithForwarder( int64_t mask, GopeedTaskEventForwarder * _Nullable listener); +FOUNDATION_EXPORT void GopeedInvokeAsyncWithResult( + NSString *method, + NSString *path, + NSString *query, + NSString *body, + int64_t requestID, + FlutterResult result); + NS_ASSUME_NONNULL_END diff --git a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m index c6bcee5e..e35bd5cf 100644 --- a/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m +++ b/ui/flutter/ios/Runner/GopeedTaskEventForwarder.m @@ -8,6 +8,45 @@ @end +@interface GopeedInvokeResultForwarder : NSObject + +@property(nonatomic, copy, nullable) FlutterResult result; + +- (instancetype)initWithResult:(FlutterResult)result; + +@end + +@implementation GopeedInvokeResultForwarder + +- (instancetype)initWithResult:(FlutterResult)result { + self = [super init]; + if (self) { + _result = [result copy]; + } + return self; +} + +- (void)onResult:(int64_t)requestID + success:(BOOL)success + payload:(NSString * _Nullable)payload { + FlutterResult result = self.result; + self.result = nil; + if (result == nil) { + return; + } + dispatch_async(dispatch_get_main_queue(), ^{ + if (success) { + result(payload ?: @""); + } else { + result([FlutterError errorWithCode:@"ERROR" + message:payload ?: @"InvokeAsync failed" + details:nil]); + } + }); +} + +@end + @implementation GopeedTaskEventForwarder - (instancetype)initWithChannel:(FlutterMethodChannel *)channel { @@ -33,3 +72,15 @@ void GopeedSubscribeTaskEventsWithForwarder( GopeedTaskEventForwarder * _Nullable listener) { LibgopeedSubscribeTaskEvents(mask, listener); } + +void GopeedInvokeAsyncWithResult( + NSString *method, + NSString *path, + NSString *query, + NSString *body, + int64_t requestID, + FlutterResult result) { + GopeedInvokeResultForwarder *forwarder = + [[GopeedInvokeResultForwarder alloc] initWithResult:result]; + LibgopeedInvokeAsync(method, path, query, body, requestID, forwarder); +} diff --git a/ui/flutter/lib/core/common/libgopeed_channel.dart b/ui/flutter/lib/core/common/libgopeed_channel.dart index 7bc32778..a6e03c46 100644 --- a/ui/flutter/lib/core/common/libgopeed_channel.dart +++ b/ui/flutter/lib/core/common/libgopeed_channel.dart @@ -11,6 +11,7 @@ import 'task_event.dart'; class LibgopeedChannel implements LibgopeedInterface { static const _channel = MethodChannel('gopeed.com/libgopeed'); final _taskEvents = StreamController.broadcast(); + int _requestID = 0; LibgopeedChannel() { _channel.setMethodCallHandler((call) async { @@ -50,7 +51,9 @@ class LibgopeedChannel implements LibgopeedInterface { @override Future invoke(String method, String path, {String query = '', String body = ''}) async { + final requestID = _requestID++; final result = await _channel.invokeMethod('invoke', { + 'requestID': requestID, 'method': method, 'path': path, 'query': query, diff --git a/ui/flutter/lib/core/ffi/libgopeed_bind.dart b/ui/flutter/lib/core/ffi/libgopeed_bind.dart index 90a3aae9..93e9b2e5 100644 --- a/ui/flutter/lib/core/ffi/libgopeed_bind.dart +++ b/ui/flutter/lib/core/ffi/libgopeed_bind.dart @@ -58,36 +58,6 @@ class LibgopeedBind { late final _RestartAPIServerPtr = _lookup Function()>>('RestartAPIServer'); late final _RestartAPIServer = _RestartAPIServerPtr.asFunction Function()>(); - ffi.Pointer Invoke( - ffi.Pointer method, - ffi.Pointer path, - ffi.Pointer query, - ffi.Pointer body, - ) { - return _Invoke(method, path, query, body); - } - - late final _InvokePtr = - _lookup< - ffi.NativeFunction< - ffi.Pointer Function( - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ) - > - >('Invoke'); - late final _Invoke = - _InvokePtr.asFunction< - ffi.Pointer Function( - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ffi.Pointer, - ) - >(); - void InvokeAsync( ffi.Pointer method, ffi.Pointer path, diff --git a/ui/flutter/pubspec.yaml b/ui/flutter/pubspec.yaml index 60c93a08..056cf870 100644 --- a/ui/flutter/pubspec.yaml +++ b/ui/flutter/pubspec.yaml @@ -208,7 +208,6 @@ ffigen: - "StartAPIServer" - "StopAPIServer" - "RestartAPIServer" - - "Invoke" - "InvokeAsync" - "SubscribeTaskEvents" - "FreeCString" diff --git a/ui/flutter/test/core/common/libgopeed_channel_test.dart b/ui/flutter/test/core/common/libgopeed_channel_test.dart new file mode 100644 index 00000000..a3b44b8e --- /dev/null +++ b/ui/flutter/test/core/common/libgopeed_channel_test.dart @@ -0,0 +1,29 @@ +import 'package:flutter/services.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:gopeed/core/common/libgopeed_channel.dart'; + +void main() { + TestWidgetsFlutterBinding.ensureInitialized(); + + const channel = MethodChannel('gopeed.com/libgopeed'); + final messenger = TestDefaultBinaryMessengerBinding.instance.defaultBinaryMessenger; + + tearDown(() { + messenger.setMockMethodCallHandler(channel, null); + }); + + test('mobile invokes carry unique request IDs for async callbacks', () async { + final requestIDs = []; + messenger.setMockMethodCallHandler(channel, (call) async { + expect(call.method, 'invoke'); + final arguments = (call.arguments as Map).cast(); + requestIDs.add(arguments['requestID']! as int); + return '{"code":0}'; + }); + + final bridge = LibgopeedChannel(); + await Future.wait([bridge.invoke('GET', '/api/v1/info'), bridge.invoke('GET', '/api/v1/config')]); + + expect(requestIDs, [0, 1]); + }); +}