mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-15 04:43:41 +00:00
refactor(resolve): use Fetcher.Close for preview cancellation
This commit is contained in:
1 parent
8161042b93
commit
882f0cbc84
3 files changed
+4
-71
No files matched your search
@@ -563,16 +563,9 @@ 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
|
||||
@@ -602,9 +595,7 @@ func (d *Downloader) ResolveWithID(id string, req *base.Request) (rr *ResolveRes
|
||||
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}
|
||||
state := &resolveState{}
|
||||
d.fetcherMapLock.Lock()
|
||||
if d.closed.Load() {
|
||||
d.fetcherMapLock.Unlock()
|
||||
@@ -645,11 +636,7 @@ func (d *Downloader) ResolveWithID(id string, req *base.Request) (rr *ResolveRes
|
||||
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)
|
||||
}
|
||||
err = f.Resolve(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -658,13 +645,13 @@ func (d *Downloader) ResolveWithID(id string, req *base.Request) (rr *ResolveRes
|
||||
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.
|
||||
// Pending resolution is interrupted according to the Fetcher's Close semantics.
|
||||
func (d *Downloader) ReleaseResolve(id string) error {
|
||||
d.fetcherMapLock.Lock()
|
||||
state := d.resolveStates[id]
|
||||
@@ -673,17 +660,10 @@ func (d *Downloader) ReleaseResolve(id string) error {
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
d.fetcherMapLock.Unlock()
|
||||
|
||||
@@ -3,50 +3,15 @@ package download
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/GopeedLab/gopeed/internal/fetcher"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/GopeedLab/gopeed/internal/fetcher"
|
||||
"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 {
|
||||
|
||||
Reference in new issue
Block a user