feat: torrent seed config (#599)

pkg/base/model.go

@@ -150,6 +150,9 @@ func (cfg *DownloaderStoreConfig) Init() *DownloaderStoreConfig {
 	if cfg.Proxy == nil {
 		cfg.Proxy = &DownloaderProxyConfig{}
 	}
+	if cfg.ProtocolConfig == nil {
+		cfg.ProtocolConfig = make(map[string]any)
+	}
 	return cfg
 }
 

pkg/download/downloader.go

@@ -42,12 +42,16 @@ var (
 type Listener func(event *Event)
 
 type Progress struct {
-	// 下载耗时(纳秒)
+	// Total download time(ns)
 	Used int64 `json:"used"`
-	// 每秒下载字节数
+	// Download speed(bytes/s)
 	Speed int64 `json:"speed"`
-	// 已下载的字节数
+	// Downloaded size(bytes)
 	Downloaded int64 `json:"downloaded"`
+	// Uploaded speed(bytes/s)
+	UploadSpeed int64 `json:"uploadSpeed"`
+	// Uploaded size(bytes)
+	Uploaded int64 `json:"uploaded"`
 }
 
 type Downloader struct {
@@ -124,11 +128,14 @@ func (d *Downloader) Setup() error {
 	}
 	// init default config
 	d.cfg.DownloaderStoreConfig.Init()
+	// init protocol config, if not exist, use default config
 	for _, fb := range d.fetcherBuilders {
-		f := fb.Build()
-		d.setupFetcher(f)
-
+		protocol := fb.Name()
+		if _, ok := d.cfg.DownloaderStoreConfig.ProtocolConfig[protocol]; !ok {
+			d.cfg.DownloaderStoreConfig.ProtocolConfig[protocol] = fb.DefaultConfig()
+		}
 	}
+
 	// load tasks from storage
 	var tasks []*Task
 	if err = d.storage.List(bucketTask, &tasks); err != nil {
@@ -166,6 +173,22 @@ func (d *Downloader) Setup() error {
 	}
 	d.extensions = extensions
 
+	// handle upload
+	go func() {
+		for _, task := range d.tasks {
+			if task.Status == base.DownloadStatusDone && task.Uploading {
+				if err := d.restoreFetcher(task); err != nil {
+					d.Logger.Error().Stack().Err(err).Msgf("task upload restore fetcher failed, task id: %s", task.ID)
+				}
+				if uploader, ok := task.fetcher.(fetcher.Uploader); ok {
+					if err := uploader.Upload(); err != nil {
+						d.Logger.Error().Stack().Err(err).Msgf("task upload failed, task id: %s", task.ID)
+					}
+				}
+			}
+		}
+	}()
+
 	// calculate download speed every tick
 	go func() {
 		for !d.closed.Load() {
@@ -174,18 +197,27 @@ func (d *Downloader) Setup() error {
 					func() {
 						task.statusLock.Lock()
 						defer task.statusLock.Unlock()
-						if task.Status != base.DownloadStatusRunning {
+						if task.Status != base.DownloadStatusRunning && !task.Uploading {
 							return
 						}
 						// check if task is deleted
-						if d.GetTask(task.ID) == nil {
+						if d.GetTask(task.ID) == nil || task.fetcher == nil {
 							return
 						}
 
 						current := task.fetcher.Progress().TotalDownloaded()
-						task.Progress.Used = task.timer.Used()
-						task.Progress.Speed = task.calcSpeed(current-task.Progress.Downloaded, float64(d.cfg.RefreshInterval)/1000)
-						task.Progress.Downloaded = current
+						tick := float64(d.cfg.RefreshInterval) / 1000
+						if task.Status == base.DownloadStatusRunning {
+							task.Progress.Used = task.timer.Used()
+							task.Progress.Speed = task.calcSpeed(current-task.Progress.Downloaded, tick)
+							task.Progress.Downloaded = current
+						}
+						if task.Uploading {
+							uploader := task.fetcher.(fetcher.Uploader)
+							currentUploaded := uploader.UploadedBytes()
+							task.Progress.UploadSpeed = task.calcSpeed(currentUploaded-task.Progress.Uploaded, tick)
+							task.Progress.Uploaded = currentUploaded
+						}
 						d.emit(EventKeyProgress, task)
 
 						// store fetcher progress
@@ -220,10 +252,10 @@ func (d *Downloader) parseFb(url string) (fetcher.FetcherBuilder, error) {
 	return nil, ErrUnSupportedProtocol
 }
 
-func (d *Downloader) setupFetcher(fetcher fetcher.Fetcher) {
+func (d *Downloader) setupFetcher(fb fetcher.FetcherBuilder, fetcher fetcher.Fetcher) {
 	ctl := controller.NewController()
-	ctl.GetConfig = func(v any) bool {
-		return d.getProtocolConfig(fetcher.Name(), v)
+	ctl.GetConfig = func(v any) {
+		d.getProtocolConfig(fb.Name(), v)
 	}
 	ctl.ProxyConfig = d.cfg.Proxy
 	fetcher.Setup(ctl)
@@ -641,6 +673,30 @@ func (d *Downloader) getProtocolConfig(name string, v any) bool {
 
 // wait task done
 func (d *Downloader) watch(task *Task) {
+	// wait task upload done
+	if task.Uploading {
+		if uploader, ok := task.fetcher.(fetcher.Uploader); ok {
+			go func() {
+				err := uploader.WaitUpload()
+				if err != nil {
+					d.Logger.Warn().Err(err).Msgf("task wait upload failed, task id: %s", task.ID)
+				}
+				d.lock.Lock()
+				defer d.lock.Unlock()
+
+				// Check if the task is deleted
+				if d.GetTask(task.ID) != nil {
+					task.Uploading = false
+					d.storage.Put(bucketTask, task.ID, task.clone())
+				}
+			}()
+		}
+	}
+
+	if task.Status == base.DownloadStatusDone {
+		return
+	}
+
 	err := task.fetcher.Wait()
 	if err != nil {
 		d.doOnError(task, err)
@@ -719,7 +775,7 @@ func (d *Downloader) restoreFetcher(task *Task) error {
 		if task.fetcher == nil {
 			task.fetcher = fb.Build()
 		}
-		d.setupFetcher(task.fetcher)
+		d.setupFetcher(fb, task.fetcher)
 		if task.fetcher.Meta().Req == nil {
 			task.fetcher.Meta().Req = task.Meta.Req
 		}
@@ -734,7 +790,7 @@ func (d *Downloader) restoreFetcher(task *Task) error {
 	return nil
 }
 
-func (d *Downloader) doCreate(fetcher fetcher.Fetcher, opts *base.Options) (taskId string, err error) {
+func (d *Downloader) doCreate(f fetcher.Fetcher, opts *base.Options) (taskId string, err error) {
 	if opts == nil {
 		opts = &base.Options{}
 	}
@@ -742,7 +798,7 @@ func (d *Downloader) doCreate(fetcher fetcher.Fetcher, opts *base.Options) (task
 		opts.SelectFiles = make([]int, 0)
 	}
 
-	meta := fetcher.Meta()
+	meta := f.Meta()
 	meta.Opts = opts
 	if opts.Path == "" {
 		storeConfig, err := d.GetConfig()
@@ -752,19 +808,20 @@ func (d *Downloader) doCreate(fetcher fetcher.Fetcher, opts *base.Options) (task
 		opts.Path = storeConfig.DownloadDir
 	}
 
-	fb, err := d.parseFb(fetcher.Meta().Req.URL)
+	fb, err := d.parseFb(f.Meta().Req.URL)
 	if err != nil {
 		return
 	}
 
 	task := NewTask()
 	task.fetcherBuilder = fb
-	task.fetcher = fetcher
-	task.Protocol = fetcher.Name()
-	task.Meta = fetcher.Meta()
+	task.fetcher = f
+	task.Protocol = fb.Name()
+	task.Meta = f.Meta()
 	task.Progress = &Progress{}
+	_, task.Uploading = f.(fetcher.Uploader)
 	initTask(task)
-	if err = fetcher.Create(opts); err != nil {
+	if err = f.Create(opts); err != nil {
 		return
 	}
 	if err = d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
@@ -867,10 +924,10 @@ func (d *Downloader) doStart(task *Task) (err error) {
 
 		task.Progress.Speed = 0
 		task.timer.Start()
-		if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
+		if err := task.fetcher.Start(); err != nil {
 			return err
 		}
-		if err := task.fetcher.Start(); err != nil {
+		if err := d.storage.Put(bucketTask, task.ID, task.clone()); err != nil {
 			return err
 		}
 		d.emit(EventKeyStart, task)
@@ -945,7 +1002,7 @@ func (d *Downloader) buildFetcher(url string) (fetcher.Fetcher, error) {
 		return nil, err
 	}
 	fetcher := fb.Build()
-	d.setupFetcher(fetcher)
+	d.setupFetcher(fb, fetcher)
 	return fetcher, nil
 }
 

pkg/download/downloader_test.go

@@ -342,8 +342,8 @@ func TestDownloader_Protocol_Config(t *testing.T) {
 
 	var httpCfg map[string]any
 	exits := downloader.getProtocolConfig("http", &httpCfg)
-	if exits {
-		t.Errorf("getProtocolConfig() got = %v, want %v", exits, false)
+	if !exits {
+		t.Errorf("getProtocolConfig() got = %v, want %v", exits, true)
 	}
 
 	storeCfg := &base.DownloaderStoreConfig{

pkg/download/extension_test.go

@@ -238,7 +238,7 @@ func TestDownloader_Extension_OnError(t *testing.T) {
 		select {
 		case err = <-errCh:
 			break
-		case <-time.After(time.Second * 30):
+		case <-time.After(time.Second * 10):
 			err = errors.New("timeout")
 		}
 

pkg/download/model.go

@@ -19,9 +19,10 @@ type ResolveResult struct {
 
 type Task struct {
 	ID        string               `json:"id"`
+	Protocol  string               `json:"protocol"`
 	Meta      *fetcher.FetcherMeta `json:"meta"`
 	Status    base.Status          `json:"status"`
-	Protocol  string               `json:"protocol"`
+	Uploading bool                 `json:"uploading"`
 	Progress  *Progress            `json:"progress"`
 	CreatedAt time.Time            `json:"createdAt"`
 	UpdatedAt time.Time            `json:"updatedAt"`
@@ -55,8 +56,10 @@ func (t *Task) updateStatus(status base.Status) {
 func (t *Task) clone() *Task {
 	return &Task{
 		ID:        t.ID,
+		Protocol:  t.Protocol,
 		Meta:      t.Meta,
 		Status:    t.Status,
+		Uploading: t.Uploading,
 		Progress:  t.Progress,
 		CreatedAt: t.CreatedAt,
 		UpdatedAt: t.UpdatedAt,

pkg/download/testdata/extensions/on_error/index.js

@@ -2,7 +2,6 @@ gopeed.events.onError(async function (ctx) {
     gopeed.logger.info("url", ctx.task.meta.req.url);
     gopeed.logger.info("error", ctx.error);
     ctx.task.meta.req.url = "https://github.com";
-    ctx.task.pause();
     ctx.task.continue();
 });
 

pkg/protocol/bt/model.go

@@ -4,11 +4,16 @@ type ReqExtra struct {
 	Trackers []string `json:"trackers"`
 }
 
-// Stats for download
+// Stats for torrent
 type Stats struct {
-	// bt stats
 	// health indicators of torrents, from large to small, ConnectedSeeders are also the key to the health of seed resources
 	TotalPeers       int `json:"totalPeers"`
 	ActivePeers      int `json:"activePeers"`
 	ConnectedSeeders int `json:"connectedSeeders"`
+	// Total seed bytes
+	SeedBytes int64 `json:"seedBytes"`
+	// Seed ratio
+	SeedRatio float64 `json:"seedRatio"`
+	// Total seed time
+	SeedTime int64 `json:"seedTime"`
 }