From dd47b52952e5b78d9135447a1886ba8ded69158c Mon Sep 17 00:00:00 2001 From: liwei Date: Wed, 5 Aug 2020 22:41:50 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95=E5=9F=BA?= =?UTF-8?q?=E7=A1=80=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- download/http/fetcher_test.go | 49 +++++++++++++++++- download/http/process.go | 97 +++++++++++++++++++++-------------- 2 files changed, 106 insertions(+), 40 deletions(-) diff --git a/download/http/fetcher_test.go b/download/http/fetcher_test.go index a05bc86f..1234ad9a 100644 --- a/download/http/fetcher_test.go +++ b/download/http/fetcher_test.go @@ -7,17 +7,62 @@ import ( "golang.org/x/net/proxy" "io" "io/ioutil" + "math/rand" + "net" "net/http" "os" "testing" "time" ) +func startMockServer() net.Listener { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + panic(err) + } + file, err := os.Create("./test.data") + if err != nil { + panic(err) + } + buf := make([]byte, 1024) + for i := 0; i < 20*1024; i++ { + _, err := rand.Read(buf) + if err != nil { + panic(err) + } + file.WriteAt(buf, int64(i*1024)) + } + + go func() { + http.Serve(listener, http.FileServer(http.Dir("./"))) + }() + return &delFileListener{ + File: file, + Listener: listener, + } +} + +type delFileListener struct { + *os.File + net.Listener +} + +func (c *delFileListener) Close() error { + c.File.Close() + err := os.Remove(c.File.Name()) + if err != nil { + fmt.Println(err) + return err + } + return c.Listener.Close() +} + func TestFetcher_Resolve(t *testing.T) { + listener := startMockServer() + defer listener.Close() fetcher := &Fetcher{} resolve, err := fetcher.Resolve(&common.Request{ - // URL: "https://github.com/monkeyWie/github-actions-demo/releases/download/v1.0.6/simple-v1.0.6-windows-x64.zip", - URL: "http://192.168.200.163:8088/go1.12.7.linux-amd64.tar.gz", + URL: "http://" + listener.Addr().String() + "/test.data", }) if err != nil { t.Fatal(err) diff --git a/download/http/process.go b/download/http/process.go index 6adcb273..cf6a46fd 100644 --- a/download/http/process.go +++ b/download/http/process.go @@ -31,11 +31,7 @@ func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *P func (p *Process) Start() error { ctl := p.fetcher.GetCtl() // 创建文件 - var filename = p.opts.Name - if filename == "" { - filename = p.res.Files[0].Name - } - name := filepath.Join(p.opts.Path, filename) + name := p.name() err := ctl.Touch(name, p.res.Size) if err != nil { return err @@ -50,31 +46,23 @@ func (p *Process) Start() error { p.chunks = make([]*model.Chunk, p.opts.Connections) p.clients = make([]*http.Response, p.opts.Connections) for i := 0; i < p.opts.Connections; i++ { - begin := chunkSize * int64(i) - end := begin + chunkSize + var ( + begin = chunkSize * int64(i) + end int64 + ) if i == p.opts.Connections-1 { - // 最后一个连接下载的分块大小需要特殊计算，保证把文件下载完 - chunkSize = end + p.res.Size%int64(p.opts.Connections) + // 最后一个分块需要保证把文件下载完 + end = p.res.Size + } else { + end = begin + chunkSize } - p.chunks[i] = model.NewChunk(begin, end) + chunk := model.NewChunk(begin, end) + p.chunks[i] = chunk go func(i int) { - errCh <- p.doFetch(i, name, begin, end) + errCh <- p.fetchChunk(i, name, chunk) }(i) } - var failErr error - for i := 0; i < p.opts.Connections; i++ { - err := <-errCh - if failErr == nil { - // 有一个连接下载失败，立即停止下载 - failErr = err - if failErr != nil { - p.Pause() - } - } - } - if failErr != nil { - return failErr - } + return p.wait(errCh) } return nil } @@ -89,49 +77,82 @@ func (p *Process) Pause() error { } func (p *Process) Continue() error { - return nil + var ( + name = p.name() + errCh = make(chan error) + ) + for i := 0; i < p.opts.Connections; i++ { + go func(i int) { + errCh <- p.fetchChunk(i, name, p.chunks[i]) + }(i) + } + return p.wait(errCh) } func (p *Process) Delete() error { panic("implement me") } -func (p *Process) doFetch(index int, filename string, begin, end int64) (err error) { +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) wait(errCh <-chan error) error { + var failErr error + for i := 0; i < p.opts.Connections; i++ { + err := <-errCh + if failErr == nil { + // 有一个连接下载失败，立即停止下载 + failErr = err + if failErr != nil { + p.Pause() + } + } + } + if failErr != nil { + return failErr + } + return nil +} + +func (p *Process) fetchChunk(index int, name string, chunk *model.Chunk) (err error) { httpReq, err := BuildRequest(p.res.Req) if err != nil { return err } - httpReq.Header.Set(common.HttpHeaderRange, fmt.Sprintf(common.HttpHeaderRangeFormat, begin, end)) - client := BuildClient() - + var ( + client = BuildClient() + buf = make([]byte, 8192) + ) // 重试5次 for i := 0; i < 5; i++ { var ( resp *http.Response retry bool ) + httpReq.Header.Set(common.HttpHeaderRange, + fmt.Sprintf(common.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End)) resp, err = client.Do(httpReq) if err != nil { // 连接失败，直接重试 continue } - p.chunks[index].Downloaded = 0 p.clients[index] = resp retry, err = func() (bool, error) { defer resp.Body.Close() - var ( - buf = make([]byte, 8192) - offset = begin - ) for { n, err := resp.Body.Read(buf) if n > 0 { - len, err := p.fetcher.GetCtl().Write(filename, offset, buf[:n]) + _, err := p.fetcher.GetCtl().Write(name, chunk.Begin+chunk.Downloaded, buf[:n]) if err != nil { return false, err } - offset += int64(len) - p.chunks[index].Downloaded = p.chunks[i].Downloaded + int64(len) + chunk.Downloaded += int64(n) } if err != nil { if err != io.EOF {