diff --git a/pkg/download/engine/engine.go b/pkg/download/engine/engine.go index 147f1f98..96420225 100644 --- a/pkg/download/engine/engine.go +++ b/pkg/download/engine/engine.go @@ -160,6 +160,8 @@ func (e *Engine) addCleanup(cleanup func()) { } type Config struct { + // TempDir supplies an app-writable directory for FFmpeg stream buffers. + TempDir string // HTTPUserAgent is a snapshot of the downloader HTTP default. Nil leaves standalone engines unchanged. HTTPUserAgent *string ProxyConfig *base.DownloaderProxyConfig @@ -207,6 +209,7 @@ func NewEngine(cfg *Config) *Engine { return } if err := ffmpegapi.Enable(runtime, loop, &ffmpegapi.Config{ + TempDir: cfg.TempDir, ProxyHandler: cfg.ProxyConfig.ToHandler(), DefaultUserAgent: cfg.HTTPUserAgent, RegisterCleanup: engine.addCleanup, diff --git a/pkg/download/engine/ffmpeg/input_test.go b/pkg/download/engine/ffmpeg/input_test.go index 2f3e1ad6..dc0aa8f8 100644 --- a/pkg/download/engine/ffmpeg/input_test.go +++ b/pkg/download/engine/ffmpeg/input_test.go @@ -8,6 +8,7 @@ import ( "net/http" "net/http/httptest" "os" + "path/filepath" "strings" "sync/atomic" "testing" @@ -117,7 +118,7 @@ func TestHTTPSequentialAndValidation(t *testing.T) { } func TestStreamRing(t *testing.T) { - s := NewStreamInput(context.Background()) + s := NewStreamInput(context.Background(), "") defer s.Close() // Start near the boundary to exercise wrap-around without a huge fixture. s.read = StreamBufferLimit - 3 @@ -157,7 +158,7 @@ func TestOutputArguments(t *testing.T) { func TestStreamBackpressureAndCancel(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - s := NewStreamInput(ctx) + s := NewStreamInput(ctx, "") defer s.Close() f, err := os.CreateTemp(t.TempDir(), "ring") if err != nil { @@ -197,3 +198,38 @@ func TestStreamBackpressureAndCancel(t *testing.T) { t.Fatal("writer did not cancel") } } + +// A mobile app can write to its storage directory even when the OS default +// temporary directory is unavailable. The buffer must also be removed on close. +func TestStreamInputAppTempDir(t *testing.T) { + root := t.TempDir() + unavailable := filepath.Join(root, "not-a-directory") + if err := os.WriteFile(unavailable, []byte("blocked"), 0600); err != nil { + t.Fatal(err) + } + t.Setenv("TMPDIR", unavailable) + t.Setenv("TMP", unavailable) + t.Setenv("TEMP", unavailable) + dir := filepath.Join(root, "app", "temp", "ffmpeg") + s := NewStreamInput(context.Background(), dir) + t.Cleanup(func() { s.Close() }) + data := []byte("media stream") + if _, err := s.Write(data); err != nil { + t.Fatal(err) + } + if filepath.Dir(s.file.Name()) != dir { + t.Fatalf("unexpected buffer: %s", s.file.Name()) + } + s.End(nil) + got, err := io.ReadAll(s) + if err != nil || !bytes.Equal(got, data) { + t.Fatalf("read %q, error %v", got, err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + entries, err := os.ReadDir(dir) + if err != nil || len(entries) != 0 { + t.Fatalf("buffer cleanup: %v, %v", entries, err) + } +} diff --git a/pkg/download/engine/ffmpeg/runtime_test.go b/pkg/download/engine/ffmpeg/runtime_test.go index ee57c180..d7ea11fe 100644 --- a/pkg/download/engine/ffmpeg/runtime_test.go +++ b/pkg/download/engine/ffmpeg/runtime_test.go @@ -128,7 +128,7 @@ func TestWebMToMP4(t *testing.T) { if err != nil { t.Fatal(err) } - inputs[i] = NewStreamInput(ctx) + inputs[i] = NewStreamInput(ctx, "") defer inputs[i].Close() if _, err = inputs[i].Write(data); err != nil { t.Fatal(err) @@ -205,7 +205,7 @@ func TestWASMMerge(t *testing.T) { }) } // A malformed input must produce an error, not successful empty output. - s := NewStreamInput(context.Background()) + s := NewStreamInput(context.Background(), "") defer s.Close() s.Write([]byte("invalid")) s.End(nil) diff --git a/pkg/download/engine/ffmpeg/stream.go b/pkg/download/engine/ffmpeg/stream.go index 14aa656d..a2c1208c 100644 --- a/pkg/download/engine/ffmpeg/stream.go +++ b/pkg/download/engine/ffmpeg/stream.go @@ -17,6 +17,7 @@ const StreamBufferLimit = 64 * 1024 * 1024 type StreamInput struct { ctx context.Context + tempDir string mu sync.Mutex file *os.File read, written int64 @@ -26,8 +27,8 @@ type StreamInput struct { space chan struct{} } -func NewStreamInput(ctx context.Context) *StreamInput { - return &StreamInput{ctx: ctx, wake: make(chan struct{}, 1), space: make(chan struct{}, 1)} +func NewStreamInput(ctx context.Context, tempDir string) *StreamInput { + return &StreamInput{ctx: ctx, tempDir: tempDir, wake: make(chan struct{}, 1), space: make(chan struct{}, 1)} } func (s *StreamInput) Write(p []byte) (int, error) { @@ -70,7 +71,13 @@ func (s *StreamInput) Write(p []byte) (int, error) { return 0, nil } if s.file == nil { - f, err := os.CreateTemp("", "gopeed-ffmpeg-*") + // Android apps cannot write to Go's default /data/local/tmp. + if s.tempDir != "" { + if err := os.MkdirAll(s.tempDir, 0700); err != nil { + return 0, err + } + } + f, err := os.CreateTemp(s.tempDir, "gopeed-ffmpeg-*") if err != nil { return 0, err } diff --git a/pkg/download/engine/inject/ffmpeg/module.go b/pkg/download/engine/inject/ffmpeg/module.go index 42dec9cb..2488d2d1 100644 --- a/pkg/download/engine/inject/ffmpeg/module.go +++ b/pkg/download/engine/inject/ffmpeg/module.go @@ -22,6 +22,7 @@ import ( var script string type Config struct { + TempDir string DefaultUserAgent *string ProxyHandler func(*http.Request) (*url.URL, error) RegisterCleanup func(func()) @@ -122,7 +123,7 @@ func Enable(vm *goja.Runtime, loop *eventloop.EventLoop, cfg *Config) error { j := &job{ctx: ctx, cancel: cancel, output: r} for i, source := range []*media.HTTPSource{opts.Video, opts.Audio} { if source == nil { - j.streams[i] = media.NewStreamInput(ctx) + j.streams[i] = media.NewStreamInput(ctx, cfg.TempDir) } } id := fmt.Sprintf("ffmpeg-%d", next.Add(1)) diff --git a/pkg/download/extension_engine_session.go b/pkg/download/extension_engine_session.go index 4f248c88..4fa6819b 100644 --- a/pkg/download/extension_engine_session.go +++ b/pkg/download/extension_engine_session.go @@ -3,6 +3,7 @@ package download import ( "context" "io" + "path/filepath" "strings" "sync" @@ -126,7 +127,12 @@ func (d *Downloader) newExtensionEngine() (*engine.Engine, *engineSession) { }{UserAgent: httpProtocol.DefaultUserAgent} d.getProtocolConfig("http", &httpConfig) userAgent := strings.TrimSpace(httpConfig.UserAgent) + tempDir := "" + if d.cfg.StorageDir != "" { + tempDir = filepath.Join(d.cfg.StorageDir, "temp", "ffmpeg") + } e := engine.NewEngine(&engine.Config{ + TempDir: tempDir, HTTPUserAgent: &userAgent, ProxyConfig: d.cfg.Proxy, StreamConfig: engineCfg,