diff --git a/download/common/constants.go b/download/base/constants.go similarity index 97% rename from download/common/constants.go rename to download/base/constants.go index 9ce2f290..8ad8d9a5 100644 --- a/download/common/constants.go +++ b/download/base/constants.go @@ -1,4 +1,4 @@ -package common +package base import "errors" diff --git a/download/common/controller.go b/download/base/controller.go similarity index 72% rename from download/common/controller.go rename to download/base/controller.go index a77d15ba..7464c74a 100644 --- a/download/common/controller.go +++ b/download/base/controller.go @@ -1,4 +1,4 @@ -package common +package base import ( "golang.org/x/net/proxy" @@ -7,6 +7,14 @@ import ( "time" ) +type Controller interface { + Touch(name string, size int64) (file *os.File, err error) + Open(name string) (file *os.File, err error) + Write(name string, offset int64, buf []byte) (int, error) + Close(name string) error + ContextDialer() (proxy.Dialer, error) +} + // 下载请求 type Request struct { // 下载链接 @@ -41,22 +49,15 @@ type FileInfo struct { Size int64 } -type Process interface { - Start() error - Pause() error - Continue() error - Delete() error -} - -type Controller struct { +type DefaultController struct { Files map[string]*os.File } -func NewController() *Controller { - return &Controller{Files: make(map[string]*os.File)} +func NewController() *DefaultController { + return &DefaultController{Files: make(map[string]*os.File)} } -func (c *Controller) Touch(name string, size int64) (file *os.File, err error) { +func (c *DefaultController) Touch(name string, size int64) (file *os.File, err error) { file, err = os.Create(name) if size > 0 { err = os.Truncate(name, size) @@ -70,7 +71,7 @@ func (c *Controller) Touch(name string, size int64) (file *os.File, err error) { return } -func (c *Controller) Open(name string) (file *os.File, err error) { +func (c *DefaultController) Open(name string) (file *os.File, err error) { file, err = os.OpenFile(name, os.O_RDWR, os.ModePerm) if err == nil { c.Files[name] = file @@ -78,17 +79,17 @@ func (c *Controller) Open(name string) (file *os.File, err error) { return } -func (c *Controller) Write(name string, offset int64, buf []byte) (int, error) { +func (c *DefaultController) Write(name string, offset int64, buf []byte) (int, error) { return c.Files[name].WriteAt(buf, offset) } -func (c *Controller) Close(name string) error { +func (c *DefaultController) Close(name string) error { err := c.Files[name].Close() delete(c.Files, name) return err } -func (c *Controller) ContextDialer() (proxy.Dialer, error) { +func (c *DefaultController) ContextDialer() (proxy.Dialer, error) { // return proxy.SOCKS5("tpc", "127.0.0.1:9999", nil, nil) var dialer proxy.Dialer return &DialerWarp{dialer: dialer}, nil diff --git a/download/common/event.go b/download/base/event.go similarity index 74% rename from download/common/event.go rename to download/base/event.go index 57899950..9cf092b5 100644 --- a/download/common/event.go +++ b/download/base/event.go @@ -1,4 +1,4 @@ -package common +package base type Event interface { Status() Status diff --git a/download/base/fetcher.go b/download/base/fetcher.go new file mode 100644 index 00000000..ea111254 --- /dev/null +++ b/download/base/fetcher.go @@ -0,0 +1,22 @@ +package base + +// 对应协议的下载支持 +type Fetcher interface { + Init(ctl Controller) + // 解析请求 + Resolve(req *Request) (*Resource, error) + // 创建任务 + Create(res *Resource, opts *Options) error + Start() error + Pause() error + Continue() error + Delete() error +} + +type DefaultFetcher struct { + Ctl Controller +} + +func (f *DefaultFetcher) Init(ctl Controller) { + f.Ctl = ctl +} diff --git a/download/common/fetcher.go b/download/common/fetcher.go deleted file mode 100644 index 038228b0..00000000 --- a/download/common/fetcher.go +++ /dev/null @@ -1,29 +0,0 @@ -package common - -// 对应协议的下载支持 -type Fetcher interface { - BaseFetch - // 支持的协议列表 - Protocols() []string - // 解析请求 - Resolve(req *Request) (*Resource, error) - // 创建任务 - Create(res *Resource, opts *Options) (Process, error) -} - -type BaseFetch interface { - InitCtl(ctl *Controller) - GetCtl() *Controller -} - -type BaseFetcher struct { - ctl *Controller -} - -func (b *BaseFetcher) InitCtl(ctl *Controller) { - b.ctl = ctl -} - -func (b *BaseFetcher) GetCtl() *Controller { - return b.ctl -} diff --git a/download/downloader.go b/download/downloader.go index c4d45415..52eb0090 100644 --- a/download/downloader.go +++ b/download/downloader.go @@ -3,65 +3,69 @@ package download import ( "errors" "github.com/google/uuid" - "github.com/monkeyWie/gopeed-core/download/common" + "github.com/monkeyWie/gopeed-core/download/base" "net/url" "strings" ) type TaskInfo struct { ID string - Res *common.Resource - Opts *common.Options - Status common.Status - Files map[string]*common.FileInfo + Res *base.Resource + Opts *base.Options + Status base.Status + Files map[string]*base.FileInfo - process common.Process + fetcher base.Fetcher } +type FetcherBuilder func() (protocols []string, builder func() base.Fetcher) + type Downloader struct { - fetches map[string]common.Fetcher - tasks map[string]*TaskInfo - *common.Controller + fetchBuilders map[string]func() base.Fetcher + tasks map[string]*TaskInfo + *base.DefaultController } -func NewDownloader(fetchers ...common.Fetcher) *Downloader { - d := &Downloader{Controller: common.NewController()} - d.fetches = make(map[string]common.Fetcher) - for _, f := range fetchers { - for _, p := range f.Protocols() { - d.fetches[strings.ToUpper(p)] = f +func NewDownloader(fbs ...FetcherBuilder) *Downloader { + d := &Downloader{DefaultController: base.NewController()} + d.fetchBuilders = make(map[string]func() base.Fetcher) + for _, f := range fbs { + protocols, builder := f() + for _, p := range protocols { + d.fetchBuilders[strings.ToUpper(p)] = builder } - f.InitCtl(d.Controller) } d.tasks = make(map[string]*TaskInfo) return d } -func (d *Downloader) getFetcher(URL string) (common.Fetcher, error) { +func (d *Downloader) buildFetcher(URL string) (base.Fetcher, error) { url, err := url.Parse(URL) if err != nil { return nil, err } - if fetch, ok := d.fetches[strings.ToUpper(url.Scheme)]; ok { - return fetch, nil + if fetchBuilder, ok := d.fetchBuilders[strings.ToUpper(url.Scheme)]; ok { + fetcher := fetchBuilder() + fetcher.Init(d.DefaultController) + return fetcher, nil } return nil, errors.New("unsupported protocol") } -func (d *Downloader) Resolve(req *common.Request) (*common.Resource, error) { - fetcher, err := d.getFetcher(req.URL) +func (d *Downloader) Resolve(req *base.Request) (*base.Resource, error) { + fetcher, err := d.buildFetcher(req.URL) if err != nil { return nil, err } return fetcher.Resolve(req) } -func (d *Downloader) Create(res *common.Resource, opts *common.Options) error { - fetcher, err := d.getFetcher(res.Req.URL) +func (d *Downloader) Create(res *base.Resource, opts *base.Options) error { + fetcher, err := d.buildFetcher(res.Req.URL) if err != nil { return err } - process, err := fetcher.Create(res, opts) + err = fetcher.Create(res, opts) if err != nil { return err } @@ -70,12 +74,12 @@ func (d *Downloader) Create(res *common.Resource, opts *common.Options) error { ID: id, Res: res, Opts: opts, - Status: common.DownloadStatusReady, - process: process, + Status: base.DownloadStatusReady, + fetcher: fetcher, } - return process.Start() + return fetcher.Start() } func (d *Downloader) Pause(id string) { - d.tasks[id].process.Pause() + d.tasks[id].fetcher.Pause() } diff --git a/download/downloader_test.go b/download/downloader_test.go index 77a07776..8280bceb 100644 --- a/download/downloader_test.go +++ b/download/downloader_test.go @@ -2,14 +2,14 @@ package download import ( "fmt" - "github.com/monkeyWie/gopeed-core/download/common" + "github.com/monkeyWie/gopeed-core/download/base" "github.com/monkeyWie/gopeed-core/download/http" "testing" ) func TestDownloader_Resolve(t *testing.T) { downloader := buildDownloader() - res, err := downloader.Resolve(&common.Request{ + res, err := downloader.Resolve(&base.Request{ URL: "http://192.168.200.163:8088/docker-compose", }) if err != nil { @@ -20,13 +20,13 @@ func TestDownloader_Resolve(t *testing.T) { func TestDownloader_Create(t *testing.T) { downloader := buildDownloader() - res, err := downloader.Resolve(&common.Request{ + res, err := downloader.Resolve(&base.Request{ URL: "http://192.168.200.163:8088/docker-compose", }) if err != nil { t.Fatal(err) } - err = downloader.Create(res, &common.Options{ + err = downloader.Create(res, &base.Options{ Path: "E:\\test\\gopeed\\http", Connections: 4, }) @@ -36,5 +36,5 @@ func TestDownloader_Create(t *testing.T) { } func buildDownloader() *Downloader { - return NewDownloader(&http.Fetcher{BaseFetcher: &common.BaseFetcher{}}) + return NewDownloader(http.FetcherBuilder) } diff --git a/download/http/envent.go b/download/http/envent.go index 02ed52b8..8bc6b907 100644 --- a/download/http/envent.go +++ b/download/http/envent.go @@ -1,11 +1,11 @@ package http -import "github.com/monkeyWie/gopeed-core/download/common" +import "github.com/monkeyWie/gopeed-core/download/base" type Event struct { - status common.Status + status base.Status } -func (e *Event) Status() common.Status { +func (e *Event) Status() base.Status { return e.status } diff --git a/download/http/fetcher.go b/download/http/fetcher.go index ff32de70..5168b200 100644 --- a/download/http/fetcher.go +++ b/download/http/fetcher.go @@ -3,15 +3,18 @@ package http import ( "bytes" "fmt" - "github.com/monkeyWie/gopeed-core/download/common" + "github.com/monkeyWie/gopeed-core/download/base" "github.com/monkeyWie/gopeed-core/download/http/model" + "io" "io/ioutil" "mime" "net/http" "net/http/cookiejar" "net/url" + "path" + "path/filepath" "strconv" - "strings" + "sync" "time" ) @@ -29,41 +32,53 @@ func (re *RequestError) Error() string { } type Fetcher struct { - *common.BaseFetcher + *base.DefaultFetcher + + res *base.Resource + opts *base.Options + status base.Status + clients []*http.Response + chunks []*model.Chunk + + pauseCond *sync.Cond } func NewFetcher() *Fetcher { - return &Fetcher{BaseFetcher: &common.BaseFetcher{}} + return &Fetcher{} } -func (h *Fetcher) Protocols() []string { - return []string{"HTTP", "HTTPS"} +var protocols = []string{"HTTP", "HTTPS"} + +func FetcherBuilder() ([]string, func() base.Fetcher) { + return protocols, func() base.Fetcher { + return NewFetcher() + } } -func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { +func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { httpReq, err := BuildRequest(req) if err != nil { return nil, err } client := BuildClient() // 只访问一个字节，测试资源是否支持Range请求 - httpReq.Header.Set(common.HttpHeaderRange, fmt.Sprintf(common.HttpHeaderRangeFormat, 0, 0)) + httpReq.Header.Set(base.HttpHeaderRange, fmt.Sprintf(base.HttpHeaderRangeFormat, 0, 0)) httpResp, err := client.Do(httpReq) if err != nil { return nil, err } // 拿到响应头就关闭，不用加defer httpResp.Body.Close() - res := &common.Resource{ + res := &base.Resource{ Req: req, Range: false, - Files: []*common.FileInfo{}, + Files: []*base.FileInfo{}, } - if common.HttpCodePartialContent == httpResp.StatusCode { + if base.HttpCodePartialContent == httpResp.StatusCode { // 返回206响应码表示支持断点下载 res.Range = true // 解析资源大小: bytes 0-1000/1001 => 1001 - contentTotal := subLastSlash(httpResp.Header.Get(common.HttpHeaderContentRange)) + contentTotal := path.Base(httpResp.Header.Get(base.HttpHeaderContentRange)) if contentTotal != "" { parse, err := strconv.ParseInt(contentTotal, 10, 64) if err != nil { @@ -71,9 +86,9 @@ func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { } res.Size = parse } - } else if common.HttpCodeOK == httpResp.StatusCode { + } else if base.HttpCodeOK == httpResp.StatusCode { // 返回200响应码，不支持断点下载，通过Content-Length头获取文件大小，获取不到的话可能是chunked编码 - contentLength := httpResp.Header.Get(common.HttpHeaderContentLength) + contentLength := httpResp.Header.Get(base.HttpHeaderContentLength) if contentLength != "" { parse, err := strconv.ParseInt(contentLength, 10, 64) if err != nil { @@ -84,10 +99,10 @@ func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { } else { return nil, NewRequestError(httpResp.StatusCode, httpResp.Status) } - file := &common.FileInfo{ + file := &base.FileInfo{ Size: res.Size, } - contentDisposition := httpResp.Header.Get(common.HttpHeaderContentDisposition) + contentDisposition := httpResp.Header.Get(base.HttpHeaderContentDisposition) if contentDisposition != "" { _, params, _ := mime.ParseMediaType(contentDisposition) filename := params["filename"] @@ -97,11 +112,7 @@ func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { } // Get file name by URL if file.Name == "" { - parse, err := url.Parse(req.URL) - if err == nil { - // e.g. /files/test.txt => test.txt - file.Name = subLastSlash(parse.Path) - } + file.Name = path.Base(req.URL) } // unknown file name if file.Name == "" { @@ -111,22 +122,221 @@ func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { return res, nil } -func (h *Fetcher) Create(res *common.Resource, opts *common.Options) (common.Process, error) { +func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error { if opts.Connections != 1 && !res.Range { opts.Connections = 1 } - return NewProcess(h, res, opts), nil + f.res = res + f.opts = opts + f.status = base.DownloadStatusReady + f.pauseCond = sync.NewCond(&sync.Mutex{}) + return nil } -func subLastSlash(str string) string { - if str == "" { - return "" +func (f *Fetcher) Start() error { + // 创建文件 + name := f.name() + _, err := f.Ctl.Touch(name, f.res.Size) + if err != nil { + return err } - index := strings.LastIndex(str, "/") - if index != -1 { - return str[index+1:] + defer f.close() + f.status = base.DownloadStatusStart + if f.res.Range { + // 每个连接平均需要下载的分块大小 + chunkSize := f.res.Size / int64(f.opts.Connections) + f.chunks = make([]*model.Chunk, f.opts.Connections) + f.clients = make([]*http.Response, f.opts.Connections) + for i := 0; i < f.opts.Connections; i++ { + var ( + begin = chunkSize * int64(i) + end int64 + ) + if i == f.opts.Connections-1 { + // 最后一个分块需要保证把文件下载完 + end = f.res.Size + } else { + end = begin + chunkSize + } + chunk := model.NewChunk(begin, end) + f.chunks[i] = chunk + } + } else { + // 只支持单连接下载 + f.chunks = make([]*model.Chunk, 1) + f.clients = make([]*http.Response, 1) + f.chunks[0] = model.NewChunk(0, 0) } - return "" + return f.fetch() +} + +func (f *Fetcher) Pause() error { + f.pauseCond.L.Lock() + defer f.pauseCond.L.Unlock() + if base.DownloadStatusStart != f.status { + return nil + } + f.status = base.DownloadStatusPause + f.stop() + // 释放锁，并等待下载结束 + f.pauseCond.Wait() + return nil +} + +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 + } + + var name = f.name() + _, err := f.Ctl.Open(name) + if err != nil { + return err + } + defer f.Ctl.Close(name) + 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 + if filename == "" { + filename = f.res.Files[0].Name + } + return filepath.Join(f.opts.Path, filename) +} + +func (f *Fetcher) fetch() error { + 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]) + }(i) + } + + var retErr error + 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() + }() + } + } + } + f.pauseCond.Signal() + return retErr +} + +func (f *Fetcher) stop() { + if len(f.clients) > 0 { + for _, client := range f.clients { + if client != nil { + client.Body.Close() + } + } + } +} + +func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err error) { + httpReq, err := BuildRequest(f.res.Req) + if err != nil { + return err + } + var ( + client = BuildClient() + buf = make([]byte, 8192) + ) + // 重试5次 + for i := 0; i < 5; i++ { + var ( + resp *http.Response + retry bool + ) + if f.res.Range { + httpReq.Header.Set(base.HttpHeaderRange, + fmt.Sprintf(base.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End)) + } else { + chunk.Downloaded = 0 + } + err = func() error { + f.pauseCond.L.Lock() + defer f.pauseCond.L.Unlock() + if f.status == base.DownloadStatusPause { + return base.PauseErr + } + resp, err = client.Do(httpReq) + if err != nil { + return err + } + f.clients[index] = resp + if resp.StatusCode != base.HttpCodeOK && resp.StatusCode != base.HttpCodePartialContent { + err = NewRequestError(resp.StatusCode, resp.Status) + return err + } + return nil + }() + if err != nil { + if err == base.PauseErr { + return err + } + // 请求失败重试 + continue + } + retry, err = func() (bool, error) { + defer resp.Body.Close() + for { + n, err := resp.Body.Read(buf) + if n > 0 { + _, err := f.Ctl.Write(name, chunk.Begin+chunk.Downloaded, buf[:n]) + if err != nil { + return false, err + } + chunk.Downloaded += int64(n) + } + if err != nil { + if err != io.EOF { + return true, err + } + break + } + } + return false, nil + }() + if err == nil || !retry { + // 下载成功，跳出重试 + break + } + } + if err != nil { + f.chunks[index].Status = base.DownloadStatusError + } else { + f.chunks[index].Status = base.DownloadStatusDone + } + return } func BuildClient() *http.Client { @@ -138,7 +348,7 @@ func BuildClient() *http.Client { } } -func BuildRequest(req *common.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 941fa90d..a3b8e7b5 100644 --- a/download/http/fetcher_test.go +++ b/download/http/fetcher_test.go @@ -4,7 +4,7 @@ import ( "crypto/md5" "encoding/hex" "fmt" - "github.com/monkeyWie/gopeed-core/download/common" + "github.com/monkeyWie/gopeed-core/download/base" "io" "math/rand" "net" @@ -23,20 +23,20 @@ const downloadName = "download.data" const downloadFile = testDir + downloadName func TestFetcher_Resolve(t *testing.T) { - testResolve(startTestFileServer, &common.Resource{ + testResolve(startTestFileServer, &base.Resource{ Size: buildSize, Range: true, - Files: []*common.FileInfo{ + Files: []*base.FileInfo{ { Name: buildName, Size: buildSize, }, }, }, t) - testResolve(startTestChunkedServer, &common.Resource{ + testResolve(startTestChunkedServer, &base.Resource{ Size: 0, Range: false, - Files: []*common.FileInfo{ + Files: []*base.FileInfo{ { Name: buildName, Size: 0, @@ -45,11 +45,11 @@ func TestFetcher_Resolve(t *testing.T) { }, t) } -func testResolve(startTestServer func() net.Listener, want *common.Resource, t *testing.T) { +func testResolve(startTestServer func() net.Listener, want *base.Resource, t *testing.T) { listener := startTestServer() defer listener.Close() fetcher := NewFetcher() - res, err := fetcher.Resolve(&common.Request{ + res, err := fetcher.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/" + buildName, }) if err != nil { @@ -88,7 +88,7 @@ func TestFetcher_DownloadChunked(t *testing.T) { listener := startTestChunkedServer() defer listener.Close() // chunked编码下载 - //downloadNormal(listener, 1, t) + downloadNormal(listener, 1, t) downloadContinue(listener, 1, t) } @@ -223,16 +223,16 @@ func ifExistAndRemove(name string) error { return nil } -func downloadReady(listener net.Listener, connections int, t *testing.T) common.Process { +func downloadReady(listener net.Listener, connections int, t *testing.T) base.Fetcher { fetcher := NewFetcher() - fetcher.InitCtl(common.NewController()) - res, err := fetcher.Resolve(&common.Request{ + fetcher.Init(base.NewController()) + res, err := fetcher.Resolve(&base.Request{ URL: "http://" + listener.Addr().String() + "/" + buildName, }) if err != nil { t.Fatal(err) } - process, err := fetcher.Create(res, &common.Options{ + err = fetcher.Create(res, &base.Options{ Name: downloadName, Path: testDir, Connections: connections, @@ -240,13 +240,13 @@ func downloadReady(listener net.Listener, connections int, t *testing.T) common. if err != nil { t.Fatal(err) } - return process + return fetcher } func downloadNormal(listener net.Listener, connections int, t *testing.T) { - process := downloadReady(listener, connections, t) - err := process.Start() + fetcher := downloadReady(listener, connections, t) + err := fetcher.Start() if err != nil { t.Fatal(err) } @@ -258,19 +258,19 @@ func downloadNormal(listener net.Listener, connections int, t *testing.T) { } func downloadContinue(listener net.Listener, connections int, t *testing.T) { - process := downloadReady(listener, connections, t) + fetcher := downloadReady(listener, connections, t) go func() { - err := process.Start() - if err != nil && err != common.PauseErr { + err := fetcher.Start() + if err != nil && err != base.PauseErr { t.Fatal(err) } }() time.Sleep(time.Millisecond * 200) - if err := process.Pause(); err != nil { + if err := fetcher.Pause(); err != nil { t.Fatal(err) } time.Sleep(time.Millisecond * 200) - if err := process.Continue(); err != nil { + if err := fetcher.Continue(); err != nil { t.Fatal(err) } want := fileMd5(buildFile) @@ -281,8 +281,8 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { } func downloadError(listener net.Listener, connections int, t *testing.T) { - process := downloadReady(listener, connections, t) - err := process.Start() + fetcher := downloadReady(listener, connections, t) + err := fetcher.Start() if err == nil { t.Errorf("Download error = %v, want %v", err, nil) } diff --git a/download/http/model/chunk.go b/download/http/model/chunk.go index c07df4dd..bd7de418 100644 --- a/download/http/model/chunk.go +++ b/download/http/model/chunk.go @@ -1,11 +1,11 @@ package model import ( - "github.com/monkeyWie/gopeed-core/download/common" + "github.com/monkeyWie/gopeed-core/download/base" ) type Chunk struct { - Status common.Status + Status base.Status Begin int64 End int64 Downloaded int64 @@ -13,7 +13,7 @@ type Chunk struct { func NewChunk(begin int64, end int64) *Chunk { return &Chunk{ - Status: common.DownloadStatusReady, + Status: base.DownloadStatusReady, Begin: begin, End: end, } diff --git a/download/http/process.go b/download/http/process.go deleted file mode 100644 index bfc6f1e1..00000000 --- a/download/http/process.go +++ /dev/null @@ -1,258 +0,0 @@ -package http - -import ( - "fmt" - "github.com/monkeyWie/gopeed-core/download/common" - "github.com/monkeyWie/gopeed-core/download/http/model" - "io" - "net/http" - "path/filepath" - "sync" -) - -type Process struct { - fetcher *Fetcher - - res *common.Resource - opts *common.Options - status common.Status - clients []*http.Response - chunks []*model.Chunk - - pauseCond *sync.Cond - eventCh chan common.Event -} - -func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *Process { - return &Process{ - fetcher: fetcher, - res: res, - opts: opts, - status: common.DownloadStatusReady, - - pauseCond: sync.NewCond(&sync.Mutex{}), - } -} - -func (p *Process) Listen() <-chan common.Event { - p.eventCh = make(chan common.Event) - return p.eventCh -} - -func (p *Process) Start() error { - ctl := p.fetcher.GetCtl() - // 创建文件 - name := p.name() - _, err := ctl.Touch(name, p.res.Size) - if err != nil { - return err - } - defer p.close() - p.status = common.DownloadStatusStart - if p.res.Range { - // 每个连接平均需要下载的分块大小 - chunkSize := p.res.Size / int64(p.opts.Connections) - p.chunks = make([]*model.Chunk, p.opts.Connections) - p.clients = make([]*http.Response, p.opts.Connections) - for i := 0; i < p.opts.Connections; i++ { - var ( - begin = chunkSize * int64(i) - end int64 - ) - if i == p.opts.Connections-1 { - // 最后一个分块需要保证把文件下载完 - end = p.res.Size - } else { - end = begin + chunkSize - } - chunk := model.NewChunk(begin, end) - p.chunks[i] = chunk - } - } else { - // 只支持单连接下载 - p.chunks = make([]*model.Chunk, 1) - p.clients = make([]*http.Response, 1) - p.chunks[0] = model.NewChunk(0, 0) - } - p.dispatch(common.DownloadStatusStart) - return p.fetch() -} - -func (p *Process) Pause() error { - p.pauseCond.L.Lock() - defer p.pauseCond.L.Unlock() - if common.DownloadStatusStart != p.status { - return nil - } - p.status = common.DownloadStatusPause - p.stop() - // 释放锁，并等待下载结束 - p.pauseCond.Wait() - return nil -} - -func (p *Process) Continue() error { - if func() bool { - p.pauseCond.L.Lock() - defer p.pauseCond.L.Unlock() - if common.DownloadStatusPause != p.status && common.DownloadStatusError != p.status { - return true - } - p.status = common.DownloadStatusStart - return false - }() { - return nil - } - - var ( - ctl = p.fetcher.GetCtl() - name = p.name() - ) - _, err := ctl.Open(name) - if err != nil { - return err - } - defer ctl.Close(name) - return p.fetch() -} - -func (p *Process) Delete() error { - panic("implement me") -} - -func (p *Process) close() error { - return p.fetcher.GetCtl().Close(p.name()) -} - -func (p *Process) name() string { - // 创建文件 - var filename = p.opts.Name - if filename == "" { - filename = p.res.Files[0].Name - } - return filepath.Join(p.opts.Path, filename) -} - -func (p *Process) fetch() error { - errCh := make(chan error, p.opts.Connections) - defer close(errCh) - - for i := 0; i < p.opts.Connections; i++ { - go func(i int) { - errCh <- p.fetchChunk(i, p.name(), p.chunks[i]) - }(i) - } - - var retErr error - for i := 0; i < p.opts.Connections; i++ { - err := <-errCh - if err != nil && retErr == nil { - retErr = err - // 有一个连接失败就立即终止下载 - if retErr != common.PauseErr { - func() { - p.pauseCond.L.Lock() - defer p.pauseCond.L.Unlock() - p.stop() - }() - } - } - } - p.pauseCond.Signal() - return retErr -} - -func (p *Process) stop() { - if len(p.clients) > 0 { - for _, client := range p.clients { - fmt.Println("client close") - if client != nil { - client.Body.Close() - } - } - } -} - -func (p *Process) fetchChunk(index int, name string, chunk *model.Chunk) (err error) { - httpReq, err := BuildRequest(p.res.Req) - if err != nil { - return err - } - var ( - client = BuildClient() - buf = make([]byte, 8192) - ) - // 重试5次 - for i := 0; i < 5; i++ { - var ( - resp *http.Response - retry bool - ) - if p.res.Range { - httpReq.Header.Set(common.HttpHeaderRange, - fmt.Sprintf(common.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End)) - } else { - chunk.Downloaded = 0 - } - err = func() error { - p.pauseCond.L.Lock() - defer p.pauseCond.L.Unlock() - if p.status == common.DownloadStatusPause { - return common.PauseErr - } - resp, err = client.Do(httpReq) - if err != nil { - return err - } - p.clients[index] = resp - if resp.StatusCode != common.HttpCodeOK && resp.StatusCode != common.HttpCodePartialContent { - err = NewRequestError(resp.StatusCode, resp.Status) - return err - } - return nil - }() - if err != nil { - if err == common.PauseErr { - return err - } - // 请求失败重试 - continue - } - retry, err = func() (bool, error) { - defer resp.Body.Close() - for { - n, err := resp.Body.Read(buf) - if n > 0 { - _, err := p.fetcher.GetCtl().Write(name, chunk.Begin+chunk.Downloaded, buf[:n]) - if err != nil { - return false, err - } - chunk.Downloaded += int64(n) - } - if err != nil { - if err != io.EOF { - return true, err - } - break - } - } - return false, nil - }() - if err == nil || !retry { - // 下载成功，跳出重试 - break - } - } - if err != nil { - p.chunks[index].Status = common.DownloadStatusError - } else { - p.chunks[index].Status = common.DownloadStatusDone - } - return -} - -func (p *Process) dispatch(status common.Status) { - if p.eventCh != nil { - p.eventCh <- &Event{status: status} - } -}