diff --git a/download/base/constants.go b/download/base/constants.go index 8ad8d9a5..6059a3e7 100644 --- a/download/base/constants.go +++ b/download/base/constants.go @@ -25,6 +25,5 @@ const ( ) var ( - PauseErr = errors.New("pause") DeleteErr = errors.New("delete") ) diff --git a/download/base/fetcher.go b/download/base/fetcher.go index 20cf334a..b4632352 100644 --- a/download/base/fetcher.go +++ b/download/base/fetcher.go @@ -2,7 +2,7 @@ package base // 对应协议的下载支持 type Fetcher interface { - Init(ctl Controller) + Setup(ctl Controller) (err <-chan error) // 解析请求 Resolve(req *Request) (res *Resource, err error) // 创建任务 @@ -13,9 +13,12 @@ type Fetcher interface { } type DefaultFetcher struct { - Ctl Controller + Ctl Controller + DoneCh chan error } -func (f *DefaultFetcher) Init(ctl Controller) { +func (f *DefaultFetcher) Setup(ctl Controller) (doneCh <-chan error) { f.Ctl = ctl + f.DoneCh = make(chan error, 1) + return f.DoneCh } diff --git a/download/downloader.go b/download/downloader.go index bcab7fa3..de3d7a1c 100644 --- a/download/downloader.go +++ b/download/downloader.go @@ -49,7 +49,7 @@ func (d *Downloader) buildFetcher(URL string) (base.Fetcher, error) { } if fetchBuilder, ok := d.fetchBuilders[strings.ToUpper(url.Scheme)]; ok { fetcher := fetchBuilder() - fetcher.Init(d.DefaultController) + fetcher.Setup(d.DefaultController) return fetcher, nil } return nil, errors.New("unsupported protocol") diff --git a/download/http/fetcher.go b/download/http/fetcher.go index 96d80679..b53ec8e8 100644 --- a/download/http/fetcher.go +++ b/download/http/fetcher.go @@ -40,14 +40,12 @@ type Fetcher struct { chunks []*model.Chunk pauseCh chan interface{} - doneCh chan error } func NewFetcher() *Fetcher { return &Fetcher{ DefaultFetcher: new(base.DefaultFetcher), pauseCh: make(chan interface{}), - doneCh: make(chan error), } } @@ -114,11 +112,11 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { file.Name = filename } } - // Get file name by URL + // Get file filename by URL if file.Name == "" { file.Name = path.Base(req.URL) } - // unknown file name + // unknown file filename if file.Name == "" { file.Name = "unknown" } @@ -138,12 +136,11 @@ func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error { func (f *Fetcher) Start() (err error) { // 创建文件 - name := f.name() + name := f.filename() _, err = f.Ctl.Touch(name, f.res.Size) if err != nil { return err } - defer f.Ctl.Close(name) f.status = base.DownloadStatusStart if f.res.Range { // 每个连接平均需要下载的分块大小 @@ -157,9 +154,9 @@ func (f *Fetcher) Start() (err error) { ) if i == f.opts.Connections-1 { // 最后一个分块需要保证把文件下载完 - end = f.res.Size + end = f.res.Size - 1 } else { - end = begin + chunkSize + end = begin + chunkSize - 1 } chunk := model.NewChunk(begin, end) f.chunks[i] = chunk @@ -170,8 +167,8 @@ func (f *Fetcher) Start() (err error) { f.clients = make([]*http.Response, 1) f.chunks[0] = model.NewChunk(0, 0) } - - return f.fetch() + f.fetch() + return } func (f *Fetcher) Pause() (err error) { @@ -181,7 +178,6 @@ func (f *Fetcher) Pause() (err error) { f.status = base.DownloadStatusPause f.stop() <-f.pauseCh - f.Ctl.Close(f.name()) return } @@ -190,16 +186,16 @@ func (f *Fetcher) Continue() (err error) { return } f.status = base.DownloadStatusStart - var name = f.name() + var name = f.filename() _, err = f.Ctl.Open(name) if err != nil { return err } - defer f.Ctl.Close(name) - return f.fetch() + f.fetch() + return } -func (f *Fetcher) name() string { +func (f *Fetcher) filename() string { // 创建文件 var filename = f.opts.Name if filename == "" { @@ -208,44 +204,42 @@ func (f *Fetcher) name() string { return filepath.Join(f.opts.Path, filename) } -func (f *Fetcher) fetch() (err error) { +func (f *Fetcher) fetch() { errCh := make(chan error, f.opts.Connections) - defer close(errCh) for i := 0; i < f.opts.Connections; i++ { go func(i int) { - errCh <- f.fetchChunk(i, f.name(), f.chunks[i]) + errCh <- f.fetchChunk(i, f.filename(), f.chunks[i]) }(i) } - stopFlag := false - for i := 0; i < f.opts.Connections; i++ { - fetchErr := <-errCh - if fetchErr != nil && !stopFlag { - // 确保如果有暂停操作返回暂停 - if err == nil || err != base.PauseErr { - err = fetchErr - } - if fetchErr != base.PauseErr { + go func() { + var err error + stopFlag := false + for i := 0; i < f.opts.Connections; i++ { + fetchErr := <-errCh + if fetchErr != nil && !stopFlag { // 有一个连接失败就立即终止下载 + err = fetchErr stopFlag = true f.stop() } } - } - if err != nil { - if err == base.PauseErr { + // 下载停止，关闭文件句柄 + f.Ctl.Close(f.filename()) + + if f.status == base.DownloadStatusPause { f.pauseCh <- nil - err = nil } else { - f.status = base.DownloadStatusError + if err != nil { + f.status = base.DownloadStatusError + } else { + f.status = base.DownloadStatusDone + } + f.DoneCh <- err } - } else { - f.status = base.DownloadStatusDone - } - f.doneCh <- err - return + }() } func (f *Fetcher) stop() { @@ -269,6 +263,14 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er ) // 重试5次 for i := 0; i < 5; i++ { + // 如果下载完成直接返回 + if chunk.Status == base.DownloadStatusDone { + return + } + // 如果已暂停直接跳出 + if f.status == base.DownloadStatusPause { + break + } var ( resp *http.Response retry bool @@ -280,9 +282,6 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er chunk.Downloaded = 0 } err = func() error { - if f.status == base.DownloadStatusPause { - return base.PauseErr - } resp, err = client.Do(httpReq) if err != nil { return err @@ -295,9 +294,6 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er return nil }() if err != nil { - if err == base.PauseErr { - return err - } // 请求失败重试 continue } @@ -326,10 +322,17 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er break } } - if err != nil { - f.chunks[index].Status = base.DownloadStatusError + + if f.status == base.DownloadStatusPause { + chunk.Status = base.DownloadStatusPause + } else if chunk.Downloaded >= chunk.End-chunk.Begin+1 { + chunk.Status = base.DownloadStatusDone } else { - f.chunks[index].Status = base.DownloadStatusDone + if err != nil { + chunk.Status = base.DownloadStatusError + } else { + chunk.Status = base.DownloadStatusDone + } } return } diff --git a/download/http/fetcher_test.go b/download/http/fetcher_test.go index 10673473..a04c814e 100644 --- a/download/http/fetcher_test.go +++ b/download/http/fetcher_test.go @@ -76,11 +76,8 @@ func TestFetcher_DownloadContinue(t *testing.T) { defer listener.Close() // 暂停继续 downloadContinue(listener, 1, t) - fmt.Println("d1") downloadContinue(listener, 5, t) - fmt.Println("d5") downloadContinue(listener, 8, t) - fmt.Println("d8") downloadContinue(listener, 16, t) } @@ -223,9 +220,9 @@ func ifExistAndRemove(name string) error { return nil } -func downloadReady(listener net.Listener, connections int, t *testing.T) base.Fetcher { +func downloadReady(listener net.Listener, connections int, t *testing.T) (base.Fetcher, <-chan error) { fetcher := NewFetcher() - fetcher.Init(base.NewController()) + doneCh := fetcher.Setup(base.NewController()) res, err := fetcher.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/" + buildName, }) @@ -240,16 +237,20 @@ func downloadReady(listener net.Listener, connections int, t *testing.T) base.Fe if err != nil { t.Fatal(err) } - return fetcher + return fetcher, doneCh } func downloadNormal(listener net.Listener, connections int, t *testing.T) { - fetcher := downloadReady(listener, connections, t) + fetcher, doneCh := downloadReady(listener, connections, t) err := fetcher.Start() if err != nil { t.Fatal(err) } + err = <-doneCh + if err != nil { + t.Fatal(err) + } want := fileMd5(buildFile) got := fileMd5(downloadFile) if want != got { @@ -258,13 +259,11 @@ func downloadNormal(listener net.Listener, connections int, t *testing.T) { } func downloadContinue(listener net.Listener, connections int, t *testing.T) { - fetcher := downloadReady(listener, connections, t) - go func() { - err := fetcher.Start() - if err != nil && err != base.PauseErr { - t.Fatal(err) - } - }() + fetcher, doneCh := downloadReady(listener, connections, t) + err := fetcher.Start() + if err != nil { + t.Fatal(err) + } time.Sleep(time.Millisecond * 500) if err := fetcher.Pause(); err != nil { t.Fatal(err) @@ -272,6 +271,10 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { if err := fetcher.Continue(); err != nil { t.Fatal(err) } + err = <-doneCh + if err != nil { + t.Fatal(err) + } want := fileMd5(buildFile) got := fileMd5(downloadFile) if want != got { @@ -280,8 +283,12 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { } func downloadError(listener net.Listener, connections int, t *testing.T) { - fetcher := downloadReady(listener, connections, t) + fetcher, doneCh := downloadReady(listener, connections, t) err := fetcher.Start() + if err != nil { + t.Fatal(err) + } + err = <-doneCh if err == nil { t.Errorf("Download error = %v, want %v", err, nil) }