laten zien hoe downloads en zo gaan
This commit is contained in:
@@ -0,0 +1,120 @@
|
||||
#lang racket/base
|
||||
|
||||
(require racket/async-channel)
|
||||
|
||||
(provide call-with-http-timeout)
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
; goal : Bound HTTP inactivity while allowing transfers of any duration.
|
||||
; pre : Seconds is positive. Proc consumes its HTTP ports before returning
|
||||
; and calls progress! whenever new response bytes arrive.
|
||||
; post : The worker has exited and its HTTP resources have been closed.
|
||||
; result : Proc's values, or its exception, or an inactivity timeout error.
|
||||
; internals: Activity restarts sync/timeout. On cancellation, a break unwinds
|
||||
; semaphore and dynamic-wind cleanup before the custodian closes
|
||||
; remaining ports. No timed-out worker can continue changing state.
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
(define (call-with-http-timeout seconds proc)
|
||||
(let* ((custodian (make-custodian))
|
||||
(activity (make-async-channel 1))
|
||||
(result (make-async-channel))
|
||||
(finished? #f)
|
||||
(worker
|
||||
(parameterize ((current-custodian custodian))
|
||||
(thread
|
||||
(λ ()
|
||||
(with-handlers ((exn:break? void))
|
||||
(parameterize-break #t
|
||||
(with-handlers
|
||||
((exn? (λ (exception)
|
||||
(async-channel-put result (cons #f exception)))))
|
||||
(let ((values-list
|
||||
(call-with-values
|
||||
(λ () (proc (λ () (async-channel-put activity #t))))
|
||||
list)))
|
||||
(async-channel-put result (cons #t values-list)))))))))))
|
||||
(dynamic-wind
|
||||
void
|
||||
(λ ()
|
||||
(let loop ()
|
||||
(let ((event (sync/timeout seconds
|
||||
(handle-evt result (λ (value) (cons 'result value)))
|
||||
(handle-evt activity (λ (_) 'progress)))))
|
||||
(cond
|
||||
((not event)
|
||||
(error 'http-transfer "No HTTP progress for ~a seconds" seconds))
|
||||
((eq? event 'progress) (loop))
|
||||
(else
|
||||
(set! finished? #t)
|
||||
(if (cadr event)
|
||||
(apply values (cddr event))
|
||||
(raise (cddr event))))))))
|
||||
(λ ()
|
||||
(unless (or finished? (thread-dead? worker))
|
||||
(break-thread worker))
|
||||
(thread-wait worker)
|
||||
(custodian-shutdown-all custodian)))))
|
||||
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
;; Tests for module http-transfer.rkt
|
||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||
|
||||
(module+ test
|
||||
(require rackunit racket/file)
|
||||
|
||||
(check-equal? (call-with-values
|
||||
(λ () (call-with-http-timeout 1 (λ (_) (values 1 2)))) list)
|
||||
'(1 2))
|
||||
(check-exn #rx"original failure"
|
||||
(λ () (call-with-http-timeout 1 (λ (_) (error "original failure")))))
|
||||
|
||||
(test-case "continuing progress allows a transfer longer than its idle limit"
|
||||
(check-equal?
|
||||
(call-with-http-timeout
|
||||
0.5
|
||||
(λ (progress!)
|
||||
(let loop ((chunks 6))
|
||||
(unless (zero? chunks)
|
||||
(sleep 0.15)
|
||||
(progress!)
|
||||
(loop (- chunks 1))))
|
||||
'downloaded))
|
||||
'downloaded))
|
||||
|
||||
(test-case "stalled HTTP workers release locks and close their ports"
|
||||
(let ((lock (make-semaphore 1))
|
||||
(output #f)
|
||||
(target (make-temporary-file "http-stall-~a")))
|
||||
(dynamic-wind
|
||||
void
|
||||
(λ ()
|
||||
(check-exn
|
||||
#rx"No HTTP progress"
|
||||
(λ ()
|
||||
(parameterize-break #f
|
||||
(call-with-http-timeout
|
||||
0.1
|
||||
(λ (_)
|
||||
(set! output (open-output-file target #:exists 'truncate))
|
||||
(call-with-semaphore lock (λ () (sync never-evt))))))))
|
||||
(check-true (semaphore-try-wait? lock))
|
||||
(check-true (port-closed? output)))
|
||||
(λ () (delete-file target)))))
|
||||
|
||||
(test-case "breaking the caller also unwinds its HTTP worker"
|
||||
(let* ((ready (make-semaphore))
|
||||
(lock (make-semaphore 1))
|
||||
(caller
|
||||
(thread
|
||||
(λ ()
|
||||
(with-handlers ((exn:break? void))
|
||||
(call-with-http-timeout
|
||||
5
|
||||
(λ (_)
|
||||
(call-with-semaphore
|
||||
lock
|
||||
(λ () (semaphore-post ready) (sync never-evt))))))))))
|
||||
(semaphore-wait ready)
|
||||
(break-thread caller)
|
||||
(check-not-false (sync/timeout 2 caller))
|
||||
(check-true (semaphore-try-wait? lock)))))
|
||||
@@ -9,7 +9,8 @@
|
||||
racket/port
|
||||
racket/string
|
||||
simple-log
|
||||
"player-agent-translate.rkt")
|
||||
"player-agent-translate.rkt"
|
||||
"http-transfer.rkt")
|
||||
|
||||
(provide (struct-out player-agent-runtime)
|
||||
make-player-agent-runtime)
|
||||
@@ -44,27 +45,30 @@
|
||||
;;; Posts JSON to an agent API endpoint and reads its JSON response.
|
||||
;;; Authorization failures receive a distinct exception for poll-loop.
|
||||
(define (post-json base path data)
|
||||
(let ((input
|
||||
(post-pure-port
|
||||
(endpoint-url base path)
|
||||
(jsexpr->bytes data)
|
||||
(list "Content-Type: application/json"
|
||||
"Cache-Control: no-store"))))
|
||||
(dynamic-wind
|
||||
void
|
||||
(λ ()
|
||||
(let ((response (read-json input)))
|
||||
(when (and (hash? response)
|
||||
(string? (hash-ref response 'error #f)))
|
||||
(if (equal? (hash-ref response 'code #f)
|
||||
"agent-not-authorized")
|
||||
(raise
|
||||
(exn:fail:agent-denied
|
||||
(hash-ref response 'error)
|
||||
(current-continuation-marks)))
|
||||
(error 'player-agent (hash-ref response 'error))))
|
||||
response))
|
||||
(λ () (close-input-port input)))))
|
||||
(call-with-http-timeout
|
||||
30
|
||||
(λ (_progress!)
|
||||
(let ((input
|
||||
(post-pure-port
|
||||
(endpoint-url base path)
|
||||
(jsexpr->bytes data)
|
||||
(list "Content-Type: application/json"
|
||||
"Cache-Control: no-store"))))
|
||||
(dynamic-wind
|
||||
void
|
||||
(λ ()
|
||||
(let ((response (read-json input)))
|
||||
(when (and (hash? response)
|
||||
(string? (hash-ref response 'error #f)))
|
||||
(if (equal? (hash-ref response 'code #f)
|
||||
"agent-not-authorized")
|
||||
(raise
|
||||
(exn:fail:agent-denied
|
||||
(hash-ref response 'error)
|
||||
(current-continuation-marks)))
|
||||
(error 'player-agent (hash-ref response 'error))))
|
||||
response))
|
||||
(λ () (close-input-port input)))))))
|
||||
|
||||
;;; Converts racket-audio states to the state names sent to the web player.
|
||||
(define (normal-state state)
|
||||
@@ -267,7 +271,7 @@
|
||||
(bytes->string/utf-8 extension))))
|
||||
(path (format "/api/agent/media/~a/~a" app-id token))
|
||||
(input #f))
|
||||
(with-handlers ((exn:fail?
|
||||
(with-handlers ((exn?
|
||||
(λ (exception)
|
||||
(when input (close-input-port input))
|
||||
(safe-delete-file target)
|
||||
@@ -281,19 +285,23 @@
|
||||
(hasheq 'filename filename 'received 0
|
||||
'total (if (exact-nonnegative-integer? size)
|
||||
size 'null)))))
|
||||
(set! input (get-pure-port (endpoint-url server-url path)))
|
||||
(call-with-output-file
|
||||
target
|
||||
(λ (output)
|
||||
(copy-media-port
|
||||
input output
|
||||
(λ (received)
|
||||
(with-agent-state
|
||||
(λ ()
|
||||
(set! download-progress
|
||||
(hash-set download-progress 'received received)))))))
|
||||
#:exists 'truncate/replace)
|
||||
(close-input-port input)
|
||||
(call-with-http-timeout
|
||||
30
|
||||
(λ (progress!)
|
||||
(set! input (get-pure-port (endpoint-url server-url path)))
|
||||
(call-with-output-file
|
||||
target
|
||||
(λ (output)
|
||||
(copy-media-port
|
||||
input output
|
||||
(λ (received)
|
||||
(progress!)
|
||||
(with-agent-state
|
||||
(λ ()
|
||||
(set! download-progress
|
||||
(hash-set download-progress 'received received)))))))
|
||||
#:exists 'truncate/replace)
|
||||
(close-input-port input)))
|
||||
(when (and (exact-nonnegative-integer? size)
|
||||
(not (= (file-size target) size)))
|
||||
(error 'player-agent
|
||||
@@ -538,21 +546,23 @@
|
||||
(set! command-worker
|
||||
(thread
|
||||
(λ ()
|
||||
(with-handlers
|
||||
((exn:fail?
|
||||
(λ (exception)
|
||||
(warn-player-agent "Command failed: ~a"
|
||||
(exn-message exception))
|
||||
(with-agent-state
|
||||
(λ ()
|
||||
(set! loading-media-key #f)
|
||||
(trim-cache!)))
|
||||
(set-agent-error! (exn-message exception)))))
|
||||
(clear-agent-error!)
|
||||
(execute-command! command))
|
||||
(set! acknowledged-command (hash-ref command 'id))
|
||||
(set! executing-command-id 0)
|
||||
(set! command-worker #f))))))
|
||||
(with-handlers ((exn:break? void))
|
||||
(parameterize-break #t
|
||||
(with-handlers
|
||||
((exn:fail?
|
||||
(λ (exception)
|
||||
(warn-player-agent "Command failed: ~a"
|
||||
(exn-message exception))
|
||||
(with-agent-state
|
||||
(λ ()
|
||||
(set! loading-media-key #f)
|
||||
(trim-cache!)))
|
||||
(set-agent-error! (exn-message exception)))))
|
||||
(clear-agent-error!)
|
||||
(execute-command! command))
|
||||
(set! acknowledged-command (hash-ref command 'id))
|
||||
(set! executing-command-id 0)
|
||||
(set! command-worker #f))))))))
|
||||
(sleep 1)
|
||||
(loop)))))
|
||||
|
||||
@@ -561,15 +571,23 @@
|
||||
(unless running
|
||||
(set! running #t)
|
||||
(status-callback (tr 'connecting))
|
||||
(set! worker (thread poll-loop))))
|
||||
(set! worker
|
||||
(thread
|
||||
(λ ()
|
||||
(with-handlers ((exn:break? void))
|
||||
(parameterize-break #t (poll-loop))))))))
|
||||
|
||||
;;; Stops polling and command workers and clears their lifecycle state.
|
||||
(define (stop!)
|
||||
(set! running #f)
|
||||
(when (and worker (not (thread-dead? worker)))
|
||||
(kill-thread worker))
|
||||
(when (and command-worker (not (thread-dead? command-worker)))
|
||||
(kill-thread command-worker))
|
||||
(let ((poll-worker worker)
|
||||
(active-command command-worker))
|
||||
(when (and poll-worker (not (thread-dead? poll-worker)))
|
||||
(break-thread poll-worker)
|
||||
(thread-wait poll-worker))
|
||||
(when (and active-command (not (thread-dead? active-command)))
|
||||
(break-thread active-command)
|
||||
(thread-wait active-command)))
|
||||
(set! worker #f)
|
||||
(set! command-worker #f)
|
||||
(with-agent-state (λ () (set! download-progress 'null)))
|
||||
|
||||
Reference in New Issue
Block a user