feat: support RESTful api (#33)

This commit is contained in:
monkeyWie authored and GitHub committed 2022-09-09 18:31:57 +08:00
1 parent e292926697
commit c6aa83194f
22 files changed
+603 -125

No files matched your search

+6 -13
View File
@@ -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")
)
+14 -16
View File
@@ -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"`
}
+75 -36
View File
@@ -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 {
+9 -9
View File
@@ -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)
}
}
+11 -8
View File
@@ -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(),
}
}
+79
View File
@@ -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))
}
+21
View File
@@ -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)
}
+8
View File
@@ -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"`
}
+49
View File
@@ -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
}
+212
View File
@@ -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))
}
}
+25
View File
@@ -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)
}
+24
View File
@@ -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 {