diff --git a/cmd/terminal/gopeed.go b/cmd/terminal/gopeed.go new file mode 100644 index 00000000..ef963a57 --- /dev/null +++ b/cmd/terminal/gopeed.go @@ -0,0 +1,24 @@ +package main + +import ( + "github.com/monkeyWie/gopeed-core/download" + "github.com/monkeyWie/gopeed-core/download/base" + "github.com/monkeyWie/gopeed-core/download/http" +) + +func main() { + downloader := download.NewDownloader(http.FetcherBuilder) + res, err := downloader.Resolve(&base.Request{ + URL: "http://192.168.200.163:8088/docker-compose", + }) + if err != nil { + panic(err) + } + err = downloader.Create(res, &base.Options{ + Path: "E:\\test\\gopeed\\http", + Connections: 4, + }) + if err != nil { + panic(err) + } +} diff --git a/download/base/controller.go b/download/base/controller.go index 7464c74a..4a6e096d 100644 --- a/download/base/controller.go +++ b/download/base/controller.go @@ -27,7 +27,7 @@ type Request struct { type Resource struct { Req *Request // 资源总大小 - Size int64 + TotalSize int64 // 是否支持断点下载 Range bool // 资源所包含的文件列表 diff --git a/download/base/fetcher.go b/download/base/fetcher.go index b4632352..576d255d 100644 --- a/download/base/fetcher.go +++ b/download/base/fetcher.go @@ -2,7 +2,7 @@ package base // 对应协议的下载支持 type Fetcher interface { - Setup(ctl Controller) (err <-chan error) + Setup(ctl Controller) // 解析请求 Resolve(req *Request) (res *Resource, err error) // 创建任务 @@ -10,6 +10,9 @@ type Fetcher interface { Start() (err error) Pause() (err error) Continue() (err error) + + // 该方法会一直阻塞，直到任务下载结束 + Wait() (err error) } type DefaultFetcher struct { @@ -17,8 +20,11 @@ type DefaultFetcher struct { DoneCh chan error } -func (f *DefaultFetcher) Setup(ctl Controller) (doneCh <-chan error) { +func (f *DefaultFetcher) Setup(ctl Controller) { f.Ctl = ctl f.DoneCh = make(chan error, 1) - return f.DoneCh +} + +func (f *DefaultFetcher) Wait() (err error) { + return <-f.DoneCh } diff --git a/download/dltest/httptest.go b/download/dltest/httptest.go new file mode 100644 index 00000000..5f069a02 --- /dev/null +++ b/download/dltest/httptest.go @@ -0,0 +1,137 @@ +package dltest + +import ( + "fmt" + "io" + "math/rand" + "net" + "net/http" + "os" +) + +const ( + BuildName = "build.data" + BuildSize = 200 * 1024 * 1024 + TestDir = "./" + BuildFile = TestDir + BuildName + + DownloadName = "download.data" + DownloadFile = TestDir + DownloadName +) + +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 StartTestErrorServer() 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 { + writer.WriteHeader(500) + return + } + }) + 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(BuildFile) + if err != nil { + panic(err) + } + // 随机生成一个文件 + l := int64(8192) + buf := make([]byte, l) + size := int64(0) + for { + _, err := rand.Read(buf) + if err != nil { + panic(err) + } + 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) + + return &delFileListener{ + File: file, + Listener: listener, + } +} + +type delFileListener struct { + *os.File + net.Listener +} + +func (c *delFileListener) Close() error { + defer func() { + c.File.Close() + if err := ifExistAndRemove(c.File.Name()); err != nil { + fmt.Println(err) + } + 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 +} diff --git a/download/dltest/util.go b/download/dltest/util.go new file mode 100644 index 00000000..26ec6148 --- /dev/null +++ b/download/dltest/util.go @@ -0,0 +1,32 @@ +package dltest + +import ( + "crypto/md5" + "encoding/hex" + "encoding/json" + "io" + "os" +) + +func FileMd5(filePath string) string { + file, err := os.Open(filePath) + if err != nil { + panic(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 "" + } + return hex.EncodeToString(hash.Sum(nil)) +} + +func ToJson(v interface{}) string { + buf, _ := json.Marshal(v) + return string(buf) +} diff --git a/download/downloader.go b/download/downloader.go index de3d7a1c..396e5334 100644 --- a/download/downloader.go +++ b/download/downloader.go @@ -81,13 +81,20 @@ func (d *Downloader) Create(res *base.Resource, opts *base.Options) (err error) fetcher: fetcher, locker: new(sync.Mutex), } + d.emit(d.tasks[id], base.DownloadStatusStart) go func() { err = fetcher.Start() if err != nil { d.emit(d.tasks[id], base.DownloadStatusError) + return + } + err = fetcher.Wait() + if err != nil { + d.emit(d.tasks[id], base.DownloadStatusError) + } else { + d.emit(d.tasks[id], base.DownloadStatusDone) } }() - d.emit(d.tasks[id], base.DownloadStatusStart) return } diff --git a/download/downloader_test.go b/download/downloader_test.go index 1cc86322..b661ab29 100644 --- a/download/downloader_test.go +++ b/download/downloader_test.go @@ -1,43 +1,75 @@ package download import ( - "fmt" "github.com/monkeyWie/gopeed-core/download/base" + "github.com/monkeyWie/gopeed-core/download/dltest" "github.com/monkeyWie/gopeed-core/download/http" + "reflect" + "sync" "testing" ) func TestDownloader_Resolve(t *testing.T) { - downloader := buildDownloader() - res, err := downloader.Resolve(&base.Request{ - URL: "http://192.168.200.163:8088/docker-compose", - }) + listener := dltest.StartTestFileServer() + defer listener.Close() + + downloader := NewDownloader(http.FetcherBuilder) + req := &base.Request{ + URL: "http://" + listener.Addr().String() + "/" + dltest.BuildName, + } + res, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } - fmt.Println(res) + want := &base.Resource{ + Req: req, + TotalSize: dltest.BuildSize, + Range: true, + Files: []*base.FileInfo{ + { + Name: dltest.BuildName, + Path: "", + Size: dltest.BuildSize, + }, + }, + } + if !reflect.DeepEqual(res, want) { + t.Errorf("Resolve error = %s, want %s", dltest.ToJson(res), dltest.ToJson(want)) + } } func TestDownloader_Create(t *testing.T) { - downloader := buildDownloader() - res, err := downloader.Resolve(&base.Request{ - URL: "http://192.168.200.163:8088/docker-compose", - }) + listener := dltest.StartTestFileServer() + defer listener.Close() + + downloader := NewDownloader(http.FetcherBuilder) + req := &base.Request{ + URL: "http://" + listener.Addr().String() + "/" + dltest.BuildName, + } + res, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } + + var wg sync.WaitGroup + wg.Add(1) downloader.Listener(func(taskInfo *TaskInfo, status base.Status) { - fmt.Printf("task:%s,status %d\n", taskInfo.ID, status) + if status == base.DownloadStatusDone { + wg.Done() + } }) err = downloader.Create(res, &base.Options{ - Path: "E:\\test\\gopeed\\http", + Path: dltest.TestDir, + Name: dltest.DownloadName, Connections: 4, }) if err != nil { t.Fatal(err) } -} - -func buildDownloader() *Downloader { - return NewDownloader(http.FetcherBuilder) + wg.Wait() + want := dltest.FileMd5(dltest.BuildFile) + got := dltest.FileMd5(dltest.DownloadFile) + if want != got { + t.Errorf("Download error = %v, want %v", got, want) + } } diff --git a/download/http/fetcher.go b/download/http/fetcher.go index 3a6b47c9..647bd930 100644 --- a/download/http/fetcher.go +++ b/download/http/fetcher.go @@ -90,7 +90,7 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { if err != nil { return nil, err } - res.Size = parse + res.TotalSize = parse } } else if base.HttpCodeOK == httpResp.StatusCode { // 返回200响应码，不支持断点下载，通过Content-Length头获取文件大小，获取不到的话可能是chunked编码 @@ -100,13 +100,13 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { if err != nil { return nil, err } - res.Size = parse + res.TotalSize = parse } } else { return nil, NewRequestError(httpResp.StatusCode, httpResp.Status) } file := &base.FileInfo{ - Size: res.Size, + Size: res.TotalSize, } contentDisposition := httpResp.Header.Get(base.HttpHeaderContentDisposition) if contentDisposition != "" { @@ -141,14 +141,14 @@ func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error { func (f *Fetcher) Start() (err error) { // 创建文件 name := f.filename() - _, err = f.Ctl.Touch(name, f.res.Size) + _, err = f.Ctl.Touch(name, f.res.TotalSize) if err != nil { return err } f.status = base.DownloadStatusStart if f.res.Range { // 每个连接平均需要下载的分块大小 - chunkSize := f.res.Size / int64(f.opts.Connections) + chunkSize := f.res.TotalSize / 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++ { @@ -158,7 +158,7 @@ func (f *Fetcher) Start() (err error) { ) if i == f.opts.Connections-1 { // 最后一个分块需要保证把文件下载完 - end = f.res.Size - 1 + end = f.res.TotalSize - 1 } else { end = begin + chunkSize - 1 } diff --git a/download/http/fetcher_test.go b/download/http/fetcher_test.go index 86b528c2..71ae6e64 100644 --- a/download/http/fetcher_test.go +++ b/download/http/fetcher_test.go @@ -1,44 +1,31 @@ package http import ( - "crypto/md5" - "encoding/hex" - "fmt" "github.com/monkeyWie/gopeed-core/download/base" - "io" - "math/rand" + "github.com/monkeyWie/gopeed-core/download/dltest" "net" - "net/http" - "os" "reflect" "testing" "time" ) -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, &base.Resource{ - Size: buildSize, - Range: true, + testResolve(dltest.StartTestFileServer, &base.Resource{ + TotalSize: dltest.BuildSize, + Range: true, Files: []*base.FileInfo{ { - Name: buildName, - Size: buildSize, + Name: dltest.BuildName, + Size: dltest.BuildSize, }, }, }, t) - testResolve(startTestChunkedServer, &base.Resource{ - Size: 0, - Range: false, + testResolve(dltest.StartTestChunkedServer, &base.Resource{ + TotalSize: 0, + Range: false, Files: []*base.FileInfo{ { - Name: buildName, + Name: dltest.BuildName, Size: 0, }, }, @@ -50,7 +37,7 @@ func testResolve(startTestServer func() net.Listener, want *base.Resource, t *te defer listener.Close() fetcher := NewFetcher() res, err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + buildName, + URL: "http://" + listener.Addr().String() + "/" + dltest.BuildName, }) if err != nil { t.Fatal(err) @@ -62,7 +49,7 @@ func testResolve(startTestServer func() net.Listener, want *base.Resource, t *te } func TestFetcher_DownloadNormal(t *testing.T) { - listener := startTestFileServer() + listener := dltest.StartTestFileServer() defer listener.Close() // 正常下载 downloadNormal(listener, 1, t) @@ -72,7 +59,7 @@ func TestFetcher_DownloadNormal(t *testing.T) { } func TestFetcher_DownloadContinue(t *testing.T) { - listener := startTestFileServer() + listener := dltest.StartTestFileServer() defer listener.Close() // 暂停继续 downloadContinue(listener, 1, t) @@ -82,7 +69,7 @@ func TestFetcher_DownloadContinue(t *testing.T) { } func TestFetcher_DownloadChunked(t *testing.T) { - listener := startTestChunkedServer() + listener := dltest.StartTestChunkedServer() defer listener.Close() // chunked编码下载 downloadNormal(listener, 1, t) @@ -90,176 +77,59 @@ func TestFetcher_DownloadChunked(t *testing.T) { } func TestFetcher_DownloadRetry(t *testing.T) { - listener := startTestRetryServer() + listener := dltest.StartTestRetryServer() defer listener.Close() // chunked编码下载 downloadNormal(listener, 1, t) } func TestFetcher_DownloadError(t *testing.T) { - listener := startTestErrorServer() + listener := dltest.StartTestErrorServer() defer listener.Close() // chunked编码下载 downloadError(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 startTestErrorServer() 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 { - writer.WriteHeader(500) - return - } - }) - 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(buildFile) - if err != nil { - panic(err) - } - // 随机生成一个文件 - l := int64(8192) - buf := make([]byte, l) - size := int64(0) - for { - _, err := rand.Read(buf) - if err != nil { - panic(err) - } - 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) - - return &delFileListener{ - File: file, - Listener: listener, - } -} - -type delFileListener struct { - *os.File - net.Listener -} - -func (c *delFileListener) Close() error { - defer func() { - c.File.Close() - if err := ifExistAndRemove(c.File.Name()); err != nil { - fmt.Println(err) - } - 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) (base.Fetcher, <-chan error) { +func downloadReady(listener net.Listener, connections int, t *testing.T) base.Fetcher { fetcher := NewFetcher() - doneCh := fetcher.Setup(base.NewController()) + fetcher.Setup(base.NewController()) res, err := fetcher.Resolve(&base.Request{ - URL: "http://" + listener.Addr().String() + "/" + buildName, + URL: "http://" + listener.Addr().String() + "/" + dltest.BuildName, }) if err != nil { t.Fatal(err) } err = fetcher.Create(res, &base.Options{ - Name: downloadName, - Path: testDir, + Name: dltest.DownloadName, + Path: dltest.TestDir, Connections: connections, }) if err != nil { t.Fatal(err) } - return fetcher, doneCh + return fetcher } func downloadNormal(listener net.Listener, connections int, t *testing.T) { - fetcher, doneCh := downloadReady(listener, connections, t) + fetcher := downloadReady(listener, connections, t) err := fetcher.Start() if err != nil { t.Fatal(err) } - err = <-doneCh + err = fetcher.Wait() if err != nil { t.Fatal(err) } - want := fileMd5(buildFile) - got := fileMd5(downloadFile) + want := dltest.FileMd5(dltest.BuildFile) + got := dltest.FileMd5(dltest.DownloadFile) if want != got { t.Errorf("Download error = %v, want %v", got, want) } } func downloadContinue(listener net.Listener, connections int, t *testing.T) { - fetcher, doneCh := downloadReady(listener, connections, t) + fetcher := downloadReady(listener, connections, t) err := fetcher.Start() if err != nil { t.Fatal(err) @@ -272,43 +142,25 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { if err := fetcher.Continue(); err != nil { t.Fatal(err) } - err = <-doneCh + err = fetcher.Wait() if err != nil { t.Fatal(err) } - want := fileMd5(buildFile) - got := fileMd5(downloadFile) + want := dltest.FileMd5(dltest.BuildFile) + got := dltest.FileMd5(dltest.DownloadFile) if want != got { t.Errorf("Download error = %v, want %v", got, want) } } func downloadError(listener net.Listener, connections int, t *testing.T) { - fetcher, doneCh := downloadReady(listener, connections, t) + fetcher := downloadReady(listener, connections, t) err := fetcher.Start() if err != nil { t.Fatal(err) } - err = <-doneCh + err = fetcher.Wait() if err == nil { t.Errorf("Download error = %v, want %v", err, nil) } } - -func fileMd5(filePath string) string { - file, err := os.Open(filePath) - if err != nil { - panic(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 "" - } - return hex.EncodeToString(hash.Sum(nil)) -}