feat: Broadcast buffer, sequential playlist, ICY metadata

- Broadcast buffer: single-producer multi-consumer ring buffer with
  per-client read cursors. 32KB burst-on-connect for fast playback.
  Never blocks producer (overwrites old data for slow clients).
- Sequential playlist: play-list runs tracks one at a time using
  Harmony's on-end callback + condition variable for completion.
- ICY metadata: set-now-playing called on each track change.
- Fixed string vs pathname bug in harmony:play (etypecase mismatch).
- Debug logging for client disconnect diagnosis.

Verified: browser plays shuffled FLAC playlist via 128kbps MP3 stream.

cl-streamer/README.org

@@ -1,6 +1,6 @@
 #+TITLE: CL-Streamer
 #+AUTHOR: Glenn Thompson
-#+DATE: 2026-03-03
+#+DATE: 2026-15-02
 
 * Overview
 

cl-streamer/buffer.lisp

@@ -1,73 +1,87 @@
 (in-package #:cl-streamer)
 
-(defclass ring-buffer ()
+;;; ---- Broadcast Ring Buffer ----
+;;; Single-producer, multi-consumer circular buffer.
+;;; The writer advances write-pos; each client has its own read cursor.
+;;; Old data is overwritten when the buffer wraps — slow clients lose data
+;;; rather than blocking the producer (appropriate for live streaming).
+
+(defclass broadcast-buffer ()
   ((data :initarg :data :accessor buffer-data)
    (size :initarg :size :reader buffer-size)
-   (read-pos :initform 0 :accessor buffer-read-pos)
    (write-pos :initform 0 :accessor buffer-write-pos)
-   (lock :initform (bt:make-lock "ring-buffer-lock") :reader buffer-lock)
+   (lock :initform (bt:make-lock "broadcast-buffer-lock") :reader buffer-lock)
    (not-empty :initform (bt:make-condition-variable :name "buffer-not-empty")
               :reader buffer-not-empty)
-   (not-full :initform (bt:make-condition-variable :name "buffer-not-full")
+   (burst-size :initarg :burst-size :reader buffer-burst-size
-             :reader buffer-not-full)))
+               :initform (* 32 1024)
+               :documentation "Bytes of recent data to send on new client connect")))
 
 (defun make-ring-buffer (size)
-  "Create a ring buffer with SIZE bytes capacity."
+  "Create a broadcast ring buffer with SIZE bytes capacity."
-  (make-instance 'ring-buffer
+  (make-instance 'broadcast-buffer
                  :data (make-array size :element-type '(unsigned-byte 8))
                  :size size))
 
-(defun buffer-available (buffer)
-  "Return the number of bytes available to read."
-  (bt:with-lock-held ((buffer-lock buffer))
-    (let ((write (buffer-write-pos buffer))
-          (read (buffer-read-pos buffer))
-          (size (buffer-size buffer)))
-      (mod (- write read) size))))
-
-(defun buffer-free-space (buffer)
-  "Return the number of bytes available to write."
-  (- (buffer-size buffer) (buffer-available buffer) 1))
-
 (defun buffer-write (buffer data &key (start 0) (end (length data)))
-  "Write bytes from DATA to BUFFER. Blocks if buffer is full."
+  "Write bytes into the broadcast buffer. Never blocks; overwrites old data."
   (let ((len (- end start)))
-    (bt:with-lock-held ((buffer-lock buffer))
+    (when (> len 0)
-      (loop while (< (buffer-free-space buffer) len)
+      (bt:with-lock-held ((buffer-lock buffer))
-            do (bt:condition-wait (buffer-not-full buffer) (buffer-lock buffer)))
+        (let ((write-pos (buffer-write-pos buffer))
-      (let ((write-pos (buffer-write-pos buffer))
+              (size (buffer-size buffer))
-            (size (buffer-size buffer))
+              (buf-data (buffer-data buffer)))
-            (buf-data (buffer-data buffer)))
+          (loop for i from start below end
-        (loop for i from start below end
+                for j = (mod write-pos size) then (mod (1+ j) size)
-              for j = write-pos then (mod (1+ j) size)
+                do (setf (aref buf-data j) (aref data i))
-              do (setf (aref buf-data j) (aref data i))
+                finally (setf (buffer-write-pos buffer) (+ write-pos len))))
-              finally (setf (buffer-write-pos buffer) (mod (1+ j) size))))
+        (bt:condition-notify (buffer-not-empty buffer))))
-      (bt:condition-notify (buffer-not-empty buffer)))
     len))
 
-(defun buffer-read (buffer output &key (start 0) (end (length output)) (blocking t))
+(defun buffer-read-from (buffer read-pos output &key (start 0) (end (length output)))
-  "Read bytes from BUFFER into OUTPUT. Returns number of bytes read.
+  "Read bytes from BUFFER starting at READ-POS into OUTPUT.
-   If BLOCKING is T, waits for data. Otherwise returns 0 if empty."
+   Returns (values bytes-read new-read-pos).
+   READ-POS is the client's absolute position in the stream."
   (let ((requested (- end start)))
     (bt:with-lock-held ((buffer-lock buffer))
-      (when blocking
+      (let* ((write-pos (buffer-write-pos buffer))
-        (loop while (zerop (buffer-available buffer))
-              do (bt:condition-wait (buffer-not-empty buffer) (buffer-lock buffer))))
-      (let* ((available (buffer-available buffer))
-             (to-read (min requested available))
-             (read-pos (buffer-read-pos buffer))
              (size (buffer-size buffer))
-             (buf-data (buffer-data buffer)))
+             (buf-data (buffer-data buffer))
-        (loop for i from start below (+ start to-read)
+             ;; Clamp read-pos: if client is too far behind, skip ahead
-              for j = read-pos then (mod (1+ j) size)
+             (oldest-available (max 0 (- write-pos size)))
-              do (setf (aref output i) (aref buf-data j))
+             (effective-read (max read-pos oldest-available))
-              finally (setf (buffer-read-pos buffer) (mod (1+ j) size)))
+             (available (- write-pos effective-read))
-        (bt:condition-notify (buffer-not-full buffer))
+             (to-read (min requested available)))
-        to-read))))
+        (if (> to-read 0)
+            (progn
+              (loop for i from start below (+ start to-read)
+                    for j = (mod effective-read size) then (mod (1+ j) size)
+                    do (setf (aref output i) (aref buf-data j)))
+              (values to-read (+ effective-read to-read)))
+            (values 0 effective-read))))))
+
+(defun buffer-wait-for-data (buffer read-pos)
+  "Block until new data is available past READ-POS."
+  (bt:with-lock-held ((buffer-lock buffer))
+    (loop while (<= (buffer-write-pos buffer) read-pos)
+          do (bt:condition-wait (buffer-not-empty buffer) (buffer-lock buffer)))))
+
+(defun buffer-current-pos (buffer)
+  "Return the current write position (for new client burst start)."
+  (bt:with-lock-held ((buffer-lock buffer))
+    (buffer-write-pos buffer)))
+
+(defun buffer-burst-start (buffer)
+  "Return a read position that gives BURST-SIZE bytes of recent data.
+   This lets new clients start playing immediately."
+  (bt:with-lock-held ((buffer-lock buffer))
+    (let* ((write-pos (buffer-write-pos buffer))
+           (size (buffer-size buffer))
+           (oldest (max 0 (- write-pos size)))
+           (burst-start (max oldest (- write-pos (buffer-burst-size buffer)))))
+      burst-start)))
 
 (defun buffer-clear (buffer)
-  "Clear all data from the buffer."
+  "Clear the buffer."
   (bt:with-lock-held ((buffer-lock buffer))
-    (setf (buffer-read-pos buffer) 0
+    (setf (buffer-write-pos buffer) 0)))
-          (buffer-write-pos buffer) 0)
-    (bt:condition-notify (buffer-not-full buffer))))

cl-streamer/harmony-backend.lisp

@@ -0,0 +1,183 @@
+(defpackage #:cl-streamer/harmony
+  (:use #:cl #:alexandria)
+  (:local-nicknames (#:harmony #:org.shirakumo.fraf.harmony)
+                    (#:mixed #:org.shirakumo.fraf.mixed))
+  (:export #:audio-pipeline
+           #:make-audio-pipeline
+           #:start-pipeline
+           #:stop-pipeline
+           #:play-file
+           #:play-list
+           #:pipeline-encoder
+           #:pipeline-server
+           #:make-streaming-server))
+
+(in-package #:cl-streamer/harmony)
+
+;;; ---- Streaming Drain ----
+;;; Custom drain that captures PCM from Harmony's pack buffer
+;;; and feeds it to the encoder/stream server, replacing the
+;;; dummy drain which just discards audio data.
+
+(defclass streaming-drain (mixed:drain)
+  ((encoder :initarg :encoder :accessor drain-encoder)
+   (mount-path :initarg :mount-path :accessor drain-mount-path :initform "/stream.mp3")
+   (channels :initarg :channels :accessor drain-channels :initform 2)))
+
+(defmethod mixed:free ((drain streaming-drain)))
+
+(defmethod mixed:start ((drain streaming-drain)))
+
+(defmethod mixed:mix ((drain streaming-drain))
+  "Read interleaved float PCM from the pack buffer, encode to MP3, write to stream.
+   The pack buffer is (unsigned-byte 8) with IEEE 754 single-floats (4 bytes each).
+   Layout: L0b0 L0b1 L0b2 L0b3 R0b0 R0b1 R0b2 R0b3 L1b0 ... (interleaved stereo)"
+  (mixed:with-buffer-tx (data start size (mixed:pack drain))
+    (when (> size 0)
+      (let* ((channels (drain-channels drain))
+             (bytes-per-sample 4)  ; single-float = 4 bytes
+             (total-floats (floor size bytes-per-sample))
+             (num-samples (floor total-floats channels))
+             (pcm-buffer (make-array (* num-samples channels)
+                                     :element-type '(signed-byte 16))))
+        ;; Convert raw bytes -> single-float -> signed-16
+        (cffi:with-pointer-to-vector-data (ptr data)
+          (loop for i below (* num-samples channels)
+                for byte-offset = (+ start (* i bytes-per-sample))
+                for sample = (cffi:mem-ref ptr :float byte-offset)
+                do (setf (aref pcm-buffer i) (float-to-s16 sample))))
+        (handler-case
+            (let ((mp3-data (cl-streamer:encode-pcm-interleaved
+                             (drain-encoder drain) pcm-buffer num-samples)))
+              (when (> (length mp3-data) 0)
+                (cl-streamer:write-audio-data (drain-mount-path drain) mp3-data)))
+          (error (e)
+            (log:warn "Encode error in drain: ~A" e)))))
+    ;; Sleep for the duration of audio we just processed
+    ;; size = bytes, each frame = channels * 4 bytes (single-float)
+    (let* ((channels (drain-channels drain))
+           (bytes-per-frame (* channels 4))
+           (frames (floor size bytes-per-frame))
+           (samplerate (mixed:samplerate (mixed:pack drain))))
+      (when (> frames 0)
+        (sleep (/ frames samplerate))))
+    (mixed:finish size)))
+
+(defmethod mixed:end ((drain streaming-drain)))
+
+;;; ---- Audio Pipeline ----
+
+(defclass audio-pipeline ()
+  ((harmony-server :initform nil :accessor pipeline-harmony-server)
+   (encoder :initarg :encoder :accessor pipeline-encoder)
+   (stream-server :initarg :stream-server :accessor pipeline-server)
+   (mount-path :initarg :mount-path :accessor pipeline-mount-path :initform "/stream.mp3")
+   (sample-rate :initarg :sample-rate :accessor pipeline-sample-rate :initform 44100)
+   (channels :initarg :channels :accessor pipeline-channels :initform 2)
+   (running :initform nil :accessor pipeline-running-p)))
+
+(defun make-audio-pipeline (&key encoder stream-server (mount-path "/stream.mp3")
+                                 (sample-rate 44100) (channels 2))
+  "Create an audio pipeline connecting Harmony to the stream server via an encoder."
+  (make-instance 'audio-pipeline
+                 :encoder encoder
+                 :stream-server stream-server
+                 :mount-path mount-path
+                 :sample-rate sample-rate
+                 :channels channels))
+
+(defun start-pipeline (pipeline)
+  "Start the audio pipeline - initializes Harmony with our streaming drain."
+  (when (pipeline-running-p pipeline)
+    (error "Pipeline already running"))
+  (mixed:init)
+  (let* ((server (harmony:make-simple-server
+                  :name "CL-Streamer"
+                  :samplerate (pipeline-sample-rate pipeline)
+                  :latency 0.05
+                  :drain :dummy
+                  :output-channels (pipeline-channels pipeline)))
+         (output (harmony:segment :output server))
+         (old-drain (harmony:segment :drain output))
+         (pack (mixed:pack old-drain))
+         (drain (make-instance 'streaming-drain
+                               :encoder (pipeline-encoder pipeline)
+                               :mount-path (pipeline-mount-path pipeline)
+                               :channels (pipeline-channels pipeline))))
+    ;; Wire our streaming drain to the same pack buffer
+    (setf (mixed:pack drain) pack)
+    ;; Swap: withdraw old dummy drain, add our streaming drain
+    (mixed:withdraw old-drain output)
+    (mixed:add drain output)
+    (setf (pipeline-harmony-server pipeline) server)
+    (mixed:start server))
+  (setf (pipeline-running-p pipeline) t)
+  (log:info "Audio pipeline started with streaming drain")
+  pipeline)
+
+(defun stop-pipeline (pipeline)
+  "Stop the audio pipeline."
+  (setf (pipeline-running-p pipeline) nil)
+  (when (pipeline-harmony-server pipeline)
+    (mixed:end (pipeline-harmony-server pipeline))
+    (setf (pipeline-harmony-server pipeline) nil))
+  (log:info "Audio pipeline stopped")
+  pipeline)
+
+(defun play-file (pipeline file-path &key (mixer :music) title (on-end :free))
+  "Play an audio file through the pipeline.
+   The file will be decoded by Harmony and encoded for streaming.
+   If TITLE is given, update ICY metadata with it.
+   FILE-PATH can be a string or pathname.
+   ON-END is passed to harmony:play (default :free)."
+  (let* ((path (pathname file-path))
+         (server (pipeline-harmony-server pipeline))
+         (harmony:*server* server)
+         (display-title (or title (pathname-name path))))
+    ;; Update ICY metadata so listeners see the track name
+    (cl-streamer:set-now-playing (pipeline-mount-path pipeline) display-title)
+    (let ((voice (harmony:play path :mixer mixer :on-end on-end)))
+      (log:info "Now playing: ~A" display-title)
+      voice)))
+
+(defun play-list (pipeline file-list &key (gap 0.5))
+  "Play a list of file paths sequentially through the pipeline.
+   Each entry can be a string (path) or a plist (:file path :title title).
+   GAP is seconds of silence between tracks."
+  (bt:make-thread
+   (lambda ()
+     (loop for entry in file-list
+           while (pipeline-running-p pipeline)
+           do (multiple-value-bind (path title)
+                  (if (listp entry)
+                      (values (getf entry :file) (getf entry :title))
+                      (values entry nil))
+                (handler-case
+                    (let* ((done-lock (bt:make-lock "track-done"))
+                           (done-cv (bt:make-condition-variable :name "track-done"))
+                           (done-p nil)
+                           (server (pipeline-harmony-server pipeline))
+                           (harmony:*server* server)
+                           (voice (play-file pipeline path
+                                             :title title
+                                             :on-end (lambda (voice)
+                                                       (declare (ignore voice))
+                                                       (bt:with-lock-held (done-lock)
+                                                         (setf done-p t)
+                                                         (bt:condition-notify done-cv))))))
+                      (declare (ignore voice))
+                      ;; Wait for the track to finish via callback
+                      (bt:with-lock-held (done-lock)
+                        (loop until (or done-p (not (pipeline-running-p pipeline)))
+                              do (bt:condition-wait done-cv done-lock)))
+                      (when (> gap 0) (sleep gap)))
+                  (error (e)
+                    (log:warn "Error playing ~A: ~A" path e)
+                    (sleep 1))))))
+   :name "cl-streamer-playlist"))
+
+(declaim (inline float-to-s16))
+(defun float-to-s16 (sample)
+  "Convert a float sample (-1.0 to 1.0) to signed 16-bit integer."
+  (let ((clamped (max -1.0 (min 1.0 sample))))
+    (the (signed-byte 16) (round (* clamped 32767.0)))))

cl-streamer/package.lisp

@@ -7,11 +7,13 @@
    #:encoding-error
    
    ;; Buffer
-   #:ring-buffer
+   #:broadcast-buffer
    #:make-ring-buffer
    #:buffer-write
-   #:buffer-read
+   #:buffer-read-from
-   #:buffer-available
+   #:buffer-wait-for-data
+   #:buffer-current-pos
+   #:buffer-burst-start
    #:buffer-clear
    
    ;; ICY Protocol
@@ -34,5 +36,23 @@
    #:listener-count
    
    ;; Main API
+   #:*server*
    #:*default-port*
-   #:*default-metaint*))
+   #:*default-metaint*
+   #:start
+   #:stop
+   #:write-audio-data
+   #:set-now-playing
+   #:get-listener-count
+   
+   ;; Encoder
+   #:make-mp3-encoder
+   #:close-encoder
+   #:encode-pcm-interleaved
+   #:encode-flush
+   #:lame-version
+   
+   ;; AAC Encoder
+   #:make-aac-encoder
+   #:close-aac-encoder
+   #:encode-aac-pcm))

cl-streamer/stream-server.lisp

@@ -29,6 +29,8 @@
    (mount :initarg :mount :accessor client-mount)
    (wants-metadata :initarg :wants-metadata :accessor client-wants-metadata-p)
    (bytes-since-meta :initform 0 :accessor client-bytes-since-meta)
+   (read-pos :initform 0 :accessor client-read-pos
+             :documentation "Client's absolute position in the broadcast buffer")
    (thread :initform nil :accessor client-thread)
    (active :initform t :accessor client-active-p)))
 
@@ -115,7 +117,9 @@
 
 (defun handle-client (server client-socket)
   "Handle a single client connection."
-  (let ((stream (usocket:socket-stream client-socket)))
+  (let ((stream (flexi-streams:make-flexi-stream
+                 (usocket:socket-stream client-socket)
+                 :external-format :latin-1)))
     (handler-case
         (let* ((request-line (read-line stream))
                (headers (read-http-headers stream)))
@@ -124,7 +128,9 @@
             (let ((mount (gethash path (server-mounts server))))
               (if mount
                   (serve-stream server client-socket stream mount wants-meta)
-                  (send-404 stream path)))))
+                  (progn
+                    (log:debug "404 for path: ~A" path)
+                    (send-404 stream path))))))
       (error (e)
         (log:debug "Client error: ~A" e)
         (ignore-errors (usocket:socket-close client-socket))))))
@@ -165,25 +171,33 @@
       (log:info "Client disconnected from ~A" (mount-path mount)))))
 
 (defun stream-to-client (client)
-  "Stream audio data to a client, inserting metadata as needed."
+  "Stream audio data to a client from the broadcast buffer.
+   Starts with a burst of recent data for fast playback start."
   (let* ((mount (client-mount client))
          (buffer (mount-buffer mount))
          (stream (client-stream client))
          (chunk-size 4096)
          (chunk (make-array chunk-size :element-type '(unsigned-byte 8))))
+    ;; Start from burst position for fast playback
+    (setf (client-read-pos client) (buffer-burst-start buffer))
     (loop while (client-active-p client)
-          do (let ((bytes-read (buffer-read buffer chunk :blocking t)))
+          do (multiple-value-bind (bytes-read new-pos)
-               (when (zerop bytes-read)
+                 (buffer-read-from buffer (client-read-pos client) chunk)
-                 (sleep 0.01)
+               (if (zerop bytes-read)
-                 (return))
+                   ;; No data yet — wait for producer
-               (handler-case
+                   (buffer-wait-for-data buffer (client-read-pos client))
-                   (if (client-wants-metadata-p client)
+                   (progn
-                       (write-with-metadata client chunk bytes-read)
+                     (setf (client-read-pos client) new-pos)
-                       (write-sequence chunk stream :end bytes-read))
+                     (handler-case
-                 (error ()
+                         (progn
-                   (setf (client-active-p client) nil)
+                           (if (client-wants-metadata-p client)
-                   (return)))
+                               (write-with-metadata client chunk bytes-read)
-               (force-output stream)))))
+                               (write-sequence chunk stream :end bytes-read))
+                           (force-output stream))
+                       (error (e)
+                         (log:debug "Client stream error: ~A" e)
+                         (setf (client-active-p client) nil)
+                         (return)))))))))
 
 (defun write-with-metadata (client data length)
   "Write audio data with ICY metadata injection."

cl-streamer/test-stream.lisp

@@ -0,0 +1,79 @@
+;;; End-to-end streaming test with playlist
+;;; Usage: sbcl --load test-stream.lisp
+;;;
+;;; Then open http://localhost:8000/stream.mp3 in VLC or browser
+;;; ICY metadata will show track names as they change.
+
+(push #p"/home/glenn/SourceCode/harmony/" asdf:*central-registry*)
+(push #p"/home/glenn/SourceCode/asteroid/cl-streamer/" asdf:*central-registry*)
+
+(ql:quickload '(:cl-streamer :cl-streamer/encoder :cl-streamer/harmony))
+
+(format t "~%=== CL-Streamer Playlist Test ===~%")
+(format t "LAME version: ~A~%" (cl-streamer::lame-version))
+
+;; 1. Create and start stream server
+(format t "~%[1] Starting stream server on port 8000...~%")
+(cl-streamer:start :port 8000)
+
+;; 2. Add mount point
+(format t "[2] Adding mount point /stream.mp3...~%")
+(cl-streamer:add-mount cl-streamer:*server* "/stream.mp3"
+                       :content-type "audio/mpeg"
+                       :bitrate 128
+                       :name "Asteroid Radio (CL-Streamer Test)")
+
+;; 3. Create MP3 encoder
+(format t "[3] Creating MP3 encoder (128kbps, 44100Hz, stereo)...~%")
+(defvar *encoder* (cl-streamer:make-mp3-encoder :sample-rate 44100
+                                                 :channels 2
+                                                 :bitrate 128))
+
+;; 4. Create and start audio pipeline
+(format t "[4] Starting audio pipeline with Harmony...~%")
+(defvar *pipeline* (cl-streamer/harmony:make-audio-pipeline
+                    :encoder *encoder*
+                    :stream-server cl-streamer:*server*
+                    :mount-path "/stream.mp3"
+                    :sample-rate 44100
+                    :channels 2))
+
+(cl-streamer/harmony:start-pipeline *pipeline*)
+
+;; 5. Build a playlist from the music library
+(format t "[5] Building playlist from music library...~%")
+(defvar *music-dir* #p"/home/glenn/SourceCode/asteroid/music/library/")
+
+(defvar *playlist*
+  (let ((files nil))
+    (dolist (dir (directory (merge-pathnames "*/" *music-dir*)))
+      (dolist (flac (directory (merge-pathnames "**/*.flac" dir)))
+        (push (list :file (namestring flac)
+                    :title (format nil "~A - ~A"
+                                   (car (last (pathname-directory flac)))
+                                   (pathname-name flac)))
+              files)))
+    ;; Shuffle and take first 10 tracks
+    (subseq (alexandria:shuffle (copy-list files))
+            0 (min 10 (length files)))))
+
+(format t "Queued ~A tracks:~%" (length *playlist*))
+(dolist (entry *playlist*)
+  (format t "  ~A~%" (getf entry :title)))
+
+;; 6. Start playlist playback
+(format t "~%[6] Starting playlist...~%")
+(cl-streamer/harmony:play-list *pipeline* *playlist*)
+
+(format t "~%=== Stream is live! ===~%")
+(format t "Listen at: http://localhost:8000/stream.mp3~%")
+(format t "~%Press Enter to stop...~%")
+
+(read-line)
+
+;; Cleanup
+(format t "Stopping...~%")
+(cl-streamer/harmony:stop-pipeline *pipeline*)
+(cl-streamer:close-encoder *encoder*)
+(cl-streamer:stop)
+(format t "Done.~%")