822 lines
35 KiB
Racket
822 lines
35 KiB
Racket
#lang racket/base
|
|
|
|
(require json
|
|
net/url
|
|
racket-audio
|
|
racket/contract
|
|
racket/file
|
|
racket/path
|
|
racket/port
|
|
racket/string
|
|
simple-log
|
|
"player-agent-translate.rkt")
|
|
|
|
(provide (struct-out player-agent-runtime)
|
|
make-player-agent-runtime)
|
|
|
|
(sl-def-log player-agent)
|
|
|
|
(struct exn:fail:agent-denied exn:fail () #:transparent)
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
; goal : Expose the small procedure-based interface of a running agent.
|
|
; pre : Constructor fields are lifecycle/query procedures and a stable ID.
|
|
; post : Creating or recognizing a value changes no external state.
|
|
; result : player-agent-runtime? recognizes values returned by the factory.
|
|
; internals: make-player-agent-runtime stores its local start!, reconnect!,
|
|
; shutdown!, snapshot and current-track procedures in this struct.
|
|
; Those procedures retain access to the factory closure, keeping the
|
|
; shared polling and audio state private without introducing a class.
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
(struct player-agent-runtime
|
|
(start! reconnect! shutdown! snapshot current-track running? app-id)
|
|
#:transparent)
|
|
|
|
;;; Normalizes a configured server address and converts it to a URL value.
|
|
(define (base-url value)
|
|
(string->url
|
|
(regexp-replace #px"/+$" (string-trim value) "")))
|
|
|
|
;;; Resolves an agent API path relative to a normalized server URL.
|
|
(define (endpoint-url base path)
|
|
(combine-url/relative (base-url base) path))
|
|
|
|
;;; 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)))))
|
|
|
|
;;; Converts racket-audio states to the state names sent to the web player.
|
|
(define (normal-state state)
|
|
(cond
|
|
((memq state '(initialized no-media)) "stopped")
|
|
((eq? state 'transitioning) "starting")
|
|
(else (symbol->string state))))
|
|
|
|
;;; Deletes a temporary media file and logs recoverable deletion failures.
|
|
(define (safe-delete-file file)
|
|
(when (and file (file-exists? file))
|
|
(with-handlers ((exn:fail?
|
|
(λ (exception)
|
|
(warn-player-agent
|
|
"Could not remove temporary media file ~a: ~a"
|
|
file
|
|
(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! #:idle-timeout [idle-timeout 60])
|
|
(let ((buffer (make-bytes 65536)))
|
|
(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)))
|
|
(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])
|
|
(let loop ((entries (sort (hash->list cache) <
|
|
#:key (λ (entry) (cadddr entry))))
|
|
(size (apply + (map cadr (hash-values cache)))))
|
|
(when (and (> size limit) (pair? entries))
|
|
(let* ((entry (car entries))
|
|
(key (car entry))
|
|
(path (cadr entry)))
|
|
(cond
|
|
((member key protected) (loop (cdr entries) size))
|
|
(else
|
|
(safe-delete-file path)
|
|
(if (file-exists? path)
|
|
(loop (cdr entries) size)
|
|
(begin
|
|
(hash-remove! cache key)
|
|
(loop (cdr entries) (- size (caddr entry)))))))))))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; Provided functions
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
; goal : Create the headless polling and audio runtime for one agent.
|
|
; pre : Server URL, display name and application ID are strings; callbacks
|
|
; accept the status/denial messages supplied to them.
|
|
; post : Mutable state is initialized but no worker thread or audio backend
|
|
; is started until the returned start! procedure is called.
|
|
; result : A player-agent-runtime containing its lifecycle/query procedures.
|
|
; internals: start! launches poll-loop, which registers through post-json and
|
|
; sends snapshots until it receives a command. A command worker runs
|
|
; execute-command! and acknowledges it only after completion.
|
|
; ensure-audio! connects racket-audio callbacks to update-from-audio!
|
|
; and advance-at-decoder-eof!. with-agent-state protects their shared
|
|
; state; stop! and shutdown! stop threads, audio and cached files.
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
(define/contract (make-player-agent-runtime initial-server-url
|
|
initial-name
|
|
app-id
|
|
#:status-callback
|
|
[status-callback void]
|
|
#:denied-callback
|
|
[denied-callback void])
|
|
(->* (string? string? string?)
|
|
(#:status-callback (-> string? any/c)
|
|
#:denied-callback (-> string? any/c))
|
|
player-agent-runtime?)
|
|
(let* ((server-url initial-server-url)
|
|
(assigned-name initial-name)
|
|
(state-lock (make-semaphore 1))
|
|
(worker #f)
|
|
(command-worker #f)
|
|
(executing-command-id 0)
|
|
(running #f)
|
|
(authorization-notified? #f)
|
|
(audio #f)
|
|
(current-media-key #f)
|
|
(loading-media-key #f)
|
|
(queued-media-key #f)
|
|
(cached-media (make-hash))
|
|
(prefetched-track #f)
|
|
(auto-started-key #f)
|
|
(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)
|
|
(agent-state
|
|
(hasheq 'state "stopped"
|
|
'position 0
|
|
'duration 'null
|
|
'rate 'null
|
|
'channels 'null
|
|
'bits 'null
|
|
'format ""
|
|
'volume logical-volume
|
|
'error 'null)))
|
|
|
|
;;; Runs a procedure while holding the semaphore for shared agent state.
|
|
(define (with-agent-state proc)
|
|
(call-with-semaphore state-lock proc))
|
|
|
|
;;; Replaces an unavailable state value with its JSON fallback.
|
|
(define (state-value value fallback)
|
|
(if (eq? value #f) fallback value))
|
|
|
|
;;; 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
|
|
(λ () (hash-set agent-state 'download download-progress))))
|
|
|
|
;;; Returns the currently audible track while holding the state semaphore.
|
|
(define (current-track)
|
|
(with-agent-state (λ () current-track-value)))
|
|
|
|
;;; Stores an error in the state reported by the next poll.
|
|
(define (set-agent-error! message)
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! agent-state (hash-set agent-state 'error message)))))
|
|
|
|
;;; Removes an earlier error from the state reported by the next poll.
|
|
(define (clear-agent-error!)
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! agent-state (hash-set agent-state 'error 'null)))))
|
|
|
|
;;; Copies racket-audio state into the agent snapshot.
|
|
;;; It also confirms when a prefetched track has become audible.
|
|
(define (update-from-audio! state full-state)
|
|
(with-agent-state
|
|
(λ ()
|
|
(let ((audible-music-id (hash-ref full-state 'at-music-id #f)))
|
|
(set! agent-state
|
|
(hasheq
|
|
'state (normal-state state)
|
|
'position (state-value (hash-ref full-state 'at-second #f) 0)
|
|
'duration (state-value (hash-ref full-state 'duration #f) 'null)
|
|
'rate (state-value (hash-ref full-state 'rate #f) 'null)
|
|
'channels (state-value (hash-ref full-state 'channels #f) 'null)
|
|
'bits (state-value (hash-ref full-state 'bits #f) 'null)
|
|
'format (let ((decoder (hash-ref full-state 'decoder #f)))
|
|
(if decoder (format "~a" decoder) ""))
|
|
'volume logical-volume
|
|
'error (hash-ref agent-state 'error 'null)))
|
|
(when (and pending-auto-music-id
|
|
(number? audible-music-id)
|
|
(= pending-auto-music-id audible-music-id))
|
|
(let ((audible-track
|
|
(hash-ref music-tracks audible-music-id #f)))
|
|
(when audible-track
|
|
(set! current-track-value audible-track)
|
|
(hash-clear! music-tracks)
|
|
(hash-set! music-tracks audible-music-id audible-track)))
|
|
(set! pending-auto-music-id #f)
|
|
(set! ended-counter (+ ended-counter 1)))))))
|
|
|
|
;;; Creates and configures the audio player on first use.
|
|
;;; Its callbacks update reported state and continue prefetched playback.
|
|
(define (ensure-audio!)
|
|
(unless audio
|
|
(set! audio
|
|
(make-audio-player
|
|
(λ (_handle state full-state)
|
|
(update-from-audio! state full-state))
|
|
(λ (handle)
|
|
(advance-at-decoder-eof! handle))
|
|
#:use-place #f
|
|
))
|
|
(audio-ao-buf-ms! audio 500)
|
|
(audio-buf-seconds! audio 4 10)
|
|
(let ((scaled (/ logical-volume 100.0)))
|
|
(audio-volume! audio (* 100.0 scaled scaled))))
|
|
audio)
|
|
|
|
;;; Downloads one protected media resource to a temporary local file.
|
|
;;; A failed download closes its port and removes its partial file.
|
|
(define (download-media-once! token filename size)
|
|
(let* ((started (current-inexact-monotonic-milliseconds))
|
|
(extension
|
|
(or (path-get-extension (string->path filename)) #""))
|
|
(target
|
|
(make-temporary-file
|
|
(string-append "rkt-player-agent-~a"
|
|
(bytes->string/utf-8 extension))))
|
|
(path (format "/api/agent/media/~a/~a" app-id token))
|
|
(input #f))
|
|
(with-handlers ((exn:fail?
|
|
(λ (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-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)
|
|
started)))
|
|
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.
|
|
;;; Older servers without version metadata only reuse the same media token.
|
|
(define (command-cache-key data)
|
|
(list server-url
|
|
(hash-ref data 'cacheKey (hash-ref data 'mediaToken))
|
|
(hash-ref data 'mediaSize #f)
|
|
(hash-ref data 'mediaModified (hash-ref data 'mediaToken))))
|
|
|
|
;;; Returns cached media or downloads and records it when absent.
|
|
(define (ensure-media-cached! data)
|
|
(let* ((key (command-cache-key data))
|
|
(found (with-agent-state
|
|
(λ ()
|
|
(set! loading-media-key key)
|
|
(hash-ref cached-media key #f)))))
|
|
(if (and found (file-exists? (car found))
|
|
(= (file-size (car found)) (cadr found)))
|
|
(begin
|
|
(info-player-agent "Cache hit: ~a" (hash-ref data 'filename "track"))
|
|
(with-agent-state
|
|
(λ ()
|
|
(hash-set! cached-media key
|
|
(list (car found) (cadr found)
|
|
(current-inexact-monotonic-milliseconds)))))
|
|
(car found))
|
|
(let ((downloaded
|
|
(download-media!
|
|
(hash-ref data 'mediaToken)
|
|
(hash-ref data 'filename "track")
|
|
(hash-ref data 'mediaSize #f))))
|
|
(with-agent-state
|
|
(λ ()
|
|
(when found (safe-delete-file (car found)))
|
|
(hash-set! cached-media key
|
|
(list downloaded (file-size downloaded)
|
|
(current-inexact-monotonic-milliseconds)))))
|
|
downloaded))))
|
|
|
|
;;; Bounds retained media while protecting the current and prepared tracks.
|
|
;;; The caller holds state-lock so decoder callbacks cannot race eviction.
|
|
(define (trim-cache!)
|
|
(trim-media-cache!
|
|
cached-media
|
|
(list current-media-key
|
|
loading-media-key
|
|
queued-media-key
|
|
(and prefetched-track (command-cache-key (car prefetched-track))))))
|
|
|
|
;;; Continues with prefetched media when the current decoder reaches EOF.
|
|
;;; Decoder EOF precedes audible EOF, so audio-play! queues behind the buffer.
|
|
(define (advance-at-decoder-eof! handle)
|
|
(let ((prepared
|
|
(with-agent-state
|
|
(λ ()
|
|
(let ((value prefetched-track))
|
|
(when value
|
|
(set! queued-media-key (command-cache-key (car value))))
|
|
(set! prefetched-track #f)
|
|
value)))))
|
|
(cond
|
|
(prepared
|
|
(let* ((data (car prepared))
|
|
(path (cdr prepared))
|
|
(key (command-cache-key data)))
|
|
(with-handlers
|
|
((exn:fail?
|
|
(λ (exception)
|
|
(warn-player-agent "Could not start prefetched track: ~a"
|
|
(exn-message exception))
|
|
(set-agent-error! (exn-message exception))
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! queued-media-key #f)
|
|
(trim-cache!)
|
|
(set! ended-counter (+ ended-counter 1)))))))
|
|
(let ((music-id (audio-play! handle path)))
|
|
(info-player-agent "Queued prefetched track ~a as music id ~a"
|
|
(hash-ref data 'filename "track")
|
|
music-id)
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! current-media-key key)
|
|
(set! queued-media-key #f)
|
|
(trim-cache!)
|
|
(hash-set! music-tracks music-id data)
|
|
(set! auto-started-key key)
|
|
(set! pending-auto-music-id music-id)))))))
|
|
(else
|
|
(warn-player-agent
|
|
"Decoder reached EOF before the next track was prefetched")
|
|
(with-agent-state
|
|
(λ () (set! ended-counter (+ ended-counter 1))))))))
|
|
|
|
;;; Applies one server command to audio, cache and reported agent state.
|
|
;;; Play and prefetch commands also maintain gapless track bookkeeping.
|
|
(define (execute-command! command)
|
|
(let ((action (hash-ref command 'action ""))
|
|
(data (hash-ref command 'data (hasheq))))
|
|
(info-player-agent "Executing command ~a: ~a"
|
|
(hash-ref command 'id) action)
|
|
(cond
|
|
((string=? action "play")
|
|
(let* ((next-key (command-cache-key data))
|
|
(already-started?
|
|
(with-agent-state
|
|
(λ ()
|
|
(let ((matches?
|
|
(and auto-started-key
|
|
(equal? auto-started-key next-key))))
|
|
(when matches?
|
|
(set! auto-started-key #f))
|
|
matches?)))))
|
|
(with-agent-state
|
|
(λ () (set! current-track-value data)))
|
|
(unless already-started?
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! prefetched-track #f)
|
|
(set! auto-started-key #f)
|
|
(set! pending-auto-music-id #f)
|
|
(set! agent-state
|
|
(hash-set
|
|
(hash-set agent-state 'state "starting")
|
|
'error 'null))))
|
|
(let* ((next-media (ensure-media-cached! data))
|
|
(handle (ensure-audio!))
|
|
(started (current-inexact-monotonic-milliseconds))
|
|
;; audio-play! interrupts and closes the previous decoder.
|
|
(music-id (audio-play! handle next-media)))
|
|
(info-player-agent "Audio accepted ~a as music id ~a in ~a ms"
|
|
(hash-ref data 'filename "track") music-id
|
|
(round (- (current-inexact-monotonic-milliseconds)
|
|
started)))
|
|
(with-agent-state
|
|
(λ ()
|
|
(hash-clear! music-tracks)
|
|
(hash-set! music-tracks music-id data)
|
|
(set! current-media-key next-key)
|
|
(set! loading-media-key #f)
|
|
(trim-cache!)))))))
|
|
((string=? action "prefetch")
|
|
(let ((path (ensure-media-cached! data)))
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! prefetched-track (cons data path))
|
|
(set! loading-media-key #f)
|
|
(trim-cache!)))
|
|
(info-player-agent "Prefetched ~a"
|
|
(hash-ref data 'filename "track"))))
|
|
((string=? action "pause")
|
|
(audio-pause! (ensure-audio!) #t))
|
|
((string=? action "resume")
|
|
(audio-pause! (ensure-audio!) #f))
|
|
((string=? action "stop")
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! prefetched-track #f)
|
|
(set! auto-started-key #f)
|
|
(set! pending-auto-music-id #f)))
|
|
(when audio
|
|
(audio-stop! audio))
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! current-media-key #f)
|
|
(trim-cache!))))
|
|
((string=? action "seek")
|
|
(audio-seek! (ensure-audio!) (hash-ref data 'percentage 0)))
|
|
((string=? action "volume")
|
|
(set! logical-volume (min 100 (max 0 (hash-ref data 'value 50))))
|
|
(let ((scaled (/ logical-volume 100.0)))
|
|
(audio-volume! (ensure-audio!) (* 100.0 scaled scaled)))
|
|
(with-agent-state
|
|
(λ ()
|
|
(set! agent-state
|
|
(hash-set agent-state 'volume logical-volume)))))
|
|
(else
|
|
(error 'player-agent "unknown command: ~a" action)))))
|
|
|
|
;;; Registers the agent and repeatedly exchanges state for server commands.
|
|
;;; Connection and authorization failures are reported before a delayed retry.
|
|
(define (poll-loop)
|
|
(with-handlers
|
|
((exn:fail:agent-denied?
|
|
(λ (exception)
|
|
(let ((message (format (tr 'denied-message) app-id)))
|
|
(warn-player-agent "Agent authorization refused: ~a"
|
|
(exn-message exception))
|
|
(set-agent-error! message)
|
|
(status-callback
|
|
(tr 'unauthorized-status))
|
|
(unless authorization-notified?
|
|
(set! authorization-notified? #t)
|
|
(denied-callback message))
|
|
(when running
|
|
(sleep 3)
|
|
(poll-loop)))))
|
|
(exn:fail?
|
|
(λ (exception)
|
|
(warn-player-agent "Connection cycle failed: ~a"
|
|
(exn-message exception))
|
|
(set-agent-error! (exn-message exception))
|
|
(status-callback
|
|
(format (tr 'disconnected) (exn-message exception)))
|
|
(when running
|
|
(sleep 3)
|
|
(poll-loop)))))
|
|
(post-json server-url
|
|
"/api/agent/register"
|
|
(hasheq 'appId app-id 'name assigned-name))
|
|
(clear-agent-error!)
|
|
(status-callback (tr 'connected))
|
|
(info-player-agent "Registered at ~a as ~a" server-url assigned-name)
|
|
(let loop ()
|
|
(when running
|
|
(let* ((response
|
|
(post-json
|
|
server-url
|
|
"/api/agent/poll"
|
|
(hasheq 'appId app-id
|
|
'name assigned-name
|
|
'ack acknowledged-command
|
|
'endedCounter ended-counter
|
|
'state (snapshot))))
|
|
(command (hash-ref response 'command 'null)))
|
|
(when (and (hash? command)
|
|
(> (hash-ref command 'id 0) acknowledged-command)
|
|
(not (= (hash-ref command 'id 0)
|
|
executing-command-id)))
|
|
(set! executing-command-id (hash-ref command 'id))
|
|
(set! command-worker
|
|
(thread
|
|
(λ ()
|
|
(with-handlers
|
|
((exn:fail?
|
|
(λ (exception)
|
|
(warn-player-agent "Command failed: ~a"
|
|
(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
|
|
(λ ()
|
|
(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!)))
|
|
(set-agent-error! (exn-message exception)))))
|
|
;; A queued prefetch must not erase a failed play's error.
|
|
(unless (equal? (hash-ref command 'action) "prefetch")
|
|
(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)))))
|
|
|
|
;;; Starts the polling worker once and reports the connecting state.
|
|
(define (start!)
|
|
(unless running
|
|
(set! running #t)
|
|
(status-callback (tr 'connecting))
|
|
(set! worker (thread 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))
|
|
(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.
|
|
(define (reconnect! new-server-url new-name)
|
|
(stop!)
|
|
(set! authorization-notified? #f)
|
|
(set! server-url (string-trim new-server-url))
|
|
(set! assigned-name (string-trim new-name))
|
|
(start!))
|
|
|
|
;;; Stops the runtime, closes audio and removes all cached media files.
|
|
(define (shutdown!)
|
|
(stop!)
|
|
(when audio
|
|
(with-handlers ((exn:fail? void))
|
|
(audio-quit! audio))
|
|
(set! audio #f))
|
|
(let loop ((paths (hash-values cached-media)))
|
|
(unless (null? paths)
|
|
(safe-delete-file (caar paths))
|
|
(loop (cdr paths))))
|
|
(hash-clear! cached-media))
|
|
|
|
(player-agent-runtime start!
|
|
reconnect!
|
|
shutdown!
|
|
snapshot
|
|
current-track
|
|
(λ () running)
|
|
app-id)))
|
|
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
;; Tests for module player-agent-core.rkt
|
|
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
|
|
|
|
(module+ test
|
|
(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
|
|
;; 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")
|
|
|
|
(let ((directory (make-temporary-file "agent-cache-test-~a" 'directory))
|
|
(cache (make-hash)))
|
|
(dynamic-wind
|
|
void
|
|
(λ ()
|
|
(for-each
|
|
(λ (key age)
|
|
(let ((path (build-path directory (symbol->string key))))
|
|
(display-to-file "1234" path)
|
|
(hash-set! cache key (list path 4 age))))
|
|
'(old recent current prepared)
|
|
'(1 4 2 3))
|
|
;; Entries below the budget survive track changes.
|
|
(trim-media-cache! cache '(current prepared) 16)
|
|
(check-equal? (hash-count cache) 4)
|
|
;; Eviction follows last use, not insertion or playback order.
|
|
(trim-media-cache! cache '(current prepared) 12)
|
|
(check-false (hash-has-key? cache 'old))
|
|
(check-false (file-exists? (build-path directory "old")))
|
|
(check-true (hash-has-key? cache 'recent))
|
|
;; Active files survive even when they exceed the entire budget.
|
|
(trim-media-cache! cache '(current prepared) 4)
|
|
(check-equal? (hash-count cache) 2)
|
|
(check-true (file-exists? (build-path directory "current")))
|
|
(check-true (file-exists? (build-path directory "prepared")))
|
|
;; Releasing active files lets the cache return to its limit.
|
|
(trim-media-cache! cache '() 4)
|
|
(check-equal? (hash-keys cache) '(prepared)))
|
|
(λ () (delete-directory/files directory))))
|
|
|
|
(let* ((app-id (make-string 64 #\a))
|
|
(runtime
|
|
(make-player-agent-runtime "http://127.0.0.1:1234"
|
|
"Test agent"
|
|
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")
|
|
(check-equal? (player-agent-runtime-app-id runtime) app-id)))
|