优化起播

internal/service/emby_remote_gate_test.go

@@ -0,0 +1,112 @@
+package service
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"go.uber.org/zap"
+
+	"github.com/truewhile/MeBox/internal/config"
+)
+
+// 远程 Emby 的并发闸门必须真的把在途请求数压在上限之内：第三方客户端首页会为
+// 每个远程媒体库各请求一次 /Items/Latest，几十个库就是几十路并发。
+func TestRemoteGateLimitsConcurrentRequests(t *testing.T) {
+	const gate = 3
+	const requests = 12
+
+	var mu sync.Mutex
+	inflight, peak := 0, 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		mu.Lock()
+		inflight++
+		if inflight > peak {
+			peak = inflight
+		}
+		mu.Unlock()
+
+		time.Sleep(20 * time.Millisecond)
+
+		mu.Lock()
+		inflight--
+		mu.Unlock()
+		_, _ = w.Write([]byte(`{"Items":[]}`))
+	}))
+	defer srv.Close()
+
+	svc := &EmbyRemoteService{http: srv.Client(), remoteGate: make(chan struct{}, gate)}
+
+	var wg sync.WaitGroup
+	for i := 0; i < requests; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			req, err := http.NewRequestWithContext(context.Background(), http.MethodGet, srv.URL+"/Items/Latest", nil)
+			if err != nil {
+				return
+			}
+			if _, _, err := svc.fetchRemoteBody(context.Background(), req, "/Items/Latest"); err != nil {
+				t.Errorf("fetchRemoteBody: %v", err)
+			}
+		}()
+	}
+	wg.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	if peak == 0 {
+		t.Fatal("test server never saw a request")
+	}
+	if peak > gate {
+		t.Fatalf("peak concurrency = %d, want <= %d", peak, gate)
+	}
+}
+
+// 闸门排队时要响应请求取消，不能把整个 HTTP 请求挂死。
+func TestRemoteGateHonoursContextCancellation(t *testing.T) {
+	svc := &EmbyRemoteService{remoteGate: make(chan struct{}, 1)}
+	svc.remoteGate <- struct{}{} // 占满名额
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := svc.enterRemoteGate(ctx); !errors.Is(err, context.Canceled) {
+		t.Fatalf("err = %v, want context.Canceled", err)
+	}
+
+	// 释放名额后必须能正常拿到。
+	<-svc.remoteGate
+	release, err := svc.enterRemoteGate(context.Background())
+	if err != nil {
+		t.Fatalf("enterRemoteGate after release: %v", err)
+	}
+	release()
+	if len(svc.remoteGate) != 0 {
+		t.Fatalf("gate leaked a permit: len = %d", len(svc.remoteGate))
+	}
+}
+
+// 未配置闸门（测试里直接构造结构体）时不应限流，保持旧行为。
+func TestRemoteGateAbsentIsUnlimited(t *testing.T) {
+	svc := &EmbyRemoteService{}
+	release, err := svc.enterRemoteGate(context.Background())
+	if err != nil {
+		t.Fatalf("enterRemoteGate: %v", err)
+	}
+	release()
+}
+
+// 生产路径构造出来的服务必须带闸门，否则上面的限制形同虚设。
+func TestNewEmbyRemoteServiceInitialisesGate(t *testing.T) {
+	svc := NewEmbyRemoteService(&config.Config{}, zap.NewNop(), nil, nil)
+	if svc.remoteGate == nil {
+		t.Fatal("remoteGate must be initialised by the constructor")
+	}
+	if cap(svc.remoteGate) != embyRemoteConcurrencyLimit {
+		t.Fatalf("gate capacity = %d, want %d", cap(svc.remoteGate), embyRemoteConcurrencyLimit)
+	}
+}