Compare commits
3 Commits
f7f4d440b8
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 59291bd8bb | |||
| 3fda5c2485 | |||
| 5f0d835501 |
@@ -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/port
|
||||||
racket/string
|
racket/string
|
||||||
simple-log
|
simple-log
|
||||||
"player-agent-translate.rkt")
|
"player-agent-translate.rkt"
|
||||||
|
"http-transfer.rkt")
|
||||||
|
|
||||||
(provide (struct-out player-agent-runtime)
|
(provide (struct-out player-agent-runtime)
|
||||||
make-player-agent-runtime)
|
make-player-agent-runtime)
|
||||||
@@ -44,6 +45,9 @@
|
|||||||
;;; Posts JSON to an agent API endpoint and reads its JSON response.
|
;;; Posts JSON to an agent API endpoint and reads its JSON response.
|
||||||
;;; Authorization failures receive a distinct exception for poll-loop.
|
;;; Authorization failures receive a distinct exception for poll-loop.
|
||||||
(define (post-json base path data)
|
(define (post-json base path data)
|
||||||
|
(call-with-http-timeout
|
||||||
|
30
|
||||||
|
(λ (_progress!)
|
||||||
(let ((input
|
(let ((input
|
||||||
(post-pure-port
|
(post-pure-port
|
||||||
(endpoint-url base path)
|
(endpoint-url base path)
|
||||||
@@ -64,7 +68,7 @@
|
|||||||
(current-continuation-marks)))
|
(current-continuation-marks)))
|
||||||
(error 'player-agent (hash-ref response 'error))))
|
(error 'player-agent (hash-ref response 'error))))
|
||||||
response))
|
response))
|
||||||
(λ () (close-input-port input)))))
|
(λ () (close-input-port input)))))))
|
||||||
|
|
||||||
;;; Converts racket-audio states to the state names sent to the web player.
|
;;; Converts racket-audio states to the state names sent to the web player.
|
||||||
(define (normal-state state)
|
(define (normal-state state)
|
||||||
@@ -86,9 +90,13 @@
|
|||||||
|
|
||||||
;;; Copies media in bounded chunks, reporting bytes written without waiting
|
;;; Copies media in bounded chunks, reporting bytes written without waiting
|
||||||
;;; for a full buffer; slow network streams therefore still show progress.
|
;;; for a full buffer; slow network streams therefore still show progress.
|
||||||
(define (copy-media-port input output progress!)
|
(define (copy-media-port input output progress! #:idle-timeout [idle-timeout 60])
|
||||||
(let ((buffer (make-bytes 65536)))
|
(let ((buffer (make-bytes 65536)))
|
||||||
(let loop ((received 0))
|
(let loop ((received 0))
|
||||||
|
(unless (sync/timeout idle-timeout input)
|
||||||
|
(error 'player-agent
|
||||||
|
"Download timeout: no data received for ~a seconds (~a bytes received)"
|
||||||
|
idle-timeout received))
|
||||||
(let ((count (read-bytes-avail! buffer input)))
|
(let ((count (read-bytes-avail! buffer input)))
|
||||||
(unless (eof-object? count)
|
(unless (eof-object? count)
|
||||||
(write-bytes buffer output 0 count)
|
(write-bytes buffer output 0 count)
|
||||||
@@ -224,7 +232,7 @@
|
|||||||
'format (let ((decoder (hash-ref full-state 'decoder #f)))
|
'format (let ((decoder (hash-ref full-state 'decoder #f)))
|
||||||
(if decoder (format "~a" decoder) ""))
|
(if decoder (format "~a" decoder) ""))
|
||||||
'volume logical-volume
|
'volume logical-volume
|
||||||
'error 'null))
|
'error (hash-ref agent-state 'error 'null)))
|
||||||
(when (and pending-auto-music-id
|
(when (and pending-auto-music-id
|
||||||
(number? audible-music-id)
|
(number? audible-music-id)
|
||||||
(= pending-auto-music-id audible-music-id))
|
(= pending-auto-music-id audible-music-id))
|
||||||
@@ -257,7 +265,7 @@
|
|||||||
|
|
||||||
;;; Downloads one protected media resource to a temporary local file.
|
;;; Downloads one protected media resource to a temporary local file.
|
||||||
;;; A failed download closes its port and removes its partial file.
|
;;; A failed download closes its port and removes its partial file.
|
||||||
(define (download-media! token filename size)
|
(define (download-media-once! token filename size)
|
||||||
(let* ((started (current-inexact-monotonic-milliseconds))
|
(let* ((started (current-inexact-monotonic-milliseconds))
|
||||||
(extension
|
(extension
|
||||||
(or (path-get-extension (string->path filename)) #""))
|
(or (path-get-extension (string->path filename)) #""))
|
||||||
@@ -267,7 +275,7 @@
|
|||||||
(bytes->string/utf-8 extension))))
|
(bytes->string/utf-8 extension))))
|
||||||
(path (format "/api/agent/media/~a/~a" app-id token))
|
(path (format "/api/agent/media/~a/~a" app-id token))
|
||||||
(input #f))
|
(input #f))
|
||||||
(with-handlers ((exn:fail?
|
(with-handlers ((exn?
|
||||||
(λ (exception)
|
(λ (exception)
|
||||||
(when input (close-input-port input))
|
(when input (close-input-port input))
|
||||||
(safe-delete-file target)
|
(safe-delete-file target)
|
||||||
@@ -281,6 +289,9 @@
|
|||||||
(hasheq 'filename filename 'received 0
|
(hasheq 'filename filename 'received 0
|
||||||
'total (if (exact-nonnegative-integer? size)
|
'total (if (exact-nonnegative-integer? size)
|
||||||
size 'null)))))
|
size 'null)))))
|
||||||
|
(call-with-http-timeout
|
||||||
|
30
|
||||||
|
(λ (progress!)
|
||||||
(set! input (get-pure-port (endpoint-url server-url path)))
|
(set! input (get-pure-port (endpoint-url server-url path)))
|
||||||
(call-with-output-file
|
(call-with-output-file
|
||||||
target
|
target
|
||||||
@@ -288,12 +299,13 @@
|
|||||||
(copy-media-port
|
(copy-media-port
|
||||||
input output
|
input output
|
||||||
(λ (received)
|
(λ (received)
|
||||||
|
(progress!)
|
||||||
(with-agent-state
|
(with-agent-state
|
||||||
(λ ()
|
(λ ()
|
||||||
(set! download-progress
|
(set! download-progress
|
||||||
(hash-set download-progress 'received received)))))))
|
(hash-set download-progress 'received received)))))))
|
||||||
#:exists 'truncate/replace)
|
#:exists 'truncate/replace)
|
||||||
(close-input-port input)
|
(close-input-port input)))
|
||||||
(when (and (exact-nonnegative-integer? size)
|
(when (and (exact-nonnegative-integer? size)
|
||||||
(not (= (file-size target) size)))
|
(not (= (file-size target) size)))
|
||||||
(error 'player-agent
|
(error 'player-agent
|
||||||
@@ -306,6 +318,22 @@
|
|||||||
started)))
|
started)))
|
||||||
target)))
|
target)))
|
||||||
|
|
||||||
|
;;; Retries interrupted downloads twice before reporting a final failure.
|
||||||
|
;;; Each attempt removes its partial file; the media token stays unacknowledged.
|
||||||
|
(define (download-media! token filename size)
|
||||||
|
(let loop ((attempt 1))
|
||||||
|
(with-handlers
|
||||||
|
((exn:fail?
|
||||||
|
(λ (exception)
|
||||||
|
(if (< attempt 3)
|
||||||
|
(begin
|
||||||
|
(warn-player-agent "Download attempt ~a failed for ~a: ~a; retrying"
|
||||||
|
attempt filename (exn-message exception))
|
||||||
|
(sleep 1)
|
||||||
|
(loop (+ attempt 1)))
|
||||||
|
(raise exception)))))
|
||||||
|
(download-media-once! token filename size))))
|
||||||
|
|
||||||
;;; Scopes media identity to its server, source path, size and modification time.
|
;;; Scopes media identity to its server, source path, size and modification time.
|
||||||
;;; Older servers without version metadata only reuse the same media token.
|
;;; Older servers without version metadata only reuse the same media token.
|
||||||
(define (command-cache-key data)
|
(define (command-cache-key data)
|
||||||
@@ -538,21 +566,39 @@
|
|||||||
(set! command-worker
|
(set! command-worker
|
||||||
(thread
|
(thread
|
||||||
(λ ()
|
(λ ()
|
||||||
|
(with-handlers ((exn:break? void))
|
||||||
|
(parameterize-break #t
|
||||||
(with-handlers
|
(with-handlers
|
||||||
((exn:fail?
|
((exn:fail?
|
||||||
(λ (exception)
|
(λ (exception)
|
||||||
(warn-player-agent "Command failed: ~a"
|
(warn-player-agent "Command failed: ~a"
|
||||||
(exn-message exception))
|
(exn-message exception))
|
||||||
|
(when (and audio
|
||||||
|
(equal? (hash-ref command 'action) "play"))
|
||||||
|
(with-handlers
|
||||||
|
((exn:fail?
|
||||||
|
(λ (stop-exception)
|
||||||
|
(warn-player-agent
|
||||||
|
"Could not stop audio after failed play: ~a"
|
||||||
|
(exn-message stop-exception)))))
|
||||||
|
(audio-stop! audio)))
|
||||||
(with-agent-state
|
(with-agent-state
|
||||||
(λ ()
|
(λ ()
|
||||||
(set! loading-media-key #f)
|
(set! loading-media-key #f)
|
||||||
|
(when (equal? (hash-ref command 'action) "play")
|
||||||
|
(set! agent-state
|
||||||
|
(hash-set* agent-state
|
||||||
|
'state "stopped"
|
||||||
|
'position 0)))
|
||||||
(trim-cache!)))
|
(trim-cache!)))
|
||||||
(set-agent-error! (exn-message exception)))))
|
(set-agent-error! (exn-message exception)))))
|
||||||
(clear-agent-error!)
|
;; A queued prefetch must not erase a failed play's error.
|
||||||
|
(unless (equal? (hash-ref command 'action) "prefetch")
|
||||||
|
(clear-agent-error!))
|
||||||
(execute-command! command))
|
(execute-command! command))
|
||||||
(set! acknowledged-command (hash-ref command 'id))
|
(set! acknowledged-command (hash-ref command 'id))
|
||||||
(set! executing-command-id 0)
|
(set! executing-command-id 0)
|
||||||
(set! command-worker #f))))))
|
(set! command-worker #f))))))))
|
||||||
(sleep 1)
|
(sleep 1)
|
||||||
(loop)))))
|
(loop)))))
|
||||||
|
|
||||||
@@ -561,15 +607,23 @@
|
|||||||
(unless running
|
(unless running
|
||||||
(set! running #t)
|
(set! running #t)
|
||||||
(status-callback (tr 'connecting))
|
(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.
|
;;; Stops polling and command workers and clears their lifecycle state.
|
||||||
(define (stop!)
|
(define (stop!)
|
||||||
(set! running #f)
|
(set! running #f)
|
||||||
(when (and worker (not (thread-dead? worker)))
|
(let ((poll-worker worker)
|
||||||
(kill-thread worker))
|
(active-command command-worker))
|
||||||
(when (and command-worker (not (thread-dead? command-worker)))
|
(when (and poll-worker (not (thread-dead? poll-worker)))
|
||||||
(kill-thread command-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! worker #f)
|
||||||
(set! command-worker #f)
|
(set! command-worker #f)
|
||||||
(with-agent-state (λ () (set! download-progress 'null)))
|
(with-agent-state (λ () (set! download-progress 'null)))
|
||||||
@@ -609,7 +663,102 @@
|
|||||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||||
|
|
||||||
(module+ test
|
(module+ test
|
||||||
(require rackunit)
|
(require rackunit
|
||||||
|
racket/async-channel
|
||||||
|
web-server/http
|
||||||
|
web-server/http/json
|
||||||
|
web-server/web-server
|
||||||
|
(prefix-in lift: web-server/dispatchers/dispatch-lift))
|
||||||
|
|
||||||
|
;; Exercise real polling after three truncated downloads. A queued prefetch
|
||||||
|
;; must preserve the error, and a subsequent Stop must still be processed.
|
||||||
|
(test-case "failed download reports stopped and accepts subsequent commands"
|
||||||
|
(let* ((ready (make-async-channel))
|
||||||
|
(done (make-semaphore 0))
|
||||||
|
(attempts 0)
|
||||||
|
(prefetch-attempts 0)
|
||||||
|
(reports (make-hash))
|
||||||
|
(runtime #f)
|
||||||
|
(stop-server
|
||||||
|
(serve
|
||||||
|
#:port 0 #:listen-ip "127.0.0.1"
|
||||||
|
#:confirmation-channel ready #:connection-close? #t
|
||||||
|
#:dispatch
|
||||||
|
(lift:make
|
||||||
|
(λ (request)
|
||||||
|
(let ((path (url->string (request-uri request))))
|
||||||
|
(cond
|
||||||
|
((equal? path "/api/agent/register")
|
||||||
|
(response/jsexpr (hasheq)))
|
||||||
|
((equal? path "/api/agent/poll")
|
||||||
|
(let* ((data (bytes->jsexpr (request-post-data/raw request)))
|
||||||
|
(ack (hash-ref data 'ack))
|
||||||
|
(command
|
||||||
|
(case ack
|
||||||
|
((0) (hasheq 'id 1 'action "play"
|
||||||
|
'data (hasheq 'mediaToken "broken"
|
||||||
|
'filename "broken.flac"
|
||||||
|
'mediaSize 8)))
|
||||||
|
((1) (hasheq 'id 2 'action "prefetch"
|
||||||
|
'data (hasheq 'mediaToken "next"
|
||||||
|
'filename "next.flac"
|
||||||
|
'mediaSize 4)))
|
||||||
|
((2) (hasheq 'id 3 'action "stop"))
|
||||||
|
(else 'null))))
|
||||||
|
(hash-set! reports ack (hash-ref data 'state))
|
||||||
|
(when (= ack 3) (semaphore-post done))
|
||||||
|
(response/jsexpr (hasheq 'command command))))
|
||||||
|
(else
|
||||||
|
(when (regexp-match? #rx"broken$" path)
|
||||||
|
(set! attempts (+ attempts 1)))
|
||||||
|
(when (regexp-match? #rx"next$" path)
|
||||||
|
(set! prefetch-attempts (+ prefetch-attempts 1)))
|
||||||
|
(let ((body (if (and (regexp-match? #rx"next$" path)
|
||||||
|
(= prefetch-attempts 1))
|
||||||
|
#"12" #"1234")))
|
||||||
|
(response/output
|
||||||
|
(λ (output) (write-bytes body output))
|
||||||
|
#:headers
|
||||||
|
(list (header #"Content-Length"
|
||||||
|
(string->bytes/utf-8
|
||||||
|
(number->string (bytes-length body)))))))))))))))
|
||||||
|
(dynamic-wind
|
||||||
|
void
|
||||||
|
(λ ()
|
||||||
|
(let ((port (async-channel-get ready)))
|
||||||
|
(when (exn? port) (raise port))
|
||||||
|
(set! runtime
|
||||||
|
(make-player-agent-runtime
|
||||||
|
(format "http://127.0.0.1:~a" port) "Test" (make-string 64 #\a)))
|
||||||
|
((player-agent-runtime-start! runtime))
|
||||||
|
(check-not-false (sync/timeout 15 done))
|
||||||
|
(check-equal? attempts 3)
|
||||||
|
(check-equal? prefetch-attempts 2)
|
||||||
|
(let ((failed (hash-ref reports 1 (hasheq)))
|
||||||
|
(prefetched (hash-ref reports 2 (hasheq)))
|
||||||
|
(stopped (hash-ref reports 3 (hasheq))))
|
||||||
|
(check-equal? (hash-ref failed 'state #f) "stopped")
|
||||||
|
(check-regexp-match #rx"received 4 of 8 bytes"
|
||||||
|
(hash-ref failed 'error ""))
|
||||||
|
(check-equal? (hash-ref failed 'download #f) 'null)
|
||||||
|
(check-equal? (hash-ref prefetched 'error #f)
|
||||||
|
(hash-ref failed 'error ""))
|
||||||
|
(check-equal? (hash-ref stopped 'error #f) 'null)
|
||||||
|
(check-true ((player-agent-runtime-running? runtime))))))
|
||||||
|
(λ ()
|
||||||
|
(when runtime ((player-agent-runtime-shutdown! runtime)))
|
||||||
|
(stop-server)))))
|
||||||
|
|
||||||
|
;; An open stream that stops delivering bytes must not occupy the worker forever.
|
||||||
|
(let-values (((input writer) (make-pipe)))
|
||||||
|
(dynamic-wind
|
||||||
|
void
|
||||||
|
(λ ()
|
||||||
|
(check-exn #rx"Download timeout: no data received"
|
||||||
|
(λ ()
|
||||||
|
(copy-media-port input (open-output-bytes) void
|
||||||
|
#:idle-timeout 0.02))))
|
||||||
|
(λ () (close-input-port input) (close-output-port writer))))
|
||||||
|
|
||||||
;; Progress must arrive before EOF, even when a slow server has sent much
|
;; Progress must arrive before EOF, even when a slow server has sent much
|
||||||
;; less than one buffer. Binary data must reach the output unchanged.
|
;; less than one buffer. Binary data must reach the output unchanged.
|
||||||
|
|||||||
@@ -98,6 +98,7 @@
|
|||||||
(download-panel #f)
|
(download-panel #f)
|
||||||
(download-message #f)
|
(download-message #f)
|
||||||
(download-gauge #f)
|
(download-gauge #f)
|
||||||
|
(error-message #f)
|
||||||
(name-field #f)
|
(name-field #f)
|
||||||
(server-field #f)
|
(server-field #f)
|
||||||
(connect-button #f)
|
(connect-button #f)
|
||||||
@@ -214,6 +215,11 @@
|
|||||||
(not (string=? format-name "")))
|
(not (string=? format-name "")))
|
||||||
format-name
|
format-name
|
||||||
"")))))
|
"")))))
|
||||||
|
(when error-message
|
||||||
|
(let ((message (hash-ref snapshot 'error 'null)))
|
||||||
|
(send error-message show (string? message))
|
||||||
|
(send error-message set-label
|
||||||
|
(if (string? message) (cap200 message) ""))))
|
||||||
(when download-panel
|
(when download-panel
|
||||||
(send download-panel show (hash? download))
|
(send download-panel show (hash? download))
|
||||||
(when (hash? download)
|
(when (hash? download)
|
||||||
@@ -431,6 +437,12 @@
|
|||||||
(range 100)
|
(range 100)
|
||||||
(stretchable-width #t)))
|
(stretchable-width #t)))
|
||||||
(send download-panel show #f)
|
(send download-panel show #f)
|
||||||
|
(set! error-message
|
||||||
|
(new ellipsis-msg%
|
||||||
|
(parent playback-panel)
|
||||||
|
(label "")
|
||||||
|
[ellipsis 'right] [auto-resize #t] [stretchable-width #t]))
|
||||||
|
(send error-message show #f)
|
||||||
(set! quit-button
|
(set! quit-button
|
||||||
(new button%
|
(new button%
|
||||||
(parent controls)
|
(parent controls)
|
||||||
|
|||||||
@@ -359,7 +359,10 @@
|
|||||||
(make-semaphore 1))))
|
(make-semaphore 1))))
|
||||||
(set-dlna-playback-monitor!
|
(set-dlna-playback-monitor!
|
||||||
playback
|
playback
|
||||||
(thread (λ () (monitor-loop playback poll-seconds))))
|
(thread (λ ()
|
||||||
|
(with-handlers ((exn:break? void))
|
||||||
|
(parameterize-break #t
|
||||||
|
(monitor-loop playback poll-seconds))))))
|
||||||
playback))
|
playback))
|
||||||
|
|
||||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||||
@@ -483,7 +486,9 @@
|
|||||||
(set-dlna-playback-running?! playback #f)
|
(set-dlna-playback-running?! playback #f)
|
||||||
(let ((monitor (dlna-playback-monitor playback)))
|
(let ((monitor (dlna-playback-monitor playback)))
|
||||||
(when (and monitor (not (thread-dead? monitor)))
|
(when (and monitor (not (thread-dead? monitor)))
|
||||||
(kill-thread monitor))
|
;; A break unwinds call-with-semaphore; kill-thread leaves its lock held.
|
||||||
|
(break-thread monitor)
|
||||||
|
(thread-wait monitor))
|
||||||
(set-dlna-playback-monitor! playback #f)
|
(set-dlna-playback-monitor! playback #f)
|
||||||
(with-lock
|
(with-lock
|
||||||
playback
|
playback
|
||||||
|
|||||||
+12
-1
@@ -1461,6 +1461,11 @@
|
|||||||
(player-renderers value))
|
(player-renderers value))
|
||||||
'rendererId (or (playback-session-selected-id session) 'null)
|
'rendererId (or (playback-session-selected-id session) 'null)
|
||||||
'currentIndex (or (playback-session-current-index session) 'null)
|
'currentIndex (or (playback-session-current-index session) 'null)
|
||||||
|
'download
|
||||||
|
(let ((backend (playback-session-backend session)))
|
||||||
|
(if (and backend (eq? (playback-session-backend-kind session) 'agent))
|
||||||
|
(hash-ref (playback-agent-reported-state backend) 'download 'null)
|
||||||
|
'null))
|
||||||
'state (symbol->string (playback-session-state session))
|
'state (symbol->string (playback-session-state session))
|
||||||
'position (playback-session-position session)
|
'position (playback-session-position session)
|
||||||
'duration (or (playback-session-duration session) 'null)
|
'duration (or (playback-session-duration session) 'null)
|
||||||
@@ -2154,7 +2159,13 @@
|
|||||||
'state
|
'state
|
||||||
(hasheq 'state "stopped"
|
(hasheq 'state "stopped"
|
||||||
'position 0
|
'position 0
|
||||||
'volume 25))))
|
'volume 25
|
||||||
|
'download (hasheq 'filename "Large.flac"
|
||||||
|
'received 65536 'total 100000000)))))
|
||||||
|
|
||||||
|
(check-equal?
|
||||||
|
(hash-ref (player-state->jsexpr example-player) 'download)
|
||||||
|
(hasheq 'filename "Large.flac" 'received 65536 'total 100000000))
|
||||||
|
|
||||||
(check-equal?
|
(check-equal?
|
||||||
(hash-ref (hash-ref first-poll 'command) 'action)
|
(hash-ref (hash-ref first-poll 'command) 'action)
|
||||||
|
|||||||
+40
-5
@@ -7,6 +7,8 @@
|
|||||||
racket/runtime-path
|
racket/runtime-path
|
||||||
racket-mimetypes
|
racket-mimetypes
|
||||||
racket/string
|
racket/string
|
||||||
|
racket/path
|
||||||
|
simple-log
|
||||||
net/url
|
net/url
|
||||||
web-server/dispatch
|
web-server/dispatch
|
||||||
web-server/http
|
web-server/http
|
||||||
@@ -19,6 +21,8 @@
|
|||||||
|
|
||||||
(provide serve-player)
|
(provide serve-player)
|
||||||
|
|
||||||
|
(sl-def-log web-player-http)
|
||||||
|
|
||||||
(define-runtime-path public-directory "../public")
|
(define-runtime-path public-directory "../public")
|
||||||
|
|
||||||
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
||||||
@@ -176,16 +180,47 @@
|
|||||||
player
|
player
|
||||||
(request-jsexpr request)))))
|
(request-jsexpr request)))))
|
||||||
|
|
||||||
|
;;; Streams an agent download without holding player locks and reports its
|
||||||
|
;;; start, progress and outcome, including the client and actual bytes sent.
|
||||||
|
(define (stream-agent-media! file output client app-id)
|
||||||
|
(let ((received 0)
|
||||||
|
(total (file-size file))
|
||||||
|
(started (current-inexact-monotonic-milliseconds))
|
||||||
|
(reported (current-inexact-monotonic-milliseconds))
|
||||||
|
(completed? #f)
|
||||||
|
(name (file-name-from-path file)))
|
||||||
|
(dynamic-wind
|
||||||
|
(λ () (info-web-player-http "Agent download started: client=~a agent=~a file=~a bytes=~a"
|
||||||
|
client (substring app-id 0 8) name total))
|
||||||
|
(λ ()
|
||||||
|
(call-with-input-file
|
||||||
|
file
|
||||||
|
(λ (input)
|
||||||
|
(let ((buffer (make-bytes 65536)))
|
||||||
|
(let loop ()
|
||||||
|
(let ((count (read-bytes-avail! buffer input)))
|
||||||
|
(unless (eof-object? count)
|
||||||
|
(write-bytes buffer output 0 count)
|
||||||
|
(set! received (+ received count))
|
||||||
|
(when (>= (- (current-inexact-monotonic-milliseconds) reported) 5000)
|
||||||
|
(set! reported (current-inexact-monotonic-milliseconds))
|
||||||
|
(info-web-player-http "Agent download progress: client=~a file=~a bytes=~a/~a"
|
||||||
|
client name received total))
|
||||||
|
(loop)))))))
|
||||||
|
(flush-output output)
|
||||||
|
(set! completed? #t))
|
||||||
|
(λ ()
|
||||||
|
(info-web-player-http "Agent download ~a: client=~a file=~a bytes=~a/~a elapsed-ms=~a"
|
||||||
|
(if completed? "sent" "interrupted") client name received total
|
||||||
|
(round (- (current-inexact-monotonic-milliseconds) started)))))))
|
||||||
|
|
||||||
;;; Streams the media file identified by an agent's opaque token.
|
;;; Streams the media file identified by an agent's opaque token.
|
||||||
(define (agent-media-handler player _request app-id token)
|
(define (agent-media-handler player request app-id token)
|
||||||
(let ((file (player-agent-media player app-id token)))
|
(let ((file (player-agent-media player app-id token)))
|
||||||
(if (and file (file-exists? file))
|
(if (and file (file-exists? file))
|
||||||
(response/output
|
(response/output
|
||||||
(λ (output)
|
(λ (output)
|
||||||
(call-with-input-file
|
(stream-agent-media! file output (request-client-ip request) app-id))
|
||||||
file
|
|
||||||
(λ (input)
|
|
||||||
(copy-port input output))))
|
|
||||||
#:mime-type
|
#:mime-type
|
||||||
(let ((mime (mimetype-for-ext file)))
|
(let ((mime (mimetype-for-ext file)))
|
||||||
(if (string? mime)
|
(if (string? mime)
|
||||||
|
|||||||
+15
-1
@@ -146,12 +146,21 @@ async function api(path, body) {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
const response = await fetch(path, options);
|
const controller = new AbortController();
|
||||||
|
const timeout = setTimeout(() => controller.abort(), 30000);
|
||||||
|
try {
|
||||||
|
const response = await fetch(path, { ...options, signal: controller.signal });
|
||||||
const data = await response.json();
|
const data = await response.json();
|
||||||
if (!response.ok) {
|
if (!response.ok) {
|
||||||
throw new ApiError(data.error || `HTTP ${response.status}`, response.status, data.code);
|
throw new ApiError(data.error || `HTTP ${response.status}`, response.status, data.code);
|
||||||
}
|
}
|
||||||
return data;
|
return data;
|
||||||
|
} catch (error) {
|
||||||
|
if (controller.signal.aborted) throw new Error("request-timeout");
|
||||||
|
throw error;
|
||||||
|
} finally {
|
||||||
|
clearTimeout(timeout);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Displays the login overlay and optionally reports why authentication failed.
|
// Displays the login overlay and optionally reports why authentication failed.
|
||||||
@@ -671,6 +680,11 @@ function render(nextState) {
|
|||||||
renderPlayer(nextState);
|
renderPlayer(nextState);
|
||||||
if (nextState.error) {
|
if (nextState.error) {
|
||||||
setStatus(t(nextState.error));
|
setStatus(t(nextState.error));
|
||||||
|
} else if (nextState.download) {
|
||||||
|
const download = nextState.download;
|
||||||
|
const received = (download.received / 1000000).toFixed(1);
|
||||||
|
const total = Number.isFinite(download.total) ? (download.total / 1000000).toFixed(1) : "?";
|
||||||
|
setStatus(t("downloadingMedia", { name: download.filename, received, total }));
|
||||||
} else {
|
} else {
|
||||||
setStatus(nextState.discovering ? t("searchingPlayers") : "");
|
setStatus(nextState.discovering ? t("searchingPlayers") : "");
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,6 +3,8 @@
|
|||||||
|
|
||||||
const translations = {
|
const translations = {
|
||||||
en: {
|
en: {
|
||||||
|
"request-timeout": "The server did not respond within 30 seconds. Status will be refreshed automatically.",
|
||||||
|
downloadingMedia: "Downloading {name}: {received} / {total} MB",
|
||||||
folders: "Folders",
|
folders: "Folders",
|
||||||
savePlaylist: "SAVE PLAYLIST",
|
savePlaylist: "SAVE PLAYLIST",
|
||||||
playlistSaved: "SAVED",
|
playlistSaved: "SAVED",
|
||||||
@@ -41,6 +43,8 @@
|
|||||||
"dlna-renderer-unreachable": "The DLNA renderer is unreachable.",
|
"dlna-renderer-unreachable": "The DLNA renderer is unreachable.",
|
||||||
},
|
},
|
||||||
nl: {
|
nl: {
|
||||||
|
"request-timeout": "De server antwoordde niet binnen 30 seconden. De status wordt automatisch opnieuw opgehaald.",
|
||||||
|
downloadingMedia: "Downloaden van {name}: {received} / {total} MB",
|
||||||
folders: "Mappen",
|
folders: "Mappen",
|
||||||
savePlaylist: "AFSPEELLIJST OPSLAAN",
|
savePlaylist: "AFSPEELLIJST OPSLAAN",
|
||||||
playlistSaved: "OPGESLAGEN",
|
playlistSaved: "OPGESLAGEN",
|
||||||
|
|||||||
@@ -143,6 +143,23 @@ the agent exits; it is not retained across restarts.
|
|||||||
A track that is not cached is still downloaded completely before playback.
|
A track that is not cached is still downloaded completely before playback.
|
||||||
The HTTP server allows up to one hour for a response, including media
|
The HTTP server allows up to one hour for a response, including media
|
||||||
transfers, so large lossless tracks can finish over slower connections.
|
transfers, so large lossless tracks can finish over slower connections.
|
||||||
|
The agent imposes no total download deadline: receiving another chunk restarts
|
||||||
|
its 30-second inactivity timer. A connection or transfer with no response bytes
|
||||||
|
for 30 seconds fails, closes its ports and removes the partial file. The private
|
||||||
|
@tt{call-with-http-timeout} procedure runs an HTTP operation with a
|
||||||
|
@tt{progress!} callback; each callback restarts this timer. Cancellation unwinds
|
||||||
|
the worker before its custodian closes the remaining HTTP resources.
|
||||||
|
|
||||||
|
The browser displays the selected agent's reported download filename and byte
|
||||||
|
counts, including prefetch downloads. Server logs record agent download start,
|
||||||
|
progress every five seconds while sending, and completion or interruption with
|
||||||
|
client address, filename and byte counts. DLNA media transfers are logged by
|
||||||
|
@tt{racket-upnp} with client address, method, range and elapsed time.
|
||||||
|
|
||||||
|
Browser API requests time out after 30 seconds, freeing the request queue and
|
||||||
|
allowing status polling to resume. Timing out a browser request does not undo a
|
||||||
|
command already accepted by the server.
|
||||||
|
|
||||||
The next track is fetched after the play command completes. Commands are
|
The next track is fetched after the play command completes. Commands are
|
||||||
processed in server order, so a pending prefetch can still delay a later manual
|
processed in server order, so a pending prefetch can still delay a later manual
|
||||||
track selection.
|
track selection.
|
||||||
@@ -351,6 +368,13 @@ During downloads, including prefetching the next track, the GUI shows the
|
|||||||
filename and received megabytes. When the server supplies the file size,
|
filename and received megabytes. When the server supplies the file size,
|
||||||
it also shows the total size, percentage and a progress bar. The display
|
it also shows the total size, percentage and a progress bar. The display
|
||||||
updates every half second and disappears when the download finishes or fails.
|
updates every half second and disappears when the download finishes or fails.
|
||||||
|
Interrupted downloads are retried twice, starting again from the beginning.
|
||||||
|
After 60 seconds without incoming media data, the agent aborts the attempt
|
||||||
|
instead of waiting indefinitely. If all three attempts fail, a failed play
|
||||||
|
returns to the stopped state and the GUI displays the error, including byte
|
||||||
|
counts for an incomplete file. Background prefetching and audio status updates
|
||||||
|
preserve this message; a new playback command clears it. Polling continues,
|
||||||
|
so subsequent commands can still be processed.
|
||||||
The GUI and its @tt{racket-tray} system tray support the same ten languages
|
The GUI and its @tt{racket-tray} system tray support the same ten languages
|
||||||
based on the operating-system language, with English as fallback. Closing or
|
based on the operating-system language, with English as fallback. Closing or
|
||||||
minimizing the window hides it in the tray. The tray menu restores the window
|
minimizing the window hides it in the tray. The tray menu restores the window
|
||||||
|
|||||||
Reference in New Issue
Block a user