syndicate-rkt/OLD-syndicate/distributed/client.rkt

114 lines
4.1 KiB
Racket

;;; SPDX-License-Identifier: LGPL-3.0-or-later
;;; SPDX-FileCopyrightText: Copyright © 2010-2021 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
#lang syndicate
(provide generic-client-session-facet)
(require "wire-protocol.rkt")
(require "internal-protocol.rkt")
(require "protocol.rkt")
(require "turn.rkt")
(require syndicate/term)
(require/activate "heartbeat.rkt")
(define-logger syndicate/distributed)
(spawn #:name 'client-factory
(during (to-server $a _) (assert (server-connection a)))
(during (observe (from-server $a _)) (assert (server-connection a)))
(during (observe (server-connected $a)) (assert (server-connection a))))
(struct sub (spec [captures #:mutable]) #:transparent)
(define (generic-client-session-facet address scope w)
(on-start (log-syndicate/distributed-info "Connected to ~v" address))
(on-stop (log-syndicate/distributed-info "Disconnected from ~v" address))
(assert (server-connected address))
(assert (server-session-connected address))
(when (log-level? syndicate/distributed-logger 'debug)
(set! w (let ((w* w))
(lambda (p)
(log-syndicate/distributed-debug "C OUT ~v ~v" address p)
(w* p)))))
(define turn (turn-recorder (lambda (items) (w (Turn items)))))
(define next-ep
(let ((counter 0))
(lambda ()
(begin0 counter
(set! counter (+ counter 1))))))
(define pubs (hash))
(define subs (hash))
(define matches (hash))
(on-start (w (Connect scope)))
(on-stop (for* [(s (in-hash-values matches)) (a (in-hash-values (sub-captures s)))] (retract! a)))
(define (instantiate s vs)
(instantiate-term->value (from-server address (sub-spec s)) vs))
(on (asserted (to-server address $a))
(define ep (next-ep))
(extend-turn! turn (Assert ep a))
(set! pubs (hash-set pubs a ep)))
(on (retracted (to-server address $a))
(define ep (hash-ref pubs a))
(extend-turn! turn (Clear ep))
(set! pubs (hash-remove pubs a)))
(on (message (to-server address $a))
(extend-turn! turn (Message a)))
(on (asserted (observe (from-server address $spec)))
(define ep (next-ep))
(extend-turn! turn (Assert ep (observe spec)))
(set! subs (hash-set subs spec ep))
(set! matches (hash-set matches ep (sub spec (hash)))))
(on (retracted (observe (from-server address $spec)))
(extend-turn! turn (Clear (hash-ref subs spec)))
(set! subs (hash-remove subs spec)))
(define reset-heartbeat! (heartbeat (list 'client address scope)
w
(lambda () (stop-current-facet))))
(on (message (server-packet address _))
(reset-heartbeat!))
(on (message (server-packet address (Ping)))
(w (Pong)))
(on (message (server-packet address (Err $detail $context)))
(log-syndicate/distributed-error "Error from ~a: ~v~a"
address
detail
(if context
(format " ~v" context)
""))
(stop-current-facet))
(on (message (server-packet address (Turn $items)))
(for [(item (in-list items))]
(match item
[(Add ep vs) (let* ((s (hash-ref matches ep))
(a (instantiate s vs)))
(set-sub-captures! s (hash-set (sub-captures s) vs a))
(assert! a))]
[(Del ep vs) (let* ((s (hash-ref matches ep))
(a (hash-ref (sub-captures s) vs)))
(retract! a)
(set-sub-captures! s (hash-remove (sub-captures s) vs)))]
[(Msg ep vs) (let* ((s (hash-ref matches ep)))
(send! (instantiate s vs)))]
[(End ep) (let* ((s (hash-ref matches ep #f)))
(when s
(for [(a (in-hash-values (sub-captures s)))] (retract! a))
(set! matches (hash-remove matches ep))))]))))