From 74d0e8e0bc39bb4b148f09d3cdd55c1596064cf6 Mon Sep 17 00:00:00 2001 From: liwei Date: Fri, 5 Mar 2021 19:52:55 +0800 Subject: [PATCH] temp --- download/base/event.go | 4 +- download/base/fetcher.go | 11 ++-- download/downloader.go | 43 +++++++++++-- download/downloader_test.go | 3 + download/http/fetcher.go | 111 ++++++++++++++++------------------ download/http/fetcher_test.go | 3 +- 6 files changed, 102 insertions(+), 73 deletions(-) diff --git a/download/base/event.go b/download/base/event.go index 9cf092b5..02c555dd 100644 --- a/download/base/event.go +++ b/download/base/event.go @@ -1,5 +1,5 @@ package base -type Event interface { - Status() Status +type Event struct { + Status } diff --git a/download/base/fetcher.go b/download/base/fetcher.go index ea111254..20cf334a 100644 --- a/download/base/fetcher.go +++ b/download/base/fetcher.go @@ -4,13 +4,12 @@ package base type Fetcher interface { Init(ctl Controller) // 解析请求 - Resolve(req *Request) (*Resource, error) + Resolve(req *Request) (res *Resource, err error) // 创建任务 - Create(res *Resource, opts *Options) error - Start() error - Pause() error - Continue() error - Delete() error + Create(res *Resource, opts *Options) (err error) + Start() (err error) + Pause() (err error) + Continue() (err error) } type DefaultFetcher struct { diff --git a/download/downloader.go b/download/downloader.go index 52eb0090..bcab7fa3 100644 --- a/download/downloader.go +++ b/download/downloader.go @@ -6,6 +6,7 @@ import ( "github.com/monkeyWie/gopeed-core/download/base" "net/url" "strings" + "sync" ) type TaskInfo struct { @@ -16,14 +17,16 @@ type TaskInfo struct { Files map[string]*base.FileInfo fetcher base.Fetcher + locker *sync.Mutex } type FetcherBuilder func() (protocols []string, builder func() base.Fetcher) type Downloader struct { + *base.DefaultController fetchBuilders map[string]func() base.Fetcher tasks map[string]*TaskInfo - *base.DefaultController + listener func(taskInfo *TaskInfo, status base.Status) } func NewDownloader(fbs ...FetcherBuilder) *Downloader { @@ -60,14 +63,14 @@ func (d *Downloader) Resolve(req *base.Request) (*base.Resource, error) { return fetcher.Resolve(req) } -func (d *Downloader) Create(res *base.Resource, opts *base.Options) error { +func (d *Downloader) Create(res *base.Resource, opts *base.Options) (err error) { fetcher, err := d.buildFetcher(res.Req.URL) if err != nil { - return err + return } err = fetcher.Create(res, opts) if err != nil { - return err + return } id := uuid.New().String() d.tasks[id] = &TaskInfo{ @@ -76,10 +79,40 @@ func (d *Downloader) Create(res *base.Resource, opts *base.Options) error { Opts: opts, Status: base.DownloadStatusReady, fetcher: fetcher, + locker: new(sync.Mutex), } - return fetcher.Start() + go func() { + err = fetcher.Start() + if err != nil { + d.emit(d.tasks[id], base.DownloadStatusError) + } + }() + d.emit(d.tasks[id], base.DownloadStatusStart) + return } func (d *Downloader) Pause(id string) { + task := d.tasks[id] + task.locker.Lock() + defer task.locker.Unlock() d.tasks[id].fetcher.Pause() + d.emit(d.tasks[id], base.DownloadStatusPause) +} + +func (d *Downloader) Continue(id string) { + task := d.tasks[id] + task.locker.Lock() + defer task.locker.Unlock() + d.tasks[id].fetcher.Continue() + d.emit(d.tasks[id], base.DownloadStatusStart) +} + +func (d *Downloader) Listener(fn func(taskInfo *TaskInfo, status base.Status)) { + d.listener = fn +} + +func (d *Downloader) emit(taskInfo *TaskInfo, status base.Status) { + if d.listener != nil { + d.listener(taskInfo, status) + } } diff --git a/download/downloader_test.go b/download/downloader_test.go index 8280bceb..1cc86322 100644 --- a/download/downloader_test.go +++ b/download/downloader_test.go @@ -26,6 +26,9 @@ func TestDownloader_Create(t *testing.T) { if err != nil { t.Fatal(err) } + downloader.Listener(func(taskInfo *TaskInfo, status base.Status) { + fmt.Printf("task:%s,status %d\n", taskInfo.ID, status) + }) err = downloader.Create(res, &base.Options{ Path: "E:\\test\\gopeed\\http", Connections: 4, diff --git a/download/http/fetcher.go b/download/http/fetcher.go index 5168b200..96d80679 100644 --- a/download/http/fetcher.go +++ b/download/http/fetcher.go @@ -14,7 +14,6 @@ import ( "path" "path/filepath" "strconv" - "sync" "time" ) @@ -40,11 +39,16 @@ type Fetcher struct { clients []*http.Response chunks []*model.Chunk - pauseCond *sync.Cond + pauseCh chan interface{} + doneCh chan error } func NewFetcher() *Fetcher { - return &Fetcher{} + return &Fetcher{ + DefaultFetcher: new(base.DefaultFetcher), + pauseCh: make(chan interface{}), + doneCh: make(chan error), + } } var protocols = []string{"HTTP", "HTTPS"} @@ -56,11 +60,11 @@ func FetcherBuilder() ([]string, func() base.Fetcher) { } func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { - httpReq, err := BuildRequest(req) + httpReq, err := buildRequest(req) if err != nil { return nil, err } - client := BuildClient() + client := buildClient() // 只访问一个字节，测试资源是否支持Range请求 httpReq.Header.Set(base.HttpHeaderRange, fmt.Sprintf(base.HttpHeaderRangeFormat, 0, 0)) httpResp, err := client.Do(httpReq) @@ -129,18 +133,17 @@ func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error { f.res = res f.opts = opts f.status = base.DownloadStatusReady - f.pauseCond = sync.NewCond(&sync.Mutex{}) return nil } -func (f *Fetcher) Start() error { +func (f *Fetcher) Start() (err error) { // 创建文件 name := f.name() - _, err := f.Ctl.Touch(name, f.res.Size) + _, err = f.Ctl.Touch(name, f.res.Size) if err != nil { return err } - defer f.close() + defer f.Ctl.Close(name) f.status = base.DownloadStatusStart if f.res.Range { // 每个连接平均需要下载的分块大小 @@ -167,37 +170,28 @@ func (f *Fetcher) Start() error { f.clients = make([]*http.Response, 1) f.chunks[0] = model.NewChunk(0, 0) } + return f.fetch() } -func (f *Fetcher) Pause() error { - f.pauseCond.L.Lock() - defer f.pauseCond.L.Unlock() +func (f *Fetcher) Pause() (err error) { if base.DownloadStatusStart != f.status { - return nil + return } f.status = base.DownloadStatusPause f.stop() - // 释放锁，并等待下载结束 - f.pauseCond.Wait() - return nil + <-f.pauseCh + f.Ctl.Close(f.name()) + return } -func (f *Fetcher) Continue() error { - if func() bool { - f.pauseCond.L.Lock() - defer f.pauseCond.L.Unlock() - if base.DownloadStatusPause != f.status && base.DownloadStatusError != f.status { - return true - } - f.status = base.DownloadStatusStart - return false - }() { - return nil +func (f *Fetcher) Continue() (err error) { + if base.DownloadStatusStart == f.status || base.DownloadStatusDone == f.status { + return } - + f.status = base.DownloadStatusStart var name = f.name() - _, err := f.Ctl.Open(name) + _, err = f.Ctl.Open(name) if err != nil { return err } @@ -205,14 +199,6 @@ func (f *Fetcher) Continue() error { return f.fetch() } -func (f *Fetcher) Delete() error { - panic("implement me") -} - -func (f *Fetcher) close() error { - return f.Ctl.Close(f.name()) -} - func (f *Fetcher) name() string { // 创建文件 var filename = f.opts.Name @@ -222,7 +208,7 @@ func (f *Fetcher) name() string { return filepath.Join(f.opts.Path, filename) } -func (f *Fetcher) fetch() error { +func (f *Fetcher) fetch() (err error) { errCh := make(chan error, f.opts.Connections) defer close(errCh) @@ -232,23 +218,34 @@ func (f *Fetcher) fetch() error { }(i) } - var retErr error + stopFlag := false for i := 0; i < f.opts.Connections; i++ { - err := <-errCh - if err != nil && retErr == nil { - retErr = err - // 有一个连接失败就立即终止下载 - if retErr != base.PauseErr { - func() { - f.pauseCond.L.Lock() - defer f.pauseCond.L.Unlock() - f.stop() - }() + fetchErr := <-errCh + if fetchErr != nil && !stopFlag { + // 确保如果有暂停操作返回暂停 + if err == nil || err != base.PauseErr { + err = fetchErr + } + if fetchErr != base.PauseErr { + // 有一个连接失败就立即终止下载 + stopFlag = true + f.stop() } } } - f.pauseCond.Signal() - return retErr + + if err != nil { + if err == base.PauseErr { + f.pauseCh <- nil + err = nil + } else { + f.status = base.DownloadStatusError + } + } else { + f.status = base.DownloadStatusDone + } + f.doneCh <- err + return } func (f *Fetcher) stop() { @@ -262,12 +259,12 @@ func (f *Fetcher) stop() { } func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err error) { - httpReq, err := BuildRequest(f.res.Req) + httpReq, err := buildRequest(f.res.Req) if err != nil { return err } var ( - client = BuildClient() + client = buildClient() buf = make([]byte, 8192) ) // 重试5次 @@ -283,8 +280,6 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er chunk.Downloaded = 0 } err = func() error { - f.pauseCond.L.Lock() - defer f.pauseCond.L.Unlock() if f.status == base.DownloadStatusPause { return base.PauseErr } @@ -326,7 +321,7 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er } return false, nil }() - if err == nil || !retry { + if !retry { // 下载成功，跳出重试 break } @@ -339,7 +334,7 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er return } -func BuildClient() *http.Client { +func buildClient() *http.Client { // Cookie handle jar, _ := cookiejar.New(nil) return &http.Client{ @@ -348,7 +343,7 @@ func BuildClient() *http.Client { } } -func BuildRequest(req *base.Request) (*http.Request, error) { +func buildRequest(req *base.Request) (*http.Request, error) { url, err := url.Parse(req.URL) if err != nil { return nil, err diff --git a/download/http/fetcher_test.go b/download/http/fetcher_test.go index a3b8e7b5..10673473 100644 --- a/download/http/fetcher_test.go +++ b/download/http/fetcher_test.go @@ -265,11 +265,10 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { t.Fatal(err) } }() - time.Sleep(time.Millisecond * 200) + time.Sleep(time.Millisecond * 500) if err := fetcher.Pause(); err != nil { t.Fatal(err) } - time.Sleep(time.Millisecond * 200) if err := fetcher.Continue(); err != nil { t.Fatal(err) }