diff --git a/cmd/extmock/main.go b/cmd/extmock/main.go new file mode 100644 index 00000000..8a79fed7 --- /dev/null +++ b/cmd/extmock/main.go @@ -0,0 +1,291 @@ +package main + +import ( + "encoding/json" + "flag" + "fmt" + "os" + "path/filepath" + + "github.com/GopeedLab/gopeed/pkg/base" + "github.com/GopeedLab/gopeed/pkg/download" +) + +type kvFlag map[string]string + +func (l *kvFlag) String() string { + if l == nil { + return "" + } + data, _ := json.Marshal(map[string]string(*l)) + return string(data) +} + +func (l *kvFlag) Set(value string) error { + if *l == nil { + *l = make(map[string]string) + } + idx := -1 + for i := 0; i < len(value); i++ { + if value[i] == '=' { + idx = i + break + } + } + if idx <= 0 { + return fmt.Errorf("invalid label %q, expected key=value", value) + } + (*l)[value[:idx]] = value[idx+1:] + return nil +} + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + + switch os.Args[1] { + case "resolve": + if err := runResolve(os.Args[2:]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + case "eval": + if err := runEval(os.Args[2:]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + case "-h", "--help", "help": + usage() + default: + fmt.Fprintf(os.Stderr, "unknown subcommand %q\n", os.Args[1]) + usage() + os.Exit(2) + } +} + +func usage() { + fmt.Fprintf(os.Stderr, `gopeed extmock CLI + +Usage: + extmock resolve [flags] + extmock eval [flags] [script-file] + +Flags: + -label key=value Request label, repeatable + -setting key=value Extension setting override, repeatable + -pretty Pretty-print JSON output + +Examples: + extmock resolve ./ https://www.google.com + extmock resolve ./ https://example.com/video -label env=test -setting token=abc + extmock eval ./test.js + extmock eval -e "new Blob(['hello']).size" +`) +} + +func runResolve(args []string) error { + fs := flag.NewFlagSet("resolve", flag.ContinueOnError) + fs.SetOutput(os.Stderr) + var labels kvFlag + var settings kvFlag + var pretty bool + fs.Var(&labels, "label", "Request label in key=value form (repeatable)") + fs.Var(&settings, "setting", "Extension setting override in key=value form (repeatable)") + fs.BoolVar(&pretty, "pretty", false, "Pretty-print JSON output") + + flagArgs, positionalArgs, err := splitResolveArgs(args) + if err != nil { + return err + } + if err := fs.Parse(flagArgs); err != nil { + return err + } + if fs.NArg() != 0 { + return fmt.Errorf("unexpected trailing args: %v", fs.Args()) + } + if len(positionalArgs) != 2 { + return fmt.Errorf("resolve expects exactly 2 args: ") + } + + extDir, err := filepath.Abs(positionalArgs[0]) + if err != nil { + return err + } + url := positionalArgs[1] + + d, cleanup, err := newExtMockDownloader() + if err != nil { + return err + } + defer cleanup() + + ext, err := d.InstallExtensionByFolder(extDir, true) + if err != nil { + return err + } + if len(settings) > 0 { + if err := d.UpdateExtensionSettings(ext.Identity, toAnyMap(settings)); err != nil { + return err + } + } + result, err := d.Resolve(&base.Request{ + URL: url, + Labels: map[string]string(labels), + }, nil) + if err != nil { + return err + } + return printJSON(result, pretty) +} + +func runEval(args []string) error { + fs := flag.NewFlagSet("eval", flag.ContinueOnError) + fs.SetOutput(os.Stderr) + var ( + script string + pretty bool + ) + fs.StringVar(&script, "e", "", "Inline JavaScript to execute") + fs.StringVar(&script, "eval", "", "Inline JavaScript to execute") + fs.BoolVar(&pretty, "pretty", false, "Pretty-print JSON output") + + flagArgs, positionalArgs, err := splitEvalArgs(args) + if err != nil { + return err + } + if err := fs.Parse(flagArgs); err != nil { + return err + } + if fs.NArg() != 0 { + return fmt.Errorf("unexpected trailing args: %v", fs.Args()) + } + if script == "" && len(positionalArgs) == 0 { + return fmt.Errorf("eval expects inline code via -e/--eval or a script file path") + } + if script != "" && len(positionalArgs) > 0 { + return fmt.Errorf("eval accepts either inline code or a script file path, not both") + } + if len(positionalArgs) > 1 { + return fmt.Errorf("eval expects at most 1 script file path") + } + + d, cleanup, err := newExtMockDownloader() + if err != nil { + return err + } + defer cleanup() + + runtime, err := d.NewExtMockRuntime() + if err != nil { + return err + } + defer runtime.Close() + + var value any + if script != "" { + value, err = runtime.Eval(script) + } else { + scriptPath, pathErr := filepath.Abs(positionalArgs[0]) + if pathErr != nil { + return pathErr + } + value, err = runtime.EvalFile(scriptPath) + } + if err != nil { + return err + } + return printJSON(value, pretty) +} + +func splitResolveArgs(args []string) ([]string, []string, error) { + var flagArgs []string + var positionalArgs []string + for i := 0; i < len(args); i++ { + arg := args[i] + switch { + case arg == "-pretty": + flagArgs = append(flagArgs, arg) + case arg == "-label": + if i+1 >= len(args) { + return nil, nil, fmt.Errorf("flag %q requires a value", arg) + } + flagArgs = append(flagArgs, arg, args[i+1]) + i++ + case arg == "-setting": + if i+1 >= len(args) { + return nil, nil, fmt.Errorf("flag %q requires a value", arg) + } + flagArgs = append(flagArgs, arg, args[i+1]) + i++ + case len(arg) > len("-label=") && arg[:len("-label=")] == "-label=": + flagArgs = append(flagArgs, arg) + case len(arg) > len("-setting=") && arg[:len("-setting=")] == "-setting=": + flagArgs = append(flagArgs, arg) + case len(arg) > 0 && arg[0] == '-': + return nil, nil, fmt.Errorf("unknown flag %q", arg) + default: + positionalArgs = append(positionalArgs, arg) + } + } + return flagArgs, positionalArgs, nil +} + +func splitEvalArgs(args []string) ([]string, []string, error) { + var flagArgs []string + var positionalArgs []string + for i := 0; i < len(args); i++ { + arg := args[i] + switch { + case arg == "-pretty": + flagArgs = append(flagArgs, arg) + case arg == "-e", arg == "--eval": + if i+1 >= len(args) { + return nil, nil, fmt.Errorf("flag %q requires a value", arg) + } + flagArgs = append(flagArgs, arg, args[i+1]) + i++ + case hasFlagValue(arg, "-e"), hasFlagValue(arg, "--eval"): + flagArgs = append(flagArgs, arg) + case len(arg) > 0 && arg[0] == '-': + return nil, nil, fmt.Errorf("unknown flag %q", arg) + default: + positionalArgs = append(positionalArgs, arg) + } + } + return flagArgs, positionalArgs, nil +} + +func hasFlagValue(arg string, prefix string) bool { + return len(arg) > len(prefix)+1 && arg[:len(prefix)+1] == prefix+"=" +} + +func toAnyMap(values kvFlag) map[string]any { + result := make(map[string]any, len(values)) + for key, value := range values { + result[key] = value + } + return result +} + +func newExtMockDownloader() (*download.Downloader, func(), error) { + return download.NewExtMockDownloader() +} + +func printJSON(v any, pretty bool) error { + var ( + data []byte + err error + ) + if pretty { + data, err = json.MarshalIndent(v, "", " ") + } else { + data, err = json.Marshal(v) + } + if err != nil { + return err + } + fmt.Println(string(data)) + return nil +} diff --git a/internal/protocol/gblob/README.md b/internal/protocol/gblob/README.md new file mode 100644 index 00000000..63276974 --- /dev/null +++ b/internal/protocol/gblob/README.md @@ -0,0 +1,233 @@ +# gblob Protocol + +## Background + +Gopeed extensions can resolve a download target into something that is not a normal remote URL. +Typical examples are: + +- text generated by an extension +- binary data assembled in JavaScript +- a stream produced incrementally inside the extension runtime + +Returning a plain `http` or `https` URL is not enough for these cases. +The `gblob` protocol provides an internal object-URL style transport so extensions can hand generated content to the downloader in a browser-like way. + +The extension-facing model is intentionally close to the browser: + +- `URL.createObjectURL(new Blob(...))` +- `URL.createObjectURL(writableStream)` +- `URL.revokeObjectURL(url)` + +The returned URL looks like: + +```text +gblob: +``` + +`gblob` is a process-local protocol. The source only exists inside the current Gopeed process. + +## Design Goals + +- make extension-generated data download as a normal task +- support both eager `Blob` data and incremental `WritableStream` data +- avoid large in-memory buffering by spooling to disk +- keep pause/continue behavior intuitive +- allow an extension to recreate a new `gblob` URL after source expiration + +## High-Level Architecture + +The implementation has two layers in this directory: + +- [`registry.go`](./registry.go): process-local source registry and spool-file manager +- [`fetcher.go`](./fetcher.go): downloader fetcher implementation for the `gblob` scheme + +The end-to-end flow is: + +1. JavaScript creates a `Blob` or `WritableStream` object URL. +2. Go registers a `gblob` source and assigns it a unique `gblob:` URL. +3. Source bytes are written into a local spool file. +4. The downloader resolves the `gblob` URL through the `gblob` fetcher. +5. The fetcher copies bytes from the spool file into the final download target. + +## Registry + +The registry owns all active `gblob` sources for the current process. + +Each source tracks: + +- source id and `gblob` URL +- local spool-file path +- source type: `blob` or `writable_stream` +- written byte count +- source state: `open`, `closed`, `aborted`, `failed` +- revoke state +- pin count +- optional extension session reference + +Key behaviors: + +- `Blob` sources are written once and immediately closed. +- `WritableStream` sources stay open until the writer closes or aborts. +- readers wait on a per-source notification channel when no new bytes are available yet. +- revocation only prevents future use; cleanup happens when the source is no longer pinned. + +## Fetcher + +The fetcher treats `gblob` as a local file-backed download source. + +### Resolve + +`Resolve()` looks up the source in the registry and builds a single-file `Resource`. + +Behavior differences by source type: + +- `Blob`: resolved size is the current written size +- `WritableStream`: resolved size is reported as `0` until the writer closes + +`Range` is always `false`. + +### Start + +`Start()` opens the final target file and reads from the source spool file. + +There are two resume modes: + +- same `gblob` id: continue from the current target-file size +- different `gblob` id: truncate the target file and restart from byte `0` + +The fetcher persists the last started source id so it can distinguish these two cases across pause/continue and task restore. + +### Pause / Continue + +`Pause()` stops the current copy loop by canceling the reader side. +It does not abort the `gblob` source. + +If the source id stays the same, `Continue()` resumes from the existing target-file offset. +This is how normal in-process `Pause -> Continue` works. + +### Source Replacement + +If an extension replaces `req.url` with a new `gblob:` before the next start, the fetcher detects the id change and restarts from scratch. + +This behavior is important for recovery flows such as: + +- original `gblob` source expired +- extension recreates a new `gblob` URL in `onError` +- task continues with the new source + +## Lifecycle Notes + +### In-process pause and continue + +For the same `gblob` id: + +- the partially downloaded output file is kept +- the fetcher resumes from the current file size +- `WritableStream` producers may continue writing while the task is paused + +### Process restart + +`gblob` sources are not persisted across process restarts. +An unfinished `gblob` task is restored without a live source behind its old URL. + +The intended recovery pattern is: + +1. user clicks continue +2. the old `gblob` source fails +3. extension `onError` runs +4. extension recreates a new `gblob` URL from `req.rawUrl` +5. extension updates `ctx.task.meta.req.url` +6. extension calls `ctx.task.continue()` + +Because the new URL uses a different `gblob` id, the fetcher restarts from byte `0`. + +## Why `rawUrl` Exists + +When an extension resolves a request into `gblob:`, the original site URL would otherwise be lost. +`Request.RawURL` keeps the original creation URL so lifecycle events can still reason about the original target. + +Extension event URL matching prefers `req.rawUrl` when present. + +This is especially useful for `onStart`, `onError`, and `onDone` on `gblob` tasks. + +## Extension Usage + +### 1. Create a `Blob` object URL + +```js +gopeed.events.onResolve(async function (ctx) { + const blob = new Blob(["hello world"], { type: "text/plain" }); + const url = URL.createObjectURL(blob); + + ctx.res = { + name: "example", + files: [ + { + name: "hello.txt", + size: 11, + req: { url } + } + ] + }; +}); +``` + +### 2. Create a streaming object URL + +```js +gopeed.events.onResolve(async function (ctx) { + const stream = new TransformStream(); + const writer = stream.writable.getWriter(); + const url = URL.createObjectURL(stream.writable); + + (async () => { + await writer.write(new TextEncoder().encode("line 1\n")); + await writer.write(new TextEncoder().encode("line 2\n")); + await writer.close(); + })(); + + ctx.res = { + name: "stream-example", + files: [ + { + name: "stream.txt", + req: { url } + } + ] + }; +}); +``` + +### 3. Recover in `onError` + +```js +gopeed.events.onError(async function (ctx) { + const req = ctx.task.meta.req; + if (!req.rawUrl) return; + if (!req.url.startsWith("gblob:")) return; + + req.url = URL.createObjectURL(new Blob(["recovered\n"])); + ctx.task.continue(); +}); +``` + +This creates a new `gblob` id, so the downloader restarts from byte `0`. + +## Limitations + +- `gblob` is not a network protocol and does not support range requests. +- unfinished `gblob` sources do not survive process restarts. +- recovery after restart depends on extension logic recreating a new source. +- the current restart behavior is full restart on source-id change, not byte-range recovery across recreated sources. +- `WritableStream` tasks with unknown total size report size `0` until the writer closes. + +## Summary + +`gblob` lets Gopeed extensions treat generated content as a first-class download source. +The registry handles source lifetime and disk spooling. +The fetcher turns those sources into normal download tasks with predictable pause/continue behavior. + +The most important rule is simple: + +- same `gblob` id means resume +- different `gblob` id means restart diff --git a/internal/protocol/gblob/fetcher.go b/internal/protocol/gblob/fetcher.go new file mode 100644 index 00000000..0545eece --- /dev/null +++ b/internal/protocol/gblob/fetcher.go @@ -0,0 +1,377 @@ +package gblob + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + + "github.com/GopeedLab/gopeed/internal/controller" + "github.com/GopeedLab/gopeed/internal/fetcher" + "github.com/GopeedLab/gopeed/pkg/base" +) + +type config struct{} + +type Fetcher struct { + ctl *controller.Controller + registry *Registry + meta *fetcher.FetcherMeta + doneCh chan error + + mu sync.Mutex + cancel context.CancelFunc + runDone chan struct{} + paused bool + closed bool + + downloaded atomic.Int64 + lastSourceID string +} + +type fetcherData struct { + LastSourceID string `json:"lastSourceId"` +} + +func (f *Fetcher) Setup(ctl *controller.Controller) { + f.ctl = ctl + if f.meta == nil { + f.meta = &fetcher.FetcherMeta{} + } + f.doneCh = make(chan error, 1) +} + +func (f *Fetcher) Resolve(req *base.Request, opts *base.Options) error { + if f.registry == nil { + return ErrSourceNotFound + } + src, err := f.registry.Get(req.URL) + if err != nil { + return err + } + f.meta.Req = req + f.meta.Opts = opts + if f.meta.Opts == nil { + f.meta.Opts = &base.Options{} + } + name := f.meta.Opts.Name + if name == "" { + name = src.ID + } + size := src.Snapshot().Written + if src.Type == SourceTypeWritableStream { + // Writable streams may continue producing bytes after Resolve, + // so treat the total size as unknown until the writer closes. + size = 0 + } + f.meta.Res = &base.Resource{ + Size: size, + Range: false, + Files: []*base.FileInfo{ + { + Name: name, + Path: "", + Size: size, + Req: req, + }, + }, + } + return nil +} + +func (f *Fetcher) Start() error { + if f.registry == nil { + return ErrSourceNotFound + } + currentSourceID, err := ParseURL(f.meta.Req.URL) + if err != nil { + return err + } + f.mu.Lock() + if f.runDone != nil { + f.mu.Unlock() + return nil + } + f.paused = false + runDone := make(chan struct{}) + ctx, cancel := context.WithCancel(context.Background()) + f.cancel = cancel + f.runDone = runDone + f.mu.Unlock() + + path := f.meta.SingleFilepath() + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { + return err + } + target, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0644) + if err != nil { + return err + } + offset := int64(0) + if f.lastSourceID != "" && f.lastSourceID != currentSourceID { + if err := target.Truncate(0); err != nil { + target.Close() + return err + } + if _, err := target.Seek(0, io.SeekStart); err != nil { + target.Close() + return err + } + } else { + stat, err := target.Stat() + if err != nil { + target.Close() + return err + } + offset = stat.Size() + if _, err := target.Seek(offset, io.SeekStart); err != nil { + target.Close() + return err + } + } + f.lastSourceID = currentSourceID + f.downloaded.Store(offset) + + go f.copyLoop(ctx, target, offset) + return nil +} + +func (f *Fetcher) copyLoop(ctx context.Context, target *os.File, offset int64) { + defer target.Close() + defer func() { + f.mu.Lock() + runDone := f.runDone + f.runDone = nil + f.cancel = nil + paused := f.paused + closed := f.closed + f.mu.Unlock() + close(runDone) + if closed { + select { + case f.doneCh <- nil: + default: + } + return + } + if paused { + return + } + }() + + src, err := f.registry.Get(f.meta.Req.URL) + if err != nil { + select { + case f.doneCh <- err: + default: + } + return + } + sourceFile, err := os.Open(src.Path) + if err != nil { + select { + case f.doneCh <- err: + default: + } + return + } + defer sourceFile.Close() + + buf := make([]byte, 32*1024) + for { + if err := f.registry.WaitForReadable(ctx, f.meta.Req.URL, offset); err != nil { + if errors.Is(err, context.Canceled) { + return + } + select { + case f.doneCh <- err: + default: + } + return + } + + snapshot, serr := f.registry.Get(f.meta.Req.URL) + if serr != nil { + select { + case f.doneCh <- serr: + default: + } + return + } + current := snapshot.Snapshot() + for offset < current.Written { + remain := current.Written - offset + readSize := int64(len(buf)) + if remain < readSize { + readSize = remain + } + n, err := sourceFile.ReadAt(buf[:readSize], offset) + if err != nil && !errors.Is(err, io.EOF) { + select { + case f.doneCh <- err: + default: + } + return + } + if n == 0 { + break + } + if _, err := target.Write(buf[:n]); err != nil { + select { + case f.doneCh <- err: + default: + } + return + } + offset += int64(n) + f.downloaded.Store(offset) + } + + if current.State == SourceStateClosed && offset >= current.Written { + if f.meta.Res != nil { + f.meta.Res.Size = current.Written + f.meta.Res.Files[0].Size = current.Written + } + select { + case f.doneCh <- nil: + default: + } + return + } + if current.State == SourceStateAborted || current.State == SourceStateFailed { + err := current.Err + if err == nil { + err = ErrSourceAborted + } + select { + case f.doneCh <- err: + default: + } + return + } + } +} + +func (f *Fetcher) Patch(req *base.Request, opts *base.Options) error { + return nil +} + +func (f *Fetcher) Pause() error { + f.mu.Lock() + f.paused = true + cancel := f.cancel + runDone := f.runDone + f.mu.Unlock() + if cancel != nil { + cancel() + } + if runDone != nil { + <-runDone + } + return nil +} + +func (f *Fetcher) Close() error { + f.mu.Lock() + f.closed = true + cancel := f.cancel + runDone := f.runDone + f.mu.Unlock() + if cancel != nil { + cancel() + } + if runDone != nil { + <-runDone + } + select { + case f.doneCh <- nil: + default: + } + return nil +} + +func (f *Fetcher) Stats() any { + return map[string]any{} +} + +func (f *Fetcher) Meta() *fetcher.FetcherMeta { + return f.meta +} + +func (f *Fetcher) Progress() fetcher.Progress { + return fetcher.Progress{f.downloaded.Load()} +} + +func (f *Fetcher) Wait() error { + return <-f.doneCh +} + +type FetcherManager struct { + registry *Registry +} + +func (fm *FetcherManager) SetRegistry(registry *Registry) { + fm.registry = registry +} + +func (fm *FetcherManager) Name() string { + return Scheme +} + +func (fm *FetcherManager) Filters() []*fetcher.SchemeFilter { + return []*fetcher.SchemeFilter{ + { + Type: fetcher.FilterTypeUrl, + Pattern: strings.ToUpper(Scheme), + }, + } +} + +func (fm *FetcherManager) Build() fetcher.Fetcher { + return &Fetcher{ + registry: fm.registry, + } +} + +func (fm *FetcherManager) ParseName(u string) string { + id, err := ParseURL(u) + if err != nil { + return "" + } + return id +} + +func (fm *FetcherManager) AutoRename() bool { + return true +} + +func (fm *FetcherManager) DefaultConfig() any { + return &config{} +} + +func (fm *FetcherManager) Store(f fetcher.Fetcher) (data any, err error) { + _f := f.(*Fetcher) + return &fetcherData{ + LastSourceID: _f.lastSourceID, + }, nil +} + +func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher) { + return &fetcherData{}, func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher { + fd := v.(*fetcherData) + return &Fetcher{ + registry: fm.registry, + meta: meta, + doneCh: make(chan error, 1), + lastSourceID: fd.LastSourceID, + } + } +} + +func (fm *FetcherManager) Close() error { + return nil +} diff --git a/internal/protocol/gblob/registry.go b/internal/protocol/gblob/registry.go new file mode 100644 index 00000000..7a1f0a8e --- /dev/null +++ b/internal/protocol/gblob/registry.go @@ -0,0 +1,447 @@ +package gblob + +import ( + "context" + "errors" + "os" + "path/filepath" + "strings" + "sync" + + gonanoid "github.com/matoous/go-nanoid/v2" +) + +const Scheme = "gblob" + +var ( + ErrInvalidURL = errors.New("invalid gblob url") + ErrSourceNotFound = errors.New("gblob source not found") + ErrSourceRevoked = errors.New("gblob source revoked") + ErrSourceClosed = errors.New("gblob source closed") + ErrSourceAborted = errors.New("gblob source aborted") +) + +type SourceType string + +const ( + SourceTypeBlob SourceType = "blob" + SourceTypeWritableStream SourceType = "writable_stream" +) + +type SourceState string + +const ( + SourceStateOpen SourceState = "open" + SourceStateClosed SourceState = "closed" + SourceStateAborted SourceState = "aborted" + SourceStateFailed SourceState = "failed" +) + +type SessionRef interface { + Retain() + Release() +} + +type Snapshot struct { + Written int64 + State SourceState + Err error + WaitCh <-chan struct{} +} + +type Source struct { + ID string + URL string + Path string + Type SourceType + ContentType string + + mu sync.Mutex + file *os.File + written int64 + state SourceState + err error + revoked bool + pins int + waitCh chan struct{} + session SessionRef + sessionReleased bool +} + +func (s *Source) Snapshot() Snapshot { + s.mu.Lock() + defer s.mu.Unlock() + return Snapshot{ + Written: s.written, + State: s.state, + Err: s.err, + WaitCh: s.waitCh, + } +} + +func (s *Source) State() SourceState { + s.mu.Lock() + defer s.mu.Unlock() + return s.state +} + +func (s *Source) IsRevoked() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.revoked +} + +func (s *Source) pin() { + s.mu.Lock() + defer s.mu.Unlock() + s.pins++ +} + +func (s *Source) unpin() int { + s.mu.Lock() + defer s.mu.Unlock() + if s.pins > 0 { + s.pins-- + } + return s.pins +} + +func (s *Source) shouldCleanupLocked() bool { + return s.revoked && s.pins == 0 && s.state != SourceStateOpen +} + +func (s *Source) notifyLocked() { + close(s.waitCh) + s.waitCh = make(chan struct{}) +} + +func (s *Source) releaseSessionLocked() { + if s.session != nil && !s.sessionReleased { + s.session.Release() + s.sessionReleased = true + } +} + +func (s *Source) write(data []byte) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.revoked { + return ErrSourceRevoked + } + if s.state != SourceStateOpen { + if s.state == SourceStateAborted || s.state == SourceStateFailed { + if s.err != nil { + return s.err + } + return ErrSourceAborted + } + return ErrSourceClosed + } + if len(data) == 0 { + return nil + } + n, err := s.file.Write(data) + if err != nil { + s.state = SourceStateFailed + s.err = err + s.file.Close() + s.file = nil + s.releaseSessionLocked() + s.notifyLocked() + return err + } + s.written += int64(n) + s.notifyLocked() + return nil +} + +func (s *Source) close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state != SourceStateOpen { + return nil + } + s.state = SourceStateClosed + if s.file != nil { + if err := s.file.Close(); err != nil { + s.err = err + } + s.file = nil + } + s.releaseSessionLocked() + s.notifyLocked() + return nil +} + +func (s *Source) abort(err error) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.state != SourceStateOpen { + return nil + } + s.state = SourceStateAborted + if err != nil { + s.err = err + } + if s.file != nil { + _ = s.file.Close() + s.file = nil + } + s.releaseSessionLocked() + s.notifyLocked() + return nil +} + +func (s *Source) revoke() bool { + s.mu.Lock() + defer s.mu.Unlock() + s.revoked = true + return s.shouldCleanupLocked() +} + +func (s *Source) waitForReadable(ctx context.Context, offset int64) error { + for { + snapshot := s.Snapshot() + if offset < snapshot.Written { + return nil + } + switch snapshot.State { + case SourceStateClosed: + return nil + case SourceStateAborted, SourceStateFailed: + if snapshot.Err != nil { + return snapshot.Err + } + return ErrSourceAborted + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-snapshot.WaitCh: + } + } +} + +type Registry struct { + dir string + + mu sync.RWMutex + sources map[string]*Source +} + +func NewRegistry(baseDir string) *Registry { + dir := baseDir + if dir == "" { + dir = filepath.Join(os.TempDir(), "gopeed-gblob") + } + return &Registry{ + dir: dir, + sources: make(map[string]*Source), + } +} + +func BuildURL(id string) string { + return Scheme + ":" + id +} + +func ParseURL(raw string) (string, error) { + if !strings.HasPrefix(strings.ToLower(raw), Scheme+":") { + return "", ErrInvalidURL + } + id := raw[len(Scheme)+1:] + if idx := strings.IndexByte(id, '/'); idx >= 0 { + id = id[:idx] + } + if id == "" { + return "", ErrInvalidURL + } + return id, nil +} + +func (r *Registry) createSource(sourceType SourceType, contentType string, session SessionRef) (*Source, error) { + if err := os.MkdirAll(r.dir, 0755); err != nil { + return nil, err + } + id, err := gonanoid.New() + if err != nil { + return nil, err + } + filePath := filepath.Join(r.dir, id) + file, err := os.Create(filePath) + if err != nil { + return nil, err + } + if session != nil { + session.Retain() + } + src := &Source{ + ID: id, + URL: BuildURL(id), + Path: filePath, + Type: sourceType, + ContentType: contentType, + file: file, + state: SourceStateOpen, + waitCh: make(chan struct{}), + session: session, + } + r.mu.Lock() + r.sources[id] = src + r.mu.Unlock() + return src, nil +} + +func (r *Registry) CreateBlob(data []byte, contentType string) (string, error) { + src, err := r.createSource(SourceTypeBlob, contentType, nil) + if err != nil { + return "", err + } + if err := src.write(data); err != nil { + _ = os.Remove(src.Path) + return "", err + } + if err := src.close(); err != nil { + _ = os.Remove(src.Path) + return "", err + } + return src.URL, nil +} + +func (r *Registry) CreateWritableStream(session SessionRef) (string, error) { + src, err := r.createSource(SourceTypeWritableStream, "", session) + if err != nil { + return "", err + } + return src.URL, nil +} + +func (r *Registry) get(raw string) (*Source, error) { + id, err := ParseURL(raw) + if err != nil { + return nil, err + } + r.mu.RLock() + defer r.mu.RUnlock() + src := r.sources[id] + if src == nil { + return nil, ErrSourceNotFound + } + return src, nil +} + +func (r *Registry) Get(raw string) (*Source, error) { + return r.get(raw) +} + +func (r *Registry) Write(raw string, data []byte) error { + src, err := r.get(raw) + if err != nil { + return err + } + return src.write(data) +} + +func (r *Registry) CloseSource(raw string) error { + src, err := r.get(raw) + if err != nil { + return err + } + if err := src.close(); err != nil { + return err + } + r.cleanupIfNeeded(src) + return nil +} + +func (r *Registry) AbortSource(raw string, err error) error { + src, gerr := r.get(raw) + if gerr != nil { + return gerr + } + if err := src.abort(err); err != nil { + return err + } + r.cleanupIfNeeded(src) + return nil +} + +func (r *Registry) Revoke(raw string) error { + src, err := r.get(raw) + if err != nil { + return err + } + if src.revoke() { + r.cleanupIfNeeded(src) + } + return nil +} + +func (r *Registry) Pin(raw string) error { + src, err := r.get(raw) + if err != nil { + return err + } + if src.IsRevoked() { + return ErrSourceRevoked + } + src.pin() + return nil +} + +func (r *Registry) Unpin(raw string) { + src, err := r.get(raw) + if err != nil { + return + } + src.unpin() + r.cleanupIfNeeded(src) +} + +func (r *Registry) WaitForReadable(ctx context.Context, raw string, offset int64) error { + src, err := r.get(raw) + if err != nil { + return err + } + return src.waitForReadable(ctx, offset) +} + +func (r *Registry) cleanupIfNeeded(src *Source) { + src.mu.Lock() + shouldCleanup := src.shouldCleanupLocked() + file := src.file + src.mu.Unlock() + if !shouldCleanup { + return + } + r.mu.Lock() + delete(r.sources, src.ID) + r.mu.Unlock() + if file != nil { + _ = file.Close() + } + _ = os.Remove(src.Path) +} + +func (r *Registry) Close() error { + r.mu.Lock() + sources := r.sources + r.sources = make(map[string]*Source) + r.mu.Unlock() + var lastErr error + for _, src := range sources { + src.mu.Lock() + if src.file != nil { + if err := src.file.Close(); err != nil { + lastErr = err + } + src.file = nil + } + src.releaseSessionLocked() + src.notifyLocked() + src.mu.Unlock() + if err := os.Remove(src.Path); err != nil && !errors.Is(err, os.ErrNotExist) { + lastErr = err + } + } + return lastErr +} diff --git a/pkg/base/model.go b/pkg/base/model.go index f385423f..c4a9d7d4 100644 --- a/pkg/base/model.go +++ b/pkg/base/model.go @@ -14,8 +14,9 @@ import ( // Request download request type Request struct { - URL string `json:"url"` - Extra any `json:"extra"` + RawURL string `json:"rawUrl"` + URL string `json:"url"` + Extra any `json:"extra"` // Labels is used to mark the download task Labels map[string]string `json:"labels"` // Proxy is special proxy config for request diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 382a5b7c..d1e0ce36 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -18,6 +18,7 @@ import ( "github.com/GopeedLab/gopeed/internal/controller" "github.com/GopeedLab/gopeed/internal/fetcher" "github.com/GopeedLab/gopeed/internal/logger" + protogblob "github.com/GopeedLab/gopeed/internal/protocol/gblob" "github.com/GopeedLab/gopeed/pkg/base" "github.com/GopeedLab/gopeed/pkg/protocol/http" "github.com/GopeedLab/gopeed/pkg/util" @@ -111,6 +112,7 @@ type Downloader struct { claimedExtractions sync.Map extensions []*Extension + gblob *protogblob.Registry } func NewDownloader(cfg *DownloaderConfig) *Downloader { @@ -142,6 +144,8 @@ func NewDownloader(cfg *DownloaderConfig) *Downloader { } func (d *Downloader) Setup() error { + d.gblob = protogblob.NewRegistry(filepath.Join(d.cfg.StorageDir, protogblob.Scheme)) + // setup storage if err := d.storage.Setup([]string{bucketTask, bucketSave, bucketProtocolState, bucketConfig, bucketExtension, bucketExtensionStorage}); err != nil { return err @@ -167,6 +171,9 @@ func (d *Downloader) Setup() error { if _, ok := d.cfg.DownloaderStoreConfig.ProtocolConfig[protocol]; !ok { d.cfg.DownloaderStoreConfig.ProtocolConfig[protocol] = fm.DefaultConfig() } + if gfm, ok := fm.(*protogblob.FetcherManager); ok { + gfm.SetRegistry(d.gblob) + } if sfm, ok := fm.(fetcher.StatefulFetcherManager); ok { sfm.SetStateStore(&protocolStateStore{ storage: d.storage, @@ -192,6 +199,10 @@ func (d *Downloader) Setup() error { } d.assignFetcherManager(task) initTask(task) + if task.Protocol == protogblob.Scheme && task.Status != base.DownloadStatusDone && task.Status != base.DownloadStatusError { + task.Status = base.DownloadStatusError + continue + } if task.Status != base.DownloadStatusDone && task.Status != base.DownloadStatusError { task.Status = base.DownloadStatusPause } @@ -384,11 +395,36 @@ func (d *Downloader) saveTask(task *Task) error { return nil } +func ensureRequestRawURL(req *base.Request) { + if req == nil { + return + } + if req.RawURL == "" { + req.RawURL = req.URL + } +} + +func ensureResourceRequestRawURLs(parentReq *base.Request, res *base.Resource) { + if res == nil { + return + } + for _, file := range res.Files { + if file == nil || file.Req == nil { + continue + } + ensureRequestRawURL(file.Req) + if parentReq != nil && parentReq.RawURL != "" && file.Req.RawURL == file.Req.URL { + file.Req.RawURL = parentReq.RawURL + } + } +} + func (d *Downloader) Resolve(req *base.Request, opts *base.Options) (rr *ResolveResult, err error) { rrId, err := gonanoid.New() if err != nil { return } + ensureRequestRawURL(req) res, err := d.triggerOnResolve(req) if err != nil { @@ -451,6 +487,7 @@ func (d *Downloader) remainRunningCount() int { } func (d *Downloader) CreateDirect(req *base.Request, opts *base.Options) (taskId string, err error) { + ensureRequestRawURL(req) var fetcher fetcher.Fetcher fetcher, err = d.buildFetcher(req.URL) if err != nil { @@ -765,6 +802,7 @@ func (d *Downloader) Stats(id string) (sr any, err error) { } func (d *Downloader) doDelete(task *Task, force bool) (err error) { + defer d.unpinGBlobTask(task) err = func() error { if err := d.storage.Delete(bucketTask, task.ID); err != nil { return err @@ -809,6 +847,9 @@ func (d *Downloader) Close() error { for _, fm := range d.cfg.FetchManagers { closeArr = append(closeArr, fm.Close) } + if d.gblob != nil { + closeArr = append(closeArr, d.gblob.Close) + } closeArr = append(closeArr, d.storage.Close) // Make sure all resources are released, if had error, return the last error var lastErr error @@ -1001,98 +1042,106 @@ func (d *Downloader) watch(task *Task) { return } - err := task.fetcher.Wait() - if err != nil { - d.doOnError(task, err) - return - } - - // When delete a not resolved task, need check if the task resource is nil - if task.Meta.Res == nil || d.GetTask(task.ID) == nil { - return - } - - task.Progress.Used = task.timer.Used() - if task.Meta.Res.Size == 0 { - task.Meta.Res.Size = task.fetcher.Progress().TotalDownloaded() - } - used := task.Progress.Used / int64(time.Second) - if used == 0 { - used = 1 - } - totalSize := task.Meta.Res.Size - task.Progress.Speed = totalSize / used - task.Progress.Downloaded = totalSize - task.updateStatus(base.DownloadStatusDone) - d.storage.Put(bucketTask, task.ID, task.clone()) - d.emit(EventKeyDone, task) - d.emit(EventKeyFinally, task, err) - d.notifyRunning() - d.triggerOnDone(task) - d.triggerWebhooks(WebhookEventDownloadDone, task, nil) - d.triggerScripts(ScriptEventDownloadDone, task, nil) - - if e, ok := task.Meta.Opts.Extra.(*http.OptsExtra); ok { - downloadFilePath := task.Meta.SingleFilepath() - - cfg, _ := d.GetConfig() - - // Determine if auto-torrent is enabled (use global config if not explicitly set) - autoTorrentEnabled := false - if e.AutoTorrent != nil { - autoTorrentEnabled = *e.AutoTorrent - } else if cfg != nil && cfg.AutoTorrent != nil { - autoTorrentEnabled = cfg.AutoTorrent.Enable + for { + err := task.fetcher.Wait() + if err != nil { + d.doOnError(task, err) + if task.Status == base.DownloadStatusRunning { + continue + } + return } - if autoTorrentEnabled && strings.HasSuffix(downloadFilePath, ".torrent") { - // Determine if should delete torrent file after creating BT task - shouldDelete := false - if e.DeleteTorrentAfterDownload != nil { - shouldDelete = *e.DeleteTorrentAfterDownload + // When delete a not resolved task, need check if the task resource is nil + if task.Meta.Res == nil || d.GetTask(task.ID) == nil { + return + } + + task.Progress.Used = task.timer.Used() + if task.Meta.Res.Size == 0 { + task.Meta.Res.Size = task.fetcher.Progress().TotalDownloaded() + } + used := task.Progress.Used / int64(time.Second) + if used == 0 { + used = 1 + } + totalSize := task.Meta.Res.Size + task.Progress.Speed = totalSize / used + task.Progress.Downloaded = totalSize + task.updateStatus(base.DownloadStatusDone) + d.storage.Put(bucketTask, task.ID, task.clone()) + d.emit(EventKeyDone, task) + d.emit(EventKeyFinally, task, err) + d.notifyRunning() + d.unpinGBlobTask(task) + d.triggerOnDone(task) + d.triggerWebhooks(WebhookEventDownloadDone, task, nil) + d.triggerScripts(ScriptEventDownloadDone, task, nil) + + if e, ok := task.Meta.Opts.Extra.(*http.OptsExtra); ok { + downloadFilePath := task.Meta.SingleFilepath() + + cfg, _ := d.GetConfig() + + // Determine if auto-torrent is enabled (use global config if not explicitly set) + autoTorrentEnabled := false + if e.AutoTorrent != nil { + autoTorrentEnabled = *e.AutoTorrent } else if cfg != nil && cfg.AutoTorrent != nil { - shouldDelete = cfg.AutoTorrent.DeleteAfterDownload + autoTorrentEnabled = cfg.AutoTorrent.Enable } - go func() { - _, err2 := d.CreateDirect( - &base.Request{ - URL: downloadFilePath, - }, - &base.Options{ - Path: task.Meta.Opts.Path, - SelectFiles: make([]int, 0), - }) - if err2 != nil { - d.Logger.Error().Err(err2).Msgf("auto create torrent task failed, task id: %s", task.ID) - return + if autoTorrentEnabled && strings.HasSuffix(downloadFilePath, ".torrent") { + // Determine if should delete torrent file after creating BT task + shouldDelete := false + if e.DeleteTorrentAfterDownload != nil { + shouldDelete = *e.DeleteTorrentAfterDownload + } else if cfg != nil && cfg.AutoTorrent != nil { + shouldDelete = cfg.AutoTorrent.DeleteAfterDownload } - if shouldDelete { - d.Delete(&TaskFilter{IDs: []string{task.ID}}, true) - } - }() - } + go func() { + _, err2 := d.CreateDirect( + &base.Request{ + URL: downloadFilePath, + }, + &base.Options{ + Path: task.Meta.Opts.Path, + SelectFiles: make([]int, 0), + }) + if err2 != nil { + d.Logger.Error().Err(err2).Msgf("auto create torrent task failed, task id: %s", task.ID) + return + } - // Determine if auto-extract is enabled (use global config if not explicitly set) - autoExtractEnabled := false - if e.AutoExtract != nil { - autoExtractEnabled = *e.AutoExtract - } else if cfg != nil && cfg.Archive != nil { - autoExtractEnabled = cfg.Archive.AutoExtract - } + if shouldDelete { + d.Delete(&TaskFilter{IDs: []string{task.ID}}, true) + } + }() + } - // Auto-extract archive files using the extraction queue - // This ensures only one extraction runs at a time to prevent resource exhaustion - if autoExtractEnabled && isArchiveFile(downloadFilePath) { - d.enqueueExtraction(task, downloadFilePath, e) + // Determine if auto-extract is enabled (use global config if not explicitly set) + autoExtractEnabled := false + if e.AutoExtract != nil { + autoExtractEnabled = *e.AutoExtract + } else if cfg != nil && cfg.Archive != nil { + autoExtractEnabled = cfg.Archive.AutoExtract + } + + // Auto-extract archive files using the extraction queue + // This ensures only one extraction runs at a time to prevent resource exhaustion + if autoExtractEnabled && isArchiveFile(downloadFilePath) { + d.enqueueExtraction(task, downloadFilePath, e) + } } + return } } func (d *Downloader) doOnError(task *Task, err error) { d.Logger.Warn().Err(err).Msgf("task download failed, task id: %s", task.ID) task.updateStatus(base.DownloadStatusError) + d.unpinGBlobTask(task) d.triggerOnError(task, err) if task.Status == base.DownloadStatusError { d.emit(EventKeyError, task, err) @@ -1141,11 +1190,22 @@ func (d *Downloader) doCreate(f fetcher.Fetcher, opts *base.Options) (taskId str if f.Meta().Opts == nil { f.Meta().Opts = opts } + ensureRequestRawURL(f.Meta().Req) fm, err := d.parseFm(f.Meta().Req.URL) if err != nil { return } + if fm.Name() == protogblob.Scheme && d.gblob != nil { + if err = d.gblob.Pin(f.Meta().Req.URL); err != nil { + return "", err + } + defer func() { + if err != nil { + d.gblob.Unpin(f.Meta().Req.URL) + } + }() + } task := NewTask() task.fetcherManager = fm @@ -1772,6 +1832,16 @@ func initTask(task *Task) { task.uploadSpeedArr = make([]int64, 0) } +func (d *Downloader) unpinGBlobTask(task *Task) { + if d.gblob == nil || task == nil || task.Meta == nil || task.Meta.Req == nil || task.Meta.Req.URL == "" { + return + } + if task.Protocol != protogblob.Scheme { + return + } + d.gblob.Unpin(task.Meta.Req.URL) +} + var defaultDownloader = NewDownloader(nil) type boot struct { diff --git a/pkg/download/engine/engine.go b/pkg/download/engine/engine.go index 93eaa60c..2355cb8d 100644 --- a/pkg/download/engine/engine.go +++ b/pkg/download/engine/engine.go @@ -3,10 +3,12 @@ package engine import ( _ "embed" "errors" + "github.com/GopeedLab/gopeed/pkg/base" gojaerror "github.com/GopeedLab/gopeed/pkg/download/engine/inject/error" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/file" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/formdata" + "github.com/GopeedLab/gopeed/pkg/download/engine/inject/stream" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/vm" "github.com/GopeedLab/gopeed/pkg/download/engine/inject/xhr" "github.com/dop251/goja" @@ -27,90 +29,62 @@ type Engine struct { // RunString executes the script and returns the go type value // if script result is promise, it will be resolved func (e *Engine) RunString(script string) (value any, err error) { - defer func() { - if r := recover(); r != nil { - err = r.(error) - } - }() - - var result goja.Value - e.loop.Run(func(runtime *goja.Runtime) { - result, err = runtime.RunString(script) - if err == nil { - go e.await(result) - } + return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) { + return runtime.RunString(script) }) - if err != nil { - return - } - return resolveResult(result) } // CallFunction calls the function and returns the go type value // if function result is promise, it will be resolved func (e *Engine) CallFunction(fn goja.Callable, args ...any) (value any, err error) { - defer func() { - if r := recover(); r != nil { - err = r.(error) - } - }() - - var result goja.Value - e.loop.Run(func(runtime *goja.Runtime) { + return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) { if args == nil { - result, err = fn(nil) - } else { - var jsArgs []goja.Value - for _, arg := range args { - jsArgs = append(jsArgs, runtime.ToValue(arg)) - } - result, err = fn(nil, jsArgs...) + return fn(nil) } - if err == nil { - go e.await(result) + var jsArgs []goja.Value + for _, arg := range args { + jsArgs = append(jsArgs, runtime.ToValue(arg)) } + return fn(nil, jsArgs...) }) - if err != nil { - return - } - return resolveResult(result) } -// loop.Run will hang if the script result has a non-stop code, such as setInterval. -// This method will stop the event loop when the promise result is resolved. -func (e *Engine) await(value any) { - if value == nil { - return +func (e *Engine) runOnLoop(fn func(runtime *goja.Runtime) (goja.Value, error)) (any, error) { + type result struct { + value goja.Value + err error } - - if v, ok := value.(goja.Value); ok { - // if result is promise, wait for it to be resolved - if p, ok := v.Export().(*goja.Promise); ok { - if p.State() != goja.PromiseStatePending { - return - } - - // check promise state every 100 milliseconds, until it is resolved - for { - time.Sleep(time.Millisecond * 100) - if p.State() == goja.PromiseStatePending { - continue + ch := make(chan result, 1) + ok := e.loop.RunOnLoop(func(runtime *goja.Runtime) { + defer func() { + if r := recover(); r != nil { + if err, ok := r.(error); ok { + ch <- result{err: err} + return } - break + ch <- result{err: errors.New("panic")} } - - // stop the event loop - e.loop.StopNoWait() - } + }() + value, err := fn(runtime) + ch <- result{value: value, err: err} + }) + if !ok { + return nil, errors.New("engine loop terminated") } + res := <-ch + if res.err != nil { + return nil, res.err + } + return resolveResult(res.value) } func (e *Engine) Close() { - e.loop.StopNoWait() + e.loop.Terminate() } type Config struct { - ProxyConfig *base.DownloaderProxyConfig + ProxyConfig *base.DownloaderProxyConfig + StreamConfig *stream.Config } func NewEngine(cfg *Config) *Engine { @@ -121,7 +95,10 @@ func NewEngine(cfg *Config) *Engine { engine := &Engine{ loop: loop, } - loop.Run(func(runtime *goja.Runtime) { + loop.Start() + done := make(chan struct{}) + loop.RunOnLoop(func(runtime *goja.Runtime) { + defer close(done) engine.Runtime = runtime runtime.SetFieldNameMapper(goja.TagFieldNameMapper("json", true)) vm.Enable(runtime) @@ -153,8 +130,12 @@ func NewEngine(cfg *Config) *Engine { if _, err := runtime.RunString("global.location = new URL('http://localhost');"); err != nil { return } + if err := stream.Enable(runtime, cfg.StreamConfig); err != nil { + return + } return }) + <-done return engine } @@ -169,6 +150,9 @@ func resolveResult(value goja.Value) (any, error) { switch export.(type) { case *goja.Promise: p := export.(*goja.Promise) + for p.State() == goja.PromiseStatePending { + time.Sleep(time.Millisecond * 10) + } switch p.State() { case goja.PromiseStatePending: return nil, nil diff --git a/pkg/download/engine/inject/stream/module.go b/pkg/download/engine/inject/stream/module.go new file mode 100644 index 00000000..a0d63be3 --- /dev/null +++ b/pkg/download/engine/inject/stream/module.go @@ -0,0 +1,92 @@ +package stream + +import ( + _ "embed" + "fmt" + + "github.com/dop251/goja" +) + +//go:embed stream.js +var script string + +type Config struct { + CreateBlobObjectURL func(data []byte, contentType string) (string, error) + CreateWritableStreamObjectURL func() (string, error) + WriteWritableStreamObjectURL func(url string, data any) error + CloseWritableStreamObjectURL func(url string) error + AbortWritableStreamObjectURL func(url string, reason string) error + RevokeObjectURL func(url string) error +} + +func Enable(runtime *goja.Runtime, cfg *Config) error { + if cfg == nil { + cfg = &Config{} + } + if err := runtime.Set("__gopeed_create_blob_object_url", func(data []byte, contentType string) string { + if cfg.CreateBlobObjectURL == nil { + panic(fmt.Errorf("gblob blob object url handler not configured")) + } + url, err := cfg.CreateBlobObjectURL(data, contentType) + if err != nil { + panic(err) + } + return url + }); err != nil { + return err + } + if err := runtime.Set("__gopeed_create_writable_stream_object_url", func() string { + if cfg.CreateWritableStreamObjectURL == nil { + panic(fmt.Errorf("gblob writable stream object url handler not configured")) + } + url, err := cfg.CreateWritableStreamObjectURL() + if err != nil { + panic(err) + } + return url + }); err != nil { + return err + } + if err := runtime.Set("__gopeed_write_writable_stream_object_url", func(url string, data any) { + if cfg.WriteWritableStreamObjectURL == nil { + panic(fmt.Errorf("gblob writable stream write handler not configured")) + } + if err := cfg.WriteWritableStreamObjectURL(url, data); err != nil { + panic(err) + } + }); err != nil { + return err + } + if err := runtime.Set("__gopeed_close_writable_stream_object_url", func(url string) { + if cfg.CloseWritableStreamObjectURL == nil { + panic(fmt.Errorf("gblob writable stream close handler not configured")) + } + if err := cfg.CloseWritableStreamObjectURL(url); err != nil { + panic(err) + } + }); err != nil { + return err + } + if err := runtime.Set("__gopeed_abort_writable_stream_object_url", func(url string, reason string) { + if cfg.AbortWritableStreamObjectURL == nil { + panic(fmt.Errorf("gblob writable stream abort handler not configured")) + } + if err := cfg.AbortWritableStreamObjectURL(url, reason); err != nil { + panic(err) + } + }); err != nil { + return err + } + if err := runtime.Set("__gopeed_revoke_object_url", func(url string) { + if cfg.RevokeObjectURL == nil { + panic(fmt.Errorf("gblob revoke handler not configured")) + } + if err := cfg.RevokeObjectURL(url); err != nil { + panic(err) + } + }); err != nil { + return err + } + _, err := runtime.RunString(script) + return err +} diff --git a/pkg/download/engine/inject/stream/stream.js b/pkg/download/engine/inject/stream/stream.js new file mode 100644 index 00000000..5492ddfd --- /dev/null +++ b/pkg/download/engine/inject/stream/stream.js @@ -0,0 +1,341 @@ +(function () { + const createBlobObjectURL = globalThis.__gopeed_create_blob_object_url; + const createWritableObjectURL = globalThis.__gopeed_create_writable_stream_object_url; + const writeWritableObjectURL = globalThis.__gopeed_write_writable_stream_object_url; + const closeWritableObjectURL = globalThis.__gopeed_close_writable_stream_object_url; + const abortWritableObjectURL = globalThis.__gopeed_abort_writable_stream_object_url; + const revokeObjectURL = globalThis.__gopeed_revoke_object_url; + + if (typeof globalThis.ReadableStream === "undefined") { + class ReadableStreamDefaultController { + constructor(stream) { + this._stream = stream; + } + + enqueue(chunk) { + this._stream._enqueue(chunk); + } + + close() { + this._stream._close(); + } + + error(err) { + this._stream._error(err); + } + } + + class ReadableStreamDefaultReader { + constructor(stream) { + this._stream = stream; + } + + read() { + return this._stream._read(); + } + + releaseLock() { + if (this._stream) { + this._stream._reader = null; + this._stream.locked = false; + this._stream = null; + } + } + } + + class ReadableStream { + constructor(source = {}) { + this._queue = []; + this._waiters = []; + this._closed = false; + this._errored = null; + this._reader = null; + this.locked = false; + const controller = new ReadableStreamDefaultController(this); + if (typeof source.start === "function") { + source.start(controller); + } + } + + _enqueue(chunk) { + if (this._closed || this._errored) { + return; + } + if (this._waiters.length > 0) { + const waiter = this._waiters.shift(); + waiter.resolve({ done: false, value: chunk }); + return; + } + this._queue.push(chunk); + } + + _close() { + this._closed = true; + while (this._waiters.length > 0) { + this._waiters.shift().resolve({ done: true, value: undefined }); + } + } + + _error(err) { + this._errored = err || new Error("ReadableStream error"); + while (this._waiters.length > 0) { + this._waiters.shift().reject(this._errored); + } + } + + _read() { + if (this._errored) { + return Promise.reject(this._errored); + } + if (this._queue.length > 0) { + return Promise.resolve({ done: false, value: this._queue.shift() }); + } + if (this._closed) { + return Promise.resolve({ done: true, value: undefined }); + } + return new Promise((resolve, reject) => { + this._waiters.push({ resolve, reject }); + }); + } + + getReader() { + if (this.locked) { + throw new TypeError("ReadableStream is locked"); + } + this.locked = true; + this._reader = new ReadableStreamDefaultReader(this); + return this._reader; + } + + async pipeTo(dest) { + const reader = this.getReader(); + const writer = dest.getWriter(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + await writer.close(); + break; + } + await writer.write(value); + } + } catch (err) { + await writer.abort(err); + throw err; + } finally { + reader.releaseLock(); + writer.releaseLock(); + } + } + } + + globalThis.ReadableStream = ReadableStream; + } + + if (typeof globalThis.WritableStream === "undefined") { + class WritableStreamDefaultWriter { + constructor(stream) { + if (stream.locked) { + throw new TypeError("WritableStream is locked"); + } + this._stream = stream; + this._stream.locked = true; + } + + write(chunk) { + return this._stream._write(chunk); + } + + close() { + return this._stream._close(); + } + + abort(reason) { + return this._stream._abort(reason); + } + + releaseLock() { + if (this._stream) { + this._stream.locked = false; + this._stream = null; + } + } + } + + class WritableStream { + constructor(sink = {}) { + this._sink = sink; + this._observers = []; + this._state = "writable"; + this.locked = false; + } + + _targets() { + return [this._sink, ...this._observers]; + } + + _write(chunk) { + if (this._state !== "writable") { + return Promise.reject(new TypeError("WritableStream is not writable")); + } + let chain = Promise.resolve(); + for (const target of this._targets()) { + if (target && typeof target.write === "function") { + chain = chain.then(() => target.write(chunk)); + } + } + return chain; + } + + _close() { + if (this._state !== "writable") { + return Promise.reject(new TypeError("WritableStream is not writable")); + } + this._state = "closed"; + let chain = Promise.resolve(); + for (const target of this._targets()) { + if (target && typeof target.close === "function") { + chain = chain.then(() => target.close()); + } + } + return chain; + } + + _abort(reason) { + if (this._state === "errored") { + return Promise.resolve(); + } + this._state = "errored"; + let chain = Promise.resolve(); + for (const target of this._targets()) { + if (target && typeof target.abort === "function") { + chain = chain.then(() => target.abort(reason)); + } + } + return chain; + } + + _addObserver(observer) { + this._observers.push(observer); + } + + getWriter() { + return new WritableStreamDefaultWriter(this); + } + } + + globalThis.WritableStream = WritableStream; + } + + if (typeof globalThis.TransformStream === "undefined") { + class TransformStream { + constructor(transformer = {}) { + let readableController; + this.readable = new ReadableStream({ + start(controller) { + readableController = controller; + } + }); + this.writable = new WritableStream({ + async write(chunk) { + if (typeof transformer.transform === "function") { + await transformer.transform(chunk, readableController); + return; + } + readableController.enqueue(chunk); + }, + async close() { + if (typeof transformer.flush === "function") { + await transformer.flush(readableController); + } + readableController.close(); + }, + async abort(reason) { + if (typeof transformer.abort === "function") { + await transformer.abort(reason); + } + readableController.error(reason || new Error("TransformStream aborted")); + } + }); + } + } + + globalThis.TransformStream = TransformStream; + } + + if (typeof globalThis.ByteLengthQueuingStrategy === "undefined") { + globalThis.ByteLengthQueuingStrategy = class ByteLengthQueuingStrategy { + constructor({ highWaterMark }) { + this.highWaterMark = highWaterMark; + } + + size(chunk) { + if (typeof chunk === "string") { + return chunk.length; + } + if (chunk && typeof chunk.byteLength === "number") { + return chunk.byteLength; + } + return 1; + } + }; + } + + if (typeof globalThis.CountQueuingStrategy === "undefined") { + globalThis.CountQueuingStrategy = class CountQueuingStrategy { + constructor({ highWaterMark }) { + this.highWaterMark = highWaterMark; + } + + size() { + return 1; + } + }; + } + + const originalCreateObjectURL = typeof URL.createObjectURL === "function" + ? URL.createObjectURL.bind(URL) + : null; + const originalRevokeObjectURL = typeof URL.revokeObjectURL === "function" + ? URL.revokeObjectURL.bind(URL) + : null; + + URL.createObjectURL = function (value) { + if (value instanceof Blob) { + return createBlobObjectURL(value._buffer, value.type || ""); + } + if (value instanceof WritableStream) { + if (value.__gopeedObjectURL) { + return value.__gopeedObjectURL; + } + const url = createWritableObjectURL(); + value._addObserver({ + write(chunk) { + return Promise.resolve(writeWritableObjectURL(url, chunk)); + }, + close() { + return Promise.resolve(closeWritableObjectURL(url)); + }, + abort(reason) { + return Promise.resolve(abortWritableObjectURL(url, reason == null ? "" : String(reason))); + } + }); + value.__gopeedObjectURL = url; + return url; + } + if (originalCreateObjectURL) { + return originalCreateObjectURL(value); + } + throw new TypeError("Unsupported object type"); + }; + + URL.revokeObjectURL = function (url) { + if (typeof url === "string" && url.indexOf("gblob:") === 0) { + revokeObjectURL(url); + return; + } + if (originalRevokeObjectURL) { + originalRevokeObjectURL(url); + } + }; +})(); diff --git a/pkg/download/extension.go b/pkg/download/extension.go index 71e4b895..1cec437f 100644 --- a/pkg/download/extension.go +++ b/pkg/download/extension.go @@ -13,7 +13,6 @@ import ( "github.com/GopeedLab/gopeed/internal/logger" "github.com/GopeedLab/gopeed/pkg/base" - "github.com/GopeedLab/gopeed/pkg/download/engine" gojaerror "github.com/GopeedLab/gopeed/pkg/download/engine/inject/error" gojautil "github.com/GopeedLab/gopeed/pkg/download/engine/util" "github.com/GopeedLab/gopeed/pkg/util" @@ -268,6 +267,7 @@ func (d *Downloader) triggerOnResolve(req *base.Request) (res *base.Resource, er for _, file := range ctx.Res.Files { file.Name = util.SafeFilename(file.Name) } + ensureResourceRequestRawURLs(req, ctx.Res) ctx.Res.CalcSize(nil) } res = ctx.Res @@ -361,10 +361,8 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c if req.Labels == nil { req.Labels = make(map[string]string) } - engine := engine.NewEngine(&engine.Config{ - ProxyConfig: d.cfg.Proxy, - }) - defer engine.Close() + engine, session := d.newExtensionEngine() + defer session.CloseIfIdle() err = engine.Runtime.Set("gopeed", gopeed) if err != nil { gopeed.Logger.logger.Error().Err(err).Msgf("[%s] engine inject failed", ext.buildIdentity()) @@ -551,8 +549,12 @@ func (s *Script) match(event ActivationEvent, req *base.Request) bool { } // match url + targetURL := req.RawURL + if targetURL == "" { + targetURL = req.URL + } for _, url := range s.Match.Urls { - if util.Match(url, req.URL) { + if util.Match(url, targetURL) { return true } } diff --git a/pkg/download/extension_engine_session.go b/pkg/download/extension_engine_session.go new file mode 100644 index 00000000..b7914cf4 --- /dev/null +++ b/pkg/download/extension_engine_session.go @@ -0,0 +1,117 @@ +package download + +import ( + "fmt" + "sync" + + "github.com/GopeedLab/gopeed/pkg/download/engine" + "github.com/GopeedLab/gopeed/pkg/download/engine/inject/stream" + "github.com/dop251/goja" +) + +type engineSession struct { + engine *engine.Engine + + mu sync.Mutex + refs int + closed bool +} + +func newEngineSession(e *engine.Engine) *engineSession { + return &engineSession{engine: e} +} + +func (s *engineSession) SetEngine(e *engine.Engine) { + s.mu.Lock() + defer s.mu.Unlock() + s.engine = e +} + +func (s *engineSession) Retain() { + s.mu.Lock() + defer s.mu.Unlock() + if s.closed { + return + } + s.refs++ +} + +func (s *engineSession) Release() { + s.mu.Lock() + if s.refs > 0 { + s.refs-- + } + shouldClose := s.refs == 0 && !s.closed + if shouldClose { + s.closed = true + } + s.mu.Unlock() + if shouldClose { + if s.engine != nil { + go s.engine.Close() + } + } +} + +func (s *engineSession) CloseIfIdle() { + s.mu.Lock() + shouldClose := s.refs == 0 && !s.closed + if shouldClose { + s.closed = true + } + s.mu.Unlock() + if shouldClose { + if s.engine != nil { + s.engine.Close() + } + } +} + +func normalizeStreamChunk(data any) ([]byte, error) { + switch v := data.(type) { + case nil: + return nil, nil + case string: + return []byte(v), nil + case []byte: + return v, nil + case goja.ArrayBuffer: + return v.Bytes(), nil + default: + return nil, fmt.Errorf("unsupported stream chunk type: %T", data) + } +} + +func (d *Downloader) newExtensionEngine() (*engine.Engine, *engineSession) { + session := newEngineSession(nil) + engineCfg := &stream.Config{ + CreateBlobObjectURL: func(data []byte, contentType string) (string, error) { + return d.gblob.CreateBlob(data, contentType) + }, + CreateWritableStreamObjectURL: func() (string, error) { + return d.gblob.CreateWritableStream(session) + }, + WriteWritableStreamObjectURL: func(url string, data any) error { + chunk, err := normalizeStreamChunk(data) + if err != nil { + return err + } + return d.gblob.Write(url, chunk) + }, + CloseWritableStreamObjectURL: func(url string) error { + return d.gblob.CloseSource(url) + }, + AbortWritableStreamObjectURL: func(url string, reason string) error { + return d.gblob.AbortSource(url, fmt.Errorf("%s", reason)) + }, + RevokeObjectURL: func(url string) error { + return d.gblob.Revoke(url) + }, + } + e := engine.NewEngine(&engine.Config{ + ProxyConfig: d.cfg.Proxy, + StreamConfig: engineCfg, + }) + session.SetEngine(e) + return e, session +} diff --git a/pkg/download/extension_test.go b/pkg/download/extension_test.go index 5290b5f5..c372dde1 100644 --- a/pkg/download/extension_test.go +++ b/pkg/download/extension_test.go @@ -3,6 +3,7 @@ package download import ( "errors" "os" + "path/filepath" "testing" "time" @@ -46,6 +47,320 @@ func TestDownloader_InstallExtensionByFolderDevMode(t *testing.T) { }) } +func TestDownloader_Extension_GBlobBlob(t *testing.T) { + setupDownloader(func(downloader *Downloader) { + if _, err := downloader.InstallExtensionByFolder("./testdata/extensions/gblob", false); err != nil { + t.Fatal(err) + } + + rr, err := downloader.Resolve(&base.Request{ + URL: "https://example.com/blob", + }, nil) + if err != nil { + t.Fatal(err) + } + if rr.ID != "" { + t.Fatalf("expected empty resolve id for extension resource, got %q", rr.ID) + } + + doneCh := make(chan error, 1) + downloader.Listener(func(event *Event) { + if event.Key == EventKeyDone || event.Key == EventKeyError { + doneCh <- event.Err + } + }) + + dir := t.TempDir() + if _, err := downloader.CreateDirect(rr.Res.Files[0].Req, &base.Options{ + Path: dir, + Name: rr.Res.Files[0].Name, + }); err != nil { + t.Fatal(err) + } + + select { + case err := <-doneCh: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for gblob blob download") + } + + data, err := os.ReadFile(filepath.Join(dir, "hello.txt")) + if err != nil { + t.Fatal(err) + } + if string(data) != "hello world" { + t.Fatalf("unexpected blob download content: %q", string(data)) + } + }) +} + +func TestDownloader_Extension_GBlobWritableStream(t *testing.T) { + setupDownloader(func(downloader *Downloader) { + if _, err := downloader.InstallExtensionByFolder("./testdata/extensions/gblob", false); err != nil { + t.Fatal(err) + } + + rr, err := downloader.Resolve(&base.Request{ + URL: "https://example.com/stream", + }, nil) + if err != nil { + t.Fatal(err) + } + if rr.ID != "" { + t.Fatalf("expected empty resolve id for extension resource, got %q", rr.ID) + } + + doneCh := make(chan error, 1) + downloader.Listener(func(event *Event) { + if event.Key == EventKeyDone || event.Key == EventKeyError { + doneCh <- event.Err + } + }) + + dir := t.TempDir() + if _, err := downloader.CreateDirect(rr.Res.Files[0].Req, &base.Options{ + Path: dir, + Name: rr.Res.Files[0].Name, + }); err != nil { + t.Fatal(err) + } + + select { + case err := <-doneCh: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for gblob writable stream download") + } + + data, err := os.ReadFile(filepath.Join(dir, "stream.txt")) + if err != nil { + t.Fatal(err) + } + if string(data) != "line 1\nline 2\n" { + t.Fatalf("unexpected stream download content: %q", string(data)) + } + }) +} + +func TestDownloader_Extension_GBlobWritableStreamUnknownSize(t *testing.T) { + setupDownloader(func(downloader *Downloader) { + if _, err := downloader.InstallExtensionByFolder("./testdata/extensions/gblob", false); err != nil { + t.Fatal(err) + } + + rr, err := downloader.Resolve(&base.Request{ + URL: "https://example.com/stream-unknown", + }, nil) + if err != nil { + t.Fatal(err) + } + if rr.ID != "" { + t.Fatalf("expected empty resolve id for extension resource, got %q", rr.ID) + } + if got := rr.Res.Files[0].Size; got != 0 { + t.Fatalf("expected unknown size in resolve result, got %d", got) + } + + doneCh := make(chan error, 1) + downloader.Listener(func(event *Event) { + if event.Key == EventKeyDone || event.Key == EventKeyError { + doneCh <- event.Err + } + }) + + dir := t.TempDir() + id, err := downloader.CreateDirect(rr.Res.Files[0].Req, &base.Options{ + Path: dir, + Name: rr.Res.Files[0].Name, + }) + if err != nil { + t.Fatal(err) + } + + time.Sleep(60 * time.Millisecond) + task := downloader.GetTask(id) + if task == nil { + t.Fatal("task not found") + } + if task.Status == base.DownloadStatusDone { + t.Fatal("task finished before writer.close()") + } + + select { + case err := <-doneCh: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for unknown-size writable stream download") + } + + data, err := os.ReadFile(filepath.Join(dir, "stream-unknown.txt")) + if err != nil { + t.Fatal(err) + } + if string(data) != "line 1\nline 2\n" { + t.Fatalf("unexpected unknown-size stream content: %q", string(data)) + } + + task = downloader.GetTask(id) + if task == nil { + t.Fatal("task not found after completion") + } + if got := task.Meta.Res.Size; got != int64(len(data)) { + t.Fatalf("unexpected final task size: got %d want %d", got, len(data)) + } + }) +} + +func TestDownloader_Extension_GBlobWritableStreamPauseAndContinue(t *testing.T) { + setupDownloader(func(downloader *Downloader) { + if _, err := downloader.InstallExtensionByFolder("./testdata/extensions/gblob", false); err != nil { + t.Fatal(err) + } + + rr, err := downloader.Resolve(&base.Request{ + URL: "https://example.com/stream-unknown", + }, nil) + if err != nil { + t.Fatal(err) + } + + doneCh := make(chan error, 1) + downloader.Listener(func(event *Event) { + if event.Key == EventKeyDone || event.Key == EventKeyError { + doneCh <- event.Err + } + }) + + dir := t.TempDir() + id, err := downloader.CreateDirect(rr.Res.Files[0].Req, &base.Options{ + Path: dir, + Name: rr.Res.Files[0].Name, + }) + if err != nil { + t.Fatal(err) + } + + filePath := filepath.Join(dir, "stream-unknown.txt") + waitForFileSizeAtLeast(t, filePath, int64(len("line 1\n")), 2*time.Second) + + if err := downloader.Pause(&TaskFilter{IDs: []string{id}}); err != nil { + t.Fatal(err) + } + + task := downloader.GetTask(id) + if task == nil { + t.Fatal("task not found after pause") + } + if task.Status != base.DownloadStatusPause { + t.Fatalf("expected paused task, got %s", task.Status) + } + + stat, err := os.Stat(filePath) + if err != nil { + t.Fatal(err) + } + pausedSize := stat.Size() + + time.Sleep(250 * time.Millisecond) + + stat, err = os.Stat(filePath) + if err != nil { + t.Fatal(err) + } + if stat.Size() != pausedSize { + t.Fatalf("expected paused file size to remain %d, got %d", pausedSize, stat.Size()) + } + + if err := downloader.Continue(&TaskFilter{IDs: []string{id}}); err != nil { + t.Fatal(err) + } + + select { + case err := <-doneCh: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for pause/continue writable stream download") + } + + data, err := os.ReadFile(filePath) + if err != nil { + t.Fatal(err) + } + if string(data) != "line 1\nline 2\n" { + t.Fatalf("unexpected pause/continue stream content: %q", string(data)) + } + }) +} + +func TestDownloader_Extension_GBlobRecoverOnError(t *testing.T) { + setupDownloader(func(downloader *Downloader) { + if _, err := downloader.InstallExtensionByFolder("./testdata/extensions/gblob_recover", false); err != nil { + t.Fatal(err) + } + + rr, err := downloader.Resolve(&base.Request{ + URL: "https://example.com/recover", + }, nil) + if err != nil { + t.Fatal(err) + } + + doneCh := make(chan error, 1) + downloader.Listener(func(event *Event) { + if event.Key == EventKeyDone || event.Key == EventKeyError { + doneCh <- event.Err + } + }) + + dir := t.TempDir() + id, err := downloader.CreateDirect(rr.Res.Files[0].Req, &base.Options{ + Path: dir, + Name: rr.Res.Files[0].Name, + }) + if err != nil { + t.Fatal(err) + } + + select { + case err := <-doneCh: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for recovered gblob download") + } + + filePath := filepath.Join(dir, "recover.txt") + data, err := os.ReadFile(filePath) + if err != nil { + t.Fatal(err) + } + if string(data) != "recovered\n" { + t.Fatalf("unexpected recovered file content: %q", string(data)) + } + + task := downloader.GetTask(id) + if task == nil { + t.Fatal("task not found after recovery") + } + if task.Status != base.DownloadStatusDone { + t.Fatalf("expected recovered task done, got %s", task.Status) + } + if task.Meta.Req.RawURL != "https://example.com/recover" { + t.Fatalf("unexpected raw url: %q", task.Meta.Req.RawURL) + } + }) +} + func TestDownloader_InstallExtensionByGit(t *testing.T) { setupDownloader(func(downloader *Downloader) { if _, err := downloader.InstallExtensionByGit("https://github.com/GopeedLab/gopeed-extension-samples#github-release-sample"); err != nil { @@ -458,3 +773,16 @@ func setupDownloader(fn func(downloader *Downloader)) { }() fn(defaultDownloader) } + +func waitForFileSizeAtLeast(t *testing.T, path string, size int64, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + info, err := os.Stat(path) + if err == nil && info.Size() >= size { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timeout waiting for file %s size >= %d", path, size) +} diff --git a/pkg/download/extmock.go b/pkg/download/extmock.go new file mode 100644 index 00000000..99d6d680 --- /dev/null +++ b/pkg/download/extmock.go @@ -0,0 +1,87 @@ +package download + +import ( + "fmt" + "os" +) + +func NewExtMockDownloader() (*Downloader, func(), error) { + d := NewDownloader(&DownloaderConfig{ + Storage: NewMemStorage(), + }) + if err := d.Setup(); err != nil { + return nil, nil, err + } + cleanup := func() { + _ = d.Clear() + } + return d, cleanup, nil +} + +type ExtMockRuntime struct { + engine engineWrapper + session *engineSession +} + +type engineWrapper interface { + Close() + RunString(script string) (any, error) +} + +func (d *Downloader) NewExtMockRuntime() (*ExtMockRuntime, error) { + engine, session := d.newExtensionEngine() + ext := &Extension{ + Name: "extmock", + Author: "gopeed", + Title: "Gopeed ExtMock Runtime", + Version: "0.0.0", + DevMode: true, + } + gopeed := &Instance{ + Events: make(InstanceEvents), + Info: NewExtensionInfo(ext), + Logger: newInstanceLogger(ext, d.ExtensionLogger), + Settings: map[string]any{}, + Storage: &ContextStorage{ + storage: d.storage, + identity: ext.buildIdentity(), + }, + } + if err := engine.Runtime.Set("gopeed", gopeed); err != nil { + engine.Close() + return nil, err + } + return &ExtMockRuntime{ + engine: engine, + session: session, + }, nil +} + +func (r *ExtMockRuntime) Eval(script string) (any, error) { + if r == nil || r.engine == nil { + return nil, fmt.Errorf("extmock runtime not initialized") + } + return r.engine.RunString(script) +} + +func (r *ExtMockRuntime) EvalFile(path string) (any, error) { + if path == "" { + return nil, fmt.Errorf("script path is empty") + } + buf, err := os.ReadFile(path) + if err != nil { + return nil, err + } + return r.Eval(string(buf)) +} + +func (r *ExtMockRuntime) Close() { + if r == nil { + return + } + if r.engine != nil { + r.engine.Close() + } + r.engine = nil + r.session = nil +} diff --git a/pkg/download/model.go b/pkg/download/model.go index bbb57ade..a1485a6b 100644 --- a/pkg/download/model.go +++ b/pkg/download/model.go @@ -9,6 +9,7 @@ import ( "github.com/GopeedLab/gopeed/internal/fetcher" "github.com/GopeedLab/gopeed/internal/protocol/bt" "github.com/GopeedLab/gopeed/internal/protocol/ed2k" + "github.com/GopeedLab/gopeed/internal/protocol/gblob" "github.com/GopeedLab/gopeed/internal/protocol/http" "github.com/GopeedLab/gopeed/pkg/base" "github.com/GopeedLab/gopeed/pkg/util" @@ -165,6 +166,7 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig { } if len(cfg.FetchManagers) == 0 { cfg.FetchManagers = []fetcher.FetcherManager{ + new(gblob.FetcherManager), new(http.FetcherManager), new(bt.FetcherManager), new(ed2k.FetcherManager), diff --git a/pkg/download/testdata/extensions/gblob/index.js b/pkg/download/testdata/extensions/gblob/index.js new file mode 100644 index 00000000..f79c1d72 --- /dev/null +++ b/pkg/download/testdata/extensions/gblob/index.js @@ -0,0 +1,62 @@ +gopeed.events.onResolve(async function (ctx) { + if (ctx.req.url.endsWith("/blob")) { + const blob = new Blob(["hello world"], {type: "text/plain"}); + const url = URL.createObjectURL(blob); + ctx.res = { + name: "gblob-blob", + files: [ + { + name: "hello.txt", + size: 11, + req: { + url, + } + } + ] + }; + return; + } + + const transform = new TransformStream(); + const writer = transform.writable.getWriter(); + const url = URL.createObjectURL(transform.writable); + + if (ctx.req.url.endsWith("/stream-unknown")) { + (async () => { + await writer.write(new TextEncoder().encode("line 1\n")); + await new Promise((resolve) => setTimeout(resolve, 120)); + await writer.write(new TextEncoder().encode("line 2\n")); + await writer.close(); + })(); + + ctx.res = { + name: "gblob-stream-unknown", + files: [ + { + name: "stream-unknown.txt", + req: { + url, + } + } + ] + }; + return; + } + + await writer.write(new TextEncoder().encode("line 1\n")); + await writer.write(new TextEncoder().encode("line 2\n")); + await writer.close(); + + ctx.res = { + name: "gblob-stream", + files: [ + { + name: "stream.txt", + size: 14, + req: { + url, + } + } + ] + }; +}); diff --git a/pkg/download/testdata/extensions/gblob/manifest.json b/pkg/download/testdata/extensions/gblob/manifest.json new file mode 100644 index 00000000..87984978 --- /dev/null +++ b/pkg/download/testdata/extensions/gblob/manifest.json @@ -0,0 +1,16 @@ +{ + "name": "gblob", + "title": "gopeed extension gblob test", + "version": "0.0.1", + "scripts": [ + { + "event": "onResolve", + "match": { + "urls": [ + "*://example.com/*" + ] + }, + "entry": "index.js" + } + ] +} diff --git a/pkg/download/testdata/extensions/gblob_recover/index.js b/pkg/download/testdata/extensions/gblob_recover/index.js new file mode 100644 index 00000000..d5b22053 --- /dev/null +++ b/pkg/download/testdata/extensions/gblob_recover/index.js @@ -0,0 +1,42 @@ +gopeed.events.onResolve(async function (ctx) { + if (!ctx.req.url.endsWith("/recover")) { + return; + } + + const transform = new TransformStream(); + const writer = transform.writable.getWriter(); + const url = URL.createObjectURL(transform.writable); + + (async () => { + await writer.write(new TextEncoder().encode("stale\n")); + await new Promise((resolve) => setTimeout(resolve, 50)); + await writer.abort("expired"); + })(); + + ctx.res = { + name: "gblob-recover", + files: [ + { + name: "recover.txt", + req: { + url, + } + } + ] + }; +}); + +gopeed.events.onError(async function (ctx) { + const req = ctx.task.meta.req; + if (!req.rawUrl || !req.rawUrl.endsWith("/recover")) { + return; + } + req.labels = req.labels || {}; + if (req.labels.recovered === "true") { + return; + } + + req.labels.recovered = "true"; + req.url = URL.createObjectURL(new Blob(["recovered\n"], { type: "text/plain" })); + ctx.task.continue(); +}); diff --git a/pkg/download/testdata/extensions/gblob_recover/manifest.json b/pkg/download/testdata/extensions/gblob_recover/manifest.json new file mode 100644 index 00000000..bfda3f44 --- /dev/null +++ b/pkg/download/testdata/extensions/gblob_recover/manifest.json @@ -0,0 +1,25 @@ +{ + "name": "gblob-recover", + "title": "gopeed extension gblob recover test", + "version": "0.0.1", + "scripts": [ + { + "event": "onResolve", + "match": { + "urls": [ + "*://example.com/*" + ] + }, + "entry": "index.js" + }, + { + "event": "onError", + "match": { + "urls": [ + "*://example.com/*" + ] + }, + "entry": "index.js" + } + ] +}