Multicast channels using IVars

mcast-channel.rkt

#lang racket/base

;; uses https://github.com/samdphillips/racket-syncvar

(require racket/contract
         racket/match
         "ivar.rkt")

(provide mcast-channel?
         mcast-output?
         make-mcast-channel

         (contract-out
          [mcast-channel-put (-> mcast-channel? any/c any)]
          [make-mcast-output (-> mcast-channel? any)]
          [mcast-output-get  (-> mcast-output? any)]))

(struct mcast-channel (req-ch rpy-ch))
(struct mcast-output (ch)
  #:property prop:evt 0)

(define (make-mcast-channel)
  (define req-ch (make-channel))
  (define rpy-ch (make-channel))
  (define (server iv)
    (match (channel-get req-ch)
      ['new-port
       (channel-put rpy-ch (make-mcast-output-tee iv))
       (server iv)]
      [(list 'message value)
       (let ([next-iv (make-ivar)])
         (ivar-put! iv (cons value next-iv))
         (server next-iv))]))
  (thread (lambda () (server (make-ivar))))
  (mcast-channel req-ch rpy-ch))

(define (make-mcast-output-tee iv)
  (define output-ch (make-channel))
  (define (tee iv)
    (match-define (cons value next-iv) (ivar-get iv))
    (channel-put output-ch value)
    (tee next-iv))
  (thread (lambda () (tee iv)))
  (mcast-output output-ch))

(define (mcast-channel-put a-mcast-channel value)
  (channel-put (mcast-channel-req-ch a-mcast-channel) (list 'message value)))

(define (make-mcast-output a-mcast-channel)
  (channel-put (mcast-channel-req-ch a-mcast-channel) 'new-port)
  (channel-get (mcast-channel-rpy-ch a-mcast-channel)))

(define (mcast-output-get a-mcast-output)
  (sync a-mcast-output))