diff --git a/.gitignore b/.gitignore index 2a726371..d005dfe3 100644 --- a/.gitignore +++ b/.gitignore @@ -21,4 +21,5 @@ bin/ .torrent.db-shm .torrent.db-wal -*.data \ No newline at end of file +*.data +*.db \ No newline at end of file diff --git a/bind/desktop/main.go b/bind/desktop/main.go new file mode 100644 index 00000000..7e219115 --- /dev/null +++ b/bind/desktop/main.go @@ -0,0 +1,30 @@ +// go build -ldflags="-w -s" -buildmode=c-shared -o bin/libgopeed.dll github.com/monkeyWie/gopeed-core/bind/desktop +// go build -ldflags="-w -s" -buildmode=c-shared -o bin/libgopeed.dylib github.com/monkeyWie/gopeed-core/bind/desktop +package main + +import "C" +import ( + "encoding/json" + "github.com/monkeyWie/gopeed-core/pkg/rest" + "github.com/monkeyWie/gopeed-core/pkg/rest/model" +) + +func main() {} + +//export Start +func Start(cfg *C.char) (int, *C.char) { + var config model.StartConfig + if err := json.Unmarshal([]byte(C.GoString(cfg)), &config); err != nil { + return 0, C.CString(err.Error()) + } + realPort, err := rest.Start(&config) + if err != nil { + return 0, C.CString(err.Error()) + } + return realPort, nil +} + +//export Stop +func Stop() { + rest.Stop() +} diff --git a/bind/mobile/main.go b/bind/mobile/main.go new file mode 100644 index 00000000..924bdd51 --- /dev/null +++ b/bind/mobile/main.go @@ -0,0 +1,20 @@ +// gomobile bind -ldflags="-w -s" -o libgopeed.aar -target=android -androidapi 19 -javapkg=com.gopeed github.com/monkeyWie/gopeed-core/bind/mobile +package libgopeed + +import ( + "encoding/json" + "github.com/monkeyWie/gopeed-core/pkg/rest" + "github.com/monkeyWie/gopeed-core/pkg/rest/model" +) + +func Start(cfg string) (int, error) { + var config model.StartConfig + if err := json.Unmarshal([]byte(cfg), &config); err != nil { + return 0, err + } + return rest.Start(&config) +} + +func Stop() { + rest.Stop() +} diff --git a/go.mod b/go.mod index d0973932..0cc07815 100644 --- a/go.mod +++ b/go.mod @@ -4,10 +4,10 @@ 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 + github.com/matoous/go-nanoid/v2 v2.0.0 + go.etcd.io/bbolt v1.3.6 + golang.org/x/sync v0.0.0-20220819030929-7fc1605a5dde ) require ( @@ -37,6 +37,7 @@ require ( github.com/dustin/go-humanize v1.0.0 // indirect github.com/edsrzf/mmap-go v1.1.0 // indirect github.com/google/btree v1.0.1 // indirect + github.com/google/uuid v1.3.0 // indirect github.com/gorilla/websocket v1.4.2 // indirect github.com/huandu/xstrings v1.3.2 // indirect github.com/lispad/go-generics-tools v1.0.0 // indirect @@ -61,10 +62,13 @@ require ( github.com/pkg/errors v0.9.1 // indirect github.com/rs/dnscache v0.0.0-20210201191234-295bba877686 // indirect github.com/tidwall/btree v1.3.1 // indirect - go.etcd.io/bbolt v1.3.6 // indirect golang.org/x/crypto v0.0.0-20220131195533-30dcbda58838 // indirect golang.org/x/exp v0.0.0-20220428152302-39d4317da171 // indirect + golang.org/x/mobile v0.0.0-20220928052126-fa6bcb076835 // indirect + golang.org/x/mod v0.6.0-dev.0.20211013180041-c96bc1413d57 // indirect + golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e // indirect golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac // indirect + golang.org/x/tools v0.1.8-0.20211029000441-d6a9af8af023 // indirect golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect ) diff --git a/go.sum b/go.sum index 40af7d12..d4342366 100644 --- a/go.sum +++ b/go.sum @@ -7,6 +7,7 @@ crawshaw.io/sqlite v0.3.3-0.20210127221821-98b1f83c5508 h1:fILCBBFnjnrQ0whVJlGhf crawshaw.io/sqlite v0.3.3-0.20210127221821-98b1f83c5508/go.mod h1:igAO5JulrQ1DbdZdtVq48mnZUBAPOeFzer7VhDWNtW4= filippo.io/edwards25519 v1.0.0-rc.1 h1:m0VOOB23frXZvAOK44usCgLWvtsxIoMCTBGJZlpmGfU= github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= +github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= github.com/RoaringBitmap/roaring v0.4.7/go.mod h1:8khRDP4HmeXns4xIj9oGrKSz7XTQiJx2zgh7AcNke4w= github.com/RoaringBitmap/roaring v0.4.17/go.mod h1:D3qVegWTmfCaX4Bl5CrBE9hfrSrrXIr8KVNvRsDi1NI= github.com/RoaringBitmap/roaring v0.4.23/go.mod h1:D0gp8kJQgE1A4LQ5wFLggQEyvDi06Mq5mKs52e1TwOo= @@ -189,6 +190,9 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/lispad/go-generics-tools v1.0.0 h1:pPtt9QERE5sGQfwvLakKe2sjhcNKKsDSnr3939ofj1E= github.com/lispad/go-generics-tools v1.0.0/go.mod h1:stn7X24ZIyFvaSyttafq3VlJzGJJJkUtLYdbgi/gopM= +github.com/matoous/go-nanoid v1.5.0/go.mod h1:zyD2a71IubI24efhpvkJz+ZwfwagzgSO6UNiFsZKN7U= +github.com/matoous/go-nanoid/v2 v2.0.0 h1:d19kur2QuLeHmJBkvYkFdhFBzLoo1XVm2GgTpL+9Tj0= +github.com/matoous/go-nanoid/v2 v2.0.0/go.mod h1:FtS4aGPVfEkxKxhdWPAspZpZSh1cOjtM7Ej/So3hR0g= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -308,6 +312,7 @@ github.com/tinylib/msgp v1.1.2/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDW github.com/willf/bitset v1.1.9/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4= github.com/willf/bitset v1.1.10/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.4.0/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= go.etcd.io/bbolt v1.3.6 h1:/ecaJf0sk1l4l6V4awd65v2C3ILy7MSj+s/x1ADCIMU= go.etcd.io/bbolt v1.3.6/go.mod h1:qXsaaIqmgQH0T+OPdb99Bf+PKfBBQVAdyD6TY9G8XM4= go.opencensus.io v0.20.1/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk= @@ -315,19 +320,30 @@ go.opencensus.io v0.20.2/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk= go.opencensus.io v0.22.3/go.mod h1:yxeiOL68Rb0Xd1ddK5vPZ/oVn4vY4Ynel7k9FzqtOIw= golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.0.0-20220131195533-30dcbda58838 h1:71vQrMauZZhcTVK6KdYM+rklehEEwb3E+ZhaE5jrPrE= golang.org/x/crypto v0.0.0-20220131195533-30dcbda58838/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= +golang.org/x/exp v0.0.0-20190731235908-ec7cb31e5a56/go.mod h1:JhuoJpWY28nO4Vef9tZUw9qufEGTyX1+7lmHxV5q5G4= golang.org/x/exp v0.0.0-20220428152302-39d4317da171 h1:TfdoLivD44QwvssI9Sv1xwa5DcL5XQr4au4sZ2F2NV4= golang.org/x/exp v0.0.0-20220428152302-39d4317da171/go.mod h1:lgLbSvA5ygNOMpwM/9anMpWVlVJ7Z+cHWq/eFuinpGE= +golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js= +golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= golang.org/x/lint v0.0.0-20190301231843-5614ed5bae6f/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE= +golang.org/x/mobile v0.0.0-20220928052126-fa6bcb076835 h1:v6yt1xSu+4xV0fPEmB8s2BHdOaoU0SSZrVQzj6pRvqk= +golang.org/x/mobile v0.0.0-20220928052126-fa6bcb076835/go.mod h1:M32cGdzp91A8Ex9qQtyZinr19EYxzkFqDjW2oyHzTDQ= +golang.org/x/mod v0.1.0/go.mod h1:0QHyrYULN0/3qlju5TqG8bIK38QM8yzMo5ekMj3DlcY= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.6.0-dev.0.20211013180041-c96bc1413d57 h1:LQmS1nU0twXLA96Kt7U9qtHJEbBk3z6Q0V4UXjZkpr4= +golang.org/x/mod v0.6.0-dev.0.20211013180041-c96bc1413d57/go.mod h1:3p9vT2HGsQu2K1YbXdKPJLVgG5VJdoTa1poYQBtP1AY= golang.org/x/net v0.0.0-20180524181706-dfa909b99c79/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= @@ -345,6 +361,7 @@ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwY golang.org/x/net v0.0.0-20201201195509-5d6afe98e0b7/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk= +golang.org/x/net v0.0.0-20210805182204-aaa1db679c0d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211201190559-0a0e4e1bb54c/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= golang.org/x/net v0.0.0-20211216030914-fe4d6282115f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= @@ -359,8 +376,9 @@ golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ= golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20220819030929-7fc1605a5dde h1:ejfdSekXMDxDLbRrJMwUk6KnSLZ2McaUCVcIKM+N6jc= +golang.org/x/sync v0.0.0-20220819030929-7fc1605a5dde/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= @@ -383,6 +401,7 @@ golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM= golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= @@ -399,9 +418,13 @@ golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGm golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20190226205152-f727befe758c/go.mod h1:9Yl7xja0Znq3iFh3HoIrodX9oNMXvdceNzlUR8zjMvY= golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20190312151545-0bb0c0a6e846/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20190312170243-e65039ee4138/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.1.8-0.20211022200916-316ba0b74098/go.mod h1:LGqMHiF4EqQNHR1JncWGqT5BVaXmza+X+BDGol+dOxo= +golang.org/x/tools v0.1.8-0.20211029000441-d6a9af8af023 h1:0c3L82FDQ5rt1bjTBlchS8t6RQ6299/+5bWMnRLh+uI= +golang.org/x/tools v0.1.8-0.20211029000441-d6a9af8af023/go.mod h1:nABZi5QlRsZVlzPpHl034qft6wpY4eDcsTt5AaioBiU= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/internal/controller/controller.go b/internal/controller/controller.go index cae683ca..dfea961e 100644 --- a/internal/controller/controller.go +++ b/internal/controller/controller.go @@ -1,18 +1,13 @@ package controller import ( - "golang.org/x/net/proxy" - "net" "os" - "time" + "path/filepath" ) 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) + //ContextDialer() (proxy.Dialer, error) } type DefaultController struct { @@ -24,6 +19,10 @@ func NewController() *DefaultController { } func (c *DefaultController) Touch(name string, size int64) (file *os.File, err error) { + dir := filepath.Dir(name) + if err = os.MkdirAll(dir, os.ModePerm); err != nil { + return + } file, err = os.Create(name) if size > 0 { err = os.Truncate(name, size) @@ -37,25 +36,7 @@ func (c *DefaultController) Touch(name string, size int64) (file *os.File, err e return } -func (c *DefaultController) Open(name string) (file *os.File, err error) { - file, err = os.OpenFile(name, os.O_RDWR, os.ModePerm) - if err == nil { - c.Files[name] = file - } - return -} - -func (c *DefaultController) Write(name string, offset int64, buf []byte) (int, error) { - return c.Files[name].WriteAt(buf, offset) -} - -func (c *DefaultController) Close(name string) error { - err := c.Files[name].Close() - delete(c.Files, name) - return err -} - -func (c *DefaultController) 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 @@ -107,4 +88,4 @@ func (d *DialerWarp) Dial(network, addr string) (c net.Conn, err error) { return nil, err } return &ConnWarp{conn: conn}, nil -} +}*/ diff --git a/internal/fetcher/fetcher.go b/internal/fetcher/fetcher.go index f57873db..01ff3592 100644 --- a/internal/fetcher/fetcher.go +++ b/internal/fetcher/fetcher.go @@ -15,6 +15,7 @@ type Fetcher interface { Start() (err error) Pause() (err error) Continue() (err error) + Close() (err error) // 获取任务各个文件下载进度 Progress() Progress @@ -25,13 +26,11 @@ type Fetcher interface { type FetcherBuilder interface { Schemes() []string Build() Fetcher -} -type FetcherSaver interface { // Store 存储任务 - Store(fetcher Fetcher) interface{} - // Resume 恢复任务 - Resume(res *base.Resource, opts *base.Options, data interface{}) Fetcher + Store(fetcher Fetcher) (any, error) + // Restore 恢复任务 + Restore() (v any, f func(res *base.Resource, opts *base.Options, v any) Fetcher) } type DefaultFetcher struct { diff --git a/internal/protocol/bt/fetcher.go b/internal/protocol/bt/fetcher.go index bfd80610..52afadc8 100644 --- a/internal/protocol/bt/fetcher.go +++ b/internal/protocol/bt/fetcher.go @@ -2,12 +2,16 @@ package bt import ( "github.com/anacrolix/torrent" + "github.com/anacrolix/torrent/metainfo" + "github.com/anacrolix/torrent/storage" "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/util" "path/filepath" "sync" + "sync/atomic" + "time" ) var client *torrent.Client @@ -16,83 +20,184 @@ type Fetcher struct { Ctl controller.Controller torrent *torrent.Torrent + res *base.Resource opts *base.Options + + ready atomic.Bool + progress fetcher.Progress + + torrentPaths map[string]string } func (f *Fetcher) Setup(ctl controller.Controller) (err error) { f.Ctl = ctl var once sync.Once once.Do(func() { - client, err = torrent.NewClient(nil) + cfg := torrent.NewDefaultClientConfig() + cfg.ListenPort = 0 + + // Support custom download path for each torrent + pieceCompletion, err := storage.NewDefaultPieceCompletionForDir(cfg.DataDir) + if err != nil { + pieceCompletion = storage.NewMapPieceCompletion() + } + clientImplCloser := storage.NewFileOpts(storage.NewFileClientOpts{ + ClientBaseDir: cfg.DataDir, + TorrentDirMaker: func(baseDir string, info *metainfo.Info, infoHash metainfo.Hash) string { + if dir, ok := f.torrentPaths[infoHash.String()]; ok { + return dir + } + return baseDir + }, + PieceCompletion: pieceCompletion, + }) + cfg.DefaultStorage = clientImplCloser + client, err = torrent.NewClient(cfg) + if err != nil { + return + } + + f.torrentPaths = make(map[string]string) }) return } func (f *Fetcher) Wait() (err error) { - client.WaitAll() + for { + if f.ready.Load() { + done := true + for _, selectIndex := range f.opts.SelectFiles { + file := f.torrent.Files()[selectIndex] + if file.BytesCompleted() < file.Length() { + done = false + break + } + } + if done { + // remove unselected files + for i, file := range f.torrent.Files() { + selected := false + for _, selectIndex := range f.opts.SelectFiles { + if i == selectIndex { + selected = true + break + } + } + if !selected { + util.SafeRemove(filepath.Join(f.opts.Path, file.Path())) + } + } + break + } + } + time.Sleep(time.Millisecond * 500) + } return nil } func (f *Fetcher) Resolve(req *base.Request) (res *base.Resource, err error) { - schema := util.ParseSchema(req.URL) - if schema == "MAGNET" { - f.torrent, err = client.AddMagnet(req.URL) - } else { - f.torrent, err = client.AddTorrentFromFile(req.URL) - } - if err != nil { + if err = f.addTorrent(req.URL, true); err != nil { return } - <-f.torrent.GotInfo() + defer f.torrent.Drop() res = &base.Resource{ Req: req, - Size: f.torrent.Length(), + Name: f.torrent.Name(), Range: true, Files: make([]*base.FileInfo, len(f.torrent.Files())), + Extra: map[string]any{ + "infoHash": f.torrent.InfoHash().String(), + }, } for i, file := range f.torrent.Files() { res.Files[i] = &base.FileInfo{ Name: filepath.Base(file.DisplayPath()), - Path: filepath.Dir(file.Path()), + Path: util.Dir(file.Path()), Size: file.Length(), } + res.Size += file.Length() } return } func (f *Fetcher) Create(res *base.Resource, opts *base.Options) (err error) { + f.res = res f.opts = opts + if len(f.opts.SelectFiles) == 0 { + f.opts.SelectFiles = make([]int, 0) + for i := range f.torrent.Files() { + f.opts.SelectFiles = append(f.opts.SelectFiles, i) + } + } + if f.opts.Path != "" { + if extra, ok := res.Extra.(map[string]any); ok { + f.torrentPaths[extra["infoHash"].(string)] = f.opts.Path + } + } + f.progress = make(fetcher.Progress, len(f.opts.SelectFiles)) + f.ready.Store(false) return nil } func (f *Fetcher) Start() (err error) { - if len(f.opts.SelectFiles) == 0 { + if err = f.addTorrent(f.res.Req.URL, false); err != nil { + return + } + files := f.torrent.Files() + if len(f.opts.SelectFiles) == len(files) { f.torrent.DownloadAll() } else { - for _, index := range f.opts.SelectFiles { - f.torrent.Files()[index].Download() + for _, selectIndex := range f.opts.SelectFiles { + file := files[selectIndex] + file.Download() } } return } func (f *Fetcher) Pause() (err error) { - f.torrent.DisallowDataDownload() + f.ready.Store(false) + f.torrent.Drop() return } func (f *Fetcher) Continue() (err error) { - f.torrent.AllowDataDownload() - return + return f.Start() +} + +func (f *Fetcher) Close() (err error) { + f.torrent.Drop() + return nil } func (f *Fetcher) Progress() fetcher.Progress { - p := make(fetcher.Progress, len(f.opts.SelectFiles)) - for i := range p { - file := f.torrent.Files()[i] - p[i] = file.BytesCompleted() + if !f.ready.Load() { + return f.progress } - return p + for i := range f.progress { + selectIndex := f.opts.SelectFiles[i] + file := f.torrent.Files()[selectIndex] + f.progress[i] = file.BytesCompleted() + } + return f.progress +} + +func (f *Fetcher) addTorrent(url string, resolve bool) (err error) { + schema := util.ParseSchema(url) + if schema == "MAGNET" { + f.torrent, err = client.AddMagnet(url) + } else { + f.torrent, err = client.AddTorrentFromFile(url) + } + if err != nil { + return + } + <-f.torrent.GotInfo() + if resolve { + return + } + f.ready.Store(true) + return } type FetcherBuilder struct { @@ -107,3 +212,11 @@ func (fb *FetcherBuilder) Schemes() []string { func (fb *FetcherBuilder) Build() fetcher.Fetcher { return &Fetcher{} } + +func (fb *FetcherBuilder) Store(f fetcher.Fetcher) (data any, err error) { + return nil, nil +} + +func (fb *FetcherBuilder) Restore() (v any, f func(res *base.Resource, opts *base.Options, v any) fetcher.Fetcher) { + return nil, nil +} diff --git a/internal/protocol/bt/fetcher_test.go b/internal/protocol/bt/fetcher_test.go index 1528daa8..1acacd2a 100644 --- a/internal/protocol/bt/fetcher_test.go +++ b/internal/protocol/bt/fetcher_test.go @@ -19,15 +19,18 @@ func TestFetcher_Resolve_Torrent(t *testing.T) { want := &base.Resource{ Req: res.Req, + Name: "ubuntu-22.04-live-server-amd64.iso", Size: 1466714112, Range: true, Files: []*base.FileInfo{ { Name: "ubuntu-22.04-live-server-amd64.iso", - Path: ".", Size: 1466714112, }, }, + Extra: map[string]any{ + "infoHash": "8a55cfbd5ca5d11507364765936c4f9e55b253ed", + }, } if !reflect.DeepEqual(want, res) { t.Errorf("Resolve() got = %v, want %v", res, want) diff --git a/internal/protocol/http/fetcher.go b/internal/protocol/http/fetcher.go index 3c8b0b23..61909a7e 100644 --- a/internal/protocol/http/fetcher.go +++ b/internal/protocol/http/fetcher.go @@ -6,12 +6,14 @@ import ( "fmt" "github.com/monkeyWie/gopeed-core/internal/fetcher" "github.com/monkeyWie/gopeed-core/pkg/base" + "github.com/monkeyWie/gopeed-core/pkg/util" "golang.org/x/sync/errgroup" "io" "mime" "net/http" "net/http/cookiejar" "net/url" + "os" "path" "path/filepath" "strconv" @@ -38,8 +40,9 @@ type Fetcher struct { res *base.Resource opts *base.Options status base.Status - chunks []*Chunk + chunks []*chunk + file *os.File ctx context.Context cancel context.CancelFunc pauseCh chan interface{} @@ -95,20 +98,21 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) { contentDisposition := httpResp.Header.Get(base.HttpHeaderContentDisposition) if contentDisposition != "" { _, params, _ := mime.ParseMediaType(contentDisposition) - filename := params["filename"] + filename := params["filePath"] if filename != "" { file.Name = filename } } - // Get file filename by URL + // Get file filePath by URL if file.Name == "" && strings.Count(req.URL, "/") > 2 { file.Name = filepath.Base(req.URL) } - // unknown file filename + // unknown file filePath if file.Name == "" { file.Name = "unknown" } res.Files = append(res.Files, file) + res.Name = file.Name return res, nil } @@ -121,35 +125,13 @@ func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error { func (f *Fetcher) Start() (err error) { // 创建文件 - name := f.filename() - _, err = f.Ctl.Touch(name, f.res.Size) + name := f.filepath() + f.file, err = f.Ctl.Touch(name, f.res.Size) if err != nil { return err } f.status = base.DownloadStatusRunning - if f.res.Range { - // 每个连接平均需要下载的分块大小 - chunkSize := f.res.Size / int64(f.opts.Connections) - f.chunks = make([]*Chunk, 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.Size - 1 - } else { - end = begin + chunkSize - 1 - } - chunk := NewChunk(begin, end) - f.chunks[i] = chunk - } - } else { - // 只支持单连接下载 - f.chunks = make([]*Chunk, 1) - f.chunks[0] = NewChunk(0, 0) - } + f.chunks = splitChunk(f.res, f.opts) f.fetch() return } @@ -169,8 +151,7 @@ func (f *Fetcher) Continue() (err error) { return } f.status = base.DownloadStatusRunning - var name = f.filename() - _, err = f.Ctl.Open(name) + f.file, err = os.OpenFile(f.filepath(), os.O_RDWR, os.ModeAppend) if err != nil { return err } @@ -178,6 +159,13 @@ func (f *Fetcher) Continue() (err error) { return } +func (f *Fetcher) Close() (err error) { + if err = f.Pause(); err != nil { + return + } + return +} + func (f *Fetcher) Progress() fetcher.Progress { p := make(fetcher.Progress, 0) if len(f.chunks) > 0 { @@ -190,13 +178,8 @@ func (f *Fetcher) Progress() fetcher.Progress { 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) filepath() string { + return util.Filepath(f.opts.Path, f.res.Files[0].Name, f.opts.Name) } func (f *Fetcher) fetch() { @@ -212,7 +195,7 @@ func (f *Fetcher) fetch() { go func() { err := eg.Wait() // 下载停止，关闭文件句柄 - f.Ctl.Close(f.filename()) + f.file.Close() if f.status == base.DownloadStatusPause { f.pauseCh <- nil } else { @@ -227,7 +210,6 @@ func (f *Fetcher) fetch() { } func (f *Fetcher) fetchChunk(index int) (err error) { - filename := f.filename() chunk := f.chunks[index] httpReq, err := buildRequest(f.ctx, f.res.Req) @@ -280,7 +262,7 @@ func (f *Fetcher) fetchChunk(index int) (err error) { for { n, err := resp.Body.Read(buf) if n > 0 { - _, err := f.Ctl.Write(filename, chunk.Begin+chunk.Downloaded, buf[:n]) + _, err := f.file.WriteAt(buf[:n], chunk.Begin+chunk.Downloaded) if err != nil { return false, err } @@ -316,6 +298,33 @@ func (f *Fetcher) fetchChunk(index int) (err error) { return } +func splitChunk(res *base.Resource, opts *base.Options) (chunks []*chunk) { + if res.Range { + // 每个连接平均需要下载的分块大小 + chunkSize := res.Size / int64(opts.Connections) + chunks = make([]*chunk, opts.Connections) + for i := 0; i < opts.Connections; i++ { + var ( + begin = chunkSize * int64(i) + end int64 + ) + if i == opts.Connections-1 { + // 最后一个分块需要保证把文件下载完 + end = res.Size - 1 + } else { + end = begin + chunkSize - 1 + } + chunk := newChunk(begin, end) + chunks[i] = chunk + } + } else { + // 只支持单连接下载 + chunks = make([]*chunk, 1) + chunks[0] = newChunk(0, 0) + } + return +} + func buildClient() *http.Client { // Cookie handle jar, _ := cookiejar.New(nil) @@ -339,7 +348,7 @@ func buildRequest(ctx context.Context, req *base.Request) (httpReq *http.Request if req.Extra == nil { method = http.MethodGet } else { - extra := req.Extra.(Extra) + extra := req.Extra.(extra) if extra.Method != "" { method = extra.Method } else { @@ -367,8 +376,8 @@ func buildRequest(ctx context.Context, req *base.Request) (httpReq *http.Request return httpReq, nil } -type FetcherData struct { - Chunks []*Chunk +type fetcherData struct { + Chunks []*chunk } type FetcherBuilder struct { @@ -387,19 +396,26 @@ func (fb *FetcherBuilder) Build() fetcher.Fetcher { } } -func (fb *FetcherBuilder) Store(f fetcher.Fetcher) interface{} { +func (fb *FetcherBuilder) Store(f fetcher.Fetcher) (data any, err error) { _f := f.(*Fetcher) - return &FetcherData{ + return &fetcherData{ Chunks: _f.chunks, - } + }, nil } -func (fb *FetcherBuilder) Resume(res *base.Resource, opts *base.Options, data interface{}) fetcher.Fetcher { - fetcher := new(FetcherBuilder).Build().(*Fetcher) - fs := data.(*FetcherData) - fetcher.res = res - fetcher.opts = opts - fetcher.chunks = fs.Chunks - fetcher.status = base.DownloadStatusPause - return fetcher +func (fb *FetcherBuilder) Restore() (v any, f func(res *base.Resource, opts *base.Options, v any) fetcher.Fetcher) { + return &fetcherData{}, func(res *base.Resource, opts *base.Options, v any) fetcher.Fetcher { + fd := v.(*fetcherData) + fb := &FetcherBuilder{} + fetcher := fb.Build().(*Fetcher) + fetcher.res = res + fetcher.opts = opts + fetcher.status = base.DownloadStatusPause + if len(fd.Chunks) == 0 { + fetcher.chunks = splitChunk(res, opts) + } else { + fetcher.chunks = fd.Chunks + } + return fetcher + } } diff --git a/internal/protocol/http/fetcher_test.go b/internal/protocol/http/fetcher_test.go index 9d359c67..66b079f6 100644 --- a/internal/protocol/http/fetcher_test.go +++ b/internal/protocol/http/fetcher_test.go @@ -4,8 +4,8 @@ import ( "fmt" "github.com/monkeyWie/gopeed-core/internal/controller" "github.com/monkeyWie/gopeed-core/internal/fetcher" + "github.com/monkeyWie/gopeed-core/internal/test" "github.com/monkeyWie/gopeed-core/pkg/base" - "github.com/monkeyWie/gopeed-core/pkg/test" "net" "reflect" "testing" @@ -14,6 +14,7 @@ import ( func TestFetcher_Resolve(t *testing.T) { testResolve(test.StartTestFileServer, &base.Resource{ + Name: test.BuildName, Size: test.BuildSize, Range: true, Files: []*base.FileInfo{ @@ -24,6 +25,7 @@ func TestFetcher_Resolve(t *testing.T) { }, }, t) testResolve(test.StartTestChunkedServer, &base.Resource{ + Name: test.BuildName, Size: 0, Range: false, Files: []*base.FileInfo{ @@ -148,7 +150,7 @@ func downloadNormal(listener net.Listener, connections int, t *testing.T) { func downloadPost(listener net.Listener, connections int, t *testing.T) { fetcher, _, _ := downloadReady(listener, connections, t) - fetcher.(*Fetcher).res.Req.Extra = Extra{ + fetcher.(*Fetcher).res.Req.Extra = extra{ Method: "POST", Header: map[string]string{ "Authorization": "Bearer 123456", @@ -176,11 +178,11 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) { if err != nil { t.Fatal(err) } - time.Sleep(time.Millisecond * 200) + time.Sleep(time.Millisecond * 50) if err := fetcher.Pause(); err != nil { t.Fatal(err) } - time.Sleep(time.Millisecond * 200) + time.Sleep(time.Millisecond * 50) if err := fetcher.Continue(); err != nil { t.Fatal(err) } @@ -215,12 +217,19 @@ func downloadResume(listener net.Listener, connections int, t *testing.T) { } fb := new(FetcherBuilder) - time.Sleep(time.Millisecond * 200) - data := fb.Store(fetcher) - time.Sleep(time.Millisecond * 200) + time.Sleep(time.Millisecond * 50) + data, err := fb.Store(fetcher) + if err != nil { + t.Fatal(err) + } + time.Sleep(time.Millisecond * 50) fetcher.Pause() - fetcher = fb.Resume(res, opts, data) + _, f := fb.Restore() + f(res, opts, data) + if err != nil { + t.Fatal(err) + } fetcher.Setup(controller.NewController()) fetcher.Continue() diff --git a/internal/protocol/http/model.go b/internal/protocol/http/model.go index 1219791d..dcf97c2e 100644 --- a/internal/protocol/http/model.go +++ b/internal/protocol/http/model.go @@ -2,22 +2,22 @@ package http import "github.com/monkeyWie/gopeed-core/pkg/base" -type Chunk struct { +type chunk struct { Status base.Status Begin int64 End int64 Downloaded int64 } -func NewChunk(begin int64, end int64) *Chunk { - return &Chunk{ +func newChunk(begin int64, end int64) *chunk { + return &chunk{ Status: base.DownloadStatusReady, Begin: begin, End: end, } } -type Extra struct { +type extra struct { Method string Header map[string]string Body string diff --git a/pkg/test/httptest.go b/internal/test/httptest.go similarity index 88% rename from pkg/test/httptest.go rename to internal/test/httptest.go index 0fed6ea5..0a371a43 100644 --- a/pkg/test/httptest.go +++ b/internal/test/httptest.go @@ -8,6 +8,7 @@ import ( "net" "net/http" "os" + "time" ) const ( @@ -26,6 +27,25 @@ func StartTestFileServer() net.Listener { }) } +type SlowFileServer struct { + delay time.Duration + handler http.Handler +} + +func (s *SlowFileServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + time.Sleep(s.delay) + s.handler.ServeHTTP(w, r) +} + +func StartTestSlowFileServer(delay time.Duration) net.Listener { + return startTestServer(func() http.Handler { + return &SlowFileServer{ + delay: delay, + handler: http.FileServer(http.Dir(Dir)), + } + }) +} + func StartTestChunkedServer() net.Listener { return startTestServer(func() http.Handler { mux := http.NewServeMux() diff --git a/pkg/test/util.go b/internal/test/util.go similarity index 61% rename from pkg/test/util.go rename to internal/test/util.go index 588b9b2b..84501269 100644 --- a/pkg/test/util.go +++ b/internal/test/util.go @@ -6,6 +6,7 @@ import ( "encoding/json" "io" "os" + "path/filepath" ) func FileMd5(filePath string) string { @@ -26,6 +27,24 @@ func FileMd5(filePath string) string { return hex.EncodeToString(hash.Sum(nil)) } +func DirMd5(dirPath string) string { + hash := md5.New() + filepath.Walk(dirPath, func(path string, info os.FileInfo, err error) error { + if info.IsDir() { + return nil + } + file, err := os.Open(path) + if err != nil { + return err + } + if _, err := io.Copy(hash, file); err != nil { + return err + } + return nil + }) + return hex.EncodeToString(hash.Sum(nil)) +} + func ToJson(v interface{}) string { buf, _ := json.Marshal(v) return string(buf) diff --git a/pkg/base/model.go b/pkg/base/model.go index 05c94bcb..f41b10c3 100644 --- a/pkg/base/model.go +++ b/pkg/base/model.go @@ -11,12 +11,16 @@ type Request struct { // 资源信息 type Resource struct { Req *Request `json:"req"` + // 资源名称 + Name string `json:"name"` // 资源总大小 Size int64 `json:"size"` // 是否支持断点下载 Range bool `json:"range"` // 资源所包含的文件列表 Files []*FileInfo `json:"files"` + // 附加信息 + Extra interface{} `json:"extra"` } type FileInfo struct { @@ -34,5 +38,5 @@ type Options struct { // 并发连接数 Connections int `json:"connections"` // 选择下载的文件下标列表 - SelectFiles []int `json:"select_files"` + SelectFiles []int `json:"selectFiles"` } diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 8904f3f0..577573f2 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -10,9 +10,16 @@ import ( "github.com/monkeyWie/gopeed-core/pkg/util" "strings" "sync" + "sync/atomic" "time" ) +const ( + bucketTask = "task" + bucketSave = "save" + bucketConfig = "config" +) + type Listener func(event *Event) type Progress struct { @@ -25,46 +32,133 @@ type Progress struct { } type Downloader struct { - *controller.DefaultController + controller *controller.DefaultController fetchBuilders map[string]fetcher.FetcherBuilder + storage Storage tasks []*Task listener Listener + + refreshInterval int + lock *sync.Mutex + closed atomic.Bool } -func NewDownloader(fbs ...fetcher.FetcherBuilder) *Downloader { - d := &Downloader{DefaultController: controller.NewController()} - d.fetchBuilders = make(map[string]fetcher.FetcherBuilder) - for _, f := range fbs { +type DownloaderConfig struct { + Controller *controller.DefaultController + FetchBuilders []fetcher.FetcherBuilder + Storage Storage + + // RefreshInterval time duration to refresh task progress(ms) + RefreshInterval int +} + +func (cfg *DownloaderConfig) Init() *DownloaderConfig { + if cfg.Controller == nil { + cfg.Controller = controller.NewController() + } + if len(cfg.FetchBuilders) == 0 { + cfg.FetchBuilders = []fetcher.FetcherBuilder{ + new(http.FetcherBuilder), + new(bt.FetcherBuilder), + } + } + if cfg.Storage == nil { + cfg.Storage = NewMemStorage() + } + if cfg.RefreshInterval == 0 { + cfg.RefreshInterval = 1000 + } + return cfg +} + +func NewDownloader(cfg *DownloaderConfig) *Downloader { + if cfg == nil { + cfg = &DownloaderConfig{} + } + cfg.Init() + + d := &Downloader{ + controller: cfg.Controller, + fetchBuilders: make(map[string]fetcher.FetcherBuilder), + refreshInterval: cfg.RefreshInterval, + storage: cfg.Storage, + lock: &sync.Mutex{}, + } + for _, f := range cfg.FetchBuilders { for _, p := range f.Schemes() { d.fetchBuilders[strings.ToUpper(p)] = f } } - d.tasks = make([]*Task, 0) - - // 每秒统计一次下载速度 - 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) { +func (d *Downloader) Setup() error { + // setup storage + if err := d.storage.Setup([]string{bucketTask, bucketSave, bucketConfig}); err != nil { + return err + } + // load tasks from storage + var tasks []*Task + if err := d.storage.List(bucketTask, &tasks); err != nil { + return err + } + if tasks == nil { + tasks = make([]*Task, 0) + } else { + for _, task := range tasks { + initTask(task) + if task.Status != base.DownloadStatusDone && task.Status != base.DownloadStatusError { + task.Status = base.DownloadStatusPause + } + } + } + d.tasks = tasks + + // 每秒统计一次下载速度 + go func() { + for !d.closed.Load() { + if len(d.tasks) > 0 { + for _, task := range d.tasks { + func() { + task.lock.Lock() + defer task.lock.Unlock() + if task.Status == base.DownloadStatusDone || + task.Status == base.DownloadStatusError || + task.Status == base.DownloadStatusPause { + return + } + // check if task is deleted + if d.GetTask(task.ID) == nil { + return + } + + 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) + + // store fetcher progress + data, err := task.fetcherBuilder.Store(task.fetcher) + if err != nil { + return // TODO log + } + if err := d.storage.Put(bucketSave, task.ID, data); err != nil { + return // TODO log + } + if err := d.storage.Put(bucketTask, task.ID, task); err != nil { + return // TODO log + } + }() + } + } + time.Sleep(time.Millisecond * time.Duration(d.refreshInterval)) + } + }() + return nil +} + +func (d *Downloader) buildFetcher(url string) (fetcher.FetcherBuilder, fetcher.Fetcher, error) { schema := util.ParseSchema(url) fetchBuilder, ok := d.fetchBuilders[schema] if !ok { @@ -72,14 +166,14 @@ func (d *Downloader) buildFetcher(url string) (fetcher.Fetcher, error) { } if ok { fetcher := fetchBuilder.Build() - fetcher.Setup(d.DefaultController) - return fetcher, nil + fetcher.Setup(d.controller) + return fetchBuilder, fetcher, nil } - return nil, errors.New("unsupported protocol") + return nil, nil, errors.New("unsupported protocol") } func (d *Downloader) Resolve(req *base.Request) (*base.Resource, error) { - fetcher, err := d.buildFetcher(req.URL) + _, fetcher, err := d.buildFetcher(req.URL) if err != nil { return nil, err } @@ -99,13 +193,13 @@ func (d *Downloader) Create(res *base.Resource, opts *base.Options) (taskId stri opts.Connections = 1 } if len(opts.SelectFiles) == 0 { - opts.SelectFiles = make([]int, len(res.Files)) - for i := range opts.SelectFiles { - opts.SelectFiles[i] = i + opts.SelectFiles = make([]int, 0) + for i := range res.Files { + opts.SelectFiles = append(opts.SelectFiles, i) } } - fetcher, err := d.buildFetcher(res.Req.URL) + fetcherBuilder, fetcher, err := d.buildFetcher(res.Req.URL) if err != nil { return } @@ -115,74 +209,144 @@ func (d *Downloader) Create(res *base.Resource, opts *base.Options) (taskId stri } task := NewTask() + task.fetcherBuilder = fetcherBuilder task.fetcher = fetcher task.Res = res task.Opts = opts 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() + // calculate select files size + for _, selectIndex := range opts.SelectFiles { + task.Size += res.Files[selectIndex].Size + } + initTask(task) + task.timer.Start() + err = func() error { + d.lock.Lock() + defer d.lock.Unlock() + if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil { + return err + } + d.tasks = append(d.tasks, task) + return nil + }() if err != nil { return } + d.emit(EventKeyStart, task) + taskId = task.ID + go func() { - err = fetcher.Wait() + err := fetcher.Start() if err != nil { - task.Status = base.DownloadStatusError d.emit(EventKeyError, task, err) - } else { - task.Progress.Used = task.timer.Used() - 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.Size / used - task.Progress.Downloaded = task.Res.Size - task.Status = base.DownloadStatusDone - d.emit(EventKeyDone, task) + return } - d.emit(EventKeyFinally, task, err) + d.watch(task) }() return } -func (d *Downloader) Pause(id string) { +func (d *Downloader) Pause(id string) (err error) { task := d.GetTask(id) if task == nil { return } - task.locker.Lock() - defer task.locker.Unlock() + task.lock.Lock() + defer task.lock.Unlock() if task.Status == base.DownloadStatusRunning { task.Status = base.DownloadStatusPause task.timer.Pause() task.fetcher.Pause() + d.storage.Put(bucketTask, task.ID, task.clone()) d.emit(EventKeyPause, task) } + return } -func (d *Downloader) Continue(id string) { +func (d *Downloader) Continue(id string) (err error) { task := d.GetTask(id) if task == nil { return } - task.locker.Lock() - defer task.locker.Unlock() + task.lock.Lock() + defer task.lock.Unlock() if task.Status == base.DownloadStatusPause || task.Status == base.DownloadStatusError { task.Status = base.DownloadStatusRunning - task.timer.Continue() - task.fetcher.Continue() + err = d.restoreFetcher(task) + if err != nil { + return + } + task.timer.Start() + err = task.fetcher.Continue() + if err != nil { + return + } + d.storage.Put(bucketTask, task.ID, task.clone()) d.emit(EventKeyContinue, task) } + return +} + +func (d *Downloader) Delete(id string, force bool) (err error) { + d.lock.Lock() + defer d.lock.Unlock() + task := d.GetTask(id) + if task == nil { + return + } + task.lock.Lock() + defer task.lock.Unlock() + if task.fetcher != nil { + err = task.fetcher.Close() + if err != nil { + return + } + } + for i, t := range d.tasks { + if t.ID == id { + d.tasks = append(d.tasks[:i], d.tasks[i+1:]...) + break + } + } + err = d.storage.Delete(bucketTask, id) + if err != nil { + return + } + err = d.storage.Delete(bucketSave, id) + if err != nil { + return + } + if force { + names := make([]string, 0) + for _, file := range task.Res.Files { + names = append(names, util.Filepath(file.Path, file.Name, task.Opts.Name)) + } + util.SafeRemoveAll(task.Opts.Path, names) + } + d.emit(EventKeyDelete, task) + task = nil + return +} + +func (d *Downloader) Close() error { + d.lock.Lock() + defer d.lock.Unlock() + d.closed.Store(true) + for _, task := range d.tasks { + d.Pause(task.ID) + } + return d.storage.Close() +} + +func (d *Downloader) Clear() error { + if err := d.Close(); err != nil { + return err + } + if err := d.storage.Clear(); err != nil { + return err + } + return nil } func (d *Downloader) Listener(fn Listener) { @@ -216,7 +380,75 @@ func (d *Downloader) GetTasks() []*Task { return d.tasks } -var defaultDownloader = NewDownloader(new(http.FetcherBuilder), new(bt.FetcherBuilder)) +func (d *Downloader) GetConfig(v any) error { + return d.storage.Get(bucketConfig, "config", v) +} + +func (d *Downloader) PutConfig(v any) error { + return d.storage.Put(bucketConfig, "config", v) +} + +func (d *Downloader) watch(task *Task) { + err := task.fetcher.Wait() + if err != nil { + task.Status = base.DownloadStatusError + d.emit(EventKeyError, task, err) + } else { + task.Progress.Used = task.timer.Used() + if task.Size == 0 { + task.Size = task.fetcher.Progress().TotalDownloaded() + task.Res.Size = task.Size + } + used := task.Progress.Used / int64(time.Second) + if used == 0 { + used = 1 + } + task.Progress.Speed = task.Size / used + task.Progress.Downloaded = task.Size + task.Status = base.DownloadStatusDone + d.storage.Put(bucketTask, task.ID, task.clone()) + d.emit(EventKeyDone, task) + } + d.emit(EventKeyFinally, task, err) +} + +func (d *Downloader) restoreFetcher(task *Task) error { + if task.fetcher == nil { + fetcherBuilder, _, err := d.buildFetcher(task.Res.Req.URL) + if err != nil { + return err + } + task.fetcherBuilder = fetcherBuilder + err = func() error { + v, f := fetcherBuilder.Restore() + if v != nil { + err := d.storage.Pop(bucketSave, task.ID, v) + if err != nil { + return err + } + task.fetcher = f(task.Res, task.Opts, v) + } + return nil + }() + if err != nil { + // TODO log + } + if task.fetcher == nil { + task.fetcher = fetcherBuilder.Build() + task.fetcher.Create(task.Res, task.Opts) + } + task.fetcher.Setup(d.controller) + go d.watch(task) + } + return nil +} + +func initTask(task *Task) { + task.timer = &util.Timer{} + task.lock = &sync.Mutex{} +} + +var defaultDownloader = NewDownloader(nil) type boot struct { url string @@ -256,5 +488,9 @@ func (b *boot) Create(opts *base.Options) (string, error) { } func Boot() *boot { + err := defaultDownloader.Setup() + if err != nil { + panic(err) + } return &boot{} } diff --git a/pkg/download/downloader_test.go b/pkg/download/downloader_test.go index 619e444b..f881693f 100644 --- a/pkg/download/downloader_test.go +++ b/pkg/download/downloader_test.go @@ -1,19 +1,23 @@ package download import ( - "github.com/monkeyWie/gopeed-core/internal/protocol/http" + "github.com/monkeyWie/gopeed-core/internal/test" "github.com/monkeyWie/gopeed-core/pkg/base" - "github.com/monkeyWie/gopeed-core/pkg/test" "reflect" "sync" "testing" + "time" ) func TestDownloader_Resolve(t *testing.T) { listener := test.StartTestFileServer() defer listener.Close() - downloader := NewDownloader(new(http.FetcherBuilder)) + downloader := NewDownloader(nil) + if err := downloader.Setup(); err != nil { + t.Fatal(err) + } + defer downloader.Clear() req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } @@ -22,6 +26,7 @@ func TestDownloader_Resolve(t *testing.T) { t.Fatal(err) } want := &base.Resource{ + Name: test.BuildName, Req: req, Size: test.BuildSize, Range: true, @@ -42,7 +47,11 @@ func TestDownloader_Create(t *testing.T) { listener := test.StartTestFileServer() defer listener.Close() - downloader := NewDownloader(new(http.FetcherBuilder)) + downloader := NewDownloader(nil) + if err := downloader.Setup(); err != nil { + t.Fatal(err) + } + defer downloader.Clear() req := &base.Request{ URL: "http://" + listener.Addr().String() + "/" + test.BuildName, } @@ -73,3 +82,69 @@ func TestDownloader_Create(t *testing.T) { t.Errorf("Download() got = %v, want %v", got, want) } } + +func TestDownloader_StoreAndRestore(t *testing.T) { + listener := test.StartTestSlowFileServer(time.Millisecond * 2000) + defer listener.Close() + + downloader := NewDownloader(&DownloaderConfig{ + Storage: NewBoltStorage("./"), + }) + if err := downloader.Setup(); err != nil { + t.Fatal(err) + } + defer downloader.Clear() + req := &base.Request{ + URL: "http://" + listener.Addr().String() + "/" + test.BuildName, + } + res, err := downloader.Resolve(req) + if err != nil { + t.Fatal(err) + } + + id, err := downloader.Create(res, &base.Options{ + Path: test.Dir, + Name: test.DownloadName, + Connections: 4, + }) + if err != nil { + t.Fatal(err) + } + time.Sleep(time.Millisecond * 1001) + err = downloader.Pause(id) + if err != nil { + t.Fatal(err) + } + downloader.Close() + + downloader = NewDownloader(&DownloaderConfig{ + Storage: NewBoltStorage("./"), + }) + if err := downloader.Setup(); err != nil { + t.Fatal(err) + } + task := downloader.GetTask(id) + + if task == nil { + t.Fatal("task is nil") + } + var wg sync.WaitGroup + wg.Add(1) + downloader.Listener(func(event *Event) { + if event.Key == EventKeyDone { + wg.Done() + } + }) + err = downloader.Continue(id) + wg.Wait() + if err != nil { + t.Fatal(err) + } + want := test.FileMd5(test.BuildFile) + got := test.FileMd5(test.DownloadFile) + if want != got { + t.Errorf("StoreAndResume() got = %v, want %v", got, want) + } + + downloader.Clear() +} diff --git a/pkg/download/event.go b/pkg/download/event.go index c5a3426d..87a4361d 100644 --- a/pkg/download/event.go +++ b/pkg/download/event.go @@ -8,6 +8,7 @@ const ( EventKeyContinue = "continue" EventKeyProgress = "progress" EventKeyError = "error" + EventKeyDelete = "delete" EventKeyDone = "done" EventKeyFinally = "finally" ) diff --git a/pkg/download/storage.go b/pkg/download/storage.go new file mode 100644 index 00000000..71692398 --- /dev/null +++ b/pkg/download/storage.go @@ -0,0 +1,219 @@ +package download + +import ( + "encoding/json" + "go.etcd.io/bbolt" + "os" + "path/filepath" + "reflect" +) + +type Storage interface { + Setup(buckets []string) error + Put(bucket string, key string, v any) error + Get(bucket string, key string, v any) error + List(bucket string, v any) error + Pop(bucket string, key string, v any) error + Delete(bucket string, key string) error + + Close() error + Clear() error +} + +func changeValue(p any, v any) { + if v == nil { + return + } + rp := reflect.ValueOf(p) + rv := reflect.ValueOf(v) + if rv.Kind() == reflect.Slice { + if rv.Len() == 0 { + return + } + // get underlying type + tp := reflect.TypeOf(p).Elem().Elem() + for i := 0; i < rv.Len(); i++ { + // convert to underlying type + vv := rv.Index(i).Elem().Convert(tp) + rp.Elem().Set(reflect.Append(rp.Elem(), vv)) + } + } else if rv.Kind() == reflect.Ptr { + rp.Elem().Set(rv.Elem()) + } else { + rp.Elem().Set(rv) + } +} + +var memData = make(map[string]map[string]any) + +type MemStorage struct { +} + +func NewMemStorage() *MemStorage { + return &MemStorage{} +} + +func (n *MemStorage) Setup(buckets []string) error { + for _, bucket := range buckets { + if _, ok := memData[bucket]; !ok { + memData[bucket] = make(map[string]any) + } + } + return nil +} + +func (n *MemStorage) Put(bucket string, key string, v any) error { + memData[bucket][key] = v + return nil +} + +func (n *MemStorage) Get(bucket string, key string, v any) error { + changeValue(v, memData[bucket][key]) + return nil +} + +func (n *MemStorage) List(bucket string, v any) error { + data := memData[bucket] + list := make([]any, 0) + for _, v := range data { + list = append(list, v) + } + changeValue(v, list) + return nil +} + +func (n *MemStorage) Pop(bucket string, key string, v any) error { + data := memData[bucket] + changeValue(v, data[key]) + delete(data, key) + return nil +} + +func (n *MemStorage) Delete(bucket string, key string) error { + delete(memData[bucket], key) + return nil +} + +func (n *MemStorage) Close() error { + return nil +} + +func (n *MemStorage) Clear() error { + memData = make(map[string]map[string]any) + return nil +} + +const ( + dbFile = "gopeed.db" +) + +type BoltStorage struct { + db *bbolt.DB + path string +} + +func NewBoltStorage(dir string) *BoltStorage { + path := filepath.Join(dir, dbFile) + db, err := bbolt.Open(path, 0600, nil) + if err != nil { + panic(err) + } + return &BoltStorage{ + db: db, + path: path, + } +} + +func (b *BoltStorage) Setup(buckets []string) error { + return b.db.Update(func(tx *bbolt.Tx) error { + for _, bucket := range buckets { + _, err := tx.CreateBucketIfNotExists([]byte(bucket)) + if err != nil { + return err + } + } + return nil + }) +} + +func (b *BoltStorage) Put(bucket string, key string, v any) error { + buf, err := json.Marshal(v) + if err != nil { + return err + } + return b.db.Update(func(tx *bbolt.Tx) error { + b := tx.Bucket([]byte(bucket)) + return b.Put([]byte(key), buf) + }) +} + +func (b *BoltStorage) Get(bucket string, key string, v any) error { + var data []byte + err := b.db.View(func(tx *bbolt.Tx) error { + b := tx.Bucket([]byte(bucket)) + data = b.Get([]byte(key)) + return nil + }) + if err != nil { + return err + } + return json.Unmarshal(data, v) +} + +func (b *BoltStorage) List(bucket string, v any) error { + list := make([]any, 0) + tv := reflect.TypeOf(v).Elem().Elem() + if err := b.db.View(func(tx *bbolt.Tx) error { + b := tx.Bucket([]byte(bucket)) + if err := b.ForEach(func(k, v []byte) error { + data := reflect.New(tv.Elem()).Interface() + if err := json.Unmarshal(v, &data); err != nil { + return err + } + list = append(list, data) + return nil + }); err != nil { + return err + } + return nil + }); err != nil { + return err + } + changeValue(v, list) + return nil +} + +func (b *BoltStorage) Pop(bucket string, key string, v any) error { + var data []byte + err := b.db.Update(func(tx *bbolt.Tx) error { + b := tx.Bucket([]byte(bucket)) + kb := []byte(key) + data = b.Get(kb) + return b.Delete(kb) + }) + if err != nil { + return err + } + return json.Unmarshal(data, v) +} + +func (b *BoltStorage) Delete(bucket string, key string) error { + return b.db.Update(func(tx *bbolt.Tx) error { + b := tx.Bucket([]byte(bucket)) + return b.Delete([]byte(key)) + }) +} + +func (b *BoltStorage) Close() error { + return b.db.Close() +} + +func (b *BoltStorage) Clear() error { + if err := b.Close(); err != nil { + return err + } + if err := os.Remove(b.path); err != nil { + return err + } + return nil +} diff --git a/pkg/download/task.go b/pkg/download/task.go index 44082922..2178586f 100644 --- a/pkg/download/task.go +++ b/pkg/download/task.go @@ -1,7 +1,7 @@ package download import ( - "github.com/google/uuid" + gonanoid "github.com/matoous/go-nanoid/v2" "github.com/monkeyWie/gopeed-core/internal/fetcher" "github.com/monkeyWie/gopeed-core/pkg/base" "github.com/monkeyWie/gopeed-core/pkg/util" @@ -10,23 +10,40 @@ import ( ) type Task struct { - 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"` + ID string `json:"id"` + Res *base.Resource `json:"res"` + Opts *base.Options `json:"opts"` + Status base.Status `json:"status"` + Progress *Progress `json:"progress"` + Size int64 `json:"size"` + CreatedAt time.Time `json:"createdAt"` - fetcher fetcher.Fetcher - timer *util.Timer - locker *sync.Mutex + fetcherBuilder fetcher.FetcherBuilder + fetcher fetcher.Fetcher + timer *util.Timer + lock *sync.Mutex } func NewTask() *Task { + id, err := gonanoid.New() + if err != nil { + panic(err) + } return &Task{ - ID: uuid.New().String(), - Status: base.DownloadStatusReady, - CreateTime: time.Now(), + ID: id, + Status: base.DownloadStatusReady, + CreatedAt: time.Now(), + } +} + +func (t *Task) clone() *Task { + return &Task{ + ID: t.ID, + Res: t.Res, + Opts: t.Opts, + Status: t.Status, + Progress: t.Progress, + Size: t.Size, + CreatedAt: t.CreatedAt, } } diff --git a/pkg/rest/api.go b/pkg/rest/api.go index c237d9d1..ad8b2437 100644 --- a/pkg/rest/api.go +++ b/pkg/rest/api.go @@ -15,7 +15,7 @@ func Resolve(w http.ResponseWriter, r *http.Request) { if util.ReadJson(w, r, &req) { resource, err := Downloader.Resolve(&req) if err != nil { - util.WriteJson(w, 500, model.NewResultWithMsg(err.Error())) + util.WriteJson(w, http.StatusInternalServerError, model.NewResultWithMsg(err.Error())) return } util.WriteJsonOk(w, model.NewResultWithData(resource)) @@ -23,11 +23,17 @@ func Resolve(w http.ResponseWriter, r *http.Request) { } func CreateTask(w http.ResponseWriter, r *http.Request) { - var req model.CreateTaskReq + var req model.CreateTask if util.ReadJson(w, r, &req) { - taskId, err := Downloader.Create(req.Resource, req.Options) + if req.Opts.Connections == 0 { + req.Opts.Connections = getServerConfig().Connections + } + if req.Opts.Path == "" { + req.Opts.Path = getServerConfig().DownloadDir + } + taskId, err := Downloader.Create(req.Res, req.Opts) if err != nil { - util.WriteJson(w, 500, model.NewResultWithMsg(err.Error())) + util.WriteJson(w, http.StatusInternalServerError, model.NewResultWithMsg(err.Error())) return } util.WriteJsonOk(w, model.NewResultWithData(taskId)) @@ -37,14 +43,31 @@ func CreateTask(w http.ResponseWriter, r *http.Request) { func PauseTask(w http.ResponseWriter, r *http.Request) { vars := mux.Vars(r) taskId := vars["id"] - Downloader.Pause(taskId) + if err := Downloader.Pause(taskId); err != nil { + util.WriteJson(w, http.StatusInternalServerError, model.NewResultWithMsg(err.Error())) + return + } util.WriteJsonOk(w, nil) } func ContinueTask(w http.ResponseWriter, r *http.Request) { vars := mux.Vars(r) taskId := vars["id"] - Downloader.Continue(taskId) + if err := Downloader.Continue(taskId); err != nil { + util.WriteJson(w, http.StatusInternalServerError, model.NewResultWithMsg(err.Error())) + return + } + util.WriteJsonOk(w, nil) +} + +func DeleteTask(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + taskId := vars["id"] + force := r.FormValue("force") + if err := Downloader.Delete(taskId, force == "true"); err != nil { + util.WriteJson(w, http.StatusInternalServerError, model.NewResultWithMsg(err.Error())) + return + } util.WriteJsonOk(w, nil) } @@ -53,7 +76,7 @@ func GetTask(w http.ResponseWriter, r *http.Request) { taskId := vars["id"] task := Downloader.GetTask(taskId) if task == nil { - util.WriteJson(w, 404, model.NewResultWithMsg("task not found")) + util.WriteJson(w, http.StatusNotFound, model.NewResultWithMsg("task not found")) return } util.WriteJsonOk(w, model.NewResultWithData(task)) @@ -62,12 +85,12 @@ func GetTask(w http.ResponseWriter, r *http.Request) { func GetTasks(w http.ResponseWriter, r *http.Request) { status := r.FormValue("status") if status == "" { - util.WriteJson(w, 400, model.NewResultWithMsg("param is required: status")) + util.WriteJson(w, http.StatusBadRequest, model.NewResultWithMsg("param is required: status")) return } statusArr := strings.Split(status, ",") tasks := Downloader.GetTasks() - var result []*download.Task + result := make([]*download.Task, 0) for _, task := range tasks { for _, s := range statusArr { if task.Status == base.Status(s) { @@ -75,5 +98,26 @@ func GetTasks(w http.ResponseWriter, r *http.Request) { } } } - util.WriteJsonOk(w, model.NewResultWithData(tasks)) + util.WriteJsonOk(w, model.NewResultWithData(result)) +} + +func GetConfig(w http.ResponseWriter, r *http.Request) { + util.WriteJsonOk(w, model.NewResultWithData(getServerConfig())) +} + +func PutConfig(w http.ResponseWriter, r *http.Request) { + var cfg model.ServerConfig + if util.ReadJson(w, r, &cfg) { + if err := Downloader.PutConfig(&cfg); err != nil { + util.WriteJson(w, http.StatusInternalServerError, model.NewResultWithMsg(err.Error())) + return + } + } + util.WriteJsonOk(w, nil) +} + +func getServerConfig() *model.ServerConfig { + var cfg model.ServerConfig + Downloader.GetConfig(&cfg) + return &cfg } diff --git a/pkg/rest/model/server.go b/pkg/rest/model/server.go new file mode 100644 index 00000000..924f9100 --- /dev/null +++ b/pkg/rest/model/server.go @@ -0,0 +1,53 @@ +package model + +type Storage string + +const ( + StorageMem Storage = "mem" + StorageBolt Storage = "bolt" +) + +type StartConfig struct { + Network string `json:"network"` + Address string `json:"address"` + Storage Storage `json:"storage"` + StorageDir string `json:"storageDir"` + RefreshInterval int `json:"refreshInterval"` +} + +func (cfg *StartConfig) Init() *StartConfig { + if cfg.Network == "" { + cfg.Network = "tcp" + } + if cfg.Storage == "" { + cfg.Storage = StorageBolt + } + if cfg.StorageDir == "" { + cfg.StorageDir = "./" + } + return cfg +} + +// ServerConfig is present in the database +type ServerConfig struct { + Host string `json:"host"` + Port int `json:"port"` + + Connections int `json:"connections"` + DownloadDir string `json:"downloadDir"` + + Extra map[string]any `json:"extra"` +} + +func (cfg *ServerConfig) Init() *ServerConfig { + if cfg.Host == "" { + cfg.Host = "127.0.0.1" + } + if cfg.Port < 0 { + cfg.Port = 0 + } + if cfg.Connections <= 0 { + cfg.Connections = 16 + } + return cfg +} diff --git a/pkg/rest/model/task.go b/pkg/rest/model/task.go index 271e0fdf..e43548df 100644 --- a/pkg/rest/model/task.go +++ b/pkg/rest/model/task.go @@ -2,7 +2,7 @@ package model import "github.com/monkeyWie/gopeed-core/pkg/base" -type CreateTaskReq struct { - Resource *base.Resource `json:"resource"` - Options *base.Options `json:"options"` +type CreateTask struct { + Res *base.Resource `json:"res"` + Opts *base.Options `json:"opts"` } diff --git a/pkg/rest/server.go b/pkg/rest/server.go index 21027e8f..d242c023 100644 --- a/pkg/rest/server.go +++ b/pkg/rest/server.go @@ -2,23 +2,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" + "github.com/monkeyWie/gopeed-core/pkg/rest/model" + "github.com/monkeyWie/gopeed-core/pkg/util" "net" "net/http" + "strconv" ) var ( srv *http.Server - Downloader = download.NewDownloader(new(fhttp.FetcherBuilder), new(bt.FetcherBuilder)) + Downloader *download.Downloader ) -func Start(addr string, port int) (int, error) { - listener, err := net.Listen("tcp", fmt.Sprintf("%s:%d", addr, port)) +func Start(startCfg *model.StartConfig) (int, error) { + if startCfg == nil { + startCfg = &model.StartConfig{} + } + startCfg.Init() + + downloadCfg := &download.DownloaderConfig{} + if startCfg.Storage == model.StorageBolt { + downloadCfg.Storage = download.NewBoltStorage(startCfg.StorageDir) + } else { + downloadCfg.Storage = download.NewMemStorage() + } + downloadCfg.RefreshInterval = startCfg.RefreshInterval + Downloader = download.NewDownloader(downloadCfg.Init()) + if err := Downloader.Setup(); err != nil { + return 0, err + } + + serverConfig := getAndPutServerConfig(startCfg) + var address string + if startCfg.Network == "unix" { + address = startCfg.Address + util.SafeRemove(address) + } else { + address = net.JoinHostPort(serverConfig.Host, strconv.Itoa(serverConfig.Port)) + } + + listener, err := net.Listen(startCfg.Network, address) if err != nil { return 0, err } @@ -28,8 +54,11 @@ func Start(addr string, port int) (int, error) { 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.MethodDelete).Path("/api/v1/tasks/{id}").HandlerFunc(DeleteTask) r.Methods(http.MethodGet).Path("/api/v1/tasks/{id}").HandlerFunc(GetTask) r.Methods(http.MethodGet).Path("/api/v1/tasks").HandlerFunc(GetTasks) + r.Methods(http.MethodGet).Path("/api/v1/config").HandlerFunc(GetConfig) + r.Methods(http.MethodPut).Path("/api/v1/config").HandlerFunc(PutConfig) srv = &http.Server{Handler: r} go func() { @@ -38,12 +67,37 @@ func Start(addr string, port int) (int, error) { panic(err) } }() - return listener.Addr().(*net.TCPAddr).Port, nil + if addr, ok := listener.Addr().(*net.TCPAddr); ok { + return addr.Port, nil + } + return 0, nil } -func Stop() error { +func Stop() { if srv != nil { - return srv.Shutdown(context.TODO()) + if err := srv.Shutdown(context.TODO()); err != nil { + // TODO log + } + } + if Downloader != nil { + if err := Downloader.Close(); err != nil { + // TODO log + } } - return nil +} + +func getAndPutServerConfig(startCfg *model.StartConfig) *model.ServerConfig { + var serverConfig model.ServerConfig + err := Downloader.GetConfig(&serverConfig) + // first start + if err != nil { + if startCfg.Network == "tcp" { + host, port, _ := net.SplitHostPort(startCfg.Address) + serverConfig.Host = host + serverConfig.Port, _ = strconv.Atoi(port) + } + serverConfig.Init() + Downloader.PutConfig(serverConfig) + } + return &serverConfig } diff --git a/pkg/rest/server_test.go b/pkg/rest/server_test.go index 83903e7e..146dd4bc 100644 --- a/pkg/rest/server_test.go +++ b/pkg/rest/server_test.go @@ -3,17 +3,20 @@ package rest import ( "bytes" "encoding/json" + "errors" "fmt" + "github.com/monkeyWie/gopeed-core/internal/test" "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" + "os" "reflect" "sync" "testing" + "time" ) var ( @@ -21,6 +24,7 @@ var ( taskReq = &base.Request{} taskRes = &base.Resource{ + Name: test.BuildName, Req: taskReq, Size: test.BuildSize, Range: true, @@ -32,9 +36,9 @@ var ( }, }, } - createReq = &model.CreateTaskReq{ - Resource: taskRes, - Options: &base.Options{ + createReq = &model.CreateTask{ + Res: taskRes, + Opts: &base.Options{ Path: test.Dir, Name: test.DownloadName, Connections: 2, @@ -130,6 +134,28 @@ func TestPauseAndContinueTask(t *testing.T) { }) } +func TestDeleteTask(t *testing.T) { + doTest(func() { + taskId := httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + httpRequestCheckOk[any](http.MethodDelete, "/api/v1/tasks/"+taskId, nil) + code, _ := httpRequest[*download.Task](http.MethodGet, "/api/v1/tasks/"+taskId, nil) + checkCode(code, http.StatusNotFound) + }) +} + +func TestDeleteTaskForce(t *testing.T) { + doTest(func() { + taskId := httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + time.Sleep(time.Millisecond * 500) + httpRequestCheckOk[any](http.MethodDelete, "/api/v1/tasks/"+taskId+"?force=true", nil) + code, _ := httpRequest[*download.Task](http.MethodGet, "/api/v1/tasks/"+taskId, nil) + checkCode(code, http.StatusNotFound) + if _, err := os.Stat(test.DownloadFile); !errors.Is(err, os.ErrNotExist) { + t.Errorf("DeleteTaskForce() got = %v, want %v", err, os.ErrNotExist) + } + }) +} + func TestGetTasks(t *testing.T) { doTest(func() { var wg sync.WaitGroup @@ -140,30 +166,65 @@ func TestGetTasks(t *testing.T) { } }) - httpRequestCheckOk[string](http.MethodPost, "/api/v1/tasks", createReq) + httpRequestCheckOk[string](http.MethodPost, fmt.Sprintf("/api/v1/tasks?status=%s,%s", + base.DownloadStatusReady, base.DownloadStatusRunning), createReq) code, _ := httpRequest[[]*download.Task](http.MethodGet, "/api/v1/tasks", nil) - checkCode(code, 400) + checkCode(code, http.StatusBadRequest) wg.Wait() - r := httpRequestCheckOk[[]*download.Task](http.MethodGet, "/api/v1/tasks?status=done", nil) + r := httpRequestCheckOk[[]*download.Task](http.MethodGet, fmt.Sprintf("/api/v1/tasks?status=%s", + base.DownloadStatusDone), nil) if r[0].Status != base.DownloadStatusDone { t.Errorf("GetTasks() got = %v, want %v", r[0].Status, base.DownloadStatusDone) } + r = httpRequestCheckOk[[]*download.Task](http.MethodGet, fmt.Sprintf("/api/v1/tasks?status=%s,%s", + base.DownloadStatusReady, base.DownloadStatusRunning), nil) + if len(r) > 0 { + t.Errorf("GetTasks() got = %v, want %v", len(r), 0) + } + }) +} + +func TestGetAndPutConfig(t *testing.T) { + doTest(func() { + cfg := httpRequestCheckOk[*model.ServerConfig](http.MethodGet, "/api/v1/config", nil) + + cfg.Port = 8888 + cfg.Connections = 32 + cfg.DownloadDir = "./download" + cfg.Extra = map[string]any{ + "theme": "dark", + } + httpRequestCheckOk[any](http.MethodPut, "/api/v1/config", cfg) + + newCfg := httpRequestCheckOk[*model.ServerConfig](http.MethodGet, "/api/v1/config", nil) + if !reflect.DeepEqual(newCfg, cfg) { + t.Errorf("GetAndPutConfig() got = %v, want %v", newCfg, cfg) + } }) } func doTest(handler func()) { - fileListener := doStart() - defer func() { - fileListener.Close() - Stop() - }() - taskReq.URL = "http://" + fileListener.Addr().String() + "/" + test.BuildName - handler() + testFunc := func(storage model.Storage) { + fileListener := doStart(storage) + defer func() { + if err := fileListener.Close(); err != nil { + panic(err) + } + Stop() + }() + defer Downloader.Clear() + taskReq.URL = "http://" + fileListener.Addr().String() + "/" + test.BuildName + handler() + } + testFunc(model.StorageMem) + testFunc(model.StorageBolt) } -func doStart() net.Listener { - port, err := Start("127.0.0.1", 0) +func doStart(storage model.Storage) net.Listener { + port, err := Start((&model.StartConfig{ + Storage: storage, + }).Init()) if err != nil { panic(err) } diff --git a/pkg/util/path.go b/pkg/util/path.go new file mode 100644 index 00000000..da19002e --- /dev/null +++ b/pkg/util/path.go @@ -0,0 +1,81 @@ +package util + +import ( + "errors" + "io" + "os" + "path/filepath" +) + +func Dir(path string) string { + dir := filepath.Dir(path) + if dir == "." { + return "" + } + return dir +} + +func Filepath(path string, originName string, customName string) string { + if customName == "" { + customName = originName + } + return filepath.Join(path, customName) +} + +// SafeRemove remove file safely, ignoring errors if the path does not exist. +func SafeRemove(name string) error { + if err := os.Remove(name); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + return nil +} + +// SafeRemoveAll remove file and parent directories safely +func SafeRemoveAll(path string, names []string) error { + for _, name := range names { + err := SafeRemove(filepath.Join(path, name)) + if err != nil { + return err + } + if err := safeRemoveParent(path, name); err != nil { + return err + } + } + return nil +} + +func safeRemoveParent(path string, subPath string) error { + currPath := filepath.Dir(subPath) + if currPath == "." { + return nil + } + // if directory is empty, remove it + dir := filepath.Join(path, filepath.Dir(subPath)) + empty, err := isEmpty(dir) + if err != nil { + return err + } + if empty { + err = SafeRemove(dir) + if err != nil { + return err + } + return safeRemoveParent(path, currPath) + } + return nil +} + +// check directory is empty +func isEmpty(name string) (bool, error) { + f, err := os.Open(name) + if err != nil { + return false, err + } + defer f.Close() + + _, err = f.Readdirnames(1) // Or f.Readdir(1) + if err == io.EOF { + return true, nil + } + return false, err // Either not empty or error, suits both cases +} diff --git a/pkg/util/path_test.go b/pkg/util/path_test.go new file mode 100644 index 00000000..b8224608 --- /dev/null +++ b/pkg/util/path_test.go @@ -0,0 +1,198 @@ +package util + +import ( + "os" + "path/filepath" + "testing" +) + +func TestDir(t *testing.T) { + type args struct { + path string + } + tests := []struct { + name string + args args + want string + }{ + { + name: "empty path", + args: args{ + path: ".", + }, + want: "", + }, + { + name: "normal path case 1", + args: args{ + path: "./a/b/c/1.txt", + }, + want: "a/b/c", + }, + { + name: "normal path case 2", + args: args{ + path: "a/b/c/1.txt", + }, + want: "a/b/c", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := Dir(tt.args.path); got != tt.want { + t.Errorf("Dir() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestFilepath(t *testing.T) { + type args struct { + path string + originName string + customName string + } + tests := []struct { + name string + args args + want string + }{ + { + name: "origin name", + args: args{ + path: "/Downloads", + originName: "1.txt", + customName: "", + }, + want: "/Downloads/1.txt", + }, + { + name: "origin name", + args: args{ + path: "/Downloads", + originName: "1.txt", + customName: "2.txt", + }, + want: "/Downloads/2.txt", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := Filepath(tt.args.path, tt.args.originName, tt.args.customName); got != tt.want { + t.Errorf("Filepath() = %v, want %v", got, tt.want) + } + }) + } +} + +func TestSafeRemove(t *testing.T) { + name := "test_safe_remove.data" + _, err := os.Create(name) + if err != nil { + t.Fatal(err) + } + if err := SafeRemove(name); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(name); err == nil { + t.Fatal(err) + } + + if err := SafeRemove("test_safe_remove_not_exist.data"); err != nil { + t.Fatal(err) + } +} + +func TestSafeRemoveAll(t *testing.T) { + testDir := "test_dir" + err := os.Mkdir(testDir, 0755) + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(testDir) + + doSafeRemoveAll(t, testDir, []string{ + "1.txt", + }, []string{}) + + doSafeRemoveAll(t, testDir, []string{ + "1.txt", + }, []string{ + "2.txt", + }, "2.txt") + + doSafeRemoveAll(t, testDir, []string{ + "1.txt", + "a/b/c/1.txt", + "a/b/1.txt", + "a/1.txt", + }, []string{}) + + doSafeRemoveAll(t, testDir, []string{ + "1.txt", + "a/b/c/1.txt", + "a/b/1.txt", + "a/1.txt", + }, []string{ + "a/b/2.txt", + }, "a/b/2.txt") + + doSafeRemoveAll(t, testDir, []string{ + "1.txt", + "a/b/c/1.txt", + "a/b/1.txt", + "a/1.txt", + }, []string{ + "a/2.txt", + }, "a/2.txt") +} + +func doSafeRemoveAll(t *testing.T, path string, downloadNames []string, otherNames []string, exist ...string) { + preCreate(t, path, downloadNames) + preCreate(t, path, otherNames) + + if err := SafeRemoveAll(path, downloadNames); err != nil { + t.Fatal(err) + } + + if len(exist) == 0 { + for _, name := range downloadNames { + filePath := filepath.Join(path, name) + if isExist(filePath) { + t.Fatalf("file %s should not exist", filePath) + } + subDirPath := filepath.Dir(name) + if subDirPath != "." { + dirPath := filepath.Join(path, subDirPath) + if isExist(dirPath) { + t.Fatalf("dir %s should not exist", dirPath) + } + } + } + return + } + + for _, name := range exist { + if !isExist(filepath.Join(path, name)) { + t.Fatalf("file %s should exist", name) + } + } +} + +func preCreate(t *testing.T, path string, names []string) { + for _, name := range names { + fullPath := filepath.Join(path, name) + if err := os.MkdirAll(filepath.Dir(fullPath), 0755); err != nil { + t.Fatal(err) + } + _, err := os.Create(fullPath) + if err != nil { + t.Fatal(err) + } + } +} + +func isExist(path string) bool { + _, err := os.Stat(path) + return err == nil || os.IsExist(err) +} diff --git a/pkg/util/timer.go b/pkg/util/timer.go index 017029d5..0a12f6d4 100644 --- a/pkg/util/timer.go +++ b/pkg/util/timer.go @@ -16,10 +16,6 @@ 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 }