diff --git a/pkg/service/soundtouchweb/handler.go b/pkg/service/soundtouchweb/handler.go index 37ea09d2..829a0808 100644 --- a/pkg/service/soundtouchweb/handler.go +++ b/pkg/service/soundtouchweb/handler.go @@ -56,6 +56,10 @@ type WebApp struct { // client cannot indefinitely block updates for healthy clients. webSocketWriteTimeout time.Duration + deviceBroadcastMu sync.Mutex + deviceBroadcastPending bool + deviceBroadcastRunning bool + Version string Commit string Date string @@ -773,6 +777,43 @@ func (app *WebApp) BroadcastDeviceList() { } } +// QueueDeviceListBroadcast schedules one device projection without blocking a +// speaker's event read loop on browser I/O. At most one worker runs per app; +// events during a slow write are coalesced into one follow-up snapshot rather +// than expanding into an unbounded queue or set of goroutines. +func (app *WebApp) QueueDeviceListBroadcast() { + app.deviceBroadcastMu.Lock() + + app.deviceBroadcastPending = true + if app.deviceBroadcastRunning { + app.deviceBroadcastMu.Unlock() + + return + } + + app.deviceBroadcastRunning = true + app.deviceBroadcastMu.Unlock() + + go app.runDeviceListBroadcasts() +} + +func (app *WebApp) runDeviceListBroadcasts() { + for { + app.deviceBroadcastMu.Lock() + if !app.deviceBroadcastPending { + app.deviceBroadcastRunning = false + app.deviceBroadcastMu.Unlock() + + return + } + + app.deviceBroadcastPending = false + app.deviceBroadcastMu.Unlock() + + app.BroadcastDeviceList() + } +} + // BroadcastDiscoveryStatus sends discovery progress updates to all connected WebSocket clients func (app *WebApp) BroadcastDiscoveryStatus(status string, deviceCount int) { discoveryStatus := &webtypes.DiscoveryStatus{ diff --git a/pkg/service/soundtouchweb/websocket.go b/pkg/service/soundtouchweb/websocket.go index f41478df..7fedd7e0 100644 --- a/pkg/service/soundtouchweb/websocket.go +++ b/pkg/service/soundtouchweb/websocket.go @@ -6,6 +6,7 @@ import ( "errors" "log" "net/http" + "reflect" "sync" "time" @@ -158,6 +159,90 @@ func (app *WebApp) registerGlobalWebSocket(conn *websocket.Conn) error { }) } +// applySpeakerStatusEvent stores one speaker event and immediately publishes +// a fresh device projection when its dashboard-visible payload changed. +func (app *WebApp) applySpeakerStatusEvent( + conn *webtypes.DeviceConnection, + mut func(*webtypes.DeviceStatus) bool, +) bool { + changed := false + + conn.ApplySpeakerEvent(func(status *webtypes.DeviceStatus) { + changed = mut(status) + }) + + if changed { + app.QueueDeviceListBroadcast() + } + + return changed +} + +func (app *WebApp) applyNowPlayingEvent( + conn *webtypes.DeviceConnection, + nowPlaying *models.NowPlaying, +) bool { + return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool { + changed := !reflect.DeepEqual(status.NowPlaying, nowPlaying) + status.NowPlaying = nowPlaying + status.LastActivity = time.Now() + + return changed + }) +} + +func (app *WebApp) applyVolumeEvent( + conn *webtypes.DeviceConnection, + volume *models.Volume, +) bool { + return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool { + changed := !reflect.DeepEqual(status.Volume, volume) + status.Volume = volume + status.LastActivity = time.Now() + + return changed + }) +} + +func (app *WebApp) applyConnectionStateEvent( + conn *webtypes.DeviceConnection, + connected bool, +) bool { + return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool { + changed := status.IsConnected != connected + status.IsConnected = connected + status.LastActivity = time.Now() + + return changed + }) +} + +func (app *WebApp) applyPresetEvent( + conn *webtypes.DeviceConnection, + presets *models.Presets, +) bool { + return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool { + changed := !reflect.DeepEqual(status.Presets, presets) + status.Presets = presets + status.LastActivity = time.Now() + + return changed + }) +} + +func (app *WebApp) applyBassEvent( + conn *webtypes.DeviceConnection, + bass *models.Bass, +) bool { + return app.applySpeakerStatusEvent(conn, func(status *webtypes.DeviceStatus) bool { + changed := !reflect.DeepEqual(status.Bass, bass) + status.Bass = bass + status.LastActivity = time.Now() + + return changed + }) +} + // HandleWebSocket handles WebSocket connections for real-time updates func (app *WebApp) HandleWebSocket(w http.ResponseWriter, r *http.Request) { conn, err := app.Upgrader.Upgrade(w, r, nil) @@ -323,35 +408,27 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device prevSource = np.Source - conn.UpdateStatus(func(s *webtypes.DeviceStatus) { - s.NowPlaying = np - s.LastActivity = time.Now() - }) + app.applyNowPlayingEvent(conn, np) }) wsClient.OnVolumeUpdated(func(event *models.VolumeUpdatedEvent) { - conn.UpdateStatus(func(s *webtypes.DeviceStatus) { - s.Volume = &event.Volume - s.LastActivity = time.Now() - }) + app.applyVolumeEvent(conn, &event.Volume) }) wsClient.OnConnectionState(func(event *models.ConnectionStateUpdatedEvent) { - conn.UpdateStatus(func(s *webtypes.DeviceStatus) { - s.IsConnected = event.ConnectionState.IsConnected() - s.LastActivity = time.Now() - }) + app.applyConnectionStateEvent(conn, event.ConnectionState.IsConnected()) }) wsClient.OnPresetUpdated(func(event *models.PresetUpdatedEvent) { - conn.UpdateStatus(func(s *webtypes.DeviceStatus) { - s.Presets = &event.Presets - s.LastActivity = time.Now() - }) + app.applyPresetEvent(conn, &event.Presets) + }) + + wsClient.OnBassUpdated(func(event *models.BassUpdatedEvent) { + app.applyBassEvent(conn, &event.Bass) }) wsClient.OnGroupUpdated(func(event *models.GroupUpdatedEvent) { - applyGroupUpdatedEvent(conn, event) + app.applyGroupUpdatedEvent(conn, event) }) if err := wsClient.Connect(); err != nil { @@ -371,9 +448,7 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device conn.WebSocket = wsClient - conn.UpdateStatus(func(s *webtypes.DeviceStatus) { - s.IsConnected = true - }) + app.applyConnectionStateEvent(conn, true) log.Printf("WebSocket connected for device %s", sanitizeLog(deviceID)) @@ -389,9 +464,7 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device // Block until the device-side WebSocket disconnects. wsClient.Wait() - conn.UpdateStatus(func(s *webtypes.DeviceStatus) { - s.IsConnected = false - }) + app.applyConnectionStateEvent(conn, false) log.Printf("WebSocket disconnected for device %s — reconnecting in %s", sanitizeLog(deviceID), backoff) @@ -434,6 +507,8 @@ func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection) return } + pollGeneration := conn.BeginStatusPoll() + // /getGroup must be gated to ST10 models -- see Client.GetGroup's doc // comment (verified against real hardware: a ST20 never replies at all, // hanging until the client's timeout instead of returning quickly). @@ -465,7 +540,7 @@ func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection) // Phase 2: fast merge. Only fields we successfully fetched // overwrite; everything else keeps the value other goroutines may // have just written. - conn.UpdateStatus(func(s *webtypes.DeviceStatus) { + conn.CompleteStatusPoll(pollGeneration, func(s *webtypes.DeviceStatus) { statusUpdated := false if nowPlayingErr == nil { @@ -509,8 +584,16 @@ func (app *WebApp) UpdateDeviceStatus(_ string, conn *webtypes.DeviceConnection) } } -func applyGroupUpdatedEvent(conn *webtypes.DeviceConnection, event *models.GroupUpdatedEvent) { - conn.ApplyGroupEvent(&event.Group, time.Now()) +func (app *WebApp) applyGroupUpdatedEvent( + conn *webtypes.DeviceConnection, + event *models.GroupUpdatedEvent, +) bool { + changed := conn.ApplyGroupEvent(&event.Group, time.Now()) + if changed { + app.QueueDeviceListBroadcast() + } + + return changed } // HandleDeviceWebSocket handles individual device WebSocket connections for real-time device-specific updates diff --git a/pkg/service/soundtouchweb/websocket_test.go b/pkg/service/soundtouchweb/websocket_test.go index c4dd607d..090635ed 100644 --- a/pkg/service/soundtouchweb/websocket_test.go +++ b/pkg/service/soundtouchweb/websocket_test.go @@ -2,6 +2,10 @@ package soundtouchweb import ( "errors" + "github.com/gesellix/bose-soundtouch/pkg/client" + "github.com/gesellix/bose-soundtouch/pkg/models" + "github.com/gesellix/bose-soundtouch/pkg/service/soundtouchweb/webtypes" + "github.com/gorilla/websocket" "net/http" "net/http/httptest" "strings" @@ -9,11 +13,6 @@ import ( "sync/atomic" "testing" "time" - - "github.com/gesellix/bose-soundtouch/pkg/client" - "github.com/gesellix/bose-soundtouch/pkg/models" - "github.com/gesellix/bose-soundtouch/pkg/service/soundtouchweb/webtypes" - "github.com/gorilla/websocket" ) func TestUpdateDeviceStatusRefreshesGroup(t *testing.T) { @@ -142,6 +141,7 @@ func TestUpdateDeviceStatusNotConnectedWhenOnlyGroupSucceeds(t *testing.T) { } func TestApplyGroupUpdatedEventReplacesGroup(t *testing.T) { + app := NewWebApp() conn := webtypes.NewDeviceConnection(nil, nil) previousActivity := time.Unix(1, 0) conn.SetStatus(&webtypes.DeviceStatus{ @@ -154,7 +154,9 @@ func TestApplyGroupUpdatedEventReplacesGroup(t *testing.T) { event := &models.GroupUpdatedEvent{ Group: models.Group{ID: "pair-new", Name: "Renamed Pair"}, } - applyGroupUpdatedEvent(conn, event) + if !app.applyGroupUpdatedEvent(conn, event) { + t.Fatal("new group event should publish a changed projection") + } status := conn.Status() if status.Group != &event.Group || status.Group.ID != "pair-new" { @@ -170,7 +172,9 @@ func TestApplyGroupUpdatedEventReplacesGroup(t *testing.T) { } teardown := &models.GroupUpdatedEvent{Group: models.Group{}} - applyGroupUpdatedEvent(conn, teardown) + if !app.applyGroupUpdatedEvent(conn, teardown) { + t.Fatal("teardown event should publish a changed projection") + } if conn.Status().Group != nil { t.Errorf("teardown event did not clear the group: %+v", conn.Status().Group) @@ -684,3 +688,211 @@ func TestWebSocketWriteBatchRefreshesDeadlineForHealthyWriter(t *testing.T) { t.Fatalf("healthy writer deadline = %v, want a fresh deadline after %v", deadline, started) } } + +func TestSpeakerEventHelpersPublishOnlyChangedPayloads(t *testing.T) { + app := NewWebApp() + conn := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + + nowPlaying := &models.NowPlaying{Source: "LOCAL_INTERNET_RADIO", Track: "Test station"} + if !app.applyNowPlayingEvent(conn, nowPlaying) { + t.Fatal("new now-playing payload was not reported as changed") + } + if app.applyNowPlayingEvent(conn, nowPlaying) { + t.Fatal("duplicate now-playing payload was reported as changed") + } + + volume := &models.Volume{ActualVolume: 25, TargetVolume: 25} + if !app.applyVolumeEvent(conn, volume) { + t.Fatal("new volume payload was not reported as changed") + } + if app.applyVolumeEvent(conn, volume) { + t.Fatal("duplicate volume payload was reported as changed") + } + + if !app.applyConnectionStateEvent(conn, true) { + t.Fatal("new connection state was not reported as changed") + } + if app.applyConnectionStateEvent(conn, true) { + t.Fatal("duplicate connection state was reported as changed") + } + + presets := &models.Presets{Preset: []models.Preset{{ID: 1}}} + if !app.applyPresetEvent(conn, presets) { + t.Fatal("new presets payload was not reported as changed") + } + if app.applyPresetEvent(conn, presets) { + t.Fatal("duplicate presets payload was reported as changed") + } + + bass := &models.Bass{ActualBass: -2} + if !app.applyBassEvent(conn, bass) { + t.Fatal("new bass payload was not reported as changed") + } + if app.applyBassEvent(conn, bass) { + t.Fatal("duplicate bass payload was reported as changed") + } +} + +func TestSpeakerEventPublishesImmediateDeviceProjection(t *testing.T) { + app := NewWebApp() + device := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + device.SetStatus(&webtypes.DeviceStatus{ + Volume: &models.Volume{ActualVolume: 10, TargetVolume: 10}, + IsConnected: true, + }) + app.AddDevice("speaker", device) + + serverConnection := make(chan *websocket.Conn, 1) + releaseServer := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := app.Upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("upgrade test WebSocket: %v", err) + + return + } + serverConnection <- conn + <-releaseServer + _ = conn.Close() + })) + + client, response, err := websocket.DefaultDialer.Dial( + "ws"+strings.TrimPrefix(server.URL, "http"), nil, + ) + if response != nil { + _ = response.Body.Close() + } + if err != nil { + close(releaseServer) + server.Close() + t.Fatalf("dial test WebSocket: %v", err) + } + remote := <-serverConnection + t.Cleanup(func() { + app.removeGlobalWebSocketClient(remote) + _ = client.Close() + close(releaseServer) + server.Close() + }) + + if err := app.registerGlobalWebSocket(remote); err != nil { + t.Fatalf("register browser WebSocket: %v", err) + } + if err := client.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatalf("set initial read deadline: %v", err) + } + var initial webtypes.WebSocketMessage + if err := client.ReadJSON(&initial); err != nil { + t.Fatalf("read initial device projection: %v", err) + } + + app.webSocketWriteMu.Lock() + writerLocked := true + defer func() { + if writerLocked { + app.webSocketWriteMu.Unlock() + } + }() + + applied := make(chan bool, 1) + go func() { + applied <- app.applyVolumeEvent(device, &models.Volume{ActualVolume: 25, TargetVolume: 25}) + }() + + select { + case changed := <-applied: + if !changed { + t.Fatal("volume event was not applied") + } + case <-time.After(250 * time.Millisecond): + t.Fatal("speaker event blocked on browser WebSocket I/O") + } + + app.webSocketWriteMu.Unlock() + writerLocked = false + + if err := client.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatalf("set event read deadline: %v", err) + } + var update webtypes.WebSocketMessage + if err := client.ReadJSON(&update); err != nil { + t.Fatalf("read immediate device projection: %v", err) + } + if update.Type != "devices" { + t.Fatalf("event frame type = %q, want devices", update.Type) + } + + devices, ok := update.Data.(map[string]interface{}) + if !ok { + t.Fatalf("event projection = %#v, want device map", update.Data) + } + speaker, ok := devices["speaker"].(map[string]interface{}) + if !ok { + t.Fatalf("speaker projection = %#v, want object", devices["speaker"]) + } + status, ok := speaker["status"].(map[string]interface{}) + if !ok { + t.Fatalf("speaker status = %#v, want object", speaker["status"]) + } + volume, ok := status["volume"].(map[string]interface{}) + if !ok || volume["ActualVolume"] != float64(25) { + t.Fatalf("projected volume = %#v, want 25", status["volume"]) + } +} + +func TestQueuedDeviceBroadcastCoalescesBurst(t *testing.T) { + app := NewWebApp() + + app.webSocketWriteMu.Lock() + writerLocked := true + defer func() { + if writerLocked { + app.webSocketWriteMu.Unlock() + } + }() + + app.QueueDeviceListBroadcast() + deadline := time.Now().Add(time.Second) + for { + app.deviceBroadcastMu.Lock() + running := app.deviceBroadcastRunning + pending := app.deviceBroadcastPending + app.deviceBroadcastMu.Unlock() + if running && !pending { + break + } + if time.Now().After(deadline) { + t.Fatal("broadcast worker did not begin its blocked write") + } + time.Sleep(time.Millisecond) + } + + for range 100 { + app.QueueDeviceListBroadcast() + } + + app.deviceBroadcastMu.Lock() + if !app.deviceBroadcastPending { + app.deviceBroadcastMu.Unlock() + t.Fatal("burst did not retain one coalesced follow-up") + } + app.deviceBroadcastMu.Unlock() + + app.webSocketWriteMu.Unlock() + writerLocked = false + + deadline = time.Now().Add(time.Second) + for { + app.deviceBroadcastMu.Lock() + running := app.deviceBroadcastRunning + pending := app.deviceBroadcastPending + app.deviceBroadcastMu.Unlock() + if !running && !pending { + break + } + if time.Now().After(deadline) { + t.Fatalf("coalesced worker did not drain: running=%v pending=%v", running, pending) + } + time.Sleep(time.Millisecond) + } +} diff --git a/pkg/service/soundtouchweb/webtypes/status_test.go b/pkg/service/soundtouchweb/webtypes/status_test.go index c3da39e7..1ee38f84 100644 --- a/pkg/service/soundtouchweb/webtypes/status_test.go +++ b/pkg/service/soundtouchweb/webtypes/status_test.go @@ -248,6 +248,64 @@ func TestApplyGroupEventIgnoresRoleOrder(t *testing.T) { } } +func TestStatusPollCannotOverwriteNewerSpeakerEvent(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + poll := conn.BeginStatusPoll() + + conn.ApplySpeakerEvent(func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 99} + }) + + if conn.CompleteStatusPoll(poll, func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 42} + }) { + t.Fatal("poll that began before a speaker event was applied") + } + + if got := conn.Status().Volume; got == nil || got.ActualVolume != 99 { + t.Fatalf("speaker event was overwritten by older poll data: %+v", got) + } +} + +func TestNewerStatusPollSupersedesOlderPoll(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + older := conn.BeginStatusPoll() + newer := conn.BeginStatusPoll() + + if !conn.CompleteStatusPoll(newer, func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 30} + }) { + t.Fatal("newer poll was not applied") + } + + if conn.CompleteStatusPoll(older, func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 10} + }) { + t.Fatal("older poll completed after a newer poll was applied") + } + + if got := conn.Status().Volume; got == nil || got.ActualVolume != 30 { + t.Fatalf("newer poll state was overwritten: %+v", got) + } +} + +func TestDuplicateSpeakerEventStillInvalidatesOlderPoll(t *testing.T) { + conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"}) + conn.SetStatus(&DeviceStatus{Volume: &models.Volume{ActualVolume: 25}}) + poll := conn.BeginStatusPoll() + + conn.ApplySpeakerEvent(func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 25} + status.LastActivity = time.Now() + }) + + if conn.CompleteStatusPoll(poll, func(status *DeviceStatus) { + status.Volume = &models.Volume{ActualVolume: 10} + }) { + t.Fatal("poll that preceded duplicate speaker evidence was applied") + } +} + func TestStatusSnapshotIsolation(t *testing.T) { // A snapshot returned by Status() must NOT change when a later // UpdateStatus replaces a pointer field. This proves the atomic diff --git a/pkg/service/soundtouchweb/webtypes/types.go b/pkg/service/soundtouchweb/webtypes/types.go index 5f71b983..01a74620 100644 --- a/pkg/service/soundtouchweb/webtypes/types.go +++ b/pkg/service/soundtouchweb/webtypes/types.go @@ -47,6 +47,12 @@ type DeviceConnection struct { status atomic.Pointer[DeviceStatus] + statusOrderMu sync.Mutex + nextStatusPollGeneration uint64 + lastStatusPollGeneration uint64 + speakerEventGeneration uint64 + statusPollEventGeneration map[uint64]uint64 + // groupMu orders polled /getGroup responses against real-time // groupUpdated events. groupGeneration is the highest generation // issued (by BeginGroupRefresh or ApplyGroupEvent); @@ -135,6 +141,66 @@ func (c *DeviceConnection) SetStatus(s *DeviceStatus) { c.status.Store(s) } +// BeginStatusPoll reserves an ordering generation before a status poll starts +// network I/O. CompleteStatusPoll uses it to reject an older poll after either +// a newer poll or a real-time speaker event has supplied fresher state. +func (c *DeviceConnection) BeginStatusPoll() uint64 { + c.statusOrderMu.Lock() + defer c.statusOrderMu.Unlock() + + c.nextStatusPollGeneration++ + if c.statusPollEventGeneration == nil { + c.statusPollEventGeneration = make(map[uint64]uint64) + } + + c.statusPollEventGeneration[c.nextStatusPollGeneration] = c.speakerEventGeneration + + return c.nextStatusPollGeneration +} + +// ApplySpeakerEvent serializes a real-time speaker event against status-poll +// completion. Even a duplicate event invalidates polls that started before it: +// the event is newer evidence than their fetched payload. +func (c *DeviceConnection) ApplySpeakerEvent(mut func(*DeviceStatus)) { + c.statusOrderMu.Lock() + defer c.statusOrderMu.Unlock() + + c.speakerEventGeneration++ + c.UpdateStatus(mut) +} + +// CompleteStatusPoll applies a poll result only when no newer poll has already +// completed and no speaker event arrived after this poll began. +func (c *DeviceConnection) CompleteStatusPoll( + generation uint64, + mut func(*DeviceStatus), +) bool { + c.statusOrderMu.Lock() + defer c.statusOrderMu.Unlock() + + eventGeneration, knownGeneration := c.statusPollEventGeneration[generation] + delete(c.statusPollEventGeneration, generation) + + if generation <= c.lastStatusPollGeneration { + return false + } + + c.lastStatusPollGeneration = generation + for olderGeneration := range c.statusPollEventGeneration { + if olderGeneration < generation { + delete(c.statusPollEventGeneration, olderGeneration) + } + } + + if !knownGeneration || eventGeneration != c.speakerEventGeneration { + return false + } + + c.UpdateStatus(mut) + + return true +} + // UpdateStatus atomically applies mut to a copy of the current status // and stores the result. If another goroutine updates the status while // mut runs, UpdateStatus retries with the newer status — so concurrent