windows快速分配

.github/workflows/build.yml

@@ -1,48 +0,0 @@
-name: build
-
-on:
-  pull_request:
-  push:
-    branches:
-      - dev
-
-jobs:
-#  lint:
-#    name: Lint
-#    runs-on: ubuntu-latest
-#    steps:
-#      - uses: actions/setup-go@v2
-#        with:
-#          go-version: '^1.15.2'
-#      - uses: actions/checkout@v2
-#      - name: Lint Go Code
-#        run: |
-#          export PATH=$PATH:$(go env GOPATH)/bin # temporary fix. See https://github.com/actions/setup-go/issues/14
-#          go get -u golang.org/x/lint/golint
-#          golint -set_exit_status ./...
-  test:
-    name: Test
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/setup-go@v2
-        with:
-          go-version: '^1.15.2'
-      - uses: actions/checkout@v2
-      - name: Run Unit tests.
-        run: go test ./... -v -covermode=count -coverprofile=coverage.txt
-      - uses: codecov/codecov-action@v1
-        with:
-          file: ./coverage.txt
-          fail_ci_if_error: true
-  build:
-    name: Build
-    runs-on: ubuntu-latest
-#    needs: [lint, test]
-    needs: [test]
-    steps:
-      - uses: actions/setup-go@v2
-        with:
-          go-version: '^1.15.2'
-      - uses: actions/checkout@v2
-      - name: Build
-        run: go build -ldflags="-s -w" -o bin/gopeed cmd/gopeed/gopeed.go

.vscode/launch.json

@@ -0,0 +1,17 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "name": "Launch",
+            "type": "go",
+            "request": "launch",
+            "mode": "auto",
+            "program": "${fileDirname}",
+            "env": {},
+            "args": []
+        }
+    ]
+}

.vscode/settings.json

@@ -0,0 +1,9 @@
+{
+  "go.useLanguageServer": true,
+  "[go]": {
+    "editor.formatOnSave": true,
+    "editor.codeActionsOnSave": {
+      "source.organizeImports": true
+    }
+  },
+}

README.md

@@ -1,38 +1,20 @@
-[![Build Status](https://github.com/monkeyWie/gopeed-core/workflows/build/badge.svg)](https://github.com/monkeyWie/gopeed-core/actions?query=workflow%3Abuild)
-[![Codecov](https://codecov.io/gh/monkeyWie/gopeed-core/branch/dev/graph/badge.svg)](https://codecov.io/gh/monkeyWie/gopeed-core)
-
 ## gopeed
-
-支持多协议（HTTP、BitTorrent）下载的客户端，提供命令行、RESTful API、WebSocket、Go 类库方式来使用。
-
-## 安装
-
-```sh
-go install github.com/monkeyWie/gopeed-core
-```
-
-## 示例
-
-```go
-
-```
+支持多协议（HTTP、BitTorrent）下载的客户端，提供命令行、RESTful API、WebSocket、Go类库方式来使用。
 
 ## TODO
-
-- [x] HTTP 下载实现
-- [ ] BitTorrent 下载实现
-  - [x] .torrent 文件解析
-  - [x] tracker 协议实现
-  - [x] peer wire protocol 协议实现
-  - [ ] DHT 协议实现
-  - [ ] 磁力链接支持
-  - [ ] uTP 协议实现
-- [x] 下载接口抽象(不关心具体协议)
+- [x] HTTP下载实现
+- [ ] BitTorrent下载实现
+    - [x] .torrent文件解析
+    - [x] tracker协议实现
+    - [x] peer wire protocol协议实现
+    - [ ] DHT协议实现
+    - [ ] 磁力链接支持
+    - [ ] uTP协议实现
+- [ ] 下载接口抽象(不关心具体协议)
 - [ ] 支持自定义配置
 - [ ] 限速功能实现
-- [x] 命令行工具提供
-- [ ] RESTful 服务提供
+- [ ] 命令行工具提供
+- [ ] RESTful服务提供
 
 ## 参与
-
-由于项目目前还未定型，代码可能随时有大的调整，所以暂不接受 PR，当然如果有什么好的想法可以在 issue 区提出来。
+由于项目目前还未定型，代码可能随时有大的调整，所以暂不接受PR，当然如果有什么好的想法可以在issue区提出来。

cmd/gopeed/flags.go

@@ -1,42 +0,0 @@
-package main
-
-import (
-	"flag"
-	"fmt"
-	"os"
-	"path/filepath"
-)
-
-type args struct {
-	url         string
-	connections *int
-	dir         *string
-}
-
-func parse() *args {
-	dir, err := filepath.Abs(filepath.Dir(os.Args[0]))
-	if err != nil {
-		panic(err)
-	}
-	var args args
-	args.connections = flag.Int("C", 16, "Concurrent connections.")
-	args.dir = flag.String("D", dir, "Save directory.")
-	flag.Parse()
-	t := flag.Args()
-	if len(t) > 0 {
-		args.url = t[0]
-	} else {
-		gPrintln("missing url parameter")
-		gPrintln("try 'gopeed -h' for more information")
-		os.Exit(1)
-	}
-	return &args
-}
-
-func gPrint(msg string) {
-	fmt.Print("gopeed: " + msg)
-}
-
-func gPrintln(msg string) {
-	gPrint(msg + "\n")
-}

cmd/gopeed/gopeed.go

@@ -1,64 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"github.com/monkeyWie/gopeed-core/internal/protocol/http"
-	"github.com/monkeyWie/gopeed-core/pkg/base"
-	"github.com/monkeyWie/gopeed-core/pkg/download"
-	"github.com/monkeyWie/gopeed-core/pkg/util"
-	"path/filepath"
-	"sync"
-)
-
-const progressWidth = 20
-
-func main() {
-	args := parse()
-
-	downloader := download.NewDownloader(http.FetcherBuilder)
-	res, err := downloader.Resolve(&base.Request{
-		URL: args.url,
-	})
-	if err != nil {
-		panic(err)
-	}
-
-	err = downloader.Create(res, &base.Options{
-		Path:        *args.dir,
-		Connections: *args.connections,
-	})
-	if err != nil {
-		panic(err)
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(1)
-	downloader.Listener(func(taskInfo *download.TaskInfo, eventKey download.EventKey) {
-		if eventKey == download.EventKeyProgress {
-			printProgress(taskInfo, "downloading...")
-		}
-		if eventKey == download.EventKeyDone {
-			printProgress(taskInfo, "complete")
-			wg.Done()
-		}
-	})
-	wg.Wait()
-	fmt.Println()
-	gPrint("saving file " + filepath.Join(*args.dir, res.Files[0].Name))
-}
-
-func printProgress(taskInfo *download.TaskInfo, title string) {
-	rate := float64(taskInfo.Progress.Downloaded) / float64(taskInfo.Res.TotalSize)
-	completeWidth := int(progressWidth * rate)
-	speed := util.ByteFmt(taskInfo.Progress.Speed)
-	totalSize := util.ByteFmt(taskInfo.Res.TotalSize)
-	fmt.Printf("\r%s [", title)
-	for i := 0; i < progressWidth; i++ {
-		if i < completeWidth {
-			fmt.Print("■")
-		} else {
-			fmt.Print("□")
-		}
-	}
-	fmt.Printf("] %.1f%%    %s/s    %s", rate*100, speed, totalSize)
-}

cmd/main.go

@@ -0,0 +1,28 @@
+package main
+
+import (
+	"github.com/monkeyWie/gopeed-core/download"
+	"github.com/monkeyWie/gopeed-core/download/common"
+	"github.com/monkeyWie/gopeed-core/download/http"
+)
+
+func main() {
+	downloader := buildDownloader()
+	res, err := downloader.Resolve(&common.Request{
+		URL: "http://192.168.200.163:8088/docker-compose",
+	})
+	if err != nil {
+		panic(err)
+	}
+	err = downloader.Create(res, &common.Options{
+		Path:        "E:\\test\\gopeed\\http",
+		Connections: 4,
+	})
+	if err != nil {
+		panic(err)
+	}
+}
+
+func buildDownloader() *download.Downloader {
+	return download.NewDownloader(&http.Fetcher{BaseFetcher: &common.BaseFetcher{}})
+}

download/common/constants.go

@@ -1,4 +1,4 @@
-package base
+package common
 
 import "errors"
 
@@ -25,5 +25,6 @@ const (
 )
 
 var (
+	PauseErr  = errors.New("pause")
 	DeleteErr = errors.New("delete")
 )

download/common/controller.go

@@ -1,4 +1,4 @@
-package controller
+package common
 
 import (
 	"golang.org/x/net/proxy"
@@ -7,29 +7,68 @@ import (
 	"time"
 )
 
-type Controller interface {
-	Touch(name string, size int64) (file *os.File, err error)
-	Open(name string) (file *os.File, err error)
-	Write(name string, offset int64, buf []byte) (int, error)
-	Close(name string) error
-	ContextDialer() (proxy.Dialer, error)
+// 下载请求
+type Request struct {
+	// 下载链接
+	URL string
+	// 附加信息
+	Extra interface{}
 }
 
-type DefaultController struct {
+// 资源信息
+type Resource struct {
+	Req *Request
+	// 资源总大小
+	Size int64
+	// 是否支持断点下载
+	Range bool
+	// 资源所包含的文件列表
+	Files []*FileInfo
+}
+
+type Options struct {
+	// 保存文件名
+	Name string
+	// 保存目录
+	Path string
+	// 并发连接数
+	Connections int
+}
+
+type FileInfo struct {
+	Name string
+	Path string
+	Size int64
+}
+
+type Process interface {
+	Start() error
+	Pause() error
+	Continue() error
+	Delete() error
+}
+
+type FileCreator interface {
+	Create(name string, size int64) (file *os.File, err error)
+}
+
+type Controller struct {
 	Files map[string]*os.File
 }
 
-func NewController() *DefaultController {
-	return &DefaultController{Files: make(map[string]*os.File)}
+func NewController() *Controller {
+	return &Controller{Files: make(map[string]*os.File)}
 }
 
-func (c *DefaultController) Touch(name string, size int64) (file *os.File, err error) {
-	file, err = os.Create(name)
+func (c *Controller) Touch(name string, size int64) (file *os.File, err error) {
 	if size > 0 {
-		err = 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)
 	}
 	if err == nil {
 		c.Files[name] = file
@@ -37,25 +76,25 @@ func (c *DefaultController) Touch(name string, size int64) (file *os.File, err e
 	return
 }
 
-func (c *DefaultController) Open(name string) (file *os.File, err error) {
-	file, err = os.OpenFile(name, os.O_RDWR, os.ModePerm)
+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 *DefaultController) Write(name string, offset int64, buf []byte) (int, error) {
+func (c *Controller) Write(name string, offset int64, buf []byte) (int, error) {
 	return c.Files[name].WriteAt(buf, offset)
 }
 
-func (c *DefaultController) Close(name string) error {
+func (c *Controller) Close(name string) error {
 	err := c.Files[name].Close()
 	delete(c.Files, name)
 	return err
 }
 
-func (c *DefaultController) ContextDialer() (proxy.Dialer, error) {
+func (c *Controller) ContextDialer() (proxy.Dialer, error) {
 	// return proxy.SOCKS5("tpc", "127.0.0.1:9999", nil, nil)
 	var dialer proxy.Dialer
 	return &DialerWarp{dialer: dialer}, nil

download/common/fetcher.go

@@ -0,0 +1,29 @@
+package common
+
+// 对应协议的下载支持
+type Fetcher interface {
+	BaseFetch
+	// 支持的协议列表
+	Protocols() []string
+	// 解析请求
+	Resolve(req *Request) (*Resource, error)
+	// 创建任务
+	Create(res *Resource, opts *Options) (Process, error)
+}
+
+type BaseFetch interface {
+	InitCtl(ctl *Controller)
+	GetCtl() *Controller
+}
+
+type BaseFetcher struct {
+	ctl *Controller
+}
+
+func (b *BaseFetcher) InitCtl(ctl *Controller) {
+	b.ctl = ctl
+}
+
+func (b *BaseFetcher) GetCtl() *Controller {
+	return b.ctl
+}

download/downloader.go

@@ -0,0 +1,81 @@
+package download
+
+import (
+	"errors"
+	"github.com/google/uuid"
+	"github.com/monkeyWie/gopeed-core/download/common"
+	"net/url"
+	"strings"
+)
+
+type TaskInfo struct {
+	ID     string
+	Res    *common.Resource
+	Opts   *common.Options
+	Status common.Status
+	Files  map[string]*common.FileInfo
+
+	process common.Process
+}
+
+type Downloader struct {
+	fetches map[string]common.Fetcher
+	tasks   map[string]*TaskInfo
+	*common.Controller
+}
+
+func NewDownloader(fetchers ...common.Fetcher) *Downloader {
+	d := &Downloader{Controller: common.NewController()}
+	d.fetches = make(map[string]common.Fetcher)
+	for _, f := range fetchers {
+		for _, p := range f.Protocols() {
+			d.fetches[strings.ToUpper(p)] = f
+		}
+		f.InitCtl(d.Controller)
+	}
+	d.tasks = make(map[string]*TaskInfo)
+	return d
+}
+
+func (d *Downloader) getFetcher(URL string) (common.Fetcher, error) {
+	url, err := url.Parse(URL)
+	if err != nil {
+		return nil, err
+	}
+	if fetch, ok := d.fetches[strings.ToUpper(url.Scheme)]; ok {
+		return fetch, nil
+	}
+	return nil, errors.New("unsupported protocol")
+}
+
+func (d *Downloader) Resolve(req *common.Request) (*common.Resource, error) {
+	fetcher, err := d.getFetcher(req.URL)
+	if err != nil {
+		return nil, err
+	}
+	return fetcher.Resolve(req)
+}
+
+func (d *Downloader) Create(res *common.Resource, opts *common.Options) error {
+	fetcher, err := d.getFetcher(res.Req.URL)
+	if err != nil {
+		return err
+	}
+	process, err := fetcher.Create(res, opts)
+	if err != nil {
+		return err
+	}
+	id := uuid.New().String()
+	d.tasks[id] = &TaskInfo{
+		ID:      id,
+		Res:     res,
+		Opts:    opts,
+		Status:  common.DownloadStatusReady,
+		process: process,
+	}
+	return process.Start()
+}
+
+func (d *Downloader) Pause(id string) {
+	d.tasks[id].process.Pause()
+}

download/downloader_test.go

@@ -0,0 +1,40 @@
+package download
+
+import (
+	"fmt"
+	"github.com/monkeyWie/gopeed-core/download/common"
+	"github.com/monkeyWie/gopeed-core/download/http"
+	"testing"
+)
+
+func TestDownloader_Resolve(t *testing.T) {
+	downloader := buildDownloader()
+	res, err := downloader.Resolve(&common.Request{
+		URL: "http://192.168.200.163:8088/docker-compose",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	fmt.Println(res)
+}
+
+func TestDownloader_Create(t *testing.T) {
+	downloader := buildDownloader()
+	res, err := downloader.Resolve(&common.Request{
+		URL: "http://192.168.200.163:8088/docker-compose",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = downloader.Create(res, &common.Options{
+		Path:        "E:\\test\\gopeed\\http",
+		Connections: 4,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+func buildDownloader() *Downloader {
+	return NewDownloader(&http.Fetcher{BaseFetcher: &common.BaseFetcher{}})
+}

download/file_allocate/file_allocate.go

@@ -0,0 +1,18 @@
+package file_allocate
+
+import (
+	"os"
+)
+
+type FileAllocate interface {
+	Allocate(name string, size int64) (file *os.File, err error)
+}
+
+type DefaultFileAllocator struct {
+}
+
+func (p *DefaultFileAllocator) Allocate(name string, size int64) (file *os.File, err error) {
+	err = os.Truncate(name, size)
+	file, err = os.OpenFile(name, os.O_RDWR, 0644)
+	return
+}

download/file_allocate/pre_file_allocate.go

@@ -0,0 +1,12 @@
+package file_allocate
+
+import "os"
+
+type PreFileAllocator struct {
+}
+
+func (p *PreFileAllocator) Allocate(name string, size int64) (file *os.File, err error) {
+	err = os.Truncate(name, size)
+	file, err = os.OpenFile(name, os.O_RDWR, 0644)
+	return
+}

download/http/fetcher.go

@@ -0,0 +1,167 @@
+package http
+
+import (
+	"bytes"
+	"fmt"
+	"github.com/monkeyWie/gopeed-core/download/common"
+	"github.com/monkeyWie/gopeed-core/download/http/model"
+	"io/ioutil"
+	"mime"
+	"net/http"
+	"net/http/cookiejar"
+	"net/url"
+	"strconv"
+	"strings"
+	"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"}
+}
+
+func (h *Fetcher) Resolve(req *common.Request) (*common.Resource, error) {
+	httpReq, err := BuildRequest(req)
+	if err != nil {
+		return nil, err
+	}
+	client := BuildClient()
+	// 只访问一个字节，测试资源是否支持Range请求
+	httpReq.Header.Set(common.HttpHeaderRange, fmt.Sprintf(common.HttpHeaderRangeFormat, 0, 0))
+	httpResp, err := client.Do(httpReq)
+	if err != nil {
+		return nil, err
+	}
+	// 拿到响应头就关闭，不用加defer
+	httpResp.Body.Close()
+	res := &common.Resource{
+		Req:   req,
+		Range: false,
+		Files: []*common.FileInfo{},
+	}
+	if common.HttpCodePartialContent == httpResp.StatusCode {
+		// 返回206响应码表示支持断点下载
+		res.Range = true
+		// 解析资源大小: bytes 0-1000/1001 => 1001
+		contentTotal := subLastSlash(httpResp.Header.Get(common.HttpHeaderContentRange))
+		if contentTotal != "" {
+			parse, err := strconv.ParseInt(contentTotal, 10, 64)
+			if err != nil {
+				return nil, err
+			}
+			res.Size = parse
+		}
+	} else if common.HttpCodeOK == httpResp.StatusCode {
+		// 返回200响应码，不支持断点下载，通过Content-Length头获取文件大小，获取不到的话可能是chunked编码
+		contentLength := httpResp.Header.Get(common.HttpHeaderContentLength)
+		if contentLength != "" {
+			parse, err := strconv.ParseInt(contentLength, 10, 64)
+			if err != nil {
+				return nil, err
+			}
+			res.Size = parse
+		}
+	} else {
+		return nil, NewRequestError(httpResp.StatusCode, httpResp.Status)
+	}
+	file := &common.FileInfo{
+		Size: res.Size,
+	}
+	contentDisposition := httpResp.Header.Get(common.HttpHeaderContentDisposition)
+	if contentDisposition != "" {
+		_, params, _ := mime.ParseMediaType(contentDisposition)
+		filename := params["filename"]
+		if filename != "" {
+			file.Name = filename
+		}
+	}
+	// Get file name by URL
+	if file.Name == "" {
+		parse, err := url.Parse(req.URL)
+		if err == nil {
+			// e.g. /files/test.txt => test.txt
+			file.Name = subLastSlash(parse.Path)
+		}
+	}
+	// unknown file name
+	if file.Name == "" {
+		file.Name = "unknown"
+	}
+	res.Files = append(res.Files, file)
+	return res, nil
+}
+
+func (h *Fetcher) Create(res *common.Resource, opts *common.Options) (common.Process, error) {
+	if opts.Connections != 1 && !res.Range {
+		opts.Connections = 1
+	}
+	return NewProcess(h, res, opts), nil
+}
+
+func subLastSlash(str string) string {
+	if str == "" {
+		return ""
+	}
+	index := strings.LastIndex(str, "/")
+	if index != -1 {
+		return str[index+1:]
+	}
+	return ""
+}
+
+func BuildClient() *http.Client {
+	// Cookie handle
+	jar, _ := cookiejar.New(nil)
+	return &http.Client{
+		Jar:     jar,
+		Timeout: time.Second * 10,
+	}
+}
+
+func BuildRequest(req *common.Request) (*http.Request, error) {
+	url, err := url.Parse(req.URL)
+	if err != nil {
+		return nil, err
+	}
+	httpReq := &http.Request{
+		URL:    url,
+		Header: map[string][]string{},
+	}
+
+	if req.Extra != nil {
+		if extra, ok := req.Extra.(model.Extra); ok {
+			if extra.Method != "" {
+				httpReq.Method = extra.Method
+			} else {
+				httpReq.Method = http.MethodGet
+			}
+			if len(extra.Header) > 0 {
+				for k, v := range extra.Header {
+					httpReq.Header[k] = []string{v}
+				}
+			}
+			httpReq.Body = ioutil.NopCloser(bytes.NewBufferString(extra.Body))
+		}
+	}
+	return httpReq, nil
+}

download/http/fetcher_test.go

@@ -0,0 +1,307 @@
+package http
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+	"github.com/monkeyWie/gopeed-core/download/common"
+	"io"
+	"math/rand"
+	"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, &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)
+	fmt.Println("d1")
+	downloadContinue(listener, 5, t)
+	fmt.Println("d5")
+	downloadContinue(listener, 8, t)
+	fmt.Println("d8")
+	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 TestFetcher_DownloadError(t *testing.T) {
+	listener := 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) 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)
+	}
+	process, err := fetcher.Create(res, &common.Options{
+		Name:        downloadName,
+		Path:        testDir,
+		Connections: connections,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return process
+
+}
+
+func downloadNormal(listener net.Listener, connections int, t *testing.T) {
+	process := downloadReady(listener, connections, t)
+	err := process.Start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := fileMd5(buildFile)
+	got := fileMd5(downloadFile)
+	if want != got {
+		t.Errorf("Download error = %v, want %v", got, want)
+	}
+}
+
+func downloadContinue(listener net.Listener, connections int, t *testing.T) {
+	process := downloadReady(listener, connections, t)
+	go func() {
+		err := process.Start()
+		if err != nil && err != common.PauseErr {
+			t.Fatal(err)
+		}
+	}()
+	time.Sleep(time.Millisecond * 200)
+	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 downloadError(listener net.Listener, connections int, t *testing.T) {
+	process := downloadReady(listener, connections, t)
+	err := process.Start()
+	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))
+}

download/http/model/chunk.go

@@ -0,0 +1,20 @@
+package model
+
+import (
+	"github.com/monkeyWie/gopeed-core/download/common"
+)
+
+type Chunk struct {
+	Status     common.Status
+	Begin      int64
+	End        int64
+	Downloaded int64
+}
+
+func NewChunk(begin int64, end int64) *Chunk {
+	return &Chunk{
+		Status: common.DownloadStatusReady,
+		Begin:  begin,
+		End:    end,
+	}
+}

download/http/model/extra.go

@@ -0,0 +1,7 @@
+package model
+
+type Extra struct {
+	Method string
+	Header map[string]string
+	Body   string
+}

download/http/process.go

@@ -0,0 +1,245 @@
+package http
+
+import (
+	"fmt"
+	"github.com/monkeyWie/gopeed-core/download/common"
+	"github.com/monkeyWie/gopeed-core/download/http/model"
+	"io"
+	"net/http"
+	"path/filepath"
+	"sync"
+)
+
+type Process struct {
+	fetcher *Fetcher
+
+	res     *common.Resource
+	opts    *common.Options
+	status  common.Status
+	clients []*http.Response
+	chunks  []*model.Chunk
+
+	pauseCond *sync.Cond
+}
+
+func NewProcess(fetcher *Fetcher, res *common.Resource, opts *common.Options) *Process {
+	return &Process{
+		fetcher: fetcher,
+		res:     res,
+		opts:    opts,
+		status:  common.DownloadStatusReady,
+
+		pauseCond: sync.NewCond(&sync.Mutex{}),
+	}
+}
+
+func (p *Process) Start() error {
+	ctl := p.fetcher.GetCtl()
+	// 创建文件
+	name := p.name()
+	_, err := ctl.Touch(name, p.res.Size)
+	if err != nil {
+		return err
+	}
+	defer p.close()
+	p.status = common.DownloadStatusStart
+	if p.res.Range {
+		// 每个连接平均需要下载的分块大小
+		chunkSize := p.res.Size / int64(p.opts.Connections)
+		p.chunks = make([]*model.Chunk, p.opts.Connections)
+		p.clients = make([]*http.Response, p.opts.Connections)
+		for i := 0; i < p.opts.Connections; i++ {
+			var (
+				begin = chunkSize * int64(i)
+				end   int64
+			)
+			if i == p.opts.Connections-1 {
+				// 最后一个分块需要保证把文件下载完
+				end = p.res.Size
+			} else {
+				end = begin + chunkSize
+			}
+			chunk := model.NewChunk(begin, end)
+			p.chunks[i] = chunk
+		}
+	} else {
+		// 只支持单连接下载
+		p.chunks = make([]*model.Chunk, 1)
+		p.clients = make([]*http.Response, 1)
+		p.chunks[0] = model.NewChunk(0, 0)
+	}
+	return p.fetch()
+}
+
+func (p *Process) Pause() error {
+	p.pauseCond.L.Lock()
+	defer p.pauseCond.L.Unlock()
+	if common.DownloadStatusStart != p.status {
+		return nil
+	}
+	p.status = common.DownloadStatusPause
+	p.stop()
+	// 释放锁，并等待下载结束
+	p.pauseCond.Wait()
+	return nil
+}
+
+func (p *Process) Continue() error {
+	if func() bool {
+		p.pauseCond.L.Lock()
+		defer p.pauseCond.L.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()
+	)
+	_, err := ctl.Open(name)
+	if err != nil {
+		return err
+	}
+	defer ctl.Close(name)
+	return p.fetch()
+}
+
+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
+	if filename == "" {
+		filename = p.res.Files[0].Name
+	}
+	return filepath.Join(p.opts.Path, filename)
+}
+
+func (p *Process) fetch() error {
+	errCh := make(chan error, p.opts.Connections)
+	defer close(errCh)
+
+	for i := 0; i < p.opts.Connections; i++ {
+		go func(i int) {
+			errCh <- p.fetchChunk(i, p.name(), p.chunks[i])
+		}(i)
+	}
+
+	var retErr error
+	for i := 0; i < p.opts.Connections; i++ {
+		err := <-errCh
+		if err != nil && retErr == nil {
+			retErr = err
+			// 有一个连接失败就立即终止下载
+			if retErr != common.PauseErr {
+				func() {
+					p.pauseCond.L.Lock()
+					defer p.pauseCond.L.Unlock()
+					p.stop()
+				}()
+			}
+		}
+	}
+	p.pauseCond.Signal()
+	return retErr
+}
+
+func (p *Process) stop() {
+	if len(p.clients) > 0 {
+		for _, client := range p.clients {
+			fmt.Println("client close")
+			if client != nil {
+				client.Body.Close()
+			}
+		}
+	}
+}
+
+func (p *Process) fetchChunk(index int, name string, chunk *model.Chunk) (err error) {
+	httpReq, err := BuildRequest(p.res.Req)
+	if err != nil {
+		return err
+	}
+	var (
+		client = BuildClient()
+		buf    = make([]byte, 8192)
+	)
+	// 重试5次
+	for i := 0; i < 5; i++ {
+		var (
+			resp  *http.Response
+			retry bool
+		)
+		if p.res.Range {
+			httpReq.Header.Set(common.HttpHeaderRange,
+				fmt.Sprintf(common.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End))
+		} else {
+			chunk.Downloaded = 0
+		}
+		err = func() error {
+			p.pauseCond.L.Lock()
+			defer p.pauseCond.L.Unlock()
+			if p.status == common.DownloadStatusPause {
+				return common.PauseErr
+			}
+			resp, err = client.Do(httpReq)
+			if err != nil {
+				return err
+			}
+			p.clients[index] = resp
+			if resp.StatusCode != common.HttpCodeOK && resp.StatusCode != common.HttpCodePartialContent {
+				err = NewRequestError(resp.StatusCode, resp.Status)
+				return err
+			}
+			return nil
+		}()
+		if err != nil {
+			if err == common.PauseErr {
+				return err
+			}
+			// 请求失败重试
+			continue
+		}
+		retry, err = func() (bool, error) {
+			defer resp.Body.Close()
+			for {
+				n, err := resp.Body.Read(buf)
+				if n > 0 {
+					_, err := p.fetcher.GetCtl().Write(name, chunk.Begin+chunk.Downloaded, buf[:n])
+					if err != nil {
+						return false, err
+					}
+					chunk.Downloaded += int64(n)
+				}
+				if err != nil {
+					if err != io.EOF {
+						return true, err
+					}
+					break
+				}
+			}
+			return false, nil
+		}()
+		if err == nil || !retry {
+			// 下载成功，跳出重试
+			break
+		}
+	}
+	if err != nil {
+		p.chunks[index].Status = common.DownloadStatusError
+	} else {
+		p.chunks[index].Status = common.DownloadStatusDone
+	}
+	return
+}

go.mod

@@ -5,13 +5,8 @@ go 1.13
 require (
 	github.com/RoaringBitmap/roaring v0.4.21
 	github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972
-	github.com/gizak/termui/v3 v3.1.0
 	github.com/google/uuid v1.1.1
 	github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e
-	github.com/mattn/go-tty v0.0.3 // indirect
 	github.com/sirupsen/logrus v1.4.2
-	github.com/superhawk610/bar v0.0.2
-	github.com/superhawk610/terminal v0.1.0 // indirect
 	golang.org/x/net v0.0.0-20200707034311-ab3426394381
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 )

go.sum

@@ -6,8 +6,6 @@ github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/gizak/termui/v3 v3.1.0 h1:ZZmVDgwHl7gR7elfKf1xc4IudXZ5qqfDh4wExk4Iajc=
-github.com/gizak/termui/v3 v3.1.0/go.mod h1:bXQEBkJpzxUAKf0+xq9MSWAvWZlE7c+aidmyFlkYTrY=
 github.com/glycerine/go-unsnap-stream v0.0.0-20181221182339-f9677308dec2 h1:Ujru1hufTHVb++eG6OuNDKMxZnGIvF6o/u8q/8h2+I4=
 github.com/glycerine/go-unsnap-stream v0.0.0-20181221182339-f9677308dec2/go.mod h1:/20jfyN9Y5QPEAprSgKAUr+glWDY39ZiUEAYOEv5dsE=
 github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31 h1:gclg6gY70GLy3PbkQ1AERPfmLMMagS60DKF78eWwLn8=
@@ -24,21 +22,8 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGi
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e h1:KMs6SK8iDSR1+ZzOK10L5wGPpWDByyvOe5nrqk51g2U=
 github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e/go.mod h1:+AHfJo5+69p+fjvMJTmYajNP9rFBHQcaTDFmuXRRATI=
-github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
-github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
-github.com/mattn/go-isatty v0.0.10 h1:qxFzApOv4WsAL965uUPIsXzAKCZxN2p9UqdhFS4ZW10=
-github.com/mattn/go-isatty v0.0.10/go.mod h1:qgIWMr58cqv1PHHyhnkY9lrL7etaEgOFcMEpPG5Rm84=
-github.com/mattn/go-runewidth v0.0.2/go.mod h1:LwmH8dsx7+W8Uxz3IHJYH5QSwggIsqBzpuz5H//U1FU=
-github.com/mattn/go-runewidth v0.0.6 h1:V2iyH+aX9C5fsYCpK60U8BYIvmhqxuOL3JZcqc1NB7k=
-github.com/mattn/go-runewidth v0.0.6/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
-github.com/mattn/go-tty v0.0.3 h1:5OfyWorkyO7xP52Mq7tB36ajHDG5OHrmBGIS/DtakQI=
-github.com/mattn/go-tty v0.0.3/go.mod h1:ihxohKRERHTVzN+aSVRwACLCeqIoZAWpoICkkvrWyR0=
-github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7 h1:DpOJ2HYzCv8LZP15IdmG+YdwD2luVPHITV96TkirNBM=
-github.com/mitchellh/go-wordwrap v0.0.0-20150314170334-ad45545899c7/go.mod h1:ZXFpozHsX6DPmq2I0TCekCxypsnAUbP2oI0UX1GXzOo=
 github.com/mschoch/smat v0.0.0-20160514031455-90eadee771ae h1:VeRdUYdCw49yizlSbMEn2SZ+gT+3IUKx8BqxyQdz+BY=
 github.com/mschoch/smat v0.0.0-20160514031455-90eadee771ae/go.mod h1:qAyveg+e4CE+eKJXWVjKXM4ck2QobLqTDytGJbLLhJg=
-github.com/nsf/termbox-go v0.0.0-20190121233118-02980233997d h1:x3S6kxmy49zXVVyhcnrFqxvNVCBPb2KZ9hV2RBdS840=
-github.com/nsf/termbox-go v0.0.0-20190121233118-02980233997d/go.mod h1:IuKpRQcYE1Tfu+oAQqaLisqDeXgjyyltCfsaoYN18NQ=
 github.com/philhofer/fwd v1.0.0 h1:UbZqGr5Y38ApvM/V/jEljVxwocdweyH+vmYvRPBnbqQ=
 github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -50,30 +35,19 @@ github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
-github.com/superhawk610/bar v0.0.2 h1:Jf9+rXkpGgzJkpogpx+BGjC4EwkFZc8L7t8QBrmyfqY=
-github.com/superhawk610/bar v0.0.2/go.mod h1:xc6t0MG+8Mbj9wnHBEdvX2TRtNIaqhM5fW3R+dF9vH8=
-github.com/superhawk610/terminal v0.1.0 h1:OpWp0+861M3l5j4ZWprm8q0fgG9Y2xXZp5tkunvCm/M=
-github.com/superhawk610/terminal v0.1.0/go.mod h1:NQ3EEKWSeofexUwENLX3lv4WR2qeGiLlbf/NveZ7ZAQ=
 github.com/tinylib/msgp v1.1.0 h1:9fQd+ICuRIu/ue4vxJZu6/LzxN0HwMds2nq/0cFvxHU=
 github.com/tinylib/msgp v1.1.0/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
 github.com/willf/bitset v1.1.10 h1:NotGKqX0KwQ72NUzqrjZq5ipPNDQex9lo3WpaS8L2sc=
 github.com/willf/bitset v1.1.10/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9 h1:psW17arqaxU48Z5kZ0CQnkZWQJsqcURM6tKiBApRjXI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381 h1:VXak5I6aEWmAXeQjA+QSZzlgNrpq9mjcfDemuexIKsU=
 golang.org/x/net v0.0.0-20200707034311-ab3426394381/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA=
-golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894 h1:Cz4ceDQGXuKRnVBDTS23GTn/pU5OE2C0WrNTOYK1Uuc=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd h1:xhmwyvizuTgC2qz7ZlMluP20uW+C3Rm0FD/WLDX8884=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

internal/fetcher/fetcher.go

@@ -1,49 +0,0 @@
-package fetcher
-
-import (
-	"github.com/monkeyWie/gopeed-core/internal/controller"
-	"github.com/monkeyWie/gopeed-core/pkg/base"
-)
-
-// 对应协议的下载支持
-type Fetcher interface {
-	Setup(ctl controller.Controller)
-	// 解析请求
-	Resolve(req *base.Request) (res *base.Resource, err error)
-	// 创建任务
-	Create(res *base.Resource, opts *base.Options) (err error)
-	Start() (err error)
-	Pause() (err error)
-	Continue() (err error)
-
-	// 获取任务各个文件下载进度
-	Progress() Progress
-	// 该方法会一直阻塞，直到任务下载结束
-	Wait() (err error)
-}
-
-type DefaultFetcher struct {
-	Ctl    controller.Controller
-	DoneCh chan error
-}
-
-func (f *DefaultFetcher) Setup(ctl controller.Controller) {
-	f.Ctl = ctl
-	f.DoneCh = make(chan error, 1)
-}
-
-func (f *DefaultFetcher) Wait() (err error) {
-	return <-f.DoneCh
-}
-
-// 获取任务中各个文件的已下载字节数
-type Progress []int64
-
-// TotalDownloaded 获取任务总下载字节数
-func (p Progress) TotalDownloaded() int64 {
-	total := int64(0)
-	for _, d := range p {
-		total += d
-	}
-	return total
-}

internal/protocol/http/fetcher.go

@@ -1,390 +0,0 @@
-package http
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"github.com/monkeyWie/gopeed-core/internal/fetcher"
-	"github.com/monkeyWie/gopeed-core/pkg/base"
-	"golang.org/x/sync/errgroup"
-	"io"
-	"io/ioutil"
-	"mime"
-	"net/http"
-	"net/http/cookiejar"
-	"net/url"
-	"path"
-	"path/filepath"
-	"strconv"
-	"strings"
-	"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 {
-	*fetcher.DefaultFetcher
-
-	res     *base.Resource
-	opts    *base.Options
-	status  base.Status
-	clients []*http.Response
-	chunks  []*Chunk
-
-	ctx     context.Context
-	cancel  context.CancelFunc
-	pauseCh chan interface{}
-}
-
-func NewFetcher() *Fetcher {
-	return &Fetcher{
-		DefaultFetcher: new(fetcher.DefaultFetcher),
-		pauseCh:        make(chan interface{}),
-	}
-}
-
-var protocols = []string{"HTTP", "HTTPS"}
-
-func FetcherBuilder() ([]string, func() fetcher.Fetcher) {
-	return protocols, func() fetcher.Fetcher {
-		return NewFetcher()
-	}
-}
-
-func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) {
-	httpReq, err := buildRequest(nil, req)
-	if err != nil {
-		return nil, err
-	}
-	client := buildClient()
-	// 只访问一个字节，测试资源是否支持Range请求
-	httpReq.Header.Set(base.HttpHeaderRange, fmt.Sprintf(base.HttpHeaderRangeFormat, 0, 0))
-	httpResp, err := client.Do(httpReq)
-	if err != nil {
-		return nil, err
-	}
-	// 拿到响应头就关闭，不用加defer
-	httpResp.Body.Close()
-	res := &base.Resource{
-		Req:   req,
-		Range: false,
-		Files: []*base.FileInfo{},
-	}
-	if base.HttpCodePartialContent == httpResp.StatusCode {
-		// 返回206响应码表示支持断点下载
-		res.Range = true
-		// 解析资源大小: bytes 0-1000/1001 => 1001
-		contentTotal := path.Base(httpResp.Header.Get(base.HttpHeaderContentRange))
-		if contentTotal != "" {
-			parse, err := strconv.ParseInt(contentTotal, 10, 64)
-			if err != nil {
-				return nil, err
-			}
-			res.TotalSize = parse
-		}
-	} else if base.HttpCodeOK == httpResp.StatusCode {
-		// 返回200响应码，不支持断点下载，通过Content-Length头获取文件大小，获取不到的话可能是chunked编码
-		contentLength := httpResp.Header.Get(base.HttpHeaderContentLength)
-		if contentLength != "" {
-			parse, err := strconv.ParseInt(contentLength, 10, 64)
-			if err != nil {
-				return nil, err
-			}
-			res.TotalSize = parse
-		}
-	} else {
-		return nil, NewRequestError(httpResp.StatusCode, httpResp.Status)
-	}
-	file := &base.FileInfo{
-		Size: res.TotalSize,
-	}
-	contentDisposition := httpResp.Header.Get(base.HttpHeaderContentDisposition)
-	if contentDisposition != "" {
-		_, params, _ := mime.ParseMediaType(contentDisposition)
-		filename := params["filename"]
-		if filename != "" {
-			file.Name = filename
-		}
-	}
-	// Get file filename by URL
-	if file.Name == "" && strings.Count(req.URL, "/") > 2 {
-		file.Name = filepath.Base(req.URL)
-	}
-	// unknown file filename
-	if file.Name == "" {
-		file.Name = "unknown"
-	}
-	res.Files = append(res.Files, file)
-	return res, nil
-}
-
-func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error {
-	if opts.Connections != 1 && !res.Range {
-		opts.Connections = 1
-	}
-	f.res = res
-	f.opts = opts
-	f.status = base.DownloadStatusReady
-	return nil
-}
-
-func (f *Fetcher) Start() (err error) {
-	// 创建文件
-	name := f.filename()
-	_, err = f.Ctl.Touch(name, f.res.TotalSize)
-	if err != nil {
-		return err
-	}
-	f.status = base.DownloadStatusStart
-	if f.res.Range {
-		// 每个连接平均需要下载的分块大小
-		chunkSize := f.res.TotalSize / int64(f.opts.Connections)
-		f.chunks = make([]*Chunk, f.opts.Connections)
-		f.clients = make([]*http.Response, f.opts.Connections)
-		for i := 0; i < f.opts.Connections; i++ {
-			var (
-				begin = chunkSize * int64(i)
-				end   int64
-			)
-			if i == f.opts.Connections-1 {
-				// 最后一个分块需要保证把文件下载完
-				end = f.res.TotalSize - 1
-			} else {
-				end = begin + chunkSize - 1
-			}
-			chunk := NewChunk(begin, end)
-			f.chunks[i] = chunk
-		}
-	} else {
-		// 只支持单连接下载
-		f.chunks = make([]*Chunk, 1)
-		f.clients = make([]*http.Response, 1)
-		f.chunks[0] = NewChunk(0, 0)
-	}
-	f.fetch()
-	return
-}
-
-func (f *Fetcher) Pause() (err error) {
-	if base.DownloadStatusStart != f.status {
-		return
-	}
-	f.status = base.DownloadStatusPause
-	f.cancel()
-	<-f.pauseCh
-	return
-}
-
-func (f *Fetcher) Continue() (err error) {
-	if base.DownloadStatusStart == f.status || base.DownloadStatusDone == f.status {
-		return
-	}
-	f.status = base.DownloadStatusStart
-	var name = f.filename()
-	_, err = f.Ctl.Open(name)
-	if err != nil {
-		return err
-	}
-	f.fetch()
-	return
-}
-
-func (f *Fetcher) Progress() fetcher.Progress {
-	p := make(fetcher.Progress, 0)
-	if len(f.chunks) > 0 {
-		total := int64(0)
-		for _, chunk := range f.chunks {
-			total += chunk.Downloaded
-		}
-		p = append(p, total)
-	}
-	return p
-}
-
-func (f *Fetcher) filename() string {
-	// 创建文件
-	var filename = f.opts.Name
-	if filename == "" {
-		filename = f.res.Files[0].Name
-	}
-	return filepath.Join(f.opts.Path, filename)
-}
-
-func (f *Fetcher) fetch() {
-	f.ctx, f.cancel = context.WithCancel(context.Background())
-	eg, _ := errgroup.WithContext(f.ctx)
-	for i := 0; i < f.opts.Connections; i++ {
-		i := i
-		eg.Go(func() error {
-			return f.fetchChunk(i)
-		})
-	}
-
-	go func() {
-		err := eg.Wait()
-		// 下载停止，关闭文件句柄
-		f.Ctl.Close(f.filename())
-		if f.status == base.DownloadStatusPause {
-			f.pauseCh <- nil
-		} else {
-			if err != nil {
-				f.status = base.DownloadStatusError
-			} else {
-				f.status = base.DownloadStatusDone
-			}
-			f.DoneCh <- err
-		}
-	}()
-}
-
-func (f *Fetcher) fetchChunk(index int) (err error) {
-	filename := f.filename()
-	chunk := f.chunks[index]
-
-	httpReq, err := buildRequest(f.ctx, f.res.Req)
-	if err != nil {
-		return err
-	}
-	var (
-		client = buildClient()
-		buf    = make([]byte, 8192)
-	)
-	// 重试5次
-	for i := 0; i < 5; i++ {
-		// 如果下载完成直接返回
-		if chunk.Status == base.DownloadStatusDone {
-			return
-		}
-		// 如果已暂停直接跳出
-		if f.status == base.DownloadStatusPause {
-			break
-		}
-		var (
-			resp  *http.Response
-			retry bool
-		)
-		if f.res.Range {
-			httpReq.Header.Set(base.HttpHeaderRange,
-				fmt.Sprintf(base.HttpHeaderRangeFormat, chunk.Begin+chunk.Downloaded, chunk.End))
-		} else {
-			chunk.Downloaded = 0
-		}
-		err = func() error {
-			resp, err = client.Do(httpReq)
-			if err != nil {
-				return err
-			}
-			f.clients[index] = resp
-			if resp.StatusCode != base.HttpCodeOK && resp.StatusCode != base.HttpCodePartialContent {
-				err = NewRequestError(resp.StatusCode, resp.Status)
-				return err
-			}
-			return nil
-		}()
-		if err != nil {
-			// 请求失败重试
-			continue
-		}
-		// 请求成功就重置错误次数，连续失败5次才终止
-		i = 0
-		retry, err = func() (bool, error) {
-			defer resp.Body.Close()
-			for {
-				n, err := resp.Body.Read(buf)
-				if n > 0 {
-					_, err := f.Ctl.Write(filename, chunk.Begin+chunk.Downloaded, buf[:n])
-					if err != nil {
-						return false, err
-					}
-					chunk.Downloaded += int64(n)
-				}
-				if err != nil {
-					if err != io.EOF {
-						return true, err
-					}
-					break
-				}
-			}
-			return false, nil
-		}()
-		if !retry {
-			// 下载成功，跳出重试
-			break
-		}
-	}
-
-	if f.status == base.DownloadStatusPause {
-		chunk.Status = base.DownloadStatusPause
-	} else if chunk.Downloaded >= chunk.End-chunk.Begin+1 {
-		chunk.Status = base.DownloadStatusDone
-	} else {
-		if err != nil {
-			chunk.Status = base.DownloadStatusError
-		} else {
-			chunk.Status = base.DownloadStatusDone
-		}
-	}
-	return
-}
-
-func buildClient() *http.Client {
-	// Cookie handle
-	jar, _ := cookiejar.New(nil)
-	return &http.Client{
-		Jar:     jar,
-		Timeout: time.Second * 10,
-	}
-}
-
-func buildRequest(ctx context.Context, req *base.Request) (httpReq *http.Request, err error) {
-	url, err := url.Parse(req.URL)
-	if err != nil {
-		return
-	}
-
-	var (
-		method string
-		body   io.Reader
-	)
-	headers := make(map[string][]string)
-	if req.Extra == nil {
-		method = http.MethodGet
-	} else {
-		extra := req.Extra.(Extra)
-		if extra.Method != "" {
-			method = extra.Method
-		} else {
-			method = http.MethodGet
-		}
-		if len(extra.Header) > 0 {
-			for k, v := range extra.Header {
-				headers[k] = []string{v}
-			}
-		}
-		if extra.Body != "" {
-			body = ioutil.NopCloser(bytes.NewBufferString(extra.Body))
-		}
-	}
-
-	if ctx != nil {
-		httpReq, err = http.NewRequestWithContext(ctx, method, url.String(), body)
-	} else {
-		httpReq, err = http.NewRequest(method, url.String(), body)
-	}
-	if err != nil {
-		return
-	}
-	httpReq.Header = headers
-	return httpReq, nil
-}

internal/protocol/http/fetcher_test.go

@@ -1,168 +0,0 @@
-package http
-
-import (
-	"github.com/monkeyWie/gopeed-core/internal/controller"
-	"github.com/monkeyWie/gopeed-core/internal/fetcher"
-	"github.com/monkeyWie/gopeed-core/pkg/base"
-	"github.com/monkeyWie/gopeed-core/pkg/test"
-	"net"
-	"reflect"
-	"testing"
-	"time"
-)
-
-func TestFetcher_Resolve(t *testing.T) {
-	testResolve(test.StartTestFileServer, &base.Resource{
-		TotalSize: test.BuildSize,
-		Range:     true,
-		Files: []*base.FileInfo{
-			{
-				Name: test.BuildName,
-				Size: test.BuildSize,
-			},
-		},
-	}, t)
-	testResolve(test.StartTestChunkedServer, &base.Resource{
-		TotalSize: 0,
-		Range:     false,
-		Files: []*base.FileInfo{
-			{
-				Name: test.BuildName,
-				Size: 0,
-			},
-		},
-	}, t)
-}
-
-func testResolve(startTestServer func() net.Listener, want *base.Resource, t *testing.T) {
-	listener := startTestServer()
-	defer listener.Close()
-	fetcher := NewFetcher()
-	res, err := fetcher.Resolve(&base.Request{
-		URL: "http://" + listener.Addr().String() + "/" + test.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 := test.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 := test.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 := test.StartTestChunkedServer()
-	defer listener.Close()
-	// chunked编码下载
-	downloadNormal(listener, 1, t)
-	downloadContinue(listener, 1, t)
-}
-
-func TestFetcher_DownloadRetry(t *testing.T) {
-	listener := test.StartTestRetryServer()
-	defer listener.Close()
-	// chunked编码下载
-	downloadNormal(listener, 1, t)
-}
-
-func TestFetcher_DownloadError(t *testing.T) {
-	listener := test.StartTestErrorServer()
-	defer listener.Close()
-	// chunked编码下载
-	downloadError(listener, 1, t)
-}
-
-func downloadReady(listener net.Listener, connections int, t *testing.T) fetcher.Fetcher {
-	fetcher := NewFetcher()
-	fetcher.Setup(controller.NewController())
-	res, err := fetcher.Resolve(&base.Request{
-		URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = fetcher.Create(res, &base.Options{
-		Name:        test.DownloadName,
-		Path:        test.Dir,
-		Connections: connections,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-	return fetcher
-
-}
-
-func downloadNormal(listener net.Listener, connections int, t *testing.T) {
-	fetcher := downloadReady(listener, connections, t)
-	err := fetcher.Start()
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = fetcher.Wait()
-	if err != nil {
-		t.Fatal(err)
-	}
-	want := test.FileMd5(test.BuildFile)
-	got := test.FileMd5(test.DownloadFile)
-	if want != got {
-		t.Errorf("Download error = %v, want %v", got, want)
-	}
-}
-
-func downloadContinue(listener net.Listener, connections int, t *testing.T) {
-	fetcher := downloadReady(listener, connections, t)
-	err := fetcher.Start()
-	if err != nil {
-		t.Fatal(err)
-	}
-	time.Sleep(time.Millisecond * 200)
-	if err := fetcher.Pause(); err != nil {
-		t.Fatal(err)
-	}
-	time.Sleep(time.Millisecond * 200)
-	if err := fetcher.Continue(); err != nil {
-		t.Fatal(err)
-	}
-	err = fetcher.Wait()
-	if err != nil {
-		t.Fatal(err)
-	}
-	want := test.FileMd5(test.BuildFile)
-	got := test.FileMd5(test.DownloadFile)
-	if want != got {
-		t.Errorf("Download error = %v, want %v", got, want)
-	}
-}
-
-func downloadError(listener net.Listener, connections int, t *testing.T) {
-	fetcher := downloadReady(listener, connections, t)
-	err := fetcher.Start()
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = fetcher.Wait()
-	if err == nil {
-		t.Errorf("Download error = %v, want %v", err, nil)
-	}
-}

internal/protocol/http/model.go

@@ -1,24 +0,0 @@
-package http
-
-import "github.com/monkeyWie/gopeed-core/pkg/base"
-
-type Chunk struct {
-	Status     base.Status
-	Begin      int64
-	End        int64
-	Downloaded int64
-}
-
-func NewChunk(begin int64, end int64) *Chunk {
-	return &Chunk{
-		Status: base.DownloadStatusReady,
-		Begin:  begin,
-		End:    end,
-	}
-}
-
-type Extra struct {
-	Method string
-	Header map[string]string
-	Body   string
-}

main.go

@@ -1,23 +1,88 @@
 package main
 
 import (
+	"bufio"
 	"fmt"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/http"
+	"os"
+	"syscall"
+	"time"
+)
+
+var (
+	modkernel32          = syscall.NewLazyDLL("kernel32.dll")
+	procSetFileValidData = modkernel32.NewProc("SetFileValidData")
 )
 
 func main() {
-	request := &http.Request{
-		Method: "get",
-		URL:    "http://github.com/proxyee-down-org/proxyee-down/releases/download/3.4/proxyee-down-main.jar",
-		Header: map[string]string{
-			"Host":            "github.com",
-			"User-Agent":      "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.121 Safari/537.36",
-			"Accept":          "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
-			"Referer":         "http://github.com/proxyee-down-org/proxyee-down/releases",
-			"Accept-Encoding": "gzip, deflate, br",
-			"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
-		},
+	var err error
+	name := "d:/test.data"
+	var size int64 = 1024 * 1024 * 1024 * 4
+
+	err = os.Truncate(name, size)
+	if err != nil {
+		panic(err)
 	}
-	got, _ := http.Resolve(request)
-	fmt.Println(got)
+
+	var hwnd HWND
+
+	err = ShellExecute(hwnd,
+		"runas",
+		"fsutil.exe",
+		"file setValidData d:\\test.data 4294967296",
+		"d:\\",
+		syscall.SW_HIDE)
+
+	//err = exec.Command("fsutil", "file", "setValidData", "d:\\test.data", "4294967296").Run()
+	if err != nil {
+		panic(err)
+	}
+
+	file, _ := os.OpenFile(name, os.O_RDWR, 0644)
+	defer os.Remove(name)
+	defer file.Close()
+	defer func() {
+		reader := bufio.NewReader(os.Stdin)
+		fmt.Println("输入回车结束")
+		reader.ReadString('\n')
+	}()
+
+	/*process, err := syscall.GetCurrentProcess()
+	if err != nil {
+		panic(err)
+	}
+	var token syscall.Token
+	err = syscall.OpenProcessToken(process, syscall.TOKEN_ADJUST_PRIVILEGES|syscall.TOKEN_QUERY, &token)
+	if err != nil {
+		panic(err)
+	}
+
+
+	_, _, en := syscall.Syscall(procSetFileValidData.Addr(), 2, file.Fd(), uintptr(size), 0)
+	if en != 0 {
+		fmt.Println(en)
+		return
+	}*/
+
+	buf := []byte{1, 2, 3, 4, 5, 6, 7, 8}
+	var (
+		tm int64
+	)
+
+	tm = time.Now().UnixNano()
+	_, err = file.WriteAt(buf, 0)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Printf("write use:%d\n", toMill(time.Now().UnixNano()-tm))
+
+	tm = time.Now().UnixNano()
+	_, err = file.WriteAt(buf, 1024*1024*1024*2)
+	if err != nil {
+		panic(err)
+	}
+	fmt.Printf("write use:%d\n", toMill(time.Now().UnixNano()-tm))
+}
+
+func toMill(nano int64) int64 {
+	return nano / int64(time.Millisecond)
 }

pkg/base/model.go

@@ -1,36 +0,0 @@
-package base
-
-// 下载请求
-type Request struct {
-	// 下载链接
-	URL string
-	// 附加信息
-	Extra interface{}
-}
-
-// 资源信息
-type Resource struct {
-	Req *Request
-	// 资源总大小
-	TotalSize int64
-	// 是否支持断点下载
-	Range bool
-	// 资源所包含的文件列表
-	Files []*FileInfo
-}
-
-type FileInfo struct {
-	Name string
-	Path string
-	Size int64
-}
-
-// 下载选项
-type Options struct {
-	// 保存文件名
-	Name string
-	// 保存目录
-	Path string
-	// 并发连接数
-	Connections int
-}

pkg/download/downloader.go

@@ -1,194 +0,0 @@
-package download
-
-import (
-	"errors"
-	"github.com/google/uuid"
-	"github.com/monkeyWie/gopeed-core/internal/controller"
-	"github.com/monkeyWie/gopeed-core/internal/fetcher"
-	"github.com/monkeyWie/gopeed-core/internal/protocol/http"
-	"github.com/monkeyWie/gopeed-core/pkg/base"
-	"github.com/monkeyWie/gopeed-core/pkg/util"
-	"net/url"
-	"strings"
-	"sync"
-	"time"
-)
-
-type FetcherBuilder func() (protocols []string, builder func() fetcher.Fetcher)
-
-type TaskInfo struct {
-	ID       string
-	Res      *base.Resource
-	Opts     *base.Options
-	Status   base.Status
-	Files    map[string]*base.FileInfo
-	Progress *Progress
-
-	fetcher fetcher.Fetcher
-	timer   *util.Timer
-	locker  *sync.Mutex
-}
-
-type Progress struct {
-	// 下载耗时(纳秒)
-	Used int64
-	// 每秒下载字节数
-	Speed int64
-	// 已下载的字节数
-	Downloaded int64
-}
-
-type Downloader struct {
-	*controller.DefaultController
-	fetchBuilders map[string]func() fetcher.Fetcher
-	tasks         map[string]*TaskInfo
-	listener      func(taskInfo *TaskInfo, eventKey EventKey)
-}
-
-var DefaultDownloader = NewDownloader(http.FetcherBuilder)
-
-func NewDownloader(fbs ...FetcherBuilder) *Downloader {
-	d := &Downloader{DefaultController: controller.NewController()}
-	d.fetchBuilders = make(map[string]func() fetcher.Fetcher)
-	for _, f := range fbs {
-		protocols, builder := f()
-		for _, p := range protocols {
-			d.fetchBuilders[strings.ToUpper(p)] = builder
-		}
-	}
-	d.tasks = make(map[string]*TaskInfo)
-
-	// 每秒统计一次下载速度
-	go func() {
-		for {
-			if len(d.tasks) > 0 {
-				for _, task := range d.tasks {
-					if task.Status == base.DownloadStatusDone ||
-						task.Status == base.DownloadStatusError ||
-						task.Status == base.DownloadStatusPause {
-						continue
-					}
-					current := task.fetcher.Progress().TotalDownloaded()
-					task.Progress.Used = task.timer.Used()
-					task.Progress.Speed = current - task.Progress.Downloaded
-					task.Progress.Downloaded = current
-					d.emit(task, EventKeyProgress)
-				}
-			}
-			time.Sleep(time.Second)
-		}
-	}()
-	return d
-}
-
-func (d *Downloader) buildFetcher(URL string) (fetcher.Fetcher, error) {
-	url, err := url.Parse(URL)
-	if err != nil {
-		return nil, err
-	}
-	if fetchBuilder, ok := d.fetchBuilders[strings.ToUpper(url.Scheme)]; ok {
-		fetcher := fetchBuilder()
-		fetcher.Setup(d.DefaultController)
-		return fetcher, nil
-	}
-	return nil, errors.New("unsupported protocol")
-}
-
-func (d *Downloader) Resolve(req *base.Request) (*base.Resource, error) {
-	fetcher, err := d.buildFetcher(req.URL)
-	if err != nil {
-		return nil, err
-	}
-	return fetcher.Resolve(req)
-}
-
-func (d *Downloader) Create(res *base.Resource, opts *base.Options) (err error) {
-	fetcher, err := d.buildFetcher(res.Req.URL)
-	if err != nil {
-		return
-	}
-	err = fetcher.Create(res, opts)
-	if err != nil {
-		return
-	}
-	id := uuid.New().String()
-	task := &TaskInfo{
-		ID:       id,
-		Res:      res,
-		Opts:     opts,
-		Status:   base.DownloadStatusReady,
-		Progress: &Progress{},
-		fetcher:  fetcher,
-		timer:    &util.Timer{},
-		locker:   new(sync.Mutex),
-	}
-	d.tasks[id] = task
-	task.timer.Start()
-	d.emit(task, EventKeyStart)
-	err = fetcher.Start()
-	if err != nil {
-		d.emit(task, EventKeyError)
-		return
-	}
-	go func() {
-		err = fetcher.Wait()
-		if err != nil {
-			d.emit(task, EventKeyError)
-		} else {
-			task.Progress.Used = task.timer.Used()
-			if task.Res.TotalSize == 0 {
-				task.Res.TotalSize = task.fetcher.Progress().TotalDownloaded()
-			}
-			used := task.Progress.Used / int64(time.Second)
-			if used == 0 {
-				used = 1
-			}
-			task.Progress.Speed = task.Res.TotalSize / used
-			task.Progress.Downloaded = task.Res.TotalSize
-			d.emit(d.tasks[id], EventKeyDone)
-		}
-	}()
-	return
-}
-
-func (d *Downloader) Pause(id string) {
-	task := d.tasks[id]
-	task.locker.Lock()
-	defer task.locker.Unlock()
-	task.timer.Pause()
-	task.fetcher.Pause()
-	d.emit(task, EventKeyPause)
-}
-
-func (d *Downloader) Continue(id string) {
-	task := d.tasks[id]
-	task.locker.Lock()
-	defer task.locker.Unlock()
-	task.timer.Continue()
-	task.fetcher.Continue()
-	d.emit(task, EventKeyContinue)
-}
-
-func (d *Downloader) Listener(fn func(taskInfo *TaskInfo, eventKey EventKey)) {
-	d.listener = fn
-}
-
-func (d *Downloader) emit(taskInfo *TaskInfo, eventKey EventKey) {
-	if d.listener != nil {
-		d.listener(taskInfo, eventKey)
-	}
-}
-
-// Resolve 解析资源
-func Resolve(url string, extra ...interface{}) (*base.Resource, error) {
-	var e interface{}
-	if len(extra) == 0 {
-		e = nil
-	} else {
-		e = extra[0]
-	}
-	return DefaultDownloader.Resolve(&base.Request{
-		URL:   url,
-		Extra: e,
-	})
-}

pkg/download/downloader_test.go

@@ -1,75 +0,0 @@
-package download
-
-import (
-	"github.com/monkeyWie/gopeed-core/internal/protocol/http"
-	"github.com/monkeyWie/gopeed-core/pkg/base"
-	"github.com/monkeyWie/gopeed-core/pkg/test"
-	"reflect"
-	"sync"
-	"testing"
-)
-
-func TestDownloader_Resolve(t *testing.T) {
-	listener := test.StartTestFileServer()
-	defer listener.Close()
-
-	downloader := NewDownloader(http.FetcherBuilder)
-	req := &base.Request{
-		URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
-	}
-	res, err := downloader.Resolve(req)
-	if err != nil {
-		t.Fatal(err)
-	}
-	want := &base.Resource{
-		Req:       req,
-		TotalSize: test.BuildSize,
-		Range:     true,
-		Files: []*base.FileInfo{
-			{
-				Name: test.BuildName,
-				Path: "",
-				Size: test.BuildSize,
-			},
-		},
-	}
-	if !reflect.DeepEqual(res, want) {
-		t.Errorf("Resolve error = %s, want %s", test.ToJson(res), test.ToJson(want))
-	}
-}
-
-func TestDownloader_Create(t *testing.T) {
-	listener := test.StartTestFileServer()
-	defer listener.Close()
-
-	downloader := NewDownloader(http.FetcherBuilder)
-	req := &base.Request{
-		URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
-	}
-	res, err := downloader.Resolve(req)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(1)
-	downloader.Listener(func(taskInfo *TaskInfo, eventKey EventKey) {
-		if eventKey == EventKeyDone {
-			wg.Done()
-		}
-	})
-	err = downloader.Create(res, &base.Options{
-		Path:        test.Dir,
-		Name:        test.DownloadName,
-		Connections: 4,
-	})
-	if err != nil {
-		t.Fatal(err)
-	}
-	wg.Wait()
-	want := test.FileMd5(test.BuildFile)
-	got := test.FileMd5(test.DownloadFile)
-	if want != got {
-		t.Errorf("Download error = %v, want %v", got, want)
-	}
-}

pkg/download/event.go

@@ -1,17 +0,0 @@
-package download
-
-type EventKey string
-
-const (
-	EventKeyStart    = "start"
-	EventKeyPause    = "pause"
-	EventKeyContinue = "continue"
-	EventKeyProgress = "progress"
-	EventKeyError    = "error"
-	EventKeyDone     = "done"
-)
-
-type Event struct {
-	EventKey
-	TaskInfo
-}

pkg/test/httptest.go

@@ -1,137 +0,0 @@
-package test
-
-import (
-	"fmt"
-	"io"
-	"math/rand"
-	"net"
-	"net/http"
-	"os"
-)
-
-const (
-	BuildName = "build.data"
-	BuildSize = 200 * 1024 * 1024
-	Dir       = "./"
-	BuildFile = Dir + BuildName
-
-	DownloadName = "download.data"
-	DownloadFile = Dir + DownloadName
-)
-
-func StartTestFileServer() net.Listener {
-	return startTestServer(func() http.Handler {
-		return http.FileServer(http.Dir(Dir))
-	})
-}
-
-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
-}

pkg/test/util.go

@@ -1,32 +0,0 @@
-package test
-
-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)
-}

pkg/util/bytefmt.go

@@ -1,23 +0,0 @@
-package util
-
-import (
-	"fmt"
-	"math"
-)
-
-var unitArr = []string{"B", "KB", "MB", "GB", "TB", "EB"}
-
-func ByteFmt(size int64) string {
-	if size == 0 {
-		return "unknown"
-	}
-	fs := float64(size)
-	p := int(math.Log(fs) / math.Log(1024))
-	val := fs / math.Pow(1024, float64(p))
-	_, frac := math.Modf(val)
-	if frac > 0 {
-		return fmt.Sprintf("%.1f%s", math.Floor(val*10)/10, unitArr[p])
-	} else {
-		return fmt.Sprintf("%d%s", int(val), unitArr[p])
-	}
-}

pkg/util/bytefmt_test.go

@@ -1,57 +0,0 @@
-package util
-
-import "testing"
-
-func TestByteFmt(t *testing.T) {
-	type args struct {
-		size int64
-	}
-	tests := []struct {
-		name string
-		args args
-		want string
-	}{
-		{
-			name: "100B",
-			args: args{size: int64(100)},
-			want: "100B",
-		},
-		{
-			name: "1KB",
-			args: args{size: int64(1024)},
-			want: "1KB",
-		},
-		{
-			name: "1.9KB",
-			args: args{size: int64(1024*2 - 1)},
-			want: "1.9KB",
-		},
-		{
-			name: "2KB",
-			args: args{size: int64(1024 * 2)},
-			want: "2KB",
-		},
-		{
-			name: "1MB",
-			args: args{size: int64(1024 * 1024)},
-			want: "1MB",
-		},
-		{
-			name: "1.9MB",
-			args: args{size: int64(1024*1024*2 - 1)},
-			want: "1.9MB",
-		},
-		{
-			name: "2MB",
-			args: args{size: int64(1024 * 1024 * 2)},
-			want: "2MB",
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := ByteFmt(tt.args.size); got != tt.want {
-				t.Errorf("ByteFmt() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}

pkg/util/timer.go

@@ -1,25 +0,0 @@
-package util
-
-import "time"
-
-// 计时器
-type Timer struct {
-	t    int64
-	used int64
-}
-
-func (t *Timer) Start() {
-	t.t = time.Now().UnixNano()
-}
-
-func (t *Timer) Pause() {
-	t.used += time.Now().UnixNano() - t.t
-}
-
-func (t *Timer) Continue() {
-	t.t = time.Now().UnixNano()
-}
-
-func (t *Timer) Used() int64 {
-	return (time.Now().UnixNano() - t.t) + t.used
-}

protocol/bt/client.go

@@ -2,8 +2,8 @@ package bt
 
 import (
 	"crypto/rand"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/torrent"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/torrent"
 	"net/url"
 )
 

protocol/bt/torrent/main/main.go

@@ -1,9 +1,9 @@
 package main
 
 import (
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/peer"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/torrent"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/torrent"
 	log "github.com/sirupsen/logrus"
 )
 

protocol/bt/torrent/peer_conn.go

@@ -15,9 +15,9 @@ import (
 	"time"
 
 	"github.com/cenkalti/mse"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/peer"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/peer/message"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/peer/message"
 	log "github.com/sirupsen/logrus"
 )
 

protocol/bt/torrent/peer_conn_test.go

@@ -2,8 +2,8 @@ package torrent
 
 import (
 	"fmt"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/peer"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/tracker"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/tracker"
 	"os"
 	"sync"
 	"testing"

protocol/bt/torrent/peer_pool.go

@@ -4,8 +4,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/peer"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/tracker"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/tracker"
 )
 
 type peerPool struct {

protocol/bt/torrent/piece_state.go

@@ -1,7 +1,7 @@
 package torrent
 
 import (
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
 	"math"
 	"sync"
 

protocol/bt/torrent/torrent.go

@@ -4,7 +4,7 @@ import (
 	"errors"
 	"time"
 
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
 	log "github.com/sirupsen/logrus"
 )
 

protocol/bt/torrent/torrent_test.go

@@ -1,8 +1,8 @@
 package torrent
 
 import (
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/peer"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
 	log "github.com/sirupsen/logrus"
 	"testing"
 )

protocol/bt/tracker/tracker.go

@@ -15,8 +15,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/peer"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
 )
 
 const (

protocol/bt/tracker/tracker_test.go

@@ -2,7 +2,7 @@ package tracker
 
 import (
 	"fmt"
-	"github.com/monkeyWie/gopeed-core/pkg/protocol/bt/metainfo"
+	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
 	"testing"
 )
 

protocol/http/client.go

@@ -6,8 +6,8 @@ import (
 	"mime"
 	"net/http"
 	"net/http/cookiejar"
+	"net/url"
 	"os"
-	"path"
 	"strconv"
 	"strings"
 	"sync"
@@ -32,15 +32,21 @@ func Resolve(request *Request) (*Response, error) {
 	}
 	ret := &Response{}
 	// Get file name by "Content-Disposition"
-	if contentDisposition := response.Header.Get("Content-Disposition"); contentDisposition != "" {
+	contentDisposition := response.Header.Get("Content-Disposition")
+	if contentDisposition != "" {
 		_, params, _ := mime.ParseMediaType(contentDisposition)
-		if filename := params["filename"]; filename != "" {
+		filename := params["filename"]
+		if filename != "" {
 			ret.Name = filename
 		}
 	}
 	// Get file name by URL
 	if ret.Name == "" {
-		ret.Name = path.Base(httpRequest.URL.Path)
+		parse, err := url.Parse(httpRequest.URL.String())
+		if err == nil {
+			// e.g. /files/test.txt => test.txt
+			ret.Name = subLastSlash(parse.Path)
+		}
 	}
 	// Unknown file name
 	if ret.Name == "" {