refactor server startup into jobs

server/ctrladmin/ctrl.go

@@ -83,26 +83,26 @@ type Controller struct {
 	sessDB    *gormstore.Store
 }
 
-func New(base *ctrlbase.Controller) *Controller {
-	sessionKey := []byte(base.DB.GetSetting("session_key"))
+func New(b *ctrlbase.Controller) *Controller {
+	sessionKey := []byte(b.DB.GetSetting("session_key"))
 	if len(sessionKey) == 0 {
 		sessionKey = securecookie.GenerateRandomKey(32)
-		base.DB.SetSetting("session_key", string(sessionKey))
+		b.DB.SetSetting("session_key", string(sessionKey))
 	}
 	tmplBase := template.
 		New("layout").
 		Funcs(sprig.FuncMap()).
 		Funcs(funcMap()).       // static
 		Funcs(template.FuncMap{ // from base
-			"path": base.Path,
+			"path": b.Path,
 		})
 	tmplBase = extendFromPaths(tmplBase, prefixPartials)
 	tmplBase = extendFromPaths(tmplBase, prefixLayouts)
-	sessDB := gormstore.New(base.DB.DB, sessionKey)
+	sessDB := gormstore.New(b.DB.DB, sessionKey)
 	sessDB.SessionOpts.HttpOnly = true
 	sessDB.SessionOpts.SameSite = http.SameSiteLaxMode
 	return &Controller{
-		Controller: base,
+		Controller: b,
 		buffPool:   bpool.NewBufferPool(64),
 		templates:  pagesFromPaths(tmplBase, prefixPages),
 		sessDB:     sessDB,

server/ctrlbase/ctrl.go

@@ -7,7 +7,6 @@ import (
 	"path"
 
 	"go.senan.xyz/gonic/db"
-	"go.senan.xyz/gonic/jukebox"
 	"go.senan.xyz/gonic/scanner"
 )
 
@@ -50,7 +49,6 @@ type Controller struct {
 	MusicPath   string
 	Scanner     *scanner.Scanner
 	ProxyPrefix string
-	Jukebox     *jukebox.Jukebox
 }
 
 // Path returns a URL path with the proxy prefix included

server/ctrlsubsonic/ctrl.go

@@ -12,6 +12,7 @@ import (
 	"go.senan.xyz/gonic/server/ctrlbase"
 	"go.senan.xyz/gonic/server/ctrlsubsonic/params"
 	"go.senan.xyz/gonic/server/ctrlsubsonic/spec"
+	"go.senan.xyz/gonic/server/jukebox"
 )
 
 type CtxKey int
@@ -24,14 +25,8 @@ const (
 
 type Controller struct {
 	*ctrlbase.Controller
-	cachePath string
-}
-
-func New(base *ctrlbase.Controller, cachePath string) *Controller {
-	return &Controller{
-		Controller: base,
-		cachePath:  cachePath,
-	}
+	CachePath string
+	Jukebox   *jukebox.Jukebox
 }
 
 type metaResponse struct {

server/ctrlsubsonic/handlers_common.go

@@ -317,22 +317,23 @@ func (c *Controller) ServeGetRandomSongs(r *http.Request) *spec.Response {
 
 func (c *Controller) ServeJukebox(r *http.Request) *spec.Response {
 	params := r.Context().Value(CtxParams).(params.Params)
-	switch params.Get("action") {
-	case "set":
+	getTracks := func() []*db.Track {
 		var tracks []*db.Track
 		ids := params.GetFirstListInt("id")
-		if len(ids) == 0 {
-			c.Jukebox.ClearTracks()
-		}
 		for _, id := range ids {
 			track := &db.Track{}
-			err := c.DB.Preload("Album").First(track, id).Error
-			if err != nil {
-				return spec.NewError(10, "couldn't find tracks with provided ids")
+			c.DB.Preload("Album").First(track, id)
+			if track.ID != 0 {
+				tracks = append(tracks, track)
 			}
-			tracks = append(tracks, track)
 		}
-		c.Jukebox.SetTracks(tracks)
+		return tracks
+	}
+	switch act := params.Get("action"); act {
+	case "set":
+		c.Jukebox.SetTracks(getTracks())
+	case "add":
+		c.Jukebox.AddTracks(getTracks())
 	case "clear":
 		c.Jukebox.ClearTracks()
 	case "remove":
@@ -356,8 +357,10 @@ func (c *Controller) ServeJukebox(r *http.Request) *spec.Response {
 		sub := spec.NewResponse()
 		sub.JukeboxPlaylist = c.Jukebox.GetTracks()
 		return sub
+	default:
+		return spec.NewError(10, "unknown value `%s` for parameter 'action'", act)
 	}
-	// All actions except get are expected to return a status
+	// all actions except get are expected to return a status
 	sub := spec.NewResponse()
 	sub.JukeboxStatus = c.Jukebox.Status()
 	return sub

server/ctrlsubsonic/handlers_raw.go

@@ -136,7 +136,7 @@ func (c *Controller) ServeStream(w http.ResponseWriter, r *http.Request) *spec.R
 	}
 	servOpts.pref = pref
 	servOpts.maxBitrate = params.GetIntOr("maxBitRate", 0)
-	servOpts.cachePath = c.cachePath
+	servOpts.cachePath = c.CachePath
 	serveTrackEncode(w, r, servOpts)
 	return nil
 }

server/jukebox/jukebox.go

@@ -0,0 +1,251 @@
+package jukebox
+
+import (
+	"fmt"
+	"os"
+	"path"
+	"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"
+	"go.senan.xyz/gonic/server/ctrlsubsonic/spec"
+)
+
+type strmInfo struct {
+	ctrlStrmr beep.Ctrl
+	strm      beep.StreamSeekCloser
+	format    beep.Format
+}
+
+type Jukebox struct {
+	playlist  []*db.Track
+	musicPath string
+	index     int
+	playing   bool
+	sr        beep.SampleRate
+	// used to notify the player to re read the members
+	updates chan struct{}
+	quit    chan struct{}
+	done    chan bool
+	info    *strmInfo
+	sync.Mutex
+}
+
+func New(musicPath string) *Jukebox {
+	return &Jukebox{
+		musicPath: musicPath,
+		sr:        beep.SampleRate(48000),
+		updates:   make(chan struct{}),
+		done:      make(chan bool),
+		quit:      make(chan struct{}),
+	}
+}
+
+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)
+	}
+	for {
+		select {
+		case <-j.quit:
+			return nil
+		case <-j.updates:
+			j.doUpdate()
+		}
+	}
+}
+
+func (j *Jukebox) Quit() {
+	j.quit <- struct{}{}
+}
+
+func (j *Jukebox) doUpdate() {
+	var streamer beep.Streamer
+	var format beep.Format
+	if j.index >= len(j.playlist) {
+		j.Lock()
+		j.index = 0
+		j.playing = false
+		j.Unlock()
+		return
+	}
+	j.Lock()
+	f, err := os.Open(path.Join(
+		j.musicPath,
+		j.playlist[j.index].RelPath(),
+	))
+	j.Unlock()
+	if err != nil {
+		j.incIndex()
+		return
+	}
+	switch j.playlist[j.index].Ext() {
+	case "mp3":
+		streamer, format, err = mp3.Decode(f)
+	case "flac":
+		streamer, format, err = flac.Decode(f)
+	default:
+		j.incIndex()
+		return
+	}
+	if err != nil {
+		j.incIndex()
+		return
+	}
+	if j.playing {
+		j.Lock()
+		{
+			j.info = &strmInfo{}
+			j.info.strm = streamer.(beep.StreamSeekCloser)
+			j.info.ctrlStrmr.Streamer = beep.Resample(
+				4, format.SampleRate,
+				j.sr, j.info.strm,
+			)
+			j.info.format = format
+		}
+		j.Unlock()
+		speaker.Play(beep.Seq(&j.info.ctrlStrmr, beep.Callback(func() {
+			j.done <- false
+		})))
+		if v := <-j.done; v {
+			return
+		}
+		j.Lock()
+		j.index++
+		if j.index >= len(j.playlist) {
+			j.index = 0
+			j.playing = false
+			j.Unlock()
+			return
+		}
+		j.Unlock()
+		// in a go routine as otherwise this hangs as the
+		go func() {
+			j.updates <- struct{}{}
+		}()
+	}
+}
+
+func (j *Jukebox) incIndex() {
+	j.Lock()
+	defer j.Unlock()
+	j.index++
+}
+
+func (j *Jukebox) SetTracks(tracks []*db.Track) {
+	j.Lock()
+	defer j.Unlock()
+	j.index = 0
+	if len(tracks) == 0 {
+		if j.playing {
+			j.done <- true
+		}
+		j.playing = false
+		j.playlist = []*db.Track{}
+		speaker.Clear()
+		return
+	}
+	if j.playing {
+		j.playlist = tracks
+		j.done <- true
+		speaker.Clear()
+		j.updates <- struct{}{}
+		return
+	}
+	j.playlist = tracks
+	j.playing = true
+	j.updates <- struct{}{}
+}
+
+func (j *Jukebox) AddTracks(tracks []*db.Track) {
+	j.Lock()
+	j.playlist = append(j.playlist, tracks...)
+	j.Unlock()
+}
+
+func (j *Jukebox) ClearTracks() {
+	j.Lock()
+	j.index = 0
+	j.playing = false
+	j.playlist = []*db.Track{}
+	j.Unlock()
+}
+
+func (j *Jukebox) RemoveTrack(i int) {
+	j.Lock()
+	defer j.Unlock()
+	if i < 0 || i > len(j.playlist) {
+		return
+	}
+	j.playlist = append(j.playlist[:i], j.playlist[i+1:]...)
+}
+
+func (j *Jukebox) Status() *spec.JukeboxStatus {
+	position := 0
+	if j.info != nil {
+		length := j.info.format.SampleRate.D(j.info.strm.Position())
+		position = int(length.Round(time.Millisecond).Seconds())
+	}
+	return &spec.JukeboxStatus{
+		CurrentIndex: j.index,
+		Playing:      j.playing,
+		Gain:         0.9,
+		Position:     position,
+	}
+}
+
+func (j *Jukebox) GetTracks() *spec.JukeboxPlaylist {
+	j.Lock()
+	defer j.Unlock()
+	jb := &spec.JukeboxPlaylist{}
+	jb.List = make([]*spec.TrackChild, len(j.playlist))
+	for i, track := range j.playlist {
+		jb.List[i] = spec.NewTrackByTags(track, track.Album)
+	}
+	jb.CurrentIndex = j.index
+	jb.Playing = j.playing
+	jb.Gain = 0.9
+	jb.Position = 0
+	if j.info != nil {
+		length := j.info.format.SampleRate.D(j.info.strm.Position())
+		jb.Position = int(length.Round(time.Millisecond).Seconds())
+	}
+	return jb
+}
+
+func (j *Jukebox) Stop() {
+	j.Lock()
+	j.playing = false
+	j.info.ctrlStrmr.Paused = true
+	j.Unlock()
+}
+
+func (j *Jukebox) Start() {
+	j.Lock()
+	j.playing = true
+	j.info.ctrlStrmr.Paused = false
+	j.Unlock()
+}
+
+func (j *Jukebox) Skip(i int, skipCurrent bool) {
+	j.Lock()
+	defer j.Unlock()
+	if i == j.index {
+		return
+	}
+	if skipCurrent {
+		j.index++
+	} else {
+		j.index = i
+	}
+	speaker.Clear()
+	if j.playing {
+		j.done <- true
+	}
+	j.updates <- struct{}{}
+}

server/server.go

@@ -11,27 +11,25 @@ import (
 	"github.com/gorilla/mux"
 
 	"go.senan.xyz/gonic/db"
-	"go.senan.xyz/gonic/jukebox"
 	"go.senan.xyz/gonic/scanner"
 	"go.senan.xyz/gonic/server/assets"
 	"go.senan.xyz/gonic/server/ctrladmin"
 	"go.senan.xyz/gonic/server/ctrlbase"
 	"go.senan.xyz/gonic/server/ctrlsubsonic"
+	"go.senan.xyz/gonic/server/jukebox"
 )
 
 type Options struct {
-	DB           *db.DB
-	MusicPath    string
-	CachePath    string
-	ListenAddr   string
-	ScanInterval time.Duration
-	ProxyPrefix  string
+	DB          *db.DB
+	MusicPath   string
+	CachePath   string
+	ProxyPrefix string
 }
 
 type Server struct {
-	*http.Server
-	scanner      *scanner.Scanner
-	scanInterval time.Duration
+	scanner *scanner.Scanner
+	jukebox *jukebox.Jukebox
+	router  *mux.Router
 }
 
 func New(opts Options) *Server {
@@ -39,7 +37,8 @@ func New(opts Options) *Server {
 	opts.MusicPath = filepath.Clean(opts.MusicPath)
 	opts.CachePath = filepath.Clean(opts.CachePath)
 	// ** begin controllers
-	scanner := scanner.New(opts.DB, opts.MusicPath)
+	scanner := scanner.New(opts.MusicPath, opts.DB)
+	jukebox := jukebox.New(opts.MusicPath)
 	// the base controller, it's fields/middlewares are embedded/used by the
 	// other two admin ui and subsonic controllers
 	base := &ctrlbase.Controller{
@@ -47,30 +46,25 @@ func New(opts Options) *Server {
 		MusicPath:   opts.MusicPath,
 		ProxyPrefix: opts.ProxyPrefix,
 		Scanner:     scanner,
-		Jukebox:     &jukebox.Jukebox{},
 	}
-	base.Jukebox.Init(opts.MusicPath)
 	// router with common wares for admin / subsonic
 	r := mux.NewRouter()
 	r.Use(base.WithLogging)
 	r.Use(base.WithCORS)
-	setupMisc(r, base)
-	setupAdminRouter := r.PathPrefix("/admin").Subrouter()
-	setupAdmin(setupAdminRouter, ctrladmin.New(base))
-	setupSubsonicRouter := r.PathPrefix("/rest").Subrouter()
-	setupSubsonic(setupSubsonicRouter, ctrlsubsonic.New(base, opts.CachePath))
-	//
-	server := &http.Server{
-		Addr:         opts.ListenAddr,
-		Handler:      r,
-		ReadTimeout:  5 * time.Second,
-		WriteTimeout: 80 * time.Second,
-		IdleTimeout:  60 * time.Second,
+	ctrlAdmin := ctrladmin.New(base)
+	ctrlSubsonic := &ctrlsubsonic.Controller{
+		Controller: base,
+		CachePath:  opts.CachePath,
+		Jukebox:    jukebox,
 	}
+	setupMisc(r, base)
+	setupAdmin(r.PathPrefix("/admin").Subrouter(), ctrlAdmin)
+	setupSubsonic(r.PathPrefix("/rest").Subrouter(), ctrlSubsonic)
+	//
 	return &Server{
-		Server:       server,
-		scanner:      scanner,
-		scanInterval: opts.ScanInterval,
+		scanner: scanner,
+		jukebox: jukebox,
+		router:  r,
 	}
 }
 
@@ -183,17 +177,54 @@ func setupSubsonic(r *mux.Router, ctrl *ctrlsubsonic.Controller) {
 	r.NotFoundHandler = notFoundRoute.GetHandler()
 }
 
-func (s *Server) Start() error {
-	if s.scanInterval > 0 {
-		log.Printf("will be scanning at intervals of %s", s.scanInterval)
-		ticker := time.NewTicker(s.scanInterval)
-		go func() {
-			for range ticker.C {
+type funcExecute func() error
+type funcInterrupt func(error)
+
+func (s *Server) StartHTTP(listenAddr string) (funcExecute, funcInterrupt) {
+	log.Print("starting job 'http'\n")
+	list := &http.Server{
+		Addr:         listenAddr,
+		Handler:      s.router,
+		ReadTimeout:  5 * time.Second,
+		WriteTimeout: 80 * time.Second,
+		IdleTimeout:  60 * time.Second,
+	}
+	execute := func() error {
+		return list.ListenAndServe()
+	}
+	return execute, func(_ error) {
+		list.Close()
+	}
+}
+
+func (s *Server) StartScanTicker(dur time.Duration) (funcExecute, funcInterrupt) {
+	log.Printf("starting job 'scan timer'\n")
+	ticker := time.NewTicker(dur)
+	done := make(chan struct{})
+	execute := func() error {
+		for {
+			select {
+			case <-done:
+				return nil
+			case <-ticker.C:
 				if err := s.scanner.Start(); err != nil {
-					log.Printf("error while scanner: %v", err)
+					log.Printf("error scanning: %v", err)
 				}
 			}
-		}()
+		}
+	}
+	return execute, func(_ error) {
+		ticker.Stop()
+		done <- struct{}{}
+	}
+}
+
+func (s *Server) StartJukebox() (funcExecute, funcInterrupt) {
+	log.Printf("starting job 'jukebox'\n")
+	execute := func() error {
+		return s.jukebox.Listen()
+	}
+	return execute, func(_ error) {
+		s.jukebox.Quit()
 	}
-	return s.ListenAndServe()
 }