nprasad2077/gopeed
1
0 Fork 0
mirror of https://github.com/GopeedLab/gopeed.git synced 2026-11-06 08:30:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

temp

Browse Source
This commit is contained in:
liwei committed 2021-03-05 19:52:55 +08:00
1 parent bdce3c84a3
commit 74d0e8e0bc
6 files changed
+102 -73
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

download/base/event.go
+2 -2
View File
@@ -1,5 +1,5 @@
package base
type Event interface {
Status() Status
type Event struct {
Status
}
download/base/fetcher.go
+5 -6
View File
@@ -4,13 +4,12 @@ package base
type Fetcher interface {
Init(ctl Controller)
// 解析请求
Resolve(req *Request) (*Resource, error)
Resolve(req *Request) (res *Resource, err error)
// 创建任务
Create(res *Resource, opts *Options) error
Start() error
Pause() error
Continue() error
Delete() error
Create(res *Resource, opts *Options) (err error)
Start() (err error)
Pause() (err error)
Continue() (err error)
}
type DefaultFetcher struct {
download/downloader.go
+38 -5
View File
@@ -6,6 +6,7 @@ import (
"github.com/monkeyWie/gopeed-core/download/base"
"net/url"
"strings"
"sync"
)
type TaskInfo struct {
@@ -16,14 +17,16 @@ type TaskInfo struct {
Files map[string]*base.FileInfo
fetcher base.Fetcher
locker *sync.Mutex
}
type FetcherBuilder func() (protocols []string, builder func() base.Fetcher)
type Downloader struct {
*base.DefaultController
fetchBuilders map[string]func() base.Fetcher
tasks map[string]*TaskInfo
*base.DefaultController
listener func(taskInfo *TaskInfo, status base.Status)
}
func NewDownloader(fbs ...FetcherBuilder) *Downloader {
@@ -60,14 +63,14 @@ func (d *Downloader) Resolve(req *base.Request) (*base.Resource, error) {
return fetcher.Resolve(req)
}
func (d *Downloader) Create(res *base.Resource, opts *base.Options) error {
func (d *Downloader) Create(res *base.Resource, opts *base.Options) (err error) {
fetcher, err := d.buildFetcher(res.Req.URL)
if err != nil {
return err
return
}
err = fetcher.Create(res, opts)
if err != nil {
return err
return
}
id := uuid.New().String()
d.tasks[id] = &TaskInfo{
@@ -76,10 +79,40 @@ func (d *Downloader) Create(res *base.Resource, opts *base.Options) error {
Opts: opts,
Status: base.DownloadStatusReady,
fetcher: fetcher,
locker: new(sync.Mutex),
}
return fetcher.Start()
go func() {
err = fetcher.Start()
if err != nil {
d.emit(d.tasks[id], base.DownloadStatusError)
}
}()
d.emit(d.tasks[id], base.DownloadStatusStart)
return
}
func (d *Downloader) Pause(id string) {
task := d.tasks[id]
task.locker.Lock()
defer task.locker.Unlock()
d.tasks[id].fetcher.Pause()
d.emit(d.tasks[id], base.DownloadStatusPause)
}
func (d *Downloader) Continue(id string) {
task := d.tasks[id]
task.locker.Lock()
defer task.locker.Unlock()
d.tasks[id].fetcher.Continue()
d.emit(d.tasks[id], base.DownloadStatusStart)
}
func (d *Downloader) Listener(fn func(taskInfo *TaskInfo, status base.Status)) {
d.listener = fn
}
func (d *Downloader) emit(taskInfo *TaskInfo, status base.Status) {
if d.listener != nil {
d.listener(taskInfo, status)
}
}
download/downloader_test.go
+3
View File
@@ -26,6 +26,9 @@ func TestDownloader_Create(t *testing.T) {
if err != nil {
t.Fatal(err)
}
downloader.Listener(func(taskInfo *TaskInfo, status base.Status) {
fmt.Printf("task:%s,status %d\n", taskInfo.ID, status)
})
err = downloader.Create(res, &base.Options{
Path: "E:\\test\\gopeed\\http",
Connections: 4,
download/http/fetcher.go
+53 -58
View File
@@ -14,7 +14,6 @@ import (
"path"
"path/filepath"
"strconv"
"sync"
"time"
)
@@ -40,11 +39,16 @@ type Fetcher struct {
clients []*http.Response
chunks []*model.Chunk
pauseCond *sync.Cond
pauseCh chan interface{}
doneCh chan error
}
func NewFetcher() *Fetcher {
return &Fetcher{}
return &Fetcher{
DefaultFetcher: new(base.DefaultFetcher),
pauseCh: make(chan interface{}),
doneCh: make(chan error),
}
}
var protocols = []string{"HTTP", "HTTPS"}
@@ -56,11 +60,11 @@ func FetcherBuilder() ([]string, func() base.Fetcher) {
}
func (f *Fetcher) Resolve(req *base.Request) (*base.Resource, error) {
httpReq, err := BuildRequest(req)
httpReq, err := buildRequest(req)
if err != nil {
return nil, err
}
client := BuildClient()
client := buildClient()
// 只访问一个字节，测试资源是否支持Range请求
httpReq.Header.Set(base.HttpHeaderRange, fmt.Sprintf(base.HttpHeaderRangeFormat, 0, 0))
httpResp, err := client.Do(httpReq)
@@ -129,18 +133,17 @@ func (f *Fetcher) Create(res *base.Resource, opts *base.Options) error {
f.res = res
f.opts = opts
f.status = base.DownloadStatusReady
f.pauseCond = sync.NewCond(&sync.Mutex{})
return nil
}
func (f *Fetcher) Start() error {
func (f *Fetcher) Start() (err error) {
// 创建文件
name := f.name()
_, err := f.Ctl.Touch(name, f.res.Size)
_, err = f.Ctl.Touch(name, f.res.Size)
if err != nil {
return err
}
defer f.close()
defer f.Ctl.Close(name)
f.status = base.DownloadStatusStart
if f.res.Range {
// 每个连接平均需要下载的分块大小
@@ -167,37 +170,28 @@ func (f *Fetcher) Start() error {
f.clients = make([]*http.Response, 1)
f.chunks[0] = model.NewChunk(0, 0)
}
return f.fetch()
}
func (f *Fetcher) Pause() error {
f.pauseCond.L.Lock()
defer f.pauseCond.L.Unlock()
func (f *Fetcher) Pause() (err error) {
if base.DownloadStatusStart != f.status {
return nil
return
}
f.status = base.DownloadStatusPause
f.stop()
// 释放锁，并等待下载结束
f.pauseCond.Wait()
return nil
<-f.pauseCh
f.Ctl.Close(f.name())
return
}
func (f *Fetcher) Continue() error {
if func() bool {
f.pauseCond.L.Lock()
defer f.pauseCond.L.Unlock()
if base.DownloadStatusPause != f.status && base.DownloadStatusError != f.status {
return true
}
f.status = base.DownloadStatusStart
return false
}() {
return nil
func (f *Fetcher) Continue() (err error) {
if base.DownloadStatusStart == f.status || base.DownloadStatusDone == f.status {
return
}
f.status = base.DownloadStatusStart
var name = f.name()
_, err := f.Ctl.Open(name)
_, err = f.Ctl.Open(name)
if err != nil {
return err
}
@@ -205,14 +199,6 @@ func (f *Fetcher) Continue() error {
return f.fetch()
}
func (f *Fetcher) Delete() error {
panic("implement me")
}
func (f *Fetcher) close() error {
return f.Ctl.Close(f.name())
}
func (f *Fetcher) name() string {
// 创建文件
var filename = f.opts.Name
@@ -222,7 +208,7 @@ func (f *Fetcher) name() string {
return filepath.Join(f.opts.Path, filename)
}
func (f *Fetcher) fetch() error {
func (f *Fetcher) fetch() (err error) {
errCh := make(chan error, f.opts.Connections)
defer close(errCh)
@@ -232,23 +218,34 @@ func (f *Fetcher) fetch() error {
}(i)
}
var retErr error
stopFlag := false
for i := 0; i < f.opts.Connections; i++ {
err := <-errCh
if err != nil && retErr == nil {
retErr = err
// 有一个连接失败就立即终止下载
if retErr != base.PauseErr {
func() {
f.pauseCond.L.Lock()
defer f.pauseCond.L.Unlock()
f.stop()
}()
fetchErr := <-errCh
if fetchErr != nil && !stopFlag {
// 确保如果有暂停操作返回暂停
if err == nil || err != base.PauseErr {
err = fetchErr
}
if fetchErr != base.PauseErr {
// 有一个连接失败就立即终止下载
stopFlag = true
f.stop()
}
}
}
f.pauseCond.Signal()
return retErr
if err != nil {
if err == base.PauseErr {
f.pauseCh <- nil
err = nil
} else {
f.status = base.DownloadStatusError
}
} else {
f.status = base.DownloadStatusDone
}
f.doneCh <- err
return
}
func (f *Fetcher) stop() {
@@ -262,12 +259,12 @@ func (f *Fetcher) stop() {
}
func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err error) {
httpReq, err := BuildRequest(f.res.Req)
httpReq, err := buildRequest(f.res.Req)
if err != nil {
return err
}
var (
client = BuildClient()
client = buildClient()
buf = make([]byte, 8192)
)
// 重试5次
@@ -283,8 +280,6 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er
chunk.Downloaded = 0
}
err = func() error {
f.pauseCond.L.Lock()
defer f.pauseCond.L.Unlock()
if f.status == base.DownloadStatusPause {
return base.PauseErr
}
@@ -326,7 +321,7 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er
}
return false, nil
}()
if err == nil || !retry {
if !retry {
// 下载成功，跳出重试
break
}
@@ -339,7 +334,7 @@ func (f *Fetcher) fetchChunk(index int, name string, chunk *model.Chunk) (err er
return
}
func BuildClient() *http.Client {
func buildClient() *http.Client {
// Cookie handle
jar, _ := cookiejar.New(nil)
return &http.Client{
@@ -348,7 +343,7 @@ func BuildClient() *http.Client {
}
}
func BuildRequest(req *base.Request) (*http.Request, error) {
func buildRequest(req *base.Request) (*http.Request, error) {
url, err := url.Parse(req.URL)
if err != nil {
return nil, err
download/http/fetcher_test.go
+1 -2
View File
@@ -265,11 +265,10 @@ func downloadContinue(listener net.Listener, connections int, t *testing.T) {
t.Fatal(err)
}
}()
time.Sleep(time.Millisecond * 200)
time.Sleep(time.Millisecond * 500)
if err := fetcher.Pause(); err != nil {
t.Fatal(err)
}
time.Sleep(time.Millisecond * 200)
if err := fetcher.Continue(); err != nil {
t.Fatal(err)
}