cl-streamer/README.org

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

cl-streamer/buffer.lisp

@@ -1,87 +1,73 @@
 (in-package #:cl-streamer)
 
-;;; ---- 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 ()
+(defclass ring-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 "broadcast-buffer-lock") :reader buffer-lock)
+   (lock :initform (bt:make-lock "ring-buffer-lock") :reader buffer-lock)
    (not-empty :initform (bt:make-condition-variable :name "buffer-not-empty")
               :reader buffer-not-empty)
-   (burst-size :initarg :burst-size :reader buffer-burst-size
-               :initform (* 32 1024)
-               :documentation "Bytes of recent data to send on new client connect")))
+   (not-full :initform (bt:make-condition-variable :name "buffer-not-full")
+             :reader buffer-not-full)))
 
 (defun make-ring-buffer (size)
-  "Create a broadcast ring buffer with SIZE bytes capacity."
-  (make-instance 'broadcast-buffer
+  "Create a ring buffer with SIZE bytes capacity."
+  (make-instance 'ring-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 into the broadcast buffer. Never blocks; overwrites old data."
+  "Write bytes from DATA to BUFFER. Blocks if buffer is full."
   (let ((len (- end start)))
-    (when (> len 0)
-      (bt:with-lock-held ((buffer-lock buffer))
-        (let ((write-pos (buffer-write-pos buffer))
-              (size (buffer-size buffer))
-              (buf-data (buffer-data buffer)))
-          (loop for i from start below end
-                for j = (mod write-pos size) then (mod (1+ j) size)
-                do (setf (aref buf-data j) (aref data i))
-                finally (setf (buffer-write-pos buffer) (+ write-pos len))))
-        (bt:condition-notify (buffer-not-empty buffer))))
+    (bt:with-lock-held ((buffer-lock buffer))
+      (loop while (< (buffer-free-space buffer) len)
+            do (bt:condition-wait (buffer-not-full buffer) (buffer-lock buffer)))
+      (let ((write-pos (buffer-write-pos buffer))
+            (size (buffer-size buffer))
+            (buf-data (buffer-data buffer)))
+        (loop for i from start below end
+              for j = write-pos then (mod (1+ j) size)
+              do (setf (aref buf-data j) (aref data i))
+              finally (setf (buffer-write-pos buffer) (mod (1+ j) size))))
+      (bt:condition-notify (buffer-not-empty buffer)))
     len))
 
-(defun buffer-read-from (buffer read-pos output &key (start 0) (end (length output)))
-  "Read bytes from BUFFER starting at READ-POS into OUTPUT.
-   Returns (values bytes-read new-read-pos).
-   READ-POS is the client's absolute position in the stream."
+(defun buffer-read (buffer output &key (start 0) (end (length output)) (blocking t))
+  "Read bytes from BUFFER into OUTPUT. Returns number of bytes read.
+   If BLOCKING is T, waits for data. Otherwise returns 0 if empty."
   (let ((requested (- end start)))
     (bt:with-lock-held ((buffer-lock buffer))
-      (let* ((write-pos (buffer-write-pos buffer))
+      (when blocking
+        (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))
-             ;; Clamp read-pos: if client is too far behind, skip ahead
-             (oldest-available (max 0 (- write-pos size)))
-             (effective-read (max read-pos oldest-available))
-             (available (- write-pos effective-read))
-             (to-read (min requested available)))
-        (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)))
+             (buf-data (buffer-data buffer)))
+        (loop for i from start below (+ start to-read)
+              for j = read-pos then (mod (1+ j) size)
+              do (setf (aref output i) (aref buf-data j))
+              finally (setf (buffer-read-pos buffer) (mod (1+ j) size)))
+        (bt:condition-notify (buffer-not-full buffer))
+        to-read))))
 
 (defun buffer-clear (buffer)
-  "Clear the buffer."
+  "Clear all data from the buffer."
   (bt:with-lock-held ((buffer-lock buffer))
-    (setf (buffer-write-pos buffer) 0)))
+    (setf (buffer-read-pos buffer) 0
+          (buffer-write-pos buffer) 0)
+    (bt:condition-notify (buffer-not-full buffer))))

cl-streamer/harmony-backend.lisp

@@ -1,183 +0,0 @@
-(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,13 +7,11 @@
    #:encoding-error
    
    ;; Buffer
-   #:broadcast-buffer
+   #:ring-buffer
    #:make-ring-buffer
    #:buffer-write
-   #:buffer-read-from
-   #:buffer-wait-for-data
-   #:buffer-current-pos
-   #:buffer-burst-start
+   #:buffer-read
+   #:buffer-available
    #:buffer-clear
    
    ;; ICY Protocol
@@ -36,23 +34,5 @@
    #:listener-count
    
    ;; Main API
-   #:*server*
    #:*default-port*
-   #:*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))
+   #:*default-metaint*))

cl-streamer/stream-server.lisp

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

cl-streamer/test-stream.lisp

@@ -1,79 +0,0 @@
-;;; 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.~%")