Introduce subscribe/fresh and use it in some places.
This commit is contained in:
parent
fe88c1cbb6
commit
a891956867
16
driver.rkt
16
driver.rkt
|
@ -42,14 +42,14 @@
|
||||||
(define boot-server
|
(define boot-server
|
||||||
(os-big-bang 'no-state
|
(os-big-bang 'no-state
|
||||||
(send-meta-message `(request create-server-socket (udp new ,port-number 512)))
|
(send-meta-message `(request create-server-socket (udp new ,port-number 512)))
|
||||||
(subscribe 'wait-for-server-socket
|
(subscribe/fresh wait-id
|
||||||
(meta-message-handlers w
|
(meta-message-handlers w
|
||||||
[`(reply create-server-socket ,s)
|
[`(reply create-server-socket ,s)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe 'wait-for-server-socket)
|
(unsubscribe wait-id)
|
||||||
(spawn (dns-read-driver s))
|
(spawn (dns-read-driver s))
|
||||||
(spawn (dns-write-driver s))
|
(spawn (dns-write-driver s))
|
||||||
(subscribe 'packet-handler (packet-handler s)))]))))
|
(subscribe 'packet-handler (packet-handler s)))]))))
|
||||||
|
|
||||||
(define (packet-handler s)
|
(define (packet-handler s)
|
||||||
(message-handlers old-state
|
(message-handlers old-state
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
(mapcar #'(lambda (x) (put x 'scheme-indent-function 1))
|
(mapcar #'(lambda (x) (put x 'scheme-indent-function 1))
|
||||||
'(transition extend-transition
|
'(transition extend-transition
|
||||||
subscribe unsubscribe
|
subscribe subscribe/fresh unsubscribe
|
||||||
os-big-bang
|
os-big-bang
|
||||||
message-handlers meta-message-handlers ground-message-handler))
|
message-handlers meta-message-handlers ground-message-handler))
|
||||||
|
|
|
@ -171,11 +171,10 @@
|
||||||
[remaining-addresses (hash-ref known-addresses
|
[remaining-addresses (hash-ref known-addresses
|
||||||
current-name)]
|
current-name)]
|
||||||
[remaining-names remaining-names]))
|
[remaining-names remaining-names]))
|
||||||
(let ((subscription-id (list 'nameserver-name-resolution current-name))
|
(let ((subq (question current-name 'a 'in))) ;; TODO: 'aaaa ?
|
||||||
(subq (question current-name 'a 'in))) ;; TODO: 'aaaa ?
|
|
||||||
(transition (struct-copy network-query-state w [remaining-names remaining-names])
|
(transition (struct-copy network-query-state w [remaining-names remaining-names])
|
||||||
(send-message subq)
|
(send-message subq)
|
||||||
(subscribe subscription-id
|
(subscribe/fresh subscription-id
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[(answered-question (== subq) ans)
|
[(answered-question (== subq) ans)
|
||||||
(define ips
|
(define ips
|
||||||
|
|
|
@ -14,16 +14,15 @@
|
||||||
(define read-line-driver-handler
|
(define read-line-driver-handler
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[`(request ,reply-addr read-line)
|
[`(request ,reply-addr read-line)
|
||||||
(define sid `(read-line-transaction ,reply-addr))
|
|
||||||
(transition w
|
(transition w
|
||||||
(subscribe sid
|
(subscribe/fresh sid
|
||||||
(ground-message-handler w
|
(ground-message-handler w
|
||||||
[((list 'read-line reply-addr)
|
[((list 'read-line reply-addr)
|
||||||
(read-line-evt (current-input-port) 'any)
|
(read-line-evt (current-input-port) 'any)
|
||||||
=> l)
|
=> l)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe sid)
|
(unsubscribe sid)
|
||||||
(send-message `(reply ,reply-addr ,l)))])))]))
|
(send-message `(reply ,reply-addr ,l)))])))]))
|
||||||
|
|
||||||
;; This should be part of racket
|
;; This should be part of racket
|
||||||
(define (time-evt msecs)
|
(define (time-evt msecs)
|
||||||
|
@ -33,14 +32,14 @@
|
||||||
(let loop ((last-tick-time 0) (counter 0))
|
(let loop ((last-tick-time 0) (counter 0))
|
||||||
(define next-time (+ last-tick-time interval))
|
(define next-time (+ last-tick-time interval))
|
||||||
(subscribe self-sid
|
(subscribe self-sid
|
||||||
(ground-message-handler w
|
(ground-message-handler w
|
||||||
[((list 'timer-alarm next-time)
|
[((list 'timer-alarm next-time)
|
||||||
(time-evt next-time)
|
(time-evt next-time)
|
||||||
=> now)
|
=> now)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe self-sid)
|
(unsubscribe self-sid)
|
||||||
(send-message `(tick ,counter ,now))
|
(send-message `(tick ,counter ,now))
|
||||||
(loop now (+ counter 1)))]))))
|
(loop now (+ counter 1)))]))))
|
||||||
|
|
||||||
(define main
|
(define main
|
||||||
(os-big-bang 'none
|
(os-big-bang 'none
|
||||||
|
|
|
@ -8,6 +8,8 @@
|
||||||
(require "os.rkt")
|
(require "os.rkt")
|
||||||
|
|
||||||
(provide (struct-out subscribe)
|
(provide (struct-out subscribe)
|
||||||
|
subscribe/fresh
|
||||||
|
|
||||||
(struct-out unsubscribe)
|
(struct-out unsubscribe)
|
||||||
(struct-out send-message)
|
(struct-out send-message)
|
||||||
(struct-out send-meta-message)
|
(struct-out send-meta-message)
|
||||||
|
@ -68,6 +70,12 @@
|
||||||
;; representation of a suspended world and its active subscriptions.
|
;; representation of a suspended world and its active subscriptions.
|
||||||
(struct world (state subscriptions) #:transparent)
|
(struct world (state subscriptions) #:transparent)
|
||||||
|
|
||||||
|
(define-syntax subscribe/fresh
|
||||||
|
(syntax-rules ()
|
||||||
|
((_ id-binder event-description)
|
||||||
|
(let ((id-binder (gensym 'id-binder)))
|
||||||
|
(subscribe id-binder event-description)))))
|
||||||
|
|
||||||
(define-syntax message-handlers*
|
(define-syntax message-handlers*
|
||||||
(syntax-rules ()
|
(syntax-rules ()
|
||||||
((_ action-constructor old-state-pattern [pattern body ...] ...)
|
((_ action-constructor old-state-pattern [pattern body ...] ...)
|
||||||
|
|
49
os-dns.rkt
49
os-dns.rkt
|
@ -20,18 +20,16 @@
|
||||||
|
|
||||||
(define (dns-read-driver s)
|
(define (dns-read-driver s)
|
||||||
(os-big-bang 'no-state
|
(os-big-bang 'no-state
|
||||||
(subscribe 'packet-reader
|
(subscribe 'packet-reader
|
||||||
(meta-message-handlers w
|
(meta-message-handlers w
|
||||||
[(udp-packet source (== s) body)
|
[(udp-packet source (== s) body)
|
||||||
(transition w
|
(transition w
|
||||||
(send-message
|
(send-message
|
||||||
(with-handlers ((exn:fail? (lambda (e)
|
(with-handlers ((exn:fail? (lambda (e) (bad-dns-packet body source s 'unparseable))))
|
||||||
(bad-dns-packet body source s
|
(define message (packet->dns-message body))
|
||||||
'unparseable))))
|
(case (dns-message-direction message)
|
||||||
(define message (packet->dns-message body))
|
((request) (dns-request message source s))
|
||||||
(case (dns-message-direction message)
|
((response) (dns-reply message source s))))))]))))
|
||||||
((request) (dns-request message source s))
|
|
||||||
((response) (dns-reply message source s))))))]))))
|
|
||||||
|
|
||||||
(define (dns-write-driver s)
|
(define (dns-write-driver s)
|
||||||
(define (translate message sink)
|
(define (translate message sink)
|
||||||
|
@ -39,21 +37,20 @@
|
||||||
(send-message (bad-dns-packet message s sink 'unencodable)))))
|
(send-message (bad-dns-packet message s sink 'unencodable)))))
|
||||||
(send-meta-message (udp-packet s sink (dns-message->packet message)))))
|
(send-meta-message (udp-packet s sink (dns-message->packet message)))))
|
||||||
(os-big-bang 'no-state
|
(os-big-bang 'no-state
|
||||||
(subscribe 'packet-writer
|
(subscribe 'packet-writer
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[(dns-request message (== s) sink)
|
[(dns-request message (== s) sink) (transition w (translate message sink))]
|
||||||
(transition w (translate message sink))]
|
[(dns-reply message (== s) sink) (transition w (translate message sink))]))))
|
||||||
[(dns-reply message (== s) sink)
|
|
||||||
(transition w (translate message sink))]))))
|
|
||||||
|
|
||||||
(require racket/pretty)
|
(require racket/pretty)
|
||||||
(define dns-spy
|
(define dns-spy
|
||||||
(os-big-bang 'none
|
(os-big-bang 'none
|
||||||
(subscribe 'spy
|
(subscribe 'spy
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[(dns-request message source sink)
|
[(dns-request message source sink)
|
||||||
(pretty-display `(DNS (,source asks ,sink)
|
(pretty-display `(DNS (,source asks ,sink) ,@(dns-message-questions message)))]
|
||||||
,@(dns-message-questions message)))]
|
[(dns-reply message source sink)
|
||||||
[(dns-reply message source sink)
|
(pretty-display `(DNS (,source answers ,sink) ,message))]
|
||||||
(pretty-display `(DNS (,source answers ,sink) ,message))]
|
[x
|
||||||
[x (write `(DNS ,x)) (newline)]))))
|
(write `(DNS ,x))
|
||||||
|
(newline)]))))
|
||||||
|
|
47
os-timer.rkt
47
os-timer.rkt
|
@ -20,29 +20,30 @@
|
||||||
|
|
||||||
(define (timer-driver [self-id 'timer-driver])
|
(define (timer-driver [self-id 'timer-driver])
|
||||||
(os-big-bang 'no-state
|
(os-big-bang 'no-state
|
||||||
(subscribe 'timer-setter
|
(subscribe 'timer-setter
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[(set-timer label msecs relative?)
|
[(set-timer reply-label msecs relative?)
|
||||||
(transition w
|
(transition w
|
||||||
(subscribe label
|
(subscribe/fresh label
|
||||||
(ground-message-handler w
|
(ground-message-handler w
|
||||||
[((list self-id label)
|
[((list self-id label)
|
||||||
(timer-evt msecs relative?)
|
(timer-evt msecs relative?)
|
||||||
=> now)
|
=> now)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe label)
|
(unsubscribe label)
|
||||||
(send-message (timer-expired label now)))])))]))))
|
(send-message (timer-expired reply-label now)))])))]))))
|
||||||
|
|
||||||
(define (timer-relay [self-id 'timer-relay])
|
(define (timer-relay [self-id 'timer-relay])
|
||||||
(os-big-bang 'no-state
|
(os-big-bang 'no-state
|
||||||
(subscribe 'timer-relay
|
(subscribe 'timer-relay
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[(set-timer label msecs relative?)
|
[(set-timer reply-label msecs relative?)
|
||||||
(transition w
|
(define timer-id (list self-id reply-label))
|
||||||
(send-meta-message (set-timer (list self-id label) msecs relative?))
|
(transition w
|
||||||
(subscribe label
|
(send-meta-message (set-timer timer-id msecs relative?))
|
||||||
(meta-message-handlers w
|
(subscribe/fresh label
|
||||||
[(timer-expired (list (== self-id) (== label)) now)
|
(meta-message-handlers w
|
||||||
(transition w
|
[(timer-expired (== timer-id) now)
|
||||||
(unsubscribe label)
|
(transition w
|
||||||
(send-message (timer-expired label now)))])))]))))
|
(unsubscribe label)
|
||||||
|
(send-message (timer-expired reply-label now)))])))]))))
|
||||||
|
|
|
@ -22,17 +22,17 @@
|
||||||
|
|
||||||
(define echoer
|
(define echoer
|
||||||
(os-big-bang 'none
|
(os-big-bang 'none
|
||||||
(send-message `(request create-echo-socket (udp new 5555 65536)))
|
(send-message `(request create-echo-socket (udp new 5555 65536)))
|
||||||
(subscribe 'echo-socket-receiver
|
(subscribe/fresh sub
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[`(reply create-echo-socket ,sname)
|
[`(reply create-echo-socket ,sname)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe 'echo-socket-receiver)
|
(unsubscribe sub)
|
||||||
(subscribe 'packet-handler (packet-handler sname)))]))))
|
(subscribe 'packet-handler (packet-handler sname)))]))))
|
||||||
|
|
||||||
(define spy
|
(define spy
|
||||||
(os-big-bang 'none
|
(os-big-bang 'none
|
||||||
(subscribe 'spy (message-handlers w [x (write `(MESSAGE ,x)) (newline)]))))
|
(subscribe 'spy (message-handlers w [x (write `(MESSAGE ,x)) (newline)]))))
|
||||||
|
|
||||||
(define (main)
|
(define (main)
|
||||||
(ground-vm
|
(ground-vm
|
||||||
|
|
15
proxy.rkt
15
proxy.rkt
|
@ -35,22 +35,22 @@
|
||||||
(spawn (timer-relay))
|
(spawn (timer-relay))
|
||||||
(spawn (query-id-allocator))
|
(spawn (query-id-allocator))
|
||||||
(send-meta-message `(request create-server-socket (udp new ,port-number 512)))
|
(send-meta-message `(request create-server-socket (udp new ,port-number 512)))
|
||||||
(subscribe 'wait-for-server-socket
|
(subscribe/fresh wait-id
|
||||||
(meta-message-handlers w
|
(meta-message-handlers w
|
||||||
[`(reply create-server-socket ,s)
|
[`(reply create-server-socket ,s)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe 'wait-for-server-socket)
|
(unsubscribe wait-id)
|
||||||
(send-meta-message
|
(send-meta-message
|
||||||
`(request create-client-socket (udp new 0 512)))
|
`(request create-client-socket (udp new 0 512)))
|
||||||
(client-socket-waiter s))]))))
|
(client-socket-waiter s))]))))
|
||||||
|
|
||||||
(define (client-socket-waiter s)
|
(define (client-socket-waiter s)
|
||||||
(subscribe 'wait-for-client-socket
|
(subscribe/fresh wait-id
|
||||||
(meta-message-handlers w
|
(meta-message-handlers w
|
||||||
[`(reply create-client-socket ,c)
|
[`(reply create-client-socket ,c)
|
||||||
(display "Ready.") (newline)
|
(display "Ready.") (newline)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe 'wait-for-client-socket)
|
(unsubscribe wait-id)
|
||||||
(spawn (dns-read-driver s))
|
(spawn (dns-read-driver s))
|
||||||
(spawn (dns-write-driver s))
|
(spawn (dns-write-driver s))
|
||||||
(spawn (dns-read-driver c))
|
(spawn (dns-read-driver c))
|
||||||
|
@ -134,11 +134,11 @@
|
||||||
with query id ,(dns-message-id request-message))) (newline)
|
with query id ,(dns-message-id request-message))) (newline)
|
||||||
(os-big-bang 'no-state/packet-relay
|
(os-big-bang 'no-state/packet-relay
|
||||||
(send-message original-question)
|
(send-message original-question)
|
||||||
(subscribe 'wait-for-answer
|
(subscribe/fresh wait-id
|
||||||
(message-handlers w
|
(message-handlers w
|
||||||
[(answered-question (== original-question) answer)
|
[(answered-question (== original-question) answer)
|
||||||
(transition w
|
(transition w
|
||||||
(unsubscribe 'wait-for-answer)
|
(unsubscribe wait-id)
|
||||||
(send-message (answer->reply original-question answer)))])))]))
|
(send-message (answer->reply original-question answer)))])))]))
|
||||||
|
|
||||||
(define (question-dispatcher zone0 client-sock)
|
(define (question-dispatcher zone0 client-sock)
|
||||||
|
@ -202,10 +202,9 @@
|
||||||
[(partial-answer base cnames)
|
[(partial-answer base cnames)
|
||||||
(transition (expanding-cnames q base (length cnames))
|
(transition (expanding-cnames q base (length cnames))
|
||||||
(map (lambda (cname)
|
(map (lambda (cname)
|
||||||
(define subscription-id (list 'cname-expander cname))
|
|
||||||
(define cname-q (question cname (question-type q) (question-class q)))
|
(define cname-q (question cname (question-type q) (question-class q)))
|
||||||
(list (send-message cname-q)
|
(list (send-message cname-q)
|
||||||
(subscribe subscription-id
|
(subscribe/fresh subscription-id
|
||||||
(message-handlers (expanding-cnames q acc remaining)
|
(message-handlers (expanding-cnames q acc remaining)
|
||||||
[(answered-question (== cname-q) ans)
|
[(answered-question (== cname-q) ans)
|
||||||
(define new-acc (if ans (merge-answers acc ans) acc))
|
(define new-acc (if ans (merge-answers acc ans) acc))
|
||||||
|
|
Loading…
Reference in New Issue