diff --git a/core/radio/metadata_manager.go b/core/radio/metadata_manager.go new file mode 100644 index 000000000..f4cefe1cd --- /dev/null +++ b/core/radio/metadata_manager.go @@ -0,0 +1,224 @@ +package radio + +import ( + "context" + "errors" + "sync" + "time" +) + +var ( + ErrInvalidSession = errors.New("invalid radio metadata session") + ErrInvalidStation = errors.New("invalid radio metadata station") +) + +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 + + sessions map[string]activeSession + readers map[string]*activeReader +} + +type ManagerOption func(*MetadataManager) + +type activeSession struct { + radioID string + streamURL string +} + +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, + 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 + } + } +} + +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 { + return nil + } + m.removeSessionLocked(sessionID) + } + + reader := m.readers[station.StreamURL] + if reader == nil { + readerCtx, cancel := context.WithCancel(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]++ + m.sessions[sessionID] = activeSession{radioID: station.ID, streamURL: station.StreamURL} + return nil +} + +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) + + 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++ + } + + delay := m.backoff(attempt) + 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 + m.mu.Unlock() + + if publish == nil { + return + } + for _, radioID := range radioIDs { + publish(reader.ctx, TitleUpdate{ + RadioID: 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 +} diff --git a/core/radio/metadata_manager_test.go b/core/radio/metadata_manager_test.go new file mode 100644 index 000000000..6b1112143 --- /dev/null +++ b/core/radio/metadata_manager_test.go @@ -0,0 +1,231 @@ +package radio + +import ( + "context" + "errors" + "sync" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("MetadataManager", func() { + 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("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)) + }) +}) + +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 +} + +func newFakePublisher() *fakePublisher { + return &fakePublisher{} +} + +func (p *fakePublisher) Publish(_ context.Context, update TitleUpdate) { + p.mu.Lock() + defer p.mu.Unlock() + p.updates = append(p.updates, update) +} + +func (p *fakePublisher) Updates() []TitleUpdate { + p.mu.Lock() + defer p.mu.Unlock() + return append([]TitleUpdate(nil), p.updates...) +} diff --git a/core/radio/radio_suite_test.go b/core/radio/radio_suite_test.go new file mode 100644 index 000000000..cf7ce602e --- /dev/null +++ b/core/radio/radio_suite_test.go @@ -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") +}