syndicate-rkt/syndicate/broker/server.rkt

#lang imperative-syndicate

(provide (struct-out server-connection)
         (struct-out server-inbound)
         (struct-out server-outbound)
         (struct-out server-proposal)
         (struct-out server-envelope))

(require "wire-protocol.rkt")
(require racket/set)

;; Internal connection protocol
(assertion-struct server-connection (connection-id scope))
(assertion-struct server-inbound (connection-id body))
(assertion-struct server-outbound (connection-id body))

;; Internal isolation
(assertion-struct server-proposal (scope body)) ;; suggestions (~ actions)
(assertion-struct server-envelope (scope body)) ;; decisions (~ events)

(spawn #:name 'server-connection-factory

       (during/spawn (server-connection _ _)
         ;; Previously, we just had server-envelope. Now, we have both
         ;; server-envelope and server-proposal. While not everything
         ;; decided is (locally) suggested, it is true that everything
         ;; suggested is decided (in this implementation at least),
         ;; and the following clause reflects this:
         (during (server-proposal $scope $assertion)
           (assert (server-envelope scope assertion))))

       (during/spawn (server-connection $id $scope)
         (define endpoints (set))

         (on (message (server-inbound id (Assert $ep $a)))
             (when (not (set-member? endpoints ep))
               (set! endpoints (set-add endpoints ep))
               (react
                (on-stop (set! endpoints (set-remove endpoints ep)))

                (field [assertion a])

                (assert (server-proposal scope (assertion)))

                (let ((! (lambda (ctor) (lambda (cs) (send! (server-outbound id (ctor ep cs)))))))
                  (add-observer-endpoint! (lambda ()
                                            (let ((a (assertion)))
                                              (when (observe? a)
                                                (server-envelope scope (observe-specification a)))))
                                          #:on-add (! Add)
                                          #:on-remove (! Del)
                                          #:on-message (! Msg)))

                (on (message (server-inbound id (Assert ep $new-a)))
                    (assertion new-a))

                (stop-when (message (server-inbound id (Clear ep)))))))

         (on (message (server-inbound id (Message $body)))
             (send! (server-envelope scope body)))

         (on (message (server-inbound id (Ping)))
             (send! (server-outbound id (Pong))))))