syndicate-lang
/
racketmq-2017
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Asynchronous verification

This commit is contained in:
Tony Garnock-Jones 2016-11-09 05:21:30 +13:00
parent 91c59988df
commit 28bb423d32
1 changed files with 59 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
rmq/server.rkt
View File

@ -47,10 +47,14 @@
;; (notification Topic Bytes (Option String)) ;; (notification Topic Bytes (Option String))
(struct notification (topic-name content content-type) #:prefab) (struct notification (topic-name content content-type) #:prefab)
(struct subscription (topic-name ;; Topic ;; (update-subscription Topic URLString (Option SubscriptionSettings))
expiry-deadline ;; Deadline (struct update-subscription (topic callback settings) #:prefab) ;; message
;; (subscription Topic URLString SubscriptionSettings)
(struct subscription (topic callback settings-) #:prefab) ;; assertion
(struct subscription-settings (expiry-deadline ;; Deadline
canonical-hub ;; URLString canonical-hub ;; URLString
callback ;; URLString
secret ;; Option Bytes secret ;; Option Bytes
poll-interval-seconds) ;; Option Number poll-interval-seconds) ;; Option Number
#:prefab) #:prefab)
@ -334,24 +338,26 @@
(actor #:name 'hub (actor #:name 'hub
(during (local-host $host-name) (during (local-host $host-name)
(on (web-request-incoming (id req) (vh host-name) 'post ("hub" ()) $body) (on (web-request-incoming (id req) (vh host-name) 'post ("hub" ()) $body)
;; Initially, I had an (actor* ...) form here for fault (asynchronous-verification-of-intent id req body)
;; isolation. However, this led to problems since I wanted (web-respond/status! id 202 #"Accepted")))
;; to use `assert!` and `retract!` to signal to the
;; `during/actor`, and the assertions were being lost as (on (message (update-subscription $topic $callback $settings))
;; the `actor*` terminated. So instead, I'm using Rackety (retract! (subscription topic callback ?))
;; `with-handlers`. (when settings (assert! (subscription topic callback settings))))
(define ok?
(with-handlers [(values (lambda (e) #f))] (during/actor (subscription $topic $callback _)
(define params (make-immutable-hash #:name (list 'subscription topic callback)
(form-urlencoded->alist (bytes->string/utf-8 body)))) #:on-crash (retract! (subscription topic callback ?))
(subscription-main topic callback)))
(define (asynchronous-verification-of-intent id req body)
(actor* (define params (make-immutable-hash (form-urlencoded->alist (bytes->string/utf-8 body))))
(define callback (hash-ref params 'hub.callback)) (define callback (hash-ref params 'hub.callback))
(define mode (define mode (match (hash-ref params 'hub.mode)
(match (hash-ref params 'hub.mode)
["subscribe" 'subscribe] ["subscribe" 'subscribe]
["unsubscribe" 'unsubscribe])) ["unsubscribe" 'unsubscribe]))
(define topic (hash-ref params 'hub.topic)) (define topic (hash-ref params 'hub.topic))
(define lease-seconds (define lease-seconds (match (hash-ref params 'hub.lease_seconds *default-lease*)
(match (hash-ref params 'hub.lease_seconds *default-lease*)
["unbounded" #f] ["unbounded" #f]
[n (string->number n)])) [n (string->number n)]))
(define poll-interval-seconds (define poll-interval-seconds
@ -361,42 +367,22 @@
(define secret-string (hash-ref params 'hub.secret #f)) (define secret-string (hash-ref params 'hub.secret #f))
(define secret-bytes (and secret-string (string->bytes/utf-8 secret-string))) (define secret-bytes (and secret-string (string->bytes/utf-8 secret-string)))
(define expiry-deadline (and lease-seconds (+ (current-seconds) lease-seconds))) (define expiry-deadline (and lease-seconds (+ (current-seconds) lease-seconds)))
(define canonical-hub (define canonical-hub (url->string (resource->url (web-request-header-resource req))))
(url->string (resource->url (web-request-header-resource req))))
;; TODO: asynchronous validation
(match mode (match mode
['subscribe ['subscribe
(if (subscription-change-validate "subscribe" (when (subscription-change-validate "subscribe"
(or lease-seconds "unbounded") (or lease-seconds "unbounded")
topic topic
callback) callback)
(begin (send! (update-subscription topic
(retract! (subscription topic ? ? callback ? ?))
(assert! (subscription topic
expiry-deadline
canonical-hub
callback callback
(subscription-settings expiry-deadline
canonical-hub
secret-bytes secret-bytes
poll-interval-seconds)) poll-interval-seconds))))]
#t)
#f)]
['unsubscribe ['unsubscribe
(if (subscription-change-validate "unsubscribe" (when (subscription-change-validate "unsubscribe" #f topic callback)
#f (send! (update-subscription topic callback #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 (subscription-change-validate mode lease topic callback)
(define challenge (random-hex-string 16)) (define challenge (random-hex-string 16))
@ -461,12 +447,11 @@
*max-retry-delay*))])))) *max-retry-delay*))]))))
(delivery-active? #f)) (delivery-active? #f))
(during (subscription partial-topic (during (subscription partial-topic callback (subscription-settings
$expiry-deadline $expiry-deadline
$canonical-hub $canonical-hub
callback
$secret-bytes $secret-bytes
$poll-interval-seconds) $poll-interval-seconds))
(define topic (url->string (combine-url/relative (canonical-topic-base-url canonical-hub) (define topic (url->string (combine-url/relative (canonical-topic-base-url canonical-hub)
partial-topic))) partial-topic)))