From d4e1fd23a1550eb6e1e69907bd7b7069e84c5550 Mon Sep 17 00:00:00 2001 From: liwei Date: Thu, 20 Aug 2020 22:19:07 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B7=91=E9=80=9A=E5=8D=95=E5=85=83=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- download/http/fetcher_test.go | 23 +++++++++------ download/http/process.go | 54 +++++++++++++++-------------------- 2 files changed, 38 insertions(+), 39 deletions(-) diff --git a/download/http/fetcher_test.go b/download/http/fetcher_test.go index 55c635b2..c95402f3 100644 --- a/download/http/fetcher_test.go +++ b/download/http/fetcher_test.go @@ -75,17 +75,17 @@ func TestFetcher_DownloadContinue(t *testing.T) { listener := startTestFileServer() defer listener.Close() // 暂停继续 - // downloadContinue(listener, 1, t) + downloadContinue(listener, 1, t) downloadContinue(listener, 5, t) - // downloadContinue(listener, 8, t) - // downloadContinue(listener, 16, t) + downloadContinue(listener, 8, t) + downloadContinue(listener, 16, t) } func TestFetcher_DownloadChunked(t *testing.T) { listener := startTestChunkedServer() defer listener.Close() // chunked编码下载 - downloadNormal(listener, 1, t) + //downloadNormal(listener, 1, t) downloadContinue(listener, 1, t) } @@ -181,16 +181,23 @@ type delFileListener struct { func (c *delFileListener) Close() error { defer func() { c.File.Close() - if err := os.Remove(c.File.Name()); err != nil { + if err := ifExistAndRemove(c.File.Name()); err != nil { fmt.Println(err) } - if err := os.Remove(downloadFile); err != nil { + if err := ifExistAndRemove(downloadFile); err != nil { fmt.Println(err) } }() return c.Listener.Close() } +func ifExistAndRemove(name string) error { + if _, err := os.Stat(name); !os.IsNotExist(err) { + return os.Remove(name) + } + return nil +} + func downloadReady(listener net.Listener, connections int, t *testing.T) common.Process { fetcher := NewFetcher() fetcher.InitCtl(common.NewController()) @@ -237,7 +244,7 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { if err := process.Pause(); err != nil { t.Fatal(err) } - /*time.Sleep(time.Millisecond * 200) + time.Sleep(time.Millisecond * 200) if err := process.Continue(); err != nil { t.Fatal(err) } @@ -245,7 +252,7 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { got := fileMd5(downloadFile) if want != got { t.Errorf("Download error = %v, want %v", got, want) - }*/ + } } func fileMd5(filePath string) string { diff --git a/download/http/process.go b/download/http/process.go index a18d3e62..4d36ca0a 100644 --- a/download/http/process.go +++ b/download/http/process.go @@ -19,7 +19,7 @@ type Process struct { clients []*http.Response chunks []*model.Chunk - pauseLock *sync.Mutex + pauseCond *sync.Cond } func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *Process { @@ -29,7 +29,7 @@ func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *P opts: opts, status: common.DownloadStatusReady, - pauseLock: &sync.Mutex{}, + pauseCond: sync.NewCond(&sync.Mutex{}), } } @@ -69,7 +69,7 @@ func (p *Process) Start() error { p.chunks = make([]*model.Chunk, 1) p.clients = make([]*http.Response, 1) p.chunks[0] = model.NewChunk(0, 0) - fetchErr = p.fetchChunk(0, name, p.chunks[0]) + fetchErr = p.fetch() } if fetchErr != nil && fetchErr != common.PauseErr { p.Pause() @@ -78,25 +78,26 @@ func (p *Process) Start() error { } func (p *Process) Pause() error { - p.pauseLock.Lock() - defer p.pauseLock.Unlock() + p.pauseCond.L.Lock() + defer p.pauseCond.L.Unlock() if common.DownloadStatusStart != p.status { return nil } - fmt.Println("============Pause============") p.status = common.DownloadStatusPause if len(p.clients) > 0 { for _, client := range p.clients { client.Body.Close() } } + // 释放锁，并等待下载结束 + p.pauseCond.Wait() return nil } func (p *Process) Continue() error { if func() bool { - p.pauseLock.Lock() - defer p.pauseLock.Unlock() + p.pauseCond.L.Lock() + defer p.pauseCond.L.Unlock() if common.DownloadStatusPause != p.status && common.DownloadStatusError != p.status { return true } @@ -105,7 +106,6 @@ func (p *Process) Continue() error { }() { return nil } - fmt.Println("============Continue============") var ( ctl = p.fetcher.GetCtl() @@ -137,35 +137,28 @@ func (p *Process) name() string { } func (p *Process) fetch() error { - errCh := make(chan error) + errCh := make(chan error, p.opts.Connections) defer close(errCh) - var done bool - var lock sync.Mutex - for i := 0; i < p.opts.Connections; i++ { go func(i int) { - err := p.fetchChunk(i, p.name(), p.chunks[i]) - lock.Lock() - if !done { - errCh <- err - } - lock.Unlock() + errCh <- p.fetchChunk(i, p.name(), p.chunks[i]) }(i) } - var err error + var retErr error for i := 0; i < p.opts.Connections; i++ { - err = <-errCh - if err != nil { - // 有一个连接下载失败，立即返回 - lock.Lock() - done = true - lock.Unlock() - break + err := <-errCh + if retErr != nil { + retErr = err + // 有一个连接失败就立即终止下载 + if retErr != common.PauseErr { + p.Pause() + } } } - return err + p.pauseCond.Signal() + return retErr } func (p *Process) fetchChunk(index int, name string, chunk *model.Chunk) (err error) { @@ -186,13 +179,12 @@ func (p *Process) fetchChunk(index int, name string, chunk *model.Chunk) (err er if p.res.Range { httpReq.Header.Set(common.HttpHeaderRange, fmt.Sprintf(common.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End)) - fmt.Printf(common.HttpHeaderRangeFormat+"\n", chunk.Begin+chunk.Downloaded, chunk.End) } else { chunk.Downloaded = 0 } err = func() error { - p.pauseLock.Lock() - defer p.pauseLock.Unlock() + p.pauseCond.L.Lock() + defer p.pauseCond.L.Unlock() if p.status == common.DownloadStatusPause { return common.PauseErr }