nprasad2077/gopeed
1
0 Fork 0
mirror of https://github.com/GopeedLab/gopeed.git synced 2026-11-15 04:43:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: integration torrent (#30)

Browse Source
This commit is contained in:
monkeyWie authored and GitHub committed 2022-08-11 14:38:16 +08:00
1 parent 43378c7dc8
commit 034137322f
20 files changed
+925 -139
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/protocol/bt/fetcher.go
+115
View File
@@ -0,0 +1,115 @@
package bt
import (
"github.com/anacrolix/torrent"
"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"
)
var client *torrent.Client
type Fetcher struct {
Ctl controller.Controller
torrent *torrent.Torrent
opts *base.Options
}
func (f *Fetcher) Setup(ctl controller.Controller) (err error) {
f.Ctl = ctl
var once sync.Once
once.Do(func() {
client, err = torrent.NewClient(nil)
})
return
}
func (f *Fetcher) Wait() (err error) {
client.WaitAll()
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 {
return
}
<-f.torrent.GotInfo()
res = &base.Resource{
Req: req,
Length: f.torrent.Length(),
Range: true,
Files: make([]*base.FileInfo, len(f.torrent.Files())),
}
for i, file := range f.torrent.Files() {
res.Files[i] = &base.FileInfo{
Name: filepath.Base(file.DisplayPath()),
Path: filepath.Dir(file.Path()),
Size: file.Length(),
}
}
return
}
func (f *Fetcher) Create(res *base.Resource, opts *base.Options) (err error) {
f.opts = opts
if len(f.opts.SelectFiles) == 0 {
f.opts.SelectFiles = make([]int, len(res.Files))
for i := range f.opts.SelectFiles {
f.opts.SelectFiles[i] = i
}
}
return nil
}
func (f *Fetcher) Start() (err error) {
if len(f.opts.SelectFiles) == 0 {
f.torrent.DownloadAll()
} else {
for _, index := range f.opts.SelectFiles {
f.torrent.Files()[index].Download()
}
}
return
}
func (f *Fetcher) Pause() (err error) {
f.torrent.DisallowDataDownload()
return
}
func (f *Fetcher) Continue() (err error) {
f.torrent.AllowDataDownload()
return
}
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()
}
return p
}
type FetcherBuilder struct {
}
var schemes = []string{"FILE", "MAGNET"}
func (fb *FetcherBuilder) Schemes() []string {
return schemes
}
func (fb *FetcherBuilder) Build() fetcher.Fetcher {
return &Fetcher{}
}
internal/protocol/bt/fetcher_test.go
+56
View File
@@ -0,0 +1,56 @@
package bt
import (
"fmt"
"github.com/monkeyWie/gopeed-core/internal/controller"
"github.com/monkeyWie/gopeed-core/internal/fetcher"
"github.com/monkeyWie/gopeed-core/pkg/base"
"testing"
)
func TestFetcher_Resolve_Magnet(t *testing.T) {
fetcher := buildFetcher()
res, err := fetcher.Resolve(&base.Request{
URL: "magnet:?xt=urn:btih:A213E40628B27B58E70F0DADAC5350CA05403AC4&dn=Microsoft%20Office%20PRO%20Plus%202021%20Retail%20Version%202108%20Build%2014326.2&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969%2Fannounce&tr=udp%3A%2F%2Ftracker.openbittorrent.com%3A6969%2Fannounce&tr=udp%3A%2F%2F9.rarbg.to%3A2710%2Fannounce&tr=udp%3A%2F%2F9.rarbg.me%3A2780%2Fannounce&tr=udp%3A%2F%2F9.rarbg.to%3A2730%2Fannounce&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=http%3A%2F%2Fp4p.arenabg.com%3A1337%2Fannounce&tr=udp%3A%2F%2Ftracker.torrent.eu.org%3A451%2Fannounce&tr=udp%3A%2F%2Ftracker.tiny-vps.com%3A6969%2Fannounce&tr=udp%3A%2F%2Fopen.stealth.si%3A80%2Fannounce",
})
if err != nil {
panic(err)
}
fmt.Println(res)
}
func TestFetcher_Resolve_Torrent(t *testing.T) {
fetcher := buildFetcher()
res, err := fetcher.Resolve(&base.Request{
URL: "d:/test/test.torrent",
})
if err != nil {
panic(err)
}
fmt.Println(res)
}
func TestFetcher_Create(t *testing.T) {
fetcher := buildFetcher()
res, err := fetcher.Resolve(&base.Request{
URL: "magnet:?xt=urn:btih:A213E40628B27B58E70F0DADAC5350CA05403AC4&dn=Microsoft%20Office%20PRO%20Plus%202021%20Retail%20Version%202108%20Build%2014326.2&tr=udp%3A%2F%2Ftracker.coppersurfer.tk%3A6969%2Fannounce&tr=udp%3A%2F%2Ftracker.openbittorrent.com%3A6969%2Fannounce&tr=udp%3A%2F%2F9.rarbg.to%3A2710%2Fannounce&tr=udp%3A%2F%2F9.rarbg.me%3A2780%2Fannounce&tr=udp%3A%2F%2F9.rarbg.to%3A2730%2Fannounce&tr=udp%3A%2F%2Ftracker.opentrackr.org%3A1337&tr=http%3A%2F%2Fp4p.arenabg.com%3A1337%2Fannounce&tr=udp%3A%2F%2Ftracker.torrent.eu.org%3A451%2Fannounce&tr=udp%3A%2F%2Ftracker.tiny-vps.com%3A6969%2Fannounce&tr=udp%3A%2F%2Fopen.stealth.si%3A80%2Fannounce",
})
if err != nil {
panic(err)
}
if err := fetcher.Create(res, &base.Options{}); err != nil {
panic(err)
}
if err := fetcher.Start(); err != nil {
panic(err)
}
if err := fetcher.Wait(); err != nil {
panic(err)
}
}
func buildFetcher() fetcher.Fetcher {
fetcher := new(FetcherBuilder).Build()
fetcher.Setup(controller.NewController())
return fetcher
}
internal/protocol/http/fetcher.go
+45 -23
View File
@@ -39,29 +39,14 @@ type Fetcher struct {
res *base.Resource
opts *base.Options
status base.Status
clients []*http.Response
chunks []*Chunk
clients []*http.Response
ctx context.Context
cancel context.CancelFunc
pauseCh chan interface{}
}
func NewFetcher() *Fetcher {
return &Fetcher{
DefaultFetcher: new(fetcher.DefaultFetcher),
pauseCh: make(chan interface{}),
}
}
var protocols = []string{"HTTP", "HTTPS"}
func FetcherBuilder() ([]string, func() fetcher.Fetcher) {
return protocols, func() fetcher.Fetcher {
return NewFetcher()
}
}
func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) {
httpReq, err := buildRequest(nil, req)
if err != nil {
@@ -91,7 +76,7 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) {
if err != nil {
return nil, err
}
res.TotalSize = parse
res.Length = parse
}
} else if base.HttpCodeOK == httpResp.StatusCode {
// 返回200响应码，不支持断点下载，通过Content-Length头获取文件大小，获取不到的话可能是chunked编码
@@ -101,13 +86,13 @@ func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) {
if err != nil {
return nil, err
}
res.TotalSize = parse
res.Length = parse
}
} else {
return nil, NewRequestError(httpResp.StatusCode, httpResp.Status)
}
file := &base.FileInfo{
Size: res.TotalSize,
Size: res.Length,
}
contentDisposition := httpResp.Header.Get(base.HttpHeaderContentDisposition)
if contentDisposition != "" {
@@ -139,14 +124,14 @@ func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error {
func (f *Fetcher) Start() (err error) {
// 创建文件
name := f.filename()
_, err = f.Ctl.Touch(name, f.res.TotalSize)
_, err = f.Ctl.Touch(name, f.res.Length)
if err != nil {
return err
}
f.status = base.DownloadStatusStart
if f.res.Range {
// 每个连接平均需要下载的分块大小
chunkSize := f.res.TotalSize / int64(f.opts.Connections)
chunkSize := f.res.Length / int64(f.opts.Connections)
f.chunks = make([]*Chunk, f.opts.Connections)
f.clients = make([]*http.Response, f.opts.Connections)
for i := 0; i < f.opts.Connections; i++ {
@@ -156,7 +141,7 @@ func (f *Fetcher) Start() (err error) {
)
if i == f.opts.Connections-1 {
// 最后一个分块需要保证把文件下载完
end = f.res.TotalSize - 1
end = f.res.Length - 1
} else {
end = begin + chunkSize - 1
}
@@ -340,7 +325,7 @@ func buildClient() *http.Client {
jar, _ := cookiejar.New(nil)
return &http.Client{
Jar: jar,
Timeout: time.Second * 10,
Timeout: time.Second * 60,
}
}
@@ -385,3 +370,40 @@ func buildRequest(ctx context.Context, req *base.Request) (httpReq *http.Request
httpReq.Header = headers
return httpReq, nil
}
type FetcherData struct {
Chunks []*Chunk
}
type FetcherBuilder struct {
}
var schemes = []string{"HTTP", "HTTPS"}
func (fb *FetcherBuilder) Schemes() []string {
return schemes
}
func (fb *FetcherBuilder) Build() fetcher.Fetcher {
return &Fetcher{
DefaultFetcher: new(fetcher.DefaultFetcher),
pauseCh: make(chan interface{}),
}
}
func (fb *FetcherBuilder) Store(f fetcher.Fetcher) interface{} {
_f := f.(*Fetcher)
return &FetcherData{
Chunks: _f.chunks,
}
}
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
}
internal/protocol/http/fetcher_test.go
+58 -15
View File
@@ -13,8 +13,8 @@ import (
func TestFetcher_Resolve(t *testing.T) {
testResolve(test.StartTestFileServer, &base.Resource{
TotalSize: test.BuildSize,
Range: true,
Length: test.BuildSize,
Range: true,
Files: []*base.FileInfo{
{
Name: test.BuildName,
@@ -23,8 +23,8 @@ func TestFetcher_Resolve(t *testing.T) {
},
}, t)
testResolve(test.StartTestChunkedServer, &base.Resource{
TotalSize: 0,
Range: false,
Length: 0,
Range: false,
Files: []*base.FileInfo{
{
Name: test.BuildName,
@@ -37,7 +37,7 @@ func TestFetcher_Resolve(t *testing.T) {
func testResolve(startTestServer func() net.Listener, want *base.Resource, t *testing.T) {
listener := startTestServer()
defer listener.Close()
fetcher := NewFetcher()
fetcher := buildFetcher()
res, err := fetcher.Resolve(&base.Request{
URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
})
@@ -92,29 +92,38 @@ func TestFetcher_DownloadError(t *testing.T) {
downloadError(listener, 1, t)
}
func downloadReady(listener net.Listener, connections int, t *testing.T) fetcher.Fetcher {
fetcher := NewFetcher()
fetcher.Setup(controller.NewController())
func TestFetcher_DownloadResume(t *testing.T) {
listener := test.StartTestFileServer()
defer listener.Close()
downloadResume(listener, 1, t)
downloadResume(listener, 5, t)
downloadResume(listener, 8, t)
downloadResume(listener, 16, t)
}
func downloadReady(listener net.Listener, connections int, t *testing.T) (fetcher fetcher.Fetcher, res *base.Resource, opts *base.Options) {
fetcher = buildFetcher()
res, err := fetcher.Resolve(&base.Request{
URL: "http://" + listener.Addr().String() + "/" + test.BuildName,
})
if err != nil {
t.Fatal(err)
}
err = fetcher.Create(res, &base.Options{
opts = &base.Options{
Name: test.DownloadName,
Path: test.Dir,
Connections: connections,
})
}
err = fetcher.Create(res, opts)
if err != nil {
t.Fatal(err)
}
return fetcher
return
}
func downloadNormal(listener net.Listener, connections int, t *testing.T) {
fetcher := downloadReady(listener, connections, t)
fetcher, _, _ := downloadReady(listener, connections, t)
err := fetcher.Start()
if err != nil {
t.Fatal(err)
@@ -131,7 +140,7 @@ func downloadNormal(listener net.Listener, connections int, t *testing.T) {
}
func downloadContinue(listener net.Listener, connections int, t *testing.T) {
fetcher := downloadReady(listener, connections, t)
fetcher, _, _ := downloadReady(listener, connections, t)
err := fetcher.Start()
if err != nil {
t.Fatal(err)
@@ -156,7 +165,7 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) {
}
func downloadError(listener net.Listener, connections int, t *testing.T) {
fetcher := downloadReady(listener, connections, t)
fetcher, _, _ := downloadReady(listener, connections, t)
err := fetcher.Start()
if err != nil {
t.Fatal(err)
@@ -166,3 +175,37 @@ func downloadError(listener net.Listener, connections int, t *testing.T) {
t.Errorf("Download error = %v, want %v", err, nil)
}
}
func downloadResume(listener net.Listener, connections int, t *testing.T) {
fetcher, res, opts := downloadReady(listener, connections, t)
err := fetcher.Start()
if err != nil {
t.Fatal(err)
}
fb := new(FetcherBuilder)
time.Sleep(time.Second * 2)
data := fb.Store(fetcher)
time.Sleep(time.Second * 1)
fetcher.Pause()
fetcher = fb.Resume(res, opts, data)
fetcher.Setup(controller.NewController())
fetcher.Continue()
err = fetcher.Wait()
if err != nil {
t.Fatal(err)
}
want := test.FileMd5(test.BuildFile)
got := test.FileMd5(test.DownloadFile)
if want != got {
t.Errorf("Download error = %v, want %v", got, want)
}
}
func buildFetcher() *Fetcher {
fetcher := new(FetcherBuilder).Build()
fetcher.Setup(controller.NewController())
return fetcher.(*Fetcher)
}