feat(jukebox): use mpv over ipc as a player backend

jukebox/jukebox.go

@@ -1,205 +1,406 @@
 // author: AlexKraak (https://github.com/alexkraak/)
+// author: sentriz (https://github.com/sentriz/)
 
 package jukebox
 
 import (
+	"errors"
 	"fmt"
-	"log"
 	"os"
+	"os/exec"
+	"path/filepath"
 	"sync"
 	"time"
 
-	"github.com/faiface/beep"
-	"github.com/faiface/beep/flac"
-	"github.com/faiface/beep/mp3"
-	"github.com/faiface/beep/speaker"
-
-	"go.senan.xyz/gonic/db"
+	"github.com/dexterlb/mpvipc"
+	"github.com/mitchellh/mapstructure"
+	"golang.org/x/exp/slices"
 )
 
-type Status struct {
-	CurrentIndex int
-	Playing      bool
-	Gain         float64
-	Position     int
+var (
+	ErrMPVTimeout      = fmt.Errorf("mpv not responding")
+	ErrMPVNeverStarted = fmt.Errorf("mpv never started")
+)
+
+func MPVArg(k string, v any) string {
+	if v, ok := v.(bool); ok {
+		if v {
+			return fmt.Sprintf("%s=yes", k)
+		}
+		return fmt.Sprintf("%s=no", k)
+	}
+	return fmt.Sprintf("%s=%v", k, v)
 }
 
 type Jukebox struct {
-	playlist []*db.Track
-	index    int
-	playing  bool
-	sr       beep.SampleRate
-	// used to notify the player to re read the members
-	quit    chan struct{}
-	done    chan bool
-	info    *strmInfo
-	speaker chan updateSpeaker
-	sync.Mutex
-}
+	cmd    *exec.Cmd
+	conn   *mpvipc.Connection
+	events <-chan *mpvipc.Event
 
-type strmInfo struct {
-	ctrlStrmr beep.Ctrl
-	strm      beep.StreamSeekCloser
-	format    beep.Format
-}
-
-type updateSpeaker struct {
-	index  int
-	offset int
+	mu sync.Mutex
 }
 
 func New() *Jukebox {
-	return &Jukebox{
-		sr:      beep.SampleRate(48000),
-		speaker: make(chan updateSpeaker, 1),
-		done:    make(chan bool),
-		quit:    make(chan struct{}),
-	}
+	return &Jukebox{}
 }
 
-func (j *Jukebox) Listen() error {
-	if err := speaker.Init(j.sr, j.sr.N(time.Second/2)); err != nil {
-		return fmt.Errorf("initing speaker: %w", err)
+func (j *Jukebox) Start(sockPath string, mpvExtraArgs []string) error {
+	const mpvName = "mpv"
+	if _, err := exec.LookPath(mpvName); err != nil {
+		return fmt.Errorf("look path: %w. did you forget to install it?", err)
 	}
+
+	var mpvArgs []string
+	mpvArgs = append(mpvArgs, "--idle", "--no-config", "--no-video", MPVArg("--audio-display", "no"), MPVArg("--input-ipc-server", sockPath))
+	mpvArgs = append(mpvArgs, mpvExtraArgs...)
+
+	j.cmd = exec.Command(mpvName, mpvArgs...)
+	if err := j.cmd.Start(); err != nil {
+		return fmt.Errorf("start mpv process: %w", err)
+	}
+
+	ok := waitUntil(5*time.Second, func() bool {
+		_, err := os.Stat(sockPath)
+		return err == nil
+	})
+	if !ok {
+		_ = j.cmd.Process.Kill()
+		return ErrMPVNeverStarted
+	}
+
+	j.conn = mpvipc.NewConnection(sockPath)
+	if err := j.conn.Open(); err != nil {
+		return fmt.Errorf("open connection: %w", err)
+	}
+	if _, err := j.conn.Call("observe_property", 0, "seekable"); err != nil {
+		return fmt.Errorf("observe property: %w", err)
+	}
+	j.events, _ = j.conn.NewEventListener()
+
+	return nil
+}
+
+func (j *Jukebox) Wait() error {
+	var exitError *exec.ExitError
+	if err := j.cmd.Wait(); err != nil && !errors.As(err, &exitError) {
+		return fmt.Errorf("wait jukebox: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) GetPlaylist() ([]string, error) {
+	defer lock(&j.mu)()
+
+	var playlist mpvPlaylist
+	if err := j.getDecode(&playlist, "playlist"); err != nil {
+		return nil, fmt.Errorf("get playlist: %w", err)
+	}
+	var items []string
+	for _, item := range playlist {
+		items = append(items, item.Filename)
+	}
+	return items, nil
+}
+
+func (j *Jukebox) SetPlaylist(items []string) error {
+	defer lock(&j.mu)()
+
+	var playlist mpvPlaylist
+	if err := j.getDecode(&playlist, "playlist"); err != nil {
+		return fmt.Errorf("get playlist: %w", err)
+	}
+	current, currentIndex := find(playlist, func(item mpvPlaylistItem) bool {
+		return item.Current
+	})
+
+	cwd, _ := os.Getwd()
+	currFilename, _ := filepath.Rel(cwd, current.Filename)
+	filteredItems, foundExistingTrack := filter(items, func(filename string) bool {
+		return filename != currFilename
+	})
+
+	tmp, cleanup, err := tmp()
+	if err != nil {
+		return fmt.Errorf("create temp file: %w", err)
+	}
+	defer cleanup()
+	for _, item := range filteredItems {
+		item, _ = filepath.Abs(item)
+		fmt.Fprintln(tmp, item)
+	}
+
+	if !foundExistingTrack {
+		// easy case - a brand new set of tracks that we can overwrite
+		if _, err := j.conn.Call("loadlist", tmp.Name(), "replace"); err != nil {
+			return fmt.Errorf("load list: %w", err)
+		}
+		return nil
+	}
+
+	// not so easy, we need to clear the playlist except what's playing, load everything
+	// except for what we're playing, then move what's playing back to its original index
+	// clear all items except what's playing (will be at index 0)
+	if _, err := j.conn.Call("playlist-clear"); err != nil {
+		return fmt.Errorf("clear playlist: %w", err)
+	}
+	if _, err := j.conn.Call("loadlist", tmp.Name(), "append-play"); err != nil {
+		return fmt.Errorf("load list: %w", err)
+	}
+	if _, err := j.conn.Call("playlist-move", 0, currentIndex+1); err != nil {
+		return fmt.Errorf("playlist move: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) AppendToPlaylist(items []string) error {
+	defer lock(&j.mu)()
+
+	tmp, cleanup, err := tmp()
+	if err != nil {
+		return fmt.Errorf("create temp file: %w", err)
+	}
+	defer cleanup()
+	for _, item := range items {
+		fmt.Fprintln(tmp, item)
+	}
+	if _, err := j.conn.Call("loadlist", tmp.Name(), "append"); err != nil {
+		return fmt.Errorf("load list: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) RemovePlaylistIndex(i int) error {
+	defer lock(&j.mu)()
+
+	if _, err := j.conn.Call("playlist-remove", i); err != nil {
+		return fmt.Errorf("playlist remove: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) SkipToPlaylistIndex(i int, offsetSecs int) error {
+	defer lock(&j.mu)()
+
+	matchEventSeekable := func(e *mpvipc.Event) bool {
+		seekable, _ := e.Data.(bool)
+		return e.Name == "property-change" &&
+			e.ExtraData["name"] == "seekable" &&
+			seekable
+	}
+
+	if offsetSecs > 0 {
+		if err := j.conn.Set("pause", true); err != nil {
+			return fmt.Errorf("pause: %w", err)
+		}
+	}
+	if _, err := j.conn.Call("playlist-play-index", i); err != nil {
+		return fmt.Errorf("playlist play index: %w", err)
+	}
+	if offsetSecs > 0 {
+		if err := waitFor(j.events, matchEventSeekable); err != nil {
+			return fmt.Errorf("waiting for file load: %w", err)
+		}
+		if _, err := j.conn.Call("seek", offsetSecs, "absolute"); err != nil {
+			return fmt.Errorf("seek: %w", err)
+		}
+		if err := j.conn.Set("pause", false); err != nil {
+			return fmt.Errorf("play: %w", err)
+		}
+	}
+	return nil
+}
+
+func (j *Jukebox) ClearPlaylist() error {
+	defer lock(&j.mu)()
+
+	if _, err := j.conn.Call("playlist-clear"); err != nil {
+		return fmt.Errorf("seek: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) Pause() error {
+	defer lock(&j.mu)()
+
+	if err := j.conn.Set("pause", true); err != nil {
+		return fmt.Errorf("pause: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) Play() error {
+	defer lock(&j.mu)()
+
+	if err := j.conn.Set("pause", false); err != nil {
+		return fmt.Errorf("pause: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) SetVolumePct(v int) error {
+	defer lock(&j.mu)()
+
+	if err := j.conn.Set("volume", v); err != nil {
+		return fmt.Errorf("set volume: %w", err)
+	}
+	return nil
+}
+
+func (j *Jukebox) GetVolumePct() (float64, error) {
+	defer lock(&j.mu)()
+
+	var volume float64
+	if err := j.getDecode(&volume, "volume"); err != nil {
+		return 0, fmt.Errorf("get volume: %w", err)
+	}
+	return volume, nil
+}
+
+type Status struct {
+	CurrentIndex    int
+	CurrentFilename string
+	Length          int
+	Playing         bool
+	GainPct         int
+	Position        int
+}
+
+func (j *Jukebox) GetStatus() (*Status, error) {
+	defer lock(&j.mu)()
+
+	var status Status
+	_ = j.getDecode(&status.Position, "time-pos") // property may not always be there
+	_ = j.getDecode(&status.GainPct, "volume")    // property may not always be there
+
+	var playlist mpvPlaylist
+	_ = j.getDecode(&playlist, "playlist")
+
+	status.CurrentIndex = slices.IndexFunc(playlist, func(pl mpvPlaylistItem) bool {
+		return pl.Current
+	})
+
+	status.Length = len(playlist)
+
+	if status.CurrentIndex < 0 {
+		return &status, nil
+	}
+
+	status.CurrentFilename = playlist[status.CurrentIndex].Filename
+
+	var paused bool
+	_ = j.getDecode(&paused, "pause") // property may not always be there
+	status.Playing = !paused
+
+	return &status, nil
+}
+
+func (j *Jukebox) Quit() error {
+	defer lock(&j.mu)()
+
+	if j.conn == nil || j.conn.IsClosed() {
+		return nil
+	}
+	if _, err := j.conn.Call("quit"); err != nil {
+		return fmt.Errorf("quit: %w", err)
+	}
+	if err := j.conn.Close(); err != nil {
+		return fmt.Errorf("close: %w", err)
+	}
+	j.conn.WaitUntilClosed()
+	return nil
+}
+
+func (j *Jukebox) getDecode(dest any, property string) error {
+	raw, err := j.conn.Get(property)
+	if err != nil {
+		return fmt.Errorf("get property: %w", err)
+	}
+	if err := mapstructure.Decode(raw, dest); err != nil {
+		return fmt.Errorf("decode: %w", err)
+	}
+	return nil
+}
+
+type mpvPlaylist []mpvPlaylistItem
+type mpvPlaylistItem struct {
+	ID       int
+	Filename string
+	Current  bool
+	Playing  bool
+}
+
+func waitUntil(timeout time.Duration, f func() bool) bool {
+	quit := time.NewTicker(timeout)
+	defer quit.Stop()
+	check := time.NewTicker(100 * time.Millisecond)
+	defer check.Stop()
+
 	for {
 		select {
-		case <-j.quit:
-			return nil
-		case speaker := <-j.speaker:
-			if err := j.doUpdateSpeaker(speaker); err != nil {
-				log.Printf("error in jukebox: %v", err)
+		case <-quit.C:
+			return false
+		case <-check.C:
+			if f() {
+				return true
 			}
 		}
 	}
 }
 
-func (j *Jukebox) Quit() {
-	j.quit <- struct{}{}
-}
+func waitFor[T any](ch <-chan T, match func(e T) bool) error {
+	quit := time.NewTicker(1 * time.Second)
+	defer quit.Stop()
 
-func (j *Jukebox) doUpdateSpeaker(su updateSpeaker) error {
-	j.Lock()
-	defer j.Unlock()
-	if su.index >= len(j.playlist) {
-		j.playing = false
-		speaker.Clear()
-		return nil
-	}
-	j.index = su.index
-	f, err := os.Open(j.playlist[su.index].AbsPath())
-	if err != nil {
-		return err
-	}
-	var streamer beep.Streamer
-	var format beep.Format
-	switch j.playlist[su.index].Ext() {
-	case "mp3":
-		streamer, format, err = mp3.Decode(f)
-	case "flac":
-		streamer, format, err = flac.Decode(f)
-	}
-	if err != nil {
-		return err
-	}
-	j.info = &strmInfo{}
-	j.info.strm = streamer.(beep.StreamSeekCloser)
-	if su.offset != 0 {
-		samples := format.SampleRate.N(time.Second * time.Duration(su.offset))
-		if err := j.info.strm.Seek(samples); err != nil {
-			return err
+	defer time.Sleep(350 * time.Millisecond)
+
+	for {
+		select {
+		case <-quit.C:
+			return ErrMPVTimeout
+		case ev := <-ch:
+			if match(ev) {
+				return nil
+			}
 		}
 	}
-	j.info.ctrlStrmr.Streamer = beep.Resample(
-		4, format.SampleRate,
-		j.sr, j.info.strm,
-	)
-	j.info.format = format
-	speaker.Play(beep.Seq(&j.info.ctrlStrmr, beep.Callback(func() {
-		j.speaker <- updateSpeaker{index: su.index + 1}
-	})))
-	return nil
 }
 
-func (j *Jukebox) SetTracks(tracks []*db.Track) {
-	j.Lock()
-	defer j.Unlock()
-	j.playlist = tracks
-}
-
-func (j *Jukebox) AddTracks(tracks []*db.Track) {
-	j.Lock()
-	if len(j.playlist) == 0 {
-		j.playlist = tracks
-		j.playing = true
-		j.index = 0
-		j.Unlock()
-		j.speaker <- updateSpeaker{index: 0}
-		return
+func tmp() (*os.File, func(), error) {
+	tmp, err := os.CreateTemp("", "")
+	if err != nil {
+		return nil, nil, fmt.Errorf("create temp file: %w", err)
 	}
-	j.playlist = append(j.playlist, tracks...)
-	j.Unlock()
-}
-
-func (j *Jukebox) RemoveTrack(i int) {
-	j.Lock()
-	defer j.Unlock()
-	if i < 0 || i >= len(j.playlist) {
-		return
+	cleanup := func() {
+		os.Remove(tmp.Name())
+		tmp.Close()
 	}
-	j.playlist = append(j.playlist[:i], j.playlist[i+1:]...)
+	return tmp, cleanup, nil
 }
 
-func (j *Jukebox) Skip(i int, offset int) {
-	speaker.Clear()
-	j.Lock()
-	j.index = i
-	j.playing = true
-	j.Unlock()
-	j.speaker <- updateSpeaker{index: j.index, offset: offset}
-}
-
-func (j *Jukebox) ClearTracks() {
-	speaker.Clear()
-	j.Lock()
-	defer j.Unlock()
-	j.playing = false
-	j.playlist = []*db.Track{}
-}
-
-func (j *Jukebox) Stop() {
-	j.Lock()
-	defer j.Unlock()
-	if j.info != nil {
-		j.playing = false
-		j.info.ctrlStrmr.Paused = true
+func find[T any](items []T, f func(T) bool) (T, int) {
+	for i, item := range items {
+		if f(item) {
+			return item, i
+		}
 	}
+	var t T
+	return t, -1
 }
 
-func (j *Jukebox) Start() {
-	if j.info != nil {
-		j.playing = true
-		j.info.ctrlStrmr.Paused = false
+func filter[T comparable](items []T, f func(T) bool) ([]T, bool) {
+	var found bool
+	var ret []T
+	for _, item := range items {
+		if !f(item) {
+			found = true
+			continue
+		}
+		ret = append(ret, item)
 	}
+	return ret, found
 }
 
-func (j *Jukebox) GetStatus() Status {
-	j.Lock()
-	defer j.Unlock()
-	position := 0
-	if j.info != nil {
-		length := j.info.format.SampleRate.D(j.info.strm.Position())
-		position = int(length.Round(time.Millisecond).Seconds())
-	}
-	return Status{
-		CurrentIndex: j.index,
-		Playing:      j.playing,
-		Gain:         0.9,
-		Position:     position,
-	}
-}
-
-func (j *Jukebox) GetTracks() []*db.Track {
-	j.Lock()
-	defer j.Unlock()
-	return j.playlist
+func lock(mu *sync.Mutex) func() {
+	mu.Lock()
+	return mu.Unlock
 }

jukebox/jukebox_test.go

@@ -0,0 +1,187 @@
+package jukebox_test
+
+import (
+	"os"
+	"path/filepath"
+	"sort"
+	"testing"
+
+	"github.com/matryer/is"
+	"go.senan.xyz/gonic/jukebox"
+)
+
+func newJukebox(t *testing.T) *jukebox.Jukebox {
+	sockPath := filepath.Join(t.TempDir(), "mpv.sock")
+
+	j := jukebox.New()
+	err := j.Start(
+		sockPath,
+		[]string{jukebox.MPVArg("--ao", "null")},
+	)
+	if err != nil {
+		t.Fatalf("start jukebox: %v", err)
+	}
+	t.Cleanup(func() {
+		j.Quit()
+	})
+	return j
+}
+
+func TestPlaySkipReset(t *testing.T) {
+	t.Parallel()
+	j := newJukebox(t)
+	is := is.New(t)
+
+	is.NoErr(j.SetPlaylist([]string{
+		testPath("tr_0.mp3"),
+		testPath("tr_1.mp3"),
+		testPath("tr_2.mp3"),
+		testPath("tr_3.mp3"),
+		testPath("tr_4.mp3"),
+	}))
+
+	status, err := j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 0)
+	is.Equal(status.CurrentFilename, testPath("tr_0.mp3"))
+	is.Equal(status.Length, 5)
+	is.Equal(status.Playing, true)
+
+	items, err := j.GetPlaylist()
+	is.NoErr(err)
+
+	itemsSorted := append([]string(nil), items...)
+	sort.Strings(itemsSorted)
+	is.Equal(items, itemsSorted)
+
+	is.NoErr(j.Play())
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.Playing, true)
+
+	is.NoErr(j.Pause())
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.Playing, false)
+
+	is.NoErr(j.Play())
+
+	// skip to 2
+	is.NoErr(j.SkipToPlaylistIndex(2, 0))
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 2)
+	is.Equal(status.CurrentFilename, testPath("tr_2.mp3"))
+	is.Equal(status.Length, 5)
+	is.Equal(status.Playing, true)
+
+	// skip to 3
+	is.NoErr(j.SkipToPlaylistIndex(3, 0))
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 3)
+	is.Equal(status.CurrentFilename, testPath("tr_3.mp3"))
+	is.Equal(status.Length, 5)
+	is.Equal(status.Playing, true)
+
+	// just add one more by overwriting the playlist like some clients do
+	// we should keep the current track unchaned if we find it
+	is.NoErr(j.SetPlaylist([]string{
+		"testdata/tr_0.mp3",
+		"testdata/tr_1.mp3",
+		"testdata/tr_2.mp3",
+		"testdata/tr_3.mp3",
+		"testdata/tr_4.mp3",
+		"testdata/tr_5.mp3",
+	}))
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 3) // index unchanged
+	is.Equal(status.CurrentFilename, testPath("tr_3.mp3"))
+	is.Equal(status.Length, 6) // we added one more track
+	is.Equal(status.Playing, true)
+
+	// skip to 3 again
+	is.NoErr(j.SkipToPlaylistIndex(3, 0))
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 3)
+	is.Equal(status.CurrentFilename, testPath("tr_3.mp3"))
+	is.Equal(status.Length, 6)
+	is.Equal(status.Playing, true)
+
+	// remove all but 3
+	is.NoErr(j.SetPlaylist([]string{
+		"testdata/tr_0.mp3",
+		"testdata/tr_1.mp3",
+		"testdata/tr_2.mp3",
+		"testdata/tr_3.mp3",
+	}))
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 3) // index unchanged
+	is.Equal(status.CurrentFilename, testPath("tr_3.mp3"))
+	is.Equal(status.Length, 4)
+	is.Equal(status.Playing, true)
+
+	// skip to 2 (5s long) in the middle of the track
+	is.NoErr(j.SkipToPlaylistIndex(2, 2))
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 2) // index unchanged
+	is.Equal(status.CurrentFilename, testPath("tr_2.mp3"))
+	is.Equal(status.Length, 4)
+	is.Equal(status.Playing, true)
+	is.Equal(status.Position, 2) // at new position
+
+	// overwrite completely
+	is.NoErr(j.SetPlaylist([]string{
+		"testdata/tr_5.mp3",
+		"testdata/tr_6.mp3",
+		"testdata/tr_7.mp3",
+		"testdata/tr_8.mp3",
+		"testdata/tr_9.mp3",
+	}))
+
+	status, err = j.GetStatus()
+	is.NoErr(err)
+	is.Equal(status.CurrentIndex, 0) // index unchanged
+	is.Equal(status.CurrentFilename, testPath("tr_5.mp3"))
+	is.Equal(status.Length, 5)
+	is.Equal(status.Playing, true)
+}
+
+func TestVolume(t *testing.T) {
+	t.Parallel()
+	j := newJukebox(t)
+	is := is.New(t)
+
+	vol, err := j.GetVolumePct()
+	is.NoErr(err)
+	is.Equal(vol, 100.0)
+
+	is.NoErr(j.SetVolumePct(69.0))
+
+	vol, err = j.GetVolumePct()
+	is.NoErr(err)
+	is.Equal(vol, 69.0)
+
+	is.NoErr(j.SetVolumePct(0.0))
+
+	vol, err = j.GetVolumePct()
+	is.NoErr(err)
+	is.Equal(vol, 0.0)
+}
+
+func testPath(path string) string {
+	cwd, _ := os.Getwd()
+	return filepath.Join(cwd, "testdata", path)
+}

jukebox/testdata/tr_0.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_1.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_2.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_3.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_4.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_5.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_6.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_7.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_8.mp3

@@ -0,0 +1 @@
+5s.mp3

jukebox/testdata/tr_9.mp3

@@ -0,0 +1 @@
+5s.mp3