refactor: fetcher模型

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)
+	}
+}

download/base/controller.go

@@ -27,7 +27,7 @@ type Request struct {
 type Resource struct {
 	Req *Request
 	// 资源总大小
-	Size int64
+	TotalSize int64
 	// 是否支持断点下载
 	Range bool
 	// 资源所包含的文件列表

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
 }

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
+}

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)
+}

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
 }
 

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)
+	}
 }

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
 			}

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))
-}