diff --git a/backend/audiocache.go b/backend/audiocache.go index 965d14d..ee9141d 100644 --- a/backend/audiocache.go +++ b/backend/audiocache.go @@ -59,6 +59,16 @@ func (a *AudioCache) PathForCachedFile(id string) string { return "" } +// PathForCachedFile returns the local filesystem path for a cached track, +// including one that is in the process of downloading. +// If it is not cached or downloading, it returns an empty string. +func (a *AudioCache) PathForCachedOrDownloadingFile(id string) string { + if _, ok := a.entries[id]; ok { + return a.pathForID(id) + } + return "" +} + // cacheFile begins downloading a file (if not already downloading) and stores it // to the cache directory under its ID as filename. The download is asynchronous. func (a *AudioCache) cacheFile(id, dlURL string) { diff --git a/backend/playbackengine.go b/backend/playbackengine.go index cd7a463..d1123ed 100644 --- a/backend/playbackengine.go +++ b/backend/playbackengine.go @@ -89,16 +89,17 @@ type playbackEngine struct { replayGainCfg ReplayGainConfig // registered callbacks - onSongChange []func(nowPlaying mediaprovider.MediaItem, justScrobbledIfAny *mediaprovider.Track) - onPlayTimeUpdate []func(float64, float64, bool) - onLoopModeChange []func(LoopMode) - onVolumeChange []func(int) - onSeek []func() - onPaused []func() - onStopped []func() - onPlaying []func() - onPlayerChange []func() - onQueueChange []func() + onBeforeSongChange []func(next mediaprovider.MediaItem) + onSongChange []func(nowPlaying mediaprovider.MediaItem, justScrobbledIfAny *mediaprovider.Track) + onPlayTimeUpdate []func(float64, float64, bool) + onLoopModeChange []func(LoopMode) + onVolumeChange []func(int) + onSeek []func() + onPaused []func() + onStopped []func() + onPlaying []func() + onPlayerChange []func() + onQueueChange []func() } func NewPlaybackEngine( @@ -246,8 +247,7 @@ func (p *playbackEngine) SetLoopMode(loopMode LoopMode) { if p.nowPlayingIdx >= 0 { // TODO - don't need when going from LoopNone to LoopAll // if not on last track - p.needToSetNextTrack = true - //p.setNextTrackBasedOnLoopMode(true) + p.handleNextTrackUpdated() } for _, cb := range p.onLoopModeChange { @@ -368,7 +368,9 @@ func (p *playbackEngine) doLoaditems(items []mediaprovider.MediaItem, insertQueu p.nowPlayingIdx = -1 p.playQueue = nil } - p.needToSetNextTrack = len(items) > 0 && (insertQueueMode == InsertNext || (insertQueueMode == Append && p.nowPlayingIdx == len(p.playQueue)-1)) + if nextChanged := len(items) > 0 && (insertQueueMode != Append || (p.nowPlayingIdx == len(p.playQueue)-1)); nextChanged { + defer p.handleNextTrackUpdated() + } if shuffle { rand.Shuffle(len(items), func(i, j int) { items[i], items[j] = items[j], items[i] }) @@ -380,10 +382,6 @@ func (p *playbackEngine) doLoaditems(items []mediaprovider.MediaItem, insertQueu } p.playQueue = append(p.playQueue[:insertIdx], append(items, p.playQueue[insertIdx:]...)...) - //if needToSetNext { - // p.setNextTrack(p.nowPlayingIdx + 1) - //} - p.invokeNoArgCallbacks(p.onQueueChange) return nil } @@ -394,7 +392,9 @@ func (p *playbackEngine) LoadRadioStation(radio *mediaprovider.RadioStation, ins p.nowPlayingIdx = -1 p.playQueue = nil } - p.needToSetNextTrack = insertMode == InsertNext || (insertMode == Append && p.nowPlayingIdx == len(p.playQueue)-1) + if nextChanged := insertMode == InsertNext || (insertMode == Append && p.nowPlayingIdx == len(p.playQueue)-1); nextChanged { + p.handleNextTrackUpdated() + } insertIdx := len(p.playQueue) if insertMode == InsertNext { insertIdx = p.nowPlayingIdx + 1 @@ -406,10 +406,6 @@ func (p *playbackEngine) LoadRadioStation(radio *mediaprovider.RadioStation, ins copy(new[len(firstHalf)+1:], p.playQueue[insertIdx:]) p.playQueue = new - //if needToSetNext { - // p.setNextTrack(p.nowPlayingIdx + 1) - //} - p.invokeNoArgCallbacks(p.onQueueChange) } @@ -468,11 +464,10 @@ func (p *playbackEngine) UpdatePlayQueue(items []mediaprovider.MediaItem) error if p.nowPlayingIdx >= 0 && newNowPlayingIdx == -1 { return p.Stop() } - p.needToSetNextTrack = p.nowPlayingIdx >= 0 + if p.nowPlayingIdx >= 0 { + p.handleNextTrackUpdated() + } p.nowPlayingIdx = newNowPlayingIdx - //if needToUpdateNext { - // p.setNextTrackBasedOnLoopMode(false) - //} p.invokeNoArgCallbacks(p.onQueueChange) return nil @@ -518,8 +513,7 @@ func (p *playbackEngine) RemoveTracksFromQueue(idxs []int) { // when we receive new track event from player } else if isNextPlayingTrackremoved { if newNowPlaying < len(newQueue)-1 { - p.needToSetNextTrack = true - //p.setNextTrack(p.nowPlayingIdx + 1) + p.handleNextTrackUpdated() } else { // no next track to play p.setNextTrack(-1) @@ -620,8 +614,7 @@ func (p *playbackEngine) handleOnTrackChange() { p.sendNowPlayingScrobble() // Must come before invokeOnChangeCallbacks b/c track may immediately be scrobbled p.invokeOnSongChangeCallbacks() p.handleTimePosUpdate(false) - p.cacheNextTracks() - p.setNextTrackBasedOnLoopMode(false) + p.handleNextTrackUpdated() } func (p *playbackEngine) handleOnStopped() { @@ -638,25 +631,35 @@ func (p *playbackEngine) handleOnStopped() { p.nowPlayingIdx = -1 } -func (p *playbackEngine) setNextTrackBasedOnLoopMode(onLoopModeChange bool) { +// to be invoked as soon as the next item in the queue that should play changes +func (p *playbackEngine) handleNextTrackUpdated() { + p.cacheNextTracks() + p.needToSetNextTrack = true + for _, cb := range p.onBeforeSongChange { + var item mediaprovider.MediaItem + if idx := p.nextPlayingIndex(); idx >= 0 { + item = p.playQueue[idx] + } + cb(item) + } +} + +func (p *playbackEngine) nextPlayingIndex() int { switch p.loopMode { case LoopNone: - if p.nowPlayingIdx < len(p.playQueue)-1 { - p.setNextTrack(p.nowPlayingIdx + 1) - } else if onLoopModeChange { - // prev was LoopOne - need to erase next track - p.setNextTrack(-1) + if p.nowPlayingIdx >= len(p.playQueue)-1 { + return -1 } + return p.nowPlayingIdx + 1 case LoopOne: - p.setNextTrack(p.nowPlayingIdx) + return p.nowPlayingIdx case LoopAll: if p.nowPlayingIdx >= len(p.playQueue)-1 { - p.setNextTrack(0) - } else if !onLoopModeChange { - // if onloopmodechange, prev mode was LoopNone and next track is already set - p.setNextTrack(p.nowPlayingIdx + 1) + return 0 } + return p.nowPlayingIdx + 1 } + return -1 // unreached } func (p *playbackEngine) setTrack(idx int, next bool, startTime float64) error { @@ -841,7 +844,7 @@ func (p *playbackEngine) handleTimePosUpdate(seeked bool) { isNearEnd := meta.Type != mediaprovider.MediaItemTypeRadioStation && s.TimePos > float64(meta.Duration)-10 if p.needToSetNextTrack && isNearEnd { p.needToSetNextTrack = false - p.setNextTrackBasedOnLoopMode(false) + p.setNextTrack(p.nextPlayingIndex()) } if p.callbacksDisabled { return diff --git a/backend/playbackmanager.go b/backend/playbackmanager.go index 21abcd9..3b7c2fe 100644 --- a/backend/playbackmanager.go +++ b/backend/playbackmanager.go @@ -47,6 +47,13 @@ type RemotePlaybackDevice struct { new func() (player.BasePlayer, error) } +var zeroWaveformImage *WaveformImage + +func init() { + zeroWaveformImage = NewWaveformImage() + GenerateWaveformImage(&WaveformData{}, zeroWaveformImage, color.White) +} + func NewPlaybackManager( ctx context.Context, s *ServerManager, @@ -81,30 +88,68 @@ func (p *PlaybackManager) addOnTrackChangeHook() { p.lastPlayTime = curTime }) + var nextWaveformImgLock sync.Mutex + var nextWaveformImg *WaveformImage + var nextWaveformImgID string + var cancel context.CancelFunc + + p.engine.onBeforeSongChange = append(p.engine.onBeforeSongChange, func(item mediaprovider.MediaItem) { + if p.cache != nil && item != nil && item.Metadata().Type == mediaprovider.MediaItemTypeTrack { + log.Println("preparing waveform image for next track ", item.Metadata().ID) + nextWaveformImgLock.Lock() + if cancel != nil { + cancel() + } + nextWaveformImgLock.Unlock() + id := item.Metadata().ID + path := p.cache.PathForCachedOrDownloadingFile(id) + go func() { + ctx, cncl := context.WithCancel(p.engine.ctx) + nextWaveformImgLock.Lock() + cancel = cncl + nextWaveformImgLock.Unlock() + wd, err := GetWaveformDataForFile(ctx, path, func() bool { + return p.cache.PathForCachedFile(id) != "" + }) + if err != nil { + log.Println(err.Error()) + } else { + im := NewWaveformImage() + GenerateWaveformImage(wd, im, color.White) + + if ctx.Err() != nil { + log.Println("canceled") + } + log.Println("have waveform image for track %s", item.Metadata().ID) + nextWaveformImgLock.Lock() + defer nextWaveformImgLock.Unlock() + nextWaveformImg = im + nextWaveformImgID = item.Metadata().ID + } + }() + } + }) + p.OnSongChange(func(item mediaprovider.MediaItem, _ *mediaprovider.Track) { // Autoplay if enabled and we are on the last track if p.autoplay && p.NowPlayingIndex() == len(p.engine.playQueue)-1 { p.enqueueAutoplayTracks() } - // TODO: make more permanent - if p.cache != nil && item != nil { - go func() { - if path := p.cache.PathForCachedFile(item.Metadata().ID); path != "" { - wd, err := GetWaveformDataForFile(context.Background(), path) - if err != nil { - log.Println(err.Error()) - } else { - im := NewWaveformImage() - GenerateWaveformImage(wd, im, color.White) - for _, cb := range p.onWaveformImgUpdate { - cb(im) - } - } - } else { - log.Println("no cached file for waveform image") - } - }() + if item != nil { + log.Println("Playing track ", item.Metadata().ID) + var im *WaveformImage + nextWaveformImgLock.Lock() + if nextWaveformImgID == item.Metadata().ID { + im = nextWaveformImg + } + nextWaveformImgLock.Unlock() + if im == nil { + im = zeroWaveformImage + } + for _, cb := range p.onWaveformImgUpdate { + cb(im) + } } if runtime.GOOS != "windows" { diff --git a/backend/util/filestreamer.go b/backend/util/filestreamer.go new file mode 100644 index 0000000..545ed51 --- /dev/null +++ b/backend/util/filestreamer.go @@ -0,0 +1,110 @@ +package util + +import ( + "context" + "io" + "log" + "net" + "net/http" + "os" + "time" +) + +type FileStreamerServer struct { + Path string + IsComplete func() bool + listener net.Listener + server *http.Server + done chan struct{} +} + +// NewFileStreamerServer creates a new server but doesn't start it yet. +func NewFileStreamerServer(path string, isComplete func() bool) (*FileStreamerServer, error) { + listener, err := net.Listen("tcp", ":0") + if err != nil { + return nil, err + } + + fs := &FileStreamerServer{ + Path: path, + IsComplete: isComplete, + listener: listener, + done: make(chan struct{}), + } + + mux := http.NewServeMux() + mux.HandleFunc("/stream", fs.streamHandler) + + fs.server = &http.Server{ + Handler: mux, + } + + return fs, nil +} + +// Addr returns the server address (host:port). +func (fs *FileStreamerServer) Addr() string { + _, port, _ := net.SplitHostPort(fs.listener.Addr().String()) + return "http://localhost:" + port + "/stream" +} + +// Serve starts serving and waits for a single request to complete. +func (fs *FileStreamerServer) Serve() error { + go func() { + _ = fs.server.Serve(fs.listener) + }() + + <-fs.done // wait for the handler to finish + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + return fs.server.Shutdown(ctx) +} + +// Handler that streams the file using chunked transfer encoding. +func (fs *FileStreamerServer) streamHandler(w http.ResponseWriter, r *http.Request) { + defer close(fs.done) // signal Serve() to shut down after this request + + file, err := os.Open(fs.Path) + if err != nil { + http.Error(w, "could not open file", http.StatusInternalServerError) + return + } + defer file.Close() + + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Transfer-Encoding", "chunked") + w.WriteHeader(http.StatusOK) + + flusher, canFlush := w.(http.Flusher) + + buf := make([]byte, 4096) + for { + n, err := file.Read(buf) + if err != nil && err != io.EOF { + log.Printf("read error: %v", err) + break + } + + if n > 0 { + _, err := w.Write(buf[:n]) + if err != nil { + log.Printf("client write error: %v", err) + break + } + if canFlush { + flusher.Flush() + } + } + + if n == 0 && fs.IsComplete() { + break + } + + // Wait for more content to be written to the source file + if n == 0 { + time.Sleep(50 * time.Millisecond) + } + } +} diff --git a/backend/waveformimage.go b/backend/waveformimage.go index 201f4a3..26edad4 100644 --- a/backend/waveformimage.go +++ b/backend/waveformimage.go @@ -6,10 +6,12 @@ import ( "image" "image/color" "io" + "log" "math" "os" "path/filepath" + "github.com/dweymouth/supersonic/backend/util" "github.com/go-audio/audio" "github.com/go-audio/wav" "github.com/supersonic-app/go-mpv" @@ -71,10 +73,20 @@ func GenerateWaveformImage(data *WaveformData, imgbuf *WaveformImage, c color.Co } } -func GetWaveformDataForFile(ctx context.Context, fpath string) (*WaveformData, error) { +func GetWaveformDataForFile(ctx context.Context, fpath string, fileIsDone func() bool) (*WaveformData, error) { dir := filepath.Dir(fpath) transcodeFile := filepath.Join(dir, filepath.Base(fpath)+"_waveform.wav") + if !fileIsDone() { + srv, err := util.NewFileStreamerServer(fpath, fileIsDone) + if err != nil { + return nil, err + } + fpath = srv.Addr() + log.Println("streaming file to MPV at ", fpath) + go srv.Serve() + } + err := convertToWav(ctx, fpath, transcodeFile) if err != nil { return nil, err @@ -82,6 +94,7 @@ func GetWaveformDataForFile(ctx context.Context, fpath string) (*WaveformData, e f, err := os.Open(transcodeFile) if err != nil { + log.Println("error opening transcoded file") return nil, err } defer f.Close() @@ -199,7 +212,6 @@ func convertToWav(ctx context.Context, inPath, outPath string) error { m.Command([]string{"loadfile", inPath, "replace"}) - //log.Println("generating wav file from %s using MPV", inPath) return mpvWaitForIdle(ctx, m) }