feat: extension support (#78)

This commit is contained in:
Levi authored and GitHub committed 2023-10-16 09:47:31 +08:00
1 parent 82d316af83
commit c073fc0a67
118 files changed
+6971 -780

No files matched your search

+28 -43
View File
@@ -9,7 +9,6 @@ import (
"github.com/GopeedLab/gopeed/pkg/util"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/metainfo"
"path"
"path/filepath"
"sync"
"sync/atomic"
@@ -38,7 +37,7 @@ func (f *Fetcher) Name() string {
return "bt"
}
func (f *Fetcher) Setup(ctl *controller.Controller) (err error) {
func (f *Fetcher) Setup(ctl *controller.Controller) {
f.ctl = ctl
if f.meta == nil {
f.meta = &fetcher.FetcherMeta{}
@@ -77,51 +76,17 @@ func (f *Fetcher) Resolve(req *base.Request) error {
if err := f.addTorrent(req); err != nil {
return err
}
go func() {
// recycle unused torrent resource
time.Sleep(time.Minute * 3)
if !f.create.Load() {
f.torrentReady.Store(false)
f.safeDrop()
}
}()
res := &base.Resource{
Name: f.torrent.Name(),
Range: true,
RootDir: f.torrent.Name(),
Files: make([]*base.FileInfo, len(f.torrent.Files())),
Hash: f.torrent.InfoHash().String(),
}
for i, file := range f.torrent.Files() {
res.Files[i] = &base.FileInfo{
Name: filepath.Base(file.DisplayPath()),
Path: util.Dir(file.Path()),
Size: file.Length(),
}
res.Size += file.Length()
}
f.updateRes()
f.meta.Req = req
f.meta.Res = res
return nil
}
func (f *Fetcher) Create(opts *base.Options) (err error) {
f.create.Store(true)
f.meta.Opts = opts
if len(opts.SelectFiles) == 0 {
opts.SelectFiles = make([]int, 0)
for i := range f.torrent.Files() {
opts.SelectFiles = append(opts.SelectFiles, i)
}
if f.meta.Res != nil {
torrentDirMap[f.meta.Res.Hash] = f.meta.FolderPath()
}
if opts.Path != "" {
torrentDirMap[f.meta.Res.Hash] = path.Join(f.meta.Opts.Path, f.meta.Res.RootDir)
if ft, ok := ftMap[f.meta.Res.Hash]; ok {
// reuse resolve fetcher
ft.setTorrentDir(torrentDirMap[f.meta.Res.Hash])
}
}
f.progress = make(fetcher.Progress, len(f.meta.Opts.SelectFiles))
return nil
}
@@ -132,6 +97,9 @@ func (f *Fetcher) Start() (err error) {
return
}
}
if ft, ok := ftMap[f.meta.Res.Hash]; ok {
ft.setTorrentDir(f.meta.FolderPath())
}
files := f.torrent.Files()
if len(f.meta.Opts.SelectFiles) == len(files) {
f.torrent.DownloadAll()
@@ -150,10 +118,6 @@ func (f *Fetcher) Pause() (err error) {
return
}
func (f *Fetcher) Continue() (err error) {
return f.Start()
}
func (f *Fetcher) Close() (err error) {
f.safeDrop()
return nil
@@ -217,6 +181,27 @@ func (f *Fetcher) Progress() fetcher.Progress {
return f.progress
}
func (f *Fetcher) updateRes() {
res := &base.Resource{
Name: f.torrent.Name(),
Range: true,
Files: make([]*base.FileInfo, len(f.torrent.Files())),
Hash: f.torrent.InfoHash().String(),
}
for i, file := range f.torrent.Files() {
res.Files[i] = &base.FileInfo{
Name: filepath.Base(file.DisplayPath()),
Path: util.Dir(file.Path()),
Size: file.Length(),
}
}
res.CalcSize()
f.meta.Res = res
if f.meta.Opts != nil {
f.meta.Opts.InitSelectFiles(len(res.Files))
}
}
func (f *Fetcher) addTorrent(req *base.Request) (err error) {
if err = f.initClient(); err != nil {
return
+6 -8
View File
@@ -33,10 +33,9 @@ func TestFetcher_Resolve_DataUri_Torrent(t *testing.T) {
}
want := &base.Resource{
Name: "ubuntu-22.04-live-server-amd64.iso",
Size: 1466714112,
Range: true,
RootDir: "ubuntu-22.04-live-server-amd64.iso",
Name: "ubuntu-22.04-live-server-amd64.iso",
Size: 1466714112,
Range: true,
Files: []*base.FileInfo{
{
Name: "ubuntu-22.04-live-server-amd64.iso",
@@ -69,10 +68,9 @@ func doResolve(t *testing.T, fetcher fetcher.Fetcher) {
}
want := &base.Resource{
Name: "ubuntu-22.04-live-server-amd64.iso",
Size: 1466714112,
Range: true,
RootDir: "ubuntu-22.04-live-server-amd64.iso",
Name: "ubuntu-22.04-live-server-amd64.iso",
Size: 1466714112,
Range: true,
Files: []*base.FileInfo{
{
Name: "ubuntu-22.04-live-server-amd64.iso",
+42 -81
View File
@@ -3,6 +3,7 @@ package http
import (
"bytes"
"context"
"errors"
"fmt"
"github.com/GopeedLab/gopeed/internal/controller"
"github.com/GopeedLab/gopeed/internal/fetcher"
@@ -33,7 +34,6 @@ func (re *RequestError) Error() string {
}
type chunk struct {
Status base.Status
Begin int64
End int64
Downloaded int64
@@ -41,9 +41,8 @@ type chunk struct {
func newChunk(begin int64, end int64) *chunk {
return &chunk{
Status: base.DownloadStatusReady,
Begin: begin,
End: end,
Begin: begin,
End: end,
}
}
@@ -52,23 +51,20 @@ type Fetcher struct {
doneCh chan error
meta *fetcher.FetcherMeta
status base.Status
chunks []*chunk
file *os.File
ctx context.Context
cancel context.CancelFunc
pauseCh chan interface{}
file *os.File
cancel context.CancelFunc
eg *errgroup.Group
}
func (f *Fetcher) Name() string {
return "http"
}
func (f *Fetcher) Setup(ctl *controller.Controller) (err error) {
func (f *Fetcher) Setup(ctl *controller.Controller) {
f.ctl = ctl
f.doneCh = make(chan error, 1)
f.pauseCh = make(chan interface{}, 1)
if f.meta == nil {
f.meta = &fetcher.FetcherMeta{}
}
@@ -139,10 +135,9 @@ func (f *Fetcher) Resolve(req *base.Request) error {
}
// unknown file filePath
if file.Name == "" || file.Name == "/" || file.Name == "." {
file.Name = "unknown"
file.Name = httpReq.URL.Hostname()
}
res.Files = append(res.Files, file)
res.Name = file.Name
f.meta.Req = req
f.meta.Res = res
return nil
@@ -150,7 +145,6 @@ func (f *Fetcher) Resolve(req *base.Request) error {
func (f *Fetcher) Create(opts *base.Options) error {
f.meta.Opts = opts
f.status = base.DownloadStatusReady
if err := base.ParseReqExtra[fhttp.ReqExtra](f.meta.Req); err != nil {
return err
@@ -178,44 +172,38 @@ func (f *Fetcher) Create(opts *base.Options) error {
}
func (f *Fetcher) Start() (err error) {
// 创建文件
name := f.filepath()
f.file, err = f.ctl.Touch(name, f.meta.Res.Size)
name := f.meta.SingleFilepath()
// if file not exist, create it, else open it
_, err = os.Stat(name)
if err != nil {
if os.IsNotExist(err) {
f.file, err = f.ctl.Touch(name, f.meta.Res.Size)
} else {
return
}
} else {
f.file, err = os.OpenFile(name, os.O_RDWR, os.ModeAppend)
}
if err != nil {
return err
}
f.status = base.DownloadStatusRunning
f.chunks = f.splitChunk()
if f.chunks == nil {
f.chunks = f.splitChunk()
}
f.fetch()
return
}
func (f *Fetcher) Pause() (err error) {
if base.DownloadStatusRunning != f.status {
return
}
f.status = base.DownloadStatusPause
if f.cancel != nil {
f.cancel()
<-f.pauseCh
// wait for pause handle complete
f.eg.Wait()
f.file.Close()
}
return
}
func (f *Fetcher) Continue() (err error) {
if base.DownloadStatusRunning == f.status || base.DownloadStatusDone == f.status {
return
}
f.status = base.DownloadStatusRunning
f.file, err = os.OpenFile(f.filepath(), os.O_RDWR, os.ModeAppend)
if err != nil {
return err
}
f.fetch()
return
}
func (f *Fetcher) Close() (err error) {
if err = f.Pause(); err != nil {
return
@@ -243,41 +231,32 @@ func (f *Fetcher) Wait() (err error) {
return <-f.doneCh
}
func (f *Fetcher) filepath() string {
return f.meta.Filepath(f.meta.Res.Files[0])
}
func (f *Fetcher) fetch() {
f.ctx, f.cancel = context.WithCancel(context.Background())
eg, _ := errgroup.WithContext(f.ctx)
var ctx context.Context
ctx, f.cancel = context.WithCancel(context.Background())
f.eg, _ = errgroup.WithContext(ctx)
for i := 0; i < len(f.chunks); i++ {
i := i
eg.Go(func() error {
return f.fetchChunk(i)
f.eg.Go(func() error {
return f.fetchChunk(i, ctx)
})
}
go func() {
err := eg.Wait()
// 下载停止,关闭文件句柄
f.file.Close()
if f.status == base.DownloadStatusPause {
f.pauseCh <- nil
} else {
if err != nil {
f.status = base.DownloadStatusError
} else {
f.status = base.DownloadStatusDone
}
f.doneCh <- err
err := f.eg.Wait()
// check if canceled
if errors.Is(err, context.Canceled) {
return
}
f.file.Close()
f.doneCh <- err
}()
}
func (f *Fetcher) fetchChunk(index int) (err error) {
func (f *Fetcher) fetchChunk(index int, ctx context.Context) (err error) {
chunk := f.chunks[index]
httpReq, err := buildRequest(f.ctx, f.meta.Req)
httpReq, err := buildRequest(ctx, f.meta.Req)
if err != nil {
return err
}
@@ -287,14 +266,10 @@ func (f *Fetcher) fetchChunk(index int) (err error) {
)
// 重试5次
for i := 0; i < 5; i++ {
// 如果下载完成直接返回
if chunk.Status == base.DownloadStatusDone {
// if chunk is completed, return
if f.meta.Res.Range && chunk.Downloaded >= chunk.End-chunk.Begin+1 {
return
}
// 如果已暂停直接跳出
if f.status == base.DownloadStatusPause {
break
}
var (
resp *http.Response
retry bool
@@ -317,7 +292,7 @@ func (f *Fetcher) fetchChunk(index int) (err error) {
return nil
}()
if err != nil {
// 请求失败重试
// retry request
continue
}
// 请求成功就重置错误次数,连续失败5次才终止
@@ -347,19 +322,6 @@ func (f *Fetcher) fetchChunk(index int) (err error) {
break
}
}
if err != nil {
chunk.Status = base.DownloadStatusError
return
}
isComplete := f.meta.Res.Range && chunk.Downloaded >= chunk.End-chunk.Begin+1
if f.status == base.DownloadStatusPause && !isComplete {
chunk.Status = base.DownloadStatusPause
return
}
chunk.Status = base.DownloadStatusDone
return
}
@@ -474,7 +436,6 @@ func (fb *FetcherBuilder) Restore() (v any, f func(meta *fetcher.FetcherMeta, v
fb := &FetcherBuilder{}
fetcher := fb.Build().(*Fetcher)
fetcher.meta = meta
fetcher.status = base.DownloadStatusPause
base.ParseReqExtra[fhttp.ReqExtra](fetcher.meta.Req)
base.ParseOptsExtra[fhttp.OptsExtra](fetcher.meta.Opts)
if len(fd.Chunks) == 0 {
+16 -6
View File
@@ -16,7 +16,6 @@ import (
func TestFetcher_Resolve(t *testing.T) {
testResolve(test.StartTestFileServer, test.BuildName, &base.Resource{
Name: test.BuildName,
Size: test.BuildSize,
Range: true,
Files: []*base.FileInfo{
@@ -27,7 +26,6 @@ func TestFetcher_Resolve(t *testing.T) {
},
}, t)
testResolve(test.StartTestCustomServer, "disposition", &base.Resource{
Name: test.BuildName,
Size: test.BuildSize,
Range: false,
Files: []*base.FileInfo{
@@ -38,7 +36,6 @@ func TestFetcher_Resolve(t *testing.T) {
},
}, t)
testResolve(test.StartTestCustomServer, test.BuildName, &base.Resource{
Name: test.BuildName,
Size: 0,
Range: false,
Files: []*base.FileInfo{
@@ -48,6 +45,17 @@ func TestFetcher_Resolve(t *testing.T) {
},
},
}, t)
fetcher := buildFetcher()
err := fetcher.Resolve(&base.Request{
URL: "http://github.com",
})
if err != nil {
t.Fatal(err)
}
if fetcher.Meta().Res.Files[0].Name != "github.com" {
t.Errorf("Resolve() got = %v, want %v", fetcher.Meta().Res, "github.com")
}
}
func testResolve(startTestServer func() net.Listener, path string, want *base.Resource, t *testing.T) {
@@ -125,7 +133,9 @@ func TestFetcher_DownloadResume(t *testing.T) {
}
func TestFetcher_Config(t *testing.T) {
fetcher := doDownloadReady(buildConfigFetcher(), test.StartTestFileServer(), 0, t)
listener := test.StartTestFileServer()
defer listener.Close()
fetcher := doDownloadReady(buildConfigFetcher(), listener, 0, t)
err := fetcher.Start()
if err != nil {
t.Fatal(err)
@@ -222,7 +232,7 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) {
t.Fatal(err)
}
time.Sleep(time.Millisecond * 50)
if err := fetcher.Continue(); err != nil {
if err := fetcher.Start(); err != nil {
t.Fatal(err)
}
err = fetcher.Wait()
@@ -270,7 +280,7 @@ func downloadResume(listener net.Listener, connections int, t *testing.T) {
t.Fatal(err)
}
fetcher.Setup(controller.NewController())
fetcher.Continue()
fetcher.Start()
err = fetcher.Wait()
if err != nil {