messagebus/messagebus.cl

(defpackage :uhub
  (:use :bt)
  (:use :cl))

(in-package :uhub)

(defclass bus ()
  ((buffer :initform (make-array 2000 :initial-element (cons 0 nil)))
   (index :initform (list 0))))

(defun subscribe (bus fun)
  (loop for msg = (with-slots (buffer) bus
                    (destructuring-bind (idx . msg) (aref buffer (mod index (array-dimension buffer 0)))
                      (when (= idx index) msg)))
        with index = 0
        if msg
                do (progn (funcall fun msg)
                          (incf index))
        else
                do (sleep 0.1)))

(defun publish (bus msg)
  (with-slots (buffer index) bus
    (let ((index (sb-ext:atomic-incf (car index))))
      (setf (aref buffer (mod index (array-dimension buffer 0))) (cons index msg)))))

(defun test ()
  (let ((bus (make-instance 'bus))
        sub)
    (unwind-protect
      (progn (setf sub (bt:make-thread (lambda () (subscribe bus (lambda (msg) (print msg))))))
             (sleep 0.1)
             (publish bus "Hello world!1")
             (publish bus "Hello world!2")
             (sleep 0.1)
             (publish bus "Hello world!3")
             (sleep 0.1))
      (when sub (sb-thread:terminate-thread sub)))))

(test)