diff --git a/.dockerignore b/.dockerignore
new file mode 100644
index 0000000..407dff4
--- /dev/null
+++ b/.dockerignore
@@ -0,0 +1,5 @@
+.gitignore
+bin/*
+/deploy.sh
+debugging-session/**
+test
\ No newline at end of file
diff --git a/.gitignore b/.gitignore
index 24e091b..bac73dd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,4 @@
bin/*
-/deploy.sh
\ No newline at end of file
+/deploy.sh
+debugging-session
+test
\ No newline at end of file
diff --git a/cmd/client/main.go b/cmd/client/main.go
index 7a7f83e..caee446 100644
--- a/cmd/client/main.go
+++ b/cmd/client/main.go
@@ -1,15 +1,18 @@
package main
import (
+ "context"
"flag"
"fmt"
"log"
"os"
"os/signal"
"path/filepath"
+ "strings"
"syscall"
"time"
+ "github.com/Sendspin/sendspin-go/pkg/discovery"
"github.com/Sendspin/sendspin-go/pkg/sendspin"
"rpi-sendspin/internal/pipewire"
@@ -21,6 +24,10 @@ func main() {
device := flag.String("device", "", "PipeWire sink node name (default: system default)")
listDevices := flag.Bool("list-devices", false, "list available PipeWire sinks and exit")
volume := flag.Int("volume", 80, "initial volume (0-100)")
+ mdnsPort := flag.Int("mdns-port", 8928, "port to advertise in mDNS (_sendspin._tcp)")
+ noMDNS := flag.Bool("no-mdns", false, "disable mDNS advertisement")
+ muteLoopback := flag.String("mute-loopback", "", "PipeWire node.name of a loopback to mute while streaming from --mute-server (requires --mute-server)")
+ muteServer := flag.String("mute-server", "", "server address that arms loopback muting; must match --server (requires --mute-loopback)")
flag.Parse()
if *listDevices {
@@ -51,8 +58,22 @@ func main() {
log.Fatalf("resolve client ID: %v", err)
}
+ if !*noMDNS {
+ disc := discovery.NewManager(discovery.Config{
+ ServiceName: *name,
+ Port: *mdnsPort,
+ })
+ if err := disc.Advertise(); err != nil {
+ log.Printf("mdns advertise: %v", err)
+ } else {
+ defer disc.Stop()
+ }
+ }
+
output := pipewire.NewOutput(*device, *name)
+ muter := buildMuter(*server, *muteServer, *muteLoopback)
+
player, err := sendspin.NewPlayer(sendspin.PlayerConfig{
ServerAddr: *server,
PlayerName: *name,
@@ -70,6 +91,11 @@ func main() {
log.Printf("now playing: %s — %s", meta.Artist, meta.Title)
}
},
+ OnStateChange: func(s sendspin.PlayerState) {
+ if muter != nil {
+ muter.SetDesiredMute(s.State == "playing")
+ }
+ },
OnError: func(err error) {
log.Printf("error: %v", err)
},
@@ -77,6 +103,15 @@ func main() {
if err != nil {
log.Fatalf("create player: %v", err)
}
+ // LIFO: muter.Stop runs AFTER player.Close, so the player has stopped
+ // emitting "playing" state by the time we force the final unmute.
+ if muter != nil {
+ defer func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ muter.Stop(ctx)
+ }()
+ }
defer player.Close()
if *device != "" {
@@ -109,6 +144,45 @@ func hostname() string {
return h
}
+// buildMuter wires up the loopback muter when both --mute-loopback and
+// --mute-server are set AND --mute-server matches the configured --server.
+// Mismatches are intentional silent no-ops so the same config file can ship
+// to machines where the server happens to be different — only logged so
+// it's visible. Returns nil when muting is disabled.
+//
+// The address check is a startup-time string compare; mDNS rediscovery
+// 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 == "" {
+ return nil
+ }
+ if muteLoopback == "" || muteServer == "" {
+ log.Printf("muter: --mute-loopback and --mute-server must be set together; loopback mute disabled")
+ return nil
+ }
+ if !sameServerAddr(server, muteServer) {
+ log.Printf("muter: --mute-server %q != --server %q; loopback mute disabled", muteServer, server)
+ return nil
+ }
+ log.Printf("muter: will mute PipeWire node %q while streaming from %s", muteLoopback, server)
+ return pipewire.NewMuter(muteLoopback)
+}
+
+// sameServerAddr normalizes two host[:port] strings for equality so
+// "localhost" and "localhost:8927" compare equal when 8927 is the default.
+func sameServerAddr(a, b string) bool {
+ const defaultPort = "8927"
+ norm := func(s string) string {
+ s = strings.TrimSpace(strings.ToLower(s))
+ if !strings.Contains(s, ":") {
+ s += ":" + defaultPort
+ }
+ return s
+ }
+ return norm(a) == norm(b)
+}
+
func clientIDFile() string {
dir, err := os.UserConfigDir()
if err != nil {
diff --git a/contrib/systemd/sendspin-client.service b/contrib/systemd/sendspin-client.service
new file mode 100644
index 0000000..aad7841
--- /dev/null
+++ b/contrib/systemd/sendspin-client.service
@@ -0,0 +1,16 @@
+[Unit]
+Description=Sendspin audio player
+After=pipewire.service network-online.target
+Wants=pipewire.service
+
+[Service]
+ExecStart=%h/.local/bin/sendspin-client \
+ --server 192.168.1.100:8927 \
+ --name %H \
+ --device alsa_output.usb-Device-00.analog-stereo \
+ --volume 80
+Restart=on-failure
+RestartSec=5
+
+[Install]
+WantedBy=default.target
diff --git a/contrib/systemd/sendspin-live-server.service b/contrib/systemd/sendspin-live-server.service
new file mode 100644
index 0000000..bd4183d
--- /dev/null
+++ b/contrib/systemd/sendspin-live-server.service
@@ -0,0 +1,17 @@
+[Unit]
+Description=Sendspin live audio streaming server
+After=pipewire.service network-online.target
+Wants=pipewire.service
+
+[Service]
+ExecStart=%h/.local/bin/sendspin-live-server \
+ --name %H \
+ --port 8927 \
+ --http-port 8080 \
+ --rate 48000 \
+ --channels 2
+Restart=on-failure
+RestartSec=5
+
+[Install]
+WantedBy=default.target
diff --git a/contrib/systemd/sendspin-server.service b/contrib/systemd/sendspin-server.service
new file mode 100644
index 0000000..b894cb9
--- /dev/null
+++ b/contrib/systemd/sendspin-server.service
@@ -0,0 +1,17 @@
+[Unit]
+Description=Sendspin audio streaming server
+After=pipewire.service network-online.target
+Wants=pipewire.service
+
+[Service]
+ExecStart=%h/.local/bin/sendspin-server \
+ --source alsa_input.usb-Device-00.analog-stereo \
+ --name %H \
+ --port 8927 \
+ --rate 48000 \
+ --channels 2
+Restart=on-failure
+RestartSec=5
+
+[Install]
+WantedBy=default.target
diff --git a/internal/live/http.go b/internal/live/http.go
index eda91a1..6dbc5d8 100644
--- a/internal/live/http.go
+++ b/internal/live/http.go
@@ -259,6 +259,19 @@ type stateView struct {
func (h *HTTPServer) getState(w http.ResponseWriter, _ *http.Request) {
active, pb, vol, members := h.mgr.State()
+ // Fill in names for offline members from the roster — the manager
+ // only knows names of currently-connected clients.
+ if len(members) > 0 {
+ roster := map[string]string{}
+ for _, e := range h.roster.List() {
+ roster[e.ClientID] = e.Name
+ }
+ for i := range members {
+ if members[i].Name == "" {
+ members[i].Name = roster[members[i].ClientID]
+ }
+ }
+ }
writeJSON(w, http.StatusOK, stateView{
ActivePreset: active,
Playback: pb,
diff --git a/internal/live/manager.go b/internal/live/manager.go
index de79e06..5e1ffdf 100644
--- a/internal/live/manager.go
+++ b/internal/live/manager.go
@@ -27,6 +27,7 @@ type StateChange struct {
// MemberState is per-speaker detail used by /api/state.
type MemberState struct {
ClientID string `json:"client_id"`
+ Name string `json:"name,omitempty"`
Connected bool `json:"connected"`
Muted bool `json:"muted"`
Volume int `json:"volume"`
@@ -416,6 +417,7 @@ func (m *Manager) State() (string, Playback, int, []MemberState) {
members[i].Connected = true
members[i].Muted = c.Muted
members[i].Volume = c.Volume
+ members[i].Name = c.Name
}
}
}
diff --git a/internal/live/web/index.html b/internal/live/web/index.html
index 21c3084..6b744a8 100644
--- a/internal/live/web/index.html
+++ b/internal/live/web/index.html
@@ -348,11 +348,16 @@ function render() {
} else {
mEl.innerHTML = "";
for (const m of state.members) {
+ const label = m.name || m.client_id;
+ const showID = m.name && m.name !== m.client_id;
const div = document.createElement("div");
div.className = "row";
div.innerHTML = `
- ${escapeHTML(m.client_id)}
+
+ ${escapeHTML(label)}
+ ${showID ? `${escapeHTML(m.client_id)}
` : ''}
+
${m.connected ? `vol ${m.volume}${m.muted ? ' · muted' : ''}` : 'offline'}
`;
mEl.appendChild(div);
diff --git a/internal/pipewire/muter.go b/internal/pipewire/muter.go
new file mode 100644
index 0000000..551c50c
--- /dev/null
+++ b/internal/pipewire/muter.go
@@ -0,0 +1,228 @@
+package pipewire
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "os/exec"
+ "strconv"
+ "sync"
+ "time"
+)
+
+// Muter keeps a single PipeWire node muted or unmuted to match a desired
+// state, re-asserting periodically so transient failures (or external
+// interference) self-heal. Resolves the numeric node ID lazily because
+// PipeWire IDs change across restarts.
+//
+// Failure modes worth knowing:
+// - wpctl missing: the muter logs once at construction and becomes a no-op.
+// - Target node not present yet: resolution is retried on every tick,
+// so it's fine to start the client before the loopback exists.
+// - Process killed with SIGKILL: nothing can unmute. SIGINT/SIGTERM and
+// normal exits run Stop, which forces unmute synchronously.
+type Muter struct {
+ nodeName string
+ hasWpctl bool
+
+ mu sync.Mutex
+ desired bool // desired muted state
+ lastApplied int // -1 = unknown, 0 = unmuted, 1 = muted
+ lastResolved int // cached node ID, -1 = unknown
+ lastResolveErrLogged bool
+
+ wake chan struct{}
+ stop chan struct{}
+ done chan struct{}
+}
+
+// NewMuter constructs a Muter for the given node.name. The control loop
+// starts immediately and the node is forced to unmuted on startup as a
+// defensive measure against a previous run that crashed mid-mute.
+func NewMuter(nodeName string) *Muter {
+ m := &Muter{
+ nodeName: nodeName,
+ lastApplied: -1,
+ lastResolved: -1,
+ wake: make(chan struct{}, 1),
+ stop: make(chan struct{}),
+ done: make(chan struct{}),
+ }
+ if _, err := exec.LookPath("wpctl"); err != nil {
+ log.Printf("muter: wpctl not found (%v); loopback mute disabled", err)
+ } else {
+ m.hasWpctl = true
+ }
+ go m.run()
+ return m
+}
+
+// SetDesiredMute updates the target state. Cheap and non-blocking; the
+// control loop applies it asynchronously and re-asserts on a ticker.
+func (m *Muter) SetDesiredMute(muted bool) {
+ m.mu.Lock()
+ changed := m.desired != muted
+ m.desired = muted
+ m.mu.Unlock()
+ if changed {
+ m.kick()
+ }
+}
+
+func (m *Muter) kick() {
+ select {
+ case m.wake <- struct{}{}:
+ default:
+ }
+}
+
+func (m *Muter) run() {
+ defer close(m.done)
+ if !m.hasWpctl {
+ <-m.stop
+ return
+ }
+
+ // Eager startup unmute so a previous crashed run can't leave the
+ // loopback stuck muted before the player has reported its state.
+ m.applyOnce(false)
+
+ ticker := time.NewTicker(3 * time.Second)
+ defer ticker.Stop()
+
+ for {
+ select {
+ case <-m.stop:
+ return
+ case <-m.wake:
+ case <-ticker.C:
+ }
+ m.mu.Lock()
+ want := m.desired
+ m.mu.Unlock()
+ m.applyOnce(want)
+ }
+}
+
+// applyOnce resolves the node ID if needed and issues `wpctl set-mute`
+// when the desired state differs from what we last successfully applied.
+// On apply failure, lastApplied is left untouched so the next tick retries.
+func (m *Muter) applyOnce(muted bool) {
+ id, err := m.resolveID()
+ if err != nil {
+ return
+ }
+
+ m.mu.Lock()
+ last := m.lastApplied
+ m.mu.Unlock()
+
+ want := 0
+ if muted {
+ want = 1
+ }
+ if last == want {
+ return
+ }
+
+ if err := setMute(id, muted); err != nil {
+ log.Printf("muter: wpctl set-mute %d %d failed: %v", id, want, err)
+ return
+ }
+ m.mu.Lock()
+ m.lastApplied = want
+ m.mu.Unlock()
+}
+
+// resolveID returns the cached node ID, refreshing from pw-dump on first
+// use or after a wpctl failure invalidated it. The "not found" case is
+// logged once per outage so a missing loopback doesn't spam the log every
+// 3 seconds, but a successful resolution rearms the warning.
+func (m *Muter) resolveID() (int, error) {
+ m.mu.Lock()
+ cached := m.lastResolved
+ m.mu.Unlock()
+ if cached >= 0 {
+ return cached, nil
+ }
+
+ id, err := FindNodeIDByName(m.nodeName)
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ if err != nil {
+ if !m.lastResolveErrLogged {
+ log.Printf("muter: %v (will keep retrying)", err)
+ m.lastResolveErrLogged = true
+ }
+ return -1, err
+ }
+ if m.lastResolveErrLogged {
+ log.Printf("muter: node.name=%q resolved to id %d", m.nodeName, id)
+ m.lastResolveErrLogged = false
+ }
+ m.lastResolved = id
+ // Force the next apply to re-issue the wpctl command, since a new ID
+ // almost certainly means a fresh node whose mute state we don't know.
+ m.lastApplied = -1
+ return id, nil
+}
+
+func setMute(id int, muted bool) error {
+ arg := "0"
+ if muted {
+ arg = "1"
+ }
+ cmd := exec.Command("wpctl", "set-mute", strconv.Itoa(id), arg)
+ out, err := cmd.CombinedOutput()
+ if err != nil {
+ return fmt.Errorf("%w: %s", err, string(out))
+ }
+ return nil
+}
+
+// Stop tears the muter down. It forces an unmute (retried a few times)
+// before returning so a normal-exit path leaves the loopback in the
+// expected state even if wpctl was briefly flaky. Safe to call multiple
+// times.
+func (m *Muter) Stop(ctx context.Context) {
+ select {
+ case <-m.stop:
+ return
+ default:
+ }
+ close(m.stop)
+ <-m.done
+
+ if !m.hasWpctl {
+ return
+ }
+ if err := m.forceUnmute(ctx); err != nil {
+ log.Printf("muter: WARNING failed to unmute %q on shutdown: %v — loopback may be stuck muted", m.nodeName, err)
+ }
+}
+
+func (m *Muter) forceUnmute(ctx context.Context) error {
+ var lastErr error
+ backoff := 100 * time.Millisecond
+ for attempt := 0; attempt < 3; attempt++ {
+ if ctx.Err() != nil {
+ return errors.Join(lastErr, ctx.Err())
+ }
+ id, err := FindNodeIDByName(m.nodeName)
+ if err != nil {
+ lastErr = err
+ } else if err := setMute(id, false); err != nil {
+ lastErr = err
+ } else {
+ return nil
+ }
+ select {
+ case <-ctx.Done():
+ return errors.Join(lastErr, ctx.Err())
+ case <-time.After(backoff):
+ }
+ backoff *= 2
+ }
+ return lastErr
+}
diff --git a/internal/pipewire/nodes.go b/internal/pipewire/nodes.go
index 7a75d59..475a702 100644
--- a/internal/pipewire/nodes.go
+++ b/internal/pipewire/nodes.go
@@ -62,6 +62,32 @@ func listAllNodes() ([]Node, error) {
return nodes, nil
}
+// FindNodeIDByName scans every PipeWire node (not just Audio/*) and returns
+// the numeric ID of the first one whose node.name matches. Used by the
+// loopback muter, where the target may be a stream/loopback rather than a
+// sink/source. Returns -1 with an error when no match is found.
+func FindNodeIDByName(name string) (int, error) {
+ out, err := exec.Command("pw-dump").Output()
+ if err != nil {
+ return -1, fmt.Errorf("pw-dump: %w", err)
+ }
+
+ var objects []pwObject
+ if err := json.Unmarshal(out, &objects); err != nil {
+ return -1, fmt.Errorf("parse pw-dump: %w", err)
+ }
+
+ for _, o := range objects {
+ if o.Type != "PipeWire:Interface:Node" {
+ continue
+ }
+ if o.Info.Props.NodeName == name {
+ return o.ID, nil
+ }
+ }
+ return -1, fmt.Errorf("no PipeWire node with node.name=%q", name)
+}
+
func ListSinks() ([]Node, error) {
all, err := listAllNodes()
if err != nil {
diff --git a/internal/pipewire/output.go b/internal/pipewire/output.go
index 689edcd..61857b8 100644
--- a/internal/pipewire/output.go
+++ b/internal/pipewire/output.go
@@ -56,7 +56,8 @@ func (o *Output) Open(sampleRate, channels, bitDepth int) error {
func (o *Output) startUnlocked() error {
nameJSON, _ := json.Marshal(o.nodeName)
- props := fmt.Sprintf(`{"node.name":%s}`, nameJSON)
+ latency := fmt.Sprintf("960/%d", o.sampleRate)
+ props := fmt.Sprintf(`{"node.name":%s,"node.latency":"%s"}`, nameJSON, latency)
args := []string{
"--playback",
@@ -64,6 +65,7 @@ func (o *Output) startUnlocked() error {
"--format", "s32",
"--rate", strconv.Itoa(o.sampleRate),
"--channels", strconv.Itoa(o.channels),
+ "--latency", latency,
"--properties", props,
}
if o.target != "" {