mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-11 02:50:27 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7cf53240c2 | ||
|
|
1bc31774cb | ||
|
|
ed125c2f19 |
No files matched your search
@@ -418,16 +418,6 @@ func (f *Fetcher) updateMaxConnTime(d time.Duration) {
|
||||
}
|
||||
|
||||
func (f *Fetcher) Resolve(req *base.Request) error {
|
||||
return f.ResolveContext(context.Background(), req)
|
||||
}
|
||||
|
||||
// ResolveContext cancels the initial request while waiting for headers. Once
|
||||
// Resolve has returned, the retained response belongs to the fetcher and its
|
||||
// existing Pause/Close lifecycle controls the body and prefetch.
|
||||
func (f *Fetcher) ResolveContext(ctx context.Context, req *base.Request) error {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := base.ParseReqExtra[fhttp.ReqExtra](req); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -439,8 +429,6 @@ func (f *Fetcher) ResolveContext(ctx context.Context, req *base.Request) error {
|
||||
// This allows the response to be reused for downloading (important for one-time URLs)
|
||||
// The body outlives Resolve, so the read timeout gets its own cancel func.
|
||||
reqCtx, reqCancel := context.WithCancel(context.Background())
|
||||
stopCancel := context.AfterFunc(ctx, reqCancel)
|
||||
defer stopCancel()
|
||||
httpReq, err := f.buildRequest(reqCtx, req)
|
||||
if err != nil {
|
||||
reqCancel()
|
||||
|
||||
@@ -1,36 +0,0 @@
|
||||
package api
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
"github.com/GopeedLab/gopeed/pkg/download"
|
||||
restmodel "github.com/GopeedLab/gopeed/pkg/rest/model"
|
||||
)
|
||||
|
||||
func TestCancelledTorrentPreviewCanBeResolvedAgain(t *testing.T) {
|
||||
d := download.NewDownloader(&download.DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir()})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Close()
|
||||
service, err := NewService(d)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
path, err := filepath.Abs("../../internal/protocol/bt/testdata/test.torrent")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req := &restmodel.ResolveTask{Req: &base.Request{URL: path}}
|
||||
first := serviceTestRequest[*download.ResolveResult](t, service, http.MethodPost, "/api/v1/resolve", req)
|
||||
serviceTestRequest[any](t, service, http.MethodDelete, "/api/v1/resolve/"+first.ID, nil)
|
||||
// Release is idempotent, including an old window disposing after Create.
|
||||
serviceTestRequest[any](t, service, http.MethodDelete, "/api/v1/resolve/"+first.ID, nil)
|
||||
second := serviceTestRequest[*download.ResolveResult](t, service, http.MethodPost, "/api/v1/resolve", req)
|
||||
if second.Res.Hash != first.Res.Hash {
|
||||
t.Fatalf("different torrent after release: %v", second)
|
||||
}
|
||||
}
|
||||
+1
-9
@@ -51,7 +51,6 @@ func NewService(downloader *download.Downloader) (*Service, error) {
|
||||
service.routes = []RouteSpec{
|
||||
newRoute(http.MethodGet, "/api/v1/info", service.info),
|
||||
newRoute(http.MethodPost, "/api/v1/resolve", service.resolve),
|
||||
newRoute(http.MethodDelete, "/api/v1/resolve/{id}", service.releaseResolve),
|
||||
newRoute(http.MethodPost, "/api/v1/tasks", service.createTask),
|
||||
newRoute(http.MethodPost, "/api/v1/tasks/batch", service.createTaskBatch),
|
||||
newRoute(http.MethodPatch, "/api/v1/tasks/{id}", service.patchTask),
|
||||
@@ -144,20 +143,13 @@ func (s *Service) resolve(ctx *Context) *Response {
|
||||
if response := decodeRequest(ctx.Body, &request); response != nil {
|
||||
return response
|
||||
}
|
||||
result, err := s.Downloader.ResolveWithID(request.ID, request.Req)
|
||||
result, err := s.Downloader.Resolve(request.Req)
|
||||
if err != nil {
|
||||
return newErrorResponse(err.Error())
|
||||
}
|
||||
return newOKResponse(result)
|
||||
}
|
||||
|
||||
func (s *Service) releaseResolve(ctx *Context) *Response {
|
||||
if err := s.Downloader.ReleaseResolve(ctx.PathParam("id")); err != nil {
|
||||
return newErrorResponse(err.Error())
|
||||
}
|
||||
return newOKResponse(nil)
|
||||
}
|
||||
|
||||
func (s *Service) createTask(ctx *Context) *Response {
|
||||
var request restmodel.CreateTask
|
||||
if response := decodeRequest(ctx.Body, &request); response != nil {
|
||||
|
||||
+32
-163
@@ -116,7 +116,6 @@ type Downloader struct {
|
||||
|
||||
cfg *DownloaderConfig
|
||||
fetcherCache map[string]fetcher.Fetcher
|
||||
resolveStates map[string]*resolveState
|
||||
storage Storage
|
||||
tasks []*Task
|
||||
waitTasks []*Task
|
||||
@@ -154,7 +153,6 @@ func NewDownloader(cfg *DownloaderConfig) *Downloader {
|
||||
cancelBackground: cancelBackground,
|
||||
cfg: cfg,
|
||||
fetcherCache: make(map[string]fetcher.Fetcher),
|
||||
resolveStates: make(map[string]*resolveState),
|
||||
waitTasks: make([]*Task, 0),
|
||||
groups: make(map[string]*TaskGroup),
|
||||
creatingGroups: make(map[string]struct{}),
|
||||
@@ -561,156 +559,42 @@ func ensureResourceRequestRawURLs(parentReq *base.Request, res *base.Resource) {
|
||||
}
|
||||
}
|
||||
|
||||
// resolveState owns a Fetcher until Create transfers it to a task.
|
||||
// State transitions are protected by fetcherMapLock; closeOnce serializes cleanup.
|
||||
type contextResolver interface {
|
||||
ResolveContext(context.Context, *base.Request) error
|
||||
}
|
||||
|
||||
type resolveState struct {
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
fetcher fetcher.Fetcher
|
||||
cancelled bool
|
||||
resolved bool
|
||||
claimed bool
|
||||
closeOnce sync.Once
|
||||
closeErr error
|
||||
}
|
||||
|
||||
func (s *resolveState) close(f fetcher.Fetcher) error {
|
||||
if f == nil {
|
||||
return nil
|
||||
func (d *Downloader) Resolve(req *base.Request) (rr *ResolveResult, err error) {
|
||||
rrId, err := gonanoid.New()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
s.closeOnce.Do(func() { s.closeErr = f.Close() })
|
||||
return s.closeErr
|
||||
}
|
||||
|
||||
func (d *Downloader) Resolve(req *base.Request) (*ResolveResult, error) {
|
||||
return d.ResolveWithID("", req)
|
||||
}
|
||||
|
||||
// ResolveWithID lets a preview cancel before metadata is available. Clients
|
||||
// that do not supply an ID retain the server-generated resource ID behavior.
|
||||
func (d *Downloader) ResolveWithID(id string, req *base.Request) (rr *ResolveResult, err error) {
|
||||
if id == "" {
|
||||
id, err = gonanoid.New()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if req == nil {
|
||||
return nil, errors.New("resolve request is required")
|
||||
}
|
||||
resolveCtx, cancel := context.WithCancel(d.backgroundCtx)
|
||||
defer cancel()
|
||||
state := &resolveState{ctx: resolveCtx, cancel: cancel}
|
||||
d.fetcherMapLock.Lock()
|
||||
if d.closed.Load() {
|
||||
d.fetcherMapLock.Unlock()
|
||||
return nil, errors.New("downloader is closed")
|
||||
}
|
||||
if d.resolveStates[id] != nil || d.fetcherCache[id] != nil {
|
||||
d.fetcherMapLock.Unlock()
|
||||
return nil, errors.New("resource id is already in use")
|
||||
}
|
||||
d.resolveStates[id] = state
|
||||
d.fetcherMapLock.Unlock()
|
||||
defer func() {
|
||||
if err != nil || state.fetcher == nil {
|
||||
d.fetcherMapLock.Lock()
|
||||
if d.resolveStates[id] == state {
|
||||
delete(d.resolveStates, id)
|
||||
}
|
||||
d.fetcherMapLock.Unlock()
|
||||
_ = state.close(state.fetcher)
|
||||
}
|
||||
}()
|
||||
ensureRequestRawURL(req)
|
||||
|
||||
res, err := d.triggerOnResolve(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return
|
||||
}
|
||||
if res != nil && len(res.Files) > 0 {
|
||||
return &ResolveResult{Res: res, SuggestedDownloadDir: d.suggestedDownloadDir(res)}, nil
|
||||
}
|
||||
f, err := d.buildFetcher(req.URL)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
d.fetcherMapLock.Lock()
|
||||
state.fetcher = f
|
||||
cancelled := state.cancelled || d.closed.Load()
|
||||
d.fetcherMapLock.Unlock()
|
||||
if cancelled {
|
||||
return nil, errors.New("resource resolution cancelled")
|
||||
}
|
||||
if resolver, ok := f.(contextResolver); ok {
|
||||
err = resolver.ResolveContext(state.ctx, req)
|
||||
} else {
|
||||
err = f.Resolve(req)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
d.fetcherMapLock.Lock()
|
||||
defer d.fetcherMapLock.Unlock()
|
||||
if state.cancelled || d.closed.Load() {
|
||||
return nil, errors.New("resource resolution cancelled")
|
||||
}
|
||||
state.resolved = true
|
||||
d.fetcherCache[id] = f
|
||||
return &ResolveResult{ID: id, Res: f.Meta().Res, SuggestedDownloadDir: d.suggestedDownloadDir(f.Meta().Res)}, nil
|
||||
}
|
||||
|
||||
// ReleaseResolve is idempotent. A resource already transferred to a task is
|
||||
// never closed by a late preview cancellation.
|
||||
func (d *Downloader) ReleaseResolve(id string) error {
|
||||
d.fetcherMapLock.Lock()
|
||||
state := d.resolveStates[id]
|
||||
f := d.fetcherCache[id]
|
||||
delete(d.fetcherCache, id)
|
||||
delete(d.resolveStates, id)
|
||||
if state != nil {
|
||||
state.cancelled = true
|
||||
state.cancel()
|
||||
if state.claimed {
|
||||
f = nil
|
||||
} else {
|
||||
f = state.fetcher
|
||||
if _, cancellable := f.(contextResolver); cancellable && !state.resolved {
|
||||
// Resolve owns unpublished response/prefetch state. Its context
|
||||
// cancels the request; the returning resolver closes it once
|
||||
// publication has stopped, including a racing response.
|
||||
f = nil
|
||||
}
|
||||
rr = &ResolveResult{
|
||||
Res: res,
|
||||
SuggestedDownloadDir: d.suggestedDownloadDir(res),
|
||||
}
|
||||
return
|
||||
}
|
||||
d.fetcherMapLock.Unlock()
|
||||
if state != nil {
|
||||
return state.close(f)
|
||||
}
|
||||
if f != nil {
|
||||
return f.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Downloader) closeResolvedResources() {
|
||||
fetcher, err := d.buildFetcher(req.URL)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
err = fetcher.Resolve(req)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
d.fetcherMapLock.Lock()
|
||||
ids := make(map[string]struct{}, len(d.resolveStates)+len(d.fetcherCache))
|
||||
for id := range d.resolveStates {
|
||||
ids[id] = struct{}{}
|
||||
}
|
||||
for id := range d.fetcherCache {
|
||||
ids[id] = struct{}{}
|
||||
}
|
||||
d.fetcherCache[rrId] = fetcher
|
||||
d.fetcherMapLock.Unlock()
|
||||
for id := range ids {
|
||||
if err := d.ReleaseResolve(id); err != nil {
|
||||
d.Logger.Warn().Err(err).Msg("close resolved resource failed")
|
||||
}
|
||||
rr = &ResolveResult{
|
||||
ID: rrId,
|
||||
Res: fetcher.Meta().Res,
|
||||
SuggestedDownloadDir: d.suggestedDownloadDir(fetcher.Meta().Res),
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (d *Downloader) notifyRunning() {
|
||||
@@ -984,34 +868,18 @@ func (d *Downloader) CreateWithOptions(rrId string, opts *base.Options) (string,
|
||||
func (d *Downloader) createResolved(rrId string, opts *base.Options, groupID string) (taskId string, err error) {
|
||||
d.fetcherMapLock.Lock()
|
||||
fetcher, ok := d.fetcherCache[rrId]
|
||||
state := d.resolveStates[rrId]
|
||||
if ok && state != nil {
|
||||
state.claimed = true
|
||||
}
|
||||
delete(d.fetcherCache, rrId)
|
||||
d.fetcherMapLock.Unlock()
|
||||
if !ok {
|
||||
return "", errors.New("invalid resource id")
|
||||
}
|
||||
// Failed creation returns ownership to the preview unless it was cancelled.
|
||||
// Only one creator may use a resolved fetcher. Failed creation makes it
|
||||
// available again; a persisted task consumes it even if starting fails.
|
||||
defer func() {
|
||||
d.fetcherMapLock.Lock()
|
||||
cancelled := d.closed.Load() || (state != nil && state.cancelled)
|
||||
if taskId == "" && !cancelled {
|
||||
if taskId == "" {
|
||||
d.fetcherMapLock.Lock()
|
||||
d.fetcherCache[rrId] = fetcher
|
||||
if state != nil {
|
||||
state.claimed = false
|
||||
}
|
||||
} else if d.resolveStates[rrId] == state {
|
||||
delete(d.resolveStates, rrId)
|
||||
}
|
||||
d.fetcherMapLock.Unlock()
|
||||
if taskId == "" && cancelled {
|
||||
if state != nil {
|
||||
_ = state.close(fetcher)
|
||||
} else {
|
||||
_ = fetcher.Close()
|
||||
}
|
||||
d.fetcherMapLock.Unlock()
|
||||
}
|
||||
}()
|
||||
if opts == nil {
|
||||
@@ -1694,7 +1562,6 @@ func (d *Downloader) Close() error {
|
||||
return nil
|
||||
}
|
||||
d.cancelBackground()
|
||||
d.closeResolvedResources()
|
||||
d.waitForExtensionGit()
|
||||
if d.protocolJobs != nil {
|
||||
d.protocolJobs.close()
|
||||
@@ -2538,7 +2405,9 @@ func (d *Downloader) doStart(task *Task) (err error) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
d.triggerOnStart(task)
|
||||
if err := d.triggerOnStart(task); err != nil {
|
||||
return err
|
||||
}
|
||||
if !d.taskIsRunningGeneration(task, generation) {
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,47 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/dop251/goja"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestCloseUnblocksPendingPromise(t *testing.T) {
|
||||
e := NewEngine(nil)
|
||||
entered := make(chan struct{})
|
||||
_, err := e.runOnLoop(func(r *goja.Runtime) (goja.Value, error) {
|
||||
return goja.Undefined(), r.Set("entered", func() { close(entered) })
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
done := make(chan error, 1)
|
||||
go func() { _, err := e.RunString(`entered(); new Promise(()=>{});`); done <- err }()
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("script did not enter")
|
||||
}
|
||||
e.Close()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err == nil {
|
||||
t.Fatal("closed execution returned success")
|
||||
}
|
||||
case <-time.After(300 * time.Millisecond):
|
||||
t.Fatal("Engine.Close left the Promise waiter blocked")
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecutionDeadlineInterruptsSynchronousScript(t *testing.T) {
|
||||
e := NewEngine(nil)
|
||||
defer e.Close()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
|
||||
defer cancel()
|
||||
_, err := e.RunStringContext(ctx, `while (true) {}`)
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("got %v, want deadline exceeded", err)
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"context"
|
||||
_ "embed"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -26,6 +27,7 @@ var polyfillScript string
|
||||
|
||||
type Engine struct {
|
||||
loop *eventloop.EventLoop
|
||||
done chan struct{}
|
||||
|
||||
closeOnce sync.Once
|
||||
cleanupMu sync.Mutex
|
||||
@@ -39,7 +41,13 @@ type JSFunction func(goja.FunctionCall) goja.Value
|
||||
// RunString executes the script and returns the go type value
|
||||
// if script result is promise, it will be resolved
|
||||
func (e *Engine) RunString(script string) (value any, err error) {
|
||||
return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) {
|
||||
return e.RunStringContext(context.Background(), script)
|
||||
}
|
||||
|
||||
// RunStringContext bounds an execution, including awaited promises. Cancelling
|
||||
// an execution closes its engine so outstanding operations cannot outlive it.
|
||||
func (e *Engine) RunStringContext(ctx context.Context, script string) (any, error) {
|
||||
return e.runOnLoopContext(ctx, func(runtime *goja.Runtime) (goja.Value, error) {
|
||||
return runtime.RunString(script)
|
||||
})
|
||||
}
|
||||
@@ -47,7 +55,11 @@ func (e *Engine) RunString(script string) (value any, err error) {
|
||||
// CallFunction calls the function and returns the go type value
|
||||
// if function result is promise, it will be resolved
|
||||
func (e *Engine) CallFunction(fn any, args ...any) (value any, err error) {
|
||||
return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) {
|
||||
return e.CallFunctionContext(context.Background(), fn, args...)
|
||||
}
|
||||
|
||||
func (e *Engine) CallFunctionContext(ctx context.Context, fn any, args ...any) (value any, err error) {
|
||||
return e.runOnLoopContext(ctx, func(runtime *goja.Runtime) (goja.Value, error) {
|
||||
var jsArgs []goja.Value
|
||||
for _, arg := range args {
|
||||
jsArgs = append(jsArgs, runtime.ToValue(arg))
|
||||
@@ -69,6 +81,20 @@ func (e *Engine) CallFunction(fn any, args ...any) (value any, err error) {
|
||||
}
|
||||
|
||||
func (e *Engine) runOnLoop(fn func(runtime *goja.Runtime) (goja.Value, error)) (any, error) {
|
||||
return e.runOnLoopContext(context.Background(), fn)
|
||||
}
|
||||
|
||||
var ErrClosed = errors.New("engine closed")
|
||||
|
||||
func (e *Engine) runOnLoopContext(ctx context.Context, fn func(runtime *goja.Runtime) (goja.Value, error)) (any, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
select {
|
||||
case <-e.done:
|
||||
return nil, ErrClosed
|
||||
default:
|
||||
}
|
||||
type result struct {
|
||||
value any
|
||||
err error
|
||||
@@ -132,15 +158,31 @@ func (e *Engine) runOnLoop(fn func(runtime *goja.Runtime) (goja.Value, error)) (
|
||||
if !ok {
|
||||
return nil, errors.New("engine loop terminated")
|
||||
}
|
||||
res := <-ch
|
||||
if res.err != nil {
|
||||
return nil, res.err
|
||||
select {
|
||||
case res := <-ch:
|
||||
return res.value, res.err
|
||||
case <-ctx.Done():
|
||||
// Interrupt synchronous JS too; a select alone only stops Promise waiting.
|
||||
e.Close()
|
||||
return nil, ctx.Err()
|
||||
case <-e.done:
|
||||
return nil, ErrClosed
|
||||
}
|
||||
return res.value, nil
|
||||
}
|
||||
|
||||
func (e *Engine) Close() {
|
||||
e.closeOnce.Do(func() {
|
||||
close(e.done)
|
||||
e.Runtime.Interrupt(ErrClosed)
|
||||
// Wait for interrupted JS to leave the loop, then allow cleanup hooks
|
||||
// to execute their JS cancellation callbacks before terminating it.
|
||||
cleared := make(chan struct{})
|
||||
if e.loop.RunOnLoop(func(runtime *goja.Runtime) {
|
||||
runtime.ClearInterrupt()
|
||||
close(cleared)
|
||||
}) {
|
||||
<-cleared
|
||||
}
|
||||
e.cleanupMu.Lock()
|
||||
cleanups := e.cleanups
|
||||
e.cleanups = nil
|
||||
@@ -177,6 +219,7 @@ func NewEngine(cfg *Config) *Engine {
|
||||
loop := eventloop.NewEventLoop()
|
||||
engine := &Engine{
|
||||
loop: loop,
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
loop.Start()
|
||||
done := make(chan struct{})
|
||||
|
||||
@@ -3,6 +3,7 @@ package download
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/url"
|
||||
@@ -327,8 +328,8 @@ func (d *Downloader) triggerOnResolve(req *base.Request) (res *base.Resource, er
|
||||
return
|
||||
}
|
||||
|
||||
func (d *Downloader) triggerOnStart(task *Task) {
|
||||
doTrigger(d,
|
||||
func (d *Downloader) triggerOnStart(task *Task) error {
|
||||
return doTrigger(d,
|
||||
EventOnStart,
|
||||
task.Meta.Req,
|
||||
&OnStartContext{
|
||||
@@ -413,6 +414,13 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
|
||||
}
|
||||
engine, session := d.newExtensionEngine()
|
||||
defer session.CloseIfIdle()
|
||||
eventCtx, cancelEvent := context.WithTimeout(d.backgroundCtx, d.cfg.ExtensionEventTimeout)
|
||||
defer cancelEvent()
|
||||
defer func() {
|
||||
if eventCtx.Err() != nil {
|
||||
session.Close()
|
||||
}
|
||||
}()
|
||||
gopeed.Runtime = &InstanceRuntime{
|
||||
WebView: d.newExtensionWebViewRuntime(session, ext.buildIdentity()),
|
||||
}
|
||||
@@ -421,13 +429,13 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
|
||||
gopeed.Logger.logger.Error().Err(err).Msgf("[%s] engine inject failed", ext.buildIdentity())
|
||||
return
|
||||
}
|
||||
_, err = engine.RunString(string(scriptBuf))
|
||||
_, err = engine.RunStringContext(eventCtx, string(scriptBuf))
|
||||
if err != nil {
|
||||
gopeed.Logger.logger.Error().Err(err).Msgf("[%s] run script failed", ext.buildIdentity())
|
||||
return
|
||||
}
|
||||
if fn, ok := gopeed.Events[event]; ok {
|
||||
_, err = engine.CallFunction(fn, ctx)
|
||||
_, err = engine.CallFunctionContext(eventCtx, fn, ctx)
|
||||
if err != nil {
|
||||
gopeed.Logger.logger.Error().Err(err).Msgf("[%s] call function failed: %s", ext.buildIdentity(), event)
|
||||
return
|
||||
@@ -437,6 +445,9 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
|
||||
}
|
||||
}
|
||||
}()
|
||||
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
package download
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/fetcher"
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
)
|
||||
|
||||
func installCancellationExtension(t *testing.T, d *Downloader, event, script string) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
manifest := fmt.Sprintf(`{"name":"cancel-%s","author":"test","title":"Cancel","version":"1.0.0","scripts":[{"event":%q,"match":{"urls":["*://example.com/*"]},"entry":"index.js"}]}`, event, event)
|
||||
for name, body := range map[string]string{"manifest.json": manifest, "index.js": script} {
|
||||
if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := d.InstallExtensionByFolder(dir, true); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
func newCancellationDownloader(t *testing.T) *Downloader {
|
||||
t.Helper()
|
||||
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{&generationTestManager{holdOpen: true}}})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
func TestResolveExtensionStopsOnShutdown(t *testing.T) {
|
||||
entered, release := make(chan struct{}), make(chan struct{})
|
||||
var once sync.Once
|
||||
defer once.Do(func() { close(release) })
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
close(entered)
|
||||
select {
|
||||
case <-release:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
w.Write([]byte("ok"))
|
||||
}))
|
||||
defer server.Close()
|
||||
d := newCancellationDownloader(t)
|
||||
defer d.Close()
|
||||
installCancellationExtension(t, d, "onResolve", fmt.Sprintf(`gopeed.events.onResolve(async ctx=>{await fetch(%q);});`, server.URL))
|
||||
done := make(chan error, 1)
|
||||
go func() { _, err := d.Resolve(&base.Request{URL: "https://example.com/cancel"}); done <- err }()
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("handler did not enter")
|
||||
}
|
||||
if err := d.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
if !errors.Is(err, context.Canceled) {
|
||||
t.Errorf("got %v, want cancellation", err)
|
||||
}
|
||||
case <-time.After(300 * time.Millisecond):
|
||||
once.Do(func() { close(release) })
|
||||
t.Fatal("shutdown did not cancel extension Resolve")
|
||||
}
|
||||
}
|
||||
func TestOnErrorCancellationReleasesTaskAPIs(t *testing.T) {
|
||||
entered, release := make(chan struct{}), make(chan struct{})
|
||||
var once sync.Once
|
||||
defer once.Do(func() { close(release) })
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
close(entered)
|
||||
select {
|
||||
case <-release:
|
||||
case <-r.Context().Done():
|
||||
}
|
||||
w.Write([]byte("ok"))
|
||||
}))
|
||||
defer server.Close()
|
||||
d := newCancellationDownloader(t)
|
||||
defer d.Close()
|
||||
parent, cancel := context.WithCancel(d.backgroundCtx)
|
||||
d.backgroundCtx = parent
|
||||
defer cancel()
|
||||
installCancellationExtension(t, d, "onError", fmt.Sprintf(`gopeed.events.onError(async ctx=>{await fetch(%q);});`, server.URL))
|
||||
d.cfg.MaxRunning = 0
|
||||
id, err := d.CreateDirect(&base.Request{URL: "generation://example.com/task"}, &base.Options{Path: t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task := d.GetTask(id)
|
||||
task.statusLock.Lock()
|
||||
task.Status = base.DownloadStatusRunning
|
||||
task.statusLock.Unlock()
|
||||
handled := make(chan struct{})
|
||||
go func() { d.doOnError(task, errors.New("source failed")); close(handled) }()
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("onError did not enter")
|
||||
}
|
||||
status, stats := make(chan struct{}), make(chan struct{})
|
||||
go func() { d.RuntimeStatus(id); close(status) }()
|
||||
go func() { d.Stats(id); close(stats) }()
|
||||
cancel()
|
||||
for name, ch := range map[string]<-chan struct{}{"onError": handled, "RuntimeStatus": status, "Stats": stats} {
|
||||
select {
|
||||
case <-ch:
|
||||
case <-time.After(300 * time.Millisecond):
|
||||
once.Do(func() { close(release) })
|
||||
<-handled
|
||||
t.Fatalf("%s stayed blocked after cancellation", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtensionEventDeadlineRejectsUnsettledResolve(t *testing.T) {
|
||||
d := newCancellationDownloader(t)
|
||||
defer d.Close()
|
||||
d.cfg.ExtensionEventTimeout = 30 * time.Millisecond
|
||||
installCancellationExtension(t, d, "onResolve", `gopeed.events.onResolve(async ctx=>{await new Promise(()=>{});});`)
|
||||
_, err := d.Resolve(&base.Request{URL: "https://example.com/timeout"})
|
||||
if !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("got %v, want event deadline exceeded", err)
|
||||
}
|
||||
}
|
||||
func TestExtensionStartDeadlinePreventsFetcherStart(t *testing.T) {
|
||||
d := newCancellationDownloader(t)
|
||||
defer d.Close()
|
||||
d.cfg.ExtensionEventTimeout = 30 * time.Millisecond
|
||||
installCancellationExtension(t, d, "onStart", `gopeed.events.onStart(async ctx=>{await new Promise(()=>{});});`)
|
||||
id, err := d.CreateDirect(&base.Request{URL: "generation://example.com/timeout"}, &base.Options{Path: t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitForTaskStatus(t, d, id, base.DownloadStatusError, 2*time.Second)
|
||||
if starts := d.cfg.FetchManagers[0].(*generationTestManager).starts.Load(); starts != 0 {
|
||||
t.Fatalf("timed out onStart still started fetcher %d times", starts)
|
||||
}
|
||||
}
|
||||
@@ -59,6 +59,21 @@ func (s *engineSession) Release() {
|
||||
}
|
||||
}
|
||||
|
||||
// Close releases retained resources when the owning event is cancelled.
|
||||
func (s *engineSession) Close() {
|
||||
s.mu.Lock()
|
||||
if s.closed {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
s.closed = true
|
||||
s.mu.Unlock()
|
||||
s.runClosers()
|
||||
if s.engine != nil {
|
||||
s.engine.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *engineSession) CloseIfIdle() {
|
||||
s.mu.Lock()
|
||||
shouldClose := s.refs == 0 && !s.closed
|
||||
|
||||
+12
-6
@@ -232,12 +232,15 @@ type DownloaderConfig struct {
|
||||
Controller *controller.Controller
|
||||
FetchManagers []fetcher.FetcherManager
|
||||
|
||||
RefreshInterval int // RefreshInterval time duration to refresh task progress(ms)
|
||||
Storage Storage
|
||||
StorageDir string
|
||||
TempDir string
|
||||
WhiteDownloadDirs []string
|
||||
WebViewProvider enginewebview.Provider
|
||||
// ExtensionEventTimeout bounds one script load and its awaited event handler.
|
||||
// Zero uses one minute. Embedders may increase it for long-running handlers.
|
||||
ExtensionEventTimeout time.Duration
|
||||
RefreshInterval int // RefreshInterval time duration to refresh task progress(ms)
|
||||
Storage Storage
|
||||
StorageDir string
|
||||
TempDir string
|
||||
WhiteDownloadDirs []string
|
||||
WebViewProvider enginewebview.Provider
|
||||
|
||||
ProductionMode bool
|
||||
|
||||
@@ -260,6 +263,9 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
|
||||
new(ed2k.FetcherManager),
|
||||
}
|
||||
}
|
||||
if cfg.ExtensionEventTimeout <= 0 {
|
||||
cfg.ExtensionEventTimeout = time.Minute
|
||||
}
|
||||
if cfg.RefreshInterval == 0 {
|
||||
cfg.RefreshInterval = 350
|
||||
}
|
||||
|
||||
@@ -1,224 +0,0 @@
|
||||
package download
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/GopeedLab/gopeed/internal/fetcher"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/GopeedLab/gopeed/pkg/base"
|
||||
)
|
||||
|
||||
func TestReleaseResolveCancelsPendingHTTPHeaders(t *testing.T) {
|
||||
entered := make(chan struct{})
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { close(entered); <-r.Context().Done() }))
|
||||
defer server.Close()
|
||||
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Close()
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := d.ResolveWithID("pending-http", &base.Request{URL: server.URL + "/file"})
|
||||
done <- err
|
||||
}()
|
||||
select {
|
||||
case <-entered:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("request did not start")
|
||||
}
|
||||
if err := d.ReleaseResolve("pending-http"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
if err == nil {
|
||||
t.Fatal("cancelled Resolve succeeded")
|
||||
}
|
||||
case <-time.After(300 * time.Millisecond):
|
||||
server.CloseClientConnections()
|
||||
t.Fatal("ReleaseResolve did not cancel the request waiting for headers")
|
||||
}
|
||||
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
|
||||
t.Fatal("cancelled operation retained resources")
|
||||
}
|
||||
}
|
||||
func TestReleaseResolveAfterCreateDoesNotCloseTask(t *testing.T) {
|
||||
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: nil})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Close()
|
||||
d.cfg.MaxRunning = 0
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) }))
|
||||
defer server.Close()
|
||||
rr, err := d.Resolve(&base.Request{URL: server.URL})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id, err := d.CreateWithOptions(rr.ID, &base.Options{Path: t.TempDir()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := d.ReleaseResolve(rr.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
task := d.GetTask(id)
|
||||
if task == nil {
|
||||
t.Fatal("late release removed task")
|
||||
}
|
||||
if err := d.doStart(task); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitForTaskStatus(t, d, id, base.DownloadStatusDone, 3*time.Second)
|
||||
}
|
||||
func TestResolveRejectsClosedDownloader(t *testing.T) {
|
||||
d := NewDownloader(&DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir()})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d.Close()
|
||||
if _, err := d.ResolveWithID("closed", &base.Request{URL: "http://127.0.0.1/"}); err == nil {
|
||||
t.Fatalf("closed downloader accepted Resolve: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestReleaseResolveCancelsPendingMagnetMetadata(t *testing.T) {
|
||||
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Close()
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := d.ResolveWithID("pending-magnet", &base.Request{URL: "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567"})
|
||||
done <- err
|
||||
}()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for {
|
||||
d.fetcherMapLock.Lock()
|
||||
state := d.resolveStates["pending-magnet"]
|
||||
ready := state != nil && state.fetcher != nil
|
||||
d.fetcherMapLock.Unlock()
|
||||
if ready {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("magnet resolve did not start")
|
||||
}
|
||||
time.Sleep(time.Millisecond)
|
||||
}
|
||||
if err := d.ReleaseResolve("pending-magnet"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case err := <-done:
|
||||
if err == nil {
|
||||
t.Fatal("cancelled magnet Resolve succeeded")
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("cancelled magnet remained in metadata wait")
|
||||
}
|
||||
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
|
||||
t.Fatal("cancelled magnet retained its resource")
|
||||
}
|
||||
}
|
||||
func TestCloseDrainsResolvedPreviewCache(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) }))
|
||||
defer server.Close()
|
||||
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir()})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := d.Resolve(&base.Request{URL: server.URL}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := d.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
|
||||
t.Fatal("shutdown retained preview resources")
|
||||
}
|
||||
}
|
||||
|
||||
type cancelledCreationFetcher struct {
|
||||
fetcher.Fetcher
|
||||
entered chan struct{}
|
||||
release chan struct{}
|
||||
err error
|
||||
}
|
||||
|
||||
func (f *cancelledCreationFetcher) Patch(*base.Request, *base.Options) error {
|
||||
close(f.entered)
|
||||
<-f.release
|
||||
return f.err
|
||||
}
|
||||
func TestReleaseResolveDuringCreationTransfersOrClosesOwnership(t *testing.T) {
|
||||
for _, fail := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("fail=%v", fail), func(t *testing.T) {
|
||||
d := NewDownloader(&DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir()})
|
||||
if err := d.Setup(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer d.Close()
|
||||
d.cfg.MaxRunning = 0
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) }))
|
||||
defer server.Close()
|
||||
rr, err := d.Resolve(&base.Request{URL: server.URL})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f := &cancelledCreationFetcher{Fetcher: d.fetcherCache[rr.ID], entered: make(chan struct{}), release: make(chan struct{})}
|
||||
if fail {
|
||||
f.err = errors.New("creation failed")
|
||||
}
|
||||
d.fetcherMapLock.Lock()
|
||||
d.fetcherCache[rr.ID] = f
|
||||
d.fetcherMapLock.Unlock()
|
||||
done := make(chan struct{})
|
||||
var id string
|
||||
var createErr error
|
||||
go func() {
|
||||
defer close(done)
|
||||
id, createErr = d.CreateWithOptions(rr.ID, &base.Options{Path: t.TempDir(), Size: 2})
|
||||
}()
|
||||
select {
|
||||
case <-f.entered:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("creation did not claim resource")
|
||||
}
|
||||
if err := d.ReleaseResolve(rr.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
close(f.release)
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("creation did not return")
|
||||
}
|
||||
if len(d.fetcherCache) != 0 || len(d.resolveStates) != 0 {
|
||||
t.Fatal("cancelled creation returned ownership to preview cache")
|
||||
}
|
||||
if fail {
|
||||
if createErr == nil {
|
||||
t.Fatal("expected creation failure")
|
||||
}
|
||||
return
|
||||
}
|
||||
if createErr != nil {
|
||||
t.Fatal(createErr)
|
||||
}
|
||||
// The protocol manager persists its concrete Fetcher type.
|
||||
task := d.GetTask(id)
|
||||
task.fetcher = f.Fetcher
|
||||
if err := d.doStart(task); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitForTaskStatus(t, d, id, base.DownloadStatusDone, 3*time.Second)
|
||||
})
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -24,7 +24,7 @@ func Info(w http.ResponseWriter, r *http.Request) {
|
||||
func Resolve(w http.ResponseWriter, r *http.Request) {
|
||||
var req model.ResolveTask
|
||||
if ReadJson(r, w, &req) {
|
||||
rr, err := Downloader.ResolveWithID(req.ID, req.Req)
|
||||
rr, err := Downloader.Resolve(req.Req)
|
||||
if err != nil {
|
||||
WriteJson(w, model.NewErrorResult(err.Error()))
|
||||
return
|
||||
|
||||
@@ -3,7 +3,6 @@ package model
|
||||
import "github.com/GopeedLab/gopeed/pkg/base"
|
||||
|
||||
type ResolveTask struct {
|
||||
ID string `json:"id,omitempty"`
|
||||
Req *base.Request `json:"req"`
|
||||
}
|
||||
|
||||
|
||||
@@ -81,9 +81,6 @@ Future<ResolveResult> resolve(ResolveTask resolveTask) {
|
||||
);
|
||||
}
|
||||
|
||||
Future<void> releaseResolve(String id) =>
|
||||
_parse<void>(() => _request('api/v1/resolve/${Uri.encodeComponent(id)}', method: 'DELETE'), null);
|
||||
|
||||
Future<String> createTask(CreateTask createTask) {
|
||||
return _parse<String>(() => _request('api/v1/tasks', method: 'POST', data: createTask), (data) => data as String);
|
||||
}
|
||||
|
||||
@@ -8,10 +8,7 @@ part 'resolve_task.g.dart';
|
||||
class ResolveTask {
|
||||
Request? req;
|
||||
|
||||
@JsonKey(includeIfNull: false)
|
||||
String? id;
|
||||
|
||||
ResolveTask({this.req, this.id});
|
||||
ResolveTask({this.req});
|
||||
|
||||
factory ResolveTask.fromJson(Map<String, dynamic> json) => _$ResolveTaskFromJson(json);
|
||||
Map<String, dynamic> toJson() => _$ResolveTaskToJson(this);
|
||||
|
||||
@@ -10,8 +10,7 @@ ResolveTask _$ResolveTaskFromJson(Map<String, dynamic> json) => ResolveTask(
|
||||
req: json['req'] == null
|
||||
? null
|
||||
: Request.fromJson(json['req'] as Map<String, dynamic>),
|
||||
id: json['id'] as String?,
|
||||
);
|
||||
|
||||
Map<String, dynamic> _$ResolveTaskToJson(ResolveTask instance) =>
|
||||
<String, dynamic>{'req': ?instance.req, 'id': ?instance.id};
|
||||
<String, dynamic>{'req': ?instance.req};
|
||||
@@ -93,10 +93,6 @@ void _bindGopeed(CapabilityRegistry registry, _ConfigStore configStore) {
|
||||
(p) => api.getMirrorCandidates(p['provider'] as String, p['url'] as String),
|
||||
)
|
||||
..bind(GopeedMethods.resolve, api.resolve)
|
||||
..bind(GopeedMethods.releaseResolve, (id) async {
|
||||
await api.releaseResolve(id);
|
||||
return const RpcUnit();
|
||||
})
|
||||
..bind(GopeedMethods.createTask, api.createTask)
|
||||
..bind(GopeedMethods.createTaskBatch, api.createTaskBatch)
|
||||
..bind(GopeedMethods.patchTask, (params) async {
|
||||
|
||||
@@ -16,7 +16,6 @@ abstract final class GopeedMethods {
|
||||
static const triggerProtocolJob = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.protocol.job.trigger');
|
||||
static const getMirrorCandidates = RpcMethod<Map<String, dynamic>, List<String>>('gopeed.mirrors.candidates');
|
||||
static const resolve = RpcMethod<ResolveTask, ResolveResult>('gopeed.resolve');
|
||||
static const releaseResolve = RpcMethod<String, RpcUnit>('gopeed.releaseResolve');
|
||||
static const createTask = RpcMethod<CreateTask, String>('gopeed.task.create');
|
||||
static const createTaskBatch = RpcMethod<CreateTaskBatch, CreateTaskBatchResult>('gopeed.task.createBatch');
|
||||
static const patchTask = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.task.patch');
|
||||
@@ -51,7 +50,6 @@ class GopeedService {
|
||||
Future<List<String>> getMirrorCandidates(String provider, String url) =>
|
||||
_invoker.invoke(GopeedMethods.getMirrorCandidates, {'provider': provider, 'url': url});
|
||||
Future<ResolveResult> resolve(ResolveTask request) => _invoker.invoke(GopeedMethods.resolve, request);
|
||||
Future<void> releaseResolve(String id) async => _invoker.invoke(GopeedMethods.releaseResolve, id);
|
||||
|
||||
Future<String> createTask(CreateTask request) => _invoker.invoke(GopeedMethods.createTask, request);
|
||||
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
import 'dart:developer' as developer;
|
||||
import 'package:uuid/uuid.dart';
|
||||
import 'package:path/path.dart' as path;
|
||||
|
||||
import '../../../api/model/create_task_batch.dart';
|
||||
@@ -18,8 +16,6 @@ class PendingDownload {
|
||||
final Request request;
|
||||
final Options options;
|
||||
ResolveResult? result;
|
||||
String? _resolveID;
|
||||
bool _cancelled = false;
|
||||
String? resolveError;
|
||||
String? createError;
|
||||
final Set<int> selected = {};
|
||||
@@ -54,16 +50,11 @@ class PendingDownload {
|
||||
bool get hasSelection => !created && result != null && selected.isNotEmpty;
|
||||
|
||||
Future<void> resolve(GopeedService service) async {
|
||||
if (_cancelled || result != null || resolving) return;
|
||||
if (result != null || resolving) return;
|
||||
resolving = true;
|
||||
resolveError = null;
|
||||
try {
|
||||
_resolveID = const Uuid().v4();
|
||||
result = await service.resolve(ResolveTask(req: request, id: _resolveID));
|
||||
if (_cancelled) {
|
||||
await _release(service, result!.id);
|
||||
return;
|
||||
}
|
||||
result = await service.resolve(ResolveTask(req: request));
|
||||
selected.addAll(List.generate(files.isEmpty ? 1 : files.length, (index) => index));
|
||||
} catch (error) {
|
||||
resolveError = error.toString();
|
||||
@@ -72,23 +63,6 @@ class PendingDownload {
|
||||
}
|
||||
}
|
||||
|
||||
/// Stop an outstanding preview and release any resource it already owns.
|
||||
/// A response arriving after cancellation is released by resolve() as well.
|
||||
Future<void> cancel(GopeedService service) async {
|
||||
if (_cancelled || created) return;
|
||||
_cancelled = true;
|
||||
await _release(service, result?.id.isNotEmpty == true ? result!.id : _resolveID);
|
||||
}
|
||||
|
||||
Future<void> _release(GopeedService service, String? id) async {
|
||||
if (id == null || id.isEmpty) return;
|
||||
try {
|
||||
await service.releaseResolve(id);
|
||||
} catch (error, stack) {
|
||||
developer.log('Failed to release resolved resource', name: 'gopeed.resolve', error: error, stackTrace: stack);
|
||||
}
|
||||
}
|
||||
|
||||
List<CreateTaskBatchItem> creationItems() {
|
||||
final indexes = selected.toList()..sort();
|
||||
if (result!.id.isNotEmpty || files.isEmpty) {
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
import 'dart:async';
|
||||
|
||||
import 'package:shadcn_flutter/shadcn_flutter.dart';
|
||||
import 'package:flutter/widgets.dart' as flutter show Flexible;
|
||||
|
||||
@@ -51,7 +49,6 @@ class _ResolveDownloadsDialogState extends State<_ResolveDownloadsDialog> {
|
||||
@override
|
||||
void dispose() {
|
||||
_cancelled = true;
|
||||
_releasePreviews();
|
||||
super.dispose();
|
||||
}
|
||||
|
||||
@@ -67,15 +64,8 @@ class _ResolveDownloadsDialogState extends State<_ResolveDownloadsDialog> {
|
||||
}
|
||||
}
|
||||
|
||||
void _releasePreviews() {
|
||||
for (final source in widget.sources) {
|
||||
unawaited(source.cancel(widget.service));
|
||||
}
|
||||
}
|
||||
|
||||
void _cancel() {
|
||||
_cancelled = true;
|
||||
_releasePreviews();
|
||||
closeOverlay(context, false);
|
||||
}
|
||||
|
||||
|
||||
@@ -64,81 +64,6 @@ Future<void> open(
|
||||
}
|
||||
|
||||
void main() {
|
||||
testWidgets('cancel releases a ready preview resource', (tester) async {
|
||||
final released = <String>[];
|
||||
final registry = CapabilityRegistry(createAppCapabilityCodecs())
|
||||
..bind(GopeedMethods.resolve, (_) => resolved('ready-rid'))
|
||||
..bind(const RpcMethod<String, RpcUnit>('gopeed.releaseResolve'), (id) {
|
||||
released.add(id);
|
||||
return const RpcUnit();
|
||||
});
|
||||
await open(tester, [pending('torrent')], registry);
|
||||
await tester.pumpAndSettle();
|
||||
await tester.tap(find.byKey(const ValueKey('resolve-cancel-button')));
|
||||
await tester.pumpAndSettle();
|
||||
expect(released, contains('ready-rid'));
|
||||
});
|
||||
|
||||
testWidgets('cancel releases a late preview result', (tester) async {
|
||||
final response = Completer<ResolveResult>();
|
||||
final released = <String>[];
|
||||
final registry = CapabilityRegistry(createAppCapabilityCodecs())
|
||||
..bind(GopeedMethods.resolve, (_) => response.future)
|
||||
..bind(const RpcMethod<String, RpcUnit>('gopeed.releaseResolve'), (id) {
|
||||
released.add(id);
|
||||
return const RpcUnit();
|
||||
});
|
||||
await open(tester, [pending('torrent')], registry);
|
||||
await tester.tap(find.byKey(const ValueKey('resolve-close-button')));
|
||||
await tester.pumpAndSettle();
|
||||
response.complete(resolved('late-rid'));
|
||||
await tester.pumpAndSettle();
|
||||
expect(released, contains('late-rid'));
|
||||
});
|
||||
|
||||
testWidgets('cancel sends the request ID before a preview completes', (tester) async {
|
||||
final response = Completer<ResolveResult>();
|
||||
final released = <String>[];
|
||||
String? requestedID;
|
||||
final registry = CapabilityRegistry(createAppCapabilityCodecs())
|
||||
..bind(GopeedMethods.resolve, (request) {
|
||||
requestedID = request.id;
|
||||
return response.future;
|
||||
})
|
||||
..bind(GopeedMethods.releaseResolve, (id) {
|
||||
released.add(id);
|
||||
return const RpcUnit();
|
||||
});
|
||||
await open(tester, [pending('torrent')], registry);
|
||||
await tester.tap(find.byKey(const ValueKey('resolve-close-button')));
|
||||
await tester.pumpAndSettle();
|
||||
expect(requestedID, isNotEmpty);
|
||||
expect(released, contains(requestedID));
|
||||
response.complete(
|
||||
ResolveResult(
|
||||
id: requestedID!,
|
||||
res: Resource(files: [FileInfo(name: 'late.txt')]),
|
||||
),
|
||||
);
|
||||
await tester.pumpAndSettle();
|
||||
expect(released.where((id) => id == requestedID), hasLength(2));
|
||||
});
|
||||
|
||||
testWidgets('successful creation keeps task-owned resources', (tester) async {
|
||||
final released = <String>[];
|
||||
final registry = CapabilityRegistry(createAppCapabilityCodecs())
|
||||
..bind(GopeedMethods.resolve, (_) => resolved('task-owned'))
|
||||
..bind(GopeedMethods.createTaskBatch, (_) => CreateTaskBatchResult(taskIds: ['task']))
|
||||
..bind(GopeedMethods.releaseResolve, (id) {
|
||||
released.add(id);
|
||||
return const RpcUnit();
|
||||
});
|
||||
await open(tester, [pending('torrent')], registry);
|
||||
await tester.pumpAndSettle();
|
||||
await tester.tap(find.byKey(const ValueKey('resolve-create-button')));
|
||||
await tester.pumpAndSettle();
|
||||
expect(released, isEmpty);
|
||||
});
|
||||
for (final width in [390.0, 700.0, 1024.0]) {
|
||||
testWidgets('file list fits and exposes ready selections at width $width', (tester) async {
|
||||
final registry = CapabilityRegistry(createAppCapabilityCodecs())
|
||||
|
||||
Reference in new issue
Block a user