feat(stream): wire TranscodingThrottle into MediaStreamer

cmd/wire_gen.go

@@ -96,7 +96,8 @@ func CreateSubsonicAPIRouter(ctx context.Context) *subsonic.Router {
 	provider := external.NewProvider(dataStore, agentsAgents)
 	artworkArtwork := artwork.NewArtwork(dataStore, fileCache, fFmpeg, provider)
 	transcodingCache := stream.GetTranscodingCache()
-	mediaStreamer := stream.NewMediaStreamer(dataStore, fFmpeg, transcodingCache)
+	transcodingThrottle := stream.GetTranscodingThrottle()
+	mediaStreamer := stream.NewMediaStreamer(dataStore, fFmpeg, transcodingCache, transcodingThrottle)
 	share := core.NewShare(dataStore)
 	archiver := core.NewArchiver(mediaStreamer, dataStore, share)
 	players := core.NewPlayers(dataStore)
@@ -124,7 +125,8 @@ func CreatePublicRouter() *public.Router {
 	provider := external.NewProvider(dataStore, agentsAgents)
 	artworkArtwork := artwork.NewArtwork(dataStore, fileCache, fFmpeg, provider)
 	transcodingCache := stream.GetTranscodingCache()
-	mediaStreamer := stream.NewMediaStreamer(dataStore, fFmpeg, transcodingCache)
+	transcodingThrottle := stream.GetTranscodingThrottle()
+	mediaStreamer := stream.NewMediaStreamer(dataStore, fFmpeg, transcodingCache, transcodingThrottle)
 	share := core.NewShare(dataStore)
 	archiver := core.NewArchiver(mediaStreamer, dataStore, share)
 	router := public.New(dataStore, artworkArtwork, mediaStreamer, share, archiver)

core/stream/media_streamer.go

@@ -27,14 +27,15 @@ type MediaStreamer interface {
 
 type TranscodingCache cache.FileCache
 
-func NewMediaStreamer(ds model.DataStore, t ffmpeg.FFmpeg, cache TranscodingCache) MediaStreamer {
+func NewMediaStreamer(ds model.DataStore, t ffmpeg.FFmpeg, cache TranscodingCache, throttle *TranscodingThrottle) MediaStreamer {
-	return &mediaStreamer{ds: ds, transcoder: t, cache: cache}
+	return &mediaStreamer{ds: ds, transcoder: t, cache: cache, throttle: throttle}
 }
 
 type mediaStreamer struct {
 	ds         model.DataStore
 	transcoder ffmpeg.FFmpeg
 	cache      cache.FileCache
+	throttle   *TranscodingThrottle
 }
 
 type streamJob struct {

core/stream/media_streamer_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"io"
 	"os"
+	"time"
 
 	"github.com/navidrome/navidrome/conf"
 	"github.com/navidrome/navidrome/conf/configtest"
@@ -31,7 +32,7 @@ var _ = Describe("MediaStreamer", func() {
 		})
 		testCache := stream.NewTranscodingCache()
 		Eventually(func() bool { return testCache.Available(context.TODO()) }).Should(BeTrue())
-		streamer = stream.NewMediaStreamer(ds, ffmpeg, testCache)
+		streamer = stream.NewMediaStreamer(ds, ffmpeg, testCache, stream.NewTranscodingThrottle(0, 100, time.Minute))
 	})
 	AfterEach(func() {
 		_ = os.RemoveAll(conf.Server.CacheFolder)

core/stream/throttle.go

@@ -8,6 +8,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/navidrome/navidrome/conf"
 	"github.com/navidrome/navidrome/log"
 	"golang.org/x/sync/semaphore"
 )
@@ -90,3 +91,12 @@ func (r *releaseOnClose) Close() error {
 	r.once.Do(r.release)
 	return err
 }
+
+// GetTranscodingThrottle creates a TranscodingThrottle from the current configuration.
+func GetTranscodingThrottle() *TranscodingThrottle {
+	return NewTranscodingThrottle(
+		conf.Server.MaxConcurrentTranscodes,
+		conf.Server.DevTranscodeThrottleBacklogLimit,
+		conf.Server.DevTranscodeThrottleBacklogTimeout,
+	)
+}

core/wire_providers.go

@@ -16,6 +16,7 @@ import (
 var Set = wire.NewSet(
 	stream.NewMediaStreamer,
 	stream.GetTranscodingCache,
+	stream.GetTranscodingThrottle,
 	NewArchiver,
 	NewPlayers,
 	NewShare,