143 lines
3.8 KiB
Go
143 lines
3.8 KiB
Go
package jobs
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"ddc/internal/explain"
|
|
"ddc/internal/extract"
|
|
"ddc/internal/fetch"
|
|
)
|
|
|
|
func (manager *Manager) Pause(id string) error {
|
|
job, err := manager.lookup(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !job.pause() {
|
|
return ErrJobBusy
|
|
}
|
|
job.logger().Info("pause requested")
|
|
return nil
|
|
}
|
|
|
|
func (manager *Manager) Resume(id string) error {
|
|
job, err := manager.lookup(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !job.begin(StatusQueued, StatusPaused, StatusFailed) {
|
|
return ErrJobBusy
|
|
}
|
|
job.logger().Info("download resumed", "has_saved_progress", job.currentDownloadState() != nil)
|
|
job.saveRecord()
|
|
manager.launchDownload(job)
|
|
return nil
|
|
}
|
|
|
|
func (manager *Manager) Reorder(id, direction string) error {
|
|
if direction != queueFirst && direction != queueUp && direction != queueDown {
|
|
return invalid(errors.New("direction must be first, up or down"))
|
|
}
|
|
job, err := manager.lookup(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if job.currentStatus() != StatusQueued || !manager.slots.reorder(id, direction) {
|
|
return ErrJobBusy
|
|
}
|
|
job.logger().Debug("queue position changed", "direction", direction, "position", manager.slots.position(id))
|
|
return nil
|
|
}
|
|
|
|
func (manager *Manager) launchDownload(job *Job) {
|
|
runContext, cancelRun := job.startRun()
|
|
manager.slots.enqueue(job.id)
|
|
manager.runInBackground(job, "download", func() {
|
|
defer cancelRun(nil)
|
|
manager.download(job, runContext)
|
|
})
|
|
}
|
|
|
|
func (manager *Manager) download(job *Job, runContext context.Context) {
|
|
if !manager.slots.acquire(runContext, job.id, manager.concurrencyLimit) {
|
|
manager.handleDownloadError(job, runContext, context.Cause(runContext))
|
|
return
|
|
}
|
|
defer manager.slots.release()
|
|
|
|
job.setStatus(StatusDownloading, "")
|
|
started := time.Now()
|
|
job.logger().Info("download started", "url", job.displayURL)
|
|
filePath, err := manager.downloader.Download(runContext, manager.fetchRequest(job))
|
|
if err != nil {
|
|
manager.handleDownloadError(job, runContext, err)
|
|
return
|
|
}
|
|
job.markDownloaded(filepath.Base(filePath))
|
|
job.logger().Info("download finished",
|
|
"file", filepath.Base(filePath),
|
|
"bytes", job.bytesDone.Load(),
|
|
"duration", time.Since(started).Round(time.Second).String(),
|
|
)
|
|
|
|
if job.autoExtract && extract.IsArchive(filePath) {
|
|
job.setStatus(StatusExtracting, "")
|
|
destination, err := manager.extractArchive(job, filePath)
|
|
if err != nil {
|
|
manager.failExtraction(job, err)
|
|
job.saveRecord()
|
|
return
|
|
}
|
|
job.SetFileName(filepath.Base(destination))
|
|
}
|
|
|
|
job.setStatus(StatusReady, "")
|
|
job.saveRecord()
|
|
manager.moveToTarget(job)
|
|
}
|
|
|
|
func (manager *Manager) fetchRequest(job *Job) fetch.Request {
|
|
current := manager.preferences.Get()
|
|
return fetch.Request{
|
|
URL: job.sourceURL,
|
|
Directory: job.directory,
|
|
FileName: job.requestedName,
|
|
ReservedBytes: current.MinFreeBytes(),
|
|
Connections: current.ConnectionsPerDownload,
|
|
State: job.currentDownloadState(),
|
|
Progress: job,
|
|
Logger: job.logger(),
|
|
Checkpoint: func(state fetch.State) {
|
|
job.setDownloadState(&state)
|
|
job.saveRecord()
|
|
},
|
|
}
|
|
}
|
|
|
|
func (manager *Manager) handleDownloadError(job *Job, runContext context.Context, err error) {
|
|
switch {
|
|
case errors.Is(context.Cause(runContext), errPaused):
|
|
job.setStatus(StatusPaused, "")
|
|
job.logger().Info("download paused", "bytes_done", job.bytesDone.Load())
|
|
case job.jobContext.Err() != nil:
|
|
job.setStatus(StatusCancelled, "")
|
|
job.logger().Info("download stopped", "bytes_done", job.bytesDone.Load())
|
|
return
|
|
default:
|
|
if errors.Is(err, fetch.ErrSourceChanged) || errors.Is(err, fetch.ErrResumeUnsupported) {
|
|
job.setDownloadState(nil)
|
|
}
|
|
job.setStatus(StatusFailed, explain.Error(err))
|
|
job.logger().Error("download failed",
|
|
"url", job.displayURL,
|
|
"bytes_done", job.bytesDone.Load(),
|
|
"bytes_total", job.bytesTotal.Load(),
|
|
"error", err,
|
|
)
|
|
}
|
|
job.saveRecord()
|
|
}
|