347 lines
14 KiB
Racket
347 lines
14 KiB
Racket
#lang syndicate/actor
|
|
|
|
(provide )
|
|
|
|
(require racket/dict)
|
|
(require racket/exn)
|
|
(require racket/format)
|
|
(require racket/set)
|
|
(require net/url)
|
|
(require net/uri-codec)
|
|
|
|
(require/activate syndicate/drivers/timer)
|
|
(require/activate syndicate/drivers/timestate)
|
|
(require/activate syndicate/drivers/web)
|
|
(require syndicate/protocol/advertise)
|
|
(require syndicate/functional-queue)
|
|
|
|
(require "private/util.rkt")
|
|
|
|
(module+ test (require rackunit))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
;; A Topic is a URIString.
|
|
|
|
;; A Deadline is an (Option Number), where #f indicates "unbounded",
|
|
;; and numbers indicate a moment in time as a Unix epoch timestamp as
|
|
;; might be returned from `current-seconds`.
|
|
|
|
;; (notification Topic Bytes String (Option URIString))
|
|
(struct notification (topic-name content content-type) #:prefab)
|
|
|
|
;; (subscription Topic Deadline URLString URLString (Option Bytes))
|
|
(struct subscription (topic-name expiry-deadline canonical-hub callback secret) #:prefab)
|
|
|
|
;; (local-topic-config Topic (Option Number) (Option Number))
|
|
(struct local-topic-config (name max-age max-count) #:prefab)
|
|
|
|
;; (topic-demand Topic)
|
|
(struct topic-demand (topic-name) #:prefab)
|
|
|
|
;; (local-topic-demand String)
|
|
(struct local-topic-demand (name) #:prefab)
|
|
|
|
;; (local-host String)
|
|
(struct local-host (name) #:prefab)
|
|
|
|
;; (canonical-local-host String)
|
|
(struct canonical-local-host (name) #:prefab)
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
;; (define (number->path n)
|
|
;; (let ((top (quotient n 256))
|
|
;; (bot (remainder n 256)))
|
|
;; (define bot-piece (~r bot #:base 16 #:min-width 2 #:pad-string "0"))
|
|
;; (if (zero? top)
|
|
;; bot-piece
|
|
;; (string-append (number->path top) "-/" bot-piece))))
|
|
|
|
;; (module+ test
|
|
;; (check-equal? (number->path 0) "00")
|
|
;; (check-equal? (number->path 123) "7b")
|
|
;; (check-equal? (number->path 12345) "30-/39")
|
|
;; (check-equal? (number->path 123456789012345678901234567890)
|
|
;; "01-/8e-/e9-/0f-/f6-/c3-/73-/e0-/ee-/4e-/3f-/0a-/d2"))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(define (vh host-name) (web-virtual-host "http" host-name 7827))
|
|
|
|
(actor #:name 'main
|
|
|
|
(assert (local-host "localhost"))
|
|
|
|
(assert (canonical-local-host "localhost"))
|
|
|
|
(during (local-host $host-name)
|
|
(assert (vh host-name))
|
|
|
|
(on (web-request-get (id req) (vh host-name) ("" ()))
|
|
(actor*
|
|
(web-respond/xexpr! id
|
|
`(html
|
|
(body
|
|
(h1 "Status"
|
|
" "
|
|
,(url->string
|
|
(resource->url
|
|
(web-request-header-resource req)))))))))))
|
|
|
|
(actor #:name 'local-topic-manager
|
|
|
|
(field [topics (set)])
|
|
|
|
(during (local-host $host-name)
|
|
(on (web-request-incoming (id req) (vh host-name) 'put ("topic" (,$topic ())))
|
|
(when (not (set-member? (topics) topic))
|
|
(topics (set-add (topics) topic))
|
|
(assert! (local-topic-demand topic))
|
|
(retract! (local-topic-config topic ? ?))
|
|
(assert! (local-topic-config topic #f #f))) ;; TODO: maximums
|
|
(web-respond/bytes! id #""))
|
|
|
|
(on (web-request-incoming (id req) (vh host-name) 'delete ("topic" (,$topic ())))
|
|
(when (set-member? (topics) topic)
|
|
(topics (set-remove (topics) topic))
|
|
(retract! (local-topic-demand topic))
|
|
(retract! (local-topic-config topic ? ?)))
|
|
(web-respond/bytes! id #"")))
|
|
|
|
(during/actor (local-topic-demand $topic)
|
|
#:name (list 'local-topic topic)
|
|
(local-topic-main topic)))
|
|
|
|
(actor #:name 'topic-demand-analyzer
|
|
(define/query-set local-hosts (local-host $host-name) host-name)
|
|
(during/actor (topic-demand $full-topic)
|
|
#:name (list 'general-topic full-topic)
|
|
#:let [(local-hosts-snapshot (local-hosts))]
|
|
(with-handlers [(exn? (lambda (e)
|
|
(log-error "Topic demand error: ~a" (exn->string e))))]
|
|
(match-define (web-resource (web-virtual-host _ topic-host _) topic-path)
|
|
(url->resource (string->url full-topic)))
|
|
(define maybe-local-topic
|
|
(match topic-path [`("topic" (,topic ())) topic] [_ #f]))
|
|
(on-start
|
|
(general-topic-main full-topic
|
|
topic-host
|
|
maybe-local-topic
|
|
(set-member? local-hosts-snapshot topic-host))))))
|
|
|
|
(define (general-topic-main full-topic topic-host maybe-local-topic start-as-local?)
|
|
(define (local-state)
|
|
(react (stop-when (retracted (local-host topic-host)) (remote-state))
|
|
(assert (local-topic-demand maybe-local-topic))))
|
|
|
|
(define (remote-state)
|
|
(react (stop-when #:when maybe-local-topic (asserted (local-host topic-host)) (local-state))
|
|
(remote-topic-main full-topic)))
|
|
|
|
(if (and maybe-local-topic start-as-local?)
|
|
(local-state)
|
|
(remote-state)))
|
|
|
|
(define (req-content-type req)
|
|
(dict-ref (web-request-header-headers req) 'content-type "application/octet-stream"))
|
|
|
|
(define (local-topic-main topic)
|
|
(field [max-age #f]
|
|
[max-count #f])
|
|
|
|
(on (asserted (local-topic-config topic $age $count))
|
|
(max-age age)
|
|
(max-count count))
|
|
|
|
(on-start (log-info "Creating local topic ~v" topic))
|
|
(begin/dataflow
|
|
(log-info "Configured local topic ~v, max-age ~v, max-count ~v"
|
|
topic
|
|
(max-age)
|
|
(max-count)))
|
|
(on-stop (log-info "Terminating local topic ~v" topic))
|
|
|
|
(during (local-host $host-name)
|
|
(on (web-request-incoming (id req) (vh host-name) 'post ("topic" (,topic ())) $body)
|
|
(actor*
|
|
(define content-type (req-content-type req))
|
|
(log-info "Local topic ~a got ~v message ~v" topic content-type body)
|
|
(define full-topic (url->string (resource->url (web-request-header-resource req))))
|
|
(send! (notification full-topic body content-type))
|
|
(web-respond/status! id 201 #"Created")))))
|
|
|
|
(define (maintain-remote-subscription full-topic canonical-host-name sub-id)
|
|
(define callback (url->string
|
|
(resource->url
|
|
(web-resource (vh canonical-host-name) `("sub" (,sub-id ()))))))
|
|
(react
|
|
(field [next-upstream-check (current-inexact-milliseconds)])
|
|
(on (asserted (later-than (next-upstream-check)))
|
|
(log-info "Rechecking upstream for ~a" full-topic)
|
|
(define-values (resp response-body)
|
|
(web-request! 'get full-topic #:redirect-budget 5))
|
|
(log-info "Resource: ~v" resp))))
|
|
|
|
(define (remote-topic-main full-topic)
|
|
(define sub-id (random-hex-string 16))
|
|
(log-info "Remote sub endpoint ~a" sub-id)
|
|
(during (canonical-local-host $canonical-host-name)
|
|
(on-start (maintain-remote-subscription full-topic canonical-host-name sub-id))
|
|
(on (web-request-incoming (id req) (vh canonical-host-name) 'post ("sub" (,sub-id ())) $body)
|
|
(actor*
|
|
(define content-type (req-content-type req))
|
|
(log-info "Remote topic ~a got ~v message ~v" full-topic content-type body)
|
|
(send! (notification full-topic body content-type))
|
|
(web-respond/status! id 201 #"Created")))))
|
|
|
|
(actor #:name 'hub
|
|
(during (local-host $host-name)
|
|
(on (web-request-incoming (id req) (vh host-name) 'post ("hub" ()) $body)
|
|
;; Initially, I had an (actor* ...) form here for fault
|
|
;; isolation. However, this led to problems since I wanted
|
|
;; to use `assert!` and `retract!` to signal to the
|
|
;; `during/actor`, and the assertions were being lost as
|
|
;; the `actor*` terminated. So instead, I'm using Rackety
|
|
;; `with-handlers`.
|
|
(define ok?
|
|
(with-handlers [(values (lambda (e) #f))]
|
|
(define params (make-immutable-hash
|
|
(form-urlencoded->alist (bytes->string/utf-8 body))))
|
|
(define callback (hash-ref params 'hub.callback))
|
|
(define mode
|
|
(match (hash-ref params 'hub.mode)
|
|
["subscribe" 'subscribe]
|
|
["unsubscribe" 'unsubscribe]))
|
|
(define topic (hash-ref params 'hub.topic))
|
|
(define lease-seconds
|
|
(match (hash-ref params 'hub.lease_seconds "unbounded")
|
|
["unbounded" #f]
|
|
[n (string->number n)]))
|
|
(define secret-string (hash-ref params 'hub.secret #f))
|
|
(define secret-bytes (and secret-string (string->bytes/utf-8 secret-string)))
|
|
(define expiry-deadline (and lease-seconds (+ (current-seconds) lease-seconds)))
|
|
(define canonical-hub
|
|
(url->string (resource->url (web-request-header-resource req))))
|
|
(match mode
|
|
['subscribe
|
|
(if (subscription-change-validate "subscribe"
|
|
(or lease-seconds "unbounded")
|
|
topic
|
|
callback)
|
|
(begin
|
|
(retract! (subscription topic ? ? callback ?))
|
|
(assert!
|
|
(subscription topic expiry-deadline canonical-hub callback secret-bytes))
|
|
#t)
|
|
#f)]
|
|
['unsubscribe
|
|
(if (subscription-change-validate "unsubscribe"
|
|
#f
|
|
topic
|
|
callback)
|
|
(begin
|
|
(retract! (subscription topic ? ? callback ?))
|
|
#t)
|
|
#f)])))
|
|
(if ok?
|
|
(web-respond/status! id 202 #"Accepted")
|
|
(web-respond/status! id 403 #"Forbidden" #"Validation failed"))))
|
|
|
|
(during/actor (subscription $partial-topic _ _ $callback _)
|
|
#:name (list 'subscription partial-topic callback)
|
|
#:on-crash (retract! (subscription partial-topic ? ? callback ?))
|
|
(subscription-main partial-topic callback)))
|
|
|
|
(define (subscription-change-validate mode lease topic callback)
|
|
(define challenge (random-hex-string 16))
|
|
(define id (gensym 'validation))
|
|
(define extra-query (list* (cons 'hub.mode mode)
|
|
(cons 'hub.topic topic)
|
|
(cons 'hub.challenge challenge)
|
|
(if lease
|
|
(list (cons 'hub.lease_seconds (~a lease)))
|
|
(list))))
|
|
(define-values (resp body)
|
|
(web-request! 'get
|
|
(let* ((u (string->url callback)))
|
|
(url->string
|
|
(struct-copy url u [query (append (url-query u) extra-query)])))))
|
|
(and (eq? (web-response-header-code-type resp) 'successful)
|
|
(equal? body (string->bytes/utf-8 challenge))))
|
|
|
|
(define (canonical-topic-base-url canonical-hub-str)
|
|
(combine-url/relative (string->url canonical-hub-str) "/topic/"))
|
|
|
|
(define (subscription-main partial-topic callback)
|
|
(field [expiry-timer-id #f]
|
|
[delivery-active? #f]
|
|
[message-queue (make-queue)]
|
|
[dead-letters (make-queue)])
|
|
|
|
(stop-when (rising-edge (> (queue-length (dead-letters)) 1))
|
|
(log-info "Too many dead letters for ~a" callback))
|
|
|
|
(define (deliver-queued-notifications canonical-hub)
|
|
(delivery-active? #t)
|
|
(let deliver-rest ()
|
|
(when (not (queue-empty? (message-queue)))
|
|
(define-values (n newq) (dequeue (message-queue)))
|
|
(message-queue newq)
|
|
(match-define (notification topic body content-type) n)
|
|
(let deliver-one ((retries-remaining 1)
|
|
(retry-delay 5.0))
|
|
(define-values (resp _body)
|
|
(web-request! 'post
|
|
callback
|
|
#:headers (list (cons 'content-type content-type)
|
|
(cons 'link (format "<~a>; rel=hub" canonical-hub))
|
|
(cons 'link (format "<~a>; rel=self" topic)))
|
|
#:body body))
|
|
(cond
|
|
[(eq? (web-response-header-code-type resp) 'successful)
|
|
(deliver-rest)]
|
|
[(zero? retries-remaining)
|
|
(log-info "Dead letter for ~v" callback)
|
|
(dead-letters (enqueue (dead-letters) n))
|
|
(deliver-rest)]
|
|
[else
|
|
(log-info
|
|
"Delivery to ~v failed; pausing for ~a seconds; ~a retries remaining. Response: ~v"
|
|
callback
|
|
retry-delay
|
|
retries-remaining
|
|
resp)
|
|
(sleep retry-delay)
|
|
(deliver-one (- retries-remaining 1)
|
|
(min (* retry-delay 1.618) 30))]))))
|
|
(delivery-active? #f))
|
|
|
|
(during (subscription partial-topic $expiry-deadline $canonical-hub callback $secret-bytes)
|
|
(define topic (url->string (combine-url/relative (canonical-topic-base-url canonical-hub)
|
|
partial-topic)))
|
|
|
|
(assert (topic-demand topic))
|
|
|
|
(on-start (log-info "Subscription configured: ~v ~v ~v ~v"
|
|
topic
|
|
expiry-deadline
|
|
canonical-hub
|
|
secret-bytes)
|
|
(cond
|
|
[expiry-deadline
|
|
(expiry-timer-id (gensym 'subscription-expiry))
|
|
(log-info "Subscription will expire at ~a" expiry-deadline)
|
|
(send! (set-timer (expiry-timer-id) (* expiry-deadline 1000.0) 'absolute))]
|
|
[else
|
|
(log-info "Subscription will not expire")
|
|
(expiry-timer-id #f)]))
|
|
|
|
(stop-when #:when (expiry-timer-id) (message (timer-expired (expiry-timer-id) _))
|
|
(log-info "Subscription expired"))
|
|
|
|
(on (message ($ n (notification topic _ _)))
|
|
(message-queue (enqueue (message-queue) n))
|
|
(when (not (delivery-active?))
|
|
(deliver-queued-notifications canonical-hub)))))
|