Merge latest PR #1527 into 2.0.0-beta.2

This commit is contained in:
Levi committed 2026-09-18 09:43:51 +08:00
commit dae13dec01
30 files changed
+1444 -194

No files matched your search

+44
View File
@@ -60,3 +60,47 @@ rejecting upstream tunnel; no test-only certificate bypass is added to the app.
Go proxy unit tests separately cover successful TLS forwarding and authentication.
These cases run in the existing `test.yml` WebView jobs; existing mobile build jobs
compile the Android/iOS bridge. No separate Flutter test application is required.
## Extension events and host version
`gopeed.info.hostVersion` is the host's build version (`dev` for development
builds). `gopeed.info.version` continues to identify the extension's own version.
See [extension-api.d.ts](extension-api.d.ts) for the event payloads.
```js
const page = await gopeed.runtime.webview.open();
const unsubscribe = await page.on("url-changed", ({ url, sameDocument }) => {
gopeed.logger.info(url, sameDocument);
});
await page.on("closed", ({ reason }) => gopeed.logger.info(reason));
await page.addInitScript("globalThis.extensionReady = true");
await page.goto("https://example.com");
unsubscribe();
await page.close();
```
Registration resolves only after the listener is installed; it does not replay
current state. `url-changed` reports actual address changes, including hash,
pushState, replaceState and history navigation. `load` follows the main document's
`window.load`, including reloads, and does not wait for business-level async work.
Iframe loads and subresource failures do not become main-page notifications.
`closed` distinguishes the user's close action from API closure and is terminal.
Unsubscribe is idempotent. Notifications cannot block navigation; the engine does
not await promises returned by handlers. Catch async errors inside the handler.
`addInitScript` applies to future document contexts only. Use `execute` to run
code on the current page. Executable functions are serialized and cannot capture
extension-side lexical variables; pass values explicitly as arguments.
The RPC host uses an NDJSON `page.events` response whose first record is
`{"ready":true}`. Subsequent records are `{event,data}`. Closing a page sends the
terminal record before ending the stream. The native backend uses document-start
notifications for main-document URL/load events and native callbacks for
navigation failures and user closure.
Additional validation:
- `go test -race ./pkg/download/engine/webview ./internal/webview/rpcprovider`
- `go test ./pkg/download -run 'TestExtension(WebViewEvents|InfoHostVersion)'`
- macOS native: `go test -tags webview_native ./internal/webview/goprovider -run TestProviderEvents`
- Running Flutter host: `go test -tags webview ./internal/webview/rpcprovider -run TestProviderEvents`
+48
View File
@@ -0,0 +1,48 @@
/** Public extension API additions. info.version remains the extension version. */
export interface ExtensionInfo {
identity: string;
name: string;
author: string;
title: string;
version: string;
hostVersion: string;
}
/** Functions are serialized; they cannot capture extension-side variables. */
export type WebviewExecutable<T = unknown> =
| string
| ((...args: unknown[]) => T | Promise<T>);
export type WebviewUnsubscribe = () => void;
/** Navigation and load notifications are for the main document only. */
export interface WebviewEventMap {
"url-changed": { url: string; sameDocument: boolean };
/** window.load, including reloads, excluding hash/History API navigation. */
"load": { url: string };
"load-error": { url: string; message: string };
"closed": { reason: "user" | "api" };
}
export interface WebviewGotoOptions {
timeoutMs?: number;
waitUntil?: "load" | "domcontentloaded";
}
/** These members extend the existing WebviewPage API. */
export interface WebviewPage {
/**
* Resolves when listening is ready. Does not replay current state.
* Register before goto() to observe navigation. Unsubscribe is idempotent.
* Closed is delivered once, then listeners are released automatically.
* Callbacks cannot cancel navigation; returned promises are not awaited.
* Callers handle errors in asynchronous callbacks.
*/
on<K extends keyof WebviewEventMap>(
event: K,
handler: (data: WebviewEventMap[K]) => void,
): Promise<WebviewUnsubscribe>;
/** Runs at document creation on future navigations, never immediately. */
addInitScript(script: string): Promise<void>;
goto(url: string, opts?: WebviewGotoOptions): Promise<void>;
}
@@ -0,0 +1,12 @@
//go:build cgo && darwin && webview_native
package goprovider
import (
"github.com/GopeedLab/gopeed/internal/webview/integrationtest"
"os"
"testing"
)
func TestMain(m *testing.M) { os.Exit(RunMainThreadLoop(m.Run)) }
func TestProviderEvents(t *testing.T) { integrationtest.RunEventContract(t, New()) }
@@ -0,0 +1,49 @@
//go:build cgo
package goprovider
import (
enginewebview "github.com/GopeedLab/gopeed/pkg/download/engine/webview"
webview "github.com/GopeedLab/webview_go"
"github.com/dop251/goja"
"testing"
"time"
)
func TestInternalErrorDocumentDoesNotReportPageEvents(t *testing.T) {
vm := goja.New()
if _, err := vm.RunString(`
var window = globalThis;
window.top = window;
var location = {protocol: "chrome-error:", href: "chrome-error://chromewebdata/"};
var notifications = [];
var listeners = [];
function notify(payload) { notifications.push(payload); }
window.addEventListener = (name) => listeners.push(name);
`); err != nil {
t.Fatal(err)
}
if _, err := vm.RunString(buildLoadNotifyScript("notify")); err != nil {
t.Fatal(err)
}
for _, name := range []string{"notifications", "listeners"} {
value, err := vm.RunString(name + ".length")
if err != nil || value.ToInteger() != 0 {
t.Fatalf("internal error page registered %s: %v (%v)", name, value, err)
}
}
// The provider must wait for the native failure, preserving the failed URL.
page := newPageWrapper(enginewebview.OpenOptions{})
got := make(chan string, 1)
_, err := page.On("load-error", func(e enginewebview.Event) { got <- e.Data["url"].(string) })
if err != nil {
t.Fatal(err)
}
page.handleNativeEvent(webview.Event{Name: "load-error", URL: "https://example.test/failure", Message: "TLS handshake failed"})
if err := page.waitForNavigation(time.Second, "load"); err == nil {
t.Fatal("native failure completed navigation successfully")
}
if url := <-got; url != "https://example.test/failure" {
t.Fatal(url)
}
}
+177 -117
View File
@@ -6,7 +6,6 @@ import (
"encoding/json"
"errors"
"fmt"
"net/url"
"strings"
"sync"
"sync/atomic"
@@ -37,7 +36,9 @@ func (p *Provider) Open(opts enginewebview.OpenOptions) (enginewebview.Page, err
}
type pageWrapper struct {
opts enginewebview.OpenOptions
opts enginewebview.OpenOptions
events enginewebview.EventHub
navigationErrors chan error
callbackName string
readyID string
@@ -49,19 +50,23 @@ type pageWrapper struct {
url string
closed bool
ready chan error
done chan struct{}
loads chan string
ready chan error
done chan struct{}
loads chan string
domReady chan struct{}
sameDocument chan struct{}
}
type evalMessage struct {
ID string `json:"id"`
Ready bool `json:"ready,omitempty"`
Load bool `json:"load,omitempty"`
URL string `json:"url,omitempty"`
OK bool `json:"ok,omitempty"`
Value any `json:"value,omitempty"`
Error string `json:"error,omitempty"`
Event string `json:"event,omitempty"`
SameDocument bool `json:"sameDocument,omitempty"`
ID string `json:"id"`
Ready bool `json:"ready,omitempty"`
Load bool `json:"load,omitempty"`
URL string `json:"url,omitempty"`
OK bool `json:"ok,omitempty"`
Value any `json:"value,omitempty"`
Error string `json:"error,omitempty"`
}
type evalResult struct {
@@ -88,13 +93,16 @@ const startupTimeout = time.Minute
func newPageWrapper(opts enginewebview.OpenOptions) *pageWrapper {
seq := providerSeq.Add(1)
return &pageWrapper{
opts: opts,
callbackName: fmt.Sprintf("__gopeedWebViewCallback_%d", seq),
readyID: fmt.Sprintf("__ready__%d", seq),
pending: make(map[string]chan evalResult),
ready: make(chan error, 1),
done: make(chan struct{}),
loads: make(chan string, 8),
opts: opts,
navigationErrors: make(chan error, 8),
callbackName: fmt.Sprintf("__gopeedWebViewCallback_%d", seq),
readyID: fmt.Sprintf("__ready__%d", seq),
pending: make(map[string]chan evalResult),
ready: make(chan error, 1),
done: make(chan struct{}),
loads: make(chan string, 8),
domReady: make(chan struct{}, 8),
sameDocument: make(chan struct{}, 8),
}
}
@@ -119,6 +127,7 @@ func (p *pageWrapper) start() error {
w = webview.New(p.opts.Debug)
}
p.view = w
w.SetEventHandler(p.handleNativeEvent)
w.SetTitle(firstNonEmpty(p.opts.Title, "Gopeed WebView"))
w.SetSize(defaultWindowDimension(p.opts.Width, 1280), defaultWindowDimension(p.opts.Height, 800), webview.HintNone)
@@ -130,7 +139,6 @@ func (p *pageWrapper) start() error {
setter.SetUserAgent(p.opts.UserAgent)
}
}
w.Init(buildLoadNotifyScript(p.callbackName))
if err := w.Bind(p.callbackName, func(payload string) error {
return p.handleCallback(payload)
@@ -141,6 +149,7 @@ func (p *pageWrapper) start() error {
return
}
w.Init(buildLoadNotifyScript(p.callbackName))
applyWindowOptions(w, p.opts)
w.SetHtml(buildBootstrapHTML(p.callbackName, p.readyID))
if runtimepkg.GOOS == "linux" {
@@ -148,6 +157,7 @@ func (p *pageWrapper) start() error {
return
}
w.Run()
p.handleNativeEvent(webview.Event{Name: "closed"})
w.Destroy()
close(p.done)
}
@@ -176,10 +186,15 @@ func (p *pageWrapper) AddInitScript(script string) error {
func (p *pageWrapper) Goto(url string, opts enginewebview.GotoOptions) error {
startedAt := time.Now()
p.drainLoads()
for {
select {
case <-p.navigationErrors:
continue
default:
}
break
}
if err := p.dispatch(func(w webview.WebView) error {
p.mu.Lock()
p.url = url
p.mu.Unlock()
w.Navigate(url)
return nil
}); err != nil {
@@ -193,7 +208,7 @@ func (p *pageWrapper) Goto(url string, opts enginewebview.GotoOptions) error {
if waitUntil == "" {
waitUntil = "load"
}
err := p.waitForNavigation(url, timeout, waitUntil)
err := p.waitForNavigation(timeout, waitUntil)
if err != nil {
p.tracef("goto failed %dms url=%s err=%v", time.Since(startedAt).Milliseconds(), url, err)
return err
@@ -233,7 +248,12 @@ func (p *pageWrapper) Execute(expression string, args ...any) (any, error) {
return nil, err
}
result := <-ch
var result evalResult
select {
case result = <-ch:
case <-p.done:
return nil, fmt.Errorf("webview page is closed")
}
if result.err != nil {
p.tracef("execute failed %dms err=%v", time.Since(startedAt).Milliseconds(), result.err)
return result.value, result.err
@@ -321,6 +341,7 @@ func (p *pageWrapper) Close() error {
return nil
}
p.closed = true
p.failPendingLocked()
w := p.view
wait := p.done
p.mu.Unlock()
@@ -343,6 +364,7 @@ func (p *pageWrapper) Close() error {
<-terminateDone
}
<-wait
p.events.Emit(enginewebview.Event{Name: "closed", Data: map[string]any{"reason": "api"}})
return nil
}
@@ -353,94 +375,29 @@ func (p *pageWrapper) tracef(format string, args ...any) {
fmt.Printf("goprovider "+format+"\n", args...)
}
func (p *pageWrapper) waitForNavigation(targetURL string, timeout time.Duration, waitUntil string) error {
deadline := time.Now().Add(timeout)
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
func (p *pageWrapper) waitForNavigation(timeout time.Duration, waitUntil string) error {
timer := time.NewTimer(timeout)
defer timer.Stop()
for {
select {
case <-p.done:
return fmt.Errorf("webview page is closed")
case err := <-p.navigationErrors:
return err
case <-timer.C:
return fmt.Errorf("webview navigation timeout")
case <-p.sameDocument:
return nil
case <-p.domReady:
if waitUntil == "domcontentloaded" {
return nil
}
case <-p.loads:
if waitUntil == "load" {
return nil
}
case <-ticker.C:
state, err := p.navigationState()
if err == nil && state.ready(targetURL, waitUntil) {
return nil
}
if time.Now().After(deadline) {
return fmt.Errorf("webview navigation timeout")
}
return nil
}
}
}
type navigationState struct {
URL string `json:"url"`
ReadyState string `json:"readyState"`
}
func (s navigationState) ready(targetURL string, waitUntil string) bool {
if s.URL == "" || s.URL == "about:blank" {
return false
}
switch waitUntil {
case "domcontentloaded":
if s.ReadyState == "loading" || s.ReadyState == "" {
return false
}
default:
if s.ReadyState != "complete" {
return false
}
}
if targetURL == "" {
return true
}
if s.URL == targetURL {
return true
}
target, err := parseURL(targetURL)
if err != nil {
return true
}
current, err := parseURL(s.URL)
if err != nil {
return true
}
if current.Scheme == target.Scheme && current.Host == target.Host && current.Path == target.Path {
return true
}
return false
}
func parseURL(raw string) (*url.URL, error) {
return url.Parse(raw)
}
func (p *pageWrapper) navigationState() (navigationState, error) {
value, err := p.Execute(`() => ({
url: String(location.href || ""),
readyState: document.readyState || "",
})`)
if err != nil {
return navigationState{}, err
}
stateMap, ok := value.(map[string]any)
if !ok {
return navigationState{}, fmt.Errorf("invalid navigation state")
}
state := navigationState{}
if urlValue, ok := stateMap["url"].(string); ok {
state.URL = urlValue
}
if readyValue, ok := stateMap["readyState"].(string); ok {
state.ReadyState = readyValue
}
return state, nil
}
func (p *pageWrapper) dispatch(fn func(w webview.WebView) error) error {
p.mu.Lock()
if p.closed {
@@ -457,7 +414,12 @@ func (p *pageWrapper) dispatch(fn func(w webview.WebView) error) error {
w.Dispatch(func() {
done <- fn(w)
})
return <-done
select {
case err := <-done:
return err
case <-p.done:
return fmt.Errorf("webview page is closed")
}
}
func (p *pageWrapper) nativeCookies() ([]nativeCookieJSON, error) {
@@ -515,7 +477,31 @@ func (p *pageWrapper) handleCallback(payload string) error {
}
return nil
}
if msg.Event == "domcontentloaded" {
select {
case p.domReady <- struct{}{}:
default:
}
return nil
}
if msg.Event == "url-changed" {
if msg.SameDocument {
select {
case p.sameDocument <- struct{}{}:
default:
}
}
p.mu.Lock()
changed := p.url != msg.URL
p.url = msg.URL
p.mu.Unlock()
if changed {
p.events.Emit(enginewebview.Event{Name: "url-changed", Data: map[string]any{"url": msg.URL, "sameDocument": msg.SameDocument}})
}
return nil
}
if msg.Load {
p.events.Emit(enginewebview.Event{Name: "load", Data: map[string]any{"url": msg.URL}})
select {
case p.loads <- msg.URL:
default:
@@ -540,6 +526,22 @@ func (p *pageWrapper) handleCallback(payload string) error {
}
func (p *pageWrapper) drainLoads() {
for {
select {
case <-p.domReady:
continue
default:
}
break
}
for {
select {
case <-p.sameDocument:
continue
default:
}
break
}
for {
select {
case <-p.loads:
@@ -601,19 +603,77 @@ func buildExecuteScript(callbackName string, reqID string, expression string, ar
func buildLoadNotifyScript(callbackName string) string {
callbackJSON, _ := json.Marshal(callbackName)
return fmt.Sprintf(`(() => {
window.addEventListener("load", () => {
const __cb = globalThis[%s];
if (typeof __cb !== "function") {
return;
}
__cb(JSON.stringify({
load: true,
url: String(location.href || ""),
}));
}, { once: true });
if (window !== window.top) return;
// WebView2 runs init scripts in its internal error document too. Its load
// must not complete goto() before the native NavigationCompleted failure.
if (location.protocol === "chrome-error:") return;
const send = (message) => {
const cb = globalThis[%s];
if (typeof cb === "function") cb(JSON.stringify(message));
};
let lastURL = String(location.href);
const changed = () => {
const url = String(location.href);
if (url === lastURL) return;
lastURL = url;
send({event: "url-changed", url, sameDocument: true});
};
send({event: "url-changed", url: lastURL, sameDocument: false});
for (const name of ["pushState", "replaceState"]) {
const original = history[name];
history[name] = function(...args) {
const result = Reflect.apply(original, this, args);
changed();
return result;
};
}
window.addEventListener("hashchange", changed);
window.addEventListener("popstate", changed);
window.addEventListener("pageshow", (event) => {
if (event.persisted) send({event: "url-changed", url: String(location.href), sameDocument: false});
else changed();
});
window.addEventListener("DOMContentLoaded", () => send({event: "domcontentloaded"}), {once:true});
window.addEventListener("load", () => send({load: true, url: String(location.href)}), {once: true});
})();`, string(callbackJSON))
}
func (p *pageWrapper) On(name string, handler func(enginewebview.Event)) (func(), error) {
return p.events.On(name, handler)
}
func (p *pageWrapper) failPendingLocked() {
for id, ch := range p.pending {
ch <- evalResult{err: fmt.Errorf("webview page is closed")}
delete(p.pending, id)
}
}
func (p *pageWrapper) handleNativeEvent(event webview.Event) {
switch event.Name {
case "load-error":
p.events.Emit(enginewebview.Event{Name: "load-error", Data: map[string]any{"url": event.URL, "message": event.Message}})
select {
case p.navigationErrors <- errors.New(event.Message):
default:
}
case "closed":
p.mu.Lock()
api := p.closed
p.closed = true
p.failPendingLocked()
p.mu.Unlock()
reason := "user"
if api {
reason = "api"
}
p.events.Emit(enginewebview.Event{Name: "closed", Data: map[string]any{"reason": reason}})
if runtimepkg.GOOS == "linux" && !api {
p.view.Dispatch(func() { p.view.Destroy(); close(p.done) })
}
}
}
func defaultWindowDimension(value int, fallback int) int {
if value > 0 {
return value
@@ -15,3 +15,5 @@ func TestProviderContract(t *testing.T) {
}
func TestProviderProfiles(t *testing.T) { integrationtest.RunProfileContract(t, New()) }
func TestProviderEvents(t *testing.T) { integrationtest.RunEventContract(t, New()) }
+12 -3
View File
@@ -108,8 +108,10 @@ func RunProviderContract(t *testing.T, provider enginewebview.Provider, opts Con
t.Fatalf("expected click count to be 1, got %v", clickCount)
}
// Hidden WebViews can throttle timers on shared CI runners. Keep the
// positive asynchronous waits within the same budget as navigation.
found, err := page.WaitForSelector("#delayed", map[string]any{
"timeoutMs": 5000,
"timeoutMs": 10000,
"pollIntervalMs": 25,
"visible": true,
})
@@ -117,11 +119,18 @@ func RunProviderContract(t *testing.T, provider enginewebview.Provider, opts Con
t.Fatalf("waitForSelector failed: %v", err)
}
if !found {
t.Fatal("expected delayed selector to appear")
diagnostic, diagnosticErr := page.Execute(`() => {
const element = document.querySelector('#delayed');
const rect = element && element.getBoundingClientRect();
return {url: location.href, readyState: document.readyState,
visibility: document.visibilityState, exists: !!element,
width: rect && rect.width, height: rect && rect.height};
}`)
t.Fatalf("expected delayed selector to appear: state=%v, diagnostic error=%v", diagnostic, diagnosticErr)
}
value, err := page.WaitForFunction(`() => window.__readyValue`, map[string]any{
"timeoutMs": 5000,
"timeoutMs": 10000,
"pollIntervalMs": 25,
})
if err != nil {
+110
View File
@@ -0,0 +1,110 @@
package integrationtest
import (
"fmt"
"strings"
"testing"
"time"
wv "github.com/GopeedLab/gopeed/pkg/download/engine/webview"
)
// RunEventContract exercises the same public semantics on native and RPC hosts.
func RunEventContract(t *testing.T, provider wv.Provider) {
t.Helper()
page, _, url := openTestPage(t, provider, wv.OpenOptions{Headless: true, Title: "WebView event contract"})
events := make(chan wv.Event, 100)
for _, name := range []string{"url-changed", "load", "load-error", "closed"} {
if _, err := page.On(name, func(e wv.Event) { events <- e }); err != nil {
t.Fatal(err)
}
}
next := func(name string) wv.Event {
t.Helper()
select {
case event := <-events:
if event.Name != name {
t.Fatalf("got event %s (%v), want %s", event.Name, event.Data, name)
}
return event
case <-time.After(10 * time.Second):
t.Fatalf("timed out waiting for %s", name)
}
return wv.Event{}
}
quiet := func() {
t.Helper()
select {
case e := <-events:
t.Fatalf("unexpected event: %+v", e)
case <-time.After(150 * time.Millisecond):
}
}
quiet()
if err := page.AddInitScript(`globalThis.__initBeforePage = document.readyState`); err != nil {
t.Fatal(err)
}
if err := page.Goto(url, map[string]any{"timeoutMs": 10000}); err != nil {
t.Fatal(err)
}
e := next("url-changed")
if strings.TrimRight(fmt.Sprint(e.Data["url"]), "/") != strings.TrimRight(url, "/") || e.Data["sameDocument"] != false {
t.Fatal(e)
}
next("load")
if result, err := page.Execute(`() => globalThis.__initBeforePage`); err != nil || result != "loading" {
t.Fatalf("init script: %v %v", result, err)
}
if err := page.AddInitScript(`globalThis.__onlyNext = true`); err != nil {
t.Fatal(err)
}
if result, err := page.Execute(`() => typeof globalThis.__onlyNext`); err != nil || result != "undefined" {
t.Fatalf("init ran in current page: %v %v", result, err)
}
for _, expression := range []string{
`() => { history.pushState({}, '', '#push'); }`,
`() => { history.replaceState({}, '', '#replace'); }`,
`() => { location.hash = 'hash'; }`,
} {
if _, err := page.Execute(expression); err != nil {
t.Fatal(err)
}
e := next("url-changed")
if e.Data["sameDocument"] != true {
t.Fatal(e)
}
quiet()
}
if _, err := page.Execute(`() => { location.reload(); }`); err != nil {
t.Fatal(err)
}
next("load") // Same address on refresh is not a URL change.
if result, err := page.Execute(`() => globalThis.__onlyNext`); err != nil || result != true {
t.Fatalf("next init: %v %v", result, err)
}
if _, err := page.Execute(`() => { const f=document.createElement('iframe'); f.src='/next'; document.body.append(f); }`); err != nil {
t.Fatal(err)
}
quiet()
// The fixture serves plain HTTP. Attempting TLS against it produces a
// real navigation failure through both HTTP CONNECT and SOCKS proxies.
// An unreachable HTTP origin would instead become a valid 502 document
// when the host's HTTP proxy handles the connection error.
failedURL := strings.Replace(url, "http://", "https://", 1) + "/failure"
if err := page.Goto(failedURL, map[string]any{"timeoutMs": 3000}); err == nil {
t.Fatal("failed navigation succeeded")
}
e = next("load-error")
if e.Data["message"] == "" || !strings.Contains(fmt.Sprint(e.Data["url"]), "/failure") {
t.Fatal(e)
}
quiet() // A handled failure must not load a browser-generated fallback page.
if err := page.Close(); err != nil {
t.Fatal(err)
}
e = next("closed")
if e.Data["reason"] != "api" {
t.Fatal(e)
}
quiet()
}
+207
View File
@@ -0,0 +1,207 @@
package rpcprovider
import (
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
wv "github.com/GopeedLab/gopeed/pkg/download/engine/webview"
)
func TestPageEventStream(t *testing.T) {
events := make(chan wv.Event, 8)
disconnected := make(chan struct{})
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var request wv.RPCRequest
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
t.Error(err)
return
}
if request.Method != "page.events" {
t.Errorf("unexpected method %s", request.Method)
return
}
if r.Header.Get("Authorization") != "Bearer test" {
t.Error("missing authorization")
}
defer close(disconnected)
enc := json.NewEncoder(w)
_ = enc.Encode(map[string]any{"ready": true})
w.(http.Flusher).Flush()
for {
select {
case e := <-events:
_ = enc.Encode(e)
w.(http.Flusher).Flush()
if e.Name == "closed" {
return
}
case <-r.Context().Done():
return
}
}
}))
defer server.Close()
p := &page{client: NewClient(wv.RPCConfig{Network: "tcp", Address: strings.TrimPrefix(server.URL, "http://"), Token: "test"}), id: "page-1"}
got := make(chan wv.Event, 8)
off, err := p.On("load", func(e wv.Event) { got <- e })
if err != nil {
t.Fatal(err)
}
_, err = p.On("closed", func(e wv.Event) { got <- e })
if err != nil {
t.Fatal(err)
}
events <- wv.Event{Name: "load", Data: map[string]any{"url": "https://example.test"}}
select {
case e := <-got:
if e.Name != "load" {
t.Fatal(e)
}
case <-time.After(time.Second):
t.Fatal("no load event")
}
off()
off()
events <- wv.Event{Name: "load"}
events <- wv.Event{Name: "closed", Data: map[string]any{"reason": "user"}}
select {
case e := <-got:
if e.Name != "closed" || e.Data["reason"] != "user" {
t.Fatal(e)
}
case <-time.After(time.Second):
t.Fatal("no closed event")
}
select {
case <-disconnected:
case <-time.After(time.Second):
t.Fatal("stream retained after close")
}
if _, err = p.On("load", func(wv.Event) {}); err == nil {
t.Fatal("registration after close")
}
}
func TestPageEventStreamHandshakeFailureCanRetry(t *testing.T) {
for _, tc := range []struct {
name string
status int
body string
}{
{"http_error", http.StatusUnauthorized, `{"ready":true}`},
{"invalid_json", http.StatusOK, `not json`},
{"not_ready", http.StatusOK, `{"ready":false}`},
} {
t.Run(tc.name, func(t *testing.T) {
var requests atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if requests.Add(1) == 1 {
w.WriteHeader(tc.status)
_, _ = io.WriteString(w, tc.body)
return
}
_, _ = io.WriteString(w, "{\"ready\":true}\n{\"event\":\"load\",\"data\":{\"url\":\"https://example.test\"}}\n")
w.(http.Flusher).Flush()
<-r.Context().Done()
}))
defer server.Close()
p := &page{client: NewClient(wv.RPCConfig{Network: "tcp", Address: strings.TrimPrefix(server.URL, "http://")}), id: "retry-page"}
if off, err := p.On("load", func(wv.Event) { t.Error("failed registration retained its listener") }); err == nil || off != nil {
t.Fatal("failed handshake accepted")
}
got := make(chan wv.Event, 1)
off, err := p.On("load", func(e wv.Event) { got <- e })
if err != nil {
t.Fatalf("retry: %v", err)
}
defer func() {
p.eventCancel()
<-p.eventDone
}()
defer off()
select {
case e := <-got:
if e.Data["url"] != "https://example.test" {
t.Fatal(e)
}
case <-time.After(time.Second):
t.Fatal("retry did not receive load notification")
}
if requests.Load() != 2 {
t.Fatalf("unexpected request count: %d", requests.Load())
}
})
}
}
func TestPageEventStreamDisconnectAndClose(t *testing.T) {
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
var request wv.RPCRequest
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
t.Error(err)
return
}
switch request.Method {
case wv.MethodPageEvents:
// End the stream without a terminal event, as if the host dropped it.
_, _ = io.WriteString(w, "{\"ready\":true}\n")
case wv.MethodPageClose:
_ = json.NewEncoder(w).Encode(wv.RPCResponse{})
default:
t.Errorf("unexpected method: %s", request.Method)
}
}))
defer server.Close()
p := &page{client: NewClient(wv.RPCConfig{Network: "tcp", Address: strings.TrimPrefix(server.URL, "http://")}), id: "disconnected-page"}
closed := make(chan wv.Event, 1)
if _, err := p.On("closed", func(e wv.Event) { closed <- e }); err != nil {
t.Fatal(err)
}
select {
case <-p.eventDone:
case <-time.After(time.Second):
t.Fatal("disconnected stream did not stop")
}
if _, err := p.On("load", func(wv.Event) {}); err == nil || !strings.Contains(err.Error(), "disconnected") {
t.Fatalf("registration hid stream failure: %v", err)
}
if err := p.Close(); err != nil {
t.Fatal(err)
}
select {
case event := <-closed:
if event.Data["reason"] != "api" {
t.Fatal(event)
}
default:
t.Fatal("API close did not release listeners after disconnect")
}
if err := p.Close(); err != nil {
t.Fatal(err)
}
select {
case event := <-closed:
t.Fatalf("duplicate close notification: %+v", event)
default:
}
}
func TestPageEventStreamConnectionFailureCanRetry(t *testing.T) {
server := httptest.NewServer(http.NotFoundHandler())
server.Close()
p := &page{client: NewClient(wv.RPCConfig{Network: "tcp", Address: strings.TrimPrefix(server.URL, "http://")}), id: "offline-page"}
for i := 0; i < 2; i++ {
if off, err := p.On("load", func(wv.Event) {}); err == nil || off != nil {
t.Fatal("registration to unavailable host succeeded")
}
}
if p.eventCancel != nil {
t.Fatal("failed connection retained an active stream")
}
}
+104 -7
View File
@@ -9,6 +9,8 @@ import (
"net"
"net/http"
"strings"
"sync"
"time"
enginewebview "github.com/GopeedLab/gopeed/pkg/download/engine/webview"
)
@@ -49,8 +51,13 @@ func (p *Provider) Open(opts enginewebview.OpenOptions) (enginewebview.Page, err
}
type page struct {
client *Client
id string
eventMu sync.Mutex
events enginewebview.EventHub
eventCancel context.CancelFunc
eventDone chan struct{}
eventErr error
client *Client
id string
}
func (p *page) AddInitScript(script string) error {
@@ -110,14 +117,28 @@ func (p *page) ClearCookies() error {
}
func (p *page) Close() error {
err := p.client.Call(enginewebview.MethodPageClose, enginewebview.PageCloseParams{
PageID: p.id,
}, nil)
err := p.client.Call(enginewebview.MethodPageClose, enginewebview.PageCloseParams{PageID: p.id}, nil)
var rpcErr *enginewebview.RPCError
if errors.As(err, &rpcErr) && rpcErr.Code == enginewebview.ErrorCodePageNotFound {
return nil
err = nil
}
return err
if err != nil {
return err
}
p.eventMu.Lock()
cancel, done := p.eventCancel, p.eventDone
p.eventMu.Unlock()
if cancel != nil {
// Drain the terminal notification before cancelling the transport, so
// already-sent navigation events keep their order before closed.
select {
case <-done:
case <-time.After(5 * time.Second):
}
cancel()
}
p.events.Emit(enginewebview.Event{Name: "closed", Data: map[string]any{"reason": "api"}})
return nil
}
type Client struct {
@@ -203,3 +224,79 @@ func (c *Client) Call(method string, params any, result any) error {
func (p *Provider) RemoveProfile(profileID, dataPath string) error {
return p.client.Call(enginewebview.MethodProfileRemove, enginewebview.ProfileRemoveParams{ProfileID: profileID}, nil)
}
// On starts one ordered notification stream per page. The ready record is sent
// only after the host has installed its subscription.
func (p *page) On(name string, handler func(enginewebview.Event)) (func(), error) {
p.eventMu.Lock()
defer p.eventMu.Unlock()
if p.eventErr != nil {
return nil, p.eventErr
}
unsubscribe, err := p.events.On(name, handler)
if err != nil {
return nil, err
}
if p.eventCancel != nil {
return unsubscribe, nil
}
ctx, cancel := context.WithCancel(context.Background())
payload, err := json.Marshal(enginewebview.RPCRequest{Method: enginewebview.MethodPageEvents, Params: enginewebview.PageCloseParams{PageID: p.id}})
if err != nil {
cancel()
unsubscribe()
return nil, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.client.endpoint, bytes.NewReader(payload))
if err != nil {
cancel()
unsubscribe()
return nil, err
}
req.Header.Set("Content-Type", "application/json")
if p.client.token != "" {
req.Header.Set("Authorization", "Bearer "+p.client.token)
}
// A broken host must not leave registration pending indefinitely.
timer := time.AfterFunc(30*time.Second, cancel)
resp, err := p.client.http.Do(req)
if err != nil {
timer.Stop()
cancel()
unsubscribe()
return nil, err
}
decoder := json.NewDecoder(resp.Body)
var ready struct {
Ready bool `json:"ready"`
}
err = decoder.Decode(&ready)
timer.Stop()
if err != nil || !ready.Ready || resp.StatusCode != http.StatusOK {
resp.Body.Close()
cancel()
unsubscribe()
return nil, fmt.Errorf("webview event stream did not become ready: %v", err)
}
p.eventCancel = cancel
p.eventDone = make(chan struct{})
go func() {
defer resp.Body.Close()
defer cancel()
defer close(p.eventDone)
for {
var event enginewebview.Event
if err := decoder.Decode(&event); err != nil {
p.eventMu.Lock()
p.eventErr = fmt.Errorf("webview event stream disconnected: %w", err)
p.eventMu.Unlock()
return
}
p.events.Emit(event)
if event.Name == "closed" {
return
}
}
}()
return unsubscribe, nil
}
@@ -42,3 +42,7 @@ func TestProviderProfiles(t *testing.T) {
provider := New(enginewebview.RPCConfig{Network: *webViewRPCNetwork, Address: *webViewRPCAddress, Token: *webViewRPCToken})
integrationtest.RunProfileContract(t, provider)
}
func TestProviderEvents(t *testing.T) {
integrationtest.RunEventContract(t, New(enginewebview.RPCConfig{Network: *webViewRPCNetwork, Address: *webViewRPCAddress, Token: *webViewRPCToken}))
}