diff --git a/download/common/controller.go b/download/common/controller.go index fe3e3d75..b02b29f9 100644 --- a/download/common/controller.go +++ b/download/common/controller.go @@ -36,10 +36,9 @@ type Options struct { } type FileInfo struct { - Name string - Path string - Size int64 - Downloaded int64 + Name string + Path string + Size int64 } type Process interface { @@ -53,25 +52,42 @@ type Controller struct { Files map[string]*os.File } -func (c *Controller) Touch(name string, size int64) error { +func NewController() *Controller { + return &Controller{Files: make(map[string]*os.File)} +} + +func (c *Controller) Touch(name string, size int64) (file *os.File, err error) { if size > 0 { - return os.Truncate(name, size) + err := os.Truncate(name, size) + if err != nil { + return nil, err + } + file, err = os.OpenFile(name, os.O_RDWR, 0666) + } else { + file, err = os.Create(name) } - _, err := os.Create(name) - return err + if err == nil { + c.Files[name] = file + } + return +} + +func (c *Controller) Open(name string) (file *os.File, err error) { + file, err = os.OpenFile(name, os.O_RDWR, 0666) + if err == nil { + c.Files[name] = file + } + return } func (c *Controller) Write(name string, offset int64, buf []byte) (int, error) { - file, ok := c.Files[name] - if !ok { - var err error - file, err = os.OpenFile(name, os.O_RDWR, 0666) - if err != nil { - return 0, err - } - c.Files[name] = file - } - return file.WriteAt(buf, offset) + return c.Files[name].WriteAt(buf, offset) +} + +func (c *Controller) Close(name string) error { + err := c.Files[name].Close() + delete(c.Files, name) + return err } func (c *Controller) ContextDialer() (proxy.Dialer, error) { diff --git a/download/downloader.go b/download/downloader.go index d6b7a688..92f83403 100644 --- a/download/downloader.go +++ b/download/downloader.go @@ -5,7 +5,6 @@ import ( "github.com/google/uuid" "github.com/monkeyWie/gopeed/download/common" "net/url" - "os" "strings" ) @@ -26,7 +25,7 @@ type Downloader struct { } func NewDownloader(fetchers ...common.Fetcher) *Downloader { - d := &Downloader{Controller: &common.Controller{Files: make(map[string]*os.File)}} + d := &Downloader{Controller: common.NewController()} d.fetches = make(map[string]common.Fetcher) for _, f := range fetchers { for _, p := range f.Protocols() { diff --git a/download/http/fetcher.go b/download/http/fetcher.go index d84745df..34c2588b 100644 --- a/download/http/fetcher.go +++ b/download/http/fetcher.go @@ -2,7 +2,6 @@ package http import ( "bytes" - "errors" "fmt" "github.com/monkeyWie/gopeed/download/common" "github.com/monkeyWie/gopeed/download/http/model" @@ -16,10 +15,27 @@ import ( "time" ) +type RequestError struct { + Code int + Msg string +} + +func NewRequestError(code int, msg string) *RequestError { + return &RequestError{Code: code, Msg: msg} +} + +func (re *RequestError) Error() string { + return fmt.Sprintf("http request fail,code:%d", re.Code) +} + type Fetcher struct { *common.BaseFetcher } +func NewFetcher() *Fetcher { + return &Fetcher{BaseFetcher: &common.BaseFetcher{}} +} + func (h *Fetcher) Protocols() []string { return []string{"HTTP", "HTTPS"} } @@ -36,7 +52,8 @@ func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { if err != nil { return nil, err } - defer httpResp.Body.Close() + // 拿到响应头就关闭，不用加defer + httpResp.Body.Close() res := &common.Resource{ Req: req, Range: false, @@ -57,13 +74,15 @@ func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) { } else if common.HttpCodeOK == httpResp.StatusCode { // 返回200响应码，不支持断点下载，通过Content-Length头获取文件大小，获取不到的话可能是chunked编码 contentLength := httpResp.Header.Get(common.HttpHeaderContentLength) - parse, err := strconv.ParseInt(contentLength, 10, 64) - if err != nil { - return nil, err + if contentLength != "" { + parse, err := strconv.ParseInt(contentLength, 10, 64) + if err != nil { + return nil, err + } + res.Size = parse } - res.Size = parse } else { - return nil, errors.New("http request fail") + return nil, NewRequestError(httpResp.StatusCode, httpResp.Status) } file := &common.FileInfo{ Size: res.Size, diff --git a/download/http/fetcher_test.go b/download/http/fetcher_test.go index 1234ad9a..7ac7446e 100644 --- a/download/http/fetcher_test.go +++ b/download/http/fetcher_test.go @@ -1,41 +1,171 @@ package http import ( - "encoding/json" - "fmt" + "crypto/md5" + "encoding/hex" "github.com/monkeyWie/gopeed/download/common" - "golang.org/x/net/proxy" "io" - "io/ioutil" "math/rand" "net" "net/http" "os" + "reflect" "testing" "time" ) -func startMockServer() net.Listener { +const testDir = "./" +const buildSize = 200 * 1024 * 1024 +const buildName = "build.data" +const buildFile = testDir + buildName +const downloadName = "download.data" +const downloadFile = testDir + downloadName + +func TestFetcher_Resolve(t *testing.T) { + testResolve(startTestFileServer, &common.Resource{ + Size: buildSize, + Range: true, + Files: []*common.FileInfo{ + { + Name: buildName, + Size: buildSize, + }, + }, + }, t) + testResolve(startTestChunkedServer, &common.Resource{ + Size: 0, + Range: false, + Files: []*common.FileInfo{ + { + Name: buildName, + Size: 0, + }, + }, + }, t) +} + +func testResolve(startTestServer func() net.Listener, want *common.Resource, t *testing.T) { + listener := startTestServer() + defer listener.Close() + fetcher := NewFetcher() + res, err := fetcher.Resolve(&common.Request{ + URL: "http://" + listener.Addr().String() + "/" + buildName, + }) + if err != nil { + t.Fatal(err) + } + res.Req = nil + if !reflect.DeepEqual(want, res) { + t.Errorf("Resolve error = %v, want %v", res, want) + } +} + +func TestFetcher_DownloadNormal(t *testing.T) { + listener := startTestFileServer() + defer listener.Close() + // 正常下载 + downloadNormal(listener, 1, t) + downloadNormal(listener, 5, t) + downloadNormal(listener, 8, t) + downloadNormal(listener, 16, t) +} + +func TestFetcher_DownloadContinue(t *testing.T) { + listener := startTestFileServer() + defer listener.Close() + // 暂停继续 + //downloadContinue(listener, 1, t) + downloadContinue(listener, 5, 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) + downloadContinue(listener, 1, t) +} + +func TestFetcher_DownloadRetry(t *testing.T) { + listener := startTestRetryServer() + defer listener.Close() + // chunked编码下载 + downloadNormal(listener, 1, t) +} + +func startTestFileServer() net.Listener { + return startTestServer(func() http.Handler { + return http.FileServer(http.Dir(testDir)) + }) +} + +func startTestChunkedServer() net.Listener { + return startTestServer(func() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/"+buildName, func(writer http.ResponseWriter, request *http.Request) { + file, err := os.Open(buildFile) + if err != nil { + panic(err) + } + defer file.Close() + io.Copy(writer, file) + }) + return mux + }) +} + +func startTestRetryServer() net.Listener { + counter := 0 + return startTestServer(func() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/"+buildName, func(writer http.ResponseWriter, request *http.Request) { + counter++ + if counter != 1 && counter < 5 { + writer.WriteHeader(500) + return + } + file, err := os.Open(buildFile) + if err != nil { + panic(err) + } + defer file.Close() + io.Copy(writer, file) + }) + return mux + }) +} + +func startTestServer(serverHandle func() http.Handler) net.Listener { listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { panic(err) } - file, err := os.Create("./test.data") + file, err := os.Create(buildFile) if err != nil { panic(err) } - buf := make([]byte, 1024) - for i := 0; i < 20*1024; i++ { + //随机生成一个文件 + l := int64(8192) + buf := make([]byte, l) + size := int64(0) + for { _, err := rand.Read(buf) if err != nil { panic(err) } - file.WriteAt(buf, int64(i*1024)) + if size+l >= buildSize { + file.WriteAt(buf[0:buildSize-size], size) + break + } + file.WriteAt(buf, size) + size += l } + server := &http.Server{} + server.Handler = serverHandle() + go server.Serve(listener) - go func() { - http.Serve(listener, http.FileServer(http.Dir("./"))) - }() return &delFileListener{ File: file, Listener: listener, @@ -48,93 +178,85 @@ type delFileListener struct { } func (c *delFileListener) Close() error { - c.File.Close() - err := os.Remove(c.File.Name()) - if err != nil { - fmt.Println(err) - return err - } + defer func() { + c.File.Close() + os.Remove(c.File.Name()) + os.Remove(downloadFile) + }() return c.Listener.Close() } -func TestFetcher_Resolve(t *testing.T) { - listener := startMockServer() - defer listener.Close() - fetcher := &Fetcher{} - resolve, err := fetcher.Resolve(&common.Request{ - URL: "http://" + listener.Addr().String() + "/test.data", +func downloadReady(listener net.Listener, connections int, t *testing.T) common.Process { + fetcher := NewFetcher() + fetcher.InitCtl(common.NewController()) + res, err := fetcher.Resolve(&common.Request{ + URL: "http://" + listener.Addr().String() + "/" + buildName, }) if err != nil { t.Fatal(err) } - buf, err := json.Marshal(resolve) + process, err := fetcher.Create(res, &common.Options{ + Name: downloadName, + Path: testDir, + Connections: connections, + }) if err != nil { t.Fatal(err) } - fmt.Printf("%s\n", string(buf)) + return process + } -func Test2(t *testing.T) { - socks5, err := proxy.SOCKS5("tcp", "127.0.0.1:1080", nil, nil) +func downloadNormal(listener net.Listener, connections int, t *testing.T) { + process := downloadReady(listener, connections, t) + err := process.Start() if err != nil { t.Fatal(err) } - conn, err := socks5.Dial("tcp", "www.baidu.com:80") - if err != nil { - t.Fatal(err) + want := fileMd5(buildFile) + got := fileMd5(downloadFile) + if want != got { + t.Errorf("Download error = %v, want %v", got, want) } - conn.Write([]byte("GET")) } -func Test3(t *testing.T) { - var u User - fmt.Println((&u).Say1()) - fmt.Println(u.Say2()) -} - -func Test4(t *testing.T) { - resp, err := http.Get("http://192.168.200.163:8088/go1.12.7.linux-amd64.tar.gz") - if err != nil { - t.Fatal(err) - } - defer resp.Body.Close() +func downloadContinue(listener net.Listener, connections int, t *testing.T) { + process := downloadReady(listener, connections, t) go func() { - time.Sleep(time.Second * 3) - err2 := resp.Body.Close() - if err2 != nil { - fmt.Println("close err2:" + err2.Error()) + err := process.Start() + if err != nil { + t.Fatal(err) } }() - _, err = io.Copy(ioutil.Discard, resp.Body) - if err != nil { + time.Sleep(time.Millisecond * 100) + if err := process.Pause(); err != nil { t.Fatal(err) } + time.Sleep(time.Millisecond * 200) + if err := process.Continue(); err != nil { + t.Fatal(err) + } + want := fileMd5(buildFile) + got := fileMd5(downloadFile) + if want != got { + t.Errorf("Download error = %v, want %v", got, want) + } } -func Test5(t *testing.T) { - name := "E:\\test\\gopeed\\http\\test.txt" - err := os.Truncate(name, 10) +func fileMd5(filePath string) string { + file, err := os.Open(filePath) if err != nil { - t.Fatal(err) + panic(err) } - file, err := os.OpenFile(name, os.O_RDWR, 0666) - if err != nil { - t.Fatal(err) - } - _, err = file.WriteAt([]byte("test"), 5) - if err != nil { - t.Fatal(err) + // Tell the program to call the following function when the current function returns + defer file.Close() + + // Open a new hash interface to write to + hash := md5.New() + + // Copy the file in the hash interface and check for any error + if _, err := io.Copy(hash, file); err != nil { + return "" } -} - -type User struct { - name string -} - -func (u *User) Say1() string { - return "hello1 " + u.name -} - -func (u User) Say2() string { - return "hello2 " + u.name + return hex.EncodeToString(hash.Sum(nil)) } diff --git a/download/http/model/chunk.go b/download/http/model/chunk.go index 6ae44152..521d1baf 100644 --- a/download/http/model/chunk.go +++ b/download/http/model/chunk.go @@ -1,11 +1,11 @@ package model import ( - common2 "github.com/monkeyWie/gopeed/download/common" + "github.com/monkeyWie/gopeed/download/common" ) type Chunk struct { - Status common2.Status + Status common.Status Begin int64 End int64 Downloaded int64 @@ -13,7 +13,7 @@ type Chunk struct { func NewChunk(begin int64, end int64) *Chunk { return &Chunk{ - Status: common2.DownloadStatusReady, + Status: common.DownloadStatusReady, Begin: begin, End: end, } diff --git a/download/http/process.go b/download/http/process.go index cf6a46fd..432edaa2 100644 --- a/download/http/process.go +++ b/download/http/process.go @@ -7,6 +7,7 @@ import ( "io" "net/http" "path/filepath" + "sync" ) type Process struct { @@ -17,6 +18,8 @@ type Process struct { status common.Status clients []*http.Response chunks []*model.Chunk + + pauseLock *sync.Mutex } func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *Process { @@ -25,6 +28,8 @@ func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *P res: res, opts: opts, status: common.DownloadStatusReady, + + pauseLock: &sync.Mutex{}, } } @@ -32,11 +37,14 @@ func (p *Process) Start() error { ctl := p.fetcher.GetCtl() // 创建文件 name := p.name() - err := ctl.Touch(name, p.res.Size) + _, err := ctl.Touch(name, p.res.Size) if err != nil { return err } - if p.res.Range && p.res.Size > 0 { + defer p.close() + p.status = common.DownloadStatusStart + var fetchErr error + if p.res.Range { // 每个连接平均需要下载的分块大小 var ( chunkSize = p.res.Size / int64(p.opts.Connections) @@ -62,12 +70,27 @@ func (p *Process) Start() error { errCh <- p.fetchChunk(i, name, chunk) }(i) } - return p.wait(errCh) + fetchErr = p.wait(errCh) + } else { + // 只支持单连接下载 + 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]) } - return nil + if p.status == common.DownloadStatusPause { + return nil + } + return fetchErr } func (p *Process) Pause() error { + p.pauseLock.Lock() + defer p.pauseLock.Unlock() + if common.DownloadStatusStart != p.status { + return nil + } + p.status = common.DownloadStatusPause if len(p.clients) > 0 { for _, client := range p.clients { client.Body.Close() @@ -77,10 +100,28 @@ func (p *Process) Pause() error { } func (p *Process) Continue() error { + if func() bool { + p.pauseLock.Lock() + defer p.pauseLock.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() errCh = make(chan error) ) + _, err := ctl.Open(name) + if err != nil { + return err + } + defer ctl.Close(name) for i := 0; i < p.opts.Connections; i++ { go func(i int) { errCh <- p.fetchChunk(i, name, p.chunks[i]) @@ -93,6 +134,10 @@ 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 @@ -120,6 +165,14 @@ func (p *Process) wait(errCh <-chan error) error { return nil } +type action int + +const ( + actionDirect action = iota + actionContinue + actionReturn +) + func (p *Process) fetchChunk(index int, name string, chunk *model.Chunk) (err error) { httpReq, err := BuildRequest(p.res.Req) if err != nil { @@ -135,14 +188,37 @@ func (p *Process) fetchChunk(index int, name string, chunk *model.Chunk) (err er 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 { - // 连接失败，直接重试 + if p.res.Range { + httpReq.Header.Set(common.HttpHeaderRange, + fmt.Sprintf(common.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End)) + } else { + chunk.Downloaded = 0 + } + var act action + act, err = func() (action, error) { + p.pauseLock.Lock() + defer p.pauseLock.Unlock() + if p.status == common.DownloadStatusPause { + return actionReturn, nil + } + resp, err = client.Do(httpReq) + if err != nil { + // 连接失败，直接重试 + return actionContinue, err + } + if resp.StatusCode != common.HttpCodeOK && resp.StatusCode != common.HttpCodePartialContent { + err = NewRequestError(resp.StatusCode, resp.Status) + return actionContinue, err + } + p.clients[index] = resp + return actionDirect, nil + }() + fmt.Printf("action:%d\n", act) + if act == actionContinue { continue + } else if act == actionReturn { + return nil } - p.clients[index] = resp retry, err = func() (bool, error) { defer resp.Body.Close() for {