diff --git a/Dockerfile.cross b/Dockerfile.cross
index 251e000..5199d53 100644
--- a/Dockerfile.cross
+++ b/Dockerfile.cross
@@ -20,6 +20,7 @@ RUN dpkg --add-architecture armhf \
WORKDIR /src
COPY go.mod go.sum ./
+COPY third_party/sendspin-go/go.mod third_party/sendspin-go/go.sum ./third_party/sendspin-go/
RUN go mod download
COPY . .
diff --git a/cmd/live-server/main.go b/cmd/live-server/main.go
index 761db57..5fdaa5d 100644
--- a/cmd/live-server/main.go
+++ b/cmd/live-server/main.go
@@ -60,6 +60,7 @@ func main() {
mqttDiscovery string
rate int
channels int
+ preset string
)
addInt(&port, 8927, "port", "p", "Sendspin WebSocket port")
@@ -74,6 +75,7 @@ func main() {
addStr(&mqttDiscovery, "homeassistant", "mqtt-discovery", "", "Home Assistant discovery prefix")
addInt(&rate, 48000, "rate", "r", "capture sample rate")
addInt(&channels, 2, "channels", "", "capture channels")
+ addStr(&preset, "", "preset", "", "activate this preset id at startup (begins playback immediately)")
flag.Usage = usage
flag.Parse()
@@ -147,6 +149,13 @@ func main() {
go func() { serverDone <- server.Start() }()
log.Printf("Sendspin server %q on :%d", name, port)
+ if preset != "" {
+ if err := mgr.SetActivePreset(preset); err != nil {
+ log.Fatalf("activate preset %q: %v", preset, err)
+ }
+ log.Printf("preset %q activated at startup", preset)
+ }
+
sig := make(chan os.Signal, 1)
signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
select {
diff --git a/dist/systemd/sendspin-client.service b/dist/systemd/sendspin-client.service
deleted file mode 100644
index aad7841..0000000
--- a/dist/systemd/sendspin-client.service
+++ /dev/null
@@ -1,16 +0,0 @@
-[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/dist/systemd/sendspin-server.service b/dist/systemd/sendspin-server.service
deleted file mode 100644
index b894cb9..0000000
--- a/dist/systemd/sendspin-server.service
+++ /dev/null
@@ -1,17 +0,0 @@
-[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/manager.go b/internal/live/manager.go
index 65620ba..de79e06 100644
--- a/internal/live/manager.go
+++ b/internal/live/manager.go
@@ -186,9 +186,11 @@ func (m *Manager) SetActivePreset(id string) error {
return nil
}
- // On entry from off, default to paused; otherwise carry over.
+ // On entry from off, auto-start playback so picking a preset is a
+ // one-click action. Switching between presets carries over the
+ // current playback state.
if prev == "" {
- m.playback = Paused
+ m.playback = Playing
}
m.src.SetTarget(nextSource)
diff --git a/internal/live/web/index.html b/internal/live/web/index.html
index 7e13570..21c3084 100644
--- a/internal/live/web/index.html
+++ b/internal/live/web/index.html
@@ -205,9 +205,7 @@
Playback
-
-
@@ -331,10 +329,12 @@ function render() {
}
sel.value = state.active_preset || "";
- // playback buttons
- $("btn-play").disabled = !state.active_preset || state.playback === "playing";
- $("btn-pause").disabled = state.playback !== "playing";
- $("btn-stop").disabled = !state.active_preset;
+ // playback toggle button: pauses when playing, resumes when paused.
+ // Disabled when no preset is active (selector controls play/stop).
+ const pauseBtn = $("btn-pause");
+ pauseBtn.disabled = !state.active_preset;
+ pauseBtn.textContent = state.playback === "paused" ? "Play" : "Pause";
+ pauseBtn.dataset.cmd = state.playback === "paused" ? "play" : "pause";
// volume slider
$("volume-label").textContent = state.volume + "%";
@@ -429,9 +429,11 @@ $("active-preset").addEventListener("change", async (e) => {
refreshAll();
});
-$("btn-play").onclick = async () => { await api("POST", "/api/playback", { command: "play" }).catch(e => alert(e.message)); refreshAll(); };
-$("btn-pause").onclick = async () => { await api("POST", "/api/playback", { command: "pause" }).catch(e => alert(e.message)); refreshAll(); };
-$("btn-stop").onclick = async () => { await api("POST", "/api/playback", { command: "stop" }).catch(e => alert(e.message)); refreshAll(); };
+$("btn-pause").onclick = async () => {
+ const cmd = $("btn-pause").dataset.cmd || "pause";
+ await api("POST", "/api/playback", { command: cmd }).catch(e => alert(e.message));
+ refreshAll();
+};
const vol = $("volume");
vol.addEventListener("input", () => { volumeDirty = true; $("volume-label").textContent = vol.value + "%"; });
diff --git a/third_party/sendspin-go/internal/server/flac_encoder.go b/third_party/sendspin-go/internal/server/flac_encoder.go
index 18ea2a2..2e91a5e 100644
--- a/third_party/sendspin-go/internal/server/flac_encoder.go
+++ b/third_party/sendspin-go/internal/server/flac_encoder.go
@@ -24,6 +24,8 @@ type FLACEncoder struct {
channels int
bitDepth int
blockSize int
+ frameNum uint64
+ chanBufs [][]int32 // reused per-channel sample buffers
}
// NewFLACEncoder creates a FLAC encoder with prediction analysis enabled.
@@ -54,6 +56,11 @@ func NewFLACEncoder(sampleRate, channels, bitDepth, blockSize int) (*FLACEncoder
copy(codecHeader, buf.Bytes())
buf.Reset()
+ chanBufs := make([][]int32, channels)
+ for ch := range chanBufs {
+ chanBufs[ch] = make([]int32, blockSize)
+ }
+
return &FLACEncoder{
encoder: enc,
buf: buf,
@@ -62,6 +69,7 @@ func NewFLACEncoder(sampleRate, channels, bitDepth, blockSize int) (*FLACEncoder
channels: channels,
bitDepth: bitDepth,
blockSize: blockSize,
+ chanBufs: chanBufs,
}, nil
}
@@ -92,20 +100,18 @@ func (e *FLACEncoder) Encode(samples []int32) ([]byte, error) {
shift = uint(24 - e.bitDepth)
}
- // De-interleave into per-channel slices.
+ // De-interleave into pre-allocated per-channel buffers.
subframes := make([]*frame.Subframe, e.channels)
for ch := 0; ch < e.channels; ch++ {
- channelSamples := make([]int32, e.blockSize)
+ buf := e.chanBufs[ch]
for i := 0; i < e.blockSize; i++ {
- channelSamples[i] = samples[i*e.channels+ch] >> shift
+ buf[i] = samples[i*e.channels+ch] >> shift
}
subframes[ch] = &frame.Subframe{
SubHeader: frame.SubHeader{
- // PredVerbatim signals the encoder to run prediction analysis
- // and pick the optimal method (Constant, Fixed, or Verbatim).
Pred: frame.PredVerbatim,
},
- Samples: channelSamples,
+ Samples: buf,
NSamples: e.blockSize,
}
}
@@ -127,9 +133,11 @@ func (e *FLACEncoder) Encode(samples []int32) ([]byte, error) {
SampleRate: uint32(e.sampleRate),
Channels: channelAssign,
BitsPerSample: uint8(e.bitDepth),
+ Num: e.frameNum,
},
Subframes: subframes,
}
+ e.frameNum++
e.buf.Reset()
if err := e.encoder.WriteFrame(f); err != nil {
diff --git a/third_party/sendspin-go/pkg/sendspin/client_dialer.go b/third_party/sendspin-go/pkg/sendspin/client_dialer.go
index 4df9664..23dc06e 100644
--- a/third_party/sendspin-go/pkg/sendspin/client_dialer.go
+++ b/third_party/sendspin-go/pkg/sendspin/client_dialer.go
@@ -128,27 +128,23 @@ func (d *clientDialer) claim(instance string) bool {
}
// release updates the instance slot after a dial attempt completes.
-// On success (dialErr == nil) the instance stays latched in the active
-// set — we never re-dial a successfully-connected instance. On error
-// it frees the slot and schedules an exponentially-backed-off cooldown
-// before the next retry.
+// The slot is always freed — dial() only returns after handleConnection
+// has returned, so the connection is already dead by the time we get
+// here, and a future mDNS re-emission should be free to redial. On
+// error it also schedules an exponentially-backed-off cooldown before
+// the next retry.
func (d *clientDialer) release(instance string, dialErr error) {
d.mu.Lock()
defer d.mu.Unlock()
+ delete(d.active, instance)
+
if dialErr == nil {
- // Latch: keep active[instance] = true so future discovery
- // events for this instance are silently ignored. The server
- // already handled the connection; re-dialing would create a
- // duplicate session.
delete(d.failures, instance)
delete(d.cooldown, instance)
return
}
- // Dial failed — release the slot so the instance can be retried
- // after the backoff period.
- delete(d.active, instance)
d.failures[instance]++
backoff := d.baseBackoff * time.Duration(1<<(d.failures[instance]-1))
if backoff > d.maxBackoff || backoff <= 0 {
diff --git a/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go b/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go
index 7bbc641..f68f42c 100644
--- a/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go
+++ b/third_party/sendspin-go/pkg/sendspin/client_dialer_test.go
@@ -54,11 +54,17 @@ func TestClientDialerDedupesByInstance(t *testing.T) {
var mu sync.Mutex
var seen []string
+ // Block each dial until the test releases it, so concurrent
+ // duplicate emissions for the same instance arrive while the
+ // dial is in flight — which is what the dedupe protects against.
+ gate := make(chan struct{})
+
dial := func(ctx context.Context, info *discovery.ClientInfo) error {
atomic.AddInt32(&dialCalls, 1)
mu.Lock()
seen = append(seen, info.Instance)
mu.Unlock()
+ <-gate
return nil
}
@@ -72,39 +78,32 @@ func TestClientDialerDedupesByInstance(t *testing.T) {
close(done)
}()
- // Emit the same instance 3 times and a different instance once
+ // Emit the same instance 3 times and a different instance once,
+ // all while dials are blocked in `gate`.
in <- &discovery.ClientInfo{Instance: "a._sendspin._tcp.local.", Host: "1.1.1.1", Port: 8928, Path: "/sendspin"}
in <- &discovery.ClientInfo{Instance: "a._sendspin._tcp.local.", Host: "1.1.1.1", Port: 8928, Path: "/sendspin"}
in <- &discovery.ClientInfo{Instance: "b._sendspin._tcp.local.", Host: "1.1.1.2", Port: 8928, Path: "/sendspin"}
in <- &discovery.ClientInfo{Instance: "a._sendspin._tcp.local.", Host: "1.1.1.1", Port: 8928, Path: "/sendspin"}
- // Wait until at least 2 unique dials have happened
- deadline := time.After(500 * time.Millisecond)
- for {
- if atomic.LoadInt32(&dialCalls) >= 2 {
- break
- }
- select {
- case <-deadline:
- t.Fatalf("timed out waiting for dial calls, got %d", atomic.LoadInt32(&dialCalls))
- case <-time.After(10 * time.Millisecond):
- }
- }
+ waitForCalls(t, &dialCalls, 2, 500*time.Millisecond)
- // Give any extra (incorrect) calls a chance to fire
+ // Give any extra (incorrect) calls a chance to fire while dials
+ // are still blocked.
time.Sleep(50 * time.Millisecond)
- cancel()
- <-done
-
if got := atomic.LoadInt32(&dialCalls); got != 2 {
- t.Errorf("dialCalls = %d, want 2 (one per unique instance)", got)
+ t.Errorf("dialCalls = %d, want 2 (one per unique instance while in flight)", got)
}
mu.Lock()
- defer mu.Unlock()
if len(seen) != 2 {
+ mu.Unlock()
t.Fatalf("seen = %v, want 2 entries", seen)
}
+ mu.Unlock()
+
+ close(gate)
+ cancel()
+ <-done
}
func waitForCalls(t *testing.T, counter *int32, want int32, timeout time.Duration) {
diff --git a/third_party/sendspin-go/pkg/sendspin/server.go b/third_party/sendspin-go/pkg/sendspin/server.go
index 06fe810..088f601 100644
--- a/third_party/sendspin-go/pkg/sendspin/server.go
+++ b/third_party/sendspin-go/pkg/sendspin/server.go
@@ -96,6 +96,12 @@ type Server struct {
audioSource AudioSource
consecutiveReadErrs int
+ // Accumulates bytes-on-wire across one streamAudio rate window
+ // (encoded chunk size, not raw PCM) so the periodic stats line can
+ // report both chunks/sec and kbps. Touched only from the streamAudio
+ // goroutine so no locking needed.
+ lastWindowBytes int64
+
mdnsManager *discovery.Manager
// server-initiated discovery dialer cancel
@@ -554,12 +560,19 @@ func (s *Server) clientWriter(c *ServerClient) {
for {
select {
case msg := <-c.sendChan:
+ queueDepth := len(c.sendChan)
switch v := msg.(type) {
case []byte:
c.conn.SetWriteDeadline(time.Now().Add(writeDeadline))
+ wStart := time.Now()
if err := c.conn.WriteMessage(websocket.BinaryMessage, v); err != nil {
+ log.Printf("clientWriter %s: binary write error: %v (queue %d)", c.name, err, queueDepth)
return
}
+ if wDur := time.Since(wStart); wDur > 5*time.Millisecond || queueDepth > 5 {
+ log.Printf("clientWriter %s: %s for %d bytes (queue %d)",
+ c.name, wDur.Round(time.Microsecond), len(v), queueDepth)
+ }
default:
data, err := json.Marshal(v)
if err != nil {
@@ -567,6 +580,7 @@ func (s *Server) clientWriter(c *ServerClient) {
}
c.conn.SetWriteDeadline(time.Now().Add(writeDeadline))
if err := c.conn.WriteMessage(websocket.TextMessage, data); err != nil {
+ log.Printf("clientWriter %s: text write error: %v", c.name, err)
return
}
}
diff --git a/third_party/sendspin-go/pkg/sendspin/server_client.go b/third_party/sendspin-go/pkg/sendspin/server_client.go
index 536416d..0577e2a 100644
--- a/third_party/sendspin-go/pkg/sendspin/server_client.go
+++ b/third_party/sendspin-go/pkg/sendspin/server_client.go
@@ -3,6 +3,7 @@
package sendspin
import (
+ "encoding/binary"
"encoding/json"
"fmt"
"log"
@@ -102,13 +103,68 @@ func (c *ServerClient) Send(msgType string, payload interface{}) error {
// enqueued frames are dropped silently. Callers should treat this as
// best-effort once they've observed a client leaving.
func (c *ServerClient) SendBinary(data []byte) error {
+ // Fast path: queue has room.
select {
case c.sendChan <- data:
return nil
default:
- return fmt.Errorf("client send buffer full (depth=%d, cap=%d)",
- len(c.sendChan), cap(c.sendChan))
}
+
+ // Queue full. Drop-oldest is timestamp-aware in effect: the engine
+ // pushes chunks in monotonically increasing playbackTime order, so
+ // the head of the queue has the earliest scheduled time = the chunk
+ // most likely to be stale by the time the writer catches up.
+ //
+ // We use the *new* chunk's playbackTime as our "now + BufferAheadMs"
+ // reference (the engine just stamped it that way), so we don't need
+ // an external clock to decide what's stale. The actual log includes
+ // how far past the scheduling headroom the dropped chunks were.
+ newTS, newOK := extractAudioPlaybackTime(data)
+ const bufferAheadUs = int64(BufferAheadMs) * 1000
+
+ dropped := 0
+ var maxLatenessUs int64
+ for dropped < cap(c.sendChan) {
+ select {
+ case head := <-c.sendChan:
+ if newOK {
+ if headBytes, ok := head.([]byte); ok {
+ if headTS, headOK := extractAudioPlaybackTime(headBytes); headOK {
+ // Lateness = how long past "now" this chunk's playback
+ // time was. "Now" ≈ newTS - bufferAhead.
+ late := (newTS - bufferAheadUs) - headTS
+ if late > maxLatenessUs {
+ maxLatenessUs = late
+ }
+ }
+ }
+ }
+ dropped++
+ default:
+ }
+ select {
+ case c.sendChan <- data:
+ if dropped > 0 {
+ log.Printf("client %s send pressure: dropped %d stale chunks (oldest was %s past schedule)",
+ c.name, dropped,
+ time.Duration(maxLatenessUs)*time.Microsecond)
+ }
+ return nil
+ default:
+ }
+ }
+ return fmt.Errorf("client send buffer full (depth=%d, cap=%d)",
+ len(c.sendChan), cap(c.sendChan))
+}
+
+// extractAudioPlaybackTime returns the playbackTime (in microseconds) embedded
+// in a sendspin audio chunk's 9-byte header. Returns false for non-audio frames.
+func extractAudioPlaybackTime(data []byte) (int64, bool) {
+ const audioChunkHeaderSize = 9
+ if len(data) < audioChunkHeaderSize || data[0] != AudioChunkMessageType {
+ return 0, false
+ }
+ return int64(binary.BigEndian.Uint64(data[1:audioChunkHeaderSize])), true
}
// State returns the client's current playback state ("synchronized",
diff --git a/third_party/sendspin-go/pkg/sendspin/server_stream.go b/third_party/sendspin-go/pkg/sendspin/server_stream.go
index 45d97dc..1fe1224 100644
--- a/third_party/sendspin-go/pkg/sendspin/server_stream.go
+++ b/third_party/sendspin-go/pkg/sendspin/server_stream.go
@@ -18,6 +18,8 @@ func (s *Server) streamAudio() {
tickBudget := time.Duration(ChunkDurationMs) * time.Millisecond
var lastTick time.Time
+ var chunkCount int
+ rateWindowStart := time.Now()
for {
select {
case t := <-ticker.C:
@@ -35,6 +37,21 @@ func (s *Server) streamAudio() {
lastTick = t
s.generateAndSendChunk()
+ chunkCount++
+
+ if elapsed := time.Since(rateWindowStart); elapsed >= 10*time.Second {
+ rate := float64(chunkCount) / elapsed.Seconds()
+ avgBytes := 0
+ if chunkCount > 0 {
+ avgBytes = int(s.lastWindowBytes) / chunkCount
+ }
+ kbps := float64(s.lastWindowBytes) * 8 / 1000.0 / elapsed.Seconds()
+ log.Printf("ENGINE STATS: %.3f chunks/sec, avg %d bytes/chunk, %.1f kbps over %s (%d chunks)",
+ rate, avgBytes, kbps, elapsed.Round(time.Millisecond), chunkCount)
+ chunkCount = 0
+ s.lastWindowBytes = 0
+ rateWindowStart = time.Now()
+ }
case <-s.stopChan:
log.Printf("Audio streaming stopping")
return
@@ -100,6 +117,10 @@ func (s *Server) generateAndSendChunk() {
tracker := c.bufferTracker
c.mu.RUnlock()
+ if codec == "" {
+ continue
+ }
+
switch codec {
case "opus":
if opusEncoder != nil {
@@ -139,6 +160,7 @@ func (s *Server) generateAndSendChunk() {
}
chunk := CreateAudioChunk(playbackTime, audioData)
+ s.lastWindowBytes += int64(len(chunk))
if tracker != nil {
chunkDurationUs := int64(ChunkDurationMs) * 1000