mirror of
https://github.com/gesellix/Bose-SoundTouch.git
synced 2026-09-07 15:07:17 +00:00
fix(player): harden device state projection
This commit is contained in:
Lukáš Lipinský
committed by
Tobias Gesellchen
Tobias Gesellchen
parent
4d6bf9e731
commit
60788bfa90
+240
-73
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/xml"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -16,17 +17,33 @@ import (
|
||||
|
||||
// WebSocketClient handles WebSocket connections to SoundTouch devices
|
||||
type WebSocketClient struct {
|
||||
client *Client
|
||||
conn *websocket.Conn
|
||||
handlers *models.WebSocketEventHandlers
|
||||
mu sync.RWMutex
|
||||
writeMu sync.Mutex // serializes all writes; gorilla/websocket allows one concurrent writer
|
||||
connected bool
|
||||
reconnect bool
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
logger Logger
|
||||
bufferSize int
|
||||
client *Client
|
||||
conn *websocket.Conn
|
||||
connection *webSocketConnection
|
||||
handlers *models.WebSocketEventHandlers
|
||||
mu sync.RWMutex
|
||||
connectMu sync.Mutex // serializes dial attempts without blocking shutdown
|
||||
writeMu sync.Mutex // serializes all writes; gorilla/websocket allows one concurrent writer
|
||||
connected bool
|
||||
reconnect bool
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
logger Logger
|
||||
bufferSize int
|
||||
dialContext webSocketDialContext
|
||||
|
||||
transportHandler func(connected bool, generation uint64)
|
||||
transportGeneration uint64
|
||||
}
|
||||
|
||||
type webSocketDialContext func(context.Context, string, http.Header) (*websocket.Conn, *http.Response, error)
|
||||
|
||||
// webSocketConnection gives each transport generation its own lifecycle so an
|
||||
// old read or ping loop cannot start using a replacement connection.
|
||||
type webSocketConnection struct {
|
||||
conn *websocket.Conn
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
// Logger interface for WebSocket logging
|
||||
@@ -159,6 +176,23 @@ func (ws *WebSocketClient) OnBassUpdated(handler models.TypedEventHandler[*model
|
||||
ws.handlers.OnBassUpdated = handler
|
||||
}
|
||||
|
||||
// OnNameUpdated sets a handler for device name update events.
|
||||
func (ws *WebSocketClient) OnNameUpdated(handler models.TypedEventHandler[*models.NameUpdatedEvent]) {
|
||||
ws.mu.Lock()
|
||||
defer ws.mu.Unlock()
|
||||
|
||||
ws.handlers.OnNameUpdated = handler
|
||||
}
|
||||
|
||||
// OnTransportState observes authoritative connection transitions. Generation
|
||||
// numbers let consumers reject callbacks that arrive out of order.
|
||||
func (ws *WebSocketClient) OnTransportState(handler func(connected bool, generation uint64)) {
|
||||
ws.mu.Lock()
|
||||
defer ws.mu.Unlock()
|
||||
|
||||
ws.transportHandler = handler
|
||||
}
|
||||
|
||||
// OnUnknownEvent sets a handler for unknown events
|
||||
func (ws *WebSocketClient) OnUnknownEvent(handler models.EventHandler) {
|
||||
ws.mu.Lock()
|
||||
@@ -198,12 +232,21 @@ func (ws *WebSocketClient) ConnectWithConfig(config *WebSocketConfig) error {
|
||||
}
|
||||
|
||||
func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error {
|
||||
ws.mu.Lock()
|
||||
defer ws.mu.Unlock()
|
||||
ws.connectMu.Lock()
|
||||
defer ws.connectMu.Unlock()
|
||||
|
||||
ws.mu.RLock()
|
||||
if ws.connected {
|
||||
ws.mu.RUnlock()
|
||||
return fmt.Errorf("already connected")
|
||||
}
|
||||
ctx := ws.ctx
|
||||
dialContext := ws.dialContext
|
||||
ws.mu.RUnlock()
|
||||
|
||||
if err := ctx.Err(); err != nil {
|
||||
return fmt.Errorf("WebSocket client is closed: %w", err)
|
||||
}
|
||||
|
||||
// Build WebSocket URL
|
||||
// Parse the base URL to extract just the hostname
|
||||
@@ -220,16 +263,20 @@ func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error {
|
||||
|
||||
ws.logger.Printf("Connecting to %s", sanitizeLog(wsURL.String()))
|
||||
|
||||
// Create dialer with custom buffer sizes and "gabbo" protocol
|
||||
dialer := websocket.Dialer{
|
||||
HandshakeTimeout: 10 * time.Second,
|
||||
ReadBufferSize: config.ReadBufferSize,
|
||||
WriteBufferSize: config.WriteBufferSize,
|
||||
Subprotocols: []string{"gabbo"}, // Required by SoundTouch API
|
||||
if dialContext == nil {
|
||||
// Create dialer with custom buffer sizes and "gabbo" protocol.
|
||||
dialer := websocket.Dialer{
|
||||
HandshakeTimeout: 10 * time.Second,
|
||||
ReadBufferSize: config.ReadBufferSize,
|
||||
WriteBufferSize: config.WriteBufferSize,
|
||||
Subprotocols: []string{"gabbo"}, // Required by SoundTouch API
|
||||
}
|
||||
dialContext = dialer.DialContext
|
||||
}
|
||||
|
||||
// Establish connection
|
||||
conn, resp, err := dialer.DialContext(ws.ctx, wsURL.String(), nil)
|
||||
// Dial without holding the state mutex so shutdown can cancel the context
|
||||
// immediately instead of waiting for the handshake timeout.
|
||||
conn, resp, err := dialContext(ctx, wsURL.String(), nil)
|
||||
if resp != nil && resp.Body != nil {
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
}
|
||||
@@ -238,8 +285,50 @@ func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error {
|
||||
return fmt.Errorf("failed to connect to WebSocket: %w", err)
|
||||
}
|
||||
|
||||
ws.mu.Lock()
|
||||
if err := ctx.Err(); err != nil {
|
||||
ws.mu.Unlock()
|
||||
_ = conn.Close()
|
||||
|
||||
return fmt.Errorf("WebSocket client closed during connect: %w", err)
|
||||
}
|
||||
if ws.connected {
|
||||
ws.mu.Unlock()
|
||||
_ = conn.Close()
|
||||
|
||||
return fmt.Errorf("already connected")
|
||||
}
|
||||
connection, transportHandler, transportGeneration := ws.activateConnectionLocked(conn)
|
||||
ws.mu.Unlock()
|
||||
|
||||
notifyTransportState(transportHandler, true, transportGeneration)
|
||||
go ws.readLoop(config, connection)
|
||||
go ws.pingLoop(config, connection)
|
||||
|
||||
ws.logger.Printf("Connected to %s", sanitizeLog(wsURL.String()))
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ws *WebSocketClient) activateConnectionLocked(
|
||||
conn *websocket.Conn,
|
||||
) (*webSocketConnection, func(bool, uint64), uint64) {
|
||||
if ws.connection != nil {
|
||||
ws.connection.cancel()
|
||||
_ = ws.connection.conn.Close()
|
||||
}
|
||||
|
||||
connectionCtx, connectionCancel := context.WithCancel(ws.ctx)
|
||||
connection := &webSocketConnection{
|
||||
conn: conn,
|
||||
ctx: connectionCtx,
|
||||
cancel: connectionCancel,
|
||||
}
|
||||
|
||||
ws.conn = conn
|
||||
ws.connection = connection
|
||||
ws.connected = true
|
||||
ws.transportGeneration++
|
||||
|
||||
// Extend the read deadline on every pong so the connection survives
|
||||
// quiet periods between speaker events. Without this, the 60-second
|
||||
@@ -250,41 +339,90 @@ func (ws *WebSocketClient) connectWithConfig(config *WebSocketConfig) error {
|
||||
return nil
|
||||
})
|
||||
|
||||
// Start background goroutines for connection management
|
||||
go ws.readLoop(config)
|
||||
go ws.pingLoop(config)
|
||||
|
||||
ws.logger.Printf("Connected to %s", sanitizeLog(wsURL.String()))
|
||||
|
||||
return nil
|
||||
return connection, ws.transportHandler, ws.transportGeneration
|
||||
}
|
||||
|
||||
// Disconnect closes the WebSocket connection
|
||||
func (ws *WebSocketClient) Disconnect() error {
|
||||
// Cancel first so an in-progress DialContext wakes without waiting for mu.
|
||||
ws.cancel()
|
||||
|
||||
ws.mu.Lock()
|
||||
defer ws.mu.Unlock()
|
||||
|
||||
if !ws.connected {
|
||||
return fmt.Errorf("not connected")
|
||||
}
|
||||
|
||||
wasConnected := ws.connected
|
||||
ws.reconnect = false
|
||||
ws.cancel() // Cancel context to stop goroutines
|
||||
if ws.connection != nil {
|
||||
ws.connection.cancel()
|
||||
ws.connection = nil
|
||||
}
|
||||
conn := ws.conn
|
||||
ws.conn = nil
|
||||
ws.connected = false
|
||||
var (
|
||||
transportHandler func(bool, uint64)
|
||||
transportGeneration uint64
|
||||
)
|
||||
if wasConnected {
|
||||
ws.transportGeneration++
|
||||
transportHandler = ws.transportHandler
|
||||
transportGeneration = ws.transportGeneration
|
||||
}
|
||||
ws.mu.Unlock()
|
||||
|
||||
if ws.conn != nil {
|
||||
err := ws.conn.Close()
|
||||
ws.conn = nil
|
||||
ws.connected = false
|
||||
if conn != nil {
|
||||
err := conn.Close()
|
||||
ws.logger.Printf("Disconnected")
|
||||
|
||||
notifyTransportState(transportHandler, false, transportGeneration)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
ws.connected = false
|
||||
notifyTransportState(transportHandler, false, transportGeneration)
|
||||
if !wasConnected {
|
||||
return fmt.Errorf("not connected")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close permanently stops this client and is idempotent. Device registries use
|
||||
// it when removal races an initial dial or an automatic reconnect.
|
||||
func (ws *WebSocketClient) Close() error {
|
||||
ws.cancel()
|
||||
|
||||
ws.mu.Lock()
|
||||
wasConnected := ws.connected
|
||||
ws.reconnect = false
|
||||
if ws.connection != nil {
|
||||
ws.connection.cancel()
|
||||
ws.connection = nil
|
||||
}
|
||||
conn := ws.conn
|
||||
ws.conn = nil
|
||||
ws.connected = false
|
||||
var (
|
||||
transportHandler func(bool, uint64)
|
||||
transportGeneration uint64
|
||||
)
|
||||
if wasConnected {
|
||||
ws.transportGeneration++
|
||||
transportHandler = ws.transportHandler
|
||||
transportGeneration = ws.transportGeneration
|
||||
}
|
||||
ws.mu.Unlock()
|
||||
|
||||
if conn == nil {
|
||||
notifyTransportState(transportHandler, false, transportGeneration)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
err := conn.Close()
|
||||
ws.logger.Printf("Disconnected")
|
||||
notifyTransportState(transportHandler, false, transportGeneration)
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// IsConnected returns true if the WebSocket is connected
|
||||
func (ws *WebSocketClient) IsConnected() bool {
|
||||
ws.mu.RLock()
|
||||
@@ -293,45 +431,61 @@ func (ws *WebSocketClient) IsConnected() bool {
|
||||
return ws.connected
|
||||
}
|
||||
|
||||
func (ws *WebSocketClient) shouldReconnect() bool {
|
||||
ws.mu.RLock()
|
||||
defer ws.mu.RUnlock()
|
||||
|
||||
return ws.reconnect
|
||||
}
|
||||
|
||||
func notifyTransportState(handler func(bool, uint64), connected bool, generation uint64) {
|
||||
if handler != nil {
|
||||
handler(connected, generation)
|
||||
}
|
||||
}
|
||||
|
||||
// readLoop continuously reads messages from the WebSocket connection
|
||||
func (ws *WebSocketClient) readLoop(config *WebSocketConfig) {
|
||||
func (ws *WebSocketClient) readLoop(config *WebSocketConfig, connection *webSocketConnection) {
|
||||
defer func() {
|
||||
ws.mu.Lock()
|
||||
if ws.connection != connection {
|
||||
ws.mu.Unlock()
|
||||
connection.cancel()
|
||||
_ = connection.conn.Close()
|
||||
|
||||
ws.connected = false
|
||||
if ws.conn != nil {
|
||||
_ = ws.conn.Close()
|
||||
ws.conn = nil
|
||||
return
|
||||
}
|
||||
|
||||
connection.cancel()
|
||||
ws.connection = nil
|
||||
ws.conn = nil
|
||||
ws.connected = false
|
||||
ws.transportGeneration++
|
||||
transportHandler := ws.transportHandler
|
||||
transportGeneration := ws.transportGeneration
|
||||
reconnect := ws.reconnect
|
||||
ws.mu.Unlock()
|
||||
_ = connection.conn.Close()
|
||||
notifyTransportState(transportHandler, false, transportGeneration)
|
||||
|
||||
// Attempt reconnection if enabled
|
||||
if ws.reconnect {
|
||||
if reconnect {
|
||||
go ws.attemptReconnect(config)
|
||||
}
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ws.ctx.Done():
|
||||
case <-connection.ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
ws.mu.RLock()
|
||||
conn := ws.conn
|
||||
ws.mu.RUnlock()
|
||||
|
||||
if conn == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Set read deadline
|
||||
_ = conn.SetReadDeadline(time.Now().Add(60 * time.Second))
|
||||
_ = connection.conn.SetReadDeadline(time.Now().Add(60 * time.Second))
|
||||
|
||||
// Read message
|
||||
messageType, data, err := conn.ReadMessage()
|
||||
messageType, data, err := connection.conn.ReadMessage()
|
||||
if err != nil {
|
||||
if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
|
||||
ws.logger.Printf("WebSocket read error: %v", err)
|
||||
@@ -351,30 +505,20 @@ func (ws *WebSocketClient) readLoop(config *WebSocketConfig) {
|
||||
}
|
||||
|
||||
// pingLoop sends periodic ping messages to keep the connection alive
|
||||
func (ws *WebSocketClient) pingLoop(config *WebSocketConfig) {
|
||||
func (ws *WebSocketClient) pingLoop(config *WebSocketConfig, connection *webSocketConnection) {
|
||||
ticker := time.NewTicker(config.PingInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ws.ctx.Done():
|
||||
case <-connection.ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
ws.mu.RLock()
|
||||
conn := ws.conn
|
||||
connected := ws.connected
|
||||
ws.mu.RUnlock()
|
||||
|
||||
if !connected || conn == nil {
|
||||
active, err := ws.writePing(connection)
|
||||
if !active {
|
||||
return
|
||||
}
|
||||
|
||||
// Set write deadline for ping
|
||||
ws.writeMu.Lock()
|
||||
_ = conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
|
||||
err := conn.WriteMessage(websocket.PingMessage, nil)
|
||||
ws.writeMu.Unlock()
|
||||
|
||||
if err != nil {
|
||||
ws.logger.Printf("Failed to send ping: %v", err)
|
||||
return
|
||||
@@ -383,10 +527,26 @@ func (ws *WebSocketClient) pingLoop(config *WebSocketConfig) {
|
||||
}
|
||||
}
|
||||
|
||||
func (ws *WebSocketClient) writePing(connection *webSocketConnection) (bool, error) {
|
||||
ws.writeMu.Lock()
|
||||
defer ws.writeMu.Unlock()
|
||||
|
||||
ws.mu.RLock()
|
||||
defer ws.mu.RUnlock()
|
||||
|
||||
if connection.ctx.Err() != nil || ws.connection != connection || !ws.connected {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
_ = connection.conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
|
||||
|
||||
return true, connection.conn.WriteMessage(websocket.PingMessage, nil)
|
||||
}
|
||||
|
||||
// attemptReconnect attempts to reconnect to the WebSocket
|
||||
func (ws *WebSocketClient) attemptReconnect(config *WebSocketConfig) {
|
||||
attempt := 0
|
||||
for ws.reconnect && (config.MaxReconnectAttempts == 0 || attempt < config.MaxReconnectAttempts) {
|
||||
for ws.shouldReconnect() && (config.MaxReconnectAttempts == 0 || attempt < config.MaxReconnectAttempts) {
|
||||
select {
|
||||
case <-ws.ctx.Done():
|
||||
return
|
||||
@@ -533,6 +693,13 @@ func (ws *WebSocketClient) dispatchTypedEventContinued(handlers *models.WebSocke
|
||||
|
||||
return true
|
||||
|
||||
case models.EventTypeNameUpdated:
|
||||
if handlers.OnNameUpdated != nil && event.NameUpdated != nil {
|
||||
handlers.OnNameUpdated(event.NameUpdated)
|
||||
}
|
||||
|
||||
return true
|
||||
|
||||
case models.EventTypeRecentsUpdated:
|
||||
return true
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
@@ -269,6 +270,139 @@ func TestWebSocketClient_Disconnect(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebSocketClient_DisconnectWhileDisconnectedStopsReconnect(t *testing.T) {
|
||||
client := NewClientFromHost("192.0.2.10")
|
||||
wsClient := client.NewWebSocketClient(nil)
|
||||
|
||||
if err := wsClient.Disconnect(); err == nil {
|
||||
t.Fatal("Disconnect() while disconnected should retain its compatibility error")
|
||||
}
|
||||
|
||||
wsClient.mu.RLock()
|
||||
reconnect := wsClient.reconnect
|
||||
wsClient.mu.RUnlock()
|
||||
if reconnect {
|
||||
t.Fatal("Disconnect() left reconnect enabled")
|
||||
}
|
||||
|
||||
select {
|
||||
case <-wsClient.ctx.Done():
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Fatal("Disconnect() did not cancel the WebSocket context")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebSocketClient_CloseCancelsInProgressDial(t *testing.T) {
|
||||
client := NewClientFromHost("192.0.2.10")
|
||||
wsClient := client.NewWebSocketClient(nil)
|
||||
dialStarted := make(chan struct{})
|
||||
|
||||
wsClient.dialContext = func(ctx context.Context, _ string, _ http.Header) (*websocket.Conn, *http.Response, error) {
|
||||
close(dialStarted)
|
||||
<-ctx.Done()
|
||||
|
||||
return nil, nil, ctx.Err()
|
||||
}
|
||||
|
||||
connectDone := make(chan error, 1)
|
||||
go func() {
|
||||
connectDone <- wsClient.Connect()
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-dialStarted:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("WebSocket dial did not start")
|
||||
}
|
||||
|
||||
if err := wsClient.Close(); err != nil {
|
||||
t.Fatalf("Close() failed: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case err := <-connectDone:
|
||||
if err == nil {
|
||||
t.Fatal("Connect() succeeded after Close() canceled its dial")
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("canceled WebSocket dial did not return")
|
||||
}
|
||||
|
||||
if err := wsClient.Close(); err != nil {
|
||||
t.Fatalf("second Close() was not idempotent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebSocketClient_StaleGenerationCannotOwnPing(t *testing.T) {
|
||||
client := NewClientFromHost("192.0.2.10")
|
||||
wsClient := client.NewWebSocketClient(nil)
|
||||
oldCtx, oldCancel := context.WithCancel(context.Background())
|
||||
currentCtx, currentCancel := context.WithCancel(context.Background())
|
||||
defer oldCancel()
|
||||
defer currentCancel()
|
||||
|
||||
oldConnection := &webSocketConnection{ctx: oldCtx, cancel: oldCancel}
|
||||
currentConnection := &webSocketConnection{ctx: currentCtx, cancel: currentCancel}
|
||||
wsClient.mu.Lock()
|
||||
wsClient.connection = currentConnection
|
||||
wsClient.connected = true
|
||||
wsClient.mu.Unlock()
|
||||
|
||||
active, err := wsClient.writePing(oldConnection)
|
||||
if err != nil {
|
||||
t.Fatalf("writePing() for stale generation returned error: %v", err)
|
||||
}
|
||||
if active {
|
||||
t.Fatal("replaced connection generation retained ping ownership")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebSocketClient_TransportCallbacksCarryMonotonicGenerations(t *testing.T) {
|
||||
server, messagesChan := setupMockWebSocketServer(t)
|
||||
defer server.Close()
|
||||
defer close(messagesChan)
|
||||
|
||||
client := NewClientFromHost("192.0.2.10")
|
||||
wsClient := client.NewWebSocketClient(nil)
|
||||
serverWebSocketURL := strings.Replace(server.URL, "http://", "ws://", 1)
|
||||
wsClient.dialContext = func(ctx context.Context, _ string, header http.Header) (*websocket.Conn, *http.Response, error) {
|
||||
return websocket.DefaultDialer.DialContext(ctx, serverWebSocketURL, header)
|
||||
}
|
||||
|
||||
type transportState struct {
|
||||
connected bool
|
||||
generation uint64
|
||||
}
|
||||
states := make(chan transportState, 2)
|
||||
wsClient.OnTransportState(func(connected bool, generation uint64) {
|
||||
states <- transportState{connected: connected, generation: generation}
|
||||
})
|
||||
nextState := func() transportState {
|
||||
t.Helper()
|
||||
select {
|
||||
case state := <-states:
|
||||
return state
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("timed out waiting for transport callback")
|
||||
return transportState{}
|
||||
}
|
||||
}
|
||||
|
||||
if err := wsClient.Connect(); err != nil {
|
||||
t.Fatalf("Connect() failed: %v", err)
|
||||
}
|
||||
if state := nextState(); !state.connected || state.generation != 1 {
|
||||
t.Fatalf("connected state = %+v, want connected generation 1", state)
|
||||
}
|
||||
|
||||
if err := wsClient.Close(); err != nil {
|
||||
t.Fatalf("Close() failed: %v", err)
|
||||
}
|
||||
if state := nextState(); state.connected || state.generation != 2 {
|
||||
t.Fatalf("disconnected state = %+v, want disconnected generation 2", state)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebSocketClient_HandleMessage(t *testing.T) {
|
||||
client := NewClientFromHost("192.0.2.10")
|
||||
wsClient := client.NewWebSocketClient(&WebSocketConfig{
|
||||
@@ -278,6 +412,7 @@ func TestWebSocketClient_HandleMessage(t *testing.T) {
|
||||
var (
|
||||
nowPlayingEvent *models.NowPlayingUpdatedEvent
|
||||
volumeEvent *models.VolumeUpdatedEvent
|
||||
nameEvent *models.NameUpdatedEvent
|
||||
)
|
||||
|
||||
wsClient.OnNowPlaying(func(event *models.NowPlayingUpdatedEvent) {
|
||||
@@ -288,6 +423,10 @@ func TestWebSocketClient_HandleMessage(t *testing.T) {
|
||||
volumeEvent = event
|
||||
})
|
||||
|
||||
wsClient.OnNameUpdated(func(event *models.NameUpdatedEvent) {
|
||||
nameEvent = event
|
||||
})
|
||||
|
||||
t.Run("HandleNowPlayingEvent", func(t *testing.T) {
|
||||
xmlData := []byte(`<?xml version="1.0" encoding="UTF-8" ?>
|
||||
<updates deviceID="689E19B8BB8A">
|
||||
@@ -343,6 +482,20 @@ func TestWebSocketClient_HandleMessage(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("HandleNameEvent", func(t *testing.T) {
|
||||
xmlData := []byte(`<updates deviceID="689E19B8BB8A"><nameUpdated deviceID="689E19B8BB8A"><name>Living Room Left</name></nameUpdated></updates>`)
|
||||
|
||||
wsClient.handleMessage(xmlData)
|
||||
|
||||
if nameEvent == nil {
|
||||
t.Fatal("Name event handler was not called")
|
||||
}
|
||||
|
||||
if nameEvent.DeviceID != "689E19B8BB8A" || nameEvent.Name.Value != "Living Room Left" {
|
||||
t.Errorf("Unexpected name event: %+v", nameEvent)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("HandleInvalidXML", func(t *testing.T) {
|
||||
logger := &mockLogger{}
|
||||
wsClient.logger = logger
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package soundtouchweb
|
||||
|
||||
import (
|
||||
"net"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -99,7 +100,7 @@ func captureDeviceProjectionEntries(snapshot []DeviceEntry) []deviceProjectionEn
|
||||
|
||||
captured = append(captured, deviceProjectionEntry{
|
||||
ID: entry.ID,
|
||||
Info: entry.Device.DeviceInfo,
|
||||
Info: entry.Device.Info(),
|
||||
Status: entry.Device.Status(),
|
||||
LastSeen: entry.LastSeen,
|
||||
})
|
||||
@@ -157,7 +158,7 @@ func projectCapturedDeviceEntries(snapshot []deviceProjectionEntry) map[string]d
|
||||
|
||||
pair := masters[entry.ID]
|
||||
devices[entry.ID] = deviceView{
|
||||
Info: projectedDeviceInfo(entry.Info, pair),
|
||||
Info: projectedDeviceInfo(entry.ID, entry.Info, pair),
|
||||
Status: entry.Status,
|
||||
LastSeen: entry.LastSeen,
|
||||
StereoPair: pair,
|
||||
@@ -236,12 +237,10 @@ func newStereoPairView(group *models.Group, byDeviceID map[string][]deviceProjec
|
||||
if entry, ok := uniqueDeviceEntry(byDeviceID, role.DeviceID); ok {
|
||||
if entry.Info != nil {
|
||||
member.Name = entry.Info.Name
|
||||
if entry.Info.IPAddress != "" {
|
||||
member.IPAddress = entry.Info.IPAddress
|
||||
}
|
||||
member.IPAddress = projectedIPAddress(entry.ID, entry.Info, member.IPAddress)
|
||||
}
|
||||
|
||||
member.Available = entry.Status != nil && entry.Status.IsConnected
|
||||
member.Available = projectedConnectivity(entry.Status) != webtypes.ConnectivityOffline
|
||||
if member.Available {
|
||||
available++
|
||||
}
|
||||
@@ -262,17 +261,56 @@ func newStereoPairView(group *models.Group, byDeviceID map[string][]deviceProjec
|
||||
}
|
||||
}
|
||||
|
||||
func projectedDeviceInfo(info *models.DeviceInfo, pair *stereoPairView) *models.DeviceInfo {
|
||||
if info == nil || pair == nil || pair.Name == "" || pair.Name == info.Name {
|
||||
func projectedDeviceInfo(controlID string, info *models.DeviceInfo, pair *stereoPairView) *models.DeviceInfo {
|
||||
if info == nil {
|
||||
return info
|
||||
}
|
||||
|
||||
address := projectedIPAddress(controlID, info, "")
|
||||
name := info.Name
|
||||
if pair != nil && pair.Name != "" {
|
||||
name = pair.Name
|
||||
}
|
||||
if address == info.IPAddress && name == info.Name {
|
||||
return info
|
||||
}
|
||||
|
||||
projected := *info
|
||||
projected.Name = pair.Name
|
||||
projected.IPAddress = address
|
||||
projected.Name = name
|
||||
|
||||
return &projected
|
||||
}
|
||||
|
||||
func projectedIPAddress(controlID string, info *models.DeviceInfo, fallback string) string {
|
||||
candidates := []string{controlID, fallback}
|
||||
if info != nil {
|
||||
candidates = append([]string{info.IPAddress}, candidates...)
|
||||
}
|
||||
|
||||
for _, candidate := range candidates {
|
||||
if ip := net.ParseIP(strings.TrimSpace(candidate)); ip != nil {
|
||||
return ip.String()
|
||||
}
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
func projectedConnectivity(status *webtypes.DeviceStatus) webtypes.Connectivity {
|
||||
if status == nil {
|
||||
return webtypes.ConnectivityOffline
|
||||
}
|
||||
if status.Connectivity != "" {
|
||||
return status.Connectivity
|
||||
}
|
||||
if status.IsConnected {
|
||||
return webtypes.ConnectivityOnline
|
||||
}
|
||||
|
||||
return webtypes.ConnectivityOffline
|
||||
}
|
||||
|
||||
func logicalPairName(groupName string, members []stereoPairMemberView) string {
|
||||
commonName := ""
|
||||
|
||||
|
||||
@@ -14,14 +14,18 @@ import (
|
||||
)
|
||||
|
||||
func projectionDevice(host, deviceID, name string, connected bool, group *models.Group) DeviceEntry {
|
||||
return projectionDeviceAt(host, host, deviceID, name, connected, group)
|
||||
}
|
||||
|
||||
func projectionDeviceAt(controlID, address, deviceID, name string, connected bool, group *models.Group) DeviceEntry {
|
||||
conn := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{
|
||||
DeviceID: deviceID,
|
||||
Name: name,
|
||||
IPAddress: host,
|
||||
IPAddress: address,
|
||||
})
|
||||
conn.SetStatus(&webtypes.DeviceStatus{IsConnected: connected, Group: group})
|
||||
|
||||
return DeviceEntry{ID: host, Device: conn, LastSeen: conn.LastSeen}
|
||||
return DeviceEntry{ID: controlID, Device: conn, LastSeen: conn.LastSeen}
|
||||
}
|
||||
|
||||
func testStereoGroup() *models.Group {
|
||||
@@ -106,6 +110,46 @@ func TestProjectDeviceEntriesNormalizesMemberRoleAndDeviceID(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestProjectDeviceEntriesKeepsHostnameControlIDSeparateFromAddress(t *testing.T) {
|
||||
entry := projectionDeviceAt("kitchen.local", "192.0.2.10", "kitchen-id", "Kitchen", true, nil)
|
||||
got := projectDeviceEntries([]DeviceEntry{entry})
|
||||
|
||||
view, ok := got["kitchen.local"]
|
||||
if !ok || len(got) != 1 {
|
||||
t.Fatalf("hostname-keyed projection = %+v", got)
|
||||
}
|
||||
if view.Info == nil || view.Info.IPAddress != "192.0.2.10" {
|
||||
t.Fatalf("presentation address = %+v, want canonical numeric IP", view.Info)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProjectDeviceEntriesUsesLatestInfoName(t *testing.T) {
|
||||
entry := projectionDevice("192.0.2.10", "kitchen-id", "Old Kitchen", true, nil)
|
||||
entry.Device.ApplyNameEvent("Kitchen")
|
||||
|
||||
view := projectDeviceEntries([]DeviceEntry{entry})["192.0.2.10"]
|
||||
if view.Info == nil || view.Info.Name != "Kitchen" {
|
||||
t.Fatalf("projected info = %+v, want latest event name", view.Info)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProjectDeviceEntriesTreatsStaleStereoMemberAsAvailable(t *testing.T) {
|
||||
group := testStereoGroup()
|
||||
left := projectionDevice("192.0.2.10", "left-id", "Living Room", false, group)
|
||||
left.Device.UpdateStatus(func(status *webtypes.DeviceStatus) {
|
||||
status.Connectivity = webtypes.ConnectivityStale
|
||||
})
|
||||
right := projectionDevice("192.0.2.11", "right-id", "Living Room", true, group)
|
||||
right.Device.UpdateStatus(func(status *webtypes.DeviceStatus) {
|
||||
status.Connectivity = webtypes.ConnectivityOnline
|
||||
})
|
||||
|
||||
pair := projectDeviceEntries([]DeviceEntry{left, right})["192.0.2.10"].StereoPair
|
||||
if pair == nil || pair.AvailableMemberCount != 2 || pair.Degraded {
|
||||
t.Fatalf("stale logical member was treated as offline: %+v", pair)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProjectDeviceEntriesShowsDegradedPairWhenMemberIsMissing(t *testing.T) {
|
||||
got := projectDeviceEntries([]DeviceEntry{
|
||||
projectionDevice("192.0.2.10", "left-id", "Living Room", true, testStereoGroup()),
|
||||
|
||||
@@ -3,6 +3,7 @@ package soundtouchweb
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -10,6 +11,7 @@ import (
|
||||
"github.com/gesellix/bose-soundtouch/pkg/client"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/config"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/discovery"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/models"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/service/soundtouchweb/webtypes"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/speaker"
|
||||
)
|
||||
@@ -85,12 +87,12 @@ func (app *WebApp) addDeviceByHost(host string, port int, source string) *webtyp
|
||||
return nil
|
||||
}
|
||||
|
||||
// Ensure IPAddress is set for the web UI
|
||||
if info.IPAddress == "" {
|
||||
info.IPAddress = host
|
||||
}
|
||||
// Keep the registry key stable for controls, but expose a canonical numeric
|
||||
// address separately for presentation and sorting.
|
||||
info.IPAddress = resolvedDeviceIPAddress(host, info)
|
||||
|
||||
conn := webtypes.NewDeviceConnection(c, info)
|
||||
conn.MarkHTTPSuccess(time.Now())
|
||||
if !app.AddDevice(host, conn) {
|
||||
// Lost a race — another goroutine inserted the same host
|
||||
// between TouchDevice and AddDevice. AddDevice bumped LastSeen
|
||||
@@ -122,6 +124,44 @@ func (app *WebApp) addDeviceByHost(host string, port int, source string) *webtyp
|
||||
return conn
|
||||
}
|
||||
|
||||
func resolvedDeviceIPAddress(host string, info *models.DeviceInfo) string {
|
||||
if info != nil {
|
||||
if address := numericIPAddress(info.IPAddress); address != "" {
|
||||
return address
|
||||
}
|
||||
|
||||
for _, network := range info.NetworkInfo {
|
||||
if address := numericIPAddress(network.IPAddress); address != "" {
|
||||
return address
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
bareHost := hostOnly(host)
|
||||
if address := numericIPAddress(bareHost); address != "" {
|
||||
return address
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
||||
defer cancel()
|
||||
|
||||
addresses, err := net.DefaultResolver.LookupNetIP(ctx, "ip4", bareHost)
|
||||
if err == nil && len(addresses) > 0 {
|
||||
return addresses[0].String()
|
||||
}
|
||||
|
||||
return ""
|
||||
}
|
||||
|
||||
func numericIPAddress(address string) string {
|
||||
ip := net.ParseIP(strings.TrimSpace(address))
|
||||
if ip == nil {
|
||||
return ""
|
||||
}
|
||||
|
||||
return ip.String()
|
||||
}
|
||||
|
||||
// SeedExtraDevices registers any devices reported by the ExtraDeviceHosts hook
|
||||
// (if set) via AddDeviceByHost, and prunes any previously-seeded host that no
|
||||
// longer appears in that set. Idempotent: already-known hosts are skipped.
|
||||
|
||||
@@ -102,6 +102,22 @@ func TestClassifySource(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolvedDeviceIPAddressSeparatesHostnameFromReportedAddress(t *testing.T) {
|
||||
info := &models.DeviceInfo{NetworkInfo: []models.NetworkInfo{
|
||||
{Type: "SCM", IPAddress: "192.0.2.42"},
|
||||
}}
|
||||
|
||||
if got := resolvedDeviceIPAddress("kitchen.local", info); got != "192.0.2.42" {
|
||||
t.Fatalf("resolved address = %q, want reported speaker address", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolvedDeviceIPAddressPreservesLiteralIP(t *testing.T) {
|
||||
if got := resolvedDeviceIPAddress("192.0.2.20", &models.DeviceInfo{}); got != "192.0.2.20" {
|
||||
t.Fatalf("resolved literal address = %q, want unchanged literal", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetryUntilReadyStopsAfterSuccess(t *testing.T) {
|
||||
attempts := 0
|
||||
retryUntilReady(context.Background(), time.Millisecond, func() bool {
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import test from 'node:test';
|
||||
|
||||
import {
|
||||
connectivityLabel,
|
||||
connectivityState,
|
||||
sortDeviceEntries,
|
||||
} from '../static/js/devicePresentation.js';
|
||||
|
||||
test('uses tri-state connectivity before the compatibility flag', () => {
|
||||
assert.equal(connectivityState({ status: { connectivity: 'online', isConnected: false } }), 'online');
|
||||
assert.equal(connectivityState({ status: { connectivity: 'stale', isConnected: true } }), 'stale');
|
||||
assert.equal(connectivityState({ status: { connectivity: 'offline', isConnected: true } }), 'offline');
|
||||
});
|
||||
|
||||
test('falls back to the legacy connected flag and supplies a label', () => {
|
||||
assert.equal(connectivityState({ status: { isConnected: true } }), 'online');
|
||||
assert.equal(connectivityState({ status: { isConnected: false } }), 'offline');
|
||||
assert.equal(connectivityState(undefined), 'offline');
|
||||
assert.equal(connectivityLabel({ status: { connectivity: 'stale' } }), 'Stale');
|
||||
});
|
||||
|
||||
test('sorts hostname control IDs by their numeric presentation IP', () => {
|
||||
const entries = [
|
||||
['speaker-a', { info: { ip_address: '192.0.2.10' } }],
|
||||
['speaker-b', { info: { ip_address: '192.0.2.2' } }],
|
||||
];
|
||||
|
||||
assert.deepEqual(sortDeviceEntries(entries, 'ip').map(([id]) => id), [
|
||||
'speaker-b',
|
||||
'speaker-a',
|
||||
]);
|
||||
});
|
||||
|
||||
test('sorts literal IP keys numerically and falls back when an address is missing', () => {
|
||||
const entries = [
|
||||
['192.0.2.10', { info: {} }],
|
||||
['192.0.2.2', {}],
|
||||
['192.0.2.1', { info: { ip_address: '' } }],
|
||||
];
|
||||
|
||||
assert.deepEqual(sortDeviceEntries(entries, 'ip').map(([id]) => id), [
|
||||
'192.0.2.1',
|
||||
'192.0.2.2',
|
||||
'192.0.2.10',
|
||||
]);
|
||||
});
|
||||
|
||||
test('uses stable control-ID tie breaks for equal names and addresses', () => {
|
||||
const byAddress = [
|
||||
['speaker-b', { info: { ip_address: '192.0.2.4' } }],
|
||||
['speaker-a', { info: { ip_address: '192.0.2.4' } }],
|
||||
];
|
||||
const byName = [
|
||||
['speaker-b', { info: { name: ' Kitchen ' } }],
|
||||
['speaker-a', { info: { name: 'kitchen' } }],
|
||||
];
|
||||
|
||||
assert.deepEqual(sortDeviceEntries(byAddress, 'ip').map(([id]) => id), [
|
||||
'speaker-a',
|
||||
'speaker-b',
|
||||
]);
|
||||
assert.deepEqual(sortDeviceEntries(byName, 'name').map(([id]) => id), [
|
||||
'speaker-a',
|
||||
'speaker-b',
|
||||
]);
|
||||
});
|
||||
@@ -0,0 +1,94 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import test from 'node:test';
|
||||
|
||||
import { removeDeviceAndRefresh } from '../static/js/deviceRemoval.js';
|
||||
|
||||
function operation(overrides = {}) {
|
||||
const calls = [];
|
||||
return {
|
||||
calls,
|
||||
options: {
|
||||
id: '192.0.2.10',
|
||||
name: 'Kitchen',
|
||||
remove: async (id) => {
|
||||
calls.push(['remove', id]);
|
||||
return { success: true };
|
||||
},
|
||||
refresh: async () => calls.push(['refresh']),
|
||||
showDeviceList: () => calls.push(['showDeviceList']),
|
||||
notify: (message) => calls.push(['notify', message]),
|
||||
...overrides,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
test('waits for successful removal before navigating and refreshing', async () => {
|
||||
let resolveRemoval;
|
||||
const pendingRemoval = new Promise(resolve => { resolveRemoval = resolve; });
|
||||
const { calls, options } = operation({
|
||||
remove: (id) => {
|
||||
calls.push(['remove', id]);
|
||||
return pendingRemoval;
|
||||
},
|
||||
});
|
||||
|
||||
const result = removeDeviceAndRefresh(options);
|
||||
await Promise.resolve();
|
||||
assert.deepEqual(calls, [['remove', '192.0.2.10']]);
|
||||
|
||||
resolveRemoval({ success: true });
|
||||
assert.equal(await result, true);
|
||||
assert.deepEqual(calls, [
|
||||
['remove', '192.0.2.10'],
|
||||
['showDeviceList'],
|
||||
['refresh'],
|
||||
['notify', 'Removed "Kitchen"'],
|
||||
]);
|
||||
});
|
||||
|
||||
test('keeps the current view and state when removal is rejected', async () => {
|
||||
const { calls, options } = operation({
|
||||
remove: async (id) => {
|
||||
calls.push(['remove', id]);
|
||||
return { success: false, error: 'Device is busy' };
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(await removeDeviceAndRefresh(options), false);
|
||||
assert.deepEqual(calls, [
|
||||
['remove', '192.0.2.10'],
|
||||
['notify', 'Device is busy'],
|
||||
]);
|
||||
});
|
||||
|
||||
test('reports transport failures without navigating or refreshing', async () => {
|
||||
const { calls, options } = operation({
|
||||
remove: async (id) => {
|
||||
calls.push(['remove', id]);
|
||||
throw new Error('network failure');
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(await removeDeviceAndRefresh(options), false);
|
||||
assert.deepEqual(calls, [
|
||||
['remove', '192.0.2.10'],
|
||||
['notify', 'Failed to remove device'],
|
||||
]);
|
||||
});
|
||||
|
||||
test('retains successful removal when the authoritative refresh fails', async () => {
|
||||
const { calls, options } = operation({
|
||||
refresh: async () => {
|
||||
calls.push(['refresh']);
|
||||
throw new Error('refresh failure');
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(await removeDeviceAndRefresh(options), true);
|
||||
assert.deepEqual(calls, [
|
||||
['remove', '192.0.2.10'],
|
||||
['showDeviceList'],
|
||||
['refresh'],
|
||||
['notify', 'Removed "Kitchen", but failed to refresh devices'],
|
||||
]);
|
||||
});
|
||||
@@ -398,7 +398,7 @@ func (app *WebApp) HandleAPIDevice(w http.ResponseWriter, r *http.Request) {
|
||||
app.UpdateDeviceStatus(deviceID, device)
|
||||
|
||||
// Connect WebSocket for real-time updates if not already connected
|
||||
if device.WebSocket == nil {
|
||||
if device.CurrentWebSocket() == nil {
|
||||
go app.ConnectDeviceWebSocket(deviceID, device)
|
||||
}
|
||||
|
||||
@@ -448,8 +448,8 @@ func (app *WebApp) HandleDeleteDevice(w http.ResponseWriter, r *http.Request) {
|
||||
// only prunes the in-memory registry below.
|
||||
if app.RemoveDeviceHook != nil {
|
||||
deviceID := ""
|
||||
if conn.DeviceInfo != nil {
|
||||
deviceID = conn.DeviceInfo.DeviceID
|
||||
if info := conn.Info(); info != nil {
|
||||
deviceID = info.DeviceID
|
||||
}
|
||||
|
||||
if err := app.RemoveDeviceHook(deviceID); err != nil {
|
||||
@@ -460,7 +460,11 @@ func (app *WebApp) HandleDeleteDevice(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
app.RemoveDevice(host)
|
||||
if !app.removeDeviceIfMatch(host, conn) {
|
||||
app.sendError(w, "Device changed during removal", http.StatusConflict)
|
||||
|
||||
return
|
||||
}
|
||||
app.BroadcastDeviceList()
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
@@ -487,7 +491,7 @@ func (app *WebApp) HandleAPIControl(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// Connect WebSocket for real-time updates if not already connected
|
||||
if device.WebSocket == nil {
|
||||
if device.CurrentWebSocket() == nil {
|
||||
go app.ConnectDeviceWebSocket(deviceID, device)
|
||||
}
|
||||
|
||||
@@ -730,7 +734,7 @@ func (app *WebApp) HandleDeviceKey(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// Connect WebSocket for real-time updates if not already connected
|
||||
if device.WebSocket == nil {
|
||||
if device.CurrentWebSocket() == nil {
|
||||
go app.ConnectDeviceWebSocket(deviceID, device)
|
||||
}
|
||||
|
||||
@@ -762,7 +766,7 @@ func (app *WebApp) HandleDirectVolumeControl(w http.ResponseWriter, r *http.Requ
|
||||
}
|
||||
|
||||
// Connect WebSocket for real-time updates if not already connected
|
||||
if device.WebSocket == nil {
|
||||
if device.CurrentWebSocket() == nil {
|
||||
go app.ConnectDeviceWebSocket(deviceID, device)
|
||||
}
|
||||
|
||||
@@ -788,7 +792,7 @@ func (app *WebApp) HandleDevicePower(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
|
||||
// Connect WebSocket for real-time updates if not already connected
|
||||
if device.WebSocket == nil {
|
||||
if device.CurrentWebSocket() == nil {
|
||||
go app.ConnectDeviceWebSocket(deviceID, device)
|
||||
}
|
||||
|
||||
@@ -1014,7 +1018,7 @@ func (app *WebApp) HandleTuneInNavigate(w http.ResponseWriter, r *http.Request)
|
||||
// Returns "" when no match is found.
|
||||
func (app *WebApp) findIPByHwID(hwID string) string {
|
||||
for _, entry := range app.DeviceSnapshot() {
|
||||
if entry.Device.DeviceInfo != nil && entry.Device.DeviceInfo.DeviceID == hwID {
|
||||
if info := entry.Device.Info(); info != nil && info.DeviceID == hwID {
|
||||
return entry.ID
|
||||
}
|
||||
}
|
||||
@@ -1053,8 +1057,10 @@ func (app *WebApp) HandleGetZone(w http.ResponseWriter, r *http.Request) {
|
||||
masterIP := app.findIPByHwID(zone.Master)
|
||||
|
||||
masterName := ""
|
||||
if conn, ok := app.GetDevice(masterIP); ok && conn.DeviceInfo != nil {
|
||||
masterName = conn.DeviceInfo.Name
|
||||
if conn, ok := app.GetDevice(masterIP); ok {
|
||||
if info := conn.Info(); info != nil {
|
||||
masterName = info.Name
|
||||
}
|
||||
}
|
||||
|
||||
type memberInfo struct {
|
||||
@@ -1067,8 +1073,10 @@ func (app *WebApp) HandleGetZone(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
for _, m := range zone.Members {
|
||||
name := ""
|
||||
if conn, ok := app.GetDevice(m.IP); ok && conn.DeviceInfo != nil {
|
||||
name = conn.DeviceInfo.Name
|
||||
if conn, ok := app.GetDevice(m.IP); ok {
|
||||
if info := conn.Info(); info != nil {
|
||||
name = info.Name
|
||||
}
|
||||
}
|
||||
|
||||
members = append(members, memberInfo{IP: m.IP, HwID: m.DeviceID, Name: name})
|
||||
|
||||
@@ -117,6 +117,39 @@ func TestHandleAPIDevices(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleDeleteDevicePreservesConcurrentReplacement(t *testing.T) {
|
||||
app := NewWebApp()
|
||||
host := "speaker.local"
|
||||
original := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{DeviceID: "ORIGINAL"})
|
||||
replacement := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{DeviceID: "REPLACEMENT"})
|
||||
app.AddDevice(host, original)
|
||||
app.RemoveDeviceHook = func(deviceID string) error {
|
||||
if deviceID != "ORIGINAL" {
|
||||
t.Fatalf("RemoveDeviceHook deviceID = %q, want ORIGINAL", deviceID)
|
||||
}
|
||||
app.RemoveDevice(host)
|
||||
app.AddDevice(host, replacement)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
req := httptest.NewRequest(http.MethodDelete, "/api/control/devices/"+host, nil)
|
||||
req = withChiParams(req, map[string]string{"id": host})
|
||||
w := httptest.NewRecorder()
|
||||
|
||||
app.HandleDeleteDevice(w, req)
|
||||
|
||||
if w.Code != http.StatusConflict {
|
||||
t.Fatalf("status = %d, want %d", w.Code, http.StatusConflict)
|
||||
}
|
||||
got, ok := app.GetDevice(host)
|
||||
if !ok || got != replacement {
|
||||
t.Fatal("concurrent replacement was removed")
|
||||
}
|
||||
|
||||
app.RemoveDevice(host)
|
||||
}
|
||||
|
||||
func TestHandleAPIDevice(t *testing.T) {
|
||||
app := createTestApp()
|
||||
|
||||
|
||||
@@ -10,7 +10,9 @@
|
||||
--accent: #000000;
|
||||
--accent-fg: #ffffff;
|
||||
--online: #22c55e;
|
||||
--stale: #f59e0b;
|
||||
--offline: #9ca3af;
|
||||
--danger: #b42318;
|
||||
--radius: 8px;
|
||||
--shadow: 0 1px 3px rgba(0,0,0,.10), 0 1px 2px rgba(0,0,0,.06);
|
||||
--nav-icon-filter: brightness(0) invert(1);
|
||||
@@ -25,6 +27,7 @@
|
||||
--text-dim: #ccc;
|
||||
--accent: #e0e0e0;
|
||||
--accent-fg:#111;
|
||||
--danger: #ff8a80;
|
||||
--nav-icon-filter: none;
|
||||
}
|
||||
}
|
||||
@@ -356,10 +359,15 @@ img { display: block; max-width: 100%; }
|
||||
}
|
||||
|
||||
.device-card {
|
||||
appearance: none;
|
||||
width: 100%;
|
||||
background: var(--surface);
|
||||
border: 1px solid var(--border);
|
||||
border-radius: var(--radius);
|
||||
padding: 1rem;
|
||||
color: inherit;
|
||||
font: inherit;
|
||||
text-align: left;
|
||||
cursor: pointer;
|
||||
transition: box-shadow .15s, transform .1s;
|
||||
box-shadow: var(--shadow);
|
||||
@@ -368,30 +376,12 @@ img { display: block; max-width: 100%; }
|
||||
min-width: 0;
|
||||
}
|
||||
.device-card:hover { box-shadow: 0 4px 12px rgba(0,0,0,.12); transform: translateY(-1px); }
|
||||
.device-card:focus-visible { outline: 2px solid var(--accent); outline-offset: 2px; }
|
||||
|
||||
.device-header { display: flex; align-items: center; justify-content: space-between; margin-bottom: .25rem; gap: .5rem; }
|
||||
.device-name { font-weight: 600; font-size: .95rem; min-width: 0; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
|
||||
.device-header-right { display: flex; align-items: center; gap: .5rem; flex-shrink: 0; }
|
||||
|
||||
/* Quiet remove affordance: invisible until the card is hovered, then dim,
|
||||
warming to red only on its own hover. Keeps the grid relaxed. */
|
||||
.device-remove {
|
||||
border: 0;
|
||||
background: none;
|
||||
padding: 0;
|
||||
width: 1.1rem;
|
||||
height: 1.1rem;
|
||||
line-height: 1;
|
||||
font-size: .8rem;
|
||||
color: var(--text-dim);
|
||||
cursor: pointer;
|
||||
opacity: 0;
|
||||
transition: opacity .15s, color .15s;
|
||||
}
|
||||
.device-card:hover .device-remove { opacity: .55; }
|
||||
.device-remove:hover { opacity: 1; color: var(--offline); }
|
||||
.device-remove:focus-visible { opacity: 1; outline: 2px solid var(--offline); outline-offset: 2px; }
|
||||
|
||||
.device-list-note {
|
||||
margin: .9rem .15rem 0;
|
||||
font-size: .78rem;
|
||||
@@ -408,6 +398,7 @@ img { display: block; max-width: 100%; }
|
||||
flex-shrink: 0;
|
||||
}
|
||||
.device-indicator.online { background: var(--online); }
|
||||
.device-indicator.stale { background: var(--stale); }
|
||||
.device-indicator.offline { background: var(--offline); }
|
||||
|
||||
.now-playing-mini { font-size: .8rem; color: var(--text-dim); margin-top: .25rem; white-space: nowrap; overflow: hidden; text-overflow: ellipsis; }
|
||||
@@ -417,6 +408,28 @@ img { display: block; max-width: 100%; }
|
||||
/* ── Device detail ───────────────────────────────────────────────────────── */
|
||||
.device-detail { max-width: 560px; }
|
||||
|
||||
.device-management-section {
|
||||
margin-top: 1.5rem;
|
||||
padding-top: 1.25rem;
|
||||
border-top: 1px solid var(--border);
|
||||
}
|
||||
|
||||
.device-remove-action {
|
||||
display: inline-flex;
|
||||
align-items: center;
|
||||
gap: .45rem;
|
||||
min-height: 44px;
|
||||
color: var(--text-dim);
|
||||
}
|
||||
.device-remove-action:hover {
|
||||
color: var(--danger);
|
||||
border-color: var(--danger);
|
||||
}
|
||||
.device-remove-action:focus-visible {
|
||||
outline: 2px solid var(--danger);
|
||||
outline-offset: 2px;
|
||||
}
|
||||
|
||||
/* ── Now playing ─────────────────────────────────────────────────────────── */
|
||||
.now-playing {
|
||||
display: flex;
|
||||
|
||||
@@ -17,10 +17,11 @@ import { TTS } from './components/TTS.js';
|
||||
import { Announcements } from './components/Announcements.js';
|
||||
import { api } from './api.js';
|
||||
import { isSoundTouch10StereoPair } from './stereoPresentation.mjs';
|
||||
import { removeDeviceAndRefresh } from './deviceRemoval.js';
|
||||
|
||||
const html = htm.bind(h);
|
||||
|
||||
function DeviceDetail({ deviceId, devices, onBack, onDevicesChanged, notify }) {
|
||||
function DeviceDetail({ deviceId, devices, onBack, onDevicesChanged, notify, onRemove }) {
|
||||
const device = devices[deviceId];
|
||||
|
||||
if (!device) {
|
||||
@@ -62,6 +63,26 @@ function DeviceDetail({ deviceId, devices, onBack, onDevicesChanged, notify }) {
|
||||
` : null}
|
||||
<${Zone} deviceId=${deviceId} devices=${devices} />
|
||||
<${Recents} deviceId=${deviceId} />
|
||||
${!device.stereoPair ? html`
|
||||
<div class="device-management-section">
|
||||
<div class="section-title">Device management</div>
|
||||
<button type="button"
|
||||
class="btn-secondary device-remove-action"
|
||||
aria-label=${`Remove ${device.info?.name || deviceId} from AfterTouch`}
|
||||
onClick=${() => onRemove(deviceId)}>
|
||||
<svg width="16" height="16" viewBox="0 0 24 24" fill="none"
|
||||
stroke="currentColor" stroke-width="2" stroke-linecap="round"
|
||||
stroke-linejoin="round" aria-hidden="true">
|
||||
<path d="M3 6h18" />
|
||||
<path d="M8 6V4h8v2" />
|
||||
<path d="M19 6l-1 14H6L5 6" />
|
||||
<path d="M10 11v5" />
|
||||
<path d="M14 11v5" />
|
||||
</svg>
|
||||
<span>Remove from AfterTouch</span>
|
||||
</button>
|
||||
</div>
|
||||
` : null}
|
||||
</div>
|
||||
`;
|
||||
}
|
||||
@@ -178,26 +199,23 @@ function App() {
|
||||
|
||||
async function refreshDevices() {
|
||||
const resp = await api.devices();
|
||||
if (resp?.success) setDevices(resp.data || {});
|
||||
if (!resp?.success) throw new Error(resp?.error || 'Failed to refresh devices');
|
||||
setDevices(resp.data || {});
|
||||
}
|
||||
|
||||
async function removeDevice(id) {
|
||||
const name = devices[id]?.info?.name || id;
|
||||
if (!confirm(`Remove "${name}"?\n\nThis clears it from AfterTouch. A device still online may reappear after the next discovery scan.`)) {
|
||||
if (!confirm(`Remove "${name}" from AfterTouch?\n\nThis does not reset the speaker. A device still online may reappear after the next discovery scan.`)) {
|
||||
return;
|
||||
}
|
||||
// Optimistically drop it; the server's devices broadcast reconciles.
|
||||
setDevices(prev => {
|
||||
const next = { ...prev };
|
||||
delete next[id];
|
||||
return next;
|
||||
await removeDeviceAndRefresh({
|
||||
id,
|
||||
name,
|
||||
remove: api.removeDevice,
|
||||
refresh: refreshDevices,
|
||||
showDeviceList: () => navigate('devices'),
|
||||
notify: showToast,
|
||||
});
|
||||
try {
|
||||
const resp = await api.removeDevice(id);
|
||||
showToast(resp?.success ? `Removed "${name}"` : (resp?.error || 'Failed to remove device'));
|
||||
} catch (err) {
|
||||
showToast('Failed to remove device');
|
||||
}
|
||||
}
|
||||
|
||||
return html`
|
||||
@@ -274,7 +292,6 @@ function App() {
|
||||
isDiscovering=${isDiscovering}
|
||||
onSelect=${(id) => navigate('device', id)}
|
||||
onDiscover=${discover}
|
||||
onRemove=${removeDevice}
|
||||
/>
|
||||
` : page === 'device' ? html`
|
||||
<${DeviceDetail}
|
||||
@@ -284,6 +301,7 @@ function App() {
|
||||
onBack=${() => navigate('devices')}
|
||||
onDevicesChanged=${refreshDevices}
|
||||
notify=${showToast}
|
||||
onRemove=${removeDevice}
|
||||
/>
|
||||
` : page === 'tunein' ? html`
|
||||
<${TuneInBrowser} key="tunein-browser" devices=${devices} />
|
||||
@@ -310,7 +328,8 @@ function App() {
|
||||
</footer>
|
||||
` : null}
|
||||
|
||||
${toast ? html`<div class="toast" key="toast">${toast}</div>` : null}
|
||||
${toast ? html`<div class="toast" role="status" aria-live="polite"
|
||||
aria-atomic="true" key="toast">${toast}</div>` : null}
|
||||
</div>
|
||||
`;
|
||||
}
|
||||
|
||||
@@ -1,45 +1,33 @@
|
||||
import { h } from 'preact';
|
||||
import { useState } from 'preact/hooks';
|
||||
import htm from 'htm';
|
||||
import {
|
||||
connectivityLabel,
|
||||
connectivityState,
|
||||
sortDeviceEntries,
|
||||
} from '../devicePresentation.js';
|
||||
|
||||
const html = htm.bind(h);
|
||||
|
||||
const SORT_LS_KEY = 'aftertouch_device_sort';
|
||||
|
||||
function sortEntries(entries, mode) {
|
||||
const copy = [...entries];
|
||||
if (mode === 'name') {
|
||||
// Sort by the speaker's display name, falling back to the map key (its IP)
|
||||
// when a device has no name yet.
|
||||
copy.sort(([idA, a], [idB, b]) =>
|
||||
(a?.info?.name || idA).localeCompare(b?.info?.name || idB, undefined, { sensitivity: 'base' }));
|
||||
} else {
|
||||
// Default: by IP (the map key), ordered numerically so .2 precedes .10.
|
||||
copy.sort(([idA], [idB]) =>
|
||||
idA.localeCompare(idB, undefined, { numeric: true, sensitivity: 'base' }));
|
||||
}
|
||||
return copy;
|
||||
}
|
||||
|
||||
function DeviceCard({ id, device, onSelect, onRemove }) {
|
||||
function DeviceCard({ id, device, onSelect }) {
|
||||
const { info, status } = device;
|
||||
const stereoPair = device.stereoPair;
|
||||
const np = status?.nowPlaying;
|
||||
const isPlaying = np?.PlayStatus === 'PLAY_STATE';
|
||||
const isStandby = !np || np.Source === 'STANDBY';
|
||||
const connectivity = connectivityState(device);
|
||||
const statusLabel = `Connectivity: ${connectivityLabel(device)}`;
|
||||
|
||||
return html`
|
||||
<div class="device-card" onClick=${() => onSelect(id)}>
|
||||
<div class="device-header">
|
||||
<button type="button" class="device-card" onClick=${() => onSelect(id)}>
|
||||
<span class="device-header">
|
||||
<span class="device-name" title=${info?.name || id}>${info?.name || id}</span>
|
||||
<span class="device-header-right">
|
||||
<span class="device-indicator ${status?.isConnected ? 'online' : 'offline'}"></span>
|
||||
${!stereoPair ? html`<button class="device-remove" title="Remove this device"
|
||||
aria-label="Remove this device"
|
||||
onClick=${(e) => { e.stopPropagation(); onRemove(id); }}>✕</button>` : null}
|
||||
</span>
|
||||
</div>
|
||||
<div class="device-type">
|
||||
<span class="device-indicator ${connectivity}" role="status"
|
||||
title=${statusLabel} aria-label=${statusLabel}></span>
|
||||
</span>
|
||||
<span class="device-type">
|
||||
${info?.type || ''}
|
||||
${info?.ip_address ? html`<span class="device-ip">(${info.ip_address})</span>` : null}
|
||||
${stereoPair ? html`
|
||||
@@ -47,20 +35,20 @@ function DeviceCard({ id, device, onSelect, onRemove }) {
|
||||
Stereo pair ${stereoPair.availableMemberCount}/${stereoPair.memberCount}
|
||||
</span>
|
||||
` : null}
|
||||
</div>
|
||||
</span>
|
||||
${!isStandby ? html`
|
||||
<div class="now-playing-mini" title=${[np.Track || np.StationName || np.Source, np.Artist].filter(Boolean).join(' - ')}>
|
||||
<span class="now-playing-mini" title=${[np.Track || np.StationName || np.Source, np.Artist].filter(Boolean).join(' - ')}>
|
||||
<span class="play-status">${isPlaying ? '▶' : '⏸'}</span>
|
||||
<span class="track-mini">${np.Track || np.StationName || np.Source}</span>
|
||||
${np.Artist ? html`<span class="artist-mini"> — ${np.Artist}</span>` : null}
|
||||
</div>
|
||||
</span>
|
||||
` : null}
|
||||
${isStandby ? html`<div class="standby-label">Standby</div>` : null}
|
||||
</div>
|
||||
${isStandby ? html`<span class="standby-label">Standby</span>` : null}
|
||||
</button>
|
||||
`;
|
||||
}
|
||||
|
||||
export function DeviceList({ devices, isDiscovering, onSelect, onDiscover, onRemove }) {
|
||||
export function DeviceList({ devices, isDiscovering, onSelect, onDiscover }) {
|
||||
const [sortMode, setSortMode] = useState(() => localStorage.getItem(SORT_LS_KEY) || 'ip');
|
||||
|
||||
function changeSort(mode) {
|
||||
@@ -68,7 +56,7 @@ export function DeviceList({ devices, isDiscovering, onSelect, onDiscover, onRem
|
||||
localStorage.setItem(SORT_LS_KEY, mode);
|
||||
}
|
||||
|
||||
const entries = sortEntries(Object.entries(devices), sortMode);
|
||||
const entries = sortDeviceEntries(Object.entries(devices), sortMode);
|
||||
|
||||
return html`
|
||||
<div class="device-list-container">
|
||||
@@ -91,13 +79,9 @@ export function DeviceList({ devices, isDiscovering, onSelect, onDiscover, onRem
|
||||
</div>
|
||||
<div class="device-grid" key="grid">
|
||||
${entries.map(([id, device]) => html`
|
||||
<${DeviceCard} key=${id} id=${id} device=${device} onSelect=${onSelect} onRemove=${onRemove} />
|
||||
<${DeviceCard} key=${id} id=${id} device=${device} onSelect=${onSelect} />
|
||||
`)}
|
||||
</div>
|
||||
<p class="device-list-note" key="note">
|
||||
Removing a device clears it here. One that is still online may
|
||||
reappear after the next discovery scan.
|
||||
</p>`
|
||||
</div>`
|
||||
}
|
||||
</div>
|
||||
`;
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
const connectivityStates = new Set(['online', 'stale', 'offline']);
|
||||
|
||||
export function connectivityState(device) {
|
||||
const reported = device?.status?.connectivity;
|
||||
if (connectivityStates.has(reported)) return reported;
|
||||
|
||||
return device?.status?.isConnected ? 'online' : 'offline';
|
||||
}
|
||||
|
||||
export function connectivityLabel(device) {
|
||||
const state = connectivityState(device);
|
||||
return state.charAt(0).toUpperCase() + state.slice(1);
|
||||
}
|
||||
|
||||
function compareText(a, b) {
|
||||
return String(a).localeCompare(String(b), undefined, {
|
||||
numeric: true,
|
||||
sensitivity: 'base',
|
||||
});
|
||||
}
|
||||
|
||||
function displayName(id, device) {
|
||||
return String(device?.info?.name || '').trim() || id;
|
||||
}
|
||||
|
||||
function presentationAddress(id, device) {
|
||||
return String(device?.info?.ip_address || '').trim() || id;
|
||||
}
|
||||
|
||||
export function sortDeviceEntries(entries, mode) {
|
||||
const sorted = [...entries];
|
||||
|
||||
sorted.sort(([idA, deviceA], [idB, deviceB]) => {
|
||||
const primary = mode === 'name'
|
||||
? compareText(displayName(idA, deviceA), displayName(idB, deviceB))
|
||||
: compareText(presentationAddress(idA, deviceA), presentationAddress(idB, deviceB));
|
||||
|
||||
return primary || compareText(idA, idB);
|
||||
});
|
||||
|
||||
return sorted;
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
export async function removeDeviceAndRefresh({
|
||||
id,
|
||||
name,
|
||||
remove,
|
||||
refresh,
|
||||
showDeviceList,
|
||||
notify,
|
||||
}) {
|
||||
let response;
|
||||
try {
|
||||
response = await remove(id);
|
||||
} catch (_) {
|
||||
notify('Failed to remove device');
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!response?.success) {
|
||||
notify(response?.error || 'Failed to remove device');
|
||||
return false;
|
||||
}
|
||||
|
||||
showDeviceList();
|
||||
try {
|
||||
await refresh();
|
||||
} catch (_) {
|
||||
notify(`Removed "${name}", but failed to refresh devices`);
|
||||
return true;
|
||||
}
|
||||
|
||||
notify(`Removed "${name}"`);
|
||||
return true;
|
||||
}
|
||||
@@ -13,6 +13,7 @@ import (
|
||||
"sync"
|
||||
"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/go-chi/chi/v5"
|
||||
@@ -482,22 +483,19 @@ func (app *WebApp) HandleAPIDiscover(w http.ResponseWriter, r *http.Request) {
|
||||
}
|
||||
}
|
||||
|
||||
// ConnectDeviceWebSocket establishes a WebSocket connection to a device
|
||||
// and keeps it alive: on disconnect or connect failure, it reconnects
|
||||
// with exponential backoff (1 s → 30 s cap, reset after each successful
|
||||
// connect). The goroutine runs for the lifetime of the device entry,
|
||||
// so status flows from the speaker keep streaming through transient
|
||||
// network blips, speaker reboots, and idle timeouts.
|
||||
//
|
||||
// conn.WebSocket is only updated on a successful (re)connect, never
|
||||
// cleared, so the duplicate-spawn guards at the callsites (which check
|
||||
// `if device.WebSocket == nil`) stay correct — once this goroutine is
|
||||
// running for a device, no second one is needed.
|
||||
// ConnectDeviceWebSocket starts the single event-transport supervisor for a
|
||||
// device. Initial connection failures are retried here; after the first
|
||||
// success WebSocketClient owns transport reconnects and this supervisor
|
||||
// observes their state until the device is removed.
|
||||
func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.DeviceConnection) {
|
||||
// Skip WebSocket connection if client is not available (e.g., in tests)
|
||||
if conn.Client == nil {
|
||||
return
|
||||
}
|
||||
if !conn.TryStartWebSocketLoop() {
|
||||
return
|
||||
}
|
||||
defer conn.FinishWebSocketLoop()
|
||||
|
||||
const (
|
||||
initialBackoff = 1 * time.Second
|
||||
@@ -524,6 +522,7 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device
|
||||
// UpdateStatus so concurrent events and the periodic poller
|
||||
// (UpdateDeviceStatus) cannot lose each other's writes.
|
||||
wsClient.OnNowPlaying(func(event *models.NowPlayingUpdatedEvent) {
|
||||
activity := time.Now()
|
||||
np := &event.NowPlaying
|
||||
|
||||
// A /select returns 200 even when the source is rejected; the
|
||||
@@ -536,29 +535,77 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device
|
||||
prevSource = np.Source
|
||||
|
||||
app.applyNowPlayingEvent(conn, np)
|
||||
conn.MarkEventStreamActivity(activity)
|
||||
})
|
||||
|
||||
wsClient.OnVolumeUpdated(func(event *models.VolumeUpdatedEvent) {
|
||||
activity := time.Now()
|
||||
app.applyVolumeEvent(conn, &event.Volume)
|
||||
conn.MarkEventStreamActivity(activity)
|
||||
})
|
||||
|
||||
wsClient.OnConnectionState(func(event *models.ConnectionStateUpdatedEvent) {
|
||||
if !speakerConnectionEventMatches(conn, event.DeviceID) {
|
||||
log.Printf("Ignoring connection state for mismatched device %s on %s",
|
||||
sanitizeLog(event.DeviceID), sanitizeLog(deviceID))
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
app.applyConnectionStateEvent(conn, event.ConnectionState.IsConnected())
|
||||
conn.ApplySpeakerConnectionEvent(webtypes.SpeakerConnectionState{
|
||||
State: event.ConnectionState.State,
|
||||
Signal: event.ConnectionState.Signal,
|
||||
}, time.Now())
|
||||
})
|
||||
|
||||
wsClient.OnPresetUpdated(func(event *models.PresetUpdatedEvent) {
|
||||
activity := time.Now()
|
||||
app.applyPresetEvent(conn, &event.Presets)
|
||||
conn.MarkEventStreamActivity(activity)
|
||||
})
|
||||
|
||||
wsClient.OnBassUpdated(func(event *models.BassUpdatedEvent) {
|
||||
activity := time.Now()
|
||||
app.applyBassEvent(conn, &event.Bass)
|
||||
conn.MarkEventStreamActivity(activity)
|
||||
})
|
||||
|
||||
wsClient.OnGroupUpdated(func(event *models.GroupUpdatedEvent) {
|
||||
app.applyGroupUpdatedEvent(conn, event)
|
||||
})
|
||||
|
||||
if err := wsClient.Connect(); err != nil {
|
||||
wsClient.OnNameUpdated(func(event *models.NameUpdatedEvent) {
|
||||
conn.MarkEventStreamActivity(time.Now())
|
||||
conn.ApplyNameEvent(event.Name.Value)
|
||||
})
|
||||
|
||||
wsClient.OnTransportState(func(connected bool, generation uint64) {
|
||||
if !conn.ObserveEventStreamTransport(generation, connected, time.Now()) {
|
||||
return
|
||||
}
|
||||
|
||||
// Drives the same FieldConnectivity fencing the old inline
|
||||
// connect/disconnect sites used to update directly -- ordered by
|
||||
// ObserveEventStreamTransport's own generation check above, so a
|
||||
// reordered transport callback can no longer apply here either.
|
||||
app.applyConnectionStateEvent(conn, connected)
|
||||
|
||||
if connected {
|
||||
if generation > 1 {
|
||||
log.Printf("WebSocket reconnected for device %s", sanitizeLog(deviceID))
|
||||
go app.UpdateDeviceStatus(deviceID, conn)
|
||||
}
|
||||
} else {
|
||||
log.Printf("WebSocket transport disconnected for device %s", sanitizeLog(deviceID))
|
||||
}
|
||||
})
|
||||
|
||||
published, err := publishAndConnectDeviceWebSocket(conn, wsClient, wsClient.Connect)
|
||||
if !published {
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
log.Printf("Failed to connect WebSocket for device %s: %v (retrying in %s)", sanitizeLog(deviceID), err, backoff)
|
||||
|
||||
if sleepOrDone(conn, backoff) {
|
||||
@@ -573,10 +620,6 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device
|
||||
continue
|
||||
}
|
||||
|
||||
conn.WebSocket = wsClient
|
||||
|
||||
app.applyConnectionStateEvent(conn, true)
|
||||
|
||||
log.Printf("WebSocket connected for device %s", sanitizeLog(deviceID))
|
||||
|
||||
// Fetch current state immediately: speakers do not replay events on
|
||||
@@ -588,24 +631,45 @@ func (app *WebApp) ConnectDeviceWebSocket(deviceID string, conn *webtypes.Device
|
||||
// starts at the lowest cadence again.
|
||||
backoff = initialBackoff
|
||||
|
||||
// Block until the device-side WebSocket disconnects.
|
||||
wsClient.Wait()
|
||||
<-conn.Done()
|
||||
|
||||
app.applyConnectionStateEvent(conn, false)
|
||||
|
||||
log.Printf("WebSocket disconnected for device %s — reconnecting in %s", sanitizeLog(deviceID), backoff)
|
||||
|
||||
if sleepOrDone(conn, backoff) {
|
||||
return
|
||||
}
|
||||
|
||||
backoff *= 2
|
||||
if backoff > maxBackoff {
|
||||
backoff = maxBackoff
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func publishAndConnectDeviceWebSocket(
|
||||
conn *webtypes.DeviceConnection,
|
||||
wsClient *client.WebSocketClient,
|
||||
connect func() error,
|
||||
) (bool, error) {
|
||||
if !conn.SetWebSocket(wsClient) {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
if err := connect(); err != nil {
|
||||
conn.ClearWebSocket(wsClient)
|
||||
_ = wsClient.Close()
|
||||
|
||||
return true, err
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func speakerConnectionEventMatches(conn *webtypes.DeviceConnection, eventDeviceID string) bool {
|
||||
eventDeviceID = strings.TrimSpace(eventDeviceID)
|
||||
if eventDeviceID == "" {
|
||||
return true
|
||||
}
|
||||
|
||||
info := conn.Info()
|
||||
if info == nil || strings.TrimSpace(info.DeviceID) == "" {
|
||||
return false
|
||||
}
|
||||
|
||||
return strings.EqualFold(eventDeviceID, strings.TrimSpace(info.DeviceID))
|
||||
}
|
||||
|
||||
// sleepOrDone waits for d to elapse or for the connection to be closed,
|
||||
// whichever comes first. It returns true if the connection was closed
|
||||
// (the caller should stop), false if the timer fired normally.
|
||||
@@ -624,13 +688,17 @@ func sleepOrDone(conn *webtypes.DeviceConnection, d time.Duration) bool {
|
||||
// UpdateDeviceStatus fetches current status from the device.
|
||||
//
|
||||
// Network calls run outside any atomic merge so slow I/O never blocks a
|
||||
// concurrent CAS retry. Each field (NowPlaying/Volume/Presets/Sources/Bass,
|
||||
// plus derived connectivity) is merged and ordered independently via its own
|
||||
// StatusField generation (BeginFieldPoll/CompleteFieldPoll/ApplyFieldEvent in
|
||||
// webtypes) -- a real-time push event, or a concurrent poll, for one field
|
||||
// can supersede only that field. A slow-but-successful fetch for one field
|
||||
// is never discarded merely because a DIFFERENT field's event or poll
|
||||
// completion happened to land first.
|
||||
// concurrent CAS retry. Each field (NowPlaying/Name/Volume/Presets/Sources/
|
||||
// Bass, plus derived connectivity) is merged and ordered independently via
|
||||
// its own StatusField generation (BeginFieldPoll/CompleteFieldPoll/
|
||||
// ApplyFieldEvent in webtypes) -- a real-time push event, or a concurrent
|
||||
// poll, for one field can supersede only that field. A slow-but-successful
|
||||
// fetch for one field is never discarded merely because a DIFFERENT field's
|
||||
// event or poll completion happened to land first. Independently, the same
|
||||
// round also feeds BeginHTTPPoll/CompleteHTTPPoll, which derives HTTP
|
||||
// reachability and the Online/Stale/Offline connectivity classification --
|
||||
// CompleteHTTPPoll is called with a nil merge func here since field merging
|
||||
// is already handled per-field above; it only records health/connectivity.
|
||||
func (app *WebApp) UpdateDeviceStatus(deviceID string, conn *webtypes.DeviceConnection) {
|
||||
app.updateDeviceStatus(deviceID, conn, nil)
|
||||
}
|
||||
@@ -658,6 +726,7 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection,
|
||||
sourcesGen := conn.BeginFieldPoll(webtypes.FieldSources)
|
||||
bassGen := conn.BeginFieldPoll(webtypes.FieldBass)
|
||||
connectivityGen := conn.BeginFieldPoll(webtypes.FieldConnectivity)
|
||||
pollGeneration := conn.BeginHTTPPoll()
|
||||
|
||||
// /getGroup must be gated to ST10 models -- see Client.GetGroup's doc
|
||||
// comment (verified against real hardware: a ST20 never replies at all,
|
||||
@@ -668,11 +737,13 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection,
|
||||
if stereoCapable && groupBaseline == nil {
|
||||
groupGeneration = conn.BeginGroupRefresh()
|
||||
}
|
||||
nameGeneration := conn.BeginNameRefresh()
|
||||
|
||||
// Phase 1: slow network fetches. Local vars only, no shared state
|
||||
// is touched yet. Errors are recorded so the merge below can tell
|
||||
// "field N stayed unchanged" apart from "field N got refreshed".
|
||||
nowPlaying, nowPlayingErr := conn.Client.GetNowPlaying()
|
||||
name, nameErr := conn.Client.GetName()
|
||||
volume, volumeErr := conn.Client.GetVolume()
|
||||
presets, presetsErr := conn.Client.GetPresets()
|
||||
sources, sourcesErr := conn.Client.GetSources()
|
||||
@@ -737,6 +808,12 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection,
|
||||
})
|
||||
}
|
||||
|
||||
if nameErr == nil {
|
||||
anyFetchSucceeded = true
|
||||
|
||||
conn.ApplyPolledName(nameGeneration, name.Value)
|
||||
}
|
||||
|
||||
// Mark as connected if we successfully got at least one status from
|
||||
// this round. Mirrors prior behaviour: deliberately does NOT fold
|
||||
// groupErr in here. GetGroup is gated to stereo-capable models and
|
||||
@@ -749,6 +826,15 @@ func (app *WebApp) updateDeviceStatus(_ string, conn *webtypes.DeviceConnection,
|
||||
s.LastActivity = time.Now()
|
||||
})
|
||||
|
||||
// Independently of the per-field merges above, also record this round
|
||||
// against the health/connectivity generation. merge is nil: field data
|
||||
// was already merged field-by-field above, so this call only derives
|
||||
// HTTPReachable/WebSocketConnected/Connectivity (Online/Stale/Offline)
|
||||
// from anyFetchSucceeded -- it must never re-apply payload fields, or a
|
||||
// concurrent speaker event landing between BeginHTTPPoll and here could
|
||||
// cause this call to silently discard part of the merge above.
|
||||
conn.CompleteHTTPPoll(pollGeneration, anyFetchSucceeded, time.Now(), nil)
|
||||
|
||||
if stereoCapable && groupErr == nil {
|
||||
if groupBaseline != nil {
|
||||
conn.ApplyPolledGroupIfBaseline(*groupBaseline, group)
|
||||
@@ -762,6 +848,8 @@ func (app *WebApp) applyGroupUpdatedEvent(
|
||||
conn *webtypes.DeviceConnection,
|
||||
event *models.GroupUpdatedEvent,
|
||||
) bool {
|
||||
conn.MarkEventStreamActivity(time.Now())
|
||||
|
||||
return app.queueBroadcastIfChanged(conn.ApplyGroupEvent(&event.Group, time.Now()))
|
||||
}
|
||||
|
||||
@@ -797,7 +885,7 @@ func (app *WebApp) HandleDeviceWebSocket(w http.ResponseWriter, r *http.Request)
|
||||
Type: "device_status",
|
||||
DeviceID: deviceID,
|
||||
Data: map[string]interface{}{
|
||||
"info": device.DeviceInfo,
|
||||
"info": device.Info(),
|
||||
"status": device.Status(),
|
||||
},
|
||||
})
|
||||
@@ -857,14 +945,14 @@ func (app *WebApp) writeDeviceWebSocketUpdate(
|
||||
Type: "device_status",
|
||||
DeviceID: deviceID,
|
||||
Data: map[string]interface{}{
|
||||
"info": device.DeviceInfo,
|
||||
"info": device.Info(),
|
||||
"status": status,
|
||||
},
|
||||
}); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if device.WebSocket == nil || !status.IsConnected {
|
||||
if device.CurrentWebSocket() == nil || !status.IsConnected {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package soundtouchweb
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
@@ -16,6 +17,54 @@ import (
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
|
||||
func TestPublishAndConnectDeviceWebSocketLetsRemovalCancelInitialDial(t *testing.T) {
|
||||
conn := webtypes.NewDeviceConnection(nil, nil)
|
||||
wsClient := client.NewClientFromHost("192.0.2.10").NewWebSocketClient(nil)
|
||||
connectStarted := make(chan struct{})
|
||||
connectDone := make(chan struct {
|
||||
published bool
|
||||
err error
|
||||
}, 1)
|
||||
|
||||
go func() {
|
||||
published, err := publishAndConnectDeviceWebSocket(conn, wsClient, func() error {
|
||||
close(connectStarted)
|
||||
wsClient.Wait()
|
||||
|
||||
return context.Canceled
|
||||
})
|
||||
connectDone <- struct {
|
||||
published bool
|
||||
err error
|
||||
}{published: published, err: err}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-connectStarted:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("initial dial did not start")
|
||||
}
|
||||
|
||||
if got := conn.CurrentWebSocket(); got != wsClient {
|
||||
t.Fatalf("CurrentWebSocket() = %p during initial dial, want %p", got, wsClient)
|
||||
}
|
||||
|
||||
conn.Close()
|
||||
|
||||
select {
|
||||
case result := <-connectDone:
|
||||
if !result.published || result.err == nil {
|
||||
t.Fatalf("publish/connect result = (%v, %v), want (true, error)", result.published, result.err)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("device removal did not cancel the initial dial")
|
||||
}
|
||||
|
||||
if got := conn.CurrentWebSocket(); got != nil {
|
||||
t.Fatalf("CurrentWebSocket() = %p after removal, want nil", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateDeviceStatusRefreshesGroup(t *testing.T) {
|
||||
server := newStatusTestServer(t, http.StatusOK, `<group id="pair-1">
|
||||
<name>Living Room</name>
|
||||
@@ -69,6 +118,101 @@ func TestUpdateDeviceStatusPreservesGroupOnError(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestSpeakerConnectionEventMatchesRegisteredHardwareID(t *testing.T) {
|
||||
conn := webtypes.NewDeviceConnection(nil, &models.DeviceInfo{DeviceID: "AA11BB22CC33"})
|
||||
|
||||
for _, eventDeviceID := range []string{"", "AA11BB22CC33", "aa11bb22cc33"} {
|
||||
if !speakerConnectionEventMatches(conn, eventDeviceID) {
|
||||
t.Fatalf("event device ID %q should match", eventDeviceID)
|
||||
}
|
||||
}
|
||||
if speakerConnectionEventMatches(conn, "DEADBEEF0000") {
|
||||
t.Fatal("mismatched speaker connection event was accepted")
|
||||
}
|
||||
if speakerConnectionEventMatches(webtypes.NewDeviceConnection(nil, nil), "AA11BB22CC33") {
|
||||
t.Fatal("identified event matched a connection without device identity")
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateDeviceStatusRefreshesDeviceName(t *testing.T) {
|
||||
server := newStatusTestServer(t, http.StatusOK, `<group/>`)
|
||||
defer server.Close()
|
||||
|
||||
conn := webtypes.NewDeviceConnection(
|
||||
client.NewClientFromHost(server.URL),
|
||||
&models.DeviceInfo{Name: "Old Name", DeviceID: "device-1"},
|
||||
)
|
||||
NewWebApp().UpdateDeviceStatus("device-1", conn)
|
||||
|
||||
if info := conn.Info(); info == nil || info.Name != "Living Room Left" {
|
||||
t.Fatalf("refreshed device info = %+v, want Living Room Left", info)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateDeviceStatusDoesNotOverwriteNewerNameEvent(t *testing.T) {
|
||||
nameRequestStarted := make(chan struct{})
|
||||
releaseNameResponse := make(chan struct{})
|
||||
responses := map[string]string{
|
||||
"/now_playing": `<nowPlaying source="STANDBY"><playStatus>STOP_STATE</playStatus></nowPlaying>`,
|
||||
"/volume": `<volume><targetvolume>10</targetvolume><actualvolume>10</actualvolume><muteenabled>false</muteenabled></volume>`,
|
||||
"/presets": `<presets/>`,
|
||||
"/sources": `<sources/>`,
|
||||
"/bass": `<bass><targetbass>0</targetbass><actualbass>0</actualbass></bass>`,
|
||||
"/getGroup": `<group/>`,
|
||||
}
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/xml")
|
||||
if r.URL.Path == "/name" {
|
||||
close(nameRequestStarted)
|
||||
<-releaseNameResponse
|
||||
_, _ = w.Write([]byte(`<name>Old Poll Result</name>`))
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
body, ok := responses[r.URL.Path]
|
||||
if !ok {
|
||||
t.Errorf("unexpected status endpoint %q", r.URL.Path)
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
|
||||
return
|
||||
}
|
||||
_, _ = w.Write([]byte(body))
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
conn := webtypes.NewDeviceConnection(
|
||||
client.NewClientFromHost(server.URL),
|
||||
&models.DeviceInfo{Name: "Initial Name", DeviceID: "device-1", Type: "SoundTouch 10"},
|
||||
)
|
||||
refreshDone := make(chan struct{})
|
||||
go func() {
|
||||
NewWebApp().UpdateDeviceStatus("device-1", conn)
|
||||
close(refreshDone)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-nameRequestStarted:
|
||||
case <-time.After(time.Second):
|
||||
close(releaseNameResponse)
|
||||
t.Fatal("name poll did not start")
|
||||
}
|
||||
|
||||
conn.MarkEventStreamActivity(time.Now())
|
||||
conn.ApplyNameEvent("New Event Name")
|
||||
close(releaseNameResponse)
|
||||
|
||||
select {
|
||||
case <-refreshDone:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("status refresh did not finish")
|
||||
}
|
||||
|
||||
if info := conn.Info(); info == nil || info.Name != "New Event Name" {
|
||||
t.Fatalf("device info after stale poll = %+v, want newer event name", info)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateDeviceStatusSkipsGroupForNonStereoModel(t *testing.T) {
|
||||
var groupRequested atomic.Bool
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
@@ -81,6 +225,7 @@ func TestUpdateDeviceStatusSkipsGroupForNonStereoModel(t *testing.T) {
|
||||
|
||||
responses := map[string]string{
|
||||
"/now_playing": `<nowPlaying source="STANDBY"><playStatus>STOP_STATE</playStatus></nowPlaying>`,
|
||||
"/name": `<name>Living Room Left</name>`,
|
||||
"/volume": `<volume><targetvolume>10</targetvolume><actualvolume>10</actualvolume><muteenabled>false</muteenabled></volume>`,
|
||||
"/presets": `<presets/>`,
|
||||
"/sources": `<sources/>`,
|
||||
@@ -278,6 +423,7 @@ func newStatusTestServer(t *testing.T, groupStatus int, groupBody string) *httpt
|
||||
|
||||
responses := map[string]string{
|
||||
"/now_playing": `<nowPlaying source="STANDBY"><playStatus>STOP_STATE</playStatus></nowPlaying>`,
|
||||
"/name": `<name>Living Room Left</name>`,
|
||||
"/volume": `<volume><targetvolume>10</targetvolume><actualvolume>10</actualvolume><muteenabled>false</muteenabled></volume>`,
|
||||
"/presets": `<presets/>`,
|
||||
"/sources": `<sources/>`,
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gesellix/bose-soundtouch/pkg/client"
|
||||
"github.com/gesellix/bose-soundtouch/pkg/models"
|
||||
)
|
||||
|
||||
@@ -23,12 +24,270 @@ func TestNewDeviceConnection_InitialStatus(t *testing.T) {
|
||||
if status.IsConnected {
|
||||
t.Error("IsConnected should default to false")
|
||||
}
|
||||
if status.Connectivity != ConnectivityOffline {
|
||||
t.Errorf("Connectivity = %q, want %q", status.Connectivity, ConnectivityOffline)
|
||||
}
|
||||
|
||||
if status.LastActivity.IsZero() {
|
||||
t.Error("LastActivity should be initialised, got zero time")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeviceConnectionRejectsWebSocketAfterClose(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, nil)
|
||||
conn.Close()
|
||||
|
||||
ws := client.NewClientFromHost("192.0.2.10").NewWebSocketClient(nil)
|
||||
if conn.SetWebSocket(ws) {
|
||||
t.Fatal("SetWebSocket() accepted a transport after Close()")
|
||||
}
|
||||
if got := conn.CurrentWebSocket(); got != nil {
|
||||
t.Fatalf("CurrentWebSocket() = %p after Close(), want nil", got)
|
||||
}
|
||||
|
||||
waitDone := make(chan struct{})
|
||||
go func() {
|
||||
ws.Wait()
|
||||
close(waitDone)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-waitDone:
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Fatal("rejected WebSocket transport was not stopped")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeviceConnectionWebSocketAccessConcurrentWithClose(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, nil)
|
||||
ws := client.NewClientFromHost("192.0.2.10").NewWebSocketClient(nil)
|
||||
start := make(chan struct{})
|
||||
var wg sync.WaitGroup
|
||||
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
for range 100 {
|
||||
conn.SetWebSocket(ws)
|
||||
_ = conn.CurrentWebSocket()
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
conn.Close()
|
||||
}()
|
||||
|
||||
close(start)
|
||||
wg.Wait()
|
||||
|
||||
if got := conn.CurrentWebSocket(); got != nil {
|
||||
t.Fatalf("CurrentWebSocket() = %p after concurrent Close(), want nil", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeviceConnectionClearWebSocketPreservesReplacement(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, nil)
|
||||
soundTouchClient := client.NewClientFromHost("192.0.2.10")
|
||||
original := soundTouchClient.NewWebSocketClient(nil)
|
||||
replacement := soundTouchClient.NewWebSocketClient(nil)
|
||||
|
||||
conn.SetWebSocket(original)
|
||||
conn.SetWebSocket(replacement)
|
||||
if conn.ClearWebSocket(original) {
|
||||
t.Fatal("ClearWebSocket() cleared a replacement transport")
|
||||
}
|
||||
if got := conn.CurrentWebSocket(); got != replacement {
|
||||
t.Fatalf("CurrentWebSocket() = %p, want replacement %p", got, replacement)
|
||||
}
|
||||
|
||||
_ = original.Close()
|
||||
conn.Close()
|
||||
}
|
||||
|
||||
func TestDeviceConnectionInfoReflectsUpdatedName(t *testing.T) {
|
||||
discovered := &models.DeviceInfo{Name: "Living Room", DeviceID: "DEVICE01"}
|
||||
conn := NewDeviceConnection(nil, discovered)
|
||||
|
||||
conn.ApplyNameEvent("Living Room Left")
|
||||
info := conn.Info()
|
||||
|
||||
if info == nil || info.Name != "Living Room Left" || info.DeviceID != "DEVICE01" {
|
||||
t.Fatalf("Info() = %+v, want updated name with original metadata", info)
|
||||
}
|
||||
if discovered.Name != "Living Room" {
|
||||
t.Fatalf("discovery snapshot was mutated: %+v", discovered)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNameEventSupersedesInFlightPoll(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "Initial"})
|
||||
generation := conn.BeginNameRefresh()
|
||||
|
||||
conn.ApplyNameEvent("Event Name")
|
||||
if conn.ApplyPolledName(generation, "Stale Poll Name") {
|
||||
t.Fatal("stale name poll was accepted")
|
||||
}
|
||||
if got := conn.Info().Name; got != "Event Name" {
|
||||
t.Fatalf("Info().Name = %q, want newer event name", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConnectivityAggregatesHTTPAndEventStreamEvidence(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
conn.ObserveEventStream(true, started)
|
||||
status := conn.Status()
|
||||
if status.Connectivity != ConnectivityOnline || status.HTTPReachable ||
|
||||
!status.WebSocketConnected || !status.IsConnected {
|
||||
t.Fatalf("stream-only success = %+v", status)
|
||||
}
|
||||
|
||||
failure := conn.BeginHTTPPoll()
|
||||
conn.CompleteHTTPPoll(failure, false, started.Add(30*time.Second), nil)
|
||||
status = conn.Status()
|
||||
if status.Connectivity != ConnectivityOnline || status.HTTPReachable ||
|
||||
!status.WebSocketConnected || !status.IsConnected {
|
||||
t.Fatalf("HTTP failure over live stream = %+v", status)
|
||||
}
|
||||
|
||||
conn.ObserveEventStream(false, started.Add(30*time.Second))
|
||||
status = conn.Status()
|
||||
if status.Connectivity != ConnectivityStale || status.WebSocketConnected || !status.IsConnected {
|
||||
t.Fatalf("direct-path loss within grace = %+v", status)
|
||||
}
|
||||
|
||||
secondFailure := conn.BeginHTTPPoll()
|
||||
conn.CompleteHTTPPoll(secondFailure, false, started.Add(60*time.Second), nil)
|
||||
status = conn.Status()
|
||||
if status.Connectivity != ConnectivityOffline || status.IsConnected {
|
||||
t.Fatalf("sustained direct-path loss = %+v", status)
|
||||
}
|
||||
|
||||
recovery := conn.BeginHTTPPoll()
|
||||
conn.CompleteHTTPPoll(recovery, true, started.Add(61*time.Second), nil)
|
||||
status = conn.Status()
|
||||
if status.Connectivity != ConnectivityOnline || !status.HTTPReachable || !status.IsConnected {
|
||||
t.Fatalf("HTTP recovery = %+v", status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOlderHTTPFailureCannotDemoteNewerStreamActivity(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC)
|
||||
conn.MarkHTTPSuccess(started)
|
||||
|
||||
older := conn.BeginHTTPPoll()
|
||||
conn.ObserveEventStream(true, started.Add(61*time.Second))
|
||||
if !conn.CompleteHTTPPoll(older, false, started.Add(62*time.Second), nil) {
|
||||
t.Fatal("latest HTTP-channel observation was unexpectedly rejected")
|
||||
}
|
||||
|
||||
status := conn.Status()
|
||||
if status.Connectivity != ConnectivityOnline || status.HTTPReachable ||
|
||||
!status.WebSocketConnected || !status.IsConnected {
|
||||
t.Fatalf("older HTTP failure demoted newer stream success: %+v", status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOlderHTTPPayloadCannotOverwriteNewerSpeakerEvent(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
poll := conn.BeginHTTPPoll()
|
||||
conn.ApplySpeakerEventAt(started.Add(time.Second), func(status *DeviceStatus) {
|
||||
status.Volume = &models.Volume{ActualVolume: 99}
|
||||
})
|
||||
conn.CompleteHTTPPoll(poll, true, started.Add(2*time.Second), func(status *DeviceStatus) {
|
||||
status.Volume = &models.Volume{ActualVolume: 42}
|
||||
})
|
||||
|
||||
status := conn.Status()
|
||||
if status.Volume == nil || status.Volume.ActualVolume != 99 {
|
||||
t.Fatalf("speaker event was overwritten by older poll data: %+v", status.Volume)
|
||||
}
|
||||
if status.Connectivity != ConnectivityOnline || !status.HTTPReachable || !status.IsConnected {
|
||||
t.Fatalf("poll health was not retained: %+v", status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOlderHTTPPollCannotOverwriteNewerSuccess(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
older := conn.BeginHTTPPoll()
|
||||
newer := conn.BeginHTTPPoll()
|
||||
if !conn.CompleteHTTPPoll(newer, true, started.Add(time.Second), nil) {
|
||||
t.Fatal("newer successful poll was unexpectedly rejected")
|
||||
}
|
||||
if conn.CompleteHTTPPoll(older, false, started.Add(2*time.Second), nil) {
|
||||
t.Fatal("older failed poll was unexpectedly accepted")
|
||||
}
|
||||
|
||||
status := conn.Status()
|
||||
if status.Connectivity != ConnectivityOnline || !status.HTTPReachable || !status.IsConnected {
|
||||
t.Fatalf("older poll overwrote newer success: %+v", status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOlderTransportCallbackCannotOverwriteNewerGeneration(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
if !conn.ObserveEventStreamTransport(3, true, started) {
|
||||
t.Fatal("newer connected transport generation was unexpectedly rejected")
|
||||
}
|
||||
conn.ApplySpeakerEventAt(started.Add(time.Second), func(status *DeviceStatus) {
|
||||
status.Volume = &models.Volume{ActualVolume: 42}
|
||||
})
|
||||
if conn.ObserveEventStreamTransport(2, false, started.Add(2*time.Second)) {
|
||||
t.Fatal("older disconnected transport callback was unexpectedly accepted")
|
||||
}
|
||||
|
||||
status := conn.Status()
|
||||
if status.Connectivity != ConnectivityOnline || !status.WebSocketConnected || !status.IsConnected {
|
||||
t.Fatalf("older transport callback demoted newer success: %+v", status)
|
||||
}
|
||||
if status.Volume == nil || status.Volume.ActualVolume != 42 {
|
||||
t.Fatalf("speaker event payload was lost: %+v", status.Volume)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInitialHTTPFailureStaysOfflineWithoutPriorSuccess(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
started := time.Date(2026, time.August, 28, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
first := conn.BeginHTTPPoll()
|
||||
conn.CompleteHTTPPoll(first, false, started, nil)
|
||||
if status := conn.Status(); status.Connectivity != ConnectivityOffline || status.IsConnected {
|
||||
t.Fatalf("first initial failure = %+v", status)
|
||||
}
|
||||
|
||||
second := conn.BeginHTTPPoll()
|
||||
conn.CompleteHTTPPoll(second, false, started.Add(time.Second), nil)
|
||||
if status := conn.Status(); status.Connectivity != ConnectivityOffline || status.IsConnected {
|
||||
t.Fatalf("second initial failure = %+v", status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWebSocketLoopHasSingleOwner(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
|
||||
if !conn.TryStartWebSocketLoop() {
|
||||
t.Fatal("first supervisor did not acquire ownership")
|
||||
}
|
||||
if conn.TryStartWebSocketLoop() {
|
||||
t.Fatal("second supervisor acquired duplicate ownership")
|
||||
}
|
||||
|
||||
conn.FinishWebSocketLoop()
|
||||
if !conn.TryStartWebSocketLoop() {
|
||||
t.Fatal("supervisor ownership was not released")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetStatus_ReplacesEntireStatus(t *testing.T) {
|
||||
conn := NewDeviceConnection(nil, &models.DeviceInfo{Name: "test"})
|
||||
conn.SetStatus(&DeviceStatus{
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
package webtypes
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
@@ -40,12 +41,40 @@ type SoundTouchClient interface {
|
||||
// / UpdateStatus rather than the private field; construct connections
|
||||
// via NewDeviceConnection to guarantee the status is initialised.
|
||||
type DeviceConnection struct {
|
||||
Client *client.Client
|
||||
WebSocket *client.WebSocketClient
|
||||
Client *client.Client
|
||||
// WebSocket is retained for source compatibility with callers that build
|
||||
// DeviceConnection values directly. Concurrent code must use
|
||||
// CurrentWebSocket and SetWebSocket.
|
||||
WebSocket *client.WebSocketClient
|
||||
// DeviceInfo is the immutable discovery snapshot. Use Info for player-facing
|
||||
// output so later nameUpdated events are reflected without racing readers.
|
||||
DeviceInfo *models.DeviceInfo
|
||||
LastSeen time.Time
|
||||
|
||||
status atomic.Pointer[DeviceStatus]
|
||||
deviceName atomic.Pointer[string]
|
||||
status atomic.Pointer[DeviceStatus]
|
||||
|
||||
webSocketMu sync.RWMutex
|
||||
webSocketLoopRunning atomic.Bool
|
||||
|
||||
nameMu sync.Mutex
|
||||
nameGen uint64
|
||||
|
||||
healthMu sync.Mutex
|
||||
|
||||
nextPollGeneration uint64
|
||||
lastPollGeneration uint64
|
||||
httpReachable bool
|
||||
consecutiveFailures int
|
||||
speakerEventGen uint64
|
||||
pollEventGen map[uint64]uint64
|
||||
|
||||
lastTransportGeneration uint64
|
||||
eventStreamConnected bool
|
||||
lastDirectSuccess time.Time
|
||||
speakerConnectionKnown bool
|
||||
speakerConnectionConnected bool
|
||||
speakerConnectionObserved time.Time
|
||||
|
||||
// fieldGenMu guards fieldGen, the per-field generation ordering used by
|
||||
// BeginFieldPoll/CompleteFieldPoll/ApplyFieldEvent. Each StatusField gets
|
||||
@@ -78,14 +107,40 @@ type DeviceConnection struct {
|
||||
|
||||
// DeviceStatus represents the current device state
|
||||
type DeviceStatus struct {
|
||||
NowPlaying *models.NowPlaying `json:"nowPlaying,omitempty"`
|
||||
Volume *models.Volume `json:"volume,omitempty"`
|
||||
Presets *models.Presets `json:"presets,omitempty"`
|
||||
Sources *models.Sources `json:"sources,omitempty"`
|
||||
Bass *models.Bass `json:"bass,omitempty"`
|
||||
Group *models.Group `json:"group,omitempty"`
|
||||
IsConnected bool `json:"isConnected"`
|
||||
LastActivity time.Time `json:"lastActivity"`
|
||||
NowPlaying *models.NowPlaying `json:"nowPlaying,omitempty"`
|
||||
Volume *models.Volume `json:"volume,omitempty"`
|
||||
Presets *models.Presets `json:"presets,omitempty"`
|
||||
Sources *models.Sources `json:"sources,omitempty"`
|
||||
Bass *models.Bass `json:"bass,omitempty"`
|
||||
Group *models.Group `json:"group,omitempty"`
|
||||
Connectivity Connectivity `json:"connectivity"`
|
||||
HTTPReachable bool `json:"httpReachable"`
|
||||
WebSocketConnected bool `json:"webSocketConnected"`
|
||||
SpeakerConnectionState *SpeakerConnectionState `json:"speakerConnectionState,omitempty"`
|
||||
IsConnected bool `json:"isConnected"`
|
||||
LastActivity time.Time `json:"lastActivity"`
|
||||
}
|
||||
|
||||
// Connectivity is the player's aggregate view of HTTP and event-stream
|
||||
// reachability. Speaker-reported connection state is retained as supporting
|
||||
// evidence but cannot override a current direct-path success.
|
||||
type Connectivity string
|
||||
|
||||
const (
|
||||
ConnectivityOnline Connectivity = "online"
|
||||
ConnectivityStale Connectivity = "stale"
|
||||
ConnectivityOffline Connectivity = "offline"
|
||||
)
|
||||
|
||||
const (
|
||||
offlineFailureThreshold = 2
|
||||
offlineGracePeriod = 60 * time.Second
|
||||
)
|
||||
|
||||
// SpeakerConnectionState is the network state reported by the speaker.
|
||||
type SpeakerConnectionState struct {
|
||||
State string `json:"state"`
|
||||
Signal string `json:"signal,omitempty"`
|
||||
}
|
||||
|
||||
// StatusField identifies one independently-racing field of DeviceStatus for
|
||||
@@ -120,13 +175,71 @@ func NewDeviceConnection(c *client.Client, info *models.DeviceInfo) *DeviceConne
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
conn.status.Store(&DeviceStatus{
|
||||
Connectivity: ConnectivityOffline,
|
||||
IsConnected: false,
|
||||
LastActivity: time.Now(),
|
||||
})
|
||||
if info != nil {
|
||||
conn.storeDeviceName(info.Name)
|
||||
}
|
||||
|
||||
return conn
|
||||
}
|
||||
|
||||
func (c *DeviceConnection) storeDeviceName(name string) {
|
||||
c.deviceName.Store(&name)
|
||||
}
|
||||
|
||||
// BeginNameRefresh starts a generation for an asynchronous /name request.
|
||||
func (c *DeviceConnection) BeginNameRefresh() uint64 {
|
||||
c.nameMu.Lock()
|
||||
defer c.nameMu.Unlock()
|
||||
|
||||
c.nameGen++
|
||||
|
||||
return c.nameGen
|
||||
}
|
||||
|
||||
// ApplyPolledName stores a /name result unless a newer poll or event won.
|
||||
func (c *DeviceConnection) ApplyPolledName(generation uint64, name string) bool {
|
||||
c.nameMu.Lock()
|
||||
defer c.nameMu.Unlock()
|
||||
|
||||
if generation != c.nameGen {
|
||||
return false
|
||||
}
|
||||
|
||||
c.storeDeviceName(name)
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// ApplyNameEvent stores a nameUpdated event and invalidates in-flight polls.
|
||||
func (c *DeviceConnection) ApplyNameEvent(name string) {
|
||||
c.nameMu.Lock()
|
||||
defer c.nameMu.Unlock()
|
||||
|
||||
c.nameGen++
|
||||
c.storeDeviceName(name)
|
||||
}
|
||||
|
||||
// Info returns a read-only metadata snapshot with the latest device name.
|
||||
func (c *DeviceConnection) Info() *models.DeviceInfo {
|
||||
if c.DeviceInfo == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
name := c.deviceName.Load()
|
||||
if name == nil || *name == c.DeviceInfo.Name {
|
||||
return c.DeviceInfo
|
||||
}
|
||||
|
||||
info := *c.DeviceInfo
|
||||
info.Name = *name
|
||||
|
||||
return &info
|
||||
}
|
||||
|
||||
// Status returns a snapshot of the current device status. The returned
|
||||
// pointer is read-only from the caller's perspective and must not be
|
||||
// mutated. Use UpdateStatus or SetStatus to apply changes. Never returns
|
||||
@@ -151,12 +264,70 @@ func (c *DeviceConnection) Close() {
|
||||
c.closeOnce.Do(func() {
|
||||
close(c.done)
|
||||
|
||||
if c.WebSocket != nil {
|
||||
_ = c.WebSocket.Disconnect()
|
||||
c.webSocketMu.Lock()
|
||||
ws := c.WebSocket
|
||||
c.WebSocket = nil
|
||||
c.webSocketMu.Unlock()
|
||||
|
||||
if ws != nil {
|
||||
_ = ws.Close()
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// CurrentWebSocket returns the current device event transport, if any.
|
||||
func (c *DeviceConnection) CurrentWebSocket() *client.WebSocketClient {
|
||||
c.webSocketMu.RLock()
|
||||
defer c.webSocketMu.RUnlock()
|
||||
|
||||
return c.WebSocket
|
||||
}
|
||||
|
||||
// SetWebSocket publishes a device event transport unless the connection was
|
||||
// already closed. A transport that loses that race is stopped immediately.
|
||||
func (c *DeviceConnection) SetWebSocket(ws *client.WebSocketClient) bool {
|
||||
c.webSocketMu.Lock()
|
||||
select {
|
||||
case <-c.done:
|
||||
c.webSocketMu.Unlock()
|
||||
if ws != nil {
|
||||
_ = ws.Close()
|
||||
}
|
||||
|
||||
return false
|
||||
default:
|
||||
c.WebSocket = ws
|
||||
c.webSocketMu.Unlock()
|
||||
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
// ClearWebSocket clears only the expected transport, preserving a replacement
|
||||
// that may already have been published under the same device entry.
|
||||
func (c *DeviceConnection) ClearWebSocket(expected *client.WebSocketClient) bool {
|
||||
c.webSocketMu.Lock()
|
||||
defer c.webSocketMu.Unlock()
|
||||
|
||||
if c.WebSocket != expected {
|
||||
return false
|
||||
}
|
||||
|
||||
c.WebSocket = nil
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// TryStartWebSocketLoop claims the single event supervisor for this device.
|
||||
func (c *DeviceConnection) TryStartWebSocketLoop() bool {
|
||||
return c.webSocketLoopRunning.CompareAndSwap(false, true)
|
||||
}
|
||||
|
||||
// FinishWebSocketLoop releases event-supervisor ownership.
|
||||
func (c *DeviceConnection) FinishWebSocketLoop() {
|
||||
c.webSocketLoopRunning.Store(false)
|
||||
}
|
||||
|
||||
// SetStatus atomically replaces the entire status. Use sparingly —
|
||||
// UpdateStatus is the preferred entry point because it preserves
|
||||
// concurrent changes from other goroutines.
|
||||
@@ -235,6 +406,214 @@ func (c *DeviceConnection) UpdateStatus(mut func(*DeviceStatus)) {
|
||||
}
|
||||
}
|
||||
|
||||
// BeginHTTPPoll reserves an ordering generation for a status poll.
|
||||
func (c *DeviceConnection) BeginHTTPPoll() uint64 {
|
||||
c.healthMu.Lock()
|
||||
defer c.healthMu.Unlock()
|
||||
|
||||
c.nextPollGeneration++
|
||||
if c.pollEventGen == nil {
|
||||
c.pollEventGen = make(map[uint64]uint64)
|
||||
}
|
||||
c.pollEventGen[c.nextPollGeneration] = c.speakerEventGen
|
||||
|
||||
return c.nextPollGeneration
|
||||
}
|
||||
|
||||
// ApplySpeakerEventAt applies event payload and records a live event stream.
|
||||
func (c *DeviceConnection) ApplySpeakerEventAt(at time.Time, mut func(*DeviceStatus)) {
|
||||
c.healthMu.Lock()
|
||||
defer c.healthMu.Unlock()
|
||||
|
||||
c.speakerEventGen++
|
||||
c.markEventStreamActivityLocked(at)
|
||||
c.UpdateStatus(func(status *DeviceStatus) {
|
||||
if mut != nil {
|
||||
mut(status)
|
||||
}
|
||||
c.applyConnectivityLocked(status, at)
|
||||
})
|
||||
}
|
||||
|
||||
// ApplySpeakerConnectionEvent stores speaker-reported diagnostic state. The
|
||||
// event itself proves the event stream is live.
|
||||
func (c *DeviceConnection) ApplySpeakerConnectionEvent(state SpeakerConnectionState, at time.Time) {
|
||||
c.healthMu.Lock()
|
||||
defer c.healthMu.Unlock()
|
||||
|
||||
c.speakerEventGen++
|
||||
c.markEventStreamActivityLocked(at)
|
||||
c.speakerConnectionObserved = at
|
||||
|
||||
switch strings.ToUpper(strings.TrimSpace(state.State)) {
|
||||
case string(models.ConnectionStateConnected):
|
||||
c.speakerConnectionKnown = true
|
||||
c.speakerConnectionConnected = true
|
||||
case string(models.ConnectionStateDisconnected):
|
||||
c.speakerConnectionKnown = true
|
||||
c.speakerConnectionConnected = false
|
||||
default:
|
||||
c.speakerConnectionKnown = false
|
||||
c.speakerConnectionConnected = false
|
||||
}
|
||||
|
||||
c.UpdateStatus(func(status *DeviceStatus) {
|
||||
reported := state
|
||||
status.SpeakerConnectionState = &reported
|
||||
status.LastActivity = at
|
||||
c.applyConnectivityLocked(status, at)
|
||||
})
|
||||
}
|
||||
|
||||
// ObserveEventStreamTransport applies an authoritative client transport
|
||||
// transition. Its generation originates in WebSocketClient, so reordered
|
||||
// callbacks cannot overwrite a newer transport state.
|
||||
func (c *DeviceConnection) ObserveEventStreamTransport(
|
||||
generation uint64,
|
||||
connected bool,
|
||||
at time.Time,
|
||||
) bool {
|
||||
c.healthMu.Lock()
|
||||
defer c.healthMu.Unlock()
|
||||
|
||||
if generation <= c.lastTransportGeneration {
|
||||
return false
|
||||
}
|
||||
|
||||
c.lastTransportGeneration = generation
|
||||
c.eventStreamConnected = connected
|
||||
if connected {
|
||||
c.recordDirectSuccessLocked(at)
|
||||
}
|
||||
c.UpdateStatus(func(status *DeviceStatus) {
|
||||
c.applyConnectivityLocked(status, at)
|
||||
})
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// ObserveEventStream records a transition when no transport generation is
|
||||
// available, primarily for non-client callers and deterministic tests.
|
||||
func (c *DeviceConnection) ObserveEventStream(connected bool, at time.Time) {
|
||||
c.healthMu.Lock()
|
||||
defer c.healthMu.Unlock()
|
||||
|
||||
c.eventStreamConnected = connected
|
||||
if connected {
|
||||
c.recordDirectSuccessLocked(at)
|
||||
}
|
||||
c.UpdateStatus(func(status *DeviceStatus) {
|
||||
c.applyConnectivityLocked(status, at)
|
||||
})
|
||||
}
|
||||
|
||||
// MarkEventStreamActivity records an event handled by a field-specific path.
|
||||
func (c *DeviceConnection) MarkEventStreamActivity(at time.Time) {
|
||||
c.ObserveEventStream(true, at)
|
||||
}
|
||||
|
||||
// CompleteHTTPPoll records health and merges payload only if no newer poll or
|
||||
// speaker event superseded it.
|
||||
func (c *DeviceConnection) CompleteHTTPPoll(
|
||||
generation uint64,
|
||||
success bool,
|
||||
at time.Time,
|
||||
merge func(*DeviceStatus),
|
||||
) bool {
|
||||
c.healthMu.Lock()
|
||||
defer c.healthMu.Unlock()
|
||||
|
||||
pollEventGeneration, knownGeneration := c.pollEventGen[generation]
|
||||
delete(c.pollEventGen, generation)
|
||||
|
||||
if generation <= c.lastPollGeneration {
|
||||
return false
|
||||
}
|
||||
|
||||
c.lastPollGeneration = generation
|
||||
for olderGeneration := range c.pollEventGen {
|
||||
if olderGeneration < generation {
|
||||
delete(c.pollEventGen, olderGeneration)
|
||||
}
|
||||
}
|
||||
|
||||
if success {
|
||||
c.httpReachable = true
|
||||
c.consecutiveFailures = 0
|
||||
c.recordDirectSuccessLocked(at)
|
||||
} else {
|
||||
c.httpReachable = false
|
||||
c.consecutiveFailures++
|
||||
}
|
||||
|
||||
c.UpdateStatus(func(status *DeviceStatus) {
|
||||
if merge != nil && knownGeneration && pollEventGeneration == c.speakerEventGen {
|
||||
merge(status)
|
||||
}
|
||||
c.applyConnectivityLocked(status, at)
|
||||
if success {
|
||||
status.LastActivity = at
|
||||
}
|
||||
})
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
func (c *DeviceConnection) markEventStreamActivityLocked(at time.Time) {
|
||||
c.eventStreamConnected = true
|
||||
c.recordDirectSuccessLocked(at)
|
||||
}
|
||||
|
||||
func (c *DeviceConnection) recordDirectSuccessLocked(at time.Time) {
|
||||
if c.lastDirectSuccess.IsZero() || at.After(c.lastDirectSuccess) {
|
||||
c.lastDirectSuccess = at
|
||||
}
|
||||
}
|
||||
|
||||
func (c *DeviceConnection) applyConnectivityLocked(status *DeviceStatus, at time.Time) {
|
||||
connectivity := c.connectivityLocked(at)
|
||||
status.Connectivity = connectivity
|
||||
status.HTTPReachable = c.httpReachable
|
||||
status.WebSocketConnected = c.eventStreamConnected
|
||||
status.IsConnected = connectivity != ConnectivityOffline
|
||||
}
|
||||
|
||||
func (c *DeviceConnection) connectivityLocked(at time.Time) Connectivity {
|
||||
if c.httpReachable || c.eventStreamConnected {
|
||||
return ConnectivityOnline
|
||||
}
|
||||
|
||||
if c.speakerConnectionKnown && c.speakerConnectionConnected &&
|
||||
withinConnectivityGrace(at, c.speakerConnectionObserved) {
|
||||
return ConnectivityStale
|
||||
}
|
||||
|
||||
if !c.lastDirectSuccess.IsZero() &&
|
||||
(c.consecutiveFailures < offlineFailureThreshold ||
|
||||
withinConnectivityGrace(at, c.lastDirectSuccess)) {
|
||||
return ConnectivityStale
|
||||
}
|
||||
|
||||
return ConnectivityOffline
|
||||
}
|
||||
|
||||
func withinConnectivityGrace(at, success time.Time) bool {
|
||||
if success.IsZero() {
|
||||
return false
|
||||
}
|
||||
if at.Before(success) {
|
||||
return true
|
||||
}
|
||||
|
||||
return at.Sub(success) < offlineGracePeriod
|
||||
}
|
||||
|
||||
// MarkHTTPSuccess records a successful out-of-band request such as /info.
|
||||
func (c *DeviceConnection) MarkHTTPSuccess(at time.Time) {
|
||||
generation := c.BeginHTTPPoll()
|
||||
c.CompleteHTTPPoll(generation, true, at, nil)
|
||||
}
|
||||
|
||||
// BeginGroupRefresh starts a new generation for an asynchronous /getGroup
|
||||
// request. Only the latest started request may later update Group.
|
||||
func (c *DeviceConnection) BeginGroupRefresh() uint64 {
|
||||
|
||||
Reference in New Issue
Block a user