mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-11 02:50:27 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8161042b93 | ||
|
|
763731eaaa |
No files matched your search
@@ -46,10 +46,6 @@ type Uploader interface {
|
|||||||
WaitUpload() error
|
WaitUpload() error
|
||||||
}
|
}
|
||||||
|
|
||||||
type DownloadCounter interface {
|
|
||||||
ReceivedBytes() int64
|
|
||||||
}
|
|
||||||
|
|
||||||
// FetcherMeta defines the meta information of a fetcher.
|
// FetcherMeta defines the meta information of a fetcher.
|
||||||
type FetcherMeta struct {
|
type FetcherMeta struct {
|
||||||
Req *base.Request `json:"req"`
|
Req *base.Request `json:"req"`
|
||||||
|
|||||||
@@ -290,82 +290,6 @@ func TestBTDownloadSelectedFiles(t *testing.T) {
|
|||||||
if f.Meta().Res.Size != 48*1024 {
|
if f.Meta().Res.Size != 48*1024 {
|
||||||
t.Fatal("selected resource size is incorrect")
|
t.Fatal("selected resource size is incorrect")
|
||||||
}
|
}
|
||||||
if downloaded := f.Progress().TotalDownloaded(); downloaded != 48*1024 || f.ReceivedBytes() <= downloaded {
|
|
||||||
t.Fatalf("selected progress = %d, received traffic = %d; want 48 KiB of progress and traffic across the shared piece", downloaded, f.ReceivedBytes())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBTProgressOnlyCountsVerifiedFileBytes(t *testing.T) {
|
|
||||||
const fileSize = 56 * 1024
|
|
||||||
fixture := makeBTFixtureWithPieceSize(t, map[string][]byte{
|
|
||||||
"a.bin": btPayload(fileSize, "verified-a"),
|
|
||||||
"b.bin": btPayload(fileSize, "verified-b"),
|
|
||||||
}, 16*1024, false, true, 64*1024)
|
|
||||||
f := btDownloadFetcher(t, nil)
|
|
||||||
btResolve(t, f, fixture.Magnet)
|
|
||||||
dir := t.TempDir()
|
|
||||||
f.Meta().Opts = &base.Options{Path: dir}
|
|
||||||
btStart(t, f)
|
|
||||||
deadline := time.Now().Add(8 * time.Second)
|
|
||||||
for f.torrent.Files()[0].BytesCompleted() == 0 && time.Now().Before(deadline) {
|
|
||||||
time.Sleep(10 * time.Millisecond)
|
|
||||||
}
|
|
||||||
btPause(t, f)
|
|
||||||
if f.torrent.Files()[0].BytesCompleted() == 0 || f.torrent.Piece(0).State().Complete {
|
|
||||||
t.Fatal("expected some received bytes in a piece that is still incomplete")
|
|
||||||
}
|
|
||||||
if f.ReceivedBytes() == 0 {
|
|
||||||
t.Fatal("received traffic must be visible before a piece is verified")
|
|
||||||
}
|
|
||||||
if progress := f.Progress(); len(progress) != 2 || progress[0] != 0 {
|
|
||||||
t.Errorf("progress before the first file's piece is verified = %v, want zero for the first file", progress)
|
|
||||||
}
|
|
||||||
fixture.Limiter.SetLimit(rate.Inf)
|
|
||||||
btStart(t, f)
|
|
||||||
btAwaitDone(t, f)
|
|
||||||
fixture.AssertFiles(t, dir)
|
|
||||||
progress := f.Progress()
|
|
||||||
if len(progress) != 2 || progress[0] != fileSize || progress[1] != fileSize {
|
|
||||||
t.Fatalf("verified progress at completion = %v, want %d for each file", progress, fileSize)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestBTUploadedBytesSamplesLiveTraffic(t *testing.T) {
|
|
||||||
payload := btPayload(48*1024, "upload-counter")
|
|
||||||
fixture := newBTFixture(t, map[string][]byte{"": payload}, 0)
|
|
||||||
const saved = 1024
|
|
||||||
f := &Fetcher{
|
|
||||||
torrent: fixture.Torrent,
|
|
||||||
data: &fetcherData{SeedBytes: saved},
|
|
||||||
seedBaseBytes: saved,
|
|
||||||
}
|
|
||||||
f.torrentReady.Store(true)
|
|
||||||
f.torrentUpload.Store(true)
|
|
||||||
|
|
||||||
cfg := torrent.TestingConfig(t)
|
|
||||||
cfg.DataDir = t.TempDir()
|
|
||||||
cfg.DisableTrackers = true
|
|
||||||
cfg.DisableUTP = true
|
|
||||||
cfg.DisableIPv6 = true
|
|
||||||
peer, err := torrent.NewClient(cfg)
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
t.Cleanup(func() { peer.Close() })
|
|
||||||
download, _, err := peer.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(fixture.MetaInfo))
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
download.AddPeers([]torrent.PeerInfo{{Addr: fixture.Seeder.ListenAddrs()[0]}})
|
|
||||||
download.DownloadAll()
|
|
||||||
select {
|
|
||||||
case <-download.Complete().On():
|
|
||||||
case <-time.After(10 * time.Second):
|
|
||||||
t.Fatal("loopback peer did not complete")
|
|
||||||
}
|
|
||||||
if got, want := f.UploadedBytes(), int64(saved+len(payload)); got != want {
|
|
||||||
t.Fatalf("uploaded bytes = %d, want %d including the saved bytes", got, want)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestBTPauseAndResume(t *testing.T) {
|
func TestBTPauseAndResume(t *testing.T) {
|
||||||
|
|||||||
@@ -10,7 +10,6 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"runtime"
|
"runtime"
|
||||||
"sort"
|
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
@@ -43,10 +42,9 @@ type Fetcher struct {
|
|||||||
ctl *controller.Controller
|
ctl *controller.Controller
|
||||||
config *config
|
config *config
|
||||||
|
|
||||||
torrent *torrent.Torrent
|
torrent *torrent.Torrent
|
||||||
meta *fetcher.FetcherMeta
|
meta *fetcher.FetcherMeta
|
||||||
data *fetcherData
|
data *fetcherData
|
||||||
seedBaseBytes int64
|
|
||||||
|
|
||||||
torrentReady atomic.Bool
|
torrentReady atomic.Bool
|
||||||
torrentUpload atomic.Bool
|
torrentUpload atomic.Bool
|
||||||
@@ -465,46 +463,14 @@ func (f *Fetcher) Progress() fetcher.Progress {
|
|||||||
if !f.torrentReady.Load() {
|
if !f.torrentReady.Load() {
|
||||||
return append(fetcher.Progress(nil), f.data.Progress...)
|
return append(fetcher.Progress(nil), f.data.Progress...)
|
||||||
}
|
}
|
||||||
pieceSize := f.torrent.Info().PieceLength
|
|
||||||
var offset int64
|
|
||||||
var verified []verifiedByteRange
|
|
||||||
for _, run := range f.torrent.PieceStateRuns() {
|
|
||||||
end := offset + int64(run.Length)*pieceSize
|
|
||||||
if pieceVerified(run.PieceState) {
|
|
||||||
verified = append(verified, verifiedByteRange{start: offset, end: end})
|
|
||||||
}
|
|
||||||
offset = end
|
|
||||||
}
|
|
||||||
files := f.torrent.Files()
|
|
||||||
for i := range f.data.Progress {
|
for i := range f.data.Progress {
|
||||||
selectIndex := f.meta.Opts.SelectFiles[i]
|
selectIndex := f.meta.Opts.SelectFiles[i]
|
||||||
file := files[selectIndex]
|
file := f.torrent.Files()[selectIndex]
|
||||||
f.data.Progress[i] = verifiedFileBytes(verified, file.Offset(), file.Length())
|
f.data.Progress[i] = file.BytesCompleted()
|
||||||
}
|
}
|
||||||
return append(fetcher.Progress(nil), f.data.Progress...)
|
return append(fetcher.Progress(nil), f.data.Progress...)
|
||||||
}
|
}
|
||||||
|
|
||||||
type verifiedByteRange struct {
|
|
||||||
start, end int64
|
|
||||||
}
|
|
||||||
|
|
||||||
func verifiedFileBytes(verified []verifiedByteRange, offset, length int64) int64 {
|
|
||||||
first := sort.Search(len(verified), func(i int) bool { return verified[i].end > offset })
|
|
||||||
fileEnd := offset + length
|
|
||||||
var completed int64
|
|
||||||
for _, run := range verified[first:] {
|
|
||||||
if run.start >= fileEnd {
|
|
||||||
break
|
|
||||||
}
|
|
||||||
completed += min(fileEnd, run.end) - max(offset, run.start)
|
|
||||||
}
|
|
||||||
return completed
|
|
||||||
}
|
|
||||||
|
|
||||||
func pieceVerified(state torrent.PieceState) bool {
|
|
||||||
return state.Ok && state.Complete && !state.Hashing && !state.QueuedForHash
|
|
||||||
}
|
|
||||||
|
|
||||||
func (f *Fetcher) Wait() error {
|
func (f *Fetcher) Wait() error {
|
||||||
f.mu.Lock()
|
f.mu.Lock()
|
||||||
var closed <-chan struct{}
|
var closed <-chan struct{}
|
||||||
@@ -621,7 +587,7 @@ func (f *Fetcher) isDoneLocked() bool {
|
|||||||
file := f.torrent.Files()[index]
|
file := f.torrent.Files()[index]
|
||||||
for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ {
|
for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ {
|
||||||
state := f.torrent.Piece(i).State()
|
state := f.torrent.Piece(i).State()
|
||||||
if !pieceVerified(state) {
|
if !state.Ok || !state.Complete || state.Hashing || state.QueuedForHash {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -719,13 +685,11 @@ func (f *Fetcher) Upload() (err error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (f *Fetcher) doUpload(fromUpload bool) {
|
func (f *Fetcher) doUpload(fromUpload bool) {
|
||||||
f.mu.Lock()
|
|
||||||
if !f.torrentUpload.CompareAndSwap(false, true) {
|
if !f.torrentUpload.CompareAndSwap(false, true) {
|
||||||
f.mu.Unlock()
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
f.seedBaseBytes = f.data.SeedBytes
|
f.mu.Lock()
|
||||||
initialTime := f.data.SeedTime
|
initialBytes, initialTime := f.data.SeedBytes, f.data.SeedTime
|
||||||
torrentClosed := f.torrent.Closed()
|
torrentClosed := f.torrent.Closed()
|
||||||
f.mu.Unlock()
|
f.mu.Unlock()
|
||||||
var doneTime int64
|
var doneTime int64
|
||||||
@@ -743,7 +707,8 @@ func (f *Fetcher) doUpload(fromUpload bool) {
|
|||||||
f.mu.Unlock()
|
f.mu.Unlock()
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
f.refreshSeedBytesLocked()
|
stats := f.torrentStats()
|
||||||
|
f.data.SeedBytes = initialBytes + stats.BytesWrittenData.Int64()
|
||||||
if !fromUpload && !f.isDoneLocked() {
|
if !fromUpload && !f.isDoneLocked() {
|
||||||
f.mu.Unlock()
|
f.mu.Unlock()
|
||||||
continue
|
continue
|
||||||
@@ -776,27 +741,9 @@ func (f *Fetcher) torrentStats() torrent.TorrentStats {
|
|||||||
func (f *Fetcher) UploadedBytes() int64 {
|
func (f *Fetcher) UploadedBytes() int64 {
|
||||||
f.mu.Lock()
|
f.mu.Lock()
|
||||||
defer f.mu.Unlock()
|
defer f.mu.Unlock()
|
||||||
if f.torrentReady.Load() && f.torrentUpload.Load() {
|
|
||||||
f.refreshSeedBytesLocked()
|
|
||||||
}
|
|
||||||
return f.data.SeedBytes
|
return f.data.SeedBytes
|
||||||
}
|
}
|
||||||
|
|
||||||
func (f *Fetcher) refreshSeedBytesLocked() {
|
|
||||||
stats := f.torrentStats()
|
|
||||||
f.data.SeedBytes = max(f.data.SeedBytes, f.seedBaseBytes+stats.BytesWrittenData.Int64())
|
|
||||||
}
|
|
||||||
|
|
||||||
func (f *Fetcher) ReceivedBytes() int64 {
|
|
||||||
f.mu.Lock()
|
|
||||||
defer f.mu.Unlock()
|
|
||||||
if !f.torrentReady.Load() {
|
|
||||||
return 0
|
|
||||||
}
|
|
||||||
stats := f.torrentStats()
|
|
||||||
return stats.BytesReadUsefulData.Int64()
|
|
||||||
}
|
|
||||||
|
|
||||||
func (f *Fetcher) WaitUpload() (err error) {
|
func (f *Fetcher) WaitUpload() (err error) {
|
||||||
<-f.uploadDoneCh
|
<-f.uploadDoneCh
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -46,11 +46,6 @@ func newBTHybridFixture(t testing.TB, payload []byte) *btFixture {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool) *btFixture {
|
func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool) *btFixture {
|
||||||
t.Helper()
|
|
||||||
return makeBTFixtureWithPieceSize(t, files, bytesPerSecond, v2, private, 16*1024)
|
|
||||||
}
|
|
||||||
|
|
||||||
func makeBTFixtureWithPieceSize(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool, pieceSize int64) *btFixture {
|
|
||||||
t.Helper()
|
t.Helper()
|
||||||
source := t.TempDir()
|
source := t.TempDir()
|
||||||
name := "payload.bin"
|
name := "payload.bin"
|
||||||
@@ -69,7 +64,7 @@ func makeBTFixtureWithPieceSize(t testing.TB, files map[string][]byte, bytesPerS
|
|||||||
}
|
}
|
||||||
expected[filepath.Join(name, filepath.FromSlash(path))] = append([]byte(nil), data...)
|
expected[filepath.Join(name, filepath.FromSlash(path))] = append([]byte(nil), data...)
|
||||||
}
|
}
|
||||||
info := metainfo.Info{PieceLength: pieceSize}
|
info := metainfo.Info{PieceLength: 16 * 1024}
|
||||||
if err := info.BuildFromFilePath(root); err != nil {
|
if err := info.BuildFromFilePath(root); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -418,6 +418,16 @@ func (f *Fetcher) updateMaxConnTime(d time.Duration) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (f *Fetcher) Resolve(req *base.Request) error {
|
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 {
|
if err := base.ParseReqExtra[fhttp.ReqExtra](req); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
@@ -429,6 +439,8 @@ func (f *Fetcher) Resolve(req *base.Request) error {
|
|||||||
// This allows the response to be reused for downloading (important for one-time URLs)
|
// 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.
|
// The body outlives Resolve, so the read timeout gets its own cancel func.
|
||||||
reqCtx, reqCancel := context.WithCancel(context.Background())
|
reqCtx, reqCancel := context.WithCancel(context.Background())
|
||||||
|
stopCancel := context.AfterFunc(ctx, reqCancel)
|
||||||
|
defer stopCancel()
|
||||||
httpReq, err := f.buildRequest(reqCtx, req)
|
httpReq, err := f.buildRequest(reqCtx, req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
reqCancel()
|
reqCancel()
|
||||||
|
|||||||
@@ -0,0 +1,36 @@
|
|||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
+9
-1
@@ -51,6 +51,7 @@ func NewService(downloader *download.Downloader) (*Service, error) {
|
|||||||
service.routes = []RouteSpec{
|
service.routes = []RouteSpec{
|
||||||
newRoute(http.MethodGet, "/api/v1/info", service.info),
|
newRoute(http.MethodGet, "/api/v1/info", service.info),
|
||||||
newRoute(http.MethodPost, "/api/v1/resolve", service.resolve),
|
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", service.createTask),
|
||||||
newRoute(http.MethodPost, "/api/v1/tasks/batch", service.createTaskBatch),
|
newRoute(http.MethodPost, "/api/v1/tasks/batch", service.createTaskBatch),
|
||||||
newRoute(http.MethodPatch, "/api/v1/tasks/{id}", service.patchTask),
|
newRoute(http.MethodPatch, "/api/v1/tasks/{id}", service.patchTask),
|
||||||
@@ -143,13 +144,20 @@ func (s *Service) resolve(ctx *Context) *Response {
|
|||||||
if response := decodeRequest(ctx.Body, &request); response != nil {
|
if response := decodeRequest(ctx.Body, &request); response != nil {
|
||||||
return response
|
return response
|
||||||
}
|
}
|
||||||
result, err := s.Downloader.Resolve(request.Req)
|
result, err := s.Downloader.ResolveWithID(request.ID, request.Req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return newErrorResponse(err.Error())
|
return newErrorResponse(err.Error())
|
||||||
}
|
}
|
||||||
return newOKResponse(result)
|
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 {
|
func (s *Service) createTask(ctx *Context) *Response {
|
||||||
var request restmodel.CreateTask
|
var request restmodel.CreateTask
|
||||||
if response := decodeRequest(ctx.Body, &request); response != nil {
|
if response := decodeRequest(ctx.Body, &request); response != nil {
|
||||||
|
|||||||
+169
-47
@@ -116,6 +116,7 @@ type Downloader struct {
|
|||||||
|
|
||||||
cfg *DownloaderConfig
|
cfg *DownloaderConfig
|
||||||
fetcherCache map[string]fetcher.Fetcher
|
fetcherCache map[string]fetcher.Fetcher
|
||||||
|
resolveStates map[string]*resolveState
|
||||||
storage Storage
|
storage Storage
|
||||||
tasks []*Task
|
tasks []*Task
|
||||||
waitTasks []*Task
|
waitTasks []*Task
|
||||||
@@ -153,6 +154,7 @@ func NewDownloader(cfg *DownloaderConfig) *Downloader {
|
|||||||
cancelBackground: cancelBackground,
|
cancelBackground: cancelBackground,
|
||||||
cfg: cfg,
|
cfg: cfg,
|
||||||
fetcherCache: make(map[string]fetcher.Fetcher),
|
fetcherCache: make(map[string]fetcher.Fetcher),
|
||||||
|
resolveStates: make(map[string]*resolveState),
|
||||||
waitTasks: make([]*Task, 0),
|
waitTasks: make([]*Task, 0),
|
||||||
groups: make(map[string]*TaskGroup),
|
groups: make(map[string]*TaskGroup),
|
||||||
creatingGroups: make(map[string]struct{}),
|
creatingGroups: make(map[string]struct{}),
|
||||||
@@ -345,12 +347,7 @@ func (d *Downloader) Setup() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
current := task.fetcher.Progress().TotalDownloaded()
|
current := task.fetcher.Progress().TotalDownloaded()
|
||||||
delta := max(0, current-task.Progress.Downloaded)
|
delta := current - task.Progress.Downloaded
|
||||||
if counter, ok := task.fetcher.(fetcher.DownloadCounter); ok && task.Status == base.DownloadStatusRunning {
|
|
||||||
received := counter.ReceivedBytes()
|
|
||||||
delta = received - task.receivedBytes
|
|
||||||
task.receivedBytes = received
|
|
||||||
}
|
|
||||||
if task.Meta != nil && task.Meta.Req != nil {
|
if task.Meta != nil && task.Meta.Req != nil {
|
||||||
if producer := d.blob.Production(task.Meta.Req.URL); producer != nil {
|
if producer := d.blob.Production(task.Meta.Req.URL); producer != nil {
|
||||||
downloaded, received := producer.Progress()
|
downloaded, received := producer.Progress()
|
||||||
@@ -358,20 +355,19 @@ func (d *Downloader) Setup() error {
|
|||||||
task.producer = producer
|
task.producer = producer
|
||||||
task.producerReceived = 0
|
task.producerReceived = 0
|
||||||
task.speedArr = nil
|
task.speedArr = nil
|
||||||
task.downloadSampledAt = time.Time{}
|
|
||||||
}
|
}
|
||||||
current = downloaded
|
current = downloaded
|
||||||
delta = max(0, received-task.producerReceived)
|
delta = max(0, received-task.producerReceived)
|
||||||
task.producerReceived = received
|
task.producerReceived = received
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
tick := float64(d.cfg.RefreshInterval) / 1000
|
||||||
downloadDataChanged := false
|
downloadDataChanged := false
|
||||||
uploading := task.Uploading
|
uploading := task.Uploading
|
||||||
if task.Status == base.DownloadStatusRunning {
|
if task.Status == base.DownloadStatusRunning {
|
||||||
downloadDataChanged = current != task.Progress.Downloaded
|
downloadDataChanged = current != task.Progress.Downloaded
|
||||||
task.Progress.Used = task.timer.Used()
|
task.Progress.Used = task.timer.Used()
|
||||||
elapsed := sampleSeconds(&task.downloadSampledAt, time.Now())
|
task.Progress.Speed = task.updateSpeed(max(0, delta), tick)
|
||||||
task.Progress.Speed = task.updateSpeed(delta, elapsed)
|
|
||||||
task.Progress.Downloaded = current
|
task.Progress.Downloaded = current
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -380,8 +376,7 @@ func (d *Downloader) Setup() error {
|
|||||||
uploader := task.fetcher.(fetcher.Uploader)
|
uploader := task.fetcher.(fetcher.Uploader)
|
||||||
currentUploaded := uploader.UploadedBytes()
|
currentUploaded := uploader.UploadedBytes()
|
||||||
uploadDataChanged = currentUploaded != task.Progress.Uploaded
|
uploadDataChanged = currentUploaded != task.Progress.Uploaded
|
||||||
elapsed := sampleSeconds(&task.uploadSampledAt, time.Now())
|
task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, tick)
|
||||||
task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, elapsed)
|
|
||||||
task.Progress.Uploaded = currentUploaded
|
task.Progress.Uploaded = currentUploaded
|
||||||
}
|
}
|
||||||
task.statusLock.Unlock()
|
task.statusLock.Unlock()
|
||||||
@@ -566,42 +561,156 @@ func ensureResourceRequestRawURLs(parentReq *base.Request, res *base.Resource) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Downloader) Resolve(req *base.Request) (rr *ResolveResult, err error) {
|
// resolveState owns a Fetcher until Create transfers it to a task.
|
||||||
rrId, err := gonanoid.New()
|
// State transitions are protected by fetcherMapLock; closeOnce serializes cleanup.
|
||||||
if err != nil {
|
type contextResolver interface {
|
||||||
return
|
ResolveContext(context.Context, *base.Request) error
|
||||||
}
|
}
|
||||||
ensureRequestRawURL(req)
|
|
||||||
|
|
||||||
|
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
|
||||||
|
}
|
||||||
|
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)
|
res, err := d.triggerOnResolve(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return nil, err
|
||||||
}
|
}
|
||||||
if res != nil && len(res.Files) > 0 {
|
if res != nil && len(res.Files) > 0 {
|
||||||
rr = &ResolveResult{
|
return &ResolveResult{Res: res, SuggestedDownloadDir: d.suggestedDownloadDir(res)}, nil
|
||||||
Res: res,
|
|
||||||
SuggestedDownloadDir: d.suggestedDownloadDir(res),
|
|
||||||
}
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
f, err := d.buildFetcher(req.URL)
|
||||||
fetcher, err := d.buildFetcher(req.URL)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return
|
return nil, err
|
||||||
}
|
|
||||||
err = fetcher.Resolve(req)
|
|
||||||
if err != nil {
|
|
||||||
return
|
|
||||||
}
|
}
|
||||||
d.fetcherMapLock.Lock()
|
d.fetcherMapLock.Lock()
|
||||||
d.fetcherCache[rrId] = fetcher
|
state.fetcher = f
|
||||||
|
cancelled := state.cancelled || d.closed.Load()
|
||||||
d.fetcherMapLock.Unlock()
|
d.fetcherMapLock.Unlock()
|
||||||
rr = &ResolveResult{
|
if cancelled {
|
||||||
ID: rrId,
|
return nil, errors.New("resource resolution cancelled")
|
||||||
Res: fetcher.Meta().Res,
|
}
|
||||||
SuggestedDownloadDir: d.suggestedDownloadDir(fetcher.Meta().Res),
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
d.fetcherMapLock.Unlock()
|
||||||
|
if state != nil {
|
||||||
|
return state.close(f)
|
||||||
|
}
|
||||||
|
if f != nil {
|
||||||
|
return f.Close()
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d *Downloader) closeResolvedResources() {
|
||||||
|
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.fetcherMapLock.Unlock()
|
||||||
|
for id := range ids {
|
||||||
|
if err := d.ReleaseResolve(id); err != nil {
|
||||||
|
d.Logger.Warn().Err(err).Msg("close resolved resource failed")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Downloader) notifyRunning() {
|
func (d *Downloader) notifyRunning() {
|
||||||
@@ -875,18 +984,34 @@ 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) {
|
func (d *Downloader) createResolved(rrId string, opts *base.Options, groupID string) (taskId string, err error) {
|
||||||
d.fetcherMapLock.Lock()
|
d.fetcherMapLock.Lock()
|
||||||
fetcher, ok := d.fetcherCache[rrId]
|
fetcher, ok := d.fetcherCache[rrId]
|
||||||
|
state := d.resolveStates[rrId]
|
||||||
|
if ok && state != nil {
|
||||||
|
state.claimed = true
|
||||||
|
}
|
||||||
delete(d.fetcherCache, rrId)
|
delete(d.fetcherCache, rrId)
|
||||||
d.fetcherMapLock.Unlock()
|
d.fetcherMapLock.Unlock()
|
||||||
if !ok {
|
if !ok {
|
||||||
return "", errors.New("invalid resource id")
|
return "", errors.New("invalid resource id")
|
||||||
}
|
}
|
||||||
// Only one creator may use a resolved fetcher. Failed creation makes it
|
// Failed creation returns ownership to the preview unless it was cancelled.
|
||||||
// available again; a persisted task consumes it even if starting fails.
|
|
||||||
defer func() {
|
defer func() {
|
||||||
if taskId == "" {
|
d.fetcherMapLock.Lock()
|
||||||
d.fetcherMapLock.Lock()
|
cancelled := d.closed.Load() || (state != nil && state.cancelled)
|
||||||
|
if taskId == "" && !cancelled {
|
||||||
d.fetcherCache[rrId] = fetcher
|
d.fetcherCache[rrId] = fetcher
|
||||||
d.fetcherMapLock.Unlock()
|
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()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
if opts == nil {
|
if opts == nil {
|
||||||
@@ -1569,6 +1694,7 @@ func (d *Downloader) Close() error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
d.cancelBackground()
|
d.cancelBackground()
|
||||||
|
d.closeResolvedResources()
|
||||||
d.waitForExtensionGit()
|
d.waitForExtensionGit()
|
||||||
if d.protocolJobs != nil {
|
if d.protocolJobs != nil {
|
||||||
d.protocolJobs.close()
|
d.protocolJobs.close()
|
||||||
@@ -2460,8 +2586,6 @@ func (d *Downloader) doStart(task *Task) (err error) {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
task.Progress.Speed = 0
|
task.Progress.Speed = 0
|
||||||
task.speedArr = nil
|
|
||||||
task.downloadSampledAt = time.Time{}
|
|
||||||
task.timer.Start()
|
task.timer.Start()
|
||||||
task.statusLock.Unlock()
|
task.statusLock.Unlock()
|
||||||
if err := task.fetcher.Start(); err != nil {
|
if err := task.fetcher.Start(); err != nil {
|
||||||
@@ -3070,10 +3194,8 @@ func initTask(task *Task) {
|
|||||||
task.lock = &sync.Mutex{}
|
task.lock = &sync.Mutex{}
|
||||||
task.statsLock = &sync.Mutex{}
|
task.statsLock = &sync.Mutex{}
|
||||||
task.blobRefLock = &sync.Mutex{}
|
task.blobRefLock = &sync.Mutex{}
|
||||||
task.speedArr = nil
|
task.speedArr = make([]int64, 0)
|
||||||
task.uploadSpeedArr = nil
|
task.uploadSpeedArr = make([]int64, 0)
|
||||||
task.downloadSampledAt = time.Time{}
|
|
||||||
task.uploadSampledAt = time.Time{}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (d *Downloader) releaseBlobTask(task *Task) {
|
func (d *Downloader) releaseBlobTask(task *Task) {
|
||||||
|
|||||||
+26
-43
@@ -75,23 +75,20 @@ type Task struct {
|
|||||||
producer production.Source
|
producer production.Source
|
||||||
producerReceived int64
|
producerReceived int64
|
||||||
|
|
||||||
fetcherManager fetcher.FetcherManager
|
fetcherManager fetcher.FetcherManager
|
||||||
fetcher fetcher.Fetcher
|
fetcher fetcher.Fetcher
|
||||||
timer *util.Timer
|
timer *util.Timer
|
||||||
statusLock *sync.Mutex
|
statusLock *sync.Mutex
|
||||||
lock *sync.Mutex
|
lock *sync.Mutex
|
||||||
statsLock *sync.Mutex
|
statsLock *sync.Mutex
|
||||||
statsLoaded bool
|
statsLoaded bool
|
||||||
stats *fetcher.Stats
|
stats *fetcher.Stats
|
||||||
statsSnapshot json.RawMessage
|
statsSnapshot json.RawMessage
|
||||||
blobRefLock *sync.Mutex
|
blobRefLock *sync.Mutex
|
||||||
blobURL string
|
blobURL string
|
||||||
runGeneration uint64
|
runGeneration uint64
|
||||||
speedArr []speedSample
|
speedArr []int64
|
||||||
uploadSpeedArr []speedSample
|
uploadSpeedArr []int64
|
||||||
downloadSampledAt time.Time
|
|
||||||
uploadSampledAt time.Time
|
|
||||||
receivedBytes int64
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// TaskRuntimeStatus is the lightweight, frequently refreshed view of a task.
|
// TaskRuntimeStatus is the lightweight, frequently refreshed view of a task.
|
||||||
@@ -198,21 +195,7 @@ func (t *Task) updateUploadSpeed(downloaded int64, usedTime float64) int64 {
|
|||||||
return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime)
|
return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime)
|
||||||
}
|
}
|
||||||
|
|
||||||
type speedSample struct {
|
func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
|
||||||
bytes int64
|
|
||||||
seconds float64
|
|
||||||
}
|
|
||||||
|
|
||||||
func sampleSeconds(lastSample *time.Time, now time.Time) float64 {
|
|
||||||
previous := *lastSample
|
|
||||||
*lastSample = now
|
|
||||||
if previous.IsZero() {
|
|
||||||
return 0
|
|
||||||
}
|
|
||||||
return now.Sub(previous).Seconds()
|
|
||||||
}
|
|
||||||
|
|
||||||
func calcSpeed(speedArr *[]speedSample, downloaded int64, usedTime float64) int64 {
|
|
||||||
if usedTime <= 0 {
|
if usedTime <= 0 {
|
||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
@@ -221,18 +204,18 @@ func calcSpeed(speedArr *[]speedSample, downloaded int64, usedTime float64) int6
|
|||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
|
|
||||||
*speedArr = append(*speedArr, speedSample{bytes: downloaded, seconds: usedTime})
|
*speedArr = append(*speedArr, downloaded)
|
||||||
var total int64
|
// Record last 5 seconds of download speed to calculate the average speed
|
||||||
var elapsed float64
|
if len(*speedArr) > int(5.0/usedTime) {
|
||||||
first := len(*speedArr)
|
*speedArr = (*speedArr)[1:]
|
||||||
for first > 0 && elapsed < 5 {
|
|
||||||
first--
|
|
||||||
sample := (*speedArr)[first]
|
|
||||||
total += sample.bytes
|
|
||||||
elapsed += sample.seconds
|
|
||||||
}
|
}
|
||||||
*speedArr = (*speedArr)[first:]
|
|
||||||
return int64(float64(total) / elapsed)
|
var total int64
|
||||||
|
for _, v := range *speedArr {
|
||||||
|
total += v
|
||||||
|
}
|
||||||
|
|
||||||
|
return int64(float64(total) / float64(len(*speedArr)) / usedTime)
|
||||||
}
|
}
|
||||||
|
|
||||||
type TaskFilter struct {
|
type TaskFilter struct {
|
||||||
|
|||||||
@@ -1,32 +1,9 @@
|
|||||||
package download
|
package download
|
||||||
|
|
||||||
import (
|
import "testing"
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
func TestTransferSpeedWithUnevenSamples(t *testing.T) {
|
|
||||||
var task Task
|
|
||||||
for _, sample := range []struct {
|
|
||||||
bytes int64
|
|
||||||
seconds float64
|
|
||||||
}{
|
|
||||||
{1000, 1},
|
|
||||||
{2000, 2},
|
|
||||||
{4000, 4},
|
|
||||||
{6000, 6},
|
|
||||||
} {
|
|
||||||
if got := task.updateSpeed(sample.bytes, sample.seconds); got != 1000 {
|
|
||||||
t.Errorf("download speed after %v seconds = %d, want 1000", sample.seconds, got)
|
|
||||||
}
|
|
||||||
if got := task.updateUploadSpeed(sample.bytes, sample.seconds); got != 1000 {
|
|
||||||
t.Errorf("upload speed after %v seconds = %d, want 1000", sample.seconds, got)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestCalcSpeedResetOnRollback(t *testing.T) {
|
func TestCalcSpeedResetOnRollback(t *testing.T) {
|
||||||
speedArr := []speedSample{{1024, 1}, {2048, 1}, {4096, 1}}
|
speedArr := []int64{1024, 2048, 4096}
|
||||||
|
|
||||||
if got := calcSpeed(&speedArr, -512, 1); got != 0 {
|
if got := calcSpeed(&speedArr, -512, 1); got != 0 {
|
||||||
t.Fatalf("calcSpeed() = %d, want 0 after rollback", got)
|
t.Fatalf("calcSpeed() = %d, want 0 after rollback", got)
|
||||||
@@ -39,17 +16,3 @@ func TestCalcSpeedResetOnRollback(t *testing.T) {
|
|||||||
t.Fatalf("calcSpeed() = %d, want 1024 after reset", got)
|
t.Fatalf("calcSpeed() = %d, want 1024 after reset", got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestSampleSecondsUsesThePreviousSuccessfulSample(t *testing.T) {
|
|
||||||
start := time.Now()
|
|
||||||
var last time.Time
|
|
||||||
if got := sampleSeconds(&last, start); got != 0 {
|
|
||||||
t.Fatalf("first sample elapsed = %v, want 0", got)
|
|
||||||
}
|
|
||||||
if got := sampleSeconds(&last, start.Add(1500*time.Millisecond)); got != 1.5 {
|
|
||||||
t.Fatalf("elapsed after a delayed sample = %v, want 1.5", got)
|
|
||||||
}
|
|
||||||
if got := sampleSeconds(&last, start.Add(2*time.Second)); got != 0.5 {
|
|
||||||
t.Fatalf("elapsed after the next sample = %v, want 0.5", got)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -0,0 +1,224 @@
|
|||||||
|
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) {
|
func Resolve(w http.ResponseWriter, r *http.Request) {
|
||||||
var req model.ResolveTask
|
var req model.ResolveTask
|
||||||
if ReadJson(r, w, &req) {
|
if ReadJson(r, w, &req) {
|
||||||
rr, err := Downloader.Resolve(req.Req)
|
rr, err := Downloader.ResolveWithID(req.ID, req.Req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
WriteJson(w, model.NewErrorResult(err.Error()))
|
WriteJson(w, model.NewErrorResult(err.Error()))
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package model
|
|||||||
import "github.com/GopeedLab/gopeed/pkg/base"
|
import "github.com/GopeedLab/gopeed/pkg/base"
|
||||||
|
|
||||||
type ResolveTask struct {
|
type ResolveTask struct {
|
||||||
|
ID string `json:"id,omitempty"`
|
||||||
Req *base.Request `json:"req"`
|
Req *base.Request `json:"req"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -81,6 +81,9 @@ 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) {
|
Future<String> createTask(CreateTask createTask) {
|
||||||
return _parse<String>(() => _request('api/v1/tasks', method: 'POST', data: createTask), (data) => data as String);
|
return _parse<String>(() => _request('api/v1/tasks', method: 'POST', data: createTask), (data) => data as String);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,7 +8,10 @@ part 'resolve_task.g.dart';
|
|||||||
class ResolveTask {
|
class ResolveTask {
|
||||||
Request? req;
|
Request? req;
|
||||||
|
|
||||||
ResolveTask({this.req});
|
@JsonKey(includeIfNull: false)
|
||||||
|
String? id;
|
||||||
|
|
||||||
|
ResolveTask({this.req, this.id});
|
||||||
|
|
||||||
factory ResolveTask.fromJson(Map<String, dynamic> json) => _$ResolveTaskFromJson(json);
|
factory ResolveTask.fromJson(Map<String, dynamic> json) => _$ResolveTaskFromJson(json);
|
||||||
Map<String, dynamic> toJson() => _$ResolveTaskToJson(this);
|
Map<String, dynamic> toJson() => _$ResolveTaskToJson(this);
|
||||||
|
|||||||
@@ -10,7 +10,8 @@ ResolveTask _$ResolveTaskFromJson(Map<String, dynamic> json) => ResolveTask(
|
|||||||
req: json['req'] == null
|
req: json['req'] == null
|
||||||
? null
|
? null
|
||||||
: Request.fromJson(json['req'] as Map<String, dynamic>),
|
: Request.fromJson(json['req'] as Map<String, dynamic>),
|
||||||
|
id: json['id'] as String?,
|
||||||
);
|
);
|
||||||
|
|
||||||
Map<String, dynamic> _$ResolveTaskToJson(ResolveTask instance) =>
|
Map<String, dynamic> _$ResolveTaskToJson(ResolveTask instance) =>
|
||||||
<String, dynamic>{'req': ?instance.req};
|
<String, dynamic>{'req': ?instance.req, 'id': ?instance.id};
|
||||||
@@ -53,7 +53,6 @@ class AppRuntimeState {
|
|||||||
class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
|
class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
|
||||||
static StartConfig? _defaultStartConfig;
|
static StartConfig? _defaultStartConfig;
|
||||||
Future<void> _preferenceWrites = Future<void>.value();
|
Future<void> _preferenceWrites = Future<void>.value();
|
||||||
bool _analyticsStarted = false;
|
|
||||||
|
|
||||||
@override
|
@override
|
||||||
Future<AppRuntimeState> build() async {
|
Future<AppRuntimeState> build() async {
|
||||||
@@ -74,7 +73,6 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
|
|||||||
startConfig
|
startConfig
|
||||||
..apiEnable = true
|
..apiEnable = true
|
||||||
..mcpEnable = config.api.mcpEnable;
|
..mcpEnable = config.api.mcpEnable;
|
||||||
await _initAnalytics(config);
|
|
||||||
} else {
|
} else {
|
||||||
startConfig
|
startConfig
|
||||||
..apiEnable = config.api.enable
|
..apiEnable = config.api.enable
|
||||||
@@ -255,8 +253,7 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
Future<void> _initAnalytics(DownloaderConfig config) async {
|
Future<void> _initAnalytics(DownloaderConfig config) async {
|
||||||
if (_analyticsStarted || !AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return;
|
if (!AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return;
|
||||||
_analyticsStarted = true;
|
|
||||||
try {
|
try {
|
||||||
final generatedClientId = await Analytics.instance.init(fallbackClientId: config.extra.deviceId);
|
final generatedClientId = await Analytics.instance.init(fallbackClientId: config.extra.deviceId);
|
||||||
if (generatedClientId != null) {
|
if (generatedClientId != null) {
|
||||||
@@ -265,7 +262,6 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
|
|||||||
}
|
}
|
||||||
unawaited(Analytics.instance.logAppOpen());
|
unawaited(Analytics.instance.logAppOpen());
|
||||||
} catch (error, stackTrace) {
|
} catch (error, stackTrace) {
|
||||||
_analyticsStarted = false;
|
|
||||||
logger.w('GA4 init failed', error, stackTrace);
|
logger.w('GA4 init failed', error, stackTrace);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ Future<DownloaderConfig> loadDownloaderConfig({
|
|||||||
config = await load();
|
config = await load();
|
||||||
} catch (error, stackTrace) {
|
} catch (error, stackTrace) {
|
||||||
onLoadError?.call(error, stackTrace);
|
onLoadError?.call(error, stackTrace);
|
||||||
config = DownloaderConfig()..extra.analyticsEnabled = false;
|
config = DownloaderConfig();
|
||||||
loaded = false;
|
loaded = false;
|
||||||
}
|
}
|
||||||
final changed = await initDownloaderConfig(config, defaultDownloadDir: defaultDownloadDir);
|
final changed = await initDownloaderConfig(config, defaultDownloadDir: defaultDownloadDir);
|
||||||
|
|||||||
@@ -93,6 +93,10 @@ void _bindGopeed(CapabilityRegistry registry, _ConfigStore configStore) {
|
|||||||
(p) => api.getMirrorCandidates(p['provider'] as String, p['url'] as String),
|
(p) => api.getMirrorCandidates(p['provider'] as String, p['url'] as String),
|
||||||
)
|
)
|
||||||
..bind(GopeedMethods.resolve, api.resolve)
|
..bind(GopeedMethods.resolve, api.resolve)
|
||||||
|
..bind(GopeedMethods.releaseResolve, (id) async {
|
||||||
|
await api.releaseResolve(id);
|
||||||
|
return const RpcUnit();
|
||||||
|
})
|
||||||
..bind(GopeedMethods.createTask, api.createTask)
|
..bind(GopeedMethods.createTask, api.createTask)
|
||||||
..bind(GopeedMethods.createTaskBatch, api.createTaskBatch)
|
..bind(GopeedMethods.createTaskBatch, api.createTaskBatch)
|
||||||
..bind(GopeedMethods.patchTask, (params) async {
|
..bind(GopeedMethods.patchTask, (params) async {
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ abstract final class GopeedMethods {
|
|||||||
static const triggerProtocolJob = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.protocol.job.trigger');
|
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 getMirrorCandidates = RpcMethod<Map<String, dynamic>, List<String>>('gopeed.mirrors.candidates');
|
||||||
static const resolve = RpcMethod<ResolveTask, ResolveResult>('gopeed.resolve');
|
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 createTask = RpcMethod<CreateTask, String>('gopeed.task.create');
|
||||||
static const createTaskBatch = RpcMethod<CreateTaskBatch, CreateTaskBatchResult>('gopeed.task.createBatch');
|
static const createTaskBatch = RpcMethod<CreateTaskBatch, CreateTaskBatchResult>('gopeed.task.createBatch');
|
||||||
static const patchTask = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.task.patch');
|
static const patchTask = RpcMethod<Map<String, dynamic>, RpcUnit>('gopeed.task.patch');
|
||||||
@@ -50,6 +51,7 @@ class GopeedService {
|
|||||||
Future<List<String>> getMirrorCandidates(String provider, String url) =>
|
Future<List<String>> getMirrorCandidates(String provider, String url) =>
|
||||||
_invoker.invoke(GopeedMethods.getMirrorCandidates, {'provider': provider, 'url': url});
|
_invoker.invoke(GopeedMethods.getMirrorCandidates, {'provider': provider, 'url': url});
|
||||||
Future<ResolveResult> resolve(ResolveTask request) => _invoker.invoke(GopeedMethods.resolve, request);
|
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);
|
Future<String> createTask(CreateTask request) => _invoker.invoke(GopeedMethods.createTask, request);
|
||||||
|
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
import 'dart:developer' as developer;
|
||||||
|
import 'package:uuid/uuid.dart';
|
||||||
import 'package:path/path.dart' as path;
|
import 'package:path/path.dart' as path;
|
||||||
|
|
||||||
import '../../../api/model/create_task_batch.dart';
|
import '../../../api/model/create_task_batch.dart';
|
||||||
@@ -16,6 +18,8 @@ class PendingDownload {
|
|||||||
final Request request;
|
final Request request;
|
||||||
final Options options;
|
final Options options;
|
||||||
ResolveResult? result;
|
ResolveResult? result;
|
||||||
|
String? _resolveID;
|
||||||
|
bool _cancelled = false;
|
||||||
String? resolveError;
|
String? resolveError;
|
||||||
String? createError;
|
String? createError;
|
||||||
final Set<int> selected = {};
|
final Set<int> selected = {};
|
||||||
@@ -50,11 +54,16 @@ class PendingDownload {
|
|||||||
bool get hasSelection => !created && result != null && selected.isNotEmpty;
|
bool get hasSelection => !created && result != null && selected.isNotEmpty;
|
||||||
|
|
||||||
Future<void> resolve(GopeedService service) async {
|
Future<void> resolve(GopeedService service) async {
|
||||||
if (result != null || resolving) return;
|
if (_cancelled || result != null || resolving) return;
|
||||||
resolving = true;
|
resolving = true;
|
||||||
resolveError = null;
|
resolveError = null;
|
||||||
try {
|
try {
|
||||||
result = await service.resolve(ResolveTask(req: request));
|
_resolveID = const Uuid().v4();
|
||||||
|
result = await service.resolve(ResolveTask(req: request, id: _resolveID));
|
||||||
|
if (_cancelled) {
|
||||||
|
await _release(service, result!.id);
|
||||||
|
return;
|
||||||
|
}
|
||||||
selected.addAll(List.generate(files.isEmpty ? 1 : files.length, (index) => index));
|
selected.addAll(List.generate(files.isEmpty ? 1 : files.length, (index) => index));
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
resolveError = error.toString();
|
resolveError = error.toString();
|
||||||
@@ -63,6 +72,23 @@ 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() {
|
List<CreateTaskBatchItem> creationItems() {
|
||||||
final indexes = selected.toList()..sort();
|
final indexes = selected.toList()..sort();
|
||||||
if (result!.id.isNotEmpty || files.isEmpty) {
|
if (result!.id.isNotEmpty || files.isEmpty) {
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
import 'dart:async';
|
||||||
|
|
||||||
import 'package:shadcn_flutter/shadcn_flutter.dart';
|
import 'package:shadcn_flutter/shadcn_flutter.dart';
|
||||||
import 'package:flutter/widgets.dart' as flutter show Flexible;
|
import 'package:flutter/widgets.dart' as flutter show Flexible;
|
||||||
|
|
||||||
@@ -49,6 +51,7 @@ class _ResolveDownloadsDialogState extends State<_ResolveDownloadsDialog> {
|
|||||||
@override
|
@override
|
||||||
void dispose() {
|
void dispose() {
|
||||||
_cancelled = true;
|
_cancelled = true;
|
||||||
|
_releasePreviews();
|
||||||
super.dispose();
|
super.dispose();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -64,8 +67,15 @@ class _ResolveDownloadsDialogState extends State<_ResolveDownloadsDialog> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void _releasePreviews() {
|
||||||
|
for (final source in widget.sources) {
|
||||||
|
unawaited(source.cancel(widget.service));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
void _cancel() {
|
void _cancel() {
|
||||||
_cancelled = true;
|
_cancelled = true;
|
||||||
|
_releasePreviews();
|
||||||
closeOverlay(context, false);
|
closeOverlay(context, false);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -32,22 +32,8 @@ void main() {
|
|||||||
// The in-memory fallback still gets usable defaults.
|
// The in-memory fallback still gets usable defaults.
|
||||||
expect(config.downloadDir, '/downloads');
|
expect(config.downloadDir, '/downloads');
|
||||||
expect(config.categories, hasLength(4));
|
expect(config.categories, hasLength(4));
|
||||||
expect(config.extra.analyticsEnabled, isFalse);
|
|
||||||
});
|
});
|
||||||
|
|
||||||
for (final analyticsEnabled in [false, true]) {
|
|
||||||
test('preserves the saved analytics preference ($analyticsEnabled)', () async {
|
|
||||||
final stored = configWith()..extra.analyticsEnabled = analyticsEnabled;
|
|
||||||
final config = await loadDownloaderConfig(
|
|
||||||
load: () async => stored,
|
|
||||||
save: (_) async {},
|
|
||||||
defaultDownloadDir: () async => '/downloads',
|
|
||||||
);
|
|
||||||
|
|
||||||
expect(config.extra.analyticsEnabled, analyticsEnabled);
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
test('skips persist when init changed nothing', () async {
|
test('skips persist when init changed nothing', () async {
|
||||||
var saved = 0;
|
var saved = 0;
|
||||||
await loadDownloaderConfig(
|
await loadDownloaderConfig(
|
||||||
|
|||||||
@@ -64,6 +64,81 @@ Future<void> open(
|
|||||||
}
|
}
|
||||||
|
|
||||||
void main() {
|
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]) {
|
for (final width in [390.0, 700.0, 1024.0]) {
|
||||||
testWidgets('file list fits and exposes ready selections at width $width', (tester) async {
|
testWidgets('file list fits and exposes ready selections at width $width', (tester) async {
|
||||||
final registry = CapabilityRegistry(createAppCapabilityCodecs())
|
final registry = CapabilityRegistry(createAppCapabilityCodecs())
|
||||||
|
|||||||
Reference in new issue
Block a user