async SSE server

(ql:quickload (list :cl-ppcre :cl-async :cl-who :parenscript :babel))
(defpackage :ts-async
  (:use :cl :cl-ppcre :cl-async :cl-who :parenscript)
  (:import-from :babel #:octets-to-string))
(in-package :ts-async)

;; Requirements:
;; apt-get install libevent-core-2.0 libevent-extra-2.0

(defparameter *channel* nil)

(defun hello-server ()
  (format t "Starting server.~%")
  (as:tcp-server "0.0.0.0" 3000
		 (lambda (socket data)
		   (let ((req (parse (octets-to-string data))))
		     (format t "~s~%~%" req)
		     (cond ((string= "/sub" req)
			    (subscribe! socket))
			   ((string= "/pub" req)
			    (publish! "Got a message!")
			    (write/close socket (cat "HTTP/1.1 200 OK" crlf
						     "Content-Type: text/plain; charset=UTF-8" crlf
						     "Cache-Control: no-cache, no-store, must-revalidate" crlf
						     "Content-Length: 10" crlf crlf
						     "Published!" crlf crlf)))
			   ((string= "/test" req)
			    (write/close socket test-page))
			   (t
			    (write/close socket (cat "HTTP/1.1 200 OK" crlf
						     "Content-Type: text/plain; charset=UTF-9" crlf
						     "Content-Length: 2" crlf crlf
						     "Ok" crlf crlf))))))
		 (lambda (err)
		   (format t "listener event: ~a" err)))
  (as:signal-handler 2 (lambda (sig)
			 (declare (ignore sig))
			 (as:exit-event-loop))))

(defparameter day-names '("Mon" "Tue" "Wed" "Thu" "Fri" "Sat" "Sun"))
(defparameter month-names '("Jan" "Feb" "Mar" "Apr" "May" "Jun" "Jul" "Aug" "Sep" "Oct" "Nov" "Dec"))

(defun http-date ()
  (multiple-value-bind (second minute hour date month year day-of-week dst-p tz)
      (get-decoded-time)
    (format nil "~a, ~a ~a ~a ~a:~a:~a GMT~@d"
	    (nth day-of-week day-names) date
	    (nth month month-names) year hour minute second (- tz))))

(defun subscribe! (sock)
  (write/keep-alive
   sock (cat "HTTP/1.1 200 OK" crlf
	     "Date: " (http-date) crlf
	     "Content-Type: text/event-stream; charset=utf-8" crlf
	     "Transfer-Encoding: chunked" crlf
	     "Connection: keep-alive" crlf
	     "Expires: Thu, 01 Jan 1970 00:00:01 GMT" crlf
	     "Cache-Control: no-cache, no-store, must-revalidate" crlf crlf))
  (push sock *channel*))

(defun publish! (msg)
  (loop for sock in *channel*
     do (handler-case
	    (if (as:socket-closed sock)
		(setf *channel* (remove sock *channel*))
		(write/keep-alive sock (cat (format nil "data: ~a" msg) crlf crlf)))
	  (error (e)
	    (declare (ignore e))
	    (format t "Removing inactive socket...")
	    (setf *channel* (remove sock *channel*))))))

(defun write/keep-alive (sock data)
  (unless (as:socket-closed sock)
    (as:write-socket-data sock data)))

(defun write/close (sock data)
  (unless (as:socket-closed sock)
    (as:write-socket-data
     sock data
     :write-cb (lambda (sock)
		 (setf (as:socket-data sock) nil)
		 (as:close-socket sock)))))

(defun start () (as:start-event-loop #'hello-server))

(defmethod parse ((req string))
  (let ((lines (split "\\r?\\n" req)))
    (second (split " " (first lines)))))

;;;;;;;;;; General purpose utilities
(defun cat (&rest seqs)
  (apply #'concatenate 'string seqs))
(defconstant crlf (list #\return #\linefeed))
(defparameter test-page
  (let ((content (with-html-output-to-string (str nil :prologue t)
		   (:html
		    (:head (:title "Test page"))
		    (:body
		     (:div :id "console")
		     (:script
		      :type "text/javascript"
		      (str (ps (defvar src (new (-event-source "/sub")))
			       (defun p (msg)
				 (let ((elem (chain document (get-element-by-id "console"))))
				   (setf (@ elem inner-h-t-m-l)
					 (+ (@ elem inner-h-t-m-l) "<p>" msg "</p>"))))
			       (setf (@ src onerror)
				     (lambda (e)
				       (p "ERROR OCCURRED...")
				       (p (chain -j-s-o-n (stringify e))))
				     (@ src onopen)
				     (lambda (e) (p "STREAM OPENED..."))
				     (@ src onmessage)
				     (lambda (e) (p "GOT MESSAGE!")))))))))))
    (cat "HTTP/1.1 200 OK" crlf
	 "Content-Type: text/html; charset=UTF-8" crlf
	 "Cache-Control: no-cache, no-store, must-revalidate" crlf
	 "Content-Length: " (write-to-string (length content)) crlf crlf
	 content crlf crlf)))