add queue.rkt

racket/queue.rkt

@@ -0,0 +1,59 @@
+#lang racket/base
+
+(require racket/async-channel racket/function racket/match racket/set)
+
+(define (multiqueue-thd)
+  (break-enabled #f)
+  (define clients (weak-seteq))
+  (define (handle-break _)
+    (void "do nothing, for now"))
+  (with-handlers ([exn:break? handle-break])
+    (let loop ()
+      (define evt (thread-receive-evt))
+      (sync/enable-break evt)
+      (match (thread-receive)
+        [(? channel? ch)
+         (define ach (make-async-channel))
+         (set-add! clients ach)
+         (channel-put ch ach)]
+        [v (for ([ach (in-set clients)])
+             (printf "sending message ~a\n" v)
+             (async-channel-put ach v))])
+      (loop))))
+
+(define (make-mq)
+  (thread multiqueue-thd))
+
+(define (mq-destroy q)
+  (break-thread q))
+
+(define (mq-register q)
+  (define ch (make-channel))
+  (thread-send q ch)
+  (channel-get ch))
+
+(define (mq-send q v)
+  (thread-send q v))
+
+(module+ main
+  (define (test-thread q x)
+    (define ach (mq-register q))
+    (let loop ()
+      (printf "~a got ~a\n" x (async-channel-get ach))
+      (loop)))
+
+  (define q (make-mq))
+  (define q1 (thread (thunk (test-thread q 1))))
+  (define q2 (thread (thunk (test-thread q 2))))
+  (define q3 (thread (thunk (test-thread q 3))))
+
+  (sleep 1)
+  (mq-send q "meow")
+  (mq-send q 'shonks)
+  (sleep 1)
+  (kill-thread q1)
+  (thread-wait q1)
+  (set! q1 #f)
+  (collect-garbage)
+  (mq-send q "meow 2")
+  (sleep 1))