mirror of
https://github.com/navidrome/navidrome.git
synced 2026-10-18 07:30:29 +00:00
Merge e25071f87e3d744bb19fda2ec253f9bc13c51b0f into 50633f839dabe49571099977753036d166734982
This commit is contained in:
commit
c50d2879dd
@ -21,6 +21,7 @@ import (
|
||||
"github.com/navidrome/navidrome/core/metrics"
|
||||
"github.com/navidrome/navidrome/core/playback"
|
||||
"github.com/navidrome/navidrome/core/playlists"
|
||||
"github.com/navidrome/navidrome/core/radio"
|
||||
"github.com/navidrome/navidrome/core/scrobbler"
|
||||
"github.com/navidrome/navidrome/core/sonic"
|
||||
"github.com/navidrome/navidrome/core/stream"
|
||||
@ -83,7 +84,9 @@ func CreateNativeAPIRouter(ctx context.Context) *nativeapi.Router {
|
||||
library := core.NewLibrary(dataStore, modelScanner, watcher, broker, manager)
|
||||
user := core.NewUser(dataStore, manager)
|
||||
maintenance := core.NewMaintenance(dataStore)
|
||||
router := nativeapi.New(dataStore, share, playlistsPlaylists, insights, library, user, maintenance, manager, imageUploadService)
|
||||
titlePublisher := events.NewRadioMetadataPublisher(broker)
|
||||
metadataManager := radio.NewMetadataManagerService(titlePublisher)
|
||||
router := nativeapi.New(dataStore, share, playlistsPlaylists, insights, library, user, maintenance, manager, imageUploadService, metadataManager)
|
||||
return router
|
||||
}
|
||||
|
||||
@ -247,7 +250,7 @@ func getPluginManager() *plugins.Manager {
|
||||
|
||||
// wire_injectors.go:
|
||||
|
||||
var allProviders = wire.NewSet(core.Set, artwork.Set, server.New, subsonic.New, jellyfin.New, nativeapi.New, public.New, persistence.New, lastfm.NewRouter, listenbrainz.NewRouter, events.GetBroker, scanner.New, scanner.GetWatcher, metrics.GetPrometheusInstance, db.Db, plugins.GetManager, sonic.New, wire.Bind(new(agents.PluginLoader), new(*plugins.Manager)), wire.Bind(new(scrobbler.PluginLoader), new(*plugins.Manager)), wire.Bind(new(lyrics.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.Engine), new(*sonic.Sonic)), wire.Bind(new(nativeapi.PluginManager), new(*plugins.Manager)), wire.Bind(new(core.PluginUnloader), new(*plugins.Manager)), wire.Bind(new(plugins.PluginMetricsRecorder), new(metrics.Metrics)), wire.Bind(new(core.Watcher), new(scanner.Watcher)))
|
||||
var allProviders = wire.NewSet(core.Set, artwork.Set, server.New, subsonic.New, jellyfin.New, nativeapi.New, public.New, persistence.New, lastfm.NewRouter, listenbrainz.NewRouter, events.GetBroker, events.NewRadioMetadataPublisher, scanner.New, scanner.GetWatcher, metrics.GetPrometheusInstance, db.Db, plugins.GetManager, sonic.New, wire.Bind(new(agents.PluginLoader), new(*plugins.Manager)), wire.Bind(new(scrobbler.PluginLoader), new(*plugins.Manager)), wire.Bind(new(lyrics.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.PluginLoader), new(*plugins.Manager)), wire.Bind(new(sonic.Engine), new(*sonic.Sonic)), wire.Bind(new(nativeapi.PluginManager), new(*plugins.Manager)), wire.Bind(new(nativeapi.RadioMetadataManager), new(*radio.MetadataManager)), wire.Bind(new(core.PluginUnloader), new(*plugins.Manager)), wire.Bind(new(plugins.PluginMetricsRecorder), new(metrics.Metrics)), wire.Bind(new(core.Watcher), new(scanner.Watcher)))
|
||||
|
||||
func GetPluginManager(ctx context.Context) *plugins.Manager {
|
||||
manager := getPluginManager()
|
||||
|
||||
@ -14,6 +14,7 @@ import (
|
||||
"github.com/navidrome/navidrome/core/lyrics"
|
||||
"github.com/navidrome/navidrome/core/metrics"
|
||||
"github.com/navidrome/navidrome/core/playback"
|
||||
"github.com/navidrome/navidrome/core/radio"
|
||||
"github.com/navidrome/navidrome/core/scrobbler"
|
||||
"github.com/navidrome/navidrome/core/sonic"
|
||||
"github.com/navidrome/navidrome/db"
|
||||
@ -41,6 +42,7 @@ var allProviders = wire.NewSet(
|
||||
lastfm.NewRouter,
|
||||
listenbrainz.NewRouter,
|
||||
events.GetBroker,
|
||||
events.NewRadioMetadataPublisher,
|
||||
scanner.New,
|
||||
scanner.GetWatcher,
|
||||
metrics.GetPrometheusInstance,
|
||||
@ -53,6 +55,7 @@ var allProviders = wire.NewSet(
|
||||
wire.Bind(new(sonic.PluginLoader), new(*plugins.Manager)),
|
||||
wire.Bind(new(sonic.Engine), new(*sonic.Sonic)),
|
||||
wire.Bind(new(nativeapi.PluginManager), new(*plugins.Manager)),
|
||||
wire.Bind(new(nativeapi.RadioMetadataManager), new(*radio.MetadataManager)),
|
||||
wire.Bind(new(core.PluginUnloader), new(*plugins.Manager)),
|
||||
wire.Bind(new(plugins.PluginMetricsRecorder), new(metrics.Metrics)),
|
||||
wire.Bind(new(core.Watcher), new(scanner.Watcher)),
|
||||
|
||||
125
core/radio/icy/parser.go
Normal file
125
core/radio/icy/parser.go
Normal file
@ -0,0 +1,125 @@
|
||||
package icy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"strings"
|
||||
"unicode/utf8"
|
||||
|
||||
"golang.org/x/text/encoding/charmap"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrInvalidMetaInt = errors.New("invalid icy metadata interval")
|
||||
ErrMissingTitleHandler = errors.New("missing icy title handler")
|
||||
errEndOfStream = errors.New("icy stream ended")
|
||||
)
|
||||
|
||||
const (
|
||||
maxMetaInt = 1024 * 1024
|
||||
maxMetadataLen = 255 * 16
|
||||
)
|
||||
|
||||
// ReadStreamTitles reads ICY metadata blocks from r and emits changed StreamTitle values.
|
||||
func ReadStreamTitles(ctx context.Context, r io.Reader, metaInt int, handleTitle func(string)) error {
|
||||
if metaInt <= 0 || metaInt > maxMetaInt {
|
||||
return ErrInvalidMetaInt
|
||||
}
|
||||
if handleTitle == nil {
|
||||
return ErrMissingTitleHandler
|
||||
}
|
||||
|
||||
length := make([]byte, 1)
|
||||
metadataBuf := make([]byte, maxMetadataLen)
|
||||
lastTitle := ""
|
||||
|
||||
for {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if _, err := io.CopyN(io.Discard, r, int64(metaInt)); err != nil {
|
||||
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := readFullOrEnd(r, length); err != nil {
|
||||
if errors.Is(err, errEndOfStream) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
metadataLen := int(length[0]) * 16
|
||||
if metadataLen == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
metadata := metadataBuf[:metadataLen]
|
||||
if err := readFullOrEnd(r, metadata); err != nil {
|
||||
if errors.Is(err, errEndOfStream) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
title := streamTitle(metadata)
|
||||
if title == "" || title == lastTitle {
|
||||
continue
|
||||
}
|
||||
|
||||
lastTitle = title
|
||||
handleTitle(title)
|
||||
}
|
||||
}
|
||||
|
||||
func readFullOrEnd(r io.Reader, buf []byte) error {
|
||||
if _, err := io.ReadFull(r, buf); err != nil {
|
||||
if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
|
||||
return errEndOfStream
|
||||
}
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func streamTitle(metadata []byte) string {
|
||||
text := decodeMetadata(bytes.TrimRight(metadata, "\x00"))
|
||||
const prefix = "StreamTitle='"
|
||||
|
||||
start := strings.Index(text, prefix)
|
||||
if start < 0 {
|
||||
return ""
|
||||
}
|
||||
|
||||
start += len(prefix)
|
||||
end := strings.Index(text[start:], "';")
|
||||
if end < 0 {
|
||||
end = strings.IndexByte(text[start:], '\'')
|
||||
if end < 0 {
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
return strings.TrimSpace(text[start : start+end])
|
||||
}
|
||||
|
||||
func decodeMetadata(metadata []byte) string {
|
||||
if utf8.Valid(metadata) {
|
||||
return string(metadata)
|
||||
}
|
||||
|
||||
text, err := charmap.ISO8859_1.NewDecoder().String(string(metadata))
|
||||
if err != nil {
|
||||
return string(bytes.ToValidUTF8(metadata, nil))
|
||||
}
|
||||
return text
|
||||
}
|
||||
17
core/radio/icy/parser_suite_test.go
Normal file
17
core/radio/icy/parser_suite_test.go
Normal file
@ -0,0 +1,17 @@
|
||||
package icy
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func TestICY(t *testing.T) {
|
||||
tests.Init(t, false)
|
||||
log.SetLevel(log.LevelFatal)
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "ICY Suite")
|
||||
}
|
||||
193
core/radio/icy/parser_test.go
Normal file
193
core/radio/icy/parser_test.go
Normal file
@ -0,0 +1,193 @@
|
||||
package icy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"strings"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("ReadStreamTitles", func() {
|
||||
const metaInt = 5
|
||||
|
||||
readTitles := func(ctx context.Context, stream []byte, interval int) ([]string, error) {
|
||||
var titles []string
|
||||
err := ReadStreamTitles(ctx, bytes.NewReader(stream), interval, func(title string) {
|
||||
titles = append(titles, title)
|
||||
})
|
||||
return titles, err
|
||||
}
|
||||
|
||||
It("emits changed StreamTitle values", func() {
|
||||
stream := icyStream(metaInt,
|
||||
"StreamTitle='Artist One - Track One';",
|
||||
"StreamTitle='Artist One - Track One';",
|
||||
"StreamTitle='Artist Two - Track Two';",
|
||||
)
|
||||
|
||||
titles, err := readTitles(context.Background(), stream, metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(Equal([]string{
|
||||
"Artist One - Track One",
|
||||
"Artist Two - Track Two",
|
||||
}))
|
||||
})
|
||||
|
||||
It("ignores empty metadata blocks", func() {
|
||||
stream := audioInterval(metaInt)
|
||||
stream = append(stream, 0)
|
||||
|
||||
titles, err := readTitles(context.Background(), stream, metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("ignores metadata blocks without StreamTitle", func() {
|
||||
stream := icyStream(metaInt, "StreamUrl='https://example.test';")
|
||||
|
||||
titles, err := readTitles(context.Background(), stream, metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("ignores empty StreamTitle values", func() {
|
||||
stream := icyStream(metaInt, "StreamTitle='';")
|
||||
|
||||
titles, err := readTitles(context.Background(), stream, metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("returns an error for invalid metadata intervals", func() {
|
||||
titles, err := readTitles(context.Background(), nil, 0)
|
||||
|
||||
Expect(err).To(MatchError(ErrInvalidMetaInt))
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("returns an error for huge metadata intervals", func() {
|
||||
titles, err := readTitles(context.Background(), nil, maxMetaInt+1)
|
||||
|
||||
Expect(err).To(MatchError(ErrInvalidMetaInt))
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("keeps single quotes inside StreamTitle values", func() {
|
||||
stream := icyStream(metaInt, "StreamTitle='L'Oiseau - Track';")
|
||||
|
||||
titles, err := readTitles(context.Background(), stream, metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(Equal([]string{"L'Oiseau - Track"}))
|
||||
})
|
||||
|
||||
It("stops when context is cancelled", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
titles, err := readTitles(ctx, audioInterval(metaInt), metaInt)
|
||||
|
||||
Expect(err).To(MatchError(context.Canceled))
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("returns cleanly when input ends mid-audio interval", func() {
|
||||
titles, err := readTitles(context.Background(), []byte("abc"), metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("returns cleanly when input ends mid-metadata block", func() {
|
||||
stream := append(audioInterval(metaInt), 2, 'S', 't', 'r')
|
||||
|
||||
titles, err := readTitles(context.Background(), stream, metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("decodes Latin-1 metadata", func() {
|
||||
metadata := []byte("StreamTitle='Beyonc")
|
||||
metadata = append(metadata, 0xe9)
|
||||
metadata = append(metadata, " - D\xe9j\xe0 Vu';"...)
|
||||
stream := icyStreamBytes(metaInt, metadata)
|
||||
|
||||
titles, err := readTitles(context.Background(), stream, metaInt)
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(Equal([]string{"Beyoncé - Déjà Vu"}))
|
||||
})
|
||||
|
||||
It("returns reader errors other than EOF", func() {
|
||||
errReader := errAfterReader{
|
||||
data: []byte("abcde"),
|
||||
err: errors.New("read failed"),
|
||||
}
|
||||
var titles []string
|
||||
|
||||
err := ReadStreamTitles(context.Background(), &errReader, metaInt, func(title string) {
|
||||
titles = append(titles, title)
|
||||
})
|
||||
|
||||
Expect(err).To(MatchError("read failed"))
|
||||
Expect(titles).To(BeEmpty())
|
||||
})
|
||||
})
|
||||
|
||||
func icyStream(metaInt int, metadataBlocks ...string) []byte {
|
||||
blocks := make([][]byte, 0, len(metadataBlocks))
|
||||
for _, block := range metadataBlocks {
|
||||
blocks = append(blocks, []byte(block))
|
||||
}
|
||||
return icyStreamBytes(metaInt, blocks...)
|
||||
}
|
||||
|
||||
func icyStreamBytes(metaInt int, metadataBlocks ...[]byte) []byte {
|
||||
var stream []byte
|
||||
for _, metadata := range metadataBlocks {
|
||||
stream = append(stream, audioInterval(metaInt)...)
|
||||
stream = append(stream, metadataBlock(metadata)...)
|
||||
}
|
||||
return stream
|
||||
}
|
||||
|
||||
func audioInterval(metaInt int) []byte {
|
||||
return []byte(strings.Repeat("a", metaInt))
|
||||
}
|
||||
|
||||
func metadataBlock(metadata []byte) []byte {
|
||||
if len(metadata) == 0 {
|
||||
return []byte{0}
|
||||
}
|
||||
blockCount := (len(metadata) + 15) / 16
|
||||
block := []byte{byte(blockCount)}
|
||||
block = append(block, metadata...)
|
||||
padding := blockCount*16 - len(metadata)
|
||||
block = append(block, bytes.Repeat([]byte{0}, padding)...)
|
||||
return block
|
||||
}
|
||||
|
||||
type errAfterReader struct {
|
||||
data []byte
|
||||
err error
|
||||
}
|
||||
|
||||
func (r *errAfterReader) Read(p []byte) (int, error) {
|
||||
if len(r.data) > 0 {
|
||||
n := copy(p, r.data)
|
||||
r.data = r.data[n:]
|
||||
return n, nil
|
||||
}
|
||||
return 0, r.err
|
||||
}
|
||||
|
||||
var _ io.Reader = (*errAfterReader)(nil)
|
||||
58
core/radio/icy/reader.go
Normal file
58
core/radio/icy/reader.go
Normal file
@ -0,0 +1,58 @@
|
||||
package icy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/navidrome/navidrome/consts"
|
||||
)
|
||||
|
||||
var ErrMissingMetaInt = errors.New("missing icy metadata interval")
|
||||
|
||||
type ResponseStatusError struct {
|
||||
StatusCode int
|
||||
Status string
|
||||
}
|
||||
|
||||
func (e *ResponseStatusError) Error() string {
|
||||
return fmt.Sprintf("icy stream returned %s", e.Status)
|
||||
}
|
||||
|
||||
// ReadHTTPStreamTitles requests ICY metadata from streamURL and emits changed StreamTitle values.
|
||||
func ReadHTTPStreamTitles(ctx context.Context, client *http.Client, streamURL string, handleTitle func(string)) error {
|
||||
if client == nil {
|
||||
client = http.DefaultClient
|
||||
}
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, streamURL, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Icy-MetaData", "1")
|
||||
req.Header.Set("User-Agent", consts.HTTPUserAgent)
|
||||
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
|
||||
return &ResponseStatusError{StatusCode: resp.StatusCode, Status: resp.Status}
|
||||
}
|
||||
|
||||
metaIntHeader := resp.Header.Get("icy-metaint")
|
||||
if metaIntHeader == "" {
|
||||
return ErrMissingMetaInt
|
||||
}
|
||||
|
||||
metaInt, err := strconv.Atoi(metaIntHeader)
|
||||
if err != nil || metaInt <= 0 {
|
||||
return fmt.Errorf("%w: %q", ErrInvalidMetaInt, metaIntHeader)
|
||||
}
|
||||
|
||||
return ReadStreamTitles(ctx, resp.Body, metaInt, handleTitle)
|
||||
}
|
||||
124
core/radio/icy/reader_test.go
Normal file
124
core/radio/icy/reader_test.go
Normal file
@ -0,0 +1,124 @@
|
||||
package icy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("ReadHTTPStreamTitles", func() {
|
||||
const metaInt = 5
|
||||
|
||||
It("requests ICY metadata and parses stream titles", func() {
|
||||
var icyHeader string
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
icyHeader = r.Header.Get("Icy-MetaData")
|
||||
w.Header().Set("icy-metaint", "5")
|
||||
_, _ = w.Write(icyStream(metaInt, "StreamTitle='Artist - Title';"))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
var titles []string
|
||||
err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(title string) {
|
||||
titles = append(titles, title)
|
||||
})
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(icyHeader).To(Equal("1"))
|
||||
Expect(titles).To(Equal([]string{"Artist - Title"}))
|
||||
})
|
||||
|
||||
It("follows redirects before reading stream metadata", func() {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if r.URL.Path == "/redirect" {
|
||||
http.Redirect(w, r, "/stream", http.StatusFound)
|
||||
return
|
||||
}
|
||||
|
||||
Expect(r.URL.Path).To(Equal("/stream"))
|
||||
Expect(r.Header.Get("Icy-MetaData")).To(Equal("1"))
|
||||
w.Header().Set("Icy-MetaInt", "5")
|
||||
_, _ = w.Write(icyStream(metaInt, "StreamTitle='Redirected - Title';"))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
var titles []string
|
||||
err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL+"/redirect", func(title string) {
|
||||
titles = append(titles, title)
|
||||
})
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(titles).To(Equal([]string{"Redirected - Title"}))
|
||||
})
|
||||
|
||||
It("returns an error when icy-metaint is missing", func() {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
_, _ = w.Write([]byte("not an icy stream"))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(string) {})
|
||||
|
||||
Expect(errors.Is(err, ErrMissingMetaInt)).To(BeTrue())
|
||||
})
|
||||
|
||||
It("returns an error when icy-metaint is invalid", func() {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("icy-metaint", "zero")
|
||||
_, _ = w.Write([]byte("not an icy stream"))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(string) {})
|
||||
|
||||
Expect(errors.Is(err, ErrInvalidMetaInt)).To(BeTrue())
|
||||
})
|
||||
|
||||
It("returns response status errors for non-2xx responses", func() {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
http.Error(w, "offline", http.StatusServiceUnavailable)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
err := ReadHTTPStreamTitles(context.Background(), server.Client(), server.URL, func(string) {})
|
||||
|
||||
var statusErr *ResponseStatusError
|
||||
Expect(errors.As(err, &statusErr)).To(BeTrue())
|
||||
Expect(statusErr.StatusCode).To(Equal(http.StatusServiceUnavailable))
|
||||
})
|
||||
|
||||
It("stops when context is cancelled", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
err := ReadHTTPStreamTitles(ctx, http.DefaultClient, "https://example.test/stream", func(string) {})
|
||||
|
||||
Expect(errors.Is(err, context.Canceled)).To(BeTrue())
|
||||
})
|
||||
|
||||
It("stops a body read when context is cancelled", func() {
|
||||
started := make(chan struct{})
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("icy-metaint", "5")
|
||||
w.(http.Flusher).Flush()
|
||||
close(started)
|
||||
<-r.Context().Done()
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
errC := make(chan error, 1)
|
||||
go func() {
|
||||
errC <- ReadHTTPStreamTitles(ctx, server.Client(), server.URL, func(string) {})
|
||||
}()
|
||||
|
||||
Eventually(started).Should(BeClosed())
|
||||
cancel()
|
||||
|
||||
Eventually(errC).Should(Receive(MatchError(ContainSubstring("context canceled"))))
|
||||
})
|
||||
})
|
||||
326
core/radio/metadata_manager.go
Normal file
326
core/radio/metadata_manager.go
Normal file
@ -0,0 +1,326 @@
|
||||
package radio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/log"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrInvalidSession = errors.New("invalid radio metadata session")
|
||||
ErrInvalidStation = errors.New("invalid radio metadata station")
|
||||
)
|
||||
|
||||
// defaultSessionTTL bounds how long a session survives without a refresh, so
|
||||
// clients that vanish without calling Stop (e.g. a closed browser tab) do not
|
||||
// keep an ICY reader connected to the station forever.
|
||||
const defaultSessionTTL = 10 * time.Minute
|
||||
|
||||
// permanentRetryDelay is used instead of the regular backoff when the reader
|
||||
// reports a permanent error (dead URL, station without ICY metadata), so we
|
||||
// do not hammer a stream that will never yield titles.
|
||||
const permanentRetryDelay = 10 * time.Minute
|
||||
|
||||
type permanentError struct{ err error }
|
||||
|
||||
func (e *permanentError) Error() string { return e.err.Error() }
|
||||
func (e *permanentError) Unwrap() error { return e.err }
|
||||
|
||||
// MarkPermanent wraps err so the retry loop switches to permanentRetryDelay.
|
||||
func MarkPermanent(err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
return &permanentError{err: err}
|
||||
}
|
||||
|
||||
func isPermanent(err error) bool {
|
||||
var p *permanentError
|
||||
return errors.As(err, &p)
|
||||
}
|
||||
|
||||
type Station struct {
|
||||
ID string
|
||||
StreamURL string
|
||||
}
|
||||
|
||||
type TitleUpdate struct {
|
||||
RadioID string `json:"radioId"`
|
||||
Title string `json:"title"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
}
|
||||
|
||||
type StreamReader func(ctx context.Context, streamURL string, handleTitle func(string)) error
|
||||
|
||||
type TitlePublisher func(ctx context.Context, update TitleUpdate)
|
||||
|
||||
type MetadataManager struct {
|
||||
mu sync.Mutex
|
||||
|
||||
reader StreamReader
|
||||
publish TitlePublisher
|
||||
now func() time.Time
|
||||
backoff func(attempt int) time.Duration
|
||||
sessionTTL time.Duration
|
||||
|
||||
sessions map[string]activeSession
|
||||
readers map[string]*activeReader
|
||||
}
|
||||
|
||||
type ManagerOption func(*MetadataManager)
|
||||
|
||||
type activeSession struct {
|
||||
radioID string
|
||||
streamURL string
|
||||
notifyCtx context.Context
|
||||
expiresAt time.Time
|
||||
expireTimer *time.Timer
|
||||
}
|
||||
|
||||
type activeReader struct {
|
||||
streamURL string
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
radioRefs map[string]int
|
||||
lastTitle string
|
||||
}
|
||||
|
||||
func NewMetadataManager(reader StreamReader, publisher TitlePublisher, opts ...ManagerOption) *MetadataManager {
|
||||
m := &MetadataManager{
|
||||
reader: reader,
|
||||
publish: publisher,
|
||||
now: time.Now,
|
||||
backoff: defaultBackoff,
|
||||
sessionTTL: defaultSessionTTL,
|
||||
sessions: map[string]activeSession{},
|
||||
readers: map[string]*activeReader{},
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(m)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func WithNow(now func() time.Time) ManagerOption {
|
||||
return func(m *MetadataManager) {
|
||||
if now != nil {
|
||||
m.now = now
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func WithRetryBackoff(backoff func(attempt int) time.Duration) ManagerOption {
|
||||
return func(m *MetadataManager) {
|
||||
if backoff != nil {
|
||||
m.backoff = backoff
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithSessionTTL sets how long a session lives without a refreshing Start
|
||||
// call. A zero or negative TTL disables expiry.
|
||||
func WithSessionTTL(ttl time.Duration) ManagerOption {
|
||||
return func(m *MetadataManager) {
|
||||
m.sessionTTL = ttl
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MetadataManager) Start(ctx context.Context, sessionID string, station Station) error {
|
||||
if sessionID == "" {
|
||||
return ErrInvalidSession
|
||||
}
|
||||
if station.ID == "" || station.StreamURL == "" {
|
||||
return ErrInvalidStation
|
||||
}
|
||||
if m.reader == nil {
|
||||
return errors.New("missing radio metadata stream reader")
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if current, ok := m.sessions[sessionID]; ok {
|
||||
if current.radioID == station.ID && current.streamURL == station.StreamURL {
|
||||
m.touchSessionLocked(sessionID, current)
|
||||
return nil
|
||||
}
|
||||
m.removeSessionLocked(sessionID)
|
||||
}
|
||||
|
||||
reader := m.readers[station.StreamURL]
|
||||
if reader == nil {
|
||||
readerCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
|
||||
reader = &activeReader{
|
||||
streamURL: station.StreamURL,
|
||||
ctx: readerCtx,
|
||||
cancel: cancel,
|
||||
radioRefs: map[string]int{},
|
||||
}
|
||||
m.readers[station.StreamURL] = reader
|
||||
go m.runReader(reader)
|
||||
}
|
||||
|
||||
reader.radioRefs[station.ID]++
|
||||
session := activeSession{
|
||||
radioID: station.ID,
|
||||
streamURL: station.StreamURL,
|
||||
notifyCtx: context.WithoutCancel(ctx),
|
||||
}
|
||||
if m.sessionTTL > 0 {
|
||||
session.expiresAt = m.now().Add(m.sessionTTL)
|
||||
session.expireTimer = time.AfterFunc(m.sessionTTL, func() { m.expireSession(sessionID) })
|
||||
}
|
||||
m.sessions[sessionID] = session
|
||||
return nil
|
||||
}
|
||||
|
||||
// touchSessionLocked extends the session deadline. The pending expiry timer
|
||||
// reschedules itself when it fires before the new deadline.
|
||||
func (m *MetadataManager) touchSessionLocked(sessionID string, session activeSession) {
|
||||
if m.sessionTTL <= 0 || session.expireTimer == nil {
|
||||
return
|
||||
}
|
||||
session.expiresAt = m.now().Add(m.sessionTTL)
|
||||
m.sessions[sessionID] = session
|
||||
}
|
||||
|
||||
func (m *MetadataManager) expireSession(sessionID string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
session, ok := m.sessions[sessionID]
|
||||
if !ok || session.expireTimer == nil {
|
||||
return
|
||||
}
|
||||
if remaining := session.expiresAt.Sub(m.now()); remaining > 0 {
|
||||
session.expireTimer.Reset(remaining)
|
||||
return
|
||||
}
|
||||
log.Debug("Radio metadata session expired without refresh", "sessionID", sessionID, "radioID", session.radioID)
|
||||
m.removeSessionLocked(sessionID)
|
||||
}
|
||||
|
||||
func (m *MetadataManager) Stop(sessionID string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.removeSessionLocked(sessionID)
|
||||
}
|
||||
|
||||
func (m *MetadataManager) removeSessionLocked(sessionID string) {
|
||||
session, ok := m.sessions[sessionID]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
delete(m.sessions, sessionID)
|
||||
if session.expireTimer != nil {
|
||||
session.expireTimer.Stop()
|
||||
}
|
||||
|
||||
reader := m.readers[session.streamURL]
|
||||
if reader == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if count := reader.radioRefs[session.radioID]; count <= 1 {
|
||||
delete(reader.radioRefs, session.radioID)
|
||||
} else {
|
||||
reader.radioRefs[session.radioID] = count - 1
|
||||
}
|
||||
|
||||
if len(reader.radioRefs) == 0 {
|
||||
delete(m.readers, session.streamURL)
|
||||
reader.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MetadataManager) runReader(reader *activeReader) {
|
||||
attempt := 0
|
||||
for {
|
||||
err := m.reader(reader.ctx, reader.streamURL, func(title string) {
|
||||
m.publishTitle(reader, title)
|
||||
})
|
||||
if reader.ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if err == nil {
|
||||
attempt = 0
|
||||
} else {
|
||||
attempt++
|
||||
log.Trace("Radio metadata reader retrying", "streamURL", reader.streamURL, "attempt", attempt, err)
|
||||
}
|
||||
|
||||
delay := m.backoff(attempt)
|
||||
if isPermanent(err) && delay < permanentRetryDelay {
|
||||
delay = permanentRetryDelay
|
||||
}
|
||||
if delay <= 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
timer := time.NewTimer(delay)
|
||||
select {
|
||||
case <-reader.ctx.Done():
|
||||
timer.Stop()
|
||||
return
|
||||
case <-timer.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *MetadataManager) publishTitle(reader *activeReader, title string) {
|
||||
m.mu.Lock()
|
||||
if reader.ctx.Err() != nil || title == "" || title == reader.lastTitle {
|
||||
m.mu.Unlock()
|
||||
return
|
||||
}
|
||||
|
||||
reader.lastTitle = title
|
||||
radioIDs := make([]string, 0, len(reader.radioRefs))
|
||||
for radioID := range reader.radioRefs {
|
||||
radioIDs = append(radioIDs, radioID)
|
||||
}
|
||||
updatedAt := m.now()
|
||||
publish := m.publish
|
||||
// Copy matching session contexts while the lock is held.
|
||||
type sessionNotify struct {
|
||||
radioID string
|
||||
notifyCtx context.Context
|
||||
}
|
||||
notifies := make([]sessionNotify, 0, len(m.sessions))
|
||||
for _, session := range m.sessions {
|
||||
if session.streamURL != reader.streamURL {
|
||||
continue
|
||||
}
|
||||
for _, radioID := range radioIDs {
|
||||
if session.radioID == radioID {
|
||||
notifies = append(notifies, sessionNotify{radioID: radioID, notifyCtx: session.notifyCtx})
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
m.mu.Unlock()
|
||||
|
||||
if publish == nil {
|
||||
return
|
||||
}
|
||||
for _, target := range notifies {
|
||||
publish(target.notifyCtx, TitleUpdate{
|
||||
RadioID: target.radioID,
|
||||
Title: title,
|
||||
UpdatedAt: updatedAt,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func defaultBackoff(attempt int) time.Duration {
|
||||
if attempt < 1 {
|
||||
attempt = 1
|
||||
}
|
||||
delay := time.Duration(attempt) * time.Second
|
||||
if delay > 30*time.Second {
|
||||
return 30 * time.Second
|
||||
}
|
||||
return delay
|
||||
}
|
||||
317
core/radio/metadata_manager_test.go
Normal file
317
core/radio/metadata_manager_test.go
Normal file
@ -0,0 +1,317 @@
|
||||
package radio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("MetadataManager", func() {
|
||||
type contextKey string
|
||||
|
||||
var (
|
||||
reader *fakeStreamReader
|
||||
publisher *fakePublisher
|
||||
manager *MetadataManager
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
reader = newFakeStreamReader()
|
||||
publisher = newFakePublisher()
|
||||
manager = NewMetadataManager(
|
||||
reader.Read,
|
||||
publisher.Publish,
|
||||
WithRetryBackoff(func(int) time.Duration { return 0 }),
|
||||
WithNow(func() time.Time { return time.Unix(123, 0).UTC() }),
|
||||
)
|
||||
})
|
||||
|
||||
It("starts a reader for an active radio session", func() {
|
||||
err := manager.Start(context.Background(), "session-1", Station{
|
||||
ID: "rd-1",
|
||||
StreamURL: "https://stream.example.test/radio",
|
||||
})
|
||||
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
Expect(reader.URLs()).To(Equal([]string{"https://stream.example.test/radio"}))
|
||||
})
|
||||
|
||||
It("keeps a reader alive after the start request context is cancelled", func() {
|
||||
ctx, cancel := context.WithCancel(context.WithValue(context.Background(), contextKey("user"), "janedoe"))
|
||||
Expect(manager.Start(ctx, "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
|
||||
cancel()
|
||||
|
||||
Consistently(reader.CancelCount).Should(Equal(0))
|
||||
reader.Emit("Live Artist - Live Track")
|
||||
Eventually(func() any {
|
||||
contexts := publisher.Contexts()
|
||||
if len(contexts) == 0 {
|
||||
return nil
|
||||
}
|
||||
return contexts[0].Value(contextKey("user"))
|
||||
}).Should(Equal("janedoe"))
|
||||
|
||||
manager.Stop("session-1")
|
||||
Eventually(reader.CancelCount).Should(Equal(1))
|
||||
})
|
||||
|
||||
It("shares one reader for sessions using the same stream URL", func() {
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
Expect(manager.Start(context.Background(), "session-2", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
Consistently(reader.StartCount).Should(Equal(1))
|
||||
})
|
||||
|
||||
It("cancels the reader when the last session stops", func() {
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
|
||||
manager.Stop("session-1")
|
||||
|
||||
Eventually(reader.CancelCount).Should(Equal(1))
|
||||
})
|
||||
|
||||
It("keeps the reader alive while another session still references the stream URL", func() {
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
Expect(manager.Start(context.Background(), "session-2", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
|
||||
manager.Stop("session-1")
|
||||
|
||||
Consistently(reader.CancelCount).Should(Equal(0))
|
||||
manager.Stop("session-2")
|
||||
Eventually(reader.CancelCount).Should(Equal(1))
|
||||
})
|
||||
|
||||
It("stops an existing session before starting that session on another station", func() {
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/one"})).To(Succeed())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-2", StreamURL: "https://stream.example.test/two"})).To(Succeed())
|
||||
|
||||
Eventually(reader.CancelCount).Should(Equal(1))
|
||||
Eventually(reader.StartCount).Should(Equal(2))
|
||||
Expect(reader.URLs()).To(Equal([]string{
|
||||
"https://stream.example.test/one",
|
||||
"https://stream.example.test/two",
|
||||
}))
|
||||
})
|
||||
|
||||
It("publishes title updates for active radio IDs", func() {
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
|
||||
reader.Emit("Live Artist - Live Track")
|
||||
|
||||
Eventually(publisher.Updates).Should(Equal([]TitleUpdate{{
|
||||
RadioID: "rd-1",
|
||||
Title: "Live Artist - Live Track",
|
||||
UpdatedAt: time.Unix(123, 0).UTC(),
|
||||
}}))
|
||||
})
|
||||
|
||||
It("does not publish duplicate consecutive titles from the same reader", func() {
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
|
||||
reader.Emit("Same Title")
|
||||
reader.Emit("Same Title")
|
||||
|
||||
Consistently(publisher.Updates).Should(Equal([]TitleUpdate{{
|
||||
RadioID: "rd-1",
|
||||
Title: "Same Title",
|
||||
UpdatedAt: time.Unix(123, 0).UTC(),
|
||||
}}))
|
||||
})
|
||||
|
||||
It("retries reader failures until stopped", func() {
|
||||
reader.err = errors.New("stream failed")
|
||||
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(Succeed())
|
||||
|
||||
Eventually(reader.StartCount).Should(BeNumerically(">", 1))
|
||||
manager.Stop("session-1")
|
||||
cancelCount := reader.CancelCount()
|
||||
Consistently(reader.StartCount).Should(BeNumerically(">=", 2))
|
||||
Expect(reader.CancelCount()).To(BeNumerically(">=", cancelCount))
|
||||
})
|
||||
|
||||
It("ignores unknown stops", func() {
|
||||
manager.Stop("missing-session")
|
||||
Expect(reader.StartCount()).To(Equal(0))
|
||||
})
|
||||
|
||||
It("rejects invalid start requests", func() {
|
||||
Expect(manager.Start(context.Background(), "", Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"})).To(MatchError(ErrInvalidSession))
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "", StreamURL: "https://stream.example.test/radio"})).To(MatchError(ErrInvalidStation))
|
||||
Expect(manager.Start(context.Background(), "session-1", Station{ID: "rd-1", StreamURL: ""})).To(MatchError(ErrInvalidStation))
|
||||
})
|
||||
|
||||
It("backs off for a long time on permanent reader errors", func() {
|
||||
reader.err = MarkPermanent(errors.New("missing icy metadata interval"))
|
||||
slow := NewMetadataManager(
|
||||
reader.Read,
|
||||
publisher.Publish,
|
||||
WithRetryBackoff(func(int) time.Duration { return time.Millisecond }),
|
||||
)
|
||||
|
||||
Expect(slow.Start(context.Background(), "session-1", Station{
|
||||
ID: "rd-1",
|
||||
StreamURL: "https://stream.example.test/radio",
|
||||
})).To(Succeed())
|
||||
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
// Regular backoff is 1ms; the permanent error must stretch it, so no
|
||||
// second attempt happens within the observation window.
|
||||
Consistently(reader.StartCount, 300*time.Millisecond, 20*time.Millisecond).Should(Equal(1))
|
||||
})
|
||||
|
||||
It("expires sessions that are never refreshed", func() {
|
||||
expiring := NewMetadataManager(
|
||||
reader.Read,
|
||||
publisher.Publish,
|
||||
WithRetryBackoff(func(int) time.Duration { return 0 }),
|
||||
WithSessionTTL(20*time.Millisecond),
|
||||
)
|
||||
|
||||
Expect(expiring.Start(context.Background(), "session-1", Station{
|
||||
ID: "rd-1",
|
||||
StreamURL: "https://stream.example.test/radio",
|
||||
})).To(Succeed())
|
||||
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
Eventually(reader.CancelCount).Should(Equal(1))
|
||||
})
|
||||
|
||||
It("keeps refreshed sessions alive past the TTL", func() {
|
||||
expiring := NewMetadataManager(
|
||||
reader.Read,
|
||||
publisher.Publish,
|
||||
WithRetryBackoff(func(int) time.Duration { return 0 }),
|
||||
WithSessionTTL(50*time.Millisecond),
|
||||
)
|
||||
station := Station{ID: "rd-1", StreamURL: "https://stream.example.test/radio"}
|
||||
|
||||
Expect(expiring.Start(context.Background(), "session-1", station)).To(Succeed())
|
||||
Eventually(reader.StartCount).Should(Equal(1))
|
||||
|
||||
// Refresh the session on every poll, well inside the TTL window.
|
||||
Consistently(func() int {
|
||||
Expect(expiring.Start(context.Background(), "session-1", station)).To(Succeed())
|
||||
return reader.CancelCount()
|
||||
}, 200*time.Millisecond, 10*time.Millisecond).Should(Equal(0))
|
||||
})
|
||||
})
|
||||
|
||||
type fakeStreamReader struct {
|
||||
mu sync.Mutex
|
||||
started chan struct{}
|
||||
cancelled chan struct{}
|
||||
handler func(string)
|
||||
urls []string
|
||||
starts int
|
||||
cancels int
|
||||
err error
|
||||
}
|
||||
|
||||
func newFakeStreamReader() *fakeStreamReader {
|
||||
return &fakeStreamReader{
|
||||
started: make(chan struct{}),
|
||||
cancelled: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *fakeStreamReader) Read(ctx context.Context, streamURL string, handleTitle func(string)) error {
|
||||
r.mu.Lock()
|
||||
r.starts++
|
||||
r.urls = append(r.urls, streamURL)
|
||||
r.handler = handleTitle
|
||||
if r.starts == 1 {
|
||||
close(r.started)
|
||||
}
|
||||
err := r.err
|
||||
r.mu.Unlock()
|
||||
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
<-ctx.Done()
|
||||
|
||||
r.mu.Lock()
|
||||
r.cancels++
|
||||
if r.cancels == 1 {
|
||||
close(r.cancelled)
|
||||
}
|
||||
r.mu.Unlock()
|
||||
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
func (r *fakeStreamReader) Emit(title string) {
|
||||
Eventually(func() func(string) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.handler
|
||||
}).ShouldNot(BeNil())
|
||||
|
||||
r.mu.Lock()
|
||||
handler := r.handler
|
||||
r.mu.Unlock()
|
||||
handler(title)
|
||||
}
|
||||
|
||||
func (r *fakeStreamReader) StartCount() int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.starts
|
||||
}
|
||||
|
||||
func (r *fakeStreamReader) CancelCount() int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.cancels
|
||||
}
|
||||
|
||||
func (r *fakeStreamReader) URLs() []string {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return append([]string(nil), r.urls...)
|
||||
}
|
||||
|
||||
type fakePublisher struct {
|
||||
mu sync.Mutex
|
||||
updates []TitleUpdate
|
||||
ctxs []context.Context
|
||||
}
|
||||
|
||||
func newFakePublisher() *fakePublisher {
|
||||
return &fakePublisher{}
|
||||
}
|
||||
|
||||
func (p *fakePublisher) Publish(ctx context.Context, update TitleUpdate) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.updates = append(p.updates, update)
|
||||
p.ctxs = append(p.ctxs, ctx)
|
||||
}
|
||||
|
||||
func (p *fakePublisher) Updates() []TitleUpdate {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return append([]TitleUpdate(nil), p.updates...)
|
||||
}
|
||||
|
||||
func (p *fakePublisher) Contexts() []context.Context {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return append([]context.Context(nil), p.ctxs...)
|
||||
}
|
||||
48
core/radio/provider.go
Normal file
48
core/radio/provider.go
Normal file
@ -0,0 +1,48 @@
|
||||
package radio
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/core/radio/icy"
|
||||
)
|
||||
|
||||
// icyHTTPClient has no overall timeout (ICY streams are long-lived), but bounds
|
||||
// connection setup and header wait so a stalled station cannot hang the reader.
|
||||
var icyHTTPClient = &http.Client{
|
||||
Transport: &http.Transport{
|
||||
DialContext: (&net.Dialer{
|
||||
Timeout: 10 * time.Second,
|
||||
KeepAlive: 30 * time.Second,
|
||||
}).DialContext,
|
||||
TLSHandshakeTimeout: 10 * time.Second,
|
||||
ResponseHeaderTimeout: 30 * time.Second,
|
||||
IdleConnTimeout: 60 * time.Second,
|
||||
MaxIdleConnsPerHost: 1,
|
||||
},
|
||||
}
|
||||
|
||||
func NewMetadataManagerService(publisher TitlePublisher) *MetadataManager {
|
||||
return NewMetadataManager(func(ctx context.Context, streamURL string, handleTitle func(string)) error {
|
||||
return classifyICYError(icy.ReadHTTPStreamTitles(ctx, icyHTTPClient, streamURL, handleTitle))
|
||||
}, publisher)
|
||||
}
|
||||
|
||||
// classifyICYError marks errors that a retry will not fix — a stream without
|
||||
// ICY metadata or a client-error HTTP status — so the reader backs off for
|
||||
// much longer instead of hammering the station.
|
||||
func classifyICYError(err error) error {
|
||||
if errors.Is(err, icy.ErrMissingMetaInt) || errors.Is(err, icy.ErrInvalidMetaInt) {
|
||||
return MarkPermanent(err)
|
||||
}
|
||||
var statusErr *icy.ResponseStatusError
|
||||
if errors.As(err, &statusErr) &&
|
||||
statusErr.StatusCode >= 400 && statusErr.StatusCode < 500 &&
|
||||
statusErr.StatusCode != http.StatusRequestTimeout && statusErr.StatusCode != http.StatusTooManyRequests {
|
||||
return MarkPermanent(err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
20
core/radio/provider_test.go
Normal file
20
core/radio/provider_test.go
Normal file
@ -0,0 +1,20 @@
|
||||
package radio
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/navidrome/navidrome/core/radio/icy"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("Metadata provider", func() {
|
||||
Describe("classifyICYError", func() {
|
||||
It("marks invalid metadata intervals as permanent", func() {
|
||||
err := classifyICYError(fmt.Errorf("%w: %q", icy.ErrInvalidMetaInt, "999999999"))
|
||||
|
||||
Expect(isPermanent(err)).To(BeTrue())
|
||||
Expect(err).To(MatchError(ContainSubstring(icy.ErrInvalidMetaInt.Error())))
|
||||
})
|
||||
})
|
||||
})
|
||||
17
core/radio/radio_suite_test.go
Normal file
17
core/radio/radio_suite_test.go
Normal file
@ -0,0 +1,17 @@
|
||||
package radio
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func TestRadio(t *testing.T) {
|
||||
tests.Init(t, false)
|
||||
log.SetLevel(log.LevelFatal)
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "Radio Suite")
|
||||
}
|
||||
@ -10,6 +10,7 @@ import (
|
||||
"github.com/navidrome/navidrome/core/metrics"
|
||||
"github.com/navidrome/navidrome/core/playback"
|
||||
"github.com/navidrome/navidrome/core/playlists"
|
||||
"github.com/navidrome/navidrome/core/radio"
|
||||
"github.com/navidrome/navidrome/core/scrobbler"
|
||||
"github.com/navidrome/navidrome/core/stream"
|
||||
)
|
||||
@ -36,4 +37,5 @@ var Set = wire.NewSet(
|
||||
playback.GetInstance,
|
||||
metrics.GetInstance,
|
||||
lyrics.NewLyrics,
|
||||
radio.NewMetadataManagerService,
|
||||
)
|
||||
|
||||
@ -7,17 +7,24 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
"unicode"
|
||||
|
||||
coreradio "github.com/navidrome/navidrome/core/radio"
|
||||
)
|
||||
|
||||
type eventCtxKey string
|
||||
|
||||
const broadcastToAllKey eventCtxKey = "broadcastToAll"
|
||||
const sendToSameUserKey eventCtxKey = "sendToSameUser"
|
||||
|
||||
// broadcastToAll is a context key that can be used to broadcast an event to all clients
|
||||
func broadcastToAll(ctx context.Context) context.Context {
|
||||
return context.WithValue(ctx, broadcastToAllKey, true)
|
||||
}
|
||||
|
||||
func sendToSameUser(ctx context.Context) context.Context {
|
||||
return context.WithValue(ctx, sendToSameUserKey, true)
|
||||
}
|
||||
|
||||
type Event interface {
|
||||
Name(Event) string
|
||||
Data(Event) string
|
||||
@ -68,6 +75,26 @@ type NowPlayingCount struct {
|
||||
Count int `json:"count"`
|
||||
}
|
||||
|
||||
type RadioNowPlaying struct {
|
||||
baseEvent
|
||||
RadioID string `json:"radioId"`
|
||||
Title string `json:"title"`
|
||||
UpdatedAt time.Time `json:"updatedAt"`
|
||||
}
|
||||
|
||||
func NewRadioMetadataPublisher(broker Broker) coreradio.TitlePublisher {
|
||||
return func(ctx context.Context, update coreradio.TitleUpdate) {
|
||||
if broker == nil {
|
||||
return
|
||||
}
|
||||
broker.SendMessage(sendToSameUser(ctx), &RadioNowPlaying{
|
||||
RadioID: update.RadioID,
|
||||
Title: update.Title,
|
||||
UpdatedAt: update.UpdatedAt,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (rr *RefreshResource) With(resource string, ids ...string) *RefreshResource {
|
||||
if rr.resources == nil {
|
||||
rr.resources = make(map[string][]string)
|
||||
|
||||
@ -1,6 +1,12 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
coreradio "github.com/navidrome/navidrome/core/radio"
|
||||
"github.com/navidrome/navidrome/model/request"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
@ -21,6 +27,49 @@ var _ = Describe("Events", func() {
|
||||
})
|
||||
})
|
||||
|
||||
Describe("RadioNowPlaying", func() {
|
||||
It("serialises with a stable event name and payload", func() {
|
||||
updatedAt := time.Date(2026, time.June, 30, 12, 34, 56, 0, time.UTC)
|
||||
event := &RadioNowPlaying{
|
||||
RadioID: "rd-1",
|
||||
Title: "Artist - Title",
|
||||
UpdatedAt: updatedAt,
|
||||
}
|
||||
|
||||
Expect(event.Name(event)).To(Equal("radioNowPlaying"))
|
||||
Expect(event.Data(event)).To(Equal(`{"radioId":"rd-1","title":"Artist - Title","updatedAt":"2026-06-30T12:34:56Z"}`))
|
||||
})
|
||||
|
||||
It("publishes metadata updates through the broker", func() {
|
||||
ctx := request.WithClientUniqueId(request.WithUsername(context.Background(), "janedoe"), "browser-1")
|
||||
updatedAt := time.Date(2026, time.June, 30, 12, 34, 56, 0, time.UTC)
|
||||
capture := &capturingBroker{}
|
||||
|
||||
publish := NewRadioMetadataPublisher(capture)
|
||||
publish(ctx, coreradio.TitleUpdate{
|
||||
RadioID: "rd-1",
|
||||
Title: "Artist - Title",
|
||||
UpdatedAt: updatedAt,
|
||||
})
|
||||
|
||||
username, hasUsername := request.UsernameFrom(capture.ctx)
|
||||
clientID, hasClientID := request.ClientUniqueIdFrom(capture.ctx)
|
||||
Expect(hasUsername).To(BeTrue())
|
||||
Expect(username).To(Equal("janedoe"))
|
||||
Expect(hasClientID).To(BeTrue())
|
||||
Expect(clientID).To(Equal("browser-1"))
|
||||
Expect(capture.broadcast).To(BeNil())
|
||||
Expect(capture.event).To(Equal(&RadioNowPlaying{
|
||||
RadioID: "rd-1",
|
||||
Title: "Artist - Title",
|
||||
UpdatedAt: updatedAt,
|
||||
}))
|
||||
routingBroker := broker{}
|
||||
Expect(routingBroker.shouldSend(message{senderCtx: capture.ctx}, client{username: "janedoe", clientUniqueId: "browser-1"})).To(BeTrue())
|
||||
Expect(routingBroker.shouldSend(message{senderCtx: capture.ctx}, client{username: "johndoe", clientUniqueId: "browser-2"})).To(BeFalse())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("RefreshResource", func() {
|
||||
var rr *RefreshResource
|
||||
BeforeEach(func() {
|
||||
@ -44,3 +93,20 @@ var _ = Describe("Events", func() {
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
type capturingBroker struct {
|
||||
ctx context.Context
|
||||
event Event
|
||||
broadcast Event
|
||||
}
|
||||
|
||||
func (b *capturingBroker) ServeHTTP(http.ResponseWriter, *http.Request) {}
|
||||
|
||||
func (b *capturingBroker) SendMessage(ctx context.Context, event Event) {
|
||||
b.ctx = ctx
|
||||
b.event = event
|
||||
}
|
||||
|
||||
func (b *capturingBroker) SendBroadcastMessage(_ context.Context, event Event) {
|
||||
b.broadcast = event
|
||||
}
|
||||
|
||||
@ -197,6 +197,15 @@ func (b *broker) shouldSend(msg message, c client) bool {
|
||||
if broadcastToAll, ok := msg.senderCtx.Value(broadcastToAllKey).(bool); ok && broadcastToAll {
|
||||
return true
|
||||
}
|
||||
if sendToSameUser, ok := msg.senderCtx.Value(sendToSameUserKey).(bool); ok && sendToSameUser {
|
||||
if username, ok := request.UsernameFrom(msg.senderCtx); ok {
|
||||
return username == c.username
|
||||
}
|
||||
if user, ok := request.UserFrom(msg.senderCtx); ok {
|
||||
return user.UserName == c.username
|
||||
}
|
||||
return true
|
||||
}
|
||||
clientUniqueId, originatedFromClient := request.ClientUniqueIdFrom(msg.senderCtx)
|
||||
if !originatedFromClient {
|
||||
return true
|
||||
|
||||
@ -38,6 +38,18 @@ var _ = Describe("Broker", func() {
|
||||
m := message{senderCtx: ctx}
|
||||
Expect(b.shouldSend(m, c)).To(BeFalse())
|
||||
})
|
||||
It("sends same-user message for same username and same clientUniqueId", func() {
|
||||
ctx = request.WithClientUniqueId(ctx, "1111")
|
||||
ctx = request.WithUsername(ctx, "janedoe")
|
||||
m := message{senderCtx: sendToSameUser(ctx)}
|
||||
Expect(b.shouldSend(m, c)).To(BeTrue())
|
||||
})
|
||||
It("does not send same-user message for a different username", func() {
|
||||
ctx = request.WithClientUniqueId(ctx, "1111")
|
||||
ctx = request.WithUsername(ctx, "johndoe")
|
||||
m := message{senderCtx: sendToSameUser(ctx)}
|
||||
Expect(b.shouldSend(m, c)).To(BeFalse())
|
||||
})
|
||||
It("does not send message for different username", func() {
|
||||
ctx = request.WithClientUniqueId(ctx, "3333")
|
||||
ctx = request.WithUsername(ctx, "johndoe")
|
||||
|
||||
@ -29,7 +29,7 @@ var _ = Describe("Config API", func() {
|
||||
conf.Server.DevUIShowConfig = true // Enable config endpoint for tests
|
||||
ds = &tests.MockDataStore{}
|
||||
auth.Init(ds)
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil)
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil)
|
||||
router = server.JWTVerifier(nativeRouter)
|
||||
|
||||
// Create test users
|
||||
|
||||
@ -31,7 +31,7 @@ var _ = Describe("Library API", func() {
|
||||
conf.Server.EnableSharing = false
|
||||
ds = &tests.MockDataStore{}
|
||||
auth.Init(ds)
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil)
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil)
|
||||
router = server.JWTVerifier(nativeRouter)
|
||||
|
||||
// Create test users
|
||||
|
||||
@ -15,6 +15,7 @@ import (
|
||||
"github.com/navidrome/navidrome/core"
|
||||
"github.com/navidrome/navidrome/core/metrics"
|
||||
playlistsvc "github.com/navidrome/navidrome/core/playlists"
|
||||
coreradio "github.com/navidrome/navidrome/core/radio"
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/model/request"
|
||||
@ -34,6 +35,11 @@ type PluginManager interface {
|
||||
UnloadDisabledPlugins(ctx context.Context)
|
||||
}
|
||||
|
||||
type RadioMetadataManager interface {
|
||||
Start(ctx context.Context, sessionID string, station coreradio.Station) error
|
||||
Stop(sessionID string)
|
||||
}
|
||||
|
||||
type Router struct {
|
||||
http.Handler
|
||||
ds model.DataStore
|
||||
@ -45,10 +51,11 @@ type Router struct {
|
||||
maintenance core.Maintenance
|
||||
pluginManager PluginManager
|
||||
imgUpload core.ImageUploadService
|
||||
radioMetadata RadioMetadataManager
|
||||
}
|
||||
|
||||
func New(ds model.DataStore, share core.Share, playlists playlistsvc.Playlists, insights metrics.Insights, libraryService core.Library, userService core.User, maintenance core.Maintenance, pluginManager PluginManager, imgUpload core.ImageUploadService) *Router {
|
||||
r := &Router{ds: ds, share: share, playlists: playlists, insights: insights, libs: libraryService, users: userService, maintenance: maintenance, pluginManager: pluginManager, imgUpload: imgUpload}
|
||||
func New(ds model.DataStore, share core.Share, playlists playlistsvc.Playlists, insights metrics.Insights, libraryService core.Library, userService core.User, maintenance core.Maintenance, pluginManager PluginManager, imgUpload core.ImageUploadService, radioMetadata RadioMetadataManager) *Router {
|
||||
r := &Router{ds: ds, share: share, playlists: playlists, insights: insights, libs: libraryService, users: userService, maintenance: maintenance, pluginManager: pluginManager, imgUpload: imgUpload, radioMetadata: radioMetadata}
|
||||
r.Handler = r.routes()
|
||||
return r
|
||||
}
|
||||
|
||||
@ -95,7 +95,7 @@ var _ = Describe("Song Endpoints", func() {
|
||||
mfRepo.SetData(testSongs)
|
||||
|
||||
// Create the native API router and wrap it with the JWTVerifier middleware
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil)
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil)
|
||||
router = server.JWTVerifier(nativeRouter)
|
||||
w = httptest.NewRecorder()
|
||||
})
|
||||
|
||||
@ -99,7 +99,7 @@ var _ = Describe("Playlist Tracks Endpoint", func() {
|
||||
err := userRepo.Put(&testUser)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
nativeRouter := New(ds, nil, plsSvc, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil)
|
||||
nativeRouter := New(ds, nil, plsSvc, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, nil)
|
||||
router = server.JWTVerifier(nativeRouter)
|
||||
w = httptest.NewRecorder()
|
||||
})
|
||||
|
||||
@ -34,7 +34,7 @@ var _ = Describe("Plugin API", func() {
|
||||
ds = &tests.MockDataStore{}
|
||||
mockManager = &tests.MockPluginManager{}
|
||||
auth.Init(ds)
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, mockManager, nil)
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, mockManager, nil, nil)
|
||||
router = server.JWTVerifier(nativeRouter)
|
||||
|
||||
// Create test users
|
||||
|
||||
141
server/nativeapi/radio_nowplaying_test.go
Normal file
141
server/nativeapi/radio_nowplaying_test.go
Normal file
@ -0,0 +1,141 @@
|
||||
package nativeapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/consts"
|
||||
"github.com/navidrome/navidrome/core/auth"
|
||||
coreradio "github.com/navidrome/navidrome/core/radio"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/model/request"
|
||||
"github.com/navidrome/navidrome/server"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("Radio now-playing metadata endpoints", func() {
|
||||
var (
|
||||
router http.Handler
|
||||
ds *tests.MockDataStore
|
||||
radioRepo *tests.MockedRadioRepo
|
||||
userRepo *tests.MockedUserRepo
|
||||
manager *fakeRadioMetadataManager
|
||||
testUser model.User
|
||||
authRequest func(method, path string) *http.Request
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
DeferCleanup(configtest.SetupConfig())
|
||||
conf.Server.SessionTimeout = time.Minute
|
||||
conf.Server.EnableSharing = false
|
||||
|
||||
radioRepo = &tests.MockedRadioRepo{
|
||||
Data: map[string]*model.Radio{
|
||||
"rd-1": {
|
||||
ID: "rd-1",
|
||||
Name: "Test Radio",
|
||||
StreamUrl: "https://stream.example.test/radio",
|
||||
},
|
||||
},
|
||||
}
|
||||
userRepo = tests.CreateMockUserRepo()
|
||||
ds = &tests.MockDataStore{
|
||||
MockedRadio: radioRepo,
|
||||
MockedUser: userRepo,
|
||||
MockedProperty: &tests.MockedPropertyRepo{},
|
||||
}
|
||||
auth.Init(ds)
|
||||
|
||||
testUser = model.User{
|
||||
ID: "user-1",
|
||||
UserName: "testuser",
|
||||
Name: "Test User",
|
||||
IsAdmin: false,
|
||||
NewPassword: "testpass",
|
||||
}
|
||||
Expect(userRepo.Put(&testUser)).To(Succeed())
|
||||
|
||||
manager = &fakeRadioMetadataManager{}
|
||||
nativeRouter := New(ds, nil, nil, nil, tests.NewMockLibraryService(), tests.NewMockUserService(), nil, nil, nil, manager)
|
||||
router = server.JWTVerifier(nativeRouter)
|
||||
|
||||
authRequest = func(method, path string) *http.Request {
|
||||
req := httptest.NewRequest(method, path, nil)
|
||||
token, err := auth.CreateToken(&testUser)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
req.Header.Set(consts.UIAuthorizationHeader, "Bearer "+token)
|
||||
return req.WithContext(request.WithClientUniqueId(req.Context(), "client-1"))
|
||||
}
|
||||
})
|
||||
|
||||
It("starts metadata reading for a stored radio stream URL", func() {
|
||||
w := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/radio/rd-1/nowplaying")
|
||||
|
||||
router.ServeHTTP(w, req)
|
||||
|
||||
Expect(w.Code).To(Equal(http.StatusNoContent))
|
||||
Expect(manager.starts).To(Equal([]metadataStart{{
|
||||
sessionID: "user-1:client-1",
|
||||
station: coreradio.Station{
|
||||
ID: "rd-1",
|
||||
StreamURL: "https://stream.example.test/radio",
|
||||
},
|
||||
}}))
|
||||
})
|
||||
|
||||
It("returns not found when starting metadata for a missing radio", func() {
|
||||
w := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodPost, "/radio/missing/nowplaying")
|
||||
|
||||
router.ServeHTTP(w, req)
|
||||
|
||||
Expect(w.Code).To(Equal(http.StatusNotFound))
|
||||
Expect(manager.starts).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("returns unauthorized when request is not authenticated", func() {
|
||||
w := httptest.NewRecorder()
|
||||
req := httptest.NewRequest(http.MethodPost, "/radio/rd-1/nowplaying", nil)
|
||||
|
||||
router.ServeHTTP(w, req)
|
||||
|
||||
Expect(w.Code).To(Equal(http.StatusUnauthorized))
|
||||
Expect(manager.starts).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("stops metadata reading idempotently for the active session", func() {
|
||||
w := httptest.NewRecorder()
|
||||
req := authRequest(http.MethodDelete, "/radio/rd-1/nowplaying")
|
||||
|
||||
router.ServeHTTP(w, req)
|
||||
|
||||
Expect(w.Code).To(Equal(http.StatusNoContent))
|
||||
Expect(manager.stops).To(Equal([]string{"user-1:client-1"}))
|
||||
})
|
||||
})
|
||||
|
||||
type metadataStart struct {
|
||||
sessionID string
|
||||
station coreradio.Station
|
||||
}
|
||||
|
||||
type fakeRadioMetadataManager struct {
|
||||
starts []metadataStart
|
||||
stops []string
|
||||
}
|
||||
|
||||
func (m *fakeRadioMetadataManager) Start(_ context.Context, sessionID string, station coreradio.Station) error {
|
||||
m.starts = append(m.starts, metadataStart{sessionID: sessionID, station: station})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *fakeRadioMetadataManager) Stop(sessionID string) {
|
||||
m.stops = append(m.stops, sessionID)
|
||||
}
|
||||
@ -9,7 +9,9 @@ import (
|
||||
"github.com/deluan/rest"
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/navidrome/navidrome/consts"
|
||||
coreradio "github.com/navidrome/navidrome/core/radio"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/model/request"
|
||||
"github.com/navidrome/navidrome/server"
|
||||
)
|
||||
|
||||
@ -27,10 +29,74 @@ func (api *Router) addRadioRoute(r chi.Router) {
|
||||
r.Delete("/", rest.Delete(constructor))
|
||||
r.Post("/image", api.uploadRadioImage())
|
||||
r.Delete("/image", api.deleteRadioImage())
|
||||
r.Post("/nowplaying", api.startRadioNowPlaying())
|
||||
r.Delete("/nowplaying", api.stopRadioNowPlaying())
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func (api *Router) startRadioNowPlaying() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
if api.radioMetadata == nil {
|
||||
http.Error(w, "radio metadata is unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
|
||||
radioID := chi.URLParam(r, "id")
|
||||
radio, err := api.ds.Radio(r.Context()).Get(radioID)
|
||||
if err != nil {
|
||||
if errors.Is(err, model.ErrNotFound) {
|
||||
http.Error(w, "not found", http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
err = api.radioMetadata.Start(r.Context(), radioNowPlayingSessionID(r.Context()), coreradio.Station{
|
||||
ID: radio.ID,
|
||||
StreamURL: radio.StreamUrl,
|
||||
})
|
||||
if err != nil {
|
||||
if errors.Is(err, coreradio.ErrInvalidSession) || errors.Is(err, coreradio.ErrInvalidStation) {
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
}
|
||||
|
||||
func (api *Router) stopRadioNowPlaying() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
if api.radioMetadata == nil {
|
||||
http.Error(w, "radio metadata is unavailable", http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
|
||||
api.radioMetadata.Stop(radioNowPlayingSessionID(r.Context()))
|
||||
w.WriteHeader(http.StatusNoContent)
|
||||
}
|
||||
}
|
||||
|
||||
func radioNowPlayingSessionID(ctx context.Context) string {
|
||||
user, hasUser := request.UserFrom(ctx)
|
||||
clientID, _ := request.ClientUniqueIdFrom(ctx)
|
||||
if clientID == "" {
|
||||
clientID = "default"
|
||||
}
|
||||
if hasUser && user.ID != "" {
|
||||
return user.ID + ":" + clientID
|
||||
}
|
||||
if username, ok := request.UsernameFrom(ctx); ok && username != "" {
|
||||
return username + ":" + clientID
|
||||
}
|
||||
return clientID
|
||||
}
|
||||
|
||||
func (api *Router) uploadRadioImage() http.HandlerFunc {
|
||||
return handleImageUpload(func(ctx context.Context, reader io.Reader, ext string) error {
|
||||
radioID := chi.URLParamFromCtx(ctx, "id")
|
||||
|
||||
@ -3,6 +3,7 @@ export const EVENT_SERVER_START = 'serverStart'
|
||||
export const EVENT_REFRESH_RESOURCE = 'refreshResource'
|
||||
export const EVENT_NOW_PLAYING_COUNT = 'nowPlayingCount'
|
||||
export const EVENT_NOW_PLAYING_COUNT_SYNC = 'nowPlayingCountSync'
|
||||
export const EVENT_RADIO_NOW_PLAYING = 'radioNowPlaying'
|
||||
export const EVENT_STREAM_RECONNECTED = 'streamReconnected'
|
||||
|
||||
export const processEvent = (type, data) => ({
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import React from 'react'
|
||||
import { useMediaQuery } from '@material-ui/core'
|
||||
import { Link } from 'react-router-dom'
|
||||
import { useSelector } from 'react-redux'
|
||||
import clsx from 'clsx'
|
||||
import { QualityInfo } from '../common'
|
||||
import { decisionService } from '../transcode'
|
||||
@ -23,6 +24,26 @@ const AudioTitle = React.memo(({ audioInfo, gainInfo, isMobile }) => {
|
||||
[song],
|
||||
)
|
||||
|
||||
// The react-jinke-music-player library keeps its own internal snapshot of
|
||||
// the queue item (audioInfo), copied when playback starts. Redux updates
|
||||
// to state.player.current never reach that snapshot, so we read the live
|
||||
// radio title from redux directly rather than relying on audioInfo alone.
|
||||
const liveRadioTitle = useSelector((state) => {
|
||||
if (!audioInfo.isRadio) {
|
||||
return undefined
|
||||
}
|
||||
if (
|
||||
state.player?.current?.trackId === audioInfo.trackId &&
|
||||
state.player.current.radioTitle
|
||||
) {
|
||||
return state.player.current.radioTitle
|
||||
}
|
||||
const queued = state.player.queue?.find(
|
||||
(item) => item.isRadio && item.trackId === audioInfo.trackId,
|
||||
)
|
||||
return queued?.radioTitle
|
||||
})
|
||||
|
||||
if (!song) {
|
||||
return ''
|
||||
}
|
||||
@ -44,8 +65,19 @@ const AudioTitle = React.memo(({ audioInfo, gainInfo, isMobile }) => {
|
||||
}
|
||||
: {}
|
||||
|
||||
const stationTitle = song.title
|
||||
const subtitle = song.tags?.['subtitle']
|
||||
const title = song.title + (subtitle ? ` (${subtitle})` : '')
|
||||
// Keep the configured station name until ICY delivers a StreamTitle. Stations that
|
||||
// already embed the current song in the stream metadata show up via radioTitle;
|
||||
// others stay on stationTitle until the backend reader picks up ICY.
|
||||
const icyTitle = (
|
||||
liveRadioTitle ||
|
||||
audioInfo.radioTitle ||
|
||||
song.radioTitle ||
|
||||
''
|
||||
).trim()
|
||||
const displayTitle = icyTitle || stationTitle
|
||||
const title = displayTitle + (subtitle ? ` (${subtitle})` : '')
|
||||
|
||||
const linkTo = audioInfo.isRadio
|
||||
? `/radio/${audioInfo.trackId}/show`
|
||||
|
||||
@ -1,6 +1,8 @@
|
||||
import React from 'react'
|
||||
import { render, screen } from '@testing-library/react'
|
||||
import { describe, it, expect, beforeEach, vi } from 'vitest'
|
||||
import { Provider } from 'react-redux'
|
||||
import { createStore } from 'redux'
|
||||
import AudioTitle from './AudioTitle'
|
||||
|
||||
vi.mock('@material-ui/core', async () => {
|
||||
@ -24,6 +26,11 @@ vi.mock('react-dnd', () => ({
|
||||
useDrag: vi.fn(() => [null, () => {}]),
|
||||
}))
|
||||
|
||||
const renderWithStore = (ui, playerState = {}) => {
|
||||
const store = createStore(() => ({ player: playerState }))
|
||||
return render(<Provider store={store}>{ui}</Provider>)
|
||||
}
|
||||
|
||||
describe('<AudioTitle />', () => {
|
||||
const baseSong = {
|
||||
id: 'song-1',
|
||||
@ -41,7 +48,9 @@ describe('<AudioTitle />', () => {
|
||||
|
||||
it('links to playlist when playlistId is provided', () => {
|
||||
const audioInfo = { trackId: 'track-1', song: baseSong }
|
||||
render(<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />)
|
||||
renderWithStore(
|
||||
<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />,
|
||||
)
|
||||
const link = screen.getByRole('link')
|
||||
expect(link.getAttribute('href')).toBe('/playlist/playlist-1/show')
|
||||
})
|
||||
@ -51,8 +60,118 @@ describe('<AudioTitle />', () => {
|
||||
trackId: 'track-1',
|
||||
song: { ...baseSong, playlistId: undefined },
|
||||
}
|
||||
render(<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />)
|
||||
renderWithStore(
|
||||
<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />,
|
||||
)
|
||||
const link = screen.getByRole('link')
|
||||
expect(link.getAttribute('href')).toBe('/album/album-1/show')
|
||||
})
|
||||
|
||||
it('shows live radio title without changing the station link', () => {
|
||||
const audioInfo = {
|
||||
trackId: 'rd-1',
|
||||
isRadio: true,
|
||||
radioTitle: 'Artist - Title',
|
||||
song: {
|
||||
id: 'rd-1',
|
||||
title: 'Test Station',
|
||||
artist: 'Test Station',
|
||||
album: 'https://stream.example.test/radio',
|
||||
},
|
||||
}
|
||||
renderWithStore(
|
||||
<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />,
|
||||
)
|
||||
|
||||
expect(screen.getByText('Artist - Title')).toBeInTheDocument()
|
||||
expect(screen.getByRole('link').getAttribute('href')).toBe(
|
||||
'/radio/rd-1/show',
|
||||
)
|
||||
})
|
||||
|
||||
it('prefers the live redux title over the stale player-library snapshot', () => {
|
||||
const audioInfo = {
|
||||
trackId: 'radio-1',
|
||||
isRadio: true,
|
||||
// Note: no radioTitle here — this simulates react-jinke-music-player's
|
||||
// stale internal snapshot, which never receives redux updates.
|
||||
song: {
|
||||
id: 'radio-1',
|
||||
title: 'Station Name',
|
||||
artist: 'Station Name',
|
||||
album: 'https://stream.example.test/radio',
|
||||
},
|
||||
}
|
||||
renderWithStore(
|
||||
<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />,
|
||||
{
|
||||
current: {
|
||||
isRadio: true,
|
||||
trackId: 'radio-1',
|
||||
radioTitle: 'Live Song - Artist',
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
expect(screen.getByText('Live Song - Artist')).toBeInTheDocument()
|
||||
expect(screen.getByRole('link').getAttribute('href')).toBe(
|
||||
'/radio/radio-1/show',
|
||||
)
|
||||
})
|
||||
|
||||
it('falls back to the station name when redux current is a different radio', () => {
|
||||
const audioInfo = {
|
||||
trackId: 'radio-1',
|
||||
isRadio: true,
|
||||
song: {
|
||||
id: 'radio-1',
|
||||
title: 'Station Name',
|
||||
// Artist deliberately differs from title so getByText stays unambiguous
|
||||
artist: 'Station Artist',
|
||||
album: 'https://stream.example.test/radio',
|
||||
},
|
||||
}
|
||||
renderWithStore(
|
||||
<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />,
|
||||
{
|
||||
current: {
|
||||
isRadio: true,
|
||||
trackId: 'radio-2',
|
||||
radioTitle: 'Some Other Live Title',
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
expect(screen.getByText('Station Name')).toBeInTheDocument()
|
||||
expect(screen.queryByText('Some Other Live Title')).not.toBeInTheDocument()
|
||||
})
|
||||
|
||||
it('shows live title from the queue when current is not set yet', () => {
|
||||
const audioInfo = {
|
||||
trackId: 'radio-1',
|
||||
isRadio: true,
|
||||
song: {
|
||||
id: 'radio-1',
|
||||
title: 'Station Name',
|
||||
artist: 'Station Artist',
|
||||
album: 'https://stream.example.test/radio',
|
||||
},
|
||||
}
|
||||
renderWithStore(
|
||||
<AudioTitle audioInfo={audioInfo} gainInfo={{}} isMobile={false} />,
|
||||
{
|
||||
current: {},
|
||||
queue: [
|
||||
{
|
||||
trackId: 'radio-1',
|
||||
isRadio: true,
|
||||
radioTitle: 'Queued Live Title',
|
||||
song: audioInfo.song,
|
||||
},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
expect(screen.getByText('Queued Live Title')).toBeInTheDocument()
|
||||
})
|
||||
})
|
||||
|
||||
@ -34,6 +34,22 @@ import { keyMap } from '../hotkeys'
|
||||
import keyHandlers from './keyHandlers'
|
||||
import { calculateGain } from '../utils/calculateReplayGain'
|
||||
import { detectBrowserProfile, decisionService } from '../transcode'
|
||||
import { httpVoid } from '../dataProvider'
|
||||
import { REST_URL } from '../consts'
|
||||
|
||||
const radioNowPlayingURL = (radioId) =>
|
||||
`${REST_URL}/radio/${encodeURIComponent(radioId)}/nowplaying`
|
||||
|
||||
const requestRadioNowPlaying = (radioId, method) => {
|
||||
if (!radioId) {
|
||||
return Promise.resolve()
|
||||
}
|
||||
return httpVoid(radioNowPlayingURL(radioId), { method }).catch(() => {})
|
||||
}
|
||||
|
||||
// Server-side now-playing sessions expire after an idle TTL; re-POST while a
|
||||
// radio session is active so long listening sessions keep their live titles.
|
||||
const radioSessionRefreshMs = 4 * 60 * 1000
|
||||
|
||||
const Player = () => {
|
||||
const theme = useCurrentTheme()
|
||||
@ -46,6 +62,8 @@ const Player = () => {
|
||||
const [heartbeatTrackId, setHeartbeatTrackId] = useState(null)
|
||||
const lastPositionMsRef = useRef(0)
|
||||
const currentTrackIdRef = useRef(null)
|
||||
const currentRadioIdRef = useRef(null)
|
||||
const [activeRadioId, setActiveRadioId] = useState(null)
|
||||
const stoppedRef = useRef(false)
|
||||
const [audioInstance, setAudioInstance] = useState(null)
|
||||
const isDesktop = useMediaQuery('(min-width:810px)')
|
||||
@ -63,6 +81,41 @@ const Player = () => {
|
||||
|
||||
currentTrackIdRef.current = currentTrackId
|
||||
|
||||
const stopRadioNowPlaying = useCallback(
|
||||
(radioId = currentRadioIdRef.current) => {
|
||||
if (!radioId) {
|
||||
return
|
||||
}
|
||||
requestRadioNowPlaying(radioId, 'DELETE')
|
||||
if (currentRadioIdRef.current === radioId) {
|
||||
currentRadioIdRef.current = null
|
||||
setActiveRadioId(null)
|
||||
}
|
||||
},
|
||||
[],
|
||||
)
|
||||
|
||||
const startRadioNowPlaying = useCallback((radioId) => {
|
||||
if (!radioId || currentRadioIdRef.current === radioId) {
|
||||
return
|
||||
}
|
||||
if (currentRadioIdRef.current) {
|
||||
requestRadioNowPlaying(currentRadioIdRef.current, 'DELETE')
|
||||
}
|
||||
currentRadioIdRef.current = radioId
|
||||
setActiveRadioId(radioId)
|
||||
requestRadioNowPlaying(radioId, 'POST')
|
||||
}, [])
|
||||
|
||||
useInterval(
|
||||
() => {
|
||||
if (currentRadioIdRef.current) {
|
||||
requestRadioNowPlaying(currentRadioIdRef.current, 'POST')
|
||||
}
|
||||
},
|
||||
activeRadioId ? radioSessionRefreshMs : null,
|
||||
)
|
||||
|
||||
useInterval(
|
||||
() => {
|
||||
if (heartbeatTrackId && !stoppedRef.current) {
|
||||
@ -180,6 +233,10 @@ const Player = () => {
|
||||
}
|
||||
|
||||
const handlePageHide = () => {
|
||||
if (playerState.current?.isRadio) {
|
||||
stopRadioNowPlaying(playerState.current.trackId)
|
||||
return
|
||||
}
|
||||
if (currentTrackIdRef.current && !playerState.current?.isRadio) {
|
||||
stoppedRef.current = true
|
||||
try {
|
||||
@ -200,7 +257,7 @@ const Player = () => {
|
||||
window.removeEventListener('beforeunload', handleBeforeUnload)
|
||||
window.removeEventListener('pagehide', handlePageHide)
|
||||
}
|
||||
}, [playerState, audioInstance])
|
||||
}, [playerState, audioInstance, stopRadioNowPlaying])
|
||||
|
||||
const defaultOptions = useMemo(
|
||||
() => ({
|
||||
@ -285,6 +342,15 @@ const Player = () => {
|
||||
}
|
||||
|
||||
dispatch(currentPlaying(info))
|
||||
if (info.isRadio) {
|
||||
startRadioNowPlaying(info.trackId)
|
||||
setHeartbeatTrackId(null)
|
||||
setCurrentTrackId(null)
|
||||
return
|
||||
}
|
||||
if (currentRadioIdRef.current) {
|
||||
stopRadioNowPlaying()
|
||||
}
|
||||
if (info.duration) {
|
||||
const song = info.song
|
||||
document.title = `${song.title} - ${song.artist} - Navidrome`
|
||||
@ -320,24 +386,44 @@ const Player = () => {
|
||||
}
|
||||
}
|
||||
},
|
||||
[context, dispatch, showNotifications, currentTrackId],
|
||||
[
|
||||
context,
|
||||
dispatch,
|
||||
showNotifications,
|
||||
currentTrackId,
|
||||
startRadioNowPlaying,
|
||||
stopRadioNowPlaying,
|
||||
],
|
||||
)
|
||||
|
||||
const onAudioPlayTrackChange = useCallback(() => {
|
||||
if (currentTrackId) {
|
||||
subsonic.reportPlayback(
|
||||
currentTrackId,
|
||||
lastPositionMsRef.current,
|
||||
'stopped',
|
||||
)
|
||||
}
|
||||
setHeartbeatTrackId(null)
|
||||
setCurrentTrackId(null)
|
||||
}, [currentTrackId])
|
||||
const onAudioPlayTrackChange = useCallback(
|
||||
(playId, audioLists, audioInfo) => {
|
||||
// Switching to another radio is handled in onAudioPlay; avoid stopping the
|
||||
// session here or the ICY reader is torn down before playback resumes.
|
||||
if (!audioInfo?.isRadio && currentRadioIdRef.current) {
|
||||
stopRadioNowPlaying()
|
||||
}
|
||||
if (currentTrackId) {
|
||||
subsonic.reportPlayback(
|
||||
currentTrackId,
|
||||
lastPositionMsRef.current,
|
||||
'stopped',
|
||||
)
|
||||
}
|
||||
setHeartbeatTrackId(null)
|
||||
setCurrentTrackId(null)
|
||||
},
|
||||
[currentTrackId, stopRadioNowPlaying],
|
||||
)
|
||||
|
||||
const onAudioPause = useCallback(
|
||||
(info) => {
|
||||
dispatch(currentPlaying(info))
|
||||
if (info.isRadio) {
|
||||
stopRadioNowPlaying(info.trackId)
|
||||
setHeartbeatTrackId(null)
|
||||
return
|
||||
}
|
||||
if (!info.isRadio && currentTrackId) {
|
||||
const posMs = Math.floor(info.currentTime * 1000)
|
||||
lastPositionMsRef.current = posMs
|
||||
@ -345,11 +431,14 @@ const Player = () => {
|
||||
}
|
||||
setHeartbeatTrackId(null)
|
||||
},
|
||||
[dispatch, currentTrackId],
|
||||
[dispatch, currentTrackId, stopRadioNowPlaying],
|
||||
)
|
||||
|
||||
const onAudioEnded = useCallback(
|
||||
(currentPlayId, audioLists, info) => {
|
||||
if (info.isRadio) {
|
||||
stopRadioNowPlaying(info.trackId)
|
||||
}
|
||||
if (currentTrackId && !info.isRadio) {
|
||||
const posMs = Math.floor((info.duration || 0) * 1000)
|
||||
subsonic.reportPlayback(currentTrackId, posMs, 'stopped')
|
||||
@ -362,7 +451,7 @@ const Player = () => {
|
||||
// eslint-disable-next-line no-console
|
||||
.catch((e) => console.log('Keepalive error:', e))
|
||||
},
|
||||
[dispatch, dataProvider, currentTrackId],
|
||||
[dispatch, dataProvider, currentTrackId, stopRadioNowPlaying],
|
||||
)
|
||||
|
||||
const onCoverClick = useCallback((mode, audioLists, audioInfo) => {
|
||||
@ -395,6 +484,9 @@ const Player = () => {
|
||||
|
||||
const onBeforeDestroy = useCallback(() => {
|
||||
return new Promise((resolve, reject) => {
|
||||
if (currentRadioIdRef.current) {
|
||||
stopRadioNowPlaying()
|
||||
}
|
||||
if (currentTrackId && !playerStateRef.current?.current?.isRadio) {
|
||||
subsonic.reportPlayback(
|
||||
currentTrackId,
|
||||
@ -407,7 +499,7 @@ const Player = () => {
|
||||
dispatch(clearQueue())
|
||||
reject()
|
||||
})
|
||||
}, [dispatch, currentTrackId])
|
||||
}, [dispatch, currentTrackId, stopRadioNowPlaying])
|
||||
|
||||
if (!visible) {
|
||||
document.title = 'Navidrome'
|
||||
|
||||
@ -4,25 +4,54 @@ import { baseUrl } from '../utils'
|
||||
import config from '../config'
|
||||
import { jwtDecode } from 'jwt-decode'
|
||||
|
||||
const customAuthorizationHeader = 'X-ND-Authorization'
|
||||
export const customAuthorizationHeader = 'X-ND-Authorization'
|
||||
export const clientUniqueIdHeader = 'X-ND-Client-Unique-Id'
|
||||
export const clientUniqueId = uuidv4()
|
||||
|
||||
const applyAuthHeaders = (headers) => {
|
||||
headers.set(clientUniqueIdHeader, clientUniqueId)
|
||||
const token = localStorage.getItem('token')
|
||||
if (token) {
|
||||
headers.set(customAuthorizationHeader, `Bearer ${token}`)
|
||||
}
|
||||
return headers
|
||||
}
|
||||
|
||||
// Endpoints that return 204 No Content (e.g. radio now-playing) must not use fetchJson.
|
||||
export const httpVoid = (url, options = {}) => {
|
||||
url = baseUrl(url)
|
||||
const headers = applyAuthHeaders(
|
||||
options.headers instanceof Headers
|
||||
? options.headers
|
||||
: new Headers({ Accept: 'application/json', ...options.headers }),
|
||||
)
|
||||
return fetch(url, { ...options, headers }).then((response) => {
|
||||
const refreshedToken = response.headers.get(customAuthorizationHeader)
|
||||
if (refreshedToken) {
|
||||
const decoded = jwtDecode(refreshedToken)
|
||||
localStorage.setItem('token', refreshedToken)
|
||||
localStorage.setItem('userId', decoded.uid)
|
||||
config.firstTime = false
|
||||
removeHomeCache()
|
||||
}
|
||||
if (!response.ok) {
|
||||
throw new Error(`Request failed with status ${response.status}`)
|
||||
}
|
||||
return response
|
||||
})
|
||||
}
|
||||
|
||||
const httpClient = (url, options = {}) => {
|
||||
url = baseUrl(url)
|
||||
if (!options.headers) {
|
||||
options.headers = new Headers({ Accept: 'application/json' })
|
||||
}
|
||||
options.headers.set(clientUniqueIdHeader, clientUniqueId)
|
||||
const token = localStorage.getItem('token')
|
||||
if (token) {
|
||||
options.headers.set(customAuthorizationHeader, `Bearer ${token}`)
|
||||
}
|
||||
applyAuthHeaders(options.headers)
|
||||
return fetchUtils.fetchJson(url, options).then((response) => {
|
||||
const token = response.headers.get(customAuthorizationHeader)
|
||||
if (token) {
|
||||
const decoded = jwtDecode(token)
|
||||
localStorage.setItem('token', token)
|
||||
const refreshedToken = response.headers.get(customAuthorizationHeader)
|
||||
if (refreshedToken) {
|
||||
const decoded = jwtDecode(refreshedToken)
|
||||
localStorage.setItem('token', refreshedToken)
|
||||
localStorage.setItem('userId', decoded.uid)
|
||||
// Avoid going to create admin dialog after logout/login without a refresh
|
||||
config.firstTime = false
|
||||
|
||||
@ -1,6 +1,10 @@
|
||||
import httpClient, { clientUniqueId, clientUniqueIdHeader } from './httpClient'
|
||||
import httpClient, {
|
||||
clientUniqueId,
|
||||
clientUniqueIdHeader,
|
||||
httpVoid,
|
||||
} from './httpClient'
|
||||
import wrapperDataProvider from './wrapperDataProvider'
|
||||
|
||||
export { httpClient, clientUniqueId, clientUniqueIdHeader }
|
||||
export { httpClient, httpVoid, clientUniqueId, clientUniqueIdHeader }
|
||||
|
||||
export default wrapperDataProvider
|
||||
|
||||
@ -1,6 +1,11 @@
|
||||
import { baseUrl } from './utils'
|
||||
import throttle from 'lodash.throttle'
|
||||
import { processEvent, serverDown, streamReconnected } from './actions'
|
||||
import {
|
||||
EVENT_RADIO_NOW_PLAYING,
|
||||
processEvent,
|
||||
serverDown,
|
||||
streamReconnected,
|
||||
} from './actions'
|
||||
import { REST_URL } from './consts'
|
||||
import config from './config'
|
||||
|
||||
@ -20,6 +25,7 @@ const setupHandlers = (stream, dispatchFn) => {
|
||||
stream.addEventListener('serverStart', eventHandler(dispatchFn))
|
||||
stream.addEventListener('scanStatus', throttledEventHandler(dispatchFn))
|
||||
stream.addEventListener('refreshResource', eventHandler(dispatchFn))
|
||||
stream.addEventListener(EVENT_RADIO_NOW_PLAYING, eventHandler(dispatchFn))
|
||||
if (config.enableNowPlaying) {
|
||||
stream.addEventListener('nowPlayingCount', eventHandler(dispatchFn))
|
||||
}
|
||||
@ -76,6 +82,10 @@ const startEventStreamLegacy = async (dispatchFn) => {
|
||||
throttledEventHandler(dispatchFn),
|
||||
)
|
||||
newStream.addEventListener('refreshResource', eventHandler(dispatchFn))
|
||||
newStream.addEventListener(
|
||||
EVENT_RADIO_NOW_PLAYING,
|
||||
eventHandler(dispatchFn),
|
||||
)
|
||||
if (config.enableNowPlaying) {
|
||||
newStream.addEventListener('nowPlayingCount', eventHandler(dispatchFn))
|
||||
}
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
import { describe, it, beforeEach, vi, expect } from 'vitest'
|
||||
import { startEventStream } from './eventStream'
|
||||
import { serverDown } from './actions'
|
||||
import { processEvent, serverDown } from './actions'
|
||||
import config from './config'
|
||||
|
||||
class MockEventSource {
|
||||
@ -48,4 +48,24 @@ describe('startEventStream', () => {
|
||||
vi.advanceTimersByTime(5000)
|
||||
expect(global.EventSource).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
|
||||
it('dispatches radio now-playing events', async () => {
|
||||
await startEventStream(dispatch)
|
||||
instance.listeners.radioNowPlaying({
|
||||
type: 'radioNowPlaying',
|
||||
data: JSON.stringify({
|
||||
radioId: 'rd-1',
|
||||
title: 'Artist - Title',
|
||||
updatedAt: '2026-06-30T12:34:56Z',
|
||||
}),
|
||||
})
|
||||
|
||||
expect(dispatch).toHaveBeenCalledWith(
|
||||
processEvent('radioNowPlaying', {
|
||||
radioId: 'rd-1',
|
||||
title: 'Artist - Title',
|
||||
updatedAt: '2026-06-30T12:34:56Z',
|
||||
}),
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@ -12,6 +12,7 @@ import {
|
||||
PLAYER_SYNC_QUEUE,
|
||||
PLAYER_SET_MODE,
|
||||
PLAYER_REFRESH_QUEUE,
|
||||
EVENT_RADIO_NOW_PLAYING,
|
||||
} from '../actions'
|
||||
import config from '../config'
|
||||
|
||||
@ -182,10 +183,28 @@ const reduceSyncQueue = (state, { data: { audioInfo, audioLists } }) => {
|
||||
}
|
||||
|
||||
const reduceCurrent = (state, { data }) => {
|
||||
const current = data.ended ? {} : data
|
||||
let current = data.ended ? {} : data
|
||||
const savedPlayIndex = state.queue.findIndex(
|
||||
(item) => item.uuid === current.uuid,
|
||||
)
|
||||
// Preserve live radio titles when the music player snapshot omits them.
|
||||
if (current.isRadio && current.trackId) {
|
||||
const queued = state.queue.find(
|
||||
(item) =>
|
||||
(current.uuid && item.uuid === current.uuid) ||
|
||||
item.trackId === current.trackId,
|
||||
)
|
||||
// Only fall back to the previous current item when it is the same track,
|
||||
// otherwise switching stations would inherit the old station's title.
|
||||
const previousTitle =
|
||||
state.current?.trackId === current.trackId
|
||||
? state.current?.radioTitle
|
||||
: undefined
|
||||
const liveTitle = queued?.radioTitle ?? previousTitle
|
||||
if (liveTitle && !current.radioTitle) {
|
||||
current = applyRadioTitle(current, liveTitle)
|
||||
}
|
||||
}
|
||||
// When a track selection is pending (playIndex is set), keep it alive
|
||||
// until the music player confirms it actually switched to the requested
|
||||
// track. Without this, a premature onAudioPlay callback for the
|
||||
@ -208,6 +227,55 @@ const reduceMode = (state, { data: { mode } }) => {
|
||||
}
|
||||
}
|
||||
|
||||
const activeQueueItem = (state) => {
|
||||
if (state.current?.isRadio) {
|
||||
return state.current
|
||||
}
|
||||
const index = state.playIndex ?? state.savedPlayIndex
|
||||
return state.queue[index]
|
||||
}
|
||||
|
||||
const applyRadioTitle = (item, title) => {
|
||||
if (!item) {
|
||||
return item
|
||||
}
|
||||
const song = { ...(item.song || {}) }
|
||||
if (title) {
|
||||
return {
|
||||
...item,
|
||||
radioTitle: title,
|
||||
song: { ...song, radioTitle: title },
|
||||
}
|
||||
}
|
||||
delete song.radioTitle
|
||||
const next = { ...item, song }
|
||||
delete next.radioTitle
|
||||
return next
|
||||
}
|
||||
|
||||
const reduceRadioNowPlaying = (state, { data }) => {
|
||||
const radioId = data?.radioId
|
||||
const active = activeQueueItem(state)
|
||||
if (!radioId || !active?.isRadio || active.trackId !== radioId) {
|
||||
return state
|
||||
}
|
||||
|
||||
const title = (data.title || '').trim()
|
||||
const matchesActive = (item) =>
|
||||
active.uuid ? item.uuid === active.uuid : item.trackId === radioId
|
||||
|
||||
return {
|
||||
...state,
|
||||
queue: state.queue.map((item) =>
|
||||
matchesActive(item) ? applyRadioTitle(item, title) : item,
|
||||
),
|
||||
current:
|
||||
state.current?.isRadio && state.current.trackId === radioId
|
||||
? applyRadioTitle(state.current, title)
|
||||
: state.current,
|
||||
}
|
||||
}
|
||||
|
||||
export const playerReducer = (previousState = initialState, payload) => {
|
||||
const { type } = payload
|
||||
switch (type) {
|
||||
@ -229,6 +297,8 @@ export const playerReducer = (previousState = initialState, payload) => {
|
||||
return reduceCurrent(previousState, payload)
|
||||
case PLAYER_SET_MODE:
|
||||
return reduceMode(previousState, payload)
|
||||
case EVENT_RADIO_NOW_PLAYING:
|
||||
return reduceRadioNowPlaying(previousState, payload)
|
||||
case PLAYER_REFRESH_QUEUE: {
|
||||
const resolvedUrls = payload.data || {}
|
||||
return {
|
||||
|
||||
@ -4,6 +4,7 @@ import {
|
||||
PLAYER_SYNC_QUEUE,
|
||||
PLAYER_CURRENT,
|
||||
PLAYER_REFRESH_QUEUE,
|
||||
EVENT_RADIO_NOW_PLAYING,
|
||||
} from '../actions'
|
||||
|
||||
describe('playerReducer', () => {
|
||||
@ -224,4 +225,161 @@ describe('playerReducer', () => {
|
||||
expect(result.playIndex).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('radioNowPlaying event', () => {
|
||||
const radioItem = {
|
||||
trackId: 'rd-1',
|
||||
uuid: 'radio-uuid',
|
||||
name: 'Test Station',
|
||||
musicSrc: 'https://stream.example.test/radio',
|
||||
isRadio: true,
|
||||
song: {
|
||||
id: 'rd-1',
|
||||
title: 'Test Station',
|
||||
artist: 'Test Station',
|
||||
album: 'https://stream.example.test/radio',
|
||||
streamUrl: 'https://stream.example.test/radio',
|
||||
},
|
||||
}
|
||||
|
||||
it('updates the active radio with live title metadata', () => {
|
||||
const state = {
|
||||
queue: [radioItem],
|
||||
current: radioItem,
|
||||
savedPlayIndex: 0,
|
||||
clear: false,
|
||||
volume: 1,
|
||||
}
|
||||
const result = playerReducer(state, {
|
||||
type: EVENT_RADIO_NOW_PLAYING,
|
||||
data: {
|
||||
radioId: 'rd-1',
|
||||
title: 'Artist - Title',
|
||||
updatedAt: '2026-06-30T12:34:56Z',
|
||||
},
|
||||
})
|
||||
|
||||
expect(result.current.radioTitle).toBe('Artist - Title')
|
||||
expect(result.current.song.radioTitle).toBe('Artist - Title')
|
||||
expect(result.current.name).toBe('Test Station')
|
||||
expect(result.current.song.title).toBe('Test Station')
|
||||
expect(result.current.musicSrc).toBe('https://stream.example.test/radio')
|
||||
expect(result.queue[0].radioTitle).toBe('Artist - Title')
|
||||
})
|
||||
|
||||
it('ignores events for non-active radios', () => {
|
||||
const state = {
|
||||
queue: [radioItem],
|
||||
current: radioItem,
|
||||
savedPlayIndex: 0,
|
||||
clear: false,
|
||||
volume: 1,
|
||||
}
|
||||
const result = playerReducer(state, {
|
||||
type: EVENT_RADIO_NOW_PLAYING,
|
||||
data: { radioId: 'rd-2', title: 'Other Station - Title' },
|
||||
})
|
||||
|
||||
expect(result).toBe(state)
|
||||
})
|
||||
|
||||
it('clears live title metadata when the title is empty', () => {
|
||||
const liveRadio = {
|
||||
...radioItem,
|
||||
radioTitle: 'Old Title',
|
||||
song: { ...radioItem.song, radioTitle: 'Old Title' },
|
||||
}
|
||||
const state = {
|
||||
queue: [liveRadio],
|
||||
current: liveRadio,
|
||||
savedPlayIndex: 0,
|
||||
clear: false,
|
||||
volume: 1,
|
||||
}
|
||||
const result = playerReducer(state, {
|
||||
type: EVENT_RADIO_NOW_PLAYING,
|
||||
data: { radioId: 'rd-1', title: '' },
|
||||
})
|
||||
|
||||
expect(result.current.radioTitle).toBeUndefined()
|
||||
expect(result.current.song.radioTitle).toBeUndefined()
|
||||
expect(result.current.song.title).toBe('Test Station')
|
||||
expect(result.current.musicSrc).toBe('https://stream.example.test/radio')
|
||||
})
|
||||
|
||||
it('updates the queued active radio before current is set', () => {
|
||||
const state = {
|
||||
queue: [radioItem],
|
||||
current: {},
|
||||
savedPlayIndex: 0,
|
||||
clear: false,
|
||||
volume: 1,
|
||||
}
|
||||
const result = playerReducer(state, {
|
||||
type: EVENT_RADIO_NOW_PLAYING,
|
||||
data: { radioId: 'rd-1', title: 'Artist - Title' },
|
||||
})
|
||||
|
||||
expect(result.current).toEqual({})
|
||||
expect(result.queue[0].radioTitle).toBe('Artist - Title')
|
||||
})
|
||||
|
||||
it('preserves live radio title when the player snapshot omits it', () => {
|
||||
const liveRadio = {
|
||||
...radioItem,
|
||||
radioTitle: 'Artist - Title',
|
||||
song: { ...radioItem.song, radioTitle: 'Artist - Title' },
|
||||
}
|
||||
const state = {
|
||||
queue: [liveRadio],
|
||||
current: {},
|
||||
savedPlayIndex: 0,
|
||||
clear: false,
|
||||
volume: 1,
|
||||
}
|
||||
const result = playerReducer(state, {
|
||||
type: PLAYER_CURRENT,
|
||||
data: { ...radioItem },
|
||||
})
|
||||
|
||||
expect(result.current.radioTitle).toBe('Artist - Title')
|
||||
expect(result.current.song.radioTitle).toBe('Artist - Title')
|
||||
})
|
||||
|
||||
it('does not inherit the previous station title when switching stations', () => {
|
||||
const previousStation = {
|
||||
...radioItem,
|
||||
radioTitle: 'Old Station Song',
|
||||
song: { ...radioItem.song, radioTitle: 'Old Station Song' },
|
||||
}
|
||||
const nextStation = {
|
||||
trackId: 'rd-2',
|
||||
uuid: 'radio-uuid-2',
|
||||
name: 'Other Station',
|
||||
musicSrc: 'https://stream.example.test/other',
|
||||
isRadio: true,
|
||||
song: {
|
||||
id: 'rd-2',
|
||||
title: 'Other Station',
|
||||
artist: 'Other Station',
|
||||
album: 'https://stream.example.test/other',
|
||||
streamUrl: 'https://stream.example.test/other',
|
||||
},
|
||||
}
|
||||
const state = {
|
||||
queue: [previousStation, nextStation],
|
||||
current: previousStation,
|
||||
savedPlayIndex: 0,
|
||||
clear: false,
|
||||
volume: 1,
|
||||
}
|
||||
const result = playerReducer(state, {
|
||||
type: PLAYER_CURRENT,
|
||||
data: { ...nextStation },
|
||||
})
|
||||
|
||||
expect(result.current.radioTitle).toBeUndefined()
|
||||
expect(result.current.song.title).toBe('Other Station')
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user