nik/navidrome
1
0
Fork 0
mirror of https://github.com/navidrome/navidrome.git synced 2026-08-31 07:30:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge aa97cb7626b8b68b66e58a90fbc0800f5f09ef24 into 3867fab4da6ea7142bfc6374c94b34a708e03b4e

Browse Source
This commit is contained in:
iheartfoss 2026-08-30 21:00:32 +05:30 committed by GitHub
commit 6dca6c2b3a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 274 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

173
core/playback/device.go
View File

@ -23,6 +23,7 @@ type Track interface {
}
type playbackDevice struct {
mutex sync.RWMutex
serviceCtx context.Context
ParentPlaybackServer PlaybackServer
Default bool
@ -31,7 +32,7 @@ type playbackDevice struct {
DeviceName string
PlaybackQueue *Queue
Gain float32
PlaybackDone chan bool
PlaybackDone chan *mpv.MpvTrack
ActiveTrack Track
startTrackSwitcher sync.Once
}
@ -45,14 +46,40 @@ type DeviceStatus struct {
const DefaultGain float32 = 1.0
func (pd *playbackDevice) getStatus() DeviceStatus {
// getStatusLocked must be called with pd.mutex held. It performs blocking IPC
// calls to mpv (Position/IsPlaying) while holding the lock; prefer getStatus
// for read-only call sites that don't already need the lock for anything else.
func (pd *playbackDevice) getStatusLocked() DeviceStatus {
pos := 0
if pd.ActiveTrack != nil {
pos = pd.ActiveTrack.Position()
}
return DeviceStatus{
CurrentIndex: pd.PlaybackQueue.Index,
Playing: pd.isPlaying(),
Playing: pd.isPlayingLocked(),
Gain: pd.Gain,
Position: pos,
}
}
// getStatus takes a read lock for the whole call, including the blocking mpv
// IPC calls (Position/IsPlaying). This lets concurrent Status/Get requests
// run in parallel with each other, while still preventing a concurrent
// mutation (Skip/Clear/Set/track-switch) from closing ActiveTrack out from
// under the IPC calls.
func (pd *playbackDevice) getStatus() DeviceStatus {
pd.mutex.RLock()
defer pd.mutex.RUnlock()
pos := 0
playing := false
if pd.ActiveTrack != nil {
pos = pd.ActiveTrack.Position()
playing = pd.ActiveTrack.IsPlaying()
}
return DeviceStatus{
CurrentIndex: pd.PlaybackQueue.Index,
Playing: playing,
Gain: pd.Gain,
Position: pos,
}
@ -70,7 +97,7 @@ func NewPlaybackDevice(ctx context.Context, playbackServer PlaybackServer, name
DeviceName: deviceName,
Gain: DefaultGain,
PlaybackQueue: NewQueue(),
PlaybackDone: make(chan bool),
PlaybackDone: make(chan *mpv.MpvTrack),
}
}
@ -80,7 +107,10 @@ func (pd *playbackDevice) String() string {
func (pd *playbackDevice) Get(ctx context.Context) (model.MediaFiles, DeviceStatus, error) {
log.Debug(ctx, "Processing Get action", "device", pd)
return pd.PlaybackQueue.Get(), pd.getStatus(), nil
pd.mutex.RLock()
items := pd.PlaybackQueue.Get()
pd.mutex.RUnlock()
return items, pd.getStatus(), nil
}
func (pd *playbackDevice) Status(ctx context.Context) (DeviceStatus, error) {
@ -92,12 +122,17 @@ func (pd *playbackDevice) Status(ctx context.Context) (DeviceStatus, error) {
func (pd *playbackDevice) Set(ctx context.Context, ids []string) (DeviceStatus, error) {
log.Debug(ctx, "Processing Set action", "ids", ids, "device", pd)
_, err := pd.Clear(ctx)
items, err := pd.fetchMediaFiles(ctx, ids)
if err != nil {
log.Error(ctx, "error setting tracks", ids)
return pd.getStatus(), err
return DeviceStatus{}, err
}
return pd.Add(ctx, ids)
pd.mutex.Lock()
pd.clearLocked()
pd.PlaybackQueue.Add(items)
pd.mutex.Unlock()
return pd.getStatus(), nil
}
func (pd *playbackDevice) Start(ctx context.Context) (DeviceStatus, error) {
@ -111,37 +146,52 @@ func (pd *playbackDevice) Start(ctx context.Context) (DeviceStatus, error) {
}()
})
pd.mutex.Lock()
defer pd.mutex.Unlock()
return pd.startLocked(ctx)
}
func (pd *playbackDevice) startLocked(ctx context.Context) (DeviceStatus, error) {
if pd.ActiveTrack != nil {
if pd.isPlaying() {
if pd.isPlayingLocked() {
log.Debug("trying to start an already playing track")
} else {
pd.ActiveTrack.Unpause()
}
} else {
if !pd.PlaybackQueue.IsEmpty() {
err := pd.switchActiveTrackByIndex(pd.PlaybackQueue.Index)
err := pd.switchActiveTrackByIndexLocked(pd.PlaybackQueue.Index)
if err != nil {
return pd.getStatus(), err
return pd.getStatusLocked(), err
}
pd.ActiveTrack.Unpause()
}
}
return pd.getStatus(), nil
return pd.getStatusLocked(), nil
}
func (pd *playbackDevice) Stop(ctx context.Context) (DeviceStatus, error) {
log.Debug(ctx, "Processing Stop action", "device", pd)
pd.mutex.Lock()
defer pd.mutex.Unlock()
return pd.stopLocked(ctx)
}
func (pd *playbackDevice) stopLocked(ctx context.Context) (DeviceStatus, error) {
if pd.ActiveTrack != nil {
pd.ActiveTrack.Pause()
}
return pd.getStatus(), nil
return pd.getStatusLocked(), nil
}
func (pd *playbackDevice) Skip(ctx context.Context, index int, offset int) (DeviceStatus, error) {
log.Debug(ctx, "Processing Skip action", "index", index, "offset", offset, "device", pd)
wasPlaying := pd.isPlaying()
pd.mutex.Lock()
defer pd.mutex.Unlock()
wasPlaying := pd.isPlayingLocked()
if pd.ActiveTrack != nil && wasPlaying {
pd.ActiveTrack.Pause()
@ -153,27 +203,27 @@ func (pd *playbackDevice) Skip(ctx context.Context, index int, offset int) (Devi
}
if pd.ActiveTrack == nil {
err := pd.switchActiveTrackByIndex(index)
err := pd.switchActiveTrackByIndexLocked(index)
if err != nil {
return pd.getStatus(), err
return pd.getStatusLocked(), err
}
}
err := pd.ActiveTrack.SetPosition(offset)
if err != nil {
log.Error(ctx, "error setting position", err)
return pd.getStatus(), err
return pd.getStatusLocked(), err
}
if wasPlaying {
_, err = pd.Start(ctx)
_, err = pd.startLocked(ctx)
if err != nil {
log.Error(ctx, "error starting new track after skipping")
return pd.getStatus(), err
return pd.getStatusLocked(), err
}
}
return pd.getStatus(), nil
return pd.getStatusLocked(), nil
}
func (pd *playbackDevice) Add(ctx context.Context, ids []string) (DeviceStatus, error) {
@ -182,40 +232,61 @@ func (pd *playbackDevice) Add(ctx context.Context, ids []string) (DeviceStatus,
return pd.getStatus(), nil
}
items := model.MediaFiles{}
for _, id := range ids {
mf, err := pd.ParentPlaybackServer.GetMediaFile(id)
if err != nil {
return DeviceStatus{}, err
}
log.Debug(ctx, "Found mediafile: "+mf.Path)
items = append(items, *mf)
items, err := pd.fetchMediaFiles(ctx, ids)
if err != nil {
return DeviceStatus{}, err
}
pd.mutex.Lock()
pd.PlaybackQueue.Add(items)
pd.mutex.Unlock()
return pd.getStatus(), nil
}
// fetchMediaFiles resolves media file IDs to model.MediaFiles. It performs
// database queries and must not be called while pd.mutex is held.
func (pd *playbackDevice) fetchMediaFiles(ctx context.Context, ids []string) (model.MediaFiles, error) {
items := model.MediaFiles{}
for _, id := range ids {
mf, err := pd.ParentPlaybackServer.GetMediaFile(id)
if err != nil {
return nil, err
}
log.Debug(ctx, "Found mediafile: "+mf.Path)
items = append(items, *mf)
}
return items, nil
}
func (pd *playbackDevice) Clear(ctx context.Context) (DeviceStatus, error) {
log.Debug(ctx, "Processing Clear action", "device", pd)
pd.mutex.Lock()
defer pd.mutex.Unlock()
pd.clearLocked()
return pd.getStatusLocked(), nil
}
func (pd *playbackDevice) clearLocked() {
if pd.ActiveTrack != nil {
pd.ActiveTrack.Pause()
pd.ActiveTrack.Close()
pd.ActiveTrack = nil
}
pd.PlaybackQueue.Clear()
return pd.getStatus(), nil
}
func (pd *playbackDevice) Remove(ctx context.Context, index int) (DeviceStatus, error) {
log.Debug(ctx, "Processing Remove action", "index", index, "device", pd)
pd.mutex.Lock()
defer pd.mutex.Unlock()
// pausing if attempting to remove running track
if pd.isPlaying() && pd.PlaybackQueue.Index == index {
_, err := pd.Stop(ctx)
if pd.isPlayingLocked() && pd.PlaybackQueue.Index == index {
_, err := pd.stopLocked(ctx)
if err != nil {
log.Error(ctx, "error stopping running track")
return pd.getStatus(), err
return pd.getStatusLocked(), err
}
}
@ -224,30 +295,36 @@ func (pd *playbackDevice) Remove(ctx context.Context, index int) (DeviceStatus,
} else {
log.Error(ctx, "Index to remove out of range: "+fmt.Sprint(index))
}
return pd.getStatus(), nil
return pd.getStatusLocked(), nil
}
func (pd *playbackDevice) Shuffle(ctx context.Context) (DeviceStatus, error) {
log.Debug(ctx, "Processing Shuffle action", "device", pd)
pd.mutex.Lock()
defer pd.mutex.Unlock()
if pd.PlaybackQueue.Size() > 1 {
pd.PlaybackQueue.Shuffle()
}
return pd.getStatus(), nil
return pd.getStatusLocked(), nil
}
// SetGain is used to control the playback volume. A float value between 0.0 and 1.0.
func (pd *playbackDevice) SetGain(ctx context.Context, gain float32) (DeviceStatus, error) {
log.Debug(ctx, "Processing SetGain action", "newGain", gain, "device", pd)
pd.mutex.Lock()
defer pd.mutex.Unlock()
if pd.ActiveTrack != nil {
pd.ActiveTrack.SetVolume(gain)
}
pd.Gain = gain
return pd.getStatus(), nil
return pd.getStatusLocked(), nil
}
func (pd *playbackDevice) isPlaying() bool {
// isPlayingLocked must be called with pd.mutex held.
func (pd *playbackDevice) isPlayingLocked() bool {
return pd.ActiveTrack != nil && pd.ActiveTrack.IsPlaying()
}
@ -255,17 +332,23 @@ func (pd *playbackDevice) trackSwitcherGoroutine() {
log.Debug("Started trackSwitcher goroutine", "device", pd)
for {
select {
case <-pd.PlaybackDone:
case finishedTrack := <-pd.PlaybackDone:
log.Debug("Track switching detected")
if pd.ActiveTrack != nil {
pd.ActiveTrack.Close()
pd.ActiveTrack = nil
pd.mutex.Lock()
if pd.ActiveTrack != finishedTrack {
// The active track was already replaced (e.g. by Skip/Clear/Set)
// since this finish signal was sent. Ignore the stale signal.
log.Debug("Ignoring stale track-finished signal")
pd.mutex.Unlock()
continue
}
pd.ActiveTrack.Close()
pd.ActiveTrack = nil
if !pd.PlaybackQueue.IsAtLastElement() {
pd.PlaybackQueue.IncreaseIndex()
log.Debug("Switching to next song", "queue", pd.PlaybackQueue.String())
err := pd.switchActiveTrackByIndex(pd.PlaybackQueue.Index)
err := pd.switchActiveTrackByIndexLocked(pd.PlaybackQueue.Index)
if err != nil {
log.Error("Error switching track", err)
}
@ -275,6 +358,7 @@ func (pd *playbackDevice) trackSwitcherGoroutine() {
} else {
log.Debug("There is no song left in the playlist. Finish.")
}
pd.mutex.Unlock()
case <-pd.serviceCtx.Done():
log.Debug("Stopping trackSwitcher goroutine", "device", pd.Name)
return
@ -282,7 +366,8 @@ func (pd *playbackDevice) trackSwitcherGoroutine() {
}
}
func (pd *playbackDevice) switchActiveTrackByIndex(index int) error {
// switchActiveTrackByIndexLocked must be called with pd.mutex held.
func (pd *playbackDevice) switchActiveTrackByIndexLocked(index int) error {
pd.PlaybackQueue.SetIndex(index)
currentTrack := pd.PlaybackQueue.Current()
if currentTrack == nil {

124
core/playback/device_test.go Normal file
View File

@ -0,0 +1,124 @@
package playback
import (
"context"
"github.com/navidrome/navidrome/core/playback/mpv"
"github.com/navidrome/navidrome/model"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
// fakeTrack is a minimal Track implementation used to exercise playbackDevice
// logic without spawning a real mpv process.
type fakeTrack struct {
playing bool
position int
positionHits int
playingHits int
}
func (f *fakeTrack) IsPlaying() bool {
f.playingHits++
return f.playing
}
func (f *fakeTrack) SetVolume(float32) {}
func (f *fakeTrack) Pause() {}
func (f *fakeTrack) Unpause() {}
func (f *fakeTrack) Position() int {
f.positionHits++
return f.position
}
func (f *fakeTrack) SetPosition(int) error { return nil }
func (f *fakeTrack) Close() {}
func (f *fakeTrack) String() string { return "fakeTrack" }
var _ = Describe("playbackDevice", func() {
var pd *playbackDevice
BeforeEach(func() {
pd = NewPlaybackDevice(context.Background(), nil, "auto", "auto")
})
Describe("getStatus", func() {
It("reflects the active track's live state", func() {
track := &fakeTrack{playing: true, position: 42}
pd.ActiveTrack = track
pd.PlaybackQueue.Add(model.MediaFiles{{ID: "1"}})
pd.Gain = 0.75
status := pd.getStatus()
Expect(status.Playing).To(BeTrue())
Expect(status.Position).To(Equal(42))
Expect(status.Gain).To(Equal(float32(0.75)))
Expect(status.CurrentIndex).To(Equal(0))
Expect(track.positionHits).To(Equal(1))
Expect(track.playingHits).To(Equal(1))
})
It("reports not-playing with no active track", func() {
status := pd.getStatus()
Expect(status.Playing).To(BeFalse())
Expect(status.Position).To(Equal(0))
})
})
Describe("trackSwitcherGoroutine", func() {
var ctx context.Context
var cancel context.CancelFunc
BeforeEach(func() {
ctx, cancel = context.WithCancel(context.Background())
pd.serviceCtx = ctx
pd.PlaybackQueue.Add(model.MediaFiles{{ID: "only-track"}})
go pd.trackSwitcherGoroutine()
})
AfterEach(func() {
cancel()
})
It("ignores a stale finish signal for a track that was already replaced", func() {
staleTrack := &mpv.MpvTrack{}
currentTrack := &mpv.MpvTrack{}
pd.mutex.Lock()
pd.ActiveTrack = currentTrack
pd.mutex.Unlock()
pd.PlaybackDone <- staleTrack
Consistently(func() bool {
pd.mutex.RLock()
defer pd.mutex.RUnlock()
return pd.ActiveTrack == currentTrack
}).Should(BeTrue())
Expect(staleTrack.CloseCalled).To(BeFalse())
Expect(currentTrack.CloseCalled).To(BeFalse())
})
It("closes and advances past a track that legitimately finished", func() {
finishedTrack := &mpv.MpvTrack{}
pd.mutex.Lock()
pd.ActiveTrack = finishedTrack
pd.mutex.Unlock()
pd.PlaybackDone <- finishedTrack
Eventually(func() bool {
return finishedTrack.CloseCalled
}).Should(BeTrue())
Eventually(func() Track {
pd.mutex.RLock()
defer pd.mutex.RUnlock()
return pd.ActiveTrack
}).Should(BeNil())
})
})
})

2
core/playback/mpv/mpv_test.go
View File

@ -349,7 +349,7 @@ var _ = Describe("MPV", func() {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
playbackDone := make(chan bool, 1)
playbackDone := make(chan *MpvTrack, 1)
_, err := NewTrack(ctx, playbackDone, "auto", testMediaFile)
Expect(err).To(HaveOccurred())
Expect(err.Error()).To(Equal("no mpv command arguments provided"))

6
core/playback/mpv/track.go
View File

@ -18,14 +18,14 @@ import (
type MpvTrack struct {
MediaFile model.MediaFile
PlaybackDone chan bool
PlaybackDone chan<- *MpvTrack
Conn *mpvipc.Connection
IPCSocketName string
Exe *Executor
CloseCalled bool
}
func NewTrack(ctx context.Context, playbackDoneChannel chan bool, deviceName string, mf model.MediaFile) (*MpvTrack, error) {
func NewTrack(ctx context.Context, playbackDoneChannel chan<- *MpvTrack, deviceName string, mf model.MediaFile) (*MpvTrack, error) {
log.Debug("Loading track", "trackPath", mf.Path, "mediaType", mf.ContentType())
if _, err := mpvCommand(); err != nil {
@ -65,7 +65,7 @@ func NewTrack(ctx context.Context, playbackDoneChannel chan bool, deviceName str
conn.WaitUntilClosed()
log.Info("Hitting end-of-stream, signalling on channel")
if !theTrack.CloseCalled {
playbackDoneChannel <- true
playbackDoneChannel <- theTrack
}
}()

7
core/playback/queue.go
View File

@ -42,9 +42,12 @@ func (pd *Queue) Current() *model.MediaFile {
return &pd.Items[pd.Index]
}
// returns the whole queue
// returns a copy of the whole queue, safe for the caller to use after
// releasing any lock protecting the Queue.
func (pd *Queue) Get() model.MediaFiles {
return pd.Items
items := make(model.MediaFiles, len(pd.Items))
copy(items, pd.Items)
return items
}
func (pd *Queue) Size() int {

12
core/playback/queue_test.go
View File

@ -116,6 +116,18 @@ var _ = Describe("Queues", func() {
queue.Clear()
Expect(queue.Size()).To(Equal(0))
})
It("returns a copy from Get, not the backing slice", func() {
snapshot := queue.Get()
Expect(snapshot).To(HaveLen(5))
queue.Remove(0)
Expect(queue.Size()).To(Equal(4))
// The previously returned snapshot must be unaffected by the mutation.
Expect(snapshot).To(HaveLen(5))
Expect(snapshot[0].ID).To(Equal("1"))
})
})
})