make example more robust for multiple clients

examples/apt-archive/proxy.ss

@@ -17,7 +17,7 @@
 (define offline? (opt 'offline))
 (define archive (or (opt 'archive) (path-combine (base-dir) "archive")))
 (define verbose? (and (opt 'verbose)))
-(define port (cond [(opt 'port) => string->number] [else 9001]))
+(define port (cond [(opt 'port) => string->number] [else 9000]))
 (unless (and (fixnum? port) (fx>= port 0))
   (errorf #f "invalid port ~a" (opt 'port)))
 (define max-retries 20)
@@ -95,7 +95,7 @@
 (define-tuple <req> target-file scheme host port method path header)
 
 (define (fetcher:start&link)
-  (define-state-tuple <fetcher> requests)
+  (define-state-tuple <fetcher> requests workers)
   (define-tuple <result> path status)
   (define (show-progress status req)
     (when verbose?
@@ -137,8 +137,20 @@
                          (put-bytevector fop data))))
                    (file-exists? target-file)]
                   [,_ (guard (memv status '(301 302 307 308))) (json:ref header 'location #f)]))))])]))
+  (define (add-worker workers path request server)
+    (define pid
+      (spawn
+       (lambda ()
+         (send server (fetch! request)))))
+    (monitor pid)
+    (ht:set workers pid path))
+  (define (reply-all requests path status)
+    (for-each (lambda (caller) (gen-server:reply caller status))
+      (ht:ref requests path '())))
   (define (init)
-    `#(ok ,(<fetcher> make [requests (ht:make string-hash string=? string?)])))
+    `#(ok ,(<fetcher> make
+             [requests (ht:make string-hash string=? string?)]
+             [workers (ht:make process-id eq? process?)])))
   (define (terminate reason state) 'ok)
   (define (handle-call msg from state)
     (match msg
@@ -152,21 +164,29 @@
          `#(reply #f ,state)]
         [else
          (let ([waiting (ht:ref ($state requests) path '())])
-           (when (null? waiting)
-             (spawn&link
-              (let ([me self])
-                (lambda ()
-                  (send me (fetch! msg))))))
            `#(no-reply
-              ,($state copy* [requests (ht:set requests path (cons from waiting))])))])]))
+              ,($state copy*
+                 [requests (ht:set requests path (cons from waiting))]
+                 [workers
+                  (if (null? waiting)
+                      (add-worker workers path msg self)
+                      workers)])))])]))
   (define (handle-cast msg state) (match msg))
   (define (handle-info msg state)
     (match msg
       [`(<result> ,path ,status)
        ($state open [requests])
-       (for-each (lambda (caller) (gen-server:reply caller status))
-         (ht:ref requests path '()))
-       `#(no-reply ,($state copy [requests (ht:delete requests path)]))]))
+       (reply-all requests path status)
+       `#(no-reply ,($state copy [requests (ht:delete requests path)]))]
+      [`(DOWN ,_ ,pid ,reason ,err)
+       ($state open [requests workers])
+       (let ([path (assert (ht:ref workers pid #f))])
+         (unless (eq? reason 'normal)
+           (reply-all requests path err))
+         `#(no-reply
+            ,($state copy
+               [requests (ht:delete requests path)]
+               [workers (ht:delete workers pid)])))]))
   (gen-server:start&link 'fetcher))
 
 (define (get-cached! target-file method scheme host port path header)
@@ -204,7 +224,9 @@
                   (match (try (get-cached! target-file method scheme host port path header))
                     [#t (http:respond-file conn 200 '() target-file)]
                     [#f (http:respond conn 404 '() #vu8())]
-                    [`(catch ,reason) (retry rpath (+ n 1) header)]
+                    [`(catch ,reason)
+                     (report "ERR ~s: ~a\n" n (exit-reason->english reason))
+                     (retry rpath (+ n 1) header)]
                     [,redirect
                      (guard (string? redirect))
                      (retry redirect (+ n 1) (copy-header header '(host accept user-agent)))])))))