ci: release on master branch

.github/release-drafter.yml

@@ -0,0 +1,35 @@
+name-template: 'v$NEXT_PATCH_VERSION 🌈'
+tag-template: 'v$NEXT_PATCH_VERSION'
+categories:
+  - title: '🆕 Features'
+    labels:
+      - 'feat'
+      - 'feature'
+      - 'enhancement'
+  - title: '🐛 Bug Fixes'
+    labels:
+      - 'fix'
+      - 'bugfix'
+      - 'bug'
+  - title: '🧰 Maintenance'
+    label: 'chore'
+  - title: '📖 Document'
+    label: 'doc'
+  - title: '🚀 CI/CD'
+    label: 'ci'
+change-template: '- $TITLE @$AUTHOR (#$NUMBER)'
+version-resolver:
+  major:
+    labels:
+      - 'major'
+  minor:
+    labels:
+      - 'minor'
+  patch:
+    labels:
+      - 'patch'
+  default: patch
+template: |
+  ## Changes
+
+  $CHANGES

.github/workflows/release.yml

@@ -0,0 +1,54 @@
+name: release
+
+on:
+  push:
+    branches:
+      - master
+jobs:
+  release:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '^1.15.2'
+      - uses: actions/checkout@v2
+      - name: Build Cli
+        run: |
+          CGO_ENABLED=0 GOOS=windows GOARCH=amd64 go build -ldflags="-w -s" -o gopeed.exe cmd/gopeed/*
+          CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -ldflags="-w -s" -o gopeed cmd/gopeed/*
+          CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build -ldflags="-w -s" -o gopeed cmd/gopeed/*
+          zip gopeed-windows-x64.zip gopeed.exe
+          zip gopeed-linux-x64.zip gopeed
+          zip gopeed-darwin-x64.zip gopeed
+      - uses: release-drafter/release-drafter@v5
+        id: create_release
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      - name: Upload Cli Windows
+        uses: actions/[email protected]
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }} # This pulls from the CREATE RELEASE step above, referencing it's ID to get its outputs object, which include a `upload_url`. See this blog post for more info: https://jasonet.co/posts/new-features-of-github-actions/#passing-data-to-future-steps
+          asset_path: ./gopeed-windows-x64.zip
+          asset_name: gopeed-${{ steps.create_release.outputs.name }}-windows-x64.zip
+          asset_content_type: application/zip
+      - name: Upload Cli Linux
+        uses: actions/[email protected]
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }} # This pulls from the CREATE RELEASE step above, referencing it's ID to get its outputs object, which include a `upload_url`. See this blog post for more info: https://jasonet.co/posts/new-features-of-github-actions/#passing-data-to-future-steps
+          asset_path: ./gopeed-linux-x64.zip
+          asset_name: gopeed-${{ steps.create_release.outputs.name }}-linux-x64.zip
+          asset_content_type: application/zip
+      - name: Upload Cli Darwin
+        uses: actions/[email protected]
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }} # This pulls from the CREATE RELEASE step above, referencing it's ID to get its outputs object, which include a `upload_url`. See this blog post for more info: https://jasonet.co/posts/new-features-of-github-actions/#passing-data-to-future-steps
+          asset_path: ./gopeed-darwin-x64.zip
+          asset_name: gopeed-${{ steps.create_release.outputs.name }}-darwin-x64.zip
+          asset_content_type: application/zip

.github/workflows/test.yml

@@ -0,0 +1,50 @@
+name: test
+
+on:
+  pull_request:
+    branches:
+      - dev
+  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 gopeed cmd/gopeed/*

.vscode/launch.json

@@ -1,17 +0,0 @@
-{
-    // 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

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

README.md

@@ -1,20 +1,46 @@
+[![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类库方式来使用。
+
+支持多协议（HTTP、BitTorrent）下载的客户端，提供命令行、RESTful API、WebSocket、Go 类库方式来使用。
+
+## 安装
+
+```sh
+go install github.com/monkeyWie/gopeed-core
+```
+
+## 示例
+
+可以在本仓库的[\_examples](_examples)目录下查看
+
+## 命令行工具
+
+### 预览
+
+![](_docs/img/cli-demo.gif)
+
+### 下载
+
+前往[releases](https://github.com/monkeyWie/gopeed-core/releases)页面
 
 ## TODO
-- [x] HTTP下载实现
-- [ ] BitTorrent下载实现
-    - [x] .torrent文件解析
-    - [x] tracker协议实现
-    - [x] peer wire protocol协议实现
-    - [ ] DHT协议实现
-    - [ ] 磁力链接支持
-    - [ ] uTP协议实现
-- [ ] 下载接口抽象(不关心具体协议)
+
+- [x] HTTP 下载实现
+- [ ] BitTorrent 下载实现
+  - [x] .torrent 文件解析
+  - [x] tracker 协议实现
+  - [x] peer wire protocol 协议实现
+  - [ ] DHT 协议实现
+  - [ ] 磁力链接支持
+  - [ ] uTP 协议实现
+- [x] 下载接口抽象(不关心具体协议)
 - [ ] 支持自定义配置
 - [ ] 限速功能实现
-- [ ] 命令行工具提供
-- [ ] RESTful服务提供
+- [x] 命令行工具提供
+- [ ] RESTful 服务提供
 
 ## 参与
-由于项目目前还未定型，代码可能随时有大的调整，所以暂不接受PR，当然如果有什么好的想法可以在issue区提出来。
+
+由于项目目前还未定型，代码可能随时有大的调整，所以暂不接受 PR，当然如果有什么好的想法可以在 issue 区提出来。

_examples/basic/main.go

@@ -0,0 +1,30 @@
+package main
+
+import (
+	"fmt"
+	"github.com/monkeyWie/gopeed-core/pkg/base"
+	"github.com/monkeyWie/gopeed-core/pkg/download"
+)
+
+func main() {
+	finallyCh := make(chan error)
+	err := download.Boot().
+		URL("https://www.baidu.com/index.html").
+		Listener(func(event *download.Event) {
+			if event.Key == download.EventKeyFinally {
+				finallyCh <- event.Err
+			}
+		}).
+		Create(&base.Options{
+			Connections: 8,
+		})
+	if err != nil {
+		panic(err)
+	}
+	err = <-finallyCh
+	if err != nil {
+		fmt.Printf("download fail:%v\n", err)
+	} else {
+		fmt.Println("download success")
+	}
+}

cmd/gopeed/flags.go

@@ -0,0 +1,42 @@
+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

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"fmt"
+	"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()
+
+	var wg sync.WaitGroup
+	err := download.Boot().
+		URL(args.url).
+		Listener(func(event *download.Event) {
+			if event.Key == download.EventKeyProgress {
+				printProgress(event.Task, "downloading...")
+			}
+			if event.Key == download.EventKeyFinally {
+				var title string
+				if event.Err != nil {
+					title = "fail"
+				} else {
+					title = "complete"
+				}
+				printProgress(event.Task, title)
+				fmt.Println()
+				if event.Err != nil {
+					gPrint("reason: " + event.Err.Error())
+				} else {
+					gPrint("saving file " + filepath.Join(*args.dir, event.Task.Res.Files[0].Name))
+				}
+				wg.Done()
+			}
+		}).
+		Create(&base.Options{
+			Path:        *args.dir,
+			Connections: *args.connections,
+		})
+	if err != nil {
+		panic(err)
+	}
+	wg.Add(1)
+	wg.Wait()
+}
+
+func printProgress(task *download.TaskInfo, title string) {
+	rate := float64(task.Progress.Downloaded) / float64(task.Res.TotalSize)
+	completeWidth := int(progressWidth * rate)
+	speed := util.ByteFmt(task.Progress.Speed)
+	totalSize := util.ByteFmt(task.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)
+}

download/common/fetcher.go

@@ -1,29 +0,0 @@
-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

@@ -1,81 +0,0 @@
-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

@@ -1,40 +0,0 @@
-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/http/fetcher.go

@@ -1,167 +0,0 @@
-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

@@ -1,307 +0,0 @@
-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

@@ -1,20 +0,0 @@
-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

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

download/http/process.go

@@ -1,245 +0,0 @@
-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

@@ -1,12 +1,9 @@
 module github.com/monkeyWie/gopeed-core
 
-go 1.13
+go 1.15
 
 require (
-	github.com/RoaringBitmap/roaring v0.4.21
-	github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972
 	github.com/google/uuid v1.1.1
-	github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e
-	github.com/sirupsen/logrus v1.4.2
 	golang.org/x/net v0.0.0-20200707034311-ab3426394381
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 )

go.sum

@@ -1,56 +1,15 @@
-github.com/RoaringBitmap/roaring v0.4.21 h1:WJ/zIlNX4wQZ9x8Ey33O1UaD9TCTakYsdLFSBcTwH+8=
-github.com/RoaringBitmap/roaring v0.4.21/go.mod h1:D0gp8kJQgE1A4LQ5wFLggQEyvDi06Mq5mKs52e1TwOo=
-github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972 h1:FvAasVrPm/VUO8M0odLDqA9Y7VbALYq3PRAwmWGPw78=
-github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972/go.mod h1:wGf99xKTJ8f/a8cC1inMqFz9cGEzISLLid5F6i5fI/g=
-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/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=
-github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31/go.mod h1:Ogl1Tioa0aV7gstGFO7KhffUsb9M4ydbEbbxpcEDc24=
-github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
-github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99 h1:twflg0XRTjwKpxb/jFExr4HGq6on2dEOmnL6FV+fgPw=
-github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
-github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
-github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
-github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk=
-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/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/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=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
-github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-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/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-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-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-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=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
-gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

internal/controller/controller.go

@@ -1,4 +1,4 @@
-package common
+package controller
 
 import (
 	"golang.org/x/net/proxy"
@@ -7,64 +7,29 @@ import (
 	"time"
 )
 
-// 下载请求
-type Request struct {
-	// 下载链接
-	URL string
-	// 附加信息
-	Extra interface{}
+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 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 Controller struct {
+type DefaultController struct {
 	Files map[string]*os.File
 }
 
-func NewController() *Controller {
-	return &Controller{Files: make(map[string]*os.File)}
+func NewController() *DefaultController {
+	return &DefaultController{Files: make(map[string]*os.File)}
 }
 
-func (c *Controller) Touch(name string, size int64) (file *os.File, err error) {
+func (c *DefaultController) Touch(name string, size int64) (file *os.File, err error) {
+	file, err = os.Create(name)
 	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
@@ -72,25 +37,25 @@ func (c *Controller) Touch(name string, size int64) (file *os.File, err error) {
 	return
 }
 
-func (c *Controller) Open(name string) (file *os.File, err error) {
-	file, err = os.OpenFile(name, os.O_RDWR, 0666)
+func (c *DefaultController) Open(name string) (file *os.File, err error) {
+	file, err = os.OpenFile(name, os.O_RDWR, os.ModePerm)
 	if err == nil {
 		c.Files[name] = file
 	}
 	return
 }
 
-func (c *Controller) Write(name string, offset int64, buf []byte) (int, error) {
+func (c *DefaultController) Write(name string, offset int64, buf []byte) (int, error) {
 	return c.Files[name].WriteAt(buf, offset)
 }
 
-func (c *Controller) Close(name string) error {
+func (c *DefaultController) Close(name string) error {
 	err := c.Files[name].Close()
 	delete(c.Files, name)
 	return err
 }
 
-func (c *Controller) ContextDialer() (proxy.Dialer, error) {
+func (c *DefaultController) ContextDialer() (proxy.Dialer, error) {
 	// return proxy.SOCKS5("tpc", "127.0.0.1:9999", nil, nil)
 	var dialer proxy.Dialer
 	return &DialerWarp{dialer: dialer}, nil

internal/fetcher/fetcher.go

@@ -0,0 +1,49 @@
+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

@@ -0,0 +1,387 @@
+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 {
+	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

@@ -0,0 +1,168 @@
+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

@@ -0,0 +1,24 @@
+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 +0,0 @@
-package main
-
-import (
-	"fmt"
-	"github.com/monkeyWie/gopeed-core/protocol/http"
-)
-
-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",
-		},
-	}
-	got, _ := http.Resolve(request)
-	fmt.Println(got)
-}

pkg/base/constants.go

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

pkg/base/model.go

@@ -0,0 +1,36 @@
+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

@@ -0,0 +1,233 @@
+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 Listener func(event *Event)
+
+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      Listener
+}
+
+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(EventKeyProgress, task)
+				}
+			}
+			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
+	}
+	if !res.Range || opts.Connections < 1 {
+		opts.Connections = 1
+	}
+	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(EventKeyStart, task)
+	err = fetcher.Start()
+	if err != nil {
+		return
+	}
+	go func() {
+		err = fetcher.Wait()
+		if err != nil {
+			d.emit(EventKeyError, task, err)
+		} 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(EventKeyDone, task)
+		}
+		d.emit(EventKeyFinally, task, err)
+	}()
+	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(EventKeyPause, task)
+}
+
+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(EventKeyContinue, task)
+}
+
+func (d *Downloader) Listener(fn Listener) {
+	d.listener = fn
+}
+
+func (d *Downloader) emit(eventKey EventKey, task *TaskInfo, errs ...error) {
+	if d.listener != nil {
+		var err error
+		if len(errs) > 0 {
+			err = errs[0]
+		}
+		d.listener(&Event{
+			Key:  eventKey,
+			Task: task,
+			Err:  err,
+		})
+	}
+}
+
+var defaultDownloader = NewDownloader(http.FetcherBuilder)
+
+type boot struct {
+	url      string
+	extra    interface{}
+	listener Listener
+}
+
+func (b *boot) URL(url string) *boot {
+	b.url = url
+	return b
+}
+
+func (b *boot) Extra(extra interface{}) *boot {
+	b.extra = extra
+	return b
+}
+
+func (b *boot) Resolve() (*base.Resource, error) {
+	return defaultDownloader.Resolve(&base.Request{
+		URL:   b.url,
+		Extra: b.extra,
+	})
+}
+
+func (b *boot) Listener(listener Listener) *boot {
+	b.listener = listener
+	return b
+}
+
+func (b *boot) Create(opts *base.Options) error {
+	res, err := b.Resolve()
+	if err != nil {
+		return err
+	}
+	defaultDownloader.Listener(b.listener)
+	return defaultDownloader.Create(res, opts)
+}
+
+func Boot() *boot {
+	return &boot{}
+}

pkg/download/downloader_test.go

@@ -0,0 +1,75 @@
+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(event *Event) {
+		if event.Key == 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

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

pkg/test/httptest.go

@@ -0,0 +1,137 @@
+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

@@ -0,0 +1,32 @@
+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

@@ -0,0 +1,23 @@
+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

@@ -0,0 +1,62 @@
+package util
+
+import "testing"
+
+func TestByteFmt(t *testing.T) {
+	type args struct {
+		size int64
+	}
+	tests := []struct {
+		name string
+		args args
+		want string
+	}{
+		{
+			name: "unknown",
+			args: args{size: int64(0)},
+			want: "unknown",
+		},
+		{
+			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

@@ -0,0 +1,25 @@
+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

@@ -1,57 +0,0 @@
-package bt
-
-import (
-	"crypto/rand"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/torrent"
-	"net/url"
-)
-
-type Client struct {
-	PeerID [20]byte
-
-	// torrents []torrent.Torrent
-}
-
-func NewClient() *Client {
-	client := &Client{}
-
-	// 生成Peer ID，规则为前三位固定字母(-GP)+SemVer(xyz)+End(-),后面随机生成
-	// 参考：https://wiki.theory.org/index.php/BitTorrentSpecification#peer_id
-	peerID := [20]byte{'-', 'G', 'P', '0', '0', '1', '-'}
-	_, err := rand.Read(peerID[7:])
-	if err != nil {
-		panic(err)
-	}
-	client.PeerID = peerID
-
-	return client
-}
-
-func (client *Client) AddTorrent(rawurl string) error {
-	parse, err := url.Parse(rawurl)
-	if err != nil {
-		return err
-	}
-	var metaInfo *metainfo.MetaInfo
-	switch parse.Scheme {
-	case "magnet":
-	// TODO 磁力链接解析
-	default:
-		metaInfo, err = metainfo.ParseFromFile(rawurl)
-		break
-	}
-	if err != nil {
-		return err
-	}
-
-	torrent := torrent.Torrent{
-		// client:   client,
-		MetaInfo: metaInfo,
-	}
-	// client.torrents = append(client.torrents, torrent)
-
-	torrent.Download("")
-
-	return nil
-}

protocol/bt/client_test.go

@@ -1,21 +0,0 @@
-package bt
-
-import (
-	"testing"
-)
-
-func TestClient_AddTorrent(t *testing.T) {
-	/*client := NewClient()
-	metaInfo, err := metainfo.ParseFromFile("testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
-	if err != nil {
-		panic(err)
-	}
-	torrent := &
-	torrent.MetaInfo.Announce = "udp://tracker.opentrackr.org:1337/announce"
-	torrent.MetaInfo.AnnounceList = [][]string{}
-	err = torrent.Tracker()
-	if err != nil {
-		panic(err)
-		return
-	}*/
-}

protocol/bt/dht/ping.go

@@ -1,34 +0,0 @@
-package dht
-
-import (
-	"github.com/marksamman/bencode"
-	"net"
-)
-
-func ping() string {
-	data := map[string]interface{}{
-		"t": "aa",
-		"y": "q",
-		"q": "ping",
-		"a": map[string]interface{}{"id": "abcdefghij0123456789"},
-	}
-	req := bencode.Encode(data)
-	// session.add_dht_router("router.utorrent.com", 6881)
-	// session.add_dht_router("router.bittorrent.com", 6881)
-	// session.add_dht_router("dht.transmissionbt.com", 6881)
-	// session.add_dht_router("router.bitcomet.com", 6881)
-	// session.add_dht_router("dht.aelitis.com", 6881)
-	dial, err := net.Dial("udp", "router.utorrent.com:6881")
-	if err != nil {
-		panic(err)
-	}
-	dial.Write(req)
-
-	reps := make([]byte, 65535)
-
-	n, err := dial.Read(reps)
-	if err != nil {
-		panic(err)
-	}
-	return string(reps[:n])
-}

protocol/bt/dht/ping_test.go

@@ -1,7 +0,0 @@
-package dht
-
-import "testing"
-
-func Test_ping(t *testing.T) {
-	println(ping())
-}

protocol/bt/metainfo/metainfo.go

@@ -1,144 +0,0 @@
-package metainfo
-
-import (
-	"crypto/sha1"
-	"encoding/json"
-	"os"
-
-	"github.com/marksamman/bencode"
-)
-
-type MetaInfo struct {
-	// tracker URL
-	// http://bittorrent.org/beps/bep_0003.html#metainfo-files
-	Announce string `json:"announce"`
-	// tracker URL
-	// http://bittorrent.org/beps/bep_0012.html
-	AnnounceList [][]string `json:"announce-list"`
-	Comment      string     `json:"comment"`
-	CreatedBy    string     `json:"created by"`
-	CreationDate int64      `json:"creation date"`
-	Encoding     string     `json:"encoding"`
-	// http://bittorrent.org/beps/bep_0019.html
-	UrlList []string `json:"url-list"`
-	Info    *Info    `json:"info"`
-
-	infoHash    [20]byte
-	totalSize   uint64
-	fileDetails []*FileDetail
-}
-
-type Info struct {
-	Name        string `json:"name"`
-	PieceLength int    `json:"piece length"`
-	Pieces      [][20]byte
-	// 单个文件时，length为该文件的长度
-	Length uint64 `json:"length"`
-	// 多个文件
-	Files []*File `json:"files"`
-}
-
-type File struct {
-	Length uint64   `json:"length"`
-	Path   []string `json:"path"`
-}
-
-type FileDetail struct {
-	Length uint64
-	Path   []string
-	Begin  int64
-	End    int64
-}
-
-type PieceMapping struct {
-	file  int
-	begin int64
-	end   int64
-}
-
-func ParseFromFile(path string) (*MetaInfo, error) {
-	file, err := os.Open(path)
-	if err != nil {
-		return nil, err
-	}
-	// http://bittorrent.org/beps/bep_0003.html#bencoding
-	btDecode, err := bencode.Decode(file)
-	if err != nil {
-		return nil, err
-	}
-	bytes, err := json.Marshal(btDecode)
-	if err != nil {
-		return nil, err
-	}
-	var metaInfo MetaInfo
-	json.Unmarshal(bytes, &metaInfo)
-	if metaInfo.Info != nil {
-		// encode info hash
-		info := btDecode["info"].(map[string]interface{})
-		infoBtEncode := bencode.Encode(info)
-		metaInfo.infoHash = sha1.Sum(infoBtEncode[:])
-		// Split pieces hash
-		pieces := info["pieces"].(string)
-		if pieces != "" {
-			metaInfo.Info.Pieces = make([][20]byte, len(pieces)/20)
-			// pieces parse
-			for i := range metaInfo.Info.Pieces {
-				begin := i * 20
-				metaInfo.Info.Pieces[i] = [20]byte{}
-				copy(metaInfo.Info.Pieces[i][:], pieces[begin:begin+20])
-			}
-		}
-	}
-	calcFileSize(&metaInfo)
-	return &metaInfo, nil
-}
-
-// 获取某个分片的大小
-func (metaInfo *MetaInfo) GetPieceLength(index int) int {
-	// 是否为最后一个分片
-	if index == len(metaInfo.Info.Pieces)-1 {
-		size := metaInfo.totalSize % uint64(metaInfo.Info.PieceLength)
-		if size > 0 {
-			return int(size)
-		}
-	}
-	return metaInfo.Info.PieceLength
-}
-
-// 获取所有文件的大小
-func (metaInfo *MetaInfo) GetTotalSize() uint64 {
-	return metaInfo.totalSize
-}
-
-// 获取info hash
-func (metaInfo *MetaInfo) GetInfoHash() [20]byte {
-	return metaInfo.infoHash
-}
-
-// 获取所有文件的开始偏移字节
-func (metaInfo *MetaInfo) GetFileDetails() []*FileDetail {
-	return metaInfo.fileDetails
-}
-
-// 计算所有文件的大小
-func calcFileSize(metaInfo *MetaInfo) {
-	if metaInfo.Info != nil {
-		fileCount := len(metaInfo.Info.Files)
-		if fileCount > 0 {
-			metaInfo.fileDetails = make([]*FileDetail, fileCount)
-			var length uint64
-			for i, f := range metaInfo.Info.Files {
-				metaInfo.fileDetails[i] = &FileDetail{
-					Length: f.Length,
-					Path:   f.Path,
-					Begin:  int64(length),
-					End:    int64(length + f.Length),
-				}
-				length += f.Length
-			}
-			metaInfo.totalSize = length
-		} else {
-			metaInfo.totalSize = metaInfo.Info.Length
-		}
-	}
-}

protocol/bt/metainfo/metainfo_test.go

@@ -1,22 +0,0 @@
-package metainfo
-
-import (
-	"encoding/hex"
-	"fmt"
-	"testing"
-)
-
-func TestParseFromFile(t *testing.T) {
-	// metaInfo, err := ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
-	metaInfo, err := ParseFromFile("E:\\bt\\win10.torrent")
-	if err != nil {
-		panic(err)
-	}
-	fmt.Printf("%+v\n", metaInfo.AnnounceList)
-	for _, file := range metaInfo.Info.Files {
-		fmt.Printf("%+v\n", file)
-	}
-	fmt.Printf("%+v\n", metaInfo.Info.PieceLength)
-	fmt.Printf("%+v\n", len(metaInfo.Info.Pieces))
-	fmt.Printf("%+v\n", hex.EncodeToString(metaInfo.infoHash[:]))
-}

protocol/bt/peer/handshake.go

@@ -1,51 +0,0 @@
-package peer
-
-import (
-	"bytes"
-	"encoding/binary"
-)
-
-const (
-	ProtocolIdentifier       = "BitTorrent protocol"
-	ProtocolIdentifierLength = 0x13
-)
-
-// Handshake of Peer wire protocol
-// Per https://wiki.theory.org/index.php/BitTorrentSpecification#Handshake
-type Handshake struct {
-	Pstrlen  byte
-	Pstr     [19]byte
-	Reserved [8]byte
-	InfoHash [20]byte
-	PeerID   [20]byte
-}
-
-func NewHandshake(reserved [8]byte, infoHash [20]byte, peerID [20]byte) *Handshake {
-	var arr [ProtocolIdentifierLength]byte
-	copy(arr[:], ProtocolIdentifier)
-	return &Handshake{
-		Pstrlen:  ProtocolIdentifierLength,
-		Pstr:     arr,
-		Reserved: reserved,
-		InfoHash: infoHash,
-		PeerID:   peerID,
-	}
-}
-
-func (handshake *Handshake) Encode() []byte {
-	writer := new(bytes.Buffer)
-	err := binary.Write(writer, binary.BigEndian, handshake)
-	if err != nil {
-		panic(err)
-	}
-	return writer.Bytes()
-}
-
-func (handshake *Handshake) Decode(buf []byte) error {
-	reader := bytes.NewReader(buf)
-	err := binary.Read(reader, binary.BigEndian, handshake)
-	if err != nil {
-		return err
-	}
-	return nil
-}

protocol/bt/peer/message/message.go

@@ -1,143 +0,0 @@
-package message
-
-import (
-	"encoding/binary"
-)
-
-// message ID
-type ID byte
-
-const (
-	IdChoke ID = iota
-	IdUnchoke
-	IdInterested
-	IdNotInterested
-	IdHave
-	IdBitfield
-	IdRequest
-	IdPiece
-	IdCancel
-)
-
-/*type Message struct {
-	Length  uint32
-	ID      MessageID
-	Payload []byte
-}*/
-
-type base struct {
-	id ID
-}
-
-func (base *base) ID() ID {
-	return base.id
-}
-
-func (base *base) Encode() []byte {
-	return encode(base, nil)
-}
-
-func (base *base) Decode(body []byte) {
-
-}
-
-type Message interface {
-	ID() ID
-	Encode() []byte
-	Decode(body []byte)
-}
-
-type TestMsg struct {
-	*base
-}
-
-func (base *TestMsg) Encode() []byte {
-	return []byte{1, 2, 3}
-}
-
-func (base *TestMsg) Decode(body []byte) {
-}
-
-func encode(base *base, body []byte) []byte {
-	buf := make([]byte, 5+len(body))
-	binary.BigEndian.PutUint32(buf, uint32(len(buf)))
-	buf[4] = byte(base.id)
-	copy(buf[4:], body)
-	return buf
-}
-
-/*func Decode(buf []byte) Message {
-	head := make([]byte, 5)
-	body := message.Encode()
-	binary.BigEndian.PutUint32(head, uint32(len(head)+len(body)))
-	head[4] = byte(message.ID())
-	return append(head, body...)
-}
-
-func (msg *Message) ToBytes() []byte {
-	buf := make([]byte, 5)
-	binary.BigEndian.PutUint32(buf, msg.Length)
-	buf[4] = byte(msg.ID)
-	if msg.Payload != nil {
-		return append(buf, msg.Payload.Encode()...)
-	} else {
-		return buf
-	}
-}
-
-func (msg *Message) FormBytes(buf []byte) {
-	msg.Length = binary.BigEndian.Uint32(buf)
-	msg.ID = ID(buf[4])
-	if msg.Payload != nil {
-		msg.Payload.Decode(buf[4:])
-	}
-}*/
-
-/*func NewMessage(id MessageID, Payload []byte) *Message {
-	message := &Message{ID: id, Payload: Payload}
-	if id == Keepalive {
-		message.Length = 0
-	} else {
-		message.Length = uint32(len(Payload) + 1)
-	}
-	return message
-}
-
-func (msg *Message) Encode() []byte {
-	if msg.ID == Keepalive {
-		return make([]byte, 4)
-	} else {
-		buf := make([]byte, 5+len(msg.Payload))
-		binary.BigEndian.PutUint32(buf, msg.Length)
-		buf[4] = byte(msg.ID)
-		copy(buf[5:], msg.Payload[:])
-		return buf
-	}
-}
-
-func (msg *Message) Decode(buf []byte) {
-	msg.Length = binary.BigEndian.Uint32(buf)
-	if msg.Length == 0 {
-		msg.ID = -1
-	} else {
-		msg.ID = MessageID(buf[4])
-		if msg.Length > 4 {
-			msg.Payload = buf[5:]
-		}
-	}
-}*/
-
-// Message protocol
-// Length prefix| message ID | Payload
-// 4-byte       | 1-byte     | remaining bytes
-// 100          | 5          | 99-bytes
-// https://wiki.theory.org/index.php/BitTorrentSpecification#Messages
-func SplitMessage(data []byte, atEOF bool) (advance int, token []byte, err error) {
-	if !atEOF && len(data) > 4 {
-		length := int(binary.BigEndian.Uint32(data))
-		if len(data)-4 >= length {
-			return length + 4, data[:length+4], nil
-		}
-	}
-	return
-}

protocol/bt/peer/message/msg_bitfield.go

@@ -1,68 +0,0 @@
-package message
-
-import (
-	"github.com/RoaringBitmap/roaring"
-	"math"
-)
-
-// bitfield: <len=0001+X><id=5><bitfield>
-type Bitfield struct {
-	*base
-	pieceCount int
-	bitmap     *roaring.Bitmap
-}
-
-func NewBitfield() *Bitfield {
-	return &Bitfield{
-		base: &base{
-			id: IdBitfield,
-		},
-	}
-}
-
-func BuildBitfield(pieceCount int, bitmap *roaring.Bitmap) *Bitfield {
-	bitfield := NewBitfield()
-	bitfield.pieceCount = pieceCount
-	bitfield.bitmap = bitmap
-	return bitfield
-}
-
-func (b *Bitfield) Encode() []byte {
-	buf := make([]byte, int(math.Ceil(float64(b.pieceCount)/8)))
-	for i := 0; i < len(buf); i++ {
-		temp := byte(0)
-		for j := 0; j < 8; j++ {
-			// 如果piece存在,位数标识为1
-			if b.bitmap.ContainsInt(i*8 + j) {
-				temp = temp | (1 << (7 - j))
-			}
-		}
-		buf[i] = temp
-	}
-	return encode(b.base, buf)
-}
-
-func (b *Bitfield) Decode(body []byte) {
-	b.bitmap = roaring.New()
-	// 初始化bitmap
-	for i := 0; i < len(body); i++ {
-		for j := 0; j < 8; j++ {
-			if body[i]&(1<<(7-j)) > 0 {
-				b.bitmap.AddInt(i*8 + j)
-			}
-		}
-	}
-}
-
-// 某个分片是否下载完成
-func (b *Bitfield) IsComplete(i int) bool {
-	return b.bitmap.ContainsInt(i)
-}
-
-// 给定一组分片下载状态，计算出当前peer能提供下载的分片下标，如果已经拥有则不用提供
-// has = 0b01111111,had = 0b10010000
-// has & ~had = 0b01111111 & 0b01101111 = 0b01101111
-func (b *Bitfield) Provide(had *roaring.Bitmap) []uint32 {
-	xor := roaring.AndNot(b.bitmap, had)
-	return xor.ToArray()
-}

protocol/bt/peer/message/msg_bitfield_test.go

@@ -1,111 +0,0 @@
-package message
-
-import (
-	"github.com/RoaringBitmap/roaring"
-	"reflect"
-	"testing"
-)
-
-func TestBitfield_IsComplete(t *testing.T) {
-	type fields struct {
-		payload *roaring.Bitmap
-	}
-	type args struct {
-		i int
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-		want   bool
-	}{
-		{
-			"case-1",
-			fields{
-				payload: roaring.BitmapOf(1, 2, 3, 4),
-			},
-			args{
-				1,
-			},
-			true,
-		},
-		{
-			"case-2",
-			fields{
-				payload: roaring.BitmapOf(1, 2, 3, 4),
-			},
-			args{
-				5,
-			},
-			false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			b := &Bitfield{
-				pieceCount: 100,
-				bitmap:     tt.fields.payload,
-			}
-			if got := b.IsComplete(tt.args.i); got != tt.want {
-				t.Errorf("Bitfield.IsComplete() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestBitfield_Provide(t *testing.T) {
-	type fields struct {
-		payload *roaring.Bitmap
-	}
-	type args struct {
-		had *roaring.Bitmap
-	}
-	tests := []struct {
-		name   string
-		fields fields
-		args   args
-		want   []uint32
-	}{
-		{
-			"case-1",
-			fields{
-				payload: roaring.BitmapOf(0, 1, 2, 3, 4),
-			},
-			args{
-				roaring.BitmapOf(0),
-			},
-			[]uint32{1, 2, 3, 4},
-		},
-		{
-			"case-2",
-			fields{
-				payload: roaring.BitmapOf(0, 1, 2, 3, 4, 5, 6, 7, 8),
-			},
-			args{
-				roaring.BitmapOf(0, 1, 3, 5, 6, 8),
-			},
-			[]uint32{2, 4, 7},
-		},
-		{
-			"case-3",
-			fields{
-				payload: roaring.BitmapOf(),
-			},
-			args{
-				roaring.BitmapOf(0, 1, 2, 3, 4, 5, 6, 7, 8, 9),
-			},
-			[]uint32{},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			b := &Bitfield{
-				pieceCount: 100,
-				bitmap:     tt.fields.payload,
-			}
-			if got := b.Provide(tt.args.had); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("Bitfield.Provide() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}

protocol/bt/peer/message/msg_choke.go

@@ -1,14 +0,0 @@
-package message
-
-// choke: <len=0001><id=0>
-type Choke struct {
-	*base
-}
-
-func NewChoke() *Choke {
-	return &Choke{
-		base: &base{
-			id: IdChoke,
-		},
-	}
-}

protocol/bt/peer/message/msg_interested.go

@@ -1,14 +0,0 @@
-package message
-
-// choke: <len=0001><id=2>
-type Interested struct {
-	*base
-}
-
-func NewInterested() *Interested {
-	return &Interested{
-		base: &base{
-			id: IdInterested,
-		},
-	}
-}

protocol/bt/peer/message/msg_keepalive.go

@@ -1,16 +0,0 @@
-package message
-
-// 	keep-alive: <len=0000>
-type Keepalive struct {
-	*base
-}
-
-var data = make([]byte, 4)
-
-func (k *Keepalive) Encode() []byte {
-	return data
-}
-
-func (k *Keepalive) Decode(body []byte) {
-
-}

protocol/bt/peer/message/msg_not_interested.go

@@ -1,14 +0,0 @@
-package message
-
-// choke: <len=0001><id=3>
-type NotInterested struct {
-	*base
-}
-
-func NewNotInterested() *NotInterested {
-	return &NotInterested{
-		base: &base{
-			id: IdNotInterested,
-		},
-	}
-}

protocol/bt/peer/message/msg_piece.go

@@ -1,42 +0,0 @@
-package message
-
-import "encoding/binary"
-
-// piece: <len=0009+X><id=7><index><begin><block>
-type Piece struct {
-	*base
-	Index uint32
-	Begin uint32
-	Block []byte
-}
-
-func NewPiece() *Piece {
-	return &Piece{
-		base: &base{
-			id: IdPiece,
-		},
-	}
-}
-
-func BuildPiece(index uint32, begin uint32, block []byte) *Piece {
-	piece := NewPiece()
-	piece.Index = index
-	piece.Begin = begin
-	piece.Block = block
-	return piece
-
-}
-
-func (p *Piece) Encode() []byte {
-	buf := make([]byte, 8+len(p.Block))
-	binary.BigEndian.PutUint32(buf[0:4], p.Index)
-	binary.BigEndian.PutUint32(buf[4:8], p.Begin)
-	copy(buf[8:], p.Block)
-	return encode(p.base, buf)
-}
-
-func (p *Piece) Decode(buf []byte) {
-	p.Index = binary.BigEndian.Uint32(buf[0:4])
-	p.Begin = binary.BigEndian.Uint32(buf[4:8])
-	p.Block = buf[8:]
-}

protocol/bt/peer/message/msg_request.go

@@ -1,43 +0,0 @@
-package message
-
-import (
-	"encoding/binary"
-)
-
-// request: <len=0013><id=6><index><begin><length>
-type Request struct {
-	*base
-	Index  uint32
-	Begin  uint32
-	Length uint32
-}
-
-func NewRequest() *Request {
-	return &Request{
-		base: &base{
-			id: IdRequest,
-		},
-	}
-}
-
-func BuildRequest(index uint32, begin uint32, length uint32) *Request {
-	request := NewRequest()
-	request.Index = index
-	request.Begin = begin
-	request.Length = length
-	return request
-}
-
-func (r *Request) Encode() []byte {
-	buf := make([]byte, 12)
-	binary.BigEndian.PutUint32(buf[0:4], r.Index)
-	binary.BigEndian.PutUint32(buf[4:8], r.Begin)
-	binary.BigEndian.PutUint32(buf[8:12], r.Length)
-	return encode(r.base, buf)
-}
-
-func (r *Request) Decode(buf []byte) {
-	r.Length = binary.BigEndian.Uint32(buf[0:4])
-	r.Begin = binary.BigEndian.Uint32(buf[4:8])
-	r.Index = binary.BigEndian.Uint32(buf[8:12])
-}

protocol/bt/peer/message/msg_unchoke.go

@@ -1,14 +0,0 @@
-package message
-
-// 	unchoke: <len=0001><id=1>
-type Unchoke struct {
-	*base
-}
-
-func NewUnchoke() *Unchoke {
-	return &Unchoke{
-		base: &base{
-			id: IdUnchoke,
-		},
-	}
-}

protocol/bt/peer/peer.go

@@ -1,30 +0,0 @@
-package peer
-
-import (
-	"encoding/binary"
-	"math/rand"
-	"net"
-	"strconv"
-)
-
-type Peer struct {
-	IP   uint32
-	Port uint16
-}
-
-func (peer *Peer) Address() string {
-	bts := make([]byte, 4)
-	binary.BigEndian.PutUint32(bts, peer.IP)
-	return net.IP(bts).String() + ":" + strconv.Itoa(int(peer.Port))
-}
-
-// 生成Peer ID，规则为前三位固定字母(-GP)+SemVer(xyz)+End(-),后面随机生成
-// 参考：https://wiki.theory.org/index.php/BitTorrentSpecification#peer_id
-func GenPeerID() [20]byte {
-	peerID := [20]byte{'-', 'G', 'P', '0', '0', '1', '-'}
-	_, err := rand.Read(peerID[7:])
-	if err != nil {
-		panic(err)
-	}
-	return peerID
-}

protocol/bt/torrent/main/main.go

@@ -1,74 +0,0 @@
-package main
-
-import (
-	"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"
-)
-
-func main() {
-	log.SetLevel(log.DebugLevel)
-	torrent := buildTorrent()
-	torrent.Download("e:/testbt/download")
-}
-
-func buildTorrent() *torrent.Torrent {
-	// metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
-	metaInfo, err := metainfo.ParseFromFile("../../testdata/office.torrent")
-	if err != nil {
-		panic(err)
-	}
-
-	metaInfo.AnnounceList[0] = append(metaInfo.AnnounceList[0], []string{
-		"udp://tracker.coppersurfer.tk:6969/announce",
-		"udp://tracker.leechers-paradise.org:6969/announce",
-		"udp://tracker.opentrackr.org:1337/announce",
-		"udp://p4p.arenabg.com:1337/announce",
-		"udp://9.rarbg.to:2710/announce",
-		"udp://9.rarbg.me:2710/announce",
-		"udp://tracker.internetwarriors.net:1337/announce",
-		"udp://exodus.desync.com:6969/announce",
-		"udp://tracker.tiny-vps.com:6969/announce",
-		"udp://tracker.sbsub.com:2710/announce",
-		"udp://retracker.lanta-net.ru:2710/announce",
-		"udp://open.stealth.si:80/announce",
-		"udp://open.demonii.si:1337/announce",
-		"udp://tracker.torrent.eu.org:451/announce",
-		"udp://tracker.moeking.me:6969/announce",
-		"udp://tracker.cyberia.is:6969/announce",
-		"udp://denis.stalker.upeer.me:6969/announce",
-		"udp://ipv4.tracker.harry.lu:80/announce",
-		"udp://explodie.org:6969/announce",
-		"udp://zephir.monocul.us:6969/announce",
-		"udp://xxxtor.com:2710/announce",
-		"udp://valakas.rollo.dnsabr.com:2710/announce",
-		"udp://tracker.zerobytes.xyz:1337/announce",
-		"udp://tracker.yoshi210.com:6969/announce",
-		"udp://tracker.uw0.xyz:6969/announce",
-		"udp://tracker.swateam.org.uk:2710/announce",
-		"udp://tracker.nyaa.uk:6969/announce",
-		"udp://tracker.filemail.com:6969/announce",
-		"udp://tracker.ds.is:6969/announce",
-		"udp://retracker.sevstar.net:2710/announce",
-		"udp://retracker.netbynet.ru:2710/announce",
-		"udp://retracker.akado-ural.ru:80/announce",
-		"udp://opentracker.i2p.rocks:6969/announce",
-		"udp://opentor.org:2710/announce",
-		"udp://open.nyap2p.com:6969/announce",
-		"udp://chihaya.toss.li:9696/announce",
-		"udp://bt2.archive.org:6969/announce",
-		"udp://bt1.archive.org:6969/announce",
-		"udp://www.loushao.net:8080/announce",
-		"udp://tracker4.itzmx.com:2710/announce",
-		"udp://tracker3.itzmx.com:6961/announce",
-		"udp://tracker2.itzmx.com:6961/announce",
-		"udp://tracker.lelux.fi:6969/announce",
-		"udp://tracker.kamigami.org:2710/announce",
-		"udp://tracker.dler.org:6969/announce",
-		"udp://tr.bangumi.moe:6969/announce",
-		"udp://qg.lorzl.gq:2710/announce",
-		"udp://bt2.54new.com:8080/announce",
-	}...)
-	return torrent.NewTorrent(peer.GenPeerID(), metaInfo)
-}

protocol/bt/torrent/peer_conn.go

@@ -1,464 +0,0 @@
-package torrent
-
-import (
-	"bufio"
-	"crypto/sha1"
-	"encoding/binary"
-	"errors"
-	"fmt"
-	"github.com/RoaringBitmap/roaring"
-	"io"
-	"net"
-	"os"
-	"path"
-	"path/filepath"
-	"time"
-
-	"github.com/cenkalti/mse"
-	"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"
-)
-
-// 现在主流客户端的block大小都是16KB
-const blockSize = 2 << 13
-const keepaliveTimeout = 60 * 2
-
-var errPieceCheckFailed = errors.New("piece check failed")
-var keepaliveData = make([]byte, 4)
-
-type peerConn struct {
-	torrent *Torrent
-	peer    *peer.Peer
-	conn    net.Conn
-	// this client is choking the peer
-	amChoking bool
-	// this client is interested in the peer
-	amInterested bool
-	// peer is choking this client
-	peerChoking bool
-	// peer is interested in this client
-	peerInterested bool
-
-	// 上一次收到包的时间
-	lastReciveTime int64
-
-	bitfield *message.Bitfield
-	readyEnd bool
-
-	downloadedCh chan error
-	disconnectCh chan error
-	// block下载队列，官方推荐为5
-	blockQueueCh chan interface{}
-}
-
-func newPeerConn(torrent *Torrent, peer *peer.Peer) *peerConn {
-	return &peerConn{
-		torrent: torrent,
-		peer:    peer,
-	}
-}
-
-// 使用MSE加密来避免运营商对bt流量的封锁，基本上现在市面上BT客户端都默认开启了，不用MSE的话很多Peer拒绝连接
-// see http://wiki.vuze.com/w/Message_Stream_Encryption
-func (pc *peerConn) dialMse() error {
-	conn, err := net.DialTimeout("tcp", pc.peer.Address(), time.Second*time.Duration(30))
-	if err != nil {
-		return err
-	}
-	mseConn := mse.WrapConn(conn)
-	infoHash := pc.torrent.MetaInfo.GetInfoHash()
-	_, err = mseConn.HandshakeOutgoing(infoHash[:], mse.PlainText, nil)
-	if err != nil {
-		mseConn.Close()
-		return err
-	}
-	pc.conn = mseConn
-	return nil
-}
-
-// Handshake of Peer wire protocol
-// see https://wiki.theory.org/index.php/BitTorrentSpecification#Handshake
-func (pc *peerConn) handshake() (*peer.Handshake, error) {
-	handshakeRes, err := func() (*peer.Handshake, error) {
-		handshakeReq := peer.NewHandshake([8]byte{}, pc.torrent.MetaInfo.GetInfoHash(), pc.torrent.PeerID)
-		_, err := pc.conn.Write(handshakeReq.Encode())
-		if err != nil {
-			return nil, err
-		}
-		var read [68]byte
-		_, err = io.ReadFull(pc.conn, read[:])
-		if err != nil {
-			return nil, err
-		}
-		handshakeRes := &peer.Handshake{}
-		err = handshakeRes.Decode(read[:])
-		if err != nil {
-			return nil, err
-		}
-		// InfoHash不匹配
-		if handshakeRes.InfoHash != handshakeReq.InfoHash {
-			return nil, fmt.Errorf("info_hash not currently serving")
-		}
-		return handshakeRes, nil
-	}()
-	if err != nil {
-		pc.conn.Close()
-		return nil, err
-	}
-	// init state
-	pc.amChoking = true
-	pc.amInterested = false
-	pc.peerChoking = true
-	pc.peerInterested = false
-	return handshakeRes, nil
-}
-
-func (pc *peerConn) handleKeepalive() {
-	pc.lastReciveTime = time.Now().Unix()
-	for {
-		time.Sleep(time.Minute * 2)
-		// 如果超过两分钟没有响应，则断开连接
-		if time.Now().Unix()-pc.lastReciveTime > keepaliveTimeout {
-			pc.conn.Close()
-			break
-		}
-		// 两分钟发送一次心跳包
-		_, err := pc.conn.Write(keepaliveData)
-		if err != nil {
-			pc.conn.Close()
-			break
-		}
-	}
-}
-
-// 准备下载
-func (pc *peerConn) ready() error {
-	if err := pc.dialMse(); err != nil {
-		return fmt.Errorf("tcp dial error %w", err)
-	}
-	if _, err := pc.handshake(); err != nil {
-		return fmt.Errorf("handshake error %w", err)
-	}
-	pc.readyEnd = false
-	readyCh := make(chan bool)
-	pc.disconnectCh = make(chan error)
-	go pc.handleKeepalive()
-	go func() {
-		scanner := bufio.NewScanner(pc.conn)
-		scanner.Split(message.SplitMessage)
-		for scanner.Scan() {
-			pc.lastReciveTime = time.Now().Unix()
-			buf := scanner.Bytes()
-			length := binary.BigEndian.Uint32(buf[:4])
-			if length == 0 {
-				// 	keepalive
-			} else {
-				switch message.ID(buf[4]) {
-				case message.IdChoke:
-					break
-				case message.IdUnchoke:
-					pc.handleUnchoke(readyCh)
-					break
-				case message.IdInterested:
-					break
-				case message.IdNotInterested:
-					break
-				case message.IdHave:
-					break
-				case message.IdBitfield:
-					pc.handleBitfield(buf)
-					break
-				case message.IdRequest:
-					break
-				case message.IdPiece:
-					pc.handlePiece(buf)
-					break
-				case message.IdCancel:
-					break
-				}
-			}
-		}
-		// 还未下载完成时连接断开
-		err := scanner.Err()
-		if err == nil {
-			err = io.EOF
-		}
-		pc.disconnectCh <- err
-		close(pc.disconnectCh)
-	}()
-	err := func() error {
-		select {
-		case status := <-readyCh:
-			if status {
-				return nil
-			}
-			return errors.New("ready fail")
-		case <-time.After(time.Second * 30):
-			// 30秒之后超时
-			return errors.New("ready time out")
-		}
-	}()
-	if err != nil {
-		pc.conn.Close()
-		return err
-	}
-	return nil
-}
-
-// 下载指定piece
-func (pc *peerConn) downloadPiece(index int) (err error) {
-	pieceLength := pc.torrent.MetaInfo.GetPieceLength(index)
-	pc.downloadedCh = make(chan error)
-	pc.blockQueueCh = make(chan interface{}, 5)
-	defer close(pc.blockQueueCh)
-	// 按块下载分片
-	blockCount := pc.torrent.pieceStates.states[index].blockCount
-	for i := 0; i < blockCount; i++ {
-		offset := blockSize * i
-		// 如果已经下载过就跳过
-		if pc.torrent.pieceStates.isBlockDone(index, offset) {
-			continue
-		}
-
-		var blockLength uint32
-		if i == blockCount-1 {
-			// 最后一个block大小需要计算出来
-			blockLength = uint32(pieceLength - offset)
-		} else {
-			blockLength = blockSize
-		}
-		// block下载排队
-		select {
-		case pc.blockQueueCh <- nil:
-			break
-		case err = <-pc.downloadedCh:
-			break
-		case err = <-pc.disconnectCh:
-			break
-		}
-		// 如果连接出现问题或下载失败直接返回异常
-		if err != nil {
-			pc.conn.Close()
-			return
-		}
-		// 发起request，对方会响应piece
-		_, err = pc.conn.Write(message.BuildRequest(uint32(index), uint32(offset), blockLength).Encode())
-		if err != nil {
-			break
-		}
-	}
-	select {
-	case err = <-pc.downloadedCh:
-		break
-	case err = <-pc.disconnectCh:
-		break
-	}
-	if err != nil {
-		pc.conn.Close()
-	}
-	return
-}
-
-func (pc *peerConn) handleUnchoke(readyCh chan<- bool) {
-	pc.peerChoking = false
-	// 已经处理过Unchoke信号
-	if pc.readyEnd {
-		return
-	}
-	pc.readyEnd = true
-	// 如果客户端对peer感兴趣并且peer没有choked客户端，就可以开始下载了
-	if pc.amInterested {
-		readyCh <- true
-	} else {
-		readyCh <- false
-	}
-	close(readyCh)
-}
-
-func (pc *peerConn) handleBitfield(buf []byte) {
-	pc.bitfield = message.NewBitfield()
-	pc.bitfield.Decode(buf)
-	have := pc.getHavePieces(pc.bitfield)
-	if len(have) > 0 {
-		// 表示对该peer感兴趣，并且不choked该peer
-		pc.conn.Write(message.NewInterested().Encode())
-		pc.amInterested = true
-
-		pc.conn.Write(message.NewUnchoke().Encode())
-		pc.amChoking = false
-	} else {
-		pc.conn.Close()
-	}
-}
-
-// 处理下载响应，每次接收到响应直接将block写入到对应文件中
-func (pc *peerConn) handlePiece(buf []byte) {
-	piece := message.NewPiece()
-	piece.Decode(buf)
-	info := pc.torrent.MetaInfo.Info
-	fds := pc.torrent.MetaInfo.GetFileDetails()
-	blockLength := int64(len(piece.Block))
-	pieceBegin := int64(piece.Index) * int64(info.PieceLength)
-	blockBegin := pieceBegin + int64(piece.Begin)
-	// 计算block对应要写到的文件偏移
-	var fileBlocks []*fileBlock
-	if len(info.Files) == 0 {
-		// 单文件
-		fileBlocks = append(fileBlocks, &fileBlock{
-			filepath:   info.Name,
-			fileSeek:   blockBegin,
-			blockBegin: 0,
-			blockEnd:   blockLength,
-		})
-	} else {
-		// 获取要写入的第一个文件和偏移
-		writeIndex := getWriteFile(blockBegin, fds)
-		// block对应种子的偏移
-		blockFileBegin := blockBegin
-		// block对应文件的偏移
-		var blockSeek int64 = 0
-		for _, f := range fds[writeIndex:] {
-			// 计算文件可剩余写入字节数
-			fileWritable := f.End - blockFileBegin
-			// 计算block剩余写入字节数
-			blockWritable := blockLength - blockSeek
-			fb := &fileBlock{
-				filepath:   filepath.Join(f.Path...),
-				fileSeek:   blockFileBegin - f.Begin,
-				blockBegin: blockSeek,
-			}
-			if fileWritable >= blockWritable {
-				// 若够block写入直接跳出循环
-				fb.blockEnd = blockSeek + blockWritable
-				fileBlocks = append(fileBlocks, fb)
-				break
-			} else {
-				// 否则计算剩余可写字节数，写入到下一个文件中
-				fb.blockEnd = blockSeek + fileWritable
-				fileBlocks = append(fileBlocks, fb)
-				blockFileBegin = f.End
-				blockSeek += fileWritable
-			}
-		}
-	}
-	// 开始写入文件
-	for _, f := range fileBlocks {
-		err := func() error {
-			name := filepath.Join(pc.torrent.Path, f.filepath)
-			file, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, 0644)
-			if err != nil {
-				return err
-			}
-			defer file.Close()
-			_, err = file.WriteAt(piece.Block[f.blockBegin:f.blockEnd], f.fileSeek)
-			if err != nil {
-				return err
-			}
-			return nil
-		}()
-		if err != nil {
-			pc.downloadedCh <- errPieceCheckFailed
-			pc.conn.Close()
-			return
-		}
-	}
-	pc.torrent.pieceStates.setBlockDone(int(piece.Index), int(piece.Begin))
-	// 出队
-	<-pc.blockQueueCh
-
-	log.Debugf("piece:%d block:%d", piece.Index, piece.Begin)
-
-	// piece全部下载完
-	if pc.torrent.pieceStates.isPieceDone(int(piece.Index)) {
-		// 计算piece对应的文件偏移
-		sha1 := sha1.New()
-		pieceLength := pc.torrent.MetaInfo.GetPieceLength(int(piece.Index))
-		writeIndex := getWriteFile(pieceBegin, fds)
-		fileBegin := pieceBegin - fds[writeIndex].Begin
-		for _, fd := range fds[writeIndex:] {
-			func() {
-				file, err := os.Open(path.Join(pc.torrent.Path, fd.Path[0]))
-				if err != nil {
-					panic(err)
-				}
-				defer file.Close()
-				_, err = file.Seek(fileBegin, 0)
-				if err != nil {
-					panic(err)
-				}
-				written, err := io.CopyN(sha1, file, int64(pieceLength))
-				if err != nil {
-					if err != io.EOF {
-						panic(err)
-					}
-				}
-				pieceLength -= int(written)
-			}()
-			if pieceLength > 0 {
-				// 	继续读下个文件
-				fileBegin = 0
-			} else {
-				break
-			}
-		}
-		// 断开连接 TODO 用连接池进行复用
-		pc.conn.Close()
-
-		// 校验piece SHA-1 hash
-		downHash := [20]byte{}
-		copy(downHash[:], sha1.Sum(nil))
-		if downHash == pc.torrent.MetaInfo.Info.Pieces[piece.Index] {
-			// piece下载完成
-			pc.downloadedCh <- nil
-		} else {
-			pc.downloadedCh <- errPieceCheckFailed
-		}
-		close(pc.downloadedCh)
-	}
-}
-
-// 获取piece对应要写入的文件
-// TODO 使用sort.Search二分查找法检索
-func getWriteFile(pieceBegin int64, fds []*metainfo.FileDetail) int {
-	for i, f := range fds {
-		if f.Begin <= pieceBegin && f.End > pieceBegin {
-			return i
-		}
-	}
-	return -1
-}
-
-type fileBlock struct {
-	filepath   string
-	fileSeek   int64
-	blockBegin int64
-	blockEnd   int64
-}
-
-// 获取peer能提供需要下载的文件分片
-func (pc *peerConn) getHavePieces(bitfield *message.Bitfield) []uint32 {
-	had := roaring.New()
-	for i := 0; i < pc.torrent.pieceStates.size(); i++ {
-		if pc.torrent.pieceStates.getState(i) == stateFinish {
-			had.AddInt(i)
-		}
-	}
-	return bitfield.Provide(had)
-}
-
-// 获取要写入到的文件
-/*func (ps *peerConn) getWriteFile(request *message.Request) string {
-	info := ps.torrent.MetaInfo.Info
-	// 单文件
-	if len(info.Files) == 0 {
-		return info.Name
-	} else {
-		request.Index * info.PieceLength + be
-		for i := 0; i < len(info.Files); i++ {
-
-		}
-	}
-}
-*/

protocol/bt/torrent/peer_conn_test.go

@@ -1,122 +0,0 @@
-package torrent
-
-import (
-	"fmt"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/tracker"
-	"os"
-	"sync"
-	"testing"
-	"time"
-)
-
-func Test_peerState_ready(t *testing.T) {
-
-	ps := getUsablePeer()
-	fmt.Println("get a peer")
-
-	// 下载前准备
-	err := ps.ready()
-	fmt.Println("ready end")
-	if err != nil {
-		fmt.Println(ps.peer.Address())
-		t.Fatal(err)
-	}
-
-}
-
-func Test_peerState_downloadPiece(t *testing.T) {
-
-	ps := getUsablePeer()
-	fmt.Println("get a peer")
-
-	// 下载前准备
-	err := ps.ready()
-	if err != nil {
-		t.Fatal(err)
-	}
-	have := ps.getHavePieces(ps.bitfield)
-	fmt.Printf("download is ready,have %d\n", len(have))
-	if len(have) > 0 {
-		// 获取分片的长度
-		for _, index := range have {
-			fmt.Printf("download index %d\n", index)
-			ps.downloadPiece(int(index))
-		}
-		return
-	}
-}
-
-// 获取一个能连接上的peer
-func getUsablePeer() *peerConn {
-	ch := getUsablePeerMore()
-	ps := <-ch
-	return ps
-}
-
-// 获取多个能连接上的peer
-func getUsablePeerMore() chan *peerConn {
-	torrent := buildTorrent()
-	tracker := &tracker.Tracker{
-		PeerID:   torrent.PeerID,
-		MetaInfo: torrent.MetaInfo,
-	}
-	peers := <-tracker.Tracker()
-	ch := make(chan *peerConn)
-	for i := range peers {
-		go peerTest(torrent, peers[i], ch)
-	}
-	return ch
-}
-
-func peerTest(torrent *Torrent, peer *peer.Peer, ch chan *peerConn) {
-	ps := &peerConn{
-		torrent: torrent,
-		peer:    peer,
-	}
-	// 连接至peer
-	err := ps.ready()
-	if err != nil {
-		return
-	}
-	ps.conn.Close()
-	ch <- ps
-}
-
-func TestSome(t *testing.T) {
-	ch := make(chan interface{})
-
-	go func() {
-		time.Sleep(time.Second)
-		select {
-		case ch <- nil:
-			fmt.Println("write")
-		case <-time.After(time.Second * 3):
-			fmt.Println("timeout")
-		}
-	}()
-
-	time.Sleep(time.Second * 5)
-	r := <-ch
-	fmt.Println(r)
-	fmt.Println("111")
-
-}
-
-func TestSome2(t *testing.T) {
-	wg := sync.WaitGroup{}
-	wg.Add(10)
-	file, err := os.OpenFile("e:/testbt/test.data", os.O_RDWR|os.O_CREATE, 0644)
-	for i := 0; i < 10; i++ {
-		go func(i int) {
-			for j := 0; j < 100; j++ {
-				if err != nil {
-					panic(err)
-				}
-				_, err = file.WriteAt([]byte{byte(i)}, int64(i*8))
-			}
-			wg.Done()
-		}(i)
-	}
-	wg.Wait()
-}

protocol/bt/torrent/peer_pool.go

@@ -1,99 +0,0 @@
-package torrent
-
-import (
-	"sync"
-	"time"
-
-	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/tracker"
-)
-
-type peerPool struct {
-	torrent *Torrent
-
-	states map[*peer.Peer]*peerState
-	lock   *sync.Mutex
-}
-
-type peerState struct {
-	// 是否在被使用
-	using bool
-	// 下载失败次数
-	errors int
-}
-
-func newPeerPool(torrent *Torrent) *peerPool {
-	return &peerPool{
-		torrent: torrent,
-		lock:    &sync.Mutex{},
-		states:  map[*peer.Peer]*peerState{},
-	}
-}
-
-func (pp *peerPool) fetch() {
-	tracker := &tracker.Tracker{
-		PeerID:   pp.torrent.PeerID,
-		MetaInfo: pp.torrent.MetaInfo,
-	}
-
-	go func() {
-		for {
-			// 当peer数量少于200个时重新发起一次tracker
-			if len(pp.states) < 200 {
-				go func() {
-					for peers := range tracker.Tracker() {
-						pp.put(peers)
-					}
-				}()
-			}
-			// 每2分钟检测一次
-			time.Sleep(time.Minute * 2)
-		}
-	}()
-}
-
-func (pp *peerPool) put(peers []*peer.Peer) {
-	pp.lock.Lock()
-	defer pp.lock.Unlock()
-	for _, peer := range peers {
-		if _, ok := pp.states[peer]; !ok {
-			pp.states[peer] = &peerState{
-				using:  false,
-				errors: 0,
-			}
-		}
-
-	}
-}
-
-func (pp *peerPool) get() *peer.Peer {
-	pp.lock.Lock()
-	defer pp.lock.Unlock()
-	for peer, state := range pp.states {
-		if !state.using {
-			state.using = true
-			return peer
-		}
-	}
-	return nil
-}
-
-// 将peer重新放回池中，等待使用
-func (pp *peerPool) release(peer *peer.Peer) {
-	pp.lock.Lock()
-	defer pp.lock.Unlock()
-	pp.states[peer].using = false
-	pp.states[peer].errors = 0
-}
-
-// 标记peer为不可用，超过3次则剔除该peer
-func (pp *peerPool) unavailable(peer *peer.Peer) {
-	pp.lock.Lock()
-	defer pp.lock.Unlock()
-	pp.states[peer].using = false
-	pp.states[peer].errors++
-	if pp.states[peer].errors > 3 {
-		delete(pp.states, peer)
-	}
-
-}

protocol/bt/torrent/piece_state.go

@@ -1,109 +0,0 @@
-package torrent
-
-import (
-	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
-	"math"
-	"sync"
-
-	"github.com/RoaringBitmap/roaring"
-)
-
-type state int
-
-const (
-	stateReady = iota
-	stateDownloading
-	stateFinish
-)
-
-// 所有piece的下载状态
-type pieceStates struct {
-	lock *sync.RWMutex
-	// 所有piece的block下载进度
-	states map[int]*pieceState
-}
-
-type pieceState struct {
-	state state
-	// piece中block的个数
-	blockCount int
-	// piece中block的下载记录
-	blockRecord *roaring.Bitmap
-}
-
-func newPiecesState(meta *metainfo.MetaInfo) *pieceStates {
-	pieceCount := len(meta.Info.Pieces)
-	ps := &pieceStates{
-		lock:   &sync.RWMutex{},
-		states: make(map[int]*pieceState, pieceCount),
-	}
-	for i := 0; i < pieceCount; i++ {
-		ps.states[i] = &pieceState{
-			state:       stateReady,
-			blockCount:  int(math.Ceil(float64(meta.GetPieceLength(i)) / blockSize)),
-			blockRecord: roaring.New(),
-		}
-	}
-	return ps
-}
-
-func (ps *pieceStates) getState(index int) state {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
-	return ps.states[index].state
-}
-
-func (ps *pieceStates) setState(index int, state state) {
-	ps.lock.Lock()
-	defer ps.lock.Unlock()
-	ps.states[index].state = state
-}
-
-func (ps *pieceStates) size() int {
-	return len(ps.states)
-}
-
-func (ps *pieceStates) getLeft() int {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
-	count := 0
-	for _, v := range ps.states {
-		if v.state == stateReady || v.state == stateDownloading {
-			count++
-		}
-	}
-	return count
-}
-
-func (ps *pieceStates) isDone() bool {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
-	for _, v := range ps.states {
-		if v.state == stateReady || v.state == stateDownloading {
-			return false
-		}
-	}
-	return true
-}
-
-func (ps *pieceStates) isPieceDone(index int) bool {
-	ps.lock.RLock()
-	defer ps.lock.RUnlock()
-	state := ps.states[index]
-	return uint64(state.blockCount) == state.blockRecord.GetCardinality()
-}
-
-// 指定piece下的某个block是否下载完成
-func (ps *pieceStates) isBlockDone(pieceIndex int, blockIndex int) bool {
-	return ps.states[pieceIndex].blockRecord.ContainsInt(blockIndex)
-}
-
-// 指定piece下的某个block下载完成
-func (ps *pieceStates) setBlockDone(pieceIndex int, blockIndex int) {
-	ps.states[pieceIndex].blockRecord.AddInt(blockIndex)
-}
-
-// 清空指定piece下所有的block下载记录
-func (ps *pieceStates) clearBlocks(index int) {
-	ps.states[index].blockRecord.Clear()
-}

protocol/bt/torrent/torrent.go

@@ -1,83 +0,0 @@
-package torrent
-
-import (
-	"errors"
-	"time"
-
-	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
-	log "github.com/sirupsen/logrus"
-)
-
-type Torrent struct {
-	PeerID   [20]byte
-	MetaInfo *metainfo.MetaInfo
-
-	pieceStates *pieceStates
-	peerPool    *peerPool
-	Path        string
-}
-
-func NewTorrent(peerID [20]byte, metaInfo *metainfo.MetaInfo) *Torrent {
-	torrent := &Torrent{
-		PeerID:      peerID,
-		MetaInfo:    metaInfo,
-		pieceStates: newPiecesState(metaInfo),
-	}
-	return torrent
-}
-
-func (t *Torrent) Download(path string) {
-	t.Path = path
-	t.peerPool = newPeerPool(t)
-	t.peerPool.fetch()
-
-	pieceQueueCh := make(chan interface{}, 128)
-	defer close(pieceQueueCh)
-	downloadedCh := make(chan interface{})
-	defer close(downloadedCh)
-
-	for i := 0; i < len(t.MetaInfo.Info.Pieces); i++ {
-		// 加入下载队列
-		pieceQueueCh <- nil
-		go func(index int) {
-			t.pieceStates.setState(index, stateDownloading)
-			// 一直下载到成功
-			for {
-				peer := t.peerPool.get()
-				if peer == nil {
-					time.Sleep(time.Second * 10)
-					continue
-				}
-				pc := newPeerConn(t, peer)
-				err := pc.ready()
-				if err != nil {
-					t.peerPool.unavailable(peer)
-				} else {
-					err = pc.downloadPiece(index)
-					if err != nil {
-						log.Debugf("piece %d download fail:%s", index, err.Error())
-						// 下载失败
-						t.peerPool.unavailable(peer)
-						if errors.Is(err, errPieceCheckFailed) {
-							// piece校验失败，需要重新下载过一遍，清空block记录
-							t.pieceStates.clearBlocks(index)
-						}
-						continue
-					} else {
-						// 下载完成
-						t.pieceStates.setState(index, stateFinish)
-						t.peerPool.release(peer)
-						log.Debugf("piece %d download success,total:%d,left:%d", index, t.pieceStates.size(), t.pieceStates.getLeft())
-						if t.pieceStates.isDone() {
-							downloadedCh <- nil
-						}
-						break
-					}
-				}
-			}
-			<-pieceQueueCh
-		}(i)
-	}
-	<-downloadedCh
-	log.Debug("download end")
-}

protocol/bt/torrent/torrent_test.go

@@ -1,77 +0,0 @@
-package torrent
-
-import (
-	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
-	log "github.com/sirupsen/logrus"
-	"testing"
-)
-
-func init() {
-	log.SetLevel(log.DebugLevel)
-}
-
-func TestTorrent_Download(t *testing.T) {
-	torrent := buildTorrent()
-	torrent.Download("e:/testbt/download")
-}
-
-func buildTorrent() *Torrent {
-	// metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
-	metaInfo, err := metainfo.ParseFromFile("../testdata/office.torrent")
-	if err != nil {
-		panic(err)
-	}
-
-	metaInfo.AnnounceList[0] = append(metaInfo.AnnounceList[0], []string{
-		"udp://tracker.coppersurfer.tk:6969/announce",
-		"udp://tracker.leechers-paradise.org:6969/announce",
-		"udp://tracker.opentrackr.org:1337/announce",
-		"udp://p4p.arenabg.com:1337/announce",
-		"udp://9.rarbg.to:2710/announce",
-		"udp://9.rarbg.me:2710/announce",
-		"udp://tracker.internetwarriors.net:1337/announce",
-		"udp://exodus.desync.com:6969/announce",
-		"udp://tracker.tiny-vps.com:6969/announce",
-		"udp://tracker.sbsub.com:2710/announce",
-		"udp://retracker.lanta-net.ru:2710/announce",
-		"udp://open.stealth.si:80/announce",
-		"udp://open.demonii.si:1337/announce",
-		"udp://tracker.torrent.eu.org:451/announce",
-		"udp://tracker.moeking.me:6969/announce",
-		"udp://tracker.cyberia.is:6969/announce",
-		"udp://denis.stalker.upeer.me:6969/announce",
-		"udp://ipv4.tracker.harry.lu:80/announce",
-		"udp://explodie.org:6969/announce",
-		"udp://zephir.monocul.us:6969/announce",
-		"udp://xxxtor.com:2710/announce",
-		"udp://valakas.rollo.dnsabr.com:2710/announce",
-		"udp://tracker.zerobytes.xyz:1337/announce",
-		"udp://tracker.yoshi210.com:6969/announce",
-		"udp://tracker.uw0.xyz:6969/announce",
-		"udp://tracker.swateam.org.uk:2710/announce",
-		"udp://tracker.nyaa.uk:6969/announce",
-		"udp://tracker.filemail.com:6969/announce",
-		"udp://tracker.ds.is:6969/announce",
-		"udp://retracker.sevstar.net:2710/announce",
-		"udp://retracker.netbynet.ru:2710/announce",
-		"udp://retracker.akado-ural.ru:80/announce",
-		"udp://opentracker.i2p.rocks:6969/announce",
-		"udp://opentor.org:2710/announce",
-		"udp://open.nyap2p.com:6969/announce",
-		"udp://chihaya.toss.li:9696/announce",
-		"udp://bt2.archive.org:6969/announce",
-		"udp://bt1.archive.org:6969/announce",
-		"udp://www.loushao.net:8080/announce",
-		"udp://tracker4.itzmx.com:2710/announce",
-		"udp://tracker3.itzmx.com:6961/announce",
-		"udp://tracker2.itzmx.com:6961/announce",
-		"udp://tracker.lelux.fi:6969/announce",
-		"udp://tracker.kamigami.org:2710/announce",
-		"udp://tracker.dler.org:6969/announce",
-		"udp://tr.bangumi.moe:6969/announce",
-		"udp://qg.lorzl.gq:2710/announce",
-		"udp://bt2.54new.com:8080/announce",
-	}...)
-	return NewTorrent(peer.GenPeerID(), metaInfo)
-}

protocol/bt/tracker/errors.go

@@ -1,29 +0,0 @@
-package tracker
-
-import (
-	"fmt"
-)
-
-type ErrorCode int
-
-const (
-	ErrTimeout  ErrorCode = -1
-	ErrResponse ErrorCode = -2
-)
-
-type BtError interface {
-	error
-}
-
-type TrackerError struct {
-	Code ErrorCode
-	Err  error
-}
-
-func (te *TrackerError) Error() string {
-	return fmt.Sprintf("%s:%d", te.Err.Error(), te.Code)
-}
-
-func NewTrackerError(code ErrorCode, err error) *TrackerError {
-	return &TrackerError{code, err}
-}

protocol/bt/tracker/tracker.go

@@ -1,412 +0,0 @@
-package tracker
-
-import (
-	"encoding/binary"
-	"errors"
-	"fmt"
-	"io"
-	"io/ioutil"
-	"math"
-	"math/rand"
-	"net"
-	"net/http"
-	"net/url"
-	"strconv"
-	"sync"
-	"time"
-
-	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/peer"
-)
-
-const (
-	udpConnectRequestMagic = 0x41727101980
-	udpConnectTimeout      = 15
-	udpConnectRetries      = 3
-	udpActionConnect       = 0
-	udpActionAnnounce      = 1
-	udpActionScrape        = 2
-	udpActionError         = 3
-)
-
-/*
-	Offset  size            Name            Value
-	0       64-bit integer  protocol_id     0x41727101980 // magic constant
-	8       32-bit integer  action          0 // connect
-	12      32-bit integer  transaction_id
-	16
-	Per https://www.libtorrent.org/udp_tracker_protocol.html#connecting
-*/
-type udpConnectRequest struct {
-	protocolId    uint64
-	action        uint32
-	transactionId uint32
-}
-
-func newUdpConnectRequest() *udpConnectRequest {
-	return &udpConnectRequest{
-		protocolId:    udpConnectRequestMagic,
-		action:        udpActionConnect,
-		transactionId: rand.Uint32(),
-	}
-}
-
-func (req *udpConnectRequest) encode() []byte {
-	buf := make([]byte, 16)
-	binary.BigEndian.PutUint64(buf[:8], req.protocolId)
-	binary.BigEndian.PutUint32(buf[8:12], req.action)
-	binary.BigEndian.PutUint32(buf[12:], req.transactionId)
-	return buf
-}
-
-/*
-	Offset  size            Name            Value
-	0       32-bit integer  action          0 // connect
-	4       32-bit integer  transaction_id
-	8       64-bit integer  connection_id
-	16
-*/
-type udpConnectResponse struct {
-	action        uint32
-	transactionId uint32
-	connectionId  uint64
-}
-
-func newUdpConnectResponse(buf []byte) *udpConnectResponse {
-	return &udpConnectResponse{
-		action:        binary.BigEndian.Uint32(buf[:4]),
-		transactionId: binary.BigEndian.Uint32(buf[4:8]),
-		connectionId:  binary.BigEndian.Uint64(buf[8:]),
-	}
-}
-
-/*
-	Offset  size    Name    Value
-	0       64-bit integer  connection_id
-	8       32-bit integer  action          1 // announce
-	12      32-bit integer  transaction_id
-	16      20-byte string  info_hash
-	36      20-byte string  peer_id
-	56      64-bit integer  downloaded
-	64      64-bit integer  left
-	72      64-bit integer  uploaded
-	80      32-bit integer  event           0 // 0: none; 1: completed; 2: started; 3: stopped
-	84      32-bit integer  ip address      0 // default
-	88      32-bit integer  key
-	92      32-bit integer  num_want        -1 // default
-	96      16-bit integer  port
-	98
-	Per https://www.libtorrent.org/udp_tracker_protocol.html#announcing
-*/
-type udpAnnounceRequest struct {
-	connectionId  uint64
-	action        uint32
-	transactionId uint32
-	infoHash      [20]byte
-	peerID        [20]byte
-	downloaded    uint64
-	left          uint64
-	uploaded      uint64
-	event         uint32
-	ip            uint32
-	key           uint32
-	numWant       uint32
-	port          uint16
-}
-
-func newUdpAnnounceRequest(connectionId uint64) *udpAnnounceRequest {
-	return &udpAnnounceRequest{
-		connectionId:  connectionId,
-		action:        udpActionAnnounce,
-		transactionId: rand.Uint32(),
-	}
-}
-
-func (req *udpAnnounceRequest) encode() []byte {
-	buf := make([]byte, 98, 98)
-	binary.BigEndian.PutUint64(buf[:8], req.connectionId)
-	binary.BigEndian.PutUint32(buf[8:12], req.action)
-	binary.BigEndian.PutUint32(buf[12:16], req.transactionId)
-	copy(buf[16:36], req.infoHash[:])
-	copy(buf[36:56], req.peerID[:])
-	binary.BigEndian.PutUint64(buf[56:64], req.downloaded)
-	binary.BigEndian.PutUint64(buf[64:72], req.left)
-	binary.BigEndian.PutUint64(buf[72:80], req.uploaded)
-	binary.BigEndian.PutUint32(buf[80:84], req.event)
-	binary.BigEndian.PutUint32(buf[84:88], req.ip)
-	binary.BigEndian.PutUint32(buf[88:92], req.key)
-	binary.BigEndian.PutUint32(buf[92:96], req.numWant)
-	binary.BigEndian.PutUint16(buf[96:98], req.port)
-	return buf
-}
-
-/*
-	Offset      size            Name            Value
-	0           32-bit integer  action          1 // announce
-	4           32-bit integer  transaction_id
-	8           32-bit integer  interval
-	12          32-bit integer  leechers
-	16          32-bit integer  seeders
-	20 + 6 * n  32-bit integer  ip address
-	24 + 6 * n  16-bit integer  TCP port
-	20 + 6 * N
-	Per https://www.libtorrent.org/udp_tracker_protocol.html#announcing
-*/
-type udpAnnounceResponse struct {
-	action        uint32
-	transactionId uint32
-	interval      uint32
-	leechers      uint32
-	seeders       uint32
-	peers         []*peer.Peer
-}
-
-func newUdpAnnounceResponse(buf []byte) *udpAnnounceResponse {
-	response := &udpAnnounceResponse{
-		action:        binary.BigEndian.Uint32(buf[:4]),
-		transactionId: binary.BigEndian.Uint32(buf[4:8]),
-		interval:      binary.BigEndian.Uint32(buf[8:12]),
-		leechers:      binary.BigEndian.Uint32(buf[12:16]),
-		seeders:       binary.BigEndian.Uint32(buf[16:20]),
-	}
-	count := (len(buf) - 20) / 6
-	response.peers = make([]*peer.Peer, count, count)
-	for i := 0; i < count; i++ {
-		ipBegin := 20 + 6*i
-		portBegin := ipBegin + 4
-		response.peers[i] = &peer.Peer{
-			IP:   binary.BigEndian.Uint32(buf[ipBegin:portBegin]),
-			Port: binary.BigEndian.Uint16(buf[portBegin : portBegin+2]),
-		}
-	}
-	return response
-}
-
-type Tracker struct {
-	PeerID   [20]byte
-	MetaInfo *metainfo.MetaInfo
-}
-
-func (tracker *Tracker) connect(conn net.Conn, timeout int64) (response *udpConnectResponse, err error) {
-	request := newUdpConnectRequest()
-	buf := request.encode()
-	conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeout)))
-	conn.Write(buf)
-	len, err := conn.Read(buf)
-	if err != nil {
-		return
-	}
-	if len != 16 {
-		err = NewTrackerError(ErrResponse, errors.New("invalid response"))
-		return
-	}
-	response = newUdpConnectResponse(buf)
-	return
-}
-
-func (tracker *Tracker) announce(conn net.Conn, timeout int64, connectionId uint64) (response *udpAnnounceResponse, err error) {
-	request := newUdpAnnounceRequest(connectionId)
-	request.infoHash = tracker.MetaInfo.GetInfoHash()
-	request.peerID = tracker.PeerID
-	request.downloaded = 0
-	request.uploaded = 0
-	request.left = tracker.MetaInfo.GetTotalSize() - request.downloaded
-	request.event = 0
-	request.ip = 0
-	request.key = 0
-	request.numWant = 200
-	request.port = 6882
-	encode := request.encode()
-	conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeout)))
-	conn.Write(encode)
-	buf := make([]byte, 20+request.numWant*6)
-	len, err := conn.Read(buf)
-	if err != nil {
-		return
-	}
-	if len < 20 {
-		err = errors.New("announce fail")
-		return
-	}
-	response = newUdpAnnounceResponse(buf[:len])
-	return
-}
-
-// 通过tracker服务器获取peer信息
-func (tracker *Tracker) Tracker() <-chan []*peer.Peer {
-	peersCh := make(chan []*peer.Peer)
-	metaInfo := tracker.MetaInfo
-
-	checkAnnounceList := func(announceList [][]string) bool {
-		if len(announceList) > 0 {
-			for _, l := range announceList {
-				if len(l) > 0 {
-					return true
-				}
-			}
-		}
-		return false
-	}
-
-	// 只要获取到了peers就直接返回
-	if checkAnnounceList(metaInfo.AnnounceList) {
-		wg := sync.WaitGroup{}
-		for _, announceArr := range metaInfo.AnnounceList {
-			if len(announceArr) > 0 {
-				for _, announce := range announceArr {
-					wg.Add(1)
-					go func(announce string) {
-						peers, err := tracker.DoTracker(announce)
-						if err == nil {
-							peersCh <- peers
-						}
-						wg.Done()
-					}(announce)
-				}
-			}
-		}
-		go func() {
-			wg.Wait()
-			close(peersCh)
-		}()
-	} else {
-		go func() {
-			peers, err := tracker.DoTracker(metaInfo.Announce)
-			if err == nil {
-				peersCh <- peers
-			}
-			close(peersCh)
-		}()
-	}
-	return peersCh
-}
-
-func (tracker *Tracker) DoTracker(announce string) (peers []*peer.Peer, err error) {
-	if announce != "" {
-		url, _ := url.Parse(announce)
-		switch url.Scheme {
-		case "http", "https":
-			return tracker.httpTracker(url)
-		case "udp", "udp4", "udp6":
-			return tracker.udpTracker(url)
-		default:
-			return nil, errors.New("unsupported protocol")
-		}
-	}
-	return nil, errors.New("empty announce")
-}
-
-// http://bittorrent.org/beps/bep_0003.html#trackers
-func (tracker *Tracker) httpTracker(url *url.URL) (peers []*peer.Peer, err error) {
-	metaInfo := tracker.MetaInfo
-	peerID := tracker.PeerID
-
-	query := url.Query()
-	infoHash := metaInfo.GetInfoHash()
-	query.Add("info_hash", string(infoHash[:]))
-	query.Add("peer_id", string(peerID[:]))
-	query.Add("port", "6882")
-	query.Add("uploaded", "0")
-	query.Add("downloaded", "0")
-	query.Add("left", strconv.FormatInt(int64(metaInfo.GetTotalSize()), 10))
-	url.RawQuery = query.Encode()
-	request, err := http.NewRequest("GET", url.String(), nil)
-	if err != nil {
-		return
-	}
-	var httpClient = &http.Client{
-		Timeout: time.Second * time.Duration(15),
-	}
-	request.Header.Set("User-Agent", "gopeed/0.0.1")
-	response, err := httpClient.Do(request)
-	if err != nil {
-		return
-	}
-	defer response.Body.Close()
-	resp, err := ioutil.ReadAll(response.Body)
-	if err != nil {
-		if err != io.EOF {
-			return
-		}
-	}
-	// TODO Parse HTTP tracker response
-	fmt.Println(resp)
-	return nil, nil
-}
-
-// http://bittorrent.org/beps/bep_0015.html
-func (tracker *Tracker) udpTracker(url *url.URL) (peers []*peer.Peer, err error) {
-	conn, err := dial(url)
-	if err != nil {
-		return
-	}
-	defer conn.Close()
-
-	announceResponse, err := func() (announceResponse *udpAnnounceResponse, err error) {
-		var connectResponse *udpConnectResponse
-		// 0:connect 1:announce
-		state := udpActionConnect
-		// 访问超时最多重试8次，当有请求成功则将重试次数重置为0
-		for n := 0; n < udpConnectRetries; n++ {
-			// 15 * 2 ^ n seconds
-			timeout := int64(udpConnectTimeout * math.Pow(2, float64(n)))
-			switch state {
-			case udpActionConnect:
-				connectResponse, err = tracker.connect(conn, timeout)
-				if err != nil {
-					if oe, ok := err.(*net.OpError); ok && oe.Timeout() {
-						break
-					} else {
-						return
-					}
-				}
-				n = -1
-				state = udpActionAnnounce
-				break
-			case udpActionAnnounce:
-				announceResponse, err = tracker.announce(conn, timeout, connectResponse.connectionId)
-				if err != nil {
-					if oe, ok := err.(*net.OpError); ok && oe.Timeout() {
-						break
-					} else {
-						return
-					}
-				}
-				return
-			}
-		}
-		return
-	}()
-
-	if err != nil {
-		return
-	}
-
-	return announceResponse.peers, nil
-}
-
-func dial(trackerUrl *url.URL) (net.Conn, error) {
-	return net.Dial(trackerUrl.Scheme, trackerUrl.Host)
-}
-
-func lookupIP(host string, scheme string) (ip net.IP, err error) {
-	ips, _ := net.LookupIP(host)
-	if len(ips) == 0 {
-		err = errors.New("no ips")
-		return
-	}
-	for _, ip = range ips {
-		switch scheme {
-		case "udp", "udp4":
-			if ip.To4() != nil {
-				return
-			}
-		case "udp6":
-			if ip.To4() == nil {
-				return
-			}
-		}
-	}
-	err = errors.New("no available ips ")
-	return
-}

protocol/bt/tracker/tracker_test.go

@@ -1,22 +0,0 @@
-package tracker
-
-import (
-	"fmt"
-	"github.com/monkeyWie/gopeed-core/protocol/bt/metainfo"
-	"testing"
-)
-
-func TestTracker_Tracker(t *testing.T) {
-	metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent")
-	if err != nil {
-		panic(err)
-	}
-	tracker := &Tracker{
-		PeerID:   [20]byte{},
-		MetaInfo: metaInfo,
-	}
-	tracker.MetaInfo.Announce = "udp://tracker.opentrackr.org:1337/announce"
-	tracker.MetaInfo.AnnounceList = [][]string{}
-	peers := <-tracker.Tracker()
-	fmt.Println(len(peers))
-}

protocol/http/client.go

@@ -1,177 +0,0 @@
-package http
-
-import (
-	"fmt"
-	"io"
-	"mime"
-	"net/http"
-	"net/http/cookiejar"
-	"net/url"
-	"os"
-	"strconv"
-	"strings"
-	"sync"
-)
-
-// Resolve return the file response to be downloaded
-func Resolve(request *Request) (*Response, error) {
-	httpRequest, err := BuildHTTPRequest(request)
-	if err != nil {
-		return nil, err
-	}
-	// Use "Range" header to resolve
-	httpRequest.Header.Add("Range", "bytes=0-0")
-	httpClient := BuildHTTPClient()
-	response, err := httpClient.Do(httpRequest)
-	if err != nil {
-		return nil, err
-	}
-	defer response.Body.Close()
-	if response.StatusCode != 200 && response.StatusCode != 206 {
-		return nil, fmt.Errorf("response status error:%d", response.StatusCode)
-	}
-	ret := &Response{}
-	// Get file name by "Content-Disposition"
-	contentDisposition := response.Header.Get("Content-Disposition")
-	if contentDisposition != "" {
-		_, params, _ := mime.ParseMediaType(contentDisposition)
-		filename := params["filename"]
-		if filename != "" {
-			ret.Name = filename
-		}
-	}
-	// Get file name by URL
-	if ret.Name == "" {
-		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 == "" {
-		ret.Name = "unknown"
-	}
-	// Is support range
-	ret.Range = response.StatusCode == 206
-	// Get file size
-	if ret.Range {
-		contentRange := response.Header.Get("Content-Range")
-		if contentRange != "" {
-			// e.g. bytes 0-1000/1001 => 1001
-			total := subLastSlash(contentRange)
-			if total != "" && total != "*" {
-				parse, err := strconv.ParseInt(total, 10, 64)
-				if err != nil {
-					return nil, err
-				}
-				ret.Size = parse
-			}
-		}
-	} else {
-		contentLength := response.Header.Get("Content-Length")
-		if contentLength != "" {
-			ret.Size, _ = strconv.ParseInt(contentLength, 10, 64)
-		}
-	}
-	return ret, nil
-}
-
-// Down
-func Down(request *Request) error {
-	response, err := Resolve(request)
-	if err != nil {
-		return err
-	}
-	// allocate file
-	file, err := os.Create(response.Name)
-	if err != nil {
-		return err
-	}
-	defer file.Close()
-	if err := file.Truncate(response.Size); err != nil {
-		return err
-	}
-	// support range
-	if response.Range {
-		cons := 16
-		chunkSize := response.Size / int64(cons)
-		waitGroup := &sync.WaitGroup{}
-		waitGroup.Add(cons)
-		for i := 0; i < cons; i++ {
-			start := int64(i) * chunkSize
-			end := start + chunkSize
-			if i == cons-1 {
-				end = response.Size
-			}
-			go downChunk(request, file, start, end-1, waitGroup)
-		}
-		waitGroup.Wait()
-	} else {
-		downChunk(request, file, 0, response.Size, nil)
-	}
-	return nil
-}
-
-func subLastSlash(str string) string {
-	index := strings.LastIndex(str, "/")
-	if index != -1 {
-		return str[index+1:]
-	}
-	return ""
-}
-
-func BuildHTTPRequest(request *Request) (*http.Request, error) {
-	// Build request
-	httpRequest, err := http.NewRequest(strings.ToUpper(request.Method), request.URL, nil)
-	if err != nil {
-		return nil, err
-	}
-	for k, v := range request.Header {
-		httpRequest.Header.Add(k, v)
-	}
-	return httpRequest, nil
-}
-
-func BuildHTTPClient() *http.Client {
-	// Cookie handle
-	jar, _ := cookiejar.New(nil)
-
-	return &http.Client{Jar: jar}
-}
-
-func downChunk(request *Request, file *os.File, start int64, end int64, waitGroup *sync.WaitGroup) {
-	if waitGroup != nil {
-		defer waitGroup.Done()
-	}
-	httpRequest, _ := BuildHTTPRequest(request)
-	httpRequest.Header.Add("Range", fmt.Sprintf("bytes=%d-%d", start, end))
-	fmt.Printf("down %d-%d\n", start, end)
-	httpClient := BuildHTTPClient()
-	httpResponse, err := httpClient.Do(httpRequest)
-	if err != nil {
-		fmt.Println(err)
-		return
-	}
-	defer httpResponse.Body.Close()
-	buf := make([]byte, 8192)
-	writeIndex := start
-	for {
-		n, err := httpResponse.Body.Read(buf)
-		if n > 0 {
-			writeSize, err := file.WriteAt(buf[0:n], writeIndex)
-			if err != nil {
-				fmt.Println(err)
-				return
-			}
-			writeIndex += int64(writeSize)
-		}
-		if err != nil {
-			if err != io.EOF {
-				fmt.Println(err)
-				return
-			}
-			break
-		}
-	}
-}

protocol/http/client_test.go

@@ -1,154 +0,0 @@
-package http
-
-import (
-	"crypto/md5"
-	"encoding/hex"
-	"fmt"
-	"io"
-	"log"
-	"os"
-	"reflect"
-	"testing"
-	"time"
-)
-
-func TestDown(t *testing.T) {
-	type args struct {
-		request *Request
-	}
-	tests := []struct {
-		name string
-		args args
-	}{
-		{
-			"proxyee-down.ico",
-			args{
-				&Request{
-					"get",
-					"https://raw.githubusercontent.com/proxyee-down-org/proxyee-down/master/front/public/favicon.ico",
-					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",
-					},
-					nil,
-				},
-			},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			os.Remove("favicon.ico")
-			err := Down(tt.args.request)
-			if err != nil {
-				t.Errorf("error down= %v", err)
-				return
-			}
-			downMd5 := fileMd5("favicon.ico")
-			os.Remove("favicon.ico")
-			if "8de7a6a2e786861013d61b77b2394012" != downMd5 {
-				t.Errorf("error md5= %v", downMd5)
-				return
-			}
-		})
-	}
-}
-
-func TestResolve(t *testing.T) {
-	// os.Setenv("HTTP_PROXY", "http://127.0.0.1:8888")
-	type args struct {
-		request *Request
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    *Response
-		wantErr bool
-	}{
-		{
-			"proxyee-down.ico",
-			args{
-				&Request{
-					"get",
-					"https://raw.githubusercontent.com/proxyee-down-org/proxyee-down/master/front/public/favicon.ico",
-					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",
-					},
-					nil,
-				},
-			},
-			&Response{"favicon.ico", 116093, true},
-			false,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := Resolve(tt.args.request)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("Resolve() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("Resolve() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestTemp(t *testing.T) {
-	f, err := os.Create("e:/testbt/test.data")
-	if err != nil {
-		log.Fatal(err)
-	}
-	if err := f.Truncate(1024 * 1024 * 512); err != nil {
-		log.Fatal(err)
-	}
-	bts := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9}
-	s := time.Now().UnixNano()
-	f.Write(bts)
-	fmt.Printf("write from 0 use: %d\n", time.Now().UnixNano()-s)
-	s = time.Now().UnixNano()
-	f.Seek(1024*1024*128, 0)
-	f.Write(bts)
-	fmt.Printf("write from 128M use %d\n", time.Now().UnixNano()-s)
-	s = time.Now().UnixNano()
-	f.Write(bts)
-	fmt.Printf("write from 128M+9b use %d\n", time.Now().UnixNano()-s)
-	s = time.Now().UnixNano()
-	f.Seek(0, 0)
-	f.Write(bts)
-	fmt.Printf("write from 0 agein use %d\n", time.Now().UnixNano()-s)
-	s = time.Now().UnixNano()
-	f.Seek(1024*1024*64, 0)
-	f.Write(bts)
-	fmt.Printf("write from agein 64M use %d\n", time.Now().UnixNano()-s)
-}
-
-func fileMd5(filePath string) string {
-	file, _ := os.Open(filePath)
-
-	// 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 TestTemp2(t *testing.T) {
-	fmt.Println("123456789"[2:3])
-}

protocol/http/request.go

@@ -1,8 +0,0 @@
-package http
-
-type Request struct {
-	Method  string
-	URL     string
-	Header  map[string]string
-	Content []byte
-}

protocol/http/response.go

@@ -1,7 +0,0 @@
-package http
-
-type Response struct {
-	Name  string
-	Size  int64
-	Range bool
-}