WIP revise proxy to forward data as it arrives

examples/apt-archive/proxy.ss

@@ -71,15 +71,24 @@
   ;; would require handling HTTP 206.
   ;;
   ;; Preserve original host in case that matters for virtual host.
+  (header->list client-header
+    (lambda (key val)
+      ;; http:read-header converts keys to lower-case symbols
+      (if (eq? key 'if-modified-since)
+          #f
+          (cons-field key val)))))
+
+;; This works for our purposes, but order could matter in HTTP header.
+(define (header->list header cons-field)
   (remq #f
     (vector->list
      (vector-map
       (lambda (cell)
         (match-define (,key . ,val) cell)
-        ;; http:read-header converts keys to lower-case symbols
-        (and (not (eq? key 'if-modified-since))
-             (cons (symbol->string key) val)))
-      (json:cells client-header)))))
+        (cons-field key val))
+      (json:cells header)))))
+
+(define (cons-field key val) (cons (symbol->string key) val))
 
 (define (choose-port scheme port)
   (cond
@@ -92,7 +101,7 @@
   (with-interrupts-disabled ;; guard against concurrent writes to console port
    (apply printf fmt args)))
 
-(define-tuple <req> target-file scheme host port method path header)
+(define-tuple <req> conn target-file scheme host port method path header)
 
 (define (fetcher:start&link)
   (define-state-tuple <fetcher> requests workers)
@@ -102,13 +111,13 @@
       (match-let* ([`(<req> ,method ,scheme ,host ,path) req])
         (report "~:@(~a~) ~a://~a ~a [~a]\n" method scheme host path status))))
   (define (fetch! request)
-    (<req> open request [target-file scheme host port method path header])
+    (<req> open request [conn target-file scheme host port method path header])
+    (define tmp-file (path-combine archive (uuid->string (osi_make_uuid))))
     (<result> make
       [path path]
       [status
        (match scheme
          ["https"
-          (define tmp-file (path-combine archive (uuid->string (osi_make_uuid))))
           (define-values (to-stdin from-stdout from-stderr os-pid)
             (spawn-os-process "curl"
               `("-s" "-o" ,tmp-file ,(format "https://~a~a" host path))
@@ -131,11 +140,27 @@
                 (show-progress status request)
                 (match status
                   [200
-                   (let ([data (get-bytevector-exactly-n ip len)])
-                     (let ([fop (open-binary-file-to-replace (make-directory-path target-file))])
-                       (on-exit (close-port fop)
-                         (put-bytevector fop data))))
-                   (file-exists? target-file)]
+                   (http:call-with-ports conn
+                     (lambda (cip cop)
+                       (let* ([buf-size (file-buffer-size)]
+                              [buf (make-bytevector buf-size)]
+                              [fop (open-binary-file-to-replace tmp-file)])
+                         (on-exit (begin (close-port fop) (delete-file tmp-file))
+                           (http:write-status cop 200)
+                           (http:write-header cop (header->list header cons-field))
+                           (let lp ([remaining len])
+                             (cond
+                              [(fx= remaining 0)
+                               (rename-path tmp-file (make-directory-path target-file))
+                               'file-sent]
+                              [else
+                               (match (get-bytevector-n! ip buf 0 (fxmin buf-size remaining))
+                                 [#!eof (raise `#(eof-while-fetching ,path))]
+                                 [,n
+                                  (put-bytevector fop buf 0 n)
+                                  (put-bytevector cop buf 0 n)
+                                  (lp (fx- remaining n))])])))))
+                     'infinity)]
                   [,_ (guard (memv status '(301 302 307 308))) (json:ref header 'location #f)]))))])]))
   (define (add-worker workers path request server)
     (define pid
@@ -189,9 +214,10 @@
                [workers (ht:delete workers pid)])))]))
   (gen-server:start&link 'fetcher))
 
-(define (get-cached! target-file method scheme host port path header)
+(define (get-cached! conn target-file method scheme host port path header)
   (gen-server:call 'fetcher
     (<req> make
+      [conn conn]
       [target-file target-file]
       [scheme scheme]
       [host host]
@@ -221,11 +247,12 @@
               (if (not (and (http:valid-path? path)
                             (string-ci=? "GET" (symbol->string method))))
                   (http:respond conn 400 '() #vu8())
-                  (match (try (get-cached! target-file method scheme host port path header))
+                  (match (try (get-cached! conn target-file method scheme host port path header))
                     [#t (http:respond-file conn 200 '() target-file)]
                     [#f (http:respond conn 404 '() #vu8())]
+                    [file-sent (void)]
                     [`(catch ,reason)
-                     (report "ERR ~s: ~a\n" n (exit-reason->english reason))
+                     (report "ERR: ~a ~a\n" (exit-reason->english reason) rpath)
                      (retry rpath (+ n 1) header)]
                     [,redirect
                      (guard (string? redirect))