Many improvements to drivers/http.rkt

syndicate-examples/http-server.rkt

@@ -52,4 +52,15 @@
                            `(h1 "Second page")
                            `(a ((href "/")) "Go back")))))
 
+      (spawn #:name 'server2
+        (at ds
+          (assert (HttpBinding #f 8080 'get (list (PathPatternElement-label "page3"))
+                               (object
+                                 [(HttpContext req res)
+                                  (at res
+                                    (log-info "+page3")
+                                    (on-stop (log-info "-page3"))
+                                    (send! (HttpResponse-header 'content-type "text/plain"))
+                                    (send! (HttpResponse-done "this is page3\n")))])))))
+
       )))

syndicate/drivers/http.rkt

@@ -110,18 +110,26 @@
          (turn! facet
                 (lambda ()
                   (react
-                    (local-connection-protocol ds connection-name conn routes req)
+                    (local-connection-protocol ds connection-name conn port routes req)
                     (on-stop (async-channel-put continue-ch 'dummy)))))
          (async-channel-get continue-ch)
          (unless close? (process-requests)))))))
 
 (define next-request-id (inexact->exact (floor (current-inexact-milliseconds))))
 
-(define (local-connection-protocol ds connection-name conn routes req)
+(define (local-connection-protocol ds connection-name conn port routes req)
   (let/ec return
 
+    (define reply-sent? #f)
+    (define (unless-reply-sent thunk)
+      (unless reply-sent?
+        (set! reply-sent? #t)
+        (thunk)))
+
     (define (respond! resp)
-      (output-response/method conn resp (request-method req))
+      (unless-reply-sent
+       (lambda ()
+         (output-response/method conn resp (request-method req))))
       (stop-current-facet))
 
     (define (decode-bytes bs)
@@ -148,10 +156,11 @@
     (define headers (headers-map (request-headers/raw req)))
 
     (define host-and-port (cond [(assq 'host (request-headers req)) => cdr] [else #f]))
-    (define-values (host port)
+    (define host
       (match host-and-port
-        [(regexp #px"(.*):(\\d+)" (list _ host port)) (values host (string->number port))]
-        [host (values host 80)]))
+        [#f (RequestHost-absent)]
+        [(regexp #px"(.*):\\d+" (list _ host)) (RequestHost-present host)]
+        [host (RequestHost-present host)]))
 
     (define method (lowercase-symbol! (decode-bytes (request-method req))))
 
@@ -198,76 +207,74 @@
                                                              (list)))
                                     (return (void)))))
           (match-define (active-handler handler handler-terminated?) (set-first handler-set))
-          (define processing #f)
-          (define pending-code 500)
-          (define pending-message "Internal Server Error")
+          (define pending-code 200)
+          (define pending-message "OK")
           (define pending-headers-rev '())
           (define chunk-ch #f)
 
-          (define (check-processing!)
-            (unless processing
-              (error 'HttpResponse "Attempt to reply before <processing> has been asserted")))
-
-          (define (send-headers!)
+          (define (start-chunked-response!)
             (unless chunk-ch
               (set! chunk-ch (make-async-channel))
               (thread
                (lambda ()
-                 (output-response/method conn
-                                         (response pending-code
-                                                   (string->bytes/utf-8 pending-message)
-                                                   (current-seconds)
-                                                   #f
-                                                   (build-headers (reverse pending-headers-rev))
-                                                   (lambda (output-port)
-                                                     (let loop ()
-                                                       (match (async-channel-get chunk-ch)
-                                                         [#f (void)]
-                                                         [bs
-                                                          (write-bytes bs output-port)
-                                                          (flush-output output-port)
-                                                          (loop)]))))
-                                         (request-method req))))))
+                 (unless-reply-sent
+                  (lambda ()
+                    (output-response/method conn
+                                            (response pending-code
+                                                      (string->bytes/utf-8 pending-message)
+                                                      (current-seconds)
+                                                      #f
+                                                      (build-headers (reverse pending-headers-rev))
+                                                      (lambda (output-port)
+                                                        (let loop ()
+                                                          (match (async-channel-get chunk-ch)
+                                                            [#f (void)]
+                                                            [bs
+                                                             (write-bytes bs output-port)
+                                                             (flush-output output-port)
+                                                             (loop)]))))
+                                            (request-method req))))))))
 
-          (define (finish-request!)
-            (send-headers!)
-            (async-channel-put chunk-ch #f)
-            (stop-current-facet))
+          (define (send-final-chunk! chunk)
+            (define bs (chunk->bytes chunk))
+            (if chunk-ch
+                (begin (async-channel-put chunk-ch bs)
+                       (async-channel-put chunk-ch #f)
+                       (stop-current-facet))
+                (respond! (response/full pending-code
+                                         (string->bytes/utf-8 pending-message)
+                                         (current-seconds)
+                                         #f
+                                         (build-headers (reverse pending-headers-rev))
+                                         (list bs)))))
 
           (begin/dataflow
             (when (handler-terminated?)
-              (finish-request!)))
+              (set! pending-code 500)
+              (set! pending-message "Internal Server Error")
+              (set! pending-headers-rev '())
+              (send-final-chunk! *empty-chunk*)))
 
           (define res (object #:name connection-name
-                              [#:asserted (HttpResponse-processing)
-                               (log-syndicate/drivers/http-debug "~v: +processing" connection-name)
-                               (set! processing #t)
-                               (set! pending-code 200)
-                               (set! pending-message "OK")
-                               #:retracted
-                               (log-syndicate/drivers/http-debug "~v: -processing" connection-name)
-                               (finish-request!)]
                               [#:message (HttpResponse-status code message)
                                (log-syndicate/drivers/http-debug "~v: status ~v ~v" connection-name
                                                                  code message)
-                               (check-processing!)
                                (set! pending-code code)
                                (set! pending-message message)]
                               [#:message (HttpResponse-header name value)
                                (log-syndicate/drivers/http-debug "~v: header ~v ~v" connection-name
                                                                  name value)
-                               (check-processing!)
                                (set! pending-headers-rev (cons (cons name value) pending-headers-rev))]
-                              [#:message (HttpResponse-body chunk)
-                               (log-syndicate/drivers/http-debug "~v: chunk ~v" connection-name
-                                                                 chunk)
-                               (check-processing!)
-                               (send-headers!)
-                               (async-channel-put chunk-ch (chunk->bytes chunk))]))
+                              [#:message (HttpResponse-chunk chunk)
+                               (start-chunked-response!)
+                               (async-channel-put chunk-ch (chunk->bytes chunk))]
+                              [#:message (HttpResponse-done chunk)
+                               (send-final-chunk! chunk)]))
           (at handler (assert (HttpContext decoded-req res)))
           (return (void)))))
 
-    (begin (try-hostname host)
+    (begin (when (RequestHost-present? host)
+             (try-hostname (RequestHost-present-value host)))
            (try-hostname #f)
            (respond! (response/full 404
                                     #"Not Found"
@@ -285,6 +292,8 @@
       [[(cons expected pat) (cons actual p)] #:when (equal? expected actual) (loop pat p)]
       [[_ _] #f])))
 
+(define *empty-chunk* (Chunk-bytes #""))
+
 (define (chunk->bytes c)
   (match c
     [(Chunk-string s) (string->bytes/utf-8 s)]
@@ -313,7 +322,6 @@
            (lambda () method)
            (lambda () (quote-path-pattern () (path-pattern-element ...)))
            (lambda (req res)
-             (at res (assert (HttpResponse-processing)))
              (match (HttpRequest-path req)
                [#,(match-quote-path-pattern #'(path-pattern-element ...))
                 (async (match (let () body ...)
@@ -381,6 +389,7 @@
 (define (send-http-response! res code message
                              #:headers [headers '()]
                              #:mime-type [mime-type #f]
+                             #:done? [done? #t]
                              . chunks)
   (send! res (HttpResponse-status code message))
   (when mime-type
@@ -390,7 +399,11 @@
   (for [(header headers)]
     (match-define (cons name value) header)
     (send! res (HttpResponse-header name value)))
-  (for [(chunk chunks)] (send! res (HttpResponse-body chunk))))
+  (let loop ((chunks chunks))
+    (match chunks
+      ['() (when done? (send! res (HttpResponse-done *empty-chunk*)))]
+      [(list ch) (send! res ((if done? HttpResponse-done HttpResponse-chunk) ch))]
+      [(cons ch more) (send! res (HttpResponse-chunk ch)) (loop more)])))
 
 ;;---------------------------------------------------------------------------
 

syndicate/private/protocols/schema-bundle.bin

@@ -2,11 +2,9 @@
 tcp-remote„´³tupleµ´³named³host´³atom³String„„´³named³port´³atom³
SignedInteger„„„„„³TcpPeerInfo´³rec´³lit³tcp-peer„´³tupleµ´³named³handle´³embedded³any„„´³named³local´³refµ„³TcpLocal„„´³named³remote´³refµ„³	TcpRemote„„„„„„³embeddedType´³refµ³	EntityRef„³Cap„„„µ³http„´³schema·³version°

³definitions·³Chunk´³orµµ±string´³atom³String„„µ±bytes´³atom³
 ByteString„„„„³Headers´³dictof´³atom³Symbol„´³atom³String„„³MimeType´³atom³Symbol„³
 QueryValue´³orµµ±string´³atom³String„„µ±file´³rec´³lit³file„´³tupleµ´³named³filename´³atom³String„„´³named³headers´³refµ„³Headers„„´³named³body´³atom³
-ByteString„„„„„„„„³HostPattern´³orµµ±host´³atom³String„„µ±any´³lit€„„„„³HttpBinding´³rec´³lit³	http-bind„´³tupleµ´³named³host´³refµ„³HostPattern„„´³named³port´³atom³
SignedInteger„„´³named³method´³refµ„³
MethodPattern„„´³named³path´³refµ„³PathPattern„„´³named³handler´³embedded´³refµ„³HttpRequest„„„„„„³HttpContext´³rec´³lit³request„´³tupleµ´³named³req´³refµ„³HttpRequest„„´³named³res´³embedded´³refµ„³HttpResponse„„„„„„³HttpRequest´³rec´³lit³http-request„´³tupleµ´³named³sequenceNumber´³atom³
SignedInteger„„´³named³host´³atom³String„„´³named³port´³atom³
SignedInteger„„´³named³method´³atom³Symbol„„´³named³path´³seqof´³atom³String„„„´³named³headers´³refµ„³Headers„„´³named³query´³dictof´³atom³Symbol„´³seqof´³refµ„³
+ByteString„„„„„„„„³HostPattern´³orµµ±host´³atom³String„„µ±any´³lit€„„„„³HttpBinding´³rec´³lit³	http-bind„´³tupleµ´³named³host´³refµ„³HostPattern„„´³named³port´³atom³
SignedInteger„„´³named³method´³refµ„³
MethodPattern„„´³named³path´³refµ„³PathPattern„„´³named³handler´³embedded´³refµ„³HttpRequest„„„„„„³HttpContext´³rec´³lit³request„´³tupleµ´³named³req´³refµ„³HttpRequest„„´³named³res´³embedded´³refµ„³HttpResponse„„„„„„³HttpRequest´³rec´³lit³http-request„´³tupleµ´³named³sequenceNumber´³atom³
SignedInteger„„´³named³host´³refµ„³RequestHost„„´³named³port´³atom³
SignedInteger„„´³named³method´³atom³Symbol„„´³named³path´³seqof´³atom³String„„„´³named³headers´³refµ„³Headers„„´³named³query´³dictof´³atom³Symbol„´³seqof´³refµ„³
 QueryValue„„„„´³named³body´³refµ„³RequestBody„„„„„³HttpService´³rec´³lit³http-service„´³tupleµ´³named³host´³refµ„³HostPattern„„´³named³port´³atom³
SignedInteger„„´³named³method´³refµ„³
MethodPattern„„´³named³path´³refµ„³PathPattern„„„„„³PathPattern´³seqof´³refµ„³PathPatternElement„„³RequestBody´³orµµ±present´³atom³
-ByteString„„µ±absent´³lit€„„„„³HttpListener´³rec´³lit³
http-listener„´³tupleµ´³named³port´³atom³
SignedInteger„„„„„³HttpResponse´³orµµ±
-processing´³rec´³lit³
-processing„´³tupleµ„„„„µ±status´³rec´³lit³status„´³tupleµ´³named³code´³atom³
SignedInteger„„´³named³message´³atom³String„„„„„„µ±header´³rec´³lit³header„´³tupleµ´³named³name´³atom³Symbol„„´³named³value´³atom³String„„„„„„µ±body´³rec´³lit³body„´³tupleµ´³named³chunk´³refµ„³Chunk„„„„„„„„³
MethodPattern´³orµµ±any´³lit€„„µ±specific´³atom³Symbol„„„„³PathPatternElement´³orµµ±label´³atom³String„„µ±wildcard´³lit³
_„„µ±rest´³lit³...„„„„„³embeddedType€„„µ³noise„´³schema·³version°

³definitions·³Packet´³orµµ±complete´³atom³
+ByteString„„µ±absent´³lit€„„„„³RequestHost´³orµµ±present´³atom³String„„µ±absent´³lit€„„„„³HttpListener´³rec´³lit³
http-listener„´³tupleµ´³named³port´³atom³
SignedInteger„„„„„³HttpResponse´³orµµ±status´³rec´³lit³status„´³tupleµ´³named³code´³atom³
SignedInteger„„´³named³message´³atom³String„„„„„„µ±header´³rec´³lit³header„´³tupleµ´³named³name´³atom³Symbol„„´³named³value´³atom³String„„„„„„µ±chunk´³rec´³lit³chunk„´³tupleµ´³named³chunk´³refµ„³Chunk„„„„„„µ±done´³rec´³lit³done„´³tupleµ´³named³chunk´³refµ„³Chunk„„„„„„„„³
MethodPattern´³orµµ±any´³lit€„„µ±specific´³atom³Symbol„„„„³PathPatternElement´³orµµ±label´³atom³String„„µ±wildcard´³lit³
_„„µ±rest´³lit³...„„„„„³embeddedType€„„µ³noise„´³schema·³version°

³definitions·³Packet´³orµµ±complete´³atom³
 ByteString„„µ±
 fragmented´³seqof´³atom³
 ByteString„„„„„³	Initiator´³rec´³lit³	initiator„´³tupleµ´³named³initiatorSession´³embedded´³refµ„³Packet„„„„„„³	NoiseSpec´³andµ´³dict·³key´³named³key´³atom³

syndicate/private/protocols/schemas/http.prs

@@ -21,7 +21,7 @@ MethodPattern = @any #f / @specific @"Lowercase" symbol .
 # Assertion in driver DS
 HttpRequest = <http-request
                @sequenceNumber int
-               @host string
+               @host RequestHost
                @port int
                @method @"Lowercase" symbol
                @path [string ...]
@@ -32,20 +32,28 @@ HttpRequest = <http-request
 Headers = {@"Lowercase" symbol: string ...:...} .
 QueryValue = @string string / <file @filename string @headers Headers @body bytes> .
 RequestBody = @present bytes / @absent #f .
+RequestHost = @present string / @absent #f .
 
 # Assertion to handler entity
 HttpContext = <request @req HttpRequest @res #:HttpResponse> .
 
+# HttpResponse protocol. Delivered to the `res` ref in `HttpContext`.
+#
+#     (status | header)* . chunk* . done
+#
+# Done triggers completion of the response and retraction of the frame by the peer. If the
+# HttpBinding responsible for the request is withdrawn mid-way through a response (i.e. when
+# chunked transfer is used and at least one chunk has been sent) the request is abruptly
+# closed; if it is withdrawn at any other moment in the lifetime of the request, a 500 Internal
+# Server Error is send to the client.
+#
 @<TODO "trailers?">
-# Messages
 HttpResponse =
-# Assertion - frames the response. When retracted, response is considered complete.
-# If retracted before `status` is delivered, response is considered a 500.
-/ <processing>
-# Remainder are messages: +processing . (status | header)* . body* . -processing
+# Messages.
 / <status @code int @message string>
 / <header @name symbol @value string>
-/ <body @chunk Chunk>
+/ <chunk @chunk Chunk>
+/ <done @chunk Chunk>
 .
 
 Chunk = @string string / @bytes bytes .