From c6aa83194f5164d990b93e8b070f6219f1d8abf7 Mon Sep 17 00:00:00 2001 From: monkeyWie Date: Fri, 9 Sep 2022 18:31:57 +0800 Subject: [PATCH] feat: support RESTful api (#33) --- .github/workflows/test.yml | 4 +- .gitignore | 2 +- _examples/basic/main.go | 2 +- cmd/gopeed/gopeed.go | 6 +- go.mod | 1 + go.sum | 2 + internal/protocol/bt/fetcher.go | 14 +- internal/protocol/bt/fetcher_test.go | 8 +- internal/protocol/http/fetcher.go | 23 ++- internal/protocol/http/fetcher_test.go | 51 ++++-- pkg/base/constants.go | 19 +-- pkg/base/model.go | 30 ++-- pkg/download/downloader.go | 111 ++++++++----- pkg/download/downloader_test.go | 18 +-- pkg/download/task.go | 19 ++- pkg/rest/api.go | 79 +++++++++ pkg/rest/model/result.go | 21 +++ pkg/rest/model/task.go | 8 + pkg/rest/server.go | 49 ++++++ pkg/rest/server_test.go | 212 +++++++++++++++++++++++++ pkg/rest/util/util.go | 25 +++ pkg/test/httptest.go | 24 +++ 22 files changed, 603 insertions(+), 125 deletions(-) create mode 100644 pkg/rest/api.go create mode 100644 pkg/rest/model/result.go create mode 100644 pkg/rest/model/task.go create mode 100644 pkg/rest/server.go create mode 100644 pkg/rest/server_test.go create mode 100644 pkg/rest/util/util.go diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 94418df2..27e3d657 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -38,9 +38,9 @@ jobs: - uses: actions/checkout@v2 - name: Run Unit tests. run: go test ./... -v -covermode=count -coverprofile=coverage.txt - - uses: codecov/codecov-action@v1 + - uses: codecov/codecov-action@v3 with: - file: ./coverage.txt + files: ./coverage.txt fail_ci_if_error: true build: name: Build diff --git a/.gitignore b/.gitignore index 7f720e5c..2a726371 100644 --- a/.gitignore +++ b/.gitignore @@ -21,4 +21,4 @@ bin/ .torrent.db-shm .torrent.db-wal -.data \ No newline at end of file +*.data \ No newline at end of file diff --git a/_examples/basic/main.go b/_examples/basic/main.go index 02b60d22..7e8fb147 100644 --- a/_examples/basic/main.go +++ b/_examples/basic/main.go @@ -8,7 +8,7 @@ import ( func main() { finallyCh := make(chan error) - err := download.Boot(). + _, err := download.Boot(). URL("https://www.baidu.com/index.html"). Listener(func(event *download.Event) { if event.Key == download.EventKeyFinally { diff --git a/cmd/gopeed/gopeed.go b/cmd/gopeed/gopeed.go index 58da59d6..8ed55895 100644 --- a/cmd/gopeed/gopeed.go +++ b/cmd/gopeed/gopeed.go @@ -15,7 +15,7 @@ func main() { args := parse() var wg sync.WaitGroup - err := download.Boot(). + _, err := download.Boot(). URL(args.url). Listener(func(event *download.Event) { if event.Key == download.EventKeyProgress { @@ -55,10 +55,10 @@ var ( ) func printProgress(task *download.Task, title string) { - rate := float64(task.Progress.Downloaded) / float64(task.Res.Length) + rate := float64(task.Progress.Downloaded) / float64(task.Res.Size) completeWidth := int(progressWidth * rate) speed := util.ByteFmt(task.Progress.Speed) - totalSize := util.ByteFmt(task.Res.Length) + totalSize := util.ByteFmt(task.Res.Size) sb.WriteString(fmt.Sprintf("\r%s [", title)) for i := 0; i < progressWidth; i++ { if i < completeWidth { diff --git a/go.mod b/go.mod index 2e9d54bf..d0973932 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,7 @@ go 1.18 require ( github.com/anacrolix/torrent v1.46.0 github.com/google/uuid v1.3.0 + github.com/gorilla/mux v1.8.0 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd golang.org/x/sync v0.0.0-20210220032951-036812b2e83c ) diff --git a/go.sum b/go.sum index 7663bdcb..40af7d12 100644 --- a/go.sum +++ b/go.sum @@ -159,6 +159,8 @@ github.com/gopherjs/gopherjs v0.0.0-20190309154008-847fc94819f9/go.mod h1:wJfORR github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= github.com/gorilla/context v1.1.1/go.mod h1:kBGZzfjB9CEq2AlWe17Uuf7NDRt0dE0s8S51q0aT7Yg= github.com/gorilla/mux v1.6.2/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs= +github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI= +github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So= github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc= github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= diff --git a/internal/protocol/bt/fetcher.go b/internal/protocol/bt/fetcher.go index b26f45da..bfd80610 100644 --- a/internal/protocol/bt/fetcher.go +++ b/internal/protocol/bt/fetcher.go @@ -45,10 +45,10 @@ func (f *Fetcher) Resolve(req *base.Request) (res *base.Resource, err error) { } <-f.torrent.GotInfo() res = &base.Resource{ - Req: req, - Length: f.torrent.Length(), - Range: true, - Files: make([]*base.FileInfo, len(f.torrent.Files())), + Req: req, + Size: f.torrent.Length(), + Range: true, + Files: make([]*base.FileInfo, len(f.torrent.Files())), } for i, file := range f.torrent.Files() { res.Files[i] = &base.FileInfo{ @@ -62,12 +62,6 @@ func (f *Fetcher) Resolve(req *base.Request) (res *base.Resource, err error) { func (f *Fetcher) Create(res *base.Resource, opts *base.Options) (err error) { f.opts = opts - if len(f.opts.SelectFiles) == 0 { - f.opts.SelectFiles = make([]int, len(res.Files)) - for i := range f.opts.SelectFiles { - f.opts.SelectFiles[i] = i - } - } return nil } diff --git a/internal/protocol/bt/fetcher_test.go b/internal/protocol/bt/fetcher_test.go index 290f203e..1528daa8 100644 --- a/internal/protocol/bt/fetcher_test.go +++ b/internal/protocol/bt/fetcher_test.go @@ -18,9 +18,9 @@ func TestFetcher_Resolve_Torrent(t *testing.T) { } want := &base.Resource{ - Req: res.Req, - Length: 1466714112, - Range: true, + Req: res.Req, + Size: 1466714112, + Range: true, Files: []*base.FileInfo{ { Name: "ubuntu-22.04-live-server-amd64.iso", @@ -30,7 +30,7 @@ func TestFetcher_Resolve_Torrent(t *testing.T) { }, } if !reflect.DeepEqual(want, res) { - t.Errorf("Resolve error = %v, want %v", res, want) + t.Errorf("Resolve() got = %v, want %v", res, want) } } diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index bebf73b7..3c8b0b23 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -8,7 +8,6 @@ import ( "github.com/monkeyWie/gopeed-core/pkg/base" "golang.org/x/sync/errgroup" "io" - "io/ioutil" "mime" "net/http" "net/http/cookiejar" @@ -75,7 +74,7 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { if err != nil { return nil, err } - res.Length = parse + res.Size = parse } } else if base.HttpCodeOK == httpResp.StatusCode { // 返回200响应码，不支持断点下载，通过Content-Length头获取文件大小，获取不到的话可能是chunked编码 @@ -85,13 +84,13 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { if err != nil { return nil, err } - res.Length = parse + res.Size = parse } } else { return nil, NewRequestError(httpResp.StatusCode, httpResp.Status) } file := &base.FileInfo{ - Size: res.Length, + Size: res.Size, } contentDisposition := httpResp.Header.Get(base.HttpHeaderContentDisposition) if contentDisposition != "" { @@ -123,14 +122,14 @@ func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error { func (f *Fetcher) Start() (err error) { // 创建文件 name := f.filename() - _, err = f.Ctl.Touch(name, f.res.Length) + _, err = f.Ctl.Touch(name, f.res.Size) if err != nil { return err } - f.status = base.DownloadStatusStart + f.status = base.DownloadStatusRunning if f.res.Range { // 每个连接平均需要下载的分块大小 - chunkSize := f.res.Length / int64(f.opts.Connections) + chunkSize := f.res.Size / int64(f.opts.Connections) f.chunks = make([]*Chunk, f.opts.Connections) for i := 0; i < f.opts.Connections; i++ { var ( @@ -139,7 +138,7 @@ func (f *Fetcher) Start() (err error) { ) if i == f.opts.Connections-1 { // 最后一个分块需要保证把文件下载完 - end = f.res.Length - 1 + end = f.res.Size - 1 } else { end = begin + chunkSize - 1 } @@ -156,7 +155,7 @@ func (f *Fetcher) Start() (err error) { } func (f *Fetcher) Pause() (err error) { - if base.DownloadStatusStart != f.status { + if base.DownloadStatusRunning != f.status { return } f.status = base.DownloadStatusPause @@ -166,10 +165,10 @@ func (f *Fetcher) Pause() (err error) { } func (f *Fetcher) Continue() (err error) { - if base.DownloadStatusStart == f.status || base.DownloadStatusDone == f.status { + if base.DownloadStatusRunning == f.status || base.DownloadStatusDone == f.status { return } - f.status = base.DownloadStatusStart + f.status = base.DownloadStatusRunning var name = f.filename() _, err = f.Ctl.Open(name) if err != nil { @@ -352,7 +351,7 @@ func buildRequest(ctx context.Context, req *base.Request) (httpReq *http.Request } } if extra.Body != "" { - body = ioutil.NopCloser(bytes.NewBufferString(extra.Body)) + body = bytes.NewBufferString(extra.Body) } } diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index 2694c2ec..9d359c67 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -1,6 +1,7 @@ package http import ( + "fmt" "github.com/monkeyWie/gopeed-core/internal/controller" "github.com/monkeyWie/gopeed-core/internal/fetcher" "github.com/monkeyWie/gopeed-core/pkg/base" @@ -13,8 +14,8 @@ import ( func TestFetcher_Resolve(t *testing.T) { testResolve(test.StartTestFileServer, &base.Resource{ - Length: test.BuildSize, - Range: true, + Size: test.BuildSize, + Range: true, Files: []*base.FileInfo{ { Name: test.BuildName, @@ -23,8 +24,8 @@ func TestFetcher_Resolve(t *testing.T) { }, }, t) testResolve(test.StartTestChunkedServer, &base.Resource{ - Length: 0, - Range: false, + Size: 0, + Range: false, Files: []*base.FileInfo{ { Name: test.BuildName, @@ -46,7 +47,7 @@ func testResolve(startTestServer func() net.Listener, want *base.Resource, t *te } res.Req = nil if !reflect.DeepEqual(want, res) { - t.Errorf("Resolve error = %v, want %v", res, want) + t.Errorf("Resolve() got = %v, want %v", res, want) } } @@ -75,7 +76,13 @@ func TestFetcher_DownloadChunked(t *testing.T) { defer listener.Close() // chunked编码下载 downloadNormal(listener, 1, t) - //downloadContinue(listener, 1, t) +} + +func TestFetcher_DownloadPost(t *testing.T) { + listener := test.StartTestPostServer() + defer listener.Close() + // post下载 + downloadPost(listener, 1, t) } func TestFetcher_DownloadRetry(t *testing.T) { @@ -135,7 +142,31 @@ func downloadNormal(listener net.Listener, connections int, t *testing.T) { want := test.FileMd5(test.BuildFile) got := test.FileMd5(test.DownloadFile) if want != got { - t.Errorf("Download error = %v, want %v", got, want) + t.Errorf("Download() got = %v, want %v", got, want) + } +} + +func downloadPost(listener net.Listener, connections int, t *testing.T) { + fetcher, _, _ := downloadReady(listener, connections, t) + fetcher.(*Fetcher).res.Req.Extra = Extra{ + Method: "POST", + Header: map[string]string{ + "Authorization": "Bearer 123456", + }, + Body: fmt.Sprintf(`{"name":"%s"}`, test.BuildName), + } + 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() got = %v, want %v", got, want) } } @@ -160,7 +191,7 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { want := test.FileMd5(test.BuildFile) got := test.FileMd5(test.DownloadFile) if want != got { - t.Errorf("Download error = %v, want %v", got, want) + t.Errorf("Download() got = %v, want %v", got, want) } } @@ -172,7 +203,7 @@ func downloadError(listener net.Listener, connections int, t *testing.T) { } err = fetcher.Wait() if err == nil { - t.Errorf("Download error = %v, want %v", err, nil) + t.Errorf("Download() got = %v, want %v", err, nil) } } @@ -200,7 +231,7 @@ func downloadResume(listener net.Listener, connections int, t *testing.T) { want := test.FileMd5(test.BuildFile) got := test.FileMd5(test.DownloadFile) if want != got { - t.Errorf("Download error = %v, want %v", got, want) + t.Errorf("Download() got = %v, want %v", got, want) } } diff --git a/pkg/base/constants.go b/pkg/base/constants.go index d3c3d7ac..3ec1db32 100644 --- a/pkg/base/constants.go +++ b/pkg/base/constants.go @@ -1,16 +1,13 @@ package base -import "errors" - -type Status int +type Status string const ( - DownloadStatusPrepare Status = iota - DownloadStatusReady - DownloadStatusStart - DownloadStatusPause - DownloadStatusError - DownloadStatusDone + DownloadStatusReady Status = "ready" + DownloadStatusRunning Status = "running" + DownloadStatusPause Status = "pause" + DownloadStatusError Status = "error" + DownloadStatusDone Status = "done" ) const ( @@ -24,7 +21,3 @@ const ( HttpHeaderRangeFormat = "bytes=%d-%d" ) - -var ( - DeleteErr = errors.New("delete") -) diff --git a/pkg/base/model.go b/pkg/base/model.go index 4a0b5d50..05c94bcb 100644 --- a/pkg/base/model.go +++ b/pkg/base/model.go @@ -3,38 +3,36 @@ package base // 下载请求 type Request struct { // 下载链接 - Schema string - // 下载链接 - URL string + URL string `json:"url"` // 附加信息 - Extra interface{} + Extra interface{} `json:"extra"` } // 资源信息 type Resource struct { - Req *Request + Req *Request `json:"req"` // 资源总大小 - Length int64 + Size int64 `json:"size"` // 是否支持断点下载 - Range bool + Range bool `json:"range"` // 资源所包含的文件列表 - Files []*FileInfo + Files []*FileInfo `json:"files"` } type FileInfo struct { - Name string - Path string - Size int64 + Name string `json:"name"` + Path string `json:"path"` + Size int64 `json:"size"` } // 下载选项 type Options struct { // 保存文件名 - Name string + Name string `json:"name"` // 保存目录 - Path string + Path string `json:"path"` // 并发连接数 - Connections int - // 下载的文件下标列表 - SelectFiles []int + Connections int `json:"connections"` + // 选择下载的文件下标列表 + SelectFiles []int `json:"select_files"` } diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 3403d18c..8904f3f0 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -17,17 +17,17 @@ type Listener func(event *Event) type Progress struct { // 下载耗时(纳秒) - Used int64 + Used int64 `json:"used"` // 每秒下载字节数 - Speed int64 + Speed int64 `json:"speed"` // 已下载的字节数 - Downloaded int64 + Downloaded int64 `json:"downloaded"` } type Downloader struct { *controller.DefaultController fetchBuilders map[string]fetcher.FetcherBuilder - tasks map[string]*Task + tasks []*Task listener Listener } @@ -39,15 +39,14 @@ func NewDownloader(fbs ...fetcher.FetcherBuilder) *Downloader { d.fetchBuilders[strings.ToUpper(p)] = f } } - d.tasks = make(map[string]*Task) + d.tasks = make([]*Task, 0) // 每秒统计一次下载速度 go func() { for { if len(d.tasks) > 0 { for _, task := range d.tasks { - if task.Status == base.DownloadStatusPrepare || - task.Status == base.DownloadStatusDone || + if task.Status == base.DownloadStatusDone || task.Status == base.DownloadStatusError || task.Status == base.DownloadStatusPause { continue @@ -79,42 +78,55 @@ func (d *Downloader) buildFetcher(url string) (fetcher.Fetcher, error) { return nil, errors.New("unsupported protocol") } -func (d *Downloader) Resolve(req *base.Request) (string, *base.Resource, error) { +func (d *Downloader) Resolve(req *base.Request) (*base.Resource, error) { fetcher, err := d.buildFetcher(req.URL) if err != nil { - return "", nil, err + return nil, err } res, err := fetcher.Resolve(req) if err != nil { - return "", nil, err + return nil, err } - task := NewTask() - task.fetcher = fetcher - task.Res = res - d.tasks[task.ID] = task - return task.ID, res, nil + + return res, nil } -func (d *Downloader) Create(taskID string, res *base.Resource, opts *base.Options) (err error) { - task := d.tasks[taskID] - fetcher := d.tasks[taskID].fetcher +func (d *Downloader) Create(res *base.Resource, opts *base.Options) (taskId string, err error) { if opts == nil { opts = &base.Options{} } if !res.Range || opts.Connections < 1 { opts.Connections = 1 } + if len(opts.SelectFiles) == 0 { + opts.SelectFiles = make([]int, len(res.Files)) + for i := range opts.SelectFiles { + opts.SelectFiles[i] = i + } + } + + fetcher, err := d.buildFetcher(res.Req.URL) + if err != nil { + return + } err = fetcher.Create(res, opts) if err != nil { return } + + task := NewTask() + task.fetcher = fetcher + task.Res = res task.Opts = opts - task.Status = base.DownloadStatusReady task.Progress = &Progress{} task.timer = &util.Timer{} task.locker = new(sync.Mutex) task.timer.Start() + task.Status = base.DownloadStatusRunning + d.tasks = append(d.tasks, task) d.emit(EventKeyStart, task) + taskId = task.ID + err = fetcher.Start() if err != nil { return @@ -122,18 +134,20 @@ func (d *Downloader) Create(taskID string, res *base.Resource, opts *base.Option go func() { err = fetcher.Wait() if err != nil { + task.Status = base.DownloadStatusError d.emit(EventKeyError, task, err) } else { task.Progress.Used = task.timer.Used() - if task.Res.Length == 0 { - task.Res.Length = task.fetcher.Progress().TotalDownloaded() + if task.Res.Size == 0 { + task.Res.Size = task.fetcher.Progress().TotalDownloaded() } used := task.Progress.Used / int64(time.Second) if used == 0 { used = 1 } - task.Progress.Speed = task.Res.Length / used - task.Progress.Downloaded = task.Res.Length + task.Progress.Speed = task.Res.Size / used + task.Progress.Downloaded = task.Res.Size + task.Status = base.DownloadStatusDone d.emit(EventKeyDone, task) } d.emit(EventKeyFinally, task, err) @@ -142,21 +156,33 @@ func (d *Downloader) Create(taskID string, res *base.Resource, opts *base.Option } func (d *Downloader) Pause(id string) { - task := d.tasks[id] + task := d.GetTask(id) + if task == nil { + return + } task.locker.Lock() defer task.locker.Unlock() - task.timer.Pause() - task.fetcher.Pause() - d.emit(EventKeyPause, task) + if task.Status == base.DownloadStatusRunning { + task.Status = base.DownloadStatusPause + task.timer.Pause() + task.fetcher.Pause() + d.emit(EventKeyPause, task) + } } func (d *Downloader) Continue(id string) { - task := d.tasks[id] + task := d.GetTask(id) + if task == nil { + return + } task.locker.Lock() defer task.locker.Unlock() - task.timer.Continue() - task.fetcher.Continue() - d.emit(EventKeyContinue, task) + if task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusError { + task.Status = base.DownloadStatusRunning + task.timer.Continue() + task.fetcher.Continue() + d.emit(EventKeyContinue, task) + } } func (d *Downloader) Listener(fn Listener) { @@ -177,6 +203,19 @@ func (d *Downloader) emit(eventKey EventKey, task *Task, errs ...error) { } } +func (d *Downloader) GetTask(id string) *Task { + for _, task := range d.tasks { + if task.ID == id { + return task + } + } + return nil +} + +func (d *Downloader) GetTasks() []*Task { + return d.tasks +} + var defaultDownloader = NewDownloader(new(http.FetcherBuilder), new(bt.FetcherBuilder)) type boot struct { @@ -195,7 +234,7 @@ func (b *boot) Extra(extra interface{}) *boot { return b } -func (b *boot) Resolve() (string, *base.Resource, error) { +func (b *boot) Resolve() (*base.Resource, error) { return defaultDownloader.Resolve(&base.Request{ URL: b.url, Extra: b.extra, @@ -207,13 +246,13 @@ func (b *boot) Listener(listener Listener) *boot { return b } -func (b *boot) Create(opts *base.Options) error { - taskID, res, err := b.Resolve() +func (b *boot) Create(opts *base.Options) (string, error) { + res, err := b.Resolve() if err != nil { - return err + return "", err } defaultDownloader.Listener(b.listener) - return defaultDownloader.Create(taskID, res, opts) + return defaultDownloader.Create(res, opts) } func Boot() *boot { diff --git a/pkg/download/downloader_test.go b/pkg/download/downloader_test.go index c0831e2e..619e444b 100644 --- a/pkg/download/downloader_test.go +++ b/pkg/download/downloader_test.go @@ -17,14 +17,14 @@ func TestDownloader_Resolve(t *testing.T) { req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } - _, res, err := downloader.Resolve(req) + res, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } want := &base.Resource{ - Req: req, - Length: test.BuildSize, - Range: true, + Req: req, + Size: test.BuildSize, + Range: true, Files: []*base.FileInfo{ { Name: test.BuildName, @@ -33,8 +33,8 @@ func TestDownloader_Resolve(t *testing.T) { }, }, } - if !reflect.DeepEqual(res, want) { - t.Errorf("Resolve error = %s, want %s", test.ToJson(res), test.ToJson(want)) + if !reflect.DeepEqual(want, res) { + t.Errorf("Resolve() got = %v, want %v", res, want) } } @@ -46,7 +46,7 @@ func TestDownloader_Create(t *testing.T) { req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } - taskID, res, err := downloader.Resolve(req) + res, err := downloader.Resolve(req) if err != nil { t.Fatal(err) } @@ -58,7 +58,7 @@ func TestDownloader_Create(t *testing.T) { wg.Done() } }) - err = downloader.Create(taskID, res, &base.Options{ + _, err = downloader.Create(res, &base.Options{ Path: test.Dir, Name: test.DownloadName, Connections: 4, @@ -70,6 +70,6 @@ func TestDownloader_Create(t *testing.T) { want := test.FileMd5(test.BuildFile) got := test.FileMd5(test.DownloadFile) if want != got { - t.Errorf("Download error = %v, want %v", got, want) + t.Errorf("Download() got = %v, want %v", got, want) } } diff --git a/pkg/download/task.go b/pkg/download/task.go index 5170f95b..44082922 100644 --- a/pkg/download/task.go +++ b/pkg/download/task.go @@ -6,15 +6,17 @@ import ( "github.com/monkeyWie/gopeed-core/pkg/base" "github.com/monkeyWie/gopeed-core/pkg/util" "sync" + "time" ) type Task struct { - ID string - Res *base.Resource - Opts *base.Options - Status base.Status - Files map[string]*base.FileInfo - Progress *Progress + ID string `json:"id"` + Res *base.Resource `json:"res"` + Opts *base.Options `json:"opts"` + Status base.Status `json:"status"` + Files map[string]*base.FileInfo `json:"files"` + Progress *Progress `json:"progress"` + CreateTime time.Time `json:"create_time"` fetcher fetcher.Fetcher timer *util.Timer @@ -23,7 +25,8 @@ type Task struct { func NewTask() *Task { return &Task{ - ID: uuid.New().String(), - Status: base.DownloadStatusPrepare, + ID: uuid.New().String(), + Status: base.DownloadStatusReady, + CreateTime: time.Now(), } } diff --git a/pkg/rest/api.go b/pkg/rest/api.go new file mode 100644 index 00000000..c237d9d1 --- /dev/null +++ b/pkg/rest/api.go @@ -0,0 +1,79 @@ +package rest + +import ( + "github.com/gorilla/mux" + "github.com/monkeyWie/gopeed-core/pkg/base" + "github.com/monkeyWie/gopeed-core/pkg/download" + "github.com/monkeyWie/gopeed-core/pkg/rest/model" + "github.com/monkeyWie/gopeed-core/pkg/rest/util" + "net/http" + "strings" +) + +func Resolve(w http.ResponseWriter, r *http.Request) { + var req base.Request + if util.ReadJson(w, r, &req) { + resource, err := Downloader.Resolve(&req) + if err != nil { + util.WriteJson(w, 500, model.NewResultWithMsg(err.Error())) + return + } + util.WriteJsonOk(w, model.NewResultWithData(resource)) + } +} + +func CreateTask(w http.ResponseWriter, r *http.Request) { + var req model.CreateTaskReq + if util.ReadJson(w, r, &req) { + taskId, err := Downloader.Create(req.Resource, req.Options) + if err != nil { + util.WriteJson(w, 500, model.NewResultWithMsg(err.Error())) + return + } + util.WriteJsonOk(w, model.NewResultWithData(taskId)) + } +} + +func PauseTask(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + taskId := vars["id"] + Downloader.Pause(taskId) + util.WriteJsonOk(w, nil) +} + +func ContinueTask(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + taskId := vars["id"] + Downloader.Continue(taskId) + util.WriteJsonOk(w, nil) +} + +func GetTask(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + taskId := vars["id"] + task := Downloader.GetTask(taskId) + if task == nil { + util.WriteJson(w, 404, model.NewResultWithMsg("task not found")) + return + } + util.WriteJsonOk(w, model.NewResultWithData(task)) +} + +func GetTasks(w http.ResponseWriter, r *http.Request) { + status := r.FormValue("status") + if status == "" { + util.WriteJson(w, 400, model.NewResultWithMsg("param is required: status")) + return + } + statusArr := strings.Split(status, ",") + tasks := Downloader.GetTasks() + var result []*download.Task + for _, task := range tasks { + for _, s := range statusArr { + if task.Status == base.Status(s) { + result = append(result, task) + } + } + } + util.WriteJsonOk(w, model.NewResultWithData(tasks)) +} diff --git a/pkg/rest/model/result.go b/pkg/rest/model/result.go new file mode 100644 index 00000000..6807e383 --- /dev/null +++ b/pkg/rest/model/result.go @@ -0,0 +1,21 @@ +package model + +type Result[T any] struct { + Msg string `json:"msg"` + Data T `json:"data"` +} + +func NewResult[T any](msg string, data T) *Result[T] { + return &Result[T]{ + Msg: msg, + Data: data, + } +} + +func NewResultWithMsg(msg string) *Result[any] { + return NewResult[any](msg, nil) +} + +func NewResultWithData[T any](data T) *Result[T] { + return NewResult[T]("", data) +} diff --git a/pkg/rest/model/task.go b/pkg/rest/model/task.go new file mode 100644 index 00000000..271e0fdf --- /dev/null +++ b/pkg/rest/model/task.go @@ -0,0 +1,8 @@ +package model + +import "github.com/monkeyWie/gopeed-core/pkg/base" + +type CreateTaskReq struct { + Resource *base.Resource `json:"resource"` + Options *base.Options `json:"options"` +} diff --git a/pkg/rest/server.go b/pkg/rest/server.go new file mode 100644 index 00000000..21027e8f --- /dev/null +++ b/pkg/rest/server.go @@ -0,0 +1,49 @@ +package rest + +import ( + "context" + "fmt" + "github.com/gorilla/mux" + "github.com/monkeyWie/gopeed-core/internal/protocol/bt" + fhttp "github.com/monkeyWie/gopeed-core/internal/protocol/http" + "github.com/monkeyWie/gopeed-core/pkg/download" + "net" + "net/http" +) + +var ( + srv *http.Server + + Downloader = download.NewDownloader(new(fhttp.FetcherBuilder), new(bt.FetcherBuilder)) +) + +func Start(addr string, port int) (int, error) { + listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", addr, port)) + if err != nil { + return 0, err + } + + var r = mux.NewRouter() + r.Methods(http.MethodPost).Path("/api/v1/resolve").HandlerFunc(Resolve) + r.Methods(http.MethodPost).Path("/api/v1/tasks").HandlerFunc(CreateTask) + r.Methods(http.MethodPut).Path("/api/v1/tasks/{id}/pause").HandlerFunc(PauseTask) + r.Methods(http.MethodPut).Path("/api/v1/tasks/{id}/continue").HandlerFunc(ContinueTask) + r.Methods(http.MethodGet).Path("/api/v1/tasks/{id}").HandlerFunc(GetTask) + r.Methods(http.MethodGet).Path("/api/v1/tasks").HandlerFunc(GetTasks) + + srv = &http.Server{Handler: r} + go func() { + if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed { + // TODO log + panic(err) + } + }() + return listener.Addr().(*net.TCPAddr).Port, nil +} + +func Stop() error { + if srv != nil { + return srv.Shutdown(context.TODO()) + } + return nil +} diff --git a/pkg/rest/server_test.go b/pkg/rest/server_test.go new file mode 100644 index 00000000..83903e7e --- /dev/null +++ b/pkg/rest/server_test.go @@ -0,0 +1,212 @@ +package rest + +import ( + "bytes" + "encoding/json" + "fmt" + "github.com/monkeyWie/gopeed-core/pkg/base" + "github.com/monkeyWie/gopeed-core/pkg/download" + "github.com/monkeyWie/gopeed-core/pkg/rest/model" + "github.com/monkeyWie/gopeed-core/pkg/test" + "io" + "net" + "net/http" + "reflect" + "sync" + "testing" +) + +var ( + restPort int + + taskReq = &base.Request{} + taskRes = &base.Resource{ + Req: taskReq, + Size: test.BuildSize, + Range: true, + Files: []*base.FileInfo{ + { + Name: test.BuildName, + Path: "", + Size: test.BuildSize, + }, + }, + } + createReq = &model.CreateTaskReq{ + Resource: taskRes, + Options: &base.Options{ + Path: test.Dir, + Name: test.DownloadName, + Connections: 2, + }, + } +) + +func TestResolve(t *testing.T) { + doTest(func() { + resp := httpRequestCheckOk[*base.Resource](http.MethodPost, "/api/v1/resolve", taskReq) + if !reflect.DeepEqual(taskRes, resp) { + t.Errorf("Resolve() got = %v, want %v", resp, taskRes) + } + }) +} + +func TestCreateTask(t *testing.T) { + doTest(func() { + var wg sync.WaitGroup + wg.Add(1) + Downloader.Listener(func(event *download.Event) { + if event.Key == download.EventKeyFinally { + wg.Done() + } + }) + + taskId := httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + if taskId == "" { + t.Fatal("create task failed") + } + + wg.Wait() + want := test.FileMd5(test.BuildFile) + got := test.FileMd5(test.DownloadFile) + if want != got { + t.Errorf("CreateTask() got = %v, want %v", got, want) + } + }) +} + +func TestPauseAndContinueTask(t *testing.T) { + doTest(func() { + type result struct { + pauseCount int + continueCount int + md5 string + } + + var got = result{ + pauseCount: 0, + continueCount: 0, + md5: "", + } + var wg sync.WaitGroup + wg.Add(1) + Downloader.Listener(func(event *download.Event) { + switch event.Key { + case download.EventKeyPause: + got.pauseCount++ + case download.EventKeyContinue: + got.continueCount++ + case download.EventKeyFinally: + wg.Done() + } + }) + + taskId := httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + t1 := httpRequestCheckOk[*download.Task](http.MethodGet, "/api/v1/tasks/"+taskId, nil) + if t1.Status != base.DownloadStatusRunning { + t.Errorf("CreateTask() got = %v, want %v", t1.Status, base.DownloadStatusRunning) + } + httpRequestCheckOk[any](http.MethodPut, "/api/v1/tasks/"+taskId+"/pause", nil) + t2 := httpRequestCheckOk[*download.Task](http.MethodGet, "/api/v1/tasks/"+taskId, nil) + if t2.Status != base.DownloadStatusPause { + t.Errorf("PauseTask() got = %v, want %v", t2.Status, base.DownloadStatusPause) + } + httpRequestCheckOk[any](http.MethodPut, "/api/v1/tasks/"+taskId+"/continue", nil) + t3 := httpRequestCheckOk[*download.Task](http.MethodGet, "/api/v1/tasks/"+taskId, nil) + if t3.Status != base.DownloadStatusRunning { + t.Errorf("ContinueTask() got = %v, want %v", t3.Status, base.DownloadStatusRunning) + } + + wg.Wait() + want := result{ + pauseCount: 1, + continueCount: 1, + md5: test.FileMd5(test.BuildFile), + } + got.md5 = test.FileMd5(test.DownloadFile) + if !reflect.DeepEqual(got, want) { + t.Errorf("PauseAndContinueTask() got = %v, want %v", got, want) + } + }) +} + +func TestGetTasks(t *testing.T) { + doTest(func() { + var wg sync.WaitGroup + wg.Add(1) + Downloader.Listener(func(event *download.Event) { + if event.Key == download.EventKeyFinally { + wg.Done() + } + }) + + httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + code, _ := httpRequest[[]*download.Task](http.MethodGet, "/api/v1/tasks", nil) + checkCode(code, 400) + + wg.Wait() + r := httpRequestCheckOk[[]*download.Task](http.MethodGet, "/api/v1/tasks?status=done", nil) + if r[0].Status != base.DownloadStatusDone { + t.Errorf("GetTasks() got = %v, want %v", r[0].Status, base.DownloadStatusDone) + } + }) +} + +func doTest(handler func()) { + fileListener := doStart() + defer func() { + fileListener.Close() + Stop() + }() + taskReq.URL = "http://" + fileListener.Addr().String() + "/" + test.BuildName + handler() +} + +func doStart() net.Listener { + port, err := Start("127.0.0.1", 0) + if err != nil { + panic(err) + } + restPort = port + return test.StartTestFileServer() +} + +func httpRequest[T any](method string, path string, req any) (int, *model.Result[T]) { + var body io.Reader + if req != nil { + buf, _ := json.Marshal(req) + body = bytes.NewBuffer(buf) + } + + request, err := http.NewRequest(method, fmt.Sprintf("http://127.0.0.1:%d%s", restPort, path), body) + if err != nil { + panic(err) + } + response, err := http.DefaultClient.Do(request) + if err != nil { + panic(err) + } + defer response.Body.Close() + + var r model.Result[T] + if err := json.NewDecoder(response.Body).Decode(&r); err != nil { + panic(err) + } + return response.StatusCode, &r +} + +func httpRequestCheckOk[T any](method string, path string, req any) T { + code, result := httpRequest[T](method, path, req) + checkOk(code) + return result.Data +} + +func checkOk(code int) { + checkCode(code, 200) +} + +func checkCode(code int, exceptCode int) { + if code != exceptCode { + panic(fmt.Sprintf("code got = %d, want %d", code, exceptCode)) + } +} diff --git a/pkg/rest/util/util.go b/pkg/rest/util/util.go new file mode 100644 index 00000000..8c86fdeb --- /dev/null +++ b/pkg/rest/util/util.go @@ -0,0 +1,25 @@ +package util + +import ( + "encoding/json" + "github.com/monkeyWie/gopeed-core/pkg/rest/model" + "net/http" +) + +func ReadJson(w http.ResponseWriter, r *http.Request, v any) bool { + if err := json.NewDecoder(r.Body).Decode(v); err != nil { + WriteJson(w, 500, model.NewResultWithMsg(err.Error())) + return false + } + return true +} + +func WriteJsonOk(w http.ResponseWriter, v any) { + WriteJson(w, 200, v) +} + +func WriteJson(w http.ResponseWriter, code int, v any) { + w.WriteHeader(code) + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(v) +} diff --git a/pkg/test/httptest.go b/pkg/test/httptest.go index 1f64f1b6..0fed6ea5 100644 --- a/pkg/test/httptest.go +++ b/pkg/test/httptest.go @@ -1,6 +1,7 @@ package test import ( + "encoding/json" "fmt" "io" "math/rand" @@ -61,6 +62,29 @@ func StartTestRetryServer() net.Listener { }) } +func StartTestPostServer() net.Listener { + return startTestServer(func() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/"+BuildName, func(writer http.ResponseWriter, request *http.Request) { + if request.Method == "POST" && request.Header.Get("Authorization") != "" { + var data map[string]interface{} + if err := json.NewDecoder(request.Body).Decode(&data); err != nil { + panic(err) + } + if data["name"] == BuildName { + 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 {