diff --git a/cmd/client/main.go b/cmd/client/main.go index caee446..9653683 100644 --- a/cmd/client/main.go +++ b/cmd/client/main.go @@ -5,21 +5,24 @@ import ( "flag" "fmt" "log" + "net/http" "os" "os/signal" "path/filepath" "strings" + "sync" "syscall" "time" "github.com/Sendspin/sendspin-go/pkg/discovery" "github.com/Sendspin/sendspin-go/pkg/sendspin" + "github.com/gorilla/websocket" "rpi-sendspin/internal/pipewire" ) func main() { - server := flag.String("server", "localhost:8927", "server address") + server := flag.String("server", "", "server address for client-initiated mode (dial out). Leave empty for server-initiated mode: advertise via mDNS and wait for the server to connect.") name := flag.String("name", hostname(), "player name") device := flag.String("device", "", "PipeWire sink node name (default: system default)") listDevices := flag.Bool("list-devices", false, "list available PipeWire sinks and exit") @@ -119,23 +122,109 @@ func main() { } else { log.Printf("output: PipeWire default sink") } - log.Printf("connecting to %s as %q...", *server, *name) - - if err := player.Connect(); err != nil { - log.Fatalf("connect: %v", err) - } - if err := player.Play(); err != nil { - log.Fatalf("play: %v", err) - } sig := make(chan os.Signal, 1) signal.Notify(sig, os.Interrupt, syscall.SIGTERM) - <-sig + + if *server == "" { + // Server-initiated mode: the server discovers us via mDNS and dials + // our /sendspin endpoint. We just listen and let player.Accept drive + // each connection. + if *noMDNS { + log.Printf("warning: --no-mdns with no --server — the server cannot discover this player") + } + runServerInitiated(player, *mdnsPort, sig) + } else { + // Client-initiated mode: dial the configured server. + log.Printf("connecting to %s as %q...", *server, *name) + if err := player.Connect(); err != nil { + log.Fatalf("connect: %v", err) + } + if err := player.Play(); err != nil { + log.Fatalf("play: %v", err) + } + <-sig + } log.Println("stopping") player.Stop() } +// runServerInitiated runs an HTTP/WebSocket server on the mDNS-advertised +// port and hands each accepted /sendspin connection to player.Accept. Only +// one session runs at a time; a newer connection evicts the current one +// (newest-wins) so a server that reconnects isn't blocked behind a stale +// socket. Blocks until a signal arrives on sig. +func runServerInitiated(player *sendspin.Player, port int, sig <-chan os.Signal) { + upgrader := websocket.Upgrader{ + CheckOrigin: func(*http.Request) bool { return true }, + } + + var ( + mu sync.Mutex // serializes sessions; held for a connection's lifetime + connMu sync.Mutex // guards curConn + curConn *websocket.Conn + ) + + handler := func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + log.Printf("websocket upgrade failed: %v", err) + return + } + + // Evict any in-flight session so its Accept returns and releases mu. + connMu.Lock() + if curConn != nil { + curConn.Close() + } + curConn = conn + connMu.Unlock() + + mu.Lock() + defer mu.Unlock() + + // We may have been superseded while waiting for the lock. + connMu.Lock() + superseded := curConn != conn + connMu.Unlock() + if superseded { + conn.Close() + return + } + + if err := player.Accept(conn); err != nil { + log.Printf("session ended: %v", err) + } + + connMu.Lock() + if curConn == conn { + curConn = nil + } + connMu.Unlock() + } + + mux := http.NewServeMux() + mux.HandleFunc("/sendspin", handler) + srv := &http.Server{Addr: fmt.Sprintf(":%d", port), Handler: mux} + + go func() { + log.Printf("server-initiated mode: listening on :%d/sendspin, waiting for a server to connect", port) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("listen: %v", err) + } + }() + + <-sig + + // Drop the active session first so the blocked handler returns, then let + // Shutdown drain the (now-unblocked) server. + player.CloseConnection() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + srv.Shutdown(ctx) +} + func hostname() string { h, _ := os.Hostname() if h == "" { @@ -154,11 +243,20 @@ func hostname() string { // during reconnect could in theory move the player to a different server, // but for the same-machine echo-cancel use case that's acceptable. func buildMuter(server, muteServer, muteLoopback string) *pipewire.Muter { - if muteLoopback == "" && muteServer == "" { + if muteLoopback == "" { + if muteServer != "" { + log.Printf("muter: --mute-server set without --mute-loopback; loopback mute disabled") + } return nil } - if muteLoopback == "" || muteServer == "" { - log.Printf("muter: --mute-loopback and --mute-server must be set together; loopback mute disabled") + // Server-initiated (listen) mode has no fixed --server to compare + // against, so arm the muter purely on --mute-loopback. + if server == "" { + log.Printf("muter: will mute PipeWire node %q while streaming (server-initiated mode)", muteLoopback) + return pipewire.NewMuter(muteLoopback) + } + if muteServer == "" { + log.Printf("muter: --mute-loopback and --mute-server must be set together in client-initiated mode; loopback mute disabled") return nil } if !sameServerAddr(server, muteServer) { diff --git a/contrib/systemd/sendspin-client.service b/contrib/systemd/sendspin-client.service index aad7841..71647e4 100644 --- a/contrib/systemd/sendspin-client.service +++ b/contrib/systemd/sendspin-client.service @@ -4,8 +4,10 @@ After=pipewire.service network-online.target Wants=pipewire.service [Service] +# No --server: server-initiated mode. The player advertises via mDNS and +# waits for Music Assistant to discover and connect to it. Pass --server +# host:port only to force the old client-initiated (dial-out) behavior. ExecStart=%h/.local/bin/sendspin-client \ - --server 192.168.1.100:8927 \ --name %H \ --device alsa_output.usb-Device-00.analog-stereo \ --volume 80 diff --git a/third_party/sendspin-go/pkg/sendspin/player.go b/third_party/sendspin-go/pkg/sendspin/player.go index 6e5f399..8978080 100644 --- a/third_party/sendspin-go/pkg/sendspin/player.go +++ b/third_party/sendspin-go/pkg/sendspin/player.go @@ -14,6 +14,7 @@ import ( "github.com/Sendspin/sendspin-go/pkg/audio/output" "github.com/Sendspin/sendspin-go/pkg/protocol" "github.com/Sendspin/sendspin-go/pkg/sync" + "github.com/gorilla/websocket" ) // ReconnectConfig controls automatic reconnect behavior after the protocol @@ -216,6 +217,55 @@ func (p *Player) Connect() error { return nil } +// Accept drives the player over a server-initiated WebSocket connection +// (the server discovered this player via mDNS and dialed it). It builds a +// receiver over the accepted conn, starts audio consumption, and BLOCKS +// until the connection drops or the player is closed — so callers running a +// listener can serialize one session at a time and re-accept on return. +// +// Unlike Connect, Accept does not start the dial-based reconnect loop: in +// server-initiated mode the server redials, so reconnection is the listener's +// responsibility, not the player's. The caller transfers ownership of conn. +func (p *Player) Accept(conn *websocket.Conn) error { + recv, err := p.buildReceiver("server-initiated") + if err != nil { + return err + } + if err := recv.AcceptConn(conn); err != nil { + return err + } + + p.receiver = recv + p.state.Connected = true + p.notifyStateChange() + + go p.consumeAudio(recv) + + // Block for the lifetime of this connection so the caller's accept loop + // serializes sessions (one server connection at a time). + select { + case <-recv.Done(): + case <-p.ctx.Done(): + } + + recv.Close() + p.state.Connected = false + p.state.State = "idle" + p.notifyStateChange() + + return nil +} + +// CloseConnection drops the currently active receiver, if any, without +// tearing down the Player or its audio output. In server-initiated mode a +// listener calls this to evict a stale/superseded session before accepting +// a newer connection; the blocked Accept for the old session then returns. +func (p *Player) CloseConnection() { + if p.receiver != nil { + p.receiver.Close() + } +} + func (p *Player) buildReceiver(addr string) (*Receiver, error) { p.ensureCapsResolved() return NewReceiver(ReceiverConfig{ diff --git a/third_party/sendspin-go/pkg/sendspin/receiver.go b/third_party/sendspin-go/pkg/sendspin/receiver.go index faf5c78..918947a 100644 --- a/third_party/sendspin-go/pkg/sendspin/receiver.go +++ b/third_party/sendspin-go/pkg/sendspin/receiver.go @@ -17,6 +17,7 @@ import ( "github.com/Sendspin/sendspin-go/pkg/audio/decode" "github.com/Sendspin/sendspin-go/pkg/protocol" "github.com/Sendspin/sendspin-go/pkg/sync" + "github.com/gorilla/websocket" ) // Time-sync burst parameters. Mirrors sendspin-cpp's TimeBurst defaults and @@ -109,6 +110,12 @@ type Receiver struct { // from r.clockSync.ServerMicrosNow so tests can drive the // timestamp-deferral path with a fake clock. clockNow func() int64 + + // closeOnce guards Close so it is idempotent: in server-initiated mode + // a listener may evict a session via CloseConnection (which closes the + // receiver) while Accept's own deferred Close also fires. Without this + // the second close(r.output) panics. + closeOnce stdsync.Once } // NewReceiver creates a new Receiver with the given configuration. @@ -189,17 +196,59 @@ func (r *Receiver) Stats() ReceiverStats { return stats } -// Connect establishes a connection to the server, performs initial clock sync, -// and starts background goroutines for connection watching and clock sync. +// Connect establishes a client-initiated connection to the server (dials +// ServerAddr), performs initial clock sync, and starts background goroutines +// for connection watching and clock sync. func (r *Receiver) Connect() error { + clientConfig, err := r.buildClientConfig() + if err != nil { + return err + } + + r.client = protocol.NewClient(clientConfig) + + if err := r.client.Connect(); err != nil { + return fmt.Errorf("connection failed: %w", err) + } + + log.Printf("Connected to server: %s", r.serverAddr) + return r.startSession() +} + +// AcceptConn drives the receiver protocol over an already-established +// WebSocket connection (server-initiated mode: the server discovered this +// player via mDNS and dialed it). The client still sends client/hello first +// per spec, so the handshake and message loop are identical to Connect once +// the socket exists. The caller transfers ownership of conn; Close() (or a +// dropped connection) closes it. +func (r *Receiver) AcceptConn(conn *websocket.Conn) error { + clientConfig, err := r.buildClientConfig() + if err != nil { + return err + } + + r.client = protocol.NewClientFromConn(clientConfig, conn) + + if err := r.client.Start(); err != nil { + return fmt.Errorf("handshake failed: %w", err) + } + + log.Printf("Accepted server-initiated connection from %s", conn.RemoteAddr()) + return r.startSession() +} + +// buildClientConfig validates required fields and assembles the protocol +// client config (advertised formats, device info, role support) shared by +// both the client-initiated (Connect) and server-initiated (AcceptConn) paths. +func (r *Receiver) buildClientConfig() (protocol.Config, error) { if r.config.ClientID == "" { - return fmt.Errorf("ReceiverConfig.ClientID is required (resolve via sendspin.ResolveClientID)") + return protocol.Config{}, fmt.Errorf("ReceiverConfig.ClientID is required (resolve via sendspin.ResolveClientID)") } supportedFormats := buildSupportedFormats(r.config.PreferredCodec, r.config.MaxSampleRate, r.config.MaxBitDepth) logAdvertisedFormats(supportedFormats, r.config.MaxSampleRate, r.config.MaxBitDepth) - clientConfig := protocol.Config{ + return protocol.Config{ ServerAddr: r.serverAddr, ClientID: r.config.ClientID, Name: r.config.PlayerName, @@ -222,15 +271,13 @@ func (r *Receiver) Connect() error { VisualizerV1Support: &protocol.VisualizerV1Support{ BufferCapacity: r.config.BufferCapacity, }, - } + }, nil +} - r.client = protocol.NewClient(clientConfig) - - if err := r.client.Connect(); err != nil { - return fmt.Errorf("connection failed: %w", err) - } - - log.Printf("Connected to server: %s", r.serverAddr) +// startSession marks the receiver connected, runs the initial clock-sync +// burst, and launches the background goroutines. Shared by Connect and +// AcceptConn; r.client must already be started. +func (r *Receiver) startSession() error { r.connected = true if err := r.performInitialSync(); err != nil { @@ -795,29 +842,31 @@ func (r *Receiver) notifyError(err error) { } func (r *Receiver) Close() error { - // Send goodbye BEFORE cancelling the context so the message - // reaches the server while the connection is still alive. - if r.client != nil { - r.client.SendGoodbye("shutdown") - } - - r.cancel() - - if r.client != nil { - r.client.Close() - } - - if r.scheduler != nil { - r.scheduler.Stop() - } - - if r.decoder != nil { - if err := r.decoder.Close(); err != nil { - log.Printf("Receiver: decoder close error: %v", err) + r.closeOnce.Do(func() { + // Send goodbye BEFORE cancelling the context so the message + // reaches the server while the connection is still alive. + if r.client != nil { + r.client.SendGoodbye("shutdown") } - } - close(r.output) + r.cancel() + + if r.client != nil { + r.client.Close() + } + + if r.scheduler != nil { + r.scheduler.Stop() + } + + if r.decoder != nil { + if err := r.decoder.Close(); err != nil { + log.Printf("Receiver: decoder close error: %v", err) + } + } + + close(r.output) + }) return nil }