Download timeouts
This commit is contained in:
@@ -84,6 +84,17 @@
|
||||
(exn-message exception)))))
|
||||
(delete-file file))))
|
||||
|
||||
;;; Copies media in bounded chunks, reporting bytes written without waiting
|
||||
;;; for a full buffer; slow network streams therefore still show progress.
|
||||
(define (copy-media-port input output progress!)
|
||||
(let ((buffer (make-bytes 65536)))
|
||||
(let loop ((received 0))
|
||||
(let ((count (read-bytes-avail! buffer input)))
|
||||
(unless (eof-object? count)
|
||||
(write-bytes buffer output 0 count)
|
||||
(progress! (+ received count))
|
||||
(loop (+ received count)))))))
|
||||
|
||||
;;; Evicts least recently used files until the cache fits its byte budget.
|
||||
;;; Entries contain path, byte size and last use; active files remain protected.
|
||||
(define (trim-media-cache! cache protected [limit 1000000000])
|
||||
@@ -151,6 +162,7 @@
|
||||
(pending-auto-music-id #f)
|
||||
(music-tracks (make-hash))
|
||||
(current-track-value #f)
|
||||
(download-progress 'null)
|
||||
(acknowledged-command 0)
|
||||
(ended-counter 0)
|
||||
(logical-volume 50)
|
||||
@@ -173,9 +185,11 @@
|
||||
(define (state-value value fallback)
|
||||
(if (eq? value #f) fallback value))
|
||||
|
||||
;;; Returns the latest audio state while holding the state semaphore.
|
||||
;;; Combines audio and download state under the semaphore so audio callbacks
|
||||
;;; cannot erase progress while the next track is being prefetched.
|
||||
(define (snapshot)
|
||||
(with-agent-state (λ () agent-state)))
|
||||
(with-agent-state
|
||||
(λ () (hash-set agent-state 'download download-progress))))
|
||||
|
||||
;;; Returns the currently audible track while holding the state semaphore.
|
||||
(define (current-track)
|
||||
@@ -243,7 +257,7 @@
|
||||
|
||||
;;; Downloads one protected media resource to a temporary local file.
|
||||
;;; A failed download closes its port and removes its partial file.
|
||||
(define (download-media! token filename)
|
||||
(define (download-media! token filename size)
|
||||
(let* ((started (current-inexact-monotonic-milliseconds))
|
||||
(extension
|
||||
(or (path-get-extension (string->path filename)) #""))
|
||||
@@ -257,14 +271,35 @@
|
||||
(λ (exception)
|
||||
(when input (close-input-port input))
|
||||
(safe-delete-file target)
|
||||
(with-agent-state
|
||||
(λ () (set! download-progress 'null)))
|
||||
(raise exception))))
|
||||
(info-player-agent "Downloading ~a" filename)
|
||||
(with-agent-state
|
||||
(λ ()
|
||||
(set! download-progress
|
||||
(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-port input output))
|
||||
(λ (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)
|
||||
(when (and (exact-nonnegative-integer? size)
|
||||
(not (= (file-size target) size)))
|
||||
(error 'player-agent
|
||||
"Incomplete media download: received ~a of ~a bytes"
|
||||
(file-size target) size))
|
||||
(with-agent-state (λ () (set! download-progress 'null)))
|
||||
(info-player-agent "Downloaded ~a: ~a bytes in ~a ms"
|
||||
filename (file-size target)
|
||||
(round (- (current-inexact-monotonic-milliseconds)
|
||||
@@ -299,12 +334,8 @@
|
||||
(let ((downloaded
|
||||
(download-media!
|
||||
(hash-ref data 'mediaToken)
|
||||
(hash-ref data 'filename "track"))))
|
||||
(when (and (hash-has-key? data 'mediaSize)
|
||||
(not (= (file-size downloaded)
|
||||
(hash-ref data 'mediaSize))))
|
||||
(safe-delete-file downloaded)
|
||||
(error 'player-agent "Downloaded media size does not match the server"))
|
||||
(hash-ref data 'filename "track")
|
||||
(hash-ref data 'mediaSize #f))))
|
||||
(with-agent-state
|
||||
(λ ()
|
||||
(when found (safe-delete-file (car found)))
|
||||
@@ -541,6 +572,7 @@
|
||||
(kill-thread command-worker))
|
||||
(set! worker #f)
|
||||
(set! command-worker #f)
|
||||
(with-agent-state (λ () (set! download-progress 'null)))
|
||||
(set! executing-command-id 0))
|
||||
|
||||
;;; Restarts the runtime with a new normalized server address and name.
|
||||
@@ -579,6 +611,35 @@
|
||||
(module+ test
|
||||
(require rackunit)
|
||||
|
||||
;; Progress must arrive before EOF, even when a slow server has sent much
|
||||
;; less than one buffer. Binary data must reach the output unchanged.
|
||||
(let-values (((input writer) (make-pipe)))
|
||||
(let* ((output (open-output-bytes))
|
||||
(updates (make-channel))
|
||||
(worker
|
||||
(thread
|
||||
(λ ()
|
||||
(copy-media-port input output
|
||||
(λ (received)
|
||||
(channel-put updates received)))))))
|
||||
(dynamic-wind
|
||||
void
|
||||
(λ ()
|
||||
(write-bytes #"\0\1\377" writer)
|
||||
(flush-output writer)
|
||||
(check-equal? (sync/timeout 5 updates) 3)
|
||||
(check-equal? (get-output-bytes output) #"\0\1\377")
|
||||
(write-bytes #"next" writer)
|
||||
(close-output-port writer)
|
||||
(check-equal? (sync/timeout 5 updates) 7)
|
||||
(check-not-false (sync/timeout 5 worker))
|
||||
(check-equal? (get-output-bytes output) #"\0\1\377next"))
|
||||
(λ ()
|
||||
(kill-thread worker)
|
||||
(close-input-port input)
|
||||
(close-output-port writer)
|
||||
(close-output-port output)))))
|
||||
|
||||
(check-equal? (normal-state 'initialized) "stopped")
|
||||
(check-equal? (normal-state 'transitioning) "starting")
|
||||
(check-equal? (normal-state 'playing) "playing")
|
||||
@@ -620,6 +681,9 @@
|
||||
app-id)))
|
||||
(check-false ((player-agent-runtime-running? runtime)))
|
||||
(check-false ((player-agent-runtime-current-track runtime)))
|
||||
(check-equal?
|
||||
(hash-ref ((player-agent-runtime-snapshot runtime)) 'download)
|
||||
'null)
|
||||
(check-equal?
|
||||
(hash-ref ((player-agent-runtime-snapshot runtime)) 'state)
|
||||
"stopped")
|
||||
|
||||
Reference in New Issue
Block a user