同步上游 0921 版 zip(新增 emby 播放等 20 处变更)

This commit is contained in:
yvan
2026-09-17 19:10:04 +08:00
parent 07dc0f1059
commit 10baea5a87
22 changed files with 731 additions and 274 deletions
+247
View File
@@ -0,0 +1,247 @@
package app
import (
"context"
"errors"
"math"
"net/http"
"net/url"
"time"
)
func (app *UIApp) activePlaybackSessionsLocked() int {
count := 0
for _, session := range app.playbacks {
if !session.emby || session.mediaWaiters > 0 {
count++
}
}
return count
}
func (app *UIApp) trimEmbyPlaybacksLocked(reserve int) []context.CancelFunc {
limit := app.mediaResources().snapshot().MaxSessions - reserve
var cancellations []context.CancelFunc
for {
count := 0
var oldest *playbackSession
for _, session := range app.playbacks {
if !session.emby || session.native == nil && (session.media == nil || session.media.generated == nil) {
continue
}
count++
if session.mediaWaiters == 0 && session.state != "opening" && (oldest == nil || session.expires.Before(oldest.expires)) {
oldest = session
}
}
if count <= limit || oldest == nil {
return cancellations
}
if oldest.cancel != nil {
cancellations = append(cancellations, oldest.cancel)
}
if oldest.media != nil {
oldest.embyProcessing = oldest.media.plan.Processing
}
oldest.cancel, oldest.native, oldest.media = nil, nil, nil
oldest.mediaReady, oldest.mediaError, oldest.state = nil, nil, "ready"
}
}
func (app *UIApp) releaseEmbyPlayback(session *playbackSession) {
app.playbackMu.Lock()
session.mediaWaiters--
var cancellations []context.CancelFunc
if app.playbacks[session.id] == session && session.mediaWaiters == 0 {
if session.state == "opening" || session.mediaError != nil {
delete(app.playbacks, session.id)
session.timer.Stop()
if session.cancel != nil {
cancellations = append(cancellations, session.cancel)
}
} else {
app.touchPlaybackLocked(session)
}
}
cancellations = append(cancellations, app.trimEmbyPlaybacksLocked(0)...)
app.playbackMu.Unlock()
for _, cancel := range cancellations {
cancel()
}
}
func (app *UIApp) writeEmbyPlaybackError(writer http.ResponseWriter, request *http.Request, status int, message string) {
if status == http.StatusTooManyRequests {
writer.Header().Set("Retry-After", "2")
}
id := request.URL.Query().Get("id")
app.downloader.recordDiagnostic(diagnosticEvent{Event: "emby.playback_failed", HTTPStatus: status,
Source: sourceFromDramaID(id), DramaID: id, Message: message})
writeJSON(writer, status, map[string]string{"error": message})
}
func (app *UIApp) acquireEmbyPlayback(writer http.ResponseWriter, request *http.Request, id, chapter, sessionID string, legacy bool) *playbackSession {
owner := request.URL.Query().Get("account")
app.playbackMu.Lock()
var session *playbackSession
if sessionID != "" {
session = app.playbacks[sessionID]
if session == nil || !session.emby || session.viewer != nil || session.accountID != owner || session.dramaID != id || len(session.tasks) != 1 || session.tasks[0].Chapter.ID != chapter {
app.playbackMu.Unlock()
app.writeEmbyPlaybackError(writer, request, http.StatusGone, "播放已过期,请重新播放")
return nil
}
} else {
for _, candidate := range app.playbacks {
if candidate.emby && candidate.embyLegacy == legacy && candidate.accountID == owner && candidate.dramaID == id && len(candidate.tasks) == 1 && candidate.tasks[0].Chapter.ID == chapter && candidate.mediaError == nil {
session = candidate
break
}
}
}
if (session == nil || session.mediaWaiters == 0) && app.activePlaybackSessionsLocked() >= app.mediaResources().snapshot().MaxSessions {
app.playbackMu.Unlock()
app.writeEmbyPlaybackError(writer, request, http.StatusTooManyRequests, "同时播放数量已达上限,请稍后再试")
return nil
}
if session == nil {
sessionID = randomHex(24)
if sessionID == "" {
app.playbackMu.Unlock()
app.writeEmbyPlaybackError(writer, request, http.StatusInternalServerError, "无法创建播放会话")
return nil
}
session = &playbackSession{id: sessionID, accountID: owner, dramaID: id, emby: true, embyLegacy: legacy, run: 1,
tasks: []Task{{DramaID: id, Chapter: Chapter{ID: chapter}}}}
if app.playbacks == nil {
app.playbacks = make(map[string]*playbackSession)
}
app.playbacks[sessionID] = session
session.timer = time.AfterFunc(playbackIdleTimeout, func() { app.expirePlayback(sessionID) })
}
session.mediaWaiters++
app.touchPlaybackLocked(session)
var cancellations []context.CancelFunc
if session.mediaWaiters == 1 {
failed := sessionID == "" && session.media != nil && (session.media.failed.Load() || session.media.ctx.Err() != nil || session.media.proxy != nil && session.media.proxy.Err() != nil)
if session.native != nil {
state, err := session.native.state()
failed = failed || err != nil || state == "stopped"
}
if failed {
if session.cancel != nil {
cancellations = append(cancellations, session.cancel)
}
session.cancel, session.native, session.media, session.mediaReady, session.mediaError = nil, nil, nil, nil, nil
}
}
var parent context.Context
if session.mediaReady == nil {
cancellations = append(cancellations, app.trimEmbyPlaybacksLocked(1)...)
parent, session.cancel = context.WithCancel(context.Background())
session.mediaReady, session.state = make(chan struct{}), "opening"
}
ready := session.mediaReady
app.playbackMu.Unlock()
for _, cancel := range cancellations {
cancel()
}
if parent != nil {
query := url.Values{"id": {id}, "chapter": {chapter}, "key": {request.URL.Query().Get("key")}}
if owner != "" {
query.Set("account", owner)
}
go app.prepareEmbyPlayback(parent, session, ready, query.Encode(), request.Header.Get("Origin"))
}
select {
case <-ready:
case <-request.Context().Done():
app.releaseEmbyPlayback(session)
return nil
}
app.playbackMu.Lock()
err := session.mediaError
if err == nil && (app.playbacks[session.id] != session || session.media == nil && session.native == nil) {
err = errors.New("播放会话已结束")
}
app.playbackMu.Unlock()
if err != nil {
app.releaseEmbyPlayback(session)
app.writeEmbyPlaybackError(writer, request, http.StatusBadGateway, "准备 Emby 播放失败:"+app.redactError(err))
return nil
}
return session
}
func (app *UIApp) prepareEmbyPlayback(parent context.Context, session *playbackSession, ready chan struct{}, query, origin string) {
ctx, cancel := context.WithTimeout(parent, 90*time.Second)
defer cancel()
task, downloadID, err := app.embyTask(ctx, session.dramaID, session.tasks[0].Chapter.ID)
var media *playbackMediaSession
var native *playbackNative
var duration float64
if err == nil && !session.embyLegacy {
media, err = app.resolveMediaSession(parent, ctx, task, downloadID, 0)
if err == nil {
base := "/api/emby/media/" + session.id + "/"
mode := "auto"
if session.embyProcessing == "remux" {
mode = "compatible"
} else if session.embyProcessing == "audio" {
mode = "audio"
}
err = app.planMedia(ctx, media, playbackClientCapabilities{MP4: true, NativeHLS: true}, mode, sourceFromDramaID(task.DramaID), origin, base, query)
if err == nil {
duration = media.plan.Duration
if media.plan.Player == "legacy" {
media.Close()
media = nil
}
}
}
}
if err == nil && media == nil {
native = newPlaybackNative(app, parent, task, downloadID, 0, 0, false)
native.start()
_, err = native.segment(ctx, 0)
if err == nil {
_, duration, _ = native.metadata()
if duration <= 0 || duration > 24*60*60 || math.IsNaN(duration) || math.IsInf(duration, 0) {
err = errors.New("未取得有效播放时长")
}
}
}
app.playbackMu.Lock()
if err == nil && (app.playbacks[session.id] != session || session.mediaReady != ready || ctx.Err() != nil) {
err = context.Canceled
}
if err == nil {
previous := session.cancel
session.cancel = func() {
previous()
if media != nil {
media.Close()
}
if native != nil {
native.Close()
}
}
session.tasks, session.media, session.native, session.duration, session.state = []Task{task}, media, native, duration, "streaming"
if media != nil {
media.plan.Run = session.run
}
app.touchPlaybackLocked(session)
} else {
session.mediaError, session.state = err, "failed"
}
close(ready)
app.playbackMu.Unlock()
if err != nil {
if media != nil {
media.Close()
}
if native != nil {
native.Close()
}
}
}
+338
View File
@@ -0,0 +1,338 @@
package app
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"net/url"
"os/exec"
"strings"
"sync"
"testing"
"time"
)
func embyScanFixture(t *testing.T, optimized bool) (*UIApp, func(int) string) {
t.Helper()
app, fixture := nativePlaybackFixture(t, "4", "160x90")
app.closePlayback(fixture.id)
app.mediaResources().settings.Enabled = optimized
app.mediaResources().settings.MaxSessions = 1
for index := 1; index <= 8; index++ {
task := fixture.tasks[0]
task.Index, task.Total = index, 8
task.Chapter.ID = fmt.Sprintf("%s:%d", task.DramaID, index)
task.Chapter.CurrentEpisode = rawEpisode(index)
id := fmt.Sprintf("scan-%d", index)
app.tasks[id] = &UITask{ID: id, DramaID: task.DramaID, Status: uiStatusSuccess, Path: task.OutPath, Source: task}
app.taskOrder = append(app.taskOrder, id)
}
key, err := app.embySigningKey(true)
if err != nil {
t.Fatal(err)
}
return app, func(index int) string {
task := app.tasks[fmt.Sprintf("scan-%d", index)].Source
query := url.Values{"id": {task.DramaID}, "chapter": {task.Chapter.ID}, "key": {embyToken(key, task.DramaID, task.Chapter.ID)}}
return "/api/emby/stream.m3u8?" + query.Encode()
}
}
func embyRequest(app *UIApp, method, address string) *httptest.ResponseRecorder {
writer := httptest.NewRecorder()
app.routes().ServeHTTP(writer, httptest.NewRequest(method, "http://localhost"+address, nil))
return writer
}
func embyFirstSegment(t *testing.T, playlist string) string {
t.Helper()
for _, line := range strings.Split(playlist, "\n") {
if strings.HasPrefix(line, "segment.ts?") {
return "/api/emby/" + line
}
}
t.Fatal("missing signed HLS segment")
return ""
}
func TestEmbySerialProbesReleaseSlotsAndReusePlayback(t *testing.T) {
probe, err := exec.LookPath("ffprobe")
if err != nil {
t.Skip("ffprobe unavailable")
}
for _, optimized := range []bool{false, true} {
t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) {
app, address := embyScanFixture(t, optimized)
server := httptest.NewServer(app.routes())
defer server.Close()
first := embyRequest(app, http.MethodGet, address(1))
firstBody, firstLocation := first.Body.String(), first.Header().Get("Location")
for index := 1; index <= 8; index++ {
if head := embyRequest(app, http.MethodHead, address(index)); head.Code != 200 || head.Body.Len() != 0 {
t.Fatal("HEAD failed", head.Code)
}
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
output, err := exec.CommandContext(ctx, probe, "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", server.URL+address(index)).CombinedOutput()
cancel()
if err != nil || !strings.Contains(string(output), "4.000000") {
t.Fatalf("serial probe %d failed: %v %s", index, err, output)
}
app.playbackMu.Lock()
active := app.activePlaybackSessionsLocked()
cached := 0
for _, session := range app.playbacks {
if session.native != nil {
cached++
}
}
app.playbackMu.Unlock()
if active != 0 || cached > 1 {
t.Fatal("finished probe retained a slot or unbounded media cache", active, cached)
}
}
if !optimized {
part := embyRequest(app, http.MethodGet, embyFirstSegment(t, firstBody))
if part.Code != 200 || part.Body.Len() < 188 || part.Body.Bytes()[0] != 0x47 {
t.Fatal("evicting an idle cache broke an existing HLS URL", part.Code, part.Body.String())
}
}
repeated := embyRequest(app, http.MethodGet, address(1))
if repeated.Code != first.Code || repeated.Body.String() != firstBody || repeated.Header().Get("Location") != firstLocation {
t.Fatal("repeated episode did not reuse its prepared session", repeated.Code)
}
web := historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"})
if web.Code != 200 {
t.Fatal("completed Emby scans blocked a new browser playback", web.Code, web.Body.String())
}
})
}
}
type embyBlockingWriter struct {
*httptest.ResponseRecorder
entered chan struct{}
resume chan struct{}
once sync.Once
}
func (writer *embyBlockingWriter) Write(body []byte) (int, error) {
writer.once.Do(func() {
close(writer.entered)
<-writer.resume
})
return writer.ResponseRecorder.Write(body)
}
func TestEmbyActiveRequestsShareLimitAndReleaseOnCompletion(t *testing.T) {
for _, optimized := range []bool{false, true} {
t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) {
app, address := embyScanFixture(t, optimized)
first := embyRequest(app, http.MethodGet, address(1))
asset := first.Header().Get("Location")
if !optimized {
asset = embyFirstSegment(t, first.Body.String())
}
writer := &embyBlockingWriter{ResponseRecorder: httptest.NewRecorder(), entered: make(chan struct{}), resume: make(chan struct{})}
done := make(chan struct{})
go func() {
defer close(done)
app.routes().ServeHTTP(writer, httptest.NewRequest(http.MethodGet, "http://localhost"+asset, nil))
}()
var resume sync.Once
defer resume.Do(func() { close(writer.resume); <-done })
select {
case <-writer.entered:
case <-time.After(10 * time.Second):
t.Fatal("media request never reached the response writer")
}
if second := embyRequest(app, http.MethodGet, address(2)); second.Code != 429 || second.Header().Get("Retry-After") == "" {
t.Fatal("active requests bypassed the configured limit", second.Code)
}
if same := embyRequest(app, http.MethodGet, address(1)); same.Code != first.Code {
t.Fatal("concurrent requests for the same episode were counted twice", same.Code)
}
web := historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"})
if web.Code != 429 {
t.Fatal("browser playback bypassed an active Emby request", web.Code)
}
resume.Do(func() { close(writer.resume); <-done })
if writer.Code != 200 {
t.Fatal("the original media request was interrupted", writer.Code)
}
if second := embyRequest(app, http.MethodGet, address(2)); second.Code != first.Code {
t.Fatal("completed request did not release its slot", second.Code)
}
web = historyJSONRequest(t, app, app.handlePlaybackOpen, "/api/ui/playback/open", map[string]string{"taskId": "scan-1"})
if web.Code != 200 {
t.Fatal("released Emby slot was not available to the browser", web.Code, web.Body.String())
}
var opened struct{ Session string }
if err := json.Unmarshal(web.Body.Bytes(), &opened); err != nil {
t.Fatal(err)
}
app.closePlayback(opened.Session)
})
}
}
func TestEmbyCanceledProbeReleasesPendingPreparation(t *testing.T) {
started, stopped := make(chan struct{}), make(chan struct{})
var startOnce, stopOnce sync.Once
d := rankingTestDownloader(t, func(request *http.Request) (*http.Response, error) {
startOnce.Do(func() { close(started) })
<-request.Context().Done()
stopOnce.Do(func() { close(stopped) })
return nil, request.Context().Err()
})
app := &UIApp{downloader: d, cfg: d.cfg}
t.Cleanup(app.closePlaybacks)
key, err := app.embySigningKey(true)
if err != nil {
t.Fatal(err)
}
id, chapter := "hongguo:7000000000000000001", "hongguo:7000000000000000001:1"
query := url.Values{"id": {id}, "chapter": {chapter}, "key": {embyToken(key, id, chapter)}}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
done := make(chan struct{})
go func() {
defer close(done)
request := httptest.NewRequest(http.MethodGet, "http://localhost/api/emby/stream.m3u8?"+query.Encode(), nil).WithContext(ctx)
app.handleEmbyStream(httptest.NewRecorder(), request)
}()
select {
case <-started:
case <-time.After(5 * time.Second):
t.Fatal("preparation did not start")
}
other, cancelOther := context.WithCancel(context.Background())
defer cancelOther()
otherDone := make(chan struct{})
go func() {
defer close(otherDone)
request := httptest.NewRequest(http.MethodGet, "http://localhost/api/emby/stream.m3u8?"+query.Encode(), nil).WithContext(other)
app.handleEmbyStream(httptest.NewRecorder(), request)
}()
deadline := time.Now().Add(5 * time.Second)
for {
app.playbackMu.Lock()
waiters := 0
for _, session := range app.playbacks {
waiters += session.mediaWaiters
}
app.playbackMu.Unlock()
if waiters == 2 {
break
}
if time.Now().After(deadline) {
t.Fatal("second request did not join the preparation")
}
time.Sleep(time.Millisecond)
}
cancel()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("first canceled request did not finish")
}
select {
case <-stopped:
t.Fatal("one canceled client stopped another client's preparation")
default:
}
cancelOther()
for _, signal := range []chan struct{}{otherDone, stopped} {
select {
case <-signal:
case <-time.After(5 * time.Second):
t.Fatal("canceling the probe left its preparation running")
}
}
app.playbackMu.Lock()
defer app.playbackMu.Unlock()
if len(app.playbacks) != 0 {
t.Fatal("canceled probe retained a playback session")
}
}
func TestEmbyEvictedRemuxRestoresExistingSegmentURLs(t *testing.T) {
app, address := embyScanFixture(t, true)
first := embyRequest(app, http.MethodGet, address(1))
if first.Code != http.StatusFound {
t.Fatal(first.Code, first.Body.String())
}
parsed, _ := url.Parse(first.Header().Get("Location"))
base := strings.TrimSuffix(parsed.Path, "file.mp4")
sessionID := strings.Split(base, "/")[4]
app.playbackMu.Lock()
session := app.playbacks[sessionID]
media := session.media
app.playbackMu.Unlock()
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
if err := app.generatePlayback(ctx, media, "remux", base, parsed.RawQuery); err != nil {
t.Fatal(err)
}
segment := media.generated.assetURL(media.generated.segments[0].Name)
before := embyRequest(app, http.MethodGet, segment)
if before.Code != 200 {
t.Fatal("prepared remux was unreadable", before.Code)
}
if next := embyRequest(app, http.MethodGet, address(2)); next.Code != http.StatusFound {
t.Fatal("could not prepare another episode", next.Code)
}
if media.ctx.Err() == nil {
t.Fatal("idle remux cache was not released")
}
after := embyRequest(app, http.MethodGet, segment)
if after.Code != 200 || !bytes.Equal(before.Body.Bytes(), after.Body.Bytes()) {
t.Fatal("reclaiming cached files changed or invalidated an existing segment", after.Code)
}
}
func TestEmbyPlaybackCacheRemainsAccountScoped(t *testing.T) {
for _, optimized := range []bool{false, true} {
t.Run(fmt.Sprintf("original=%t", optimized), func(t *testing.T) {
app, address := embyScanFixture(t, optimized)
store := app.browserViewers().accountStore()
store.mu.Lock()
for _, owner := range []string{"owner-a", "owner-b"} {
store.state.Accounts[owner] = accountRecord{ID: owner, Sources: []string{sourceHongguo}}
}
store.mu.Unlock()
key, err := app.embySigningKey(false)
if err != nil {
t.Fatal(err)
}
sign := func(path, owner string) string {
parsed, _ := url.Parse(path)
query := parsed.Query()
query.Set("account", owner)
query.Set("key", embyToken(key, query.Get("id"), query.Get("chapter"), owner))
parsed.RawQuery = query.Encode()
return parsed.String()
}
first := embyRequest(app, http.MethodGet, sign(address(1), "owner-a"))
asset := first.Header().Get("Location")
if !optimized {
asset = embyFirstSegment(t, first.Body.String())
}
if forged := embyRequest(app, http.MethodGet, sign(asset, "owner-b")); forged.Code != http.StatusGone {
t.Fatal("valid credentials for another account reused a private cache", forged.Code)
}
second := embyRequest(app, http.MethodGet, sign(address(1), "owner-b"))
if second.Code != first.Code {
t.Fatal("second account could not prepare its own session", second.Code)
}
app.playbackMu.Lock()
count := len(app.playbacks)
app.playbackMu.Unlock()
if count != 2 {
t.Fatal("accounts shared a playback session", count)
}
})
}
}
Binary file not shown.
+9 -3
View File
@@ -45,7 +45,7 @@ func playbackGeneratedArgs(media providerMedia, input, directory, processing str
}
return append(args, "-avoid_negative_ts", "make_zero", "-f", "hls", "-hls_time", "2", "-hls_list_size", "0",
"-hls_playlist_type", "vod", "-hls_segment_type", "fmp4", "-hls_fmp4_init_filename", "init.mp4",
"-hls_flags", "temp_file", "-hls_segment_filename", filepath.Join(directory, "%06d.m4s"), filepath.Join(directory, "index.m3u8"))
"-hls_flags", "temp_file", "-hls_segment_filename", filepath.ToSlash(filepath.Join(directory, "%06d.m4s")), filepath.ToSlash(filepath.Join(directory, "index.m3u8")))
}
func (app *UIApp) generatePlayback(ctx context.Context, media *playbackMediaSession, processing, base, query string) error {
@@ -209,8 +209,14 @@ func (generated *playbackGenerated) index(playlist string) error {
}
for name := range generated.files {
info, err := os.Stat(filepath.Join(generated.directory, name))
if err != nil || !info.Mode().IsRegular() || info.Size() <= 0 {
return errors.New("媒体分片未完整生成")
if err != nil {
return fmt.Errorf("媒体分片未完整生成(%s):%w", name, err)
}
if !info.Mode().IsRegular() {
return fmt.Errorf("媒体分片未完整生成(%s 不是普通文件)", name)
}
if info.Size() <= 0 {
return fmt.Errorf("媒体分片未完整生成(%s 为空文件)", name)
}
}
generated.playlist = strings.Join(lines, "\n")
+28 -2
View File
@@ -21,6 +21,7 @@ import (
func plannedFixture(t *testing.T, app *UIApp, episode, version int, mode string) playbackMediaPlan {
t.Helper()
app.mediaResources().settings.Enabled = true
body, _ := json.Marshal(map[string]any{"session": "fixture", "episode": episode, "version": version, "start": 1.25, "mode": mode,
"client": playbackClientCapabilities{MP4: true, NativeHLS: true, HlsJS: true, Video: []string{"h264"}, Audio: []string{"aac"}}})
request := httptest.NewRequest(http.MethodPost, "http://localhost/api/ui/playback/plan", bytes.NewReader(body))
@@ -165,6 +166,28 @@ func playbackDecodedFrames(t *testing.T, ffmpeg, address string) []string {
return frames
}
func TestPlaybackGeneratedInitializationPathMatchesPlaylist(t *testing.T) {
directory := filepath.Join(t.TempDir(), "media cache")
args := playbackGeneratedArgs(providerMedia{}, "input.mp4", directory, "remux")
playlist := args[len(args)-1]
var init, segment string
for i := 0; i+1 < len(args); i++ {
switch args[i] {
case "-hls_fmp4_init_filename":
init = args[i+1]
case "-hls_segment_filename":
segment = args[i+1]
}
}
if init != "init.mp4" || strings.ContainsAny(playlist+segment, "\\") || !strings.HasSuffix(segment, "/%06d.m4s") {
t.Fatalf("FFmpeg output paths are not portable: %q", args)
}
resolvedInit := playlist[:strings.LastIndex(playlist, "/")+1] + init
if resolvedInit != filepath.ToSlash(filepath.Join(directory, "init.mp4")) {
t.Fatalf("FFmpeg would write its initialization segment outside the media directory: %q", resolvedInit)
}
}
func TestPlaybackGeneratedPreservesLongGOPAndAudioOnlyVideo(t *testing.T) {
app, fixture := nativePlaybackFixture(t, "10", "320x180")
ffmpeg := app.downloader.cfg.FFmpeg
@@ -296,14 +319,16 @@ func TestPlaybackOriginalHLSURIHeadersAndByteRanges(t *testing.T) {
func TestPlaybackResourceBudgetsCancellationAndPersistence(t *testing.T) {
app, admin := administratorFixture(t)
settings := defaultPlaybackSettings()
settings.Enabled = true
settings.MaxVideoTranscodes, settings.MaxMediaRequests = 0, 1
response := admin.request(t, http.MethodPost, "/api/ui/admin/playback", settings)
if response.Code != 200 {
t.Fatal(response.Code, response.Body.String())
}
restarted := &UIApp{cfg: app.cfg}
if restarted.mediaResources().snapshot().MaxVideoTranscodes != 0 {
t.Fatal("disabled video processing did not persist")
saved := restarted.mediaResources().snapshot()
if !saved.Enabled || saved.MaxVideoTranscodes != 0 {
t.Fatal("explicit playback settings did not persist")
}
resources := app.mediaResources()
if _, err := resources.acquire(context.Background(), "video", false); err == nil {
@@ -338,6 +363,7 @@ func TestPlaybackResourceBudgetsCancellationAndPersistence(t *testing.T) {
func TestEmbyOriginalMP4SignedRangeWithoutCookies(t *testing.T) {
app, fixture := nativePlaybackFixture(t, "4", "320x180")
app.mediaResources().settings.Enabled = true
key, err := app.embySigningKey(true)
if err != nil {
t.Fatal(err)
+1 -1
View File
@@ -35,7 +35,7 @@ type playbackResources struct {
}
func defaultPlaybackSettings() playbackSettings {
return playbackSettings{Enabled: true, DirectSources: []string{}, MaxSessions: 4, MaxMediaRequests: 8,
return playbackSettings{Enabled: false, DirectSources: []string{}, MaxSessions: 4, MaxMediaRequests: 8,
MaxRemuxJobs: 2, MaxAudioTranscodes: 1, MaxVideoTranscodes: 1, MaxPrefetchJobs: 1, CacheMB: 512}
}
+2 -2
View File
@@ -122,8 +122,8 @@ func playbackInputArgs(media providerMedia, input string, offset float64) []stri
func playbackEncodingArgs(media providerMedia, input string, offset float64) []string {
return append(playbackInputArgs(media, input, offset),
"-vf", "fps=30,scale=trunc(iw/2)*2:trunc(ih/2)*2,setsar=1",
"-c:v", "libx264", "-preset", "ultrafast", "-tune", "zerolatency", "-profile:v", "baseline",
"-pix_fmt", "yuv420p", "-crf", "22", "-maxrate", "12000k", "-bufsize", "24000k", "-threads", "4",
"-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency", "-profile:v", "baseline",
"-pix_fmt", "yuv420p", "-crf", "23", "-maxrate", "12000k", "-bufsize", "24000k", "-threads", "2",
"-g", "30", "-keyint_min", "30", "-sc_threshold", "0",
"-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2")
}
+22 -69
View File
@@ -242,58 +242,16 @@ func (app *UIApp) handleEmbyLegacyStream(writer http.ResponseWriter, request *ht
writer.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
return
}
sessionID := randomHex(24)
if sessionID == "" {
writeJSON(writer, http.StatusInternalServerError, map[string]string{"error": "无法创建播放会话"})
return
}
ctx, cancel := context.WithTimeout(request.Context(), 90*time.Second)
defer cancel()
app.playbackMu.Lock()
if len(app.playbacks) >= app.mediaResources().snapshot().MaxSessions {
app.playbackMu.Unlock()
writeJSON(writer, http.StatusTooManyRequests, map[string]string{"error": "同时播放数量已达上限,请稍后再试"})
return
}
if app.playbacks == nil {
app.playbacks = make(map[string]*playbackSession)
}
session := &playbackSession{accountID: request.URL.Query().Get("account"), dramaID: id, id: sessionID, run: 1, state: "opening", expires: time.Now().Add(playbackIdleTimeout), cancel: cancel}
session.timer = time.AfterFunc(playbackIdleTimeout, func() { app.expirePlayback(sessionID) })
app.playbacks[sessionID] = session
app.playbackMu.Unlock()
ready := false
defer func() {
if !ready {
app.closePlayback(sessionID)
}
}()
task, downloadID, err := app.embyTask(ctx, id, chapter)
if err != nil {
writeJSON(writer, http.StatusBadGateway, map[string]string{"error": "解析 Emby 分集失败:" + app.redactError(err)})
return
}
cache := newPlaybackNative(app, context.Background(), task, downloadID, 0, 0, false)
app.playbackMu.Lock()
if app.playbacks[sessionID] != session || ctx.Err() != nil {
app.playbackMu.Unlock()
cache.Close()
return
}
session.tasks = []Task{task}
session.native = cache
session.cancel = cache.Close
app.playbackMu.Unlock()
cache.start()
if _, err = cache.segment(ctx, 0); err != nil {
writeJSON(writer, http.StatusBadGateway, map[string]string{"error": "准备 Emby 播放失败:" + app.redactError(err)})
return
}
_, duration, _ := cache.metadata()
if duration <= 0 || duration > 24*60*60 || math.IsNaN(duration) || math.IsInf(duration, 0) {
writeJSON(writer, http.StatusBadGateway, map[string]string{"error": "未取得有效播放时长"})
session := app.acquireEmbyPlayback(writer, request, id, chapter, "", true)
if session == nil {
return
}
defer app.releaseEmbyPlayback(session)
app.serveEmbyPlaylist(writer, request, session)
}
func (app *UIApp) serveEmbyPlaylist(writer http.ResponseWriter, request *http.Request, session *playbackSession) {
id, chapter, sessionID, duration := session.dramaID, session.tasks[0].Chapter.ID, session.id, session.duration
var playlist strings.Builder
fmt.Fprintf(&playlist, "#EXTM3U\n#EXT-X-VERSION:3\n#EXT-X-TARGETDURATION:%d\n#EXT-X-MEDIA-SEQUENCE:0\n#EXT-X-PLAYLIST-TYPE:VOD\n#EXT-X-INDEPENDENT-SEGMENTS\n", playbackNativeSegmentSeconds)
for part, count := 0, int(math.Ceil(duration/playbackNativeSegmentSeconds)); part < count; part++ {
@@ -304,17 +262,6 @@ func (app *UIApp) handleEmbyLegacyStream(writer http.ResponseWriter, request *ht
fmt.Fprintf(&playlist, "#EXTINF:%.6f,\nsegment.ts?%s\n", math.Min(playbackNativeSegmentSeconds, duration-float64(part*playbackNativeSegmentSeconds)), query.Encode())
}
playlist.WriteString("#EXT-X-ENDLIST\n")
app.playbackMu.Lock()
if app.playbacks[sessionID] == session {
session.state = "streaming"
session.duration = duration
app.touchPlaybackLocked(session)
ready = true
}
app.playbackMu.Unlock()
if !ready {
return
}
writer.Header().Set("Content-Type", "application/vnd.apple.mpegurl")
http.ServeContent(writer, request, "index.m3u8", time.Time{}, strings.NewReader(playlist.String()))
}
@@ -330,23 +277,29 @@ func (app *UIApp) handleEmbySegment(writer http.ResponseWriter, request *http.Re
writeJSON(writer, http.StatusBadRequest, map[string]string{"error": "分片编号无效"})
return
}
app.playbackMu.Lock()
session := app.playbacks[query.Get("session")]
if session == nil || session.native == nil || len(session.tasks) != 1 || session.tasks[0].DramaID != id || session.tasks[0].Chapter.ID != chapter {
app.playbackMu.Unlock()
writeJSON(writer, http.StatusGone, map[string]string{"error": "播放已过期,请重新播放"})
if query.Get("session") == "" {
app.writeEmbyPlaybackError(writer, request, http.StatusGone, "播放已过期,请重新播放")
return
}
session := app.acquireEmbyPlayback(writer, request, id, chapter, query.Get("session"), true)
if session == nil {
return
}
defer app.releaseEmbyPlayback(session)
cache := session.native
app.touchPlaybackLocked(session)
app.playbackMu.Unlock()
if cache == nil {
app.writeEmbyPlaybackError(writer, request, http.StatusGone, "播放格式已变化,请重新播放")
return
}
ctx, cancel := context.WithTimeout(request.Context(), 90*time.Second)
defer cancel()
body, err := cache.segment(ctx, part)
if err != nil {
writeJSON(writer, http.StatusBadGateway, map[string]string{"error": "读取 Emby 分片失败:" + app.redactError(err)})
app.writeEmbyPlaybackError(writer, request, http.StatusBadGateway, "读取 Emby 分片失败:"+app.redactError(err))
return
}
active := &playbackRunContext{id: session.id, run: session.run, session: session, ctx: cache.ctx}
writer = &playbackActivityWriter{ResponseWriter: writer, app: app, run: active}
writer.Header().Set("Content-Type", "video/mp2t")
http.ServeContent(writer, request, "segment.ts", time.Time{}, bytes.NewReader(body))
}
+13 -117
View File
@@ -1,12 +1,8 @@
package app
import (
"context"
"errors"
"net/http"
"net/url"
"strings"
"time"
)
func (app *UIApp) handleEmbyStream(writer http.ResponseWriter, request *http.Request) {
@@ -22,141 +18,41 @@ func (app *UIApp) handleEmbyStream(writer http.ResponseWriter, request *http.Req
writer.Header().Set("Content-Type", "application/octet-stream")
return
}
owner := request.URL.Query().Get("account")
app.playbackMu.Lock()
var session *playbackSession
for _, candidate := range app.playbacks {
if candidate.viewer == nil && candidate.accountID == owner && candidate.mediaReady != nil && candidate.dramaID == id && len(candidate.tasks) == 1 && candidate.tasks[0].Chapter.ID == chapter && candidate.mediaError == nil && (candidate.media == nil || !candidate.media.failed.Load() && candidate.media.ctx.Err() == nil && (candidate.media.proxy == nil || candidate.media.proxy.Err() == nil)) {
session = candidate
break
}
}
session := app.acquireEmbyPlayback(writer, request, id, chapter, "", false)
if session == nil {
if len(app.playbacks) >= app.mediaResources().snapshot().MaxSessions {
app.playbackMu.Unlock()
writeJSON(writer, http.StatusTooManyRequests, map[string]string{"error": "同时播放数量已达上限,请稍后再试"})
return
}
sessionID := randomHex(24)
if sessionID == "" {
app.playbackMu.Unlock()
writer.WriteHeader(http.StatusInternalServerError)
return
}
parent, cancel := context.WithCancel(context.Background())
session = &playbackSession{id: sessionID, accountID: owner, dramaID: id, run: 1, state: "opening", mediaReady: make(chan struct{}), cancel: cancel,
tasks: []Task{{DramaID: id, Chapter: Chapter{ID: chapter}}}, expires: time.Now().Add(playbackIdleTimeout)}
if app.playbacks == nil {
app.playbacks = make(map[string]*playbackSession)
}
app.playbacks[sessionID] = session
session.timer = time.AfterFunc(playbackIdleTimeout, func() { app.expirePlayback(sessionID) })
query := url.Values{"id": {id}, "chapter": {chapter}, "key": {request.URL.Query().Get("key")}}
if owner != "" {
query.Set("account", owner)
}
origin := request.Header.Get("Origin")
go app.prepareEmbyMedia(parent, session, query.Encode(), origin)
}
session.mediaWaiters++
app.touchPlaybackLocked(session)
app.playbackMu.Unlock()
defer func() {
app.playbackMu.Lock()
session.mediaWaiters--
var cancel context.CancelFunc
if session.mediaWaiters == 0 && session.state == "opening" && app.playbacks[session.id] == session {
delete(app.playbacks, session.id)
session.timer.Stop()
cancel = session.cancel
}
app.playbackMu.Unlock()
if cancel != nil {
cancel()
}
}()
select {
case <-session.mediaReady:
case <-request.Context().Done():
return
}
app.playbackMu.Lock()
media, err := session.media, session.mediaError
app.playbackMu.Unlock()
if err != nil || media == nil {
if err == nil {
err = errors.New("播放会话已结束")
}
writeJSON(writer, http.StatusBadGateway, map[string]string{"error": "准备 Emby 播放失败:" + app.redactError(err)})
defer app.releaseEmbyPlayback(session)
if session.native != nil {
app.serveEmbyPlaylist(writer, request, session)
return
}
if media.plan.Player == "legacy" {
app.closePlayback(session.id)
app.handleEmbyLegacyStream(writer, request)
return
}
writer.Header().Set("Location", media.plan.URL)
writer.Header().Set("Location", session.media.plan.URL)
writer.Header().Set("Cache-Control", "private, no-store")
writer.WriteHeader(http.StatusFound)
}
func (app *UIApp) prepareEmbyMedia(parent context.Context, session *playbackSession, query, origin string) {
ctx, cancel := context.WithTimeout(parent, 90*time.Second)
defer cancel()
task, downloadID, err := app.embyTask(ctx, session.dramaID, session.tasks[0].Chapter.ID)
var media *playbackMediaSession
if err == nil {
media, err = app.resolveMediaSession(parent, ctx, task, downloadID, 0)
}
if err == nil {
base := "/api/emby/media/" + session.id + "/"
err = app.planMedia(ctx, media, playbackClientCapabilities{MP4: true, NativeHLS: true}, "auto", sourceFromDramaID(task.DramaID), origin, base, query)
}
app.playbackMu.Lock()
if err == nil && (app.playbacks[session.id] != session || ctx.Err() != nil) {
err = context.Canceled
}
if err == nil {
previous := session.cancel
session.cancel = func() { previous(); media.Close() }
session.tasks, session.media, session.state = []Task{task}, media, "streaming"
session.duration = media.plan.Duration
media.plan.Run = session.run
app.touchPlaybackLocked(session)
} else {
session.mediaError, session.state = err, "failed"
}
close(session.mediaReady)
app.playbackMu.Unlock()
if err != nil {
if media != nil {
media.Close()
}
app.closePlayback(session.id)
}
}
func (app *UIApp) handleEmbyMediaAsset(writer http.ResponseWriter, request *http.Request) {
id, chapter, ok := app.authorizeEmby(writer, request)
if !ok {
return
}
parts := strings.Split(strings.TrimPrefix(request.URL.Path, "/api/emby/media/"), "/")
if len(parts) != 2 {
if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
http.NotFound(writer, request)
return
}
app.playbackMu.Lock()
session := app.playbacks[parts[0]]
if session == nil || session.viewer != nil || session.media == nil || session.accountID != request.URL.Query().Get("account") || len(session.tasks) != 1 || session.tasks[0].DramaID != id || session.tasks[0].Chapter.ID != chapter {
app.playbackMu.Unlock()
writer.WriteHeader(http.StatusGone)
session := app.acquireEmbyPlayback(writer, request, id, chapter, parts[0], false)
if session == nil {
return
}
defer app.releaseEmbyPlayback(session)
media := session.media
if media == nil {
app.writeEmbyPlaybackError(writer, request, http.StatusGone, "播放格式已变化,请重新播放")
return
}
active := &playbackRunContext{id: session.id, run: session.run, session: session, ctx: media.ctx}
app.touchPlaybackLocked(session)
app.playbackMu.Unlock()
writer = &playbackActivityWriter{ResponseWriter: writer, app: app, run: active}
app.serveMediaSession(writer, request, media)
}
+4 -2
View File
@@ -15,7 +15,6 @@ import (
)
const playbackIdleTimeout = 10 * time.Minute
const playbackSessionLimit = 4
const playbackMIME = `video/mp4; codecs="avc1.42C01F, mp4a.40.2"`
type playbackSession struct {
@@ -41,6 +40,9 @@ type playbackSession struct {
mediaReady chan struct{}
mediaError error
mediaWaiters int
emby bool
embyLegacy bool
embyProcessing string
quality int
streamVersion uint64
openedAt time.Time
@@ -252,7 +254,7 @@ func (app *UIApp) handlePlaybackOpen(writer http.ResponseWriter, request *http.R
return
}
app.playbackMu.Lock()
if len(app.playbacks) >= app.mediaResources().snapshot().MaxSessions {
if app.activePlaybackSessionsLocked() >= app.mediaResources().snapshot().MaxSessions {
app.playbackMu.Unlock()
writeJSON(writer, http.StatusTooManyRequests, map[string]string{"error": "同时播放数量已达上限,请稍后再试"})
return
+3 -3
View File
@@ -149,8 +149,8 @@
</div>
<div class="toolbar"><button id="saveSettingsBtn">保存并应用</button><button id="checkNetworkBtn" class="secondary">检测当前连接</button><span id="settingsStatus" class="small" aria-live="polite"></span></div>
<details class="playback-settings"><summary>播放与资源限制</summary>
<label class="playback-option"><input id="originalPlaybackEnabled" type="checkbox" checked><span>优先原始 MP4 / HLS 播放</span></label>
<p class="small">兼容内容免转码播放;关闭后使用原有兼容链路。需要处理时优先保留视频,只在必要时转码。</p>
<label class="playback-option"><input id="originalPlaybackEnabled" type="checkbox"><span>优先原始 MP4 / HLS 播放</span></label>
<p class="small">默认关闭,使用原有兼容播放链路;开启后兼容内容免转码,需要处理时优先保留视频。</p>
<div class="settings-grid">
<label>播放会话上限<input id="playback-maxSessions" type="number" min="1" max="64" value="4"></label>
<label>上游媒体连接上限<input id="playback-maxMediaRequests" type="number" min="1" max="64" value="8"></label>
@@ -160,7 +160,7 @@
<label>下一集预备并发<input id="playback-maxPrefetchJobs" type="number" min="0" max="4" value="1"></label>
<label>封装缓存(MB)<input id="playback-cacheMB" type="number" min="64" max="8192" value="512"></label>
</div>
<p class="small">处理并发设为 0 可禁用该类任务。原始播放不占转码名额;新链路的下一集预备仅解析地址。</p>
<p class="small">Emby 探测或取流结束即释放会话名额。处理并发设为 0 可禁用该类任务;原始播放不占转码名额。</p>
<p>允许尝试客户端直连的站源</p><div id="playbackDirectSources" class="playback-direct-sources"></div>
<p class="small">默认关闭。检查通过后尝试直连,失败退回中转;已发出的源站链接可能在账号权限变更后继续有效。</p>
<div class="toolbar"><button id="savePlaybackSettings" class="secondary">保存播放设置</button><span id="playbackSettingsStatus" class="small" role="status"></span></div>