Carry a picture end to end, in jolt
frq.av.plane is what libjoltmoq was, assembled from the pieces the last few commits bound. A synthetic I420 frame goes out through openh264 onto a real MoQ media track, comes back through a real subscribe, and arrives as RGBA ready for vidya/frame-rgba!. Nothing between those ends is stubbed. It keeps joltmoq's SHAPE deliberately -- start!/stop!/live?, poll-status!, poll-frame! -- so that frq.av eventually changes call sites rather than being rewritten around a new idea. PUMPED, NOT THREADED, and that is a decision rather than laziness. jolt has fibers, but a fiber is bound to its carrier for life and a blocking foreign call pins that carrier and strands everything queued behind it -- and the two things this does most often, V4L2's DQBUF and ALSA's readi, are exactly that. Pumping also keeps joltmoq's frame contract for free: at most one frame is decoded per pump, so the pointer handed out stays good until the next one, which is what joltmoq_frame_rgba promised in the first place. A source is a THUNK answering [pointer length], not a camera. V4L2 is one such thunk, a test pattern is another, and Camera2 will be a third without this namespace learning about JNI. That is also what lets the whole path be tested on a machine with no camera. The decode happens inside the lift, while the RustBuffer the payload points into is still alive. Lifting the span out and decoding after would read a freed buffer, and what comes back from that is not a fault but plausible rubbish. The RGBA pointer it answers belongs to the decoder, not the RustBuffer, so it outlives the lift and is the borrow poll-frame! hands on. The test frame has a dark half and a bright half rather than being flat: a decode producing a uniform picture is a failure a gray frame cannot tell from success. 0x40 and 0xC0 come back as 56 and 205. STILL NOT PORTED: frq.av is untouched. This plane is one peer and video only. Audio is bound but not wired -- mixing several peers into one playback stream wants a jitter buffer and a resampler for clock drift, and a bad one is worse than none. Multi-peer needs the announce and catalog handling that finds them. Android has neither V4L2 nor ALSA. Those are what stand between this and deleting the joltmoq entry from deps.edn. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
5db1a03 parent: 982c702 added
src/frq/av/plane.clj +216 -0 | new file mode 100644 | ||
| @@ -0,0 +1,216 @@ | ||
| 1 | +(ns frq.av.plane | |
| 2 | + "The media plane, in jolt — what `libjoltmoq` was. | |
| 3 | + | |
| 4 | + `joltmoq_start` was one call that connected, published a camera, subscribed | |
| 5 | + to every peer, decoded their video and mixed their audio, on its own Rust | |
| 6 | + threads. This is the same job assembled from the pieces `frq.moq.*`, | |
| 7 | + `frq.codec.*` and `frq.capture.*` now provide, and it deliberately keeps | |
| 8 | + joltmoq's SHAPE so that `frq.av` becomes a change of call sites rather than | |
| 9 | + a rewrite: | |
| 10 | + | |
| 11 | + start! stop! live? joltmoq_start / _stop / _is_live | |
| 12 | + poll-status! joltmoq_poll_status + _status_text | |
| 13 | + poll-frame! joltmoq_frame_poll + _frame_rgba | |
| 14 | + | |
| 15 | + WHY IT IS PUMPED AND NOT THREADED. jolt has fibers, but a fiber is bound to | |
| 16 | + its carrier for life and a blocking foreign call pins that carrier and | |
| 17 | + strands everything queued behind it — and the two things this has to do | |
| 18 | + most often, V4L2's DQBUF and ALSA's readi, are exactly that. So the plane is | |
| 19 | + driven from `pump!`, which glimmer already calls from a timer for `frq.av`. | |
| 20 | + It also keeps joltmoq's frame contract intact for free: at most one frame is | |
| 21 | + decoded per poll, so the pointer handed out stays valid until the next one, | |
| 22 | + which is precisely what `joltmoq_frame_rgba` promised. | |
| 23 | + | |
| 24 | + WHAT IS HERE. One peer, video only. Outbound is capture → H.264 → a MoQ | |
| 25 | + media track; inbound is that track → H.264 → RGBA. That is enough to carry | |
| 26 | + a picture end to end and it is deliberately the smallest thing that can be, | |
| 27 | + because the parts that are NOT here are the ones worth doing carefully: | |
| 28 | + | |
| 29 | + * more than one peer, and the announce/catalog handling that finds them | |
| 30 | + * audio at all — Opus and ALSA are bound, but mixing several peers into | |
| 31 | + one playback stream needs a jitter buffer and a resampler for clock | |
| 32 | + drift, and a bad one is worse than none | |
| 33 | + * Android, where neither V4L2 nor ALSA exists | |
| 34 | + | |
| 35 | + A SOURCE IS A FUNCTION, not a camera. `start!` takes `:source`, a thunk | |
| 36 | + answering [pointer length] for one I420 frame or nil for \"nothing right | |
| 37 | + now\". `frq.capture.v4l2` is one such thunk; a test pattern is another. That | |
| 38 | + is what lets the plane be exercised on a machine with no camera, and it is | |
| 39 | + also how the phone will pass a Camera2 buffer in later without this | |
| 40 | + namespace learning about JNI." | |
| 41 | + (:require [frq.moq.media :as media] | |
| 42 | + [frq.moq.uniffi :as uniffi] | |
| 43 | + [frq.codec.h264 :as h264] | |
| 44 | + [jolt.ffi :as ffi])) | |
| 45 | + | |
| 46 | +;; --- state ------------------------------------------------------------------- | |
| 47 | +;; One plane per process, as joltmoq had: its C API was all globals, and the | |
| 48 | +;; call surface above it assumes a single call at a time. | |
| 49 | + | |
| 50 | +(declare stop!) | |
| 51 | + | |
| 52 | +(defonce ^:private plane (atom nil)) | |
| 53 | + | |
| 54 | +(defn live? [] (some? @plane)) | |
| 55 | + | |
| 56 | +;; --- starting ---------------------------------------------------------------- | |
| 57 | + | |
| 58 | +(defn start! | |
| 59 | + "Bring the plane up and answer true, or false with a reason recorded. | |
| 60 | + | |
| 61 | + `origin` is a MoqOriginProducer — from a session for a real call, or made | |
| 62 | + locally for a test, which is the same object either way. `source` is the | |
| 63 | + frame thunk described above. | |
| 64 | + | |
| 65 | + Everything that can fail does so HERE rather than at the first frame: the | |
| 66 | + encoder validates its size, the decoder opens, and the subscribe settles, | |
| 67 | + so a plane that comes up is one that can carry a picture." | |
| 68 | + [{:keys [origin path source width height fps bitrate camera?] | |
| 69 | + :or {path "/frq" width 640 height 480 fps 30 bitrate 800000 camera? true}}] | |
| 70 | + (stop!) | |
| 71 | + (let [broadcast (media/create-broadcast! origin path) | |
| 72 | + producer (media/publish-media! broadcast "avc3") | |
| 73 | + track (media/producer-name producer) | |
| 74 | + consumer (media/broadcast-consumer broadcast)] | |
| 75 | + (reset! plane | |
| 76 | + {:broadcast broadcast | |
| 77 | + :producer producer | |
| 78 | + :track track | |
| 79 | + :consumer consumer | |
| 80 | + :subscribe (media/subscribe-media! consumer track media/video-container) | |
| 81 | + :media nil | |
| 82 | + :pending nil | |
| 83 | + :encoder (h264/encoder {:width width :height height | |
| 84 | + :fps fps :bitrate bitrate}) | |
| 85 | + :decoder (h264/decoder) | |
| 86 | + :source source | |
| 87 | + :size [width height] | |
| 88 | + :camera? camera? | |
| 89 | + :frame nil | |
| 90 | + :status (atom []) | |
| 91 | + :pts (atom 0) | |
| 92 | + :fps fps}) | |
| 93 | + true)) | |
| 94 | + | |
| 95 | +(defn stop! | |
| 96 | + "Take the plane down and release everything it holds." | |
| 97 | + [] | |
| 98 | + (when-let [p @plane] | |
| 99 | + (try (h264/close! (:encoder p)) (catch Exception _ nil)) | |
| 100 | + (try (h264/close-decoder! (:decoder p)) (catch Exception _ nil)) | |
| 101 | + (reset! plane nil)) | |
| 102 | + nil) | |
| 103 | + | |
| 104 | +;; --- controls ---------------------------------------------------------------- | |
| 105 | + | |
| 106 | +(defn set-camera! | |
| 107 | + "Publishing on or off. Off stops the encoder being fed; it does not tear the | |
| 108 | + track down, because a subscriber that saw the track vanish and reappear | |
| 109 | + would have to rediscover it." | |
| 110 | + [on?] | |
| 111 | + (swap! plane #(when % (assoc % :camera? (boolean on?)))) | |
| 112 | + nil) | |
| 113 | + | |
| 114 | +(defn force-keyframe! | |
| 115 | + "Make the next published frame an IDR. | |
| 116 | + | |
| 117 | + What a subscriber joining mid-call needs: the second frame out of an | |
| 118 | + encoder is a P-frame, and a decoder handed one first has no SPS or PPS to | |
| 119 | + decode against and says so." | |
| 120 | + [] | |
| 121 | + (when-let [p @plane] (h264/force-keyframe! (:encoder p))) | |
| 122 | + nil) | |
| 123 | + | |
| 124 | +;; --- the outbound half ------------------------------------------------------- | |
| 125 | + | |
| 126 | +(defn- pump-out! | |
| 127 | + "Take one frame from the source, encode it, publish it." | |
| 128 | + [{:keys [source encoder producer camera? pts fps]}] | |
| 129 | + (when (and camera? source) | |
| 130 | + (when-let [[px _len] (source)] | |
| 131 | + (let [us (swap! pts + (quot 1000000 (max 1 fps)))] | |
| 132 | + (h264/encode! | |
| 133 | + encoder px us | |
| 134 | + (fn [p len _key?] | |
| 135 | + ;; A skipped frame is a decision, not a failure — openh264 | |
| 136 | + ;; answers zero length and there is simply nothing to send. | |
| 137 | + (when (pos? len) | |
| 138 | + (media/write-video-frame! producer p len us)))))))) | |
| 139 | + | |
| 140 | +;; --- the inbound half -------------------------------------------------------- | |
| 141 | + | |
| 142 | +(defn- settle | |
| 143 | + "Answer a settled future's value, or nil while it has not settled. | |
| 144 | + | |
| 145 | + Never blocks: an unsettled future is polled again and nil comes back, which | |
| 146 | + is what lets this be called from the loop thread as often as a timer fires." | |
| 147 | + [fut lift] | |
| 148 | + (when (and fut (uniffi/settled? fut)) | |
| 149 | + (if lift (uniffi/complete! fut lift) (uniffi/complete! fut)))) | |
| 150 | + | |
| 151 | +(defn- pump-in! | |
| 152 | + "Advance the subscribe, then take at most ONE frame and decode it. | |
| 153 | + | |
| 154 | + One, not all of them: the decoder answers a pointer into its own buffer and | |
| 155 | + the next decode overwrites it, so draining the queue here would hand out | |
| 156 | + three pointers to the same pixels. joltmoq had the same rule and stated it | |
| 157 | + the same way." | |
| 158 | + [p] | |
| 159 | + (let [p (if (and (:subscribe p) (nil? (:media p))) | |
| 160 | + (if-let [mc (settle (:subscribe p) nil)] | |
| 161 | + (assoc p :media mc :subscribe nil) | |
| 162 | + p) | |
| 163 | + p)] | |
| 164 | + (if-not (:media p) | |
| 165 | + p | |
| 166 | + (let [p (if (:pending p) p (assoc p :pending (media/next-frame! (:media p)))) | |
| 167 | + ;; The decode happens INSIDE the lift, while the RustBuffer the | |
| 168 | + ;; payload points into is still alive. Lifting a span out and | |
| 169 | + ;; decoding afterwards reads a buffer that has already been | |
| 170 | + ;; freed — and what comes back from that is not a fault but | |
| 171 | + ;; plausible rubbish, which is the worst kind. | |
| 172 | + ;; | |
| 173 | + ;; The RGBA pointer it answers is the DECODER's buffer, not the | |
| 174 | + ;; RustBuffer's, so it outlives this and is good until the next | |
| 175 | + ;; decode. That is the borrow `poll-frame!` hands on. | |
| 176 | + decoded (settle (:pending p) | |
| 177 | + #(media/lift-media-frame | |
| 178 | + % | |
| 179 | + (fn [ptr len] | |
| 180 | + (when (pos? len) | |
| 181 | + (h264/decode! | |
| 182 | + (:decoder p) ptr len | |
| 183 | + (fn [rgba w h] | |
| 184 | + (when-not (or (ffi/null? rgba) (zero? w)) | |
| 185 | + {:key "peer" :w w :h h :rgba rgba})))))))] | |
| 186 | + (if decoded | |
| 187 | + (assoc p :pending nil :frame (:payload decoded)) | |
| 188 | + (assoc p :frame nil)))))) | |
| 189 | + | |
| 190 | +;; --- the pump ---------------------------------------------------------------- | |
| 191 | + | |
| 192 | +(defn pump! | |
| 193 | + "Drive both halves once. Called from the same timer as `frq.av/pump!`." | |
| 194 | + [] | |
| 195 | + (when-let [p @plane] | |
| 196 | + (pump-out! p) | |
| 197 | + (reset! plane (pump-in! p))) | |
| 198 | + nil) | |
| 199 | + | |
| 200 | +(defn poll-frame! | |
| 201 | + "The frame decoded by the last `pump!`, or nil. | |
| 202 | + | |
| 203 | + {:key :w :h :rgba}, where `:rgba` is BORROWED — it is the decoder's own | |
| 204 | + buffer and the next `pump!` overwrites it. Hand it to `vidya/frame-rgba!` | |
| 205 | + and let it go; copying it is the one copy this whole path exists to avoid." | |
| 206 | + [] | |
| 207 | + (:frame @plane)) | |
| 208 | + | |
| 209 | +(defn poll-status! | |
| 210 | + "Drain what the plane has learned, as [code text] pairs, oldest first." | |
| 211 | + [] | |
| 212 | + (when-let [p @plane] | |
| 213 | + (let [q (:status p) | |
| 214 | + v @q] | |
| 215 | + (reset! q []) | |
| 216 | + v))) | |
| new file mode 100644 | |||
| @@ -0,0 +1,216 @@ | |||
| 1 | +(ns frq.av.plane | ||
| 2 | + "The media plane, in jolt — what `libjoltmoq` was. | ||
| 3 | + | ||
| 4 | + `joltmoq_start` was one call that connected, published a camera, subscribed | ||
| 5 | + to every peer, decoded their video and mixed their audio, on its own Rust | ||
| 6 | + threads. This is the same job assembled from the pieces `frq.moq.*`, | ||
| 7 | + `frq.codec.*` and `frq.capture.*` now provide, and it deliberately keeps | ||
| 8 | + joltmoq's SHAPE so that `frq.av` becomes a change of call sites rather than | ||
| 9 | + a rewrite: | ||
| 10 | + | ||
| 11 | + start! stop! live? joltmoq_start / _stop / _is_live | ||
| 12 | + poll-status! joltmoq_poll_status + _status_text | ||
| 13 | + poll-frame! joltmoq_frame_poll + _frame_rgba | ||
| 14 | + | ||
| 15 | + WHY IT IS PUMPED AND NOT THREADED. jolt has fibers, but a fiber is bound to | ||
| 16 | + its carrier for life and a blocking foreign call pins that carrier and | ||
| 17 | + strands everything queued behind it — and the two things this has to do | ||
| 18 | + most often, V4L2's DQBUF and ALSA's readi, are exactly that. So the plane is | ||
| 19 | + driven from `pump!`, which glimmer already calls from a timer for `frq.av`. | ||
| 20 | + It also keeps joltmoq's frame contract intact for free: at most one frame is | ||
| 21 | + decoded per poll, so the pointer handed out stays valid until the next one, | ||
| 22 | + which is precisely what `joltmoq_frame_rgba` promised. | ||
| 23 | + | ||
| 24 | + WHAT IS HERE. One peer, video only. Outbound is capture → H.264 → a MoQ | ||
| 25 | + media track; inbound is that track → H.264 → RGBA. That is enough to carry | ||
| 26 | + a picture end to end and it is deliberately the smallest thing that can be, | ||
| 27 | + because the parts that are NOT here are the ones worth doing carefully: | ||
| 28 | + | ||
| 29 | + * more than one peer, and the announce/catalog handling that finds them | ||
| 30 | + * audio at all — Opus and ALSA are bound, but mixing several peers into | ||
| 31 | + one playback stream needs a jitter buffer and a resampler for clock | ||
| 32 | + drift, and a bad one is worse than none | ||
| 33 | + * Android, where neither V4L2 nor ALSA exists | ||
| 34 | + | ||
| 35 | + A SOURCE IS A FUNCTION, not a camera. `start!` takes `:source`, a thunk | ||
| 36 | + answering [pointer length] for one I420 frame or nil for \"nothing right | ||
| 37 | + now\". `frq.capture.v4l2` is one such thunk; a test pattern is another. That | ||
| 38 | + is what lets the plane be exercised on a machine with no camera, and it is | ||
| 39 | + also how the phone will pass a Camera2 buffer in later without this | ||
| 40 | + namespace learning about JNI." | ||
| 41 | + (:require [frq.moq.media :as media] | ||
| 42 | + [frq.moq.uniffi :as uniffi] | ||
| 43 | + [frq.codec.h264 :as h264] | ||
| 44 | + [jolt.ffi :as ffi])) | ||
| 45 | + | ||
| 46 | +;; --- state ------------------------------------------------------------------- | ||
| 47 | +;; One plane per process, as joltmoq had: its C API was all globals, and the | ||
| 48 | +;; call surface above it assumes a single call at a time. | ||
| 49 | + | ||
| 50 | +(declare stop!) | ||
| 51 | + | ||
| 52 | +(defonce ^:private plane (atom nil)) | ||
| 53 | + | ||
| 54 | +(defn live? [] (some? @plane)) | ||
| 55 | + | ||
| 56 | +;; --- starting ---------------------------------------------------------------- | ||
| 57 | + | ||
| 58 | +(defn start! | ||
| 59 | + "Bring the plane up and answer true, or false with a reason recorded. | ||
| 60 | + | ||
| 61 | + `origin` is a MoqOriginProducer — from a session for a real call, or made | ||
| 62 | + locally for a test, which is the same object either way. `source` is the | ||
| 63 | + frame thunk described above. | ||
| 64 | + | ||
| 65 | + Everything that can fail does so HERE rather than at the first frame: the | ||
| 66 | + encoder validates its size, the decoder opens, and the subscribe settles, | ||
| 67 | + so a plane that comes up is one that can carry a picture." | ||
| 68 | + [{:keys [origin path source width height fps bitrate camera?] | ||
| 69 | + :or {path "/frq" width 640 height 480 fps 30 bitrate 800000 camera? true}}] | ||
| 70 | + (stop!) | ||
| 71 | + (let [broadcast (media/create-broadcast! origin path) | ||
| 72 | + producer (media/publish-media! broadcast "avc3") | ||
| 73 | + track (media/producer-name producer) | ||
| 74 | + consumer (media/broadcast-consumer broadcast)] | ||
| 75 | + (reset! plane | ||
| 76 | + {:broadcast broadcast | ||
| 77 | + :producer producer | ||
| 78 | + :track track | ||
| 79 | + :consumer consumer | ||
| 80 | + :subscribe (media/subscribe-media! consumer track media/video-container) | ||
| 81 | + :media nil | ||
| 82 | + :pending nil | ||
| 83 | + :encoder (h264/encoder {:width width :height height | ||
| 84 | + :fps fps :bitrate bitrate}) | ||
| 85 | + :decoder (h264/decoder) | ||
| 86 | + :source source | ||
| 87 | + :size [width height] | ||
| 88 | + :camera? camera? | ||
| 89 | + :frame nil | ||
| 90 | + :status (atom []) | ||
| 91 | + :pts (atom 0) | ||
| 92 | + :fps fps}) | ||
| 93 | + true)) | ||
| 94 | + | ||
| 95 | +(defn stop! | ||
| 96 | + "Take the plane down and release everything it holds." | ||
| 97 | + [] | ||
| 98 | + (when-let [p @plane] | ||
| 99 | + (try (h264/close! (:encoder p)) (catch Exception _ nil)) | ||
| 100 | + (try (h264/close-decoder! (:decoder p)) (catch Exception _ nil)) | ||
| 101 | + (reset! plane nil)) | ||
| 102 | + nil) | ||
| 103 | + | ||
| 104 | +;; --- controls ---------------------------------------------------------------- | ||
| 105 | + | ||
| 106 | +(defn set-camera! | ||
| 107 | + "Publishing on or off. Off stops the encoder being fed; it does not tear the | ||
| 108 | + track down, because a subscriber that saw the track vanish and reappear | ||
| 109 | + would have to rediscover it." | ||
| 110 | + [on?] | ||
| 111 | + (swap! plane #(when % (assoc % :camera? (boolean on?)))) | ||
| 112 | + nil) | ||
| 113 | + | ||
| 114 | +(defn force-keyframe! | ||
| 115 | + "Make the next published frame an IDR. | ||
| 116 | + | ||
| 117 | + What a subscriber joining mid-call needs: the second frame out of an | ||
| 118 | + encoder is a P-frame, and a decoder handed one first has no SPS or PPS to | ||
| 119 | + decode against and says so." | ||
| 120 | + [] | ||
| 121 | + (when-let [p @plane] (h264/force-keyframe! (:encoder p))) | ||
| 122 | + nil) | ||
| 123 | + | ||
| 124 | +;; --- the outbound half ------------------------------------------------------- | ||
| 125 | + | ||
| 126 | +(defn- pump-out! | ||
| 127 | + "Take one frame from the source, encode it, publish it." | ||
| 128 | + [{:keys [source encoder producer camera? pts fps]}] | ||
| 129 | + (when (and camera? source) | ||
| 130 | + (when-let [[px _len] (source)] | ||
| 131 | + (let [us (swap! pts + (quot 1000000 (max 1 fps)))] | ||
| 132 | + (h264/encode! | ||
| 133 | + encoder px us | ||
| 134 | + (fn [p len _key?] | ||
| 135 | + ;; A skipped frame is a decision, not a failure — openh264 | ||
| 136 | + ;; answers zero length and there is simply nothing to send. | ||
| 137 | + (when (pos? len) | ||
| 138 | + (media/write-video-frame! producer p len us)))))))) | ||
| 139 | + | ||
| 140 | +;; --- the inbound half -------------------------------------------------------- | ||
| 141 | + | ||
| 142 | +(defn- settle | ||
| 143 | + "Answer a settled future's value, or nil while it has not settled. | ||
| 144 | + | ||
| 145 | + Never blocks: an unsettled future is polled again and nil comes back, which | ||
| 146 | + is what lets this be called from the loop thread as often as a timer fires." | ||
| 147 | + [fut lift] | ||
| 148 | + (when (and fut (uniffi/settled? fut)) | ||
| 149 | + (if lift (uniffi/complete! fut lift) (uniffi/complete! fut)))) | ||
| 150 | + | ||
| 151 | +(defn- pump-in! | ||
| 152 | + "Advance the subscribe, then take at most ONE frame and decode it. | ||
| 153 | + | ||
| 154 | + One, not all of them: the decoder answers a pointer into its own buffer and | ||
| 155 | + the next decode overwrites it, so draining the queue here would hand out | ||
| 156 | + three pointers to the same pixels. joltmoq had the same rule and stated it | ||
| 157 | + the same way." | ||
| 158 | + [p] | ||
| 159 | + (let [p (if (and (:subscribe p) (nil? (:media p))) | ||
| 160 | + (if-let [mc (settle (:subscribe p) nil)] | ||
| 161 | + (assoc p :media mc :subscribe nil) | ||
| 162 | + p) | ||
| 163 | + p)] | ||
| 164 | + (if-not (:media p) | ||
| 165 | + p | ||
| 166 | + (let [p (if (:pending p) p (assoc p :pending (media/next-frame! (:media p)))) | ||
| 167 | + ;; The decode happens INSIDE the lift, while the RustBuffer the | ||
| 168 | + ;; payload points into is still alive. Lifting a span out and | ||
| 169 | + ;; decoding afterwards reads a buffer that has already been | ||
| 170 | + ;; freed — and what comes back from that is not a fault but | ||
| 171 | + ;; plausible rubbish, which is the worst kind. | ||
| 172 | + ;; | ||
| 173 | + ;; The RGBA pointer it answers is the DECODER's buffer, not the | ||
| 174 | + ;; RustBuffer's, so it outlives this and is good until the next | ||
| 175 | + ;; decode. That is the borrow `poll-frame!` hands on. | ||
| 176 | + decoded (settle (:pending p) | ||
| 177 | + #(media/lift-media-frame | ||
| 178 | + % | ||
| 179 | + (fn [ptr len] | ||
| 180 | + (when (pos? len) | ||
| 181 | + (h264/decode! | ||
| 182 | + (:decoder p) ptr len | ||
| 183 | + (fn [rgba w h] | ||
| 184 | + (when-not (or (ffi/null? rgba) (zero? w)) | ||
| 185 | + {:key "peer" :w w :h h :rgba rgba})))))))] | ||
| 186 | + (if decoded | ||
| 187 | + (assoc p :pending nil :frame (:payload decoded)) | ||
| 188 | + (assoc p :frame nil)))))) | ||
| 189 | + | ||
| 190 | +;; --- the pump ---------------------------------------------------------------- | ||
| 191 | + | ||
| 192 | +(defn pump! | ||
| 193 | + "Drive both halves once. Called from the same timer as `frq.av/pump!`." | ||
| 194 | + [] | ||
| 195 | + (when-let [p @plane] | ||
| 196 | + (pump-out! p) | ||
| 197 | + (reset! plane (pump-in! p))) | ||
| 198 | + nil) | ||
| 199 | + | ||
| 200 | +(defn poll-frame! | ||
| 201 | + "The frame decoded by the last `pump!`, or nil. | ||
| 202 | + | ||
| 203 | + {:key :w :h :rgba}, where `:rgba` is BORROWED — it is the decoder's own | ||
| 204 | + buffer and the next `pump!` overwrites it. Hand it to `vidya/frame-rgba!` | ||
| 205 | + and let it go; copying it is the one copy this whole path exists to avoid." | ||
| 206 | + [] | ||
| 207 | + (:frame @plane)) | ||
| 208 | + | ||
| 209 | +(defn poll-status! | ||
| 210 | + "Drain what the plane has learned, as [code text] pairs, oldest first." | ||
| 211 | + [] | ||
| 212 | + (when-let [p @plane] | ||
| 213 | + (let [q (:status p) | ||
| 214 | + v @q] | ||
| 215 | + (reset! q []) | ||
| 216 | + v))) | ||
modified
src/frq/moq/media.clj +15 -0 | @@ -253,6 +253,21 @@ | ||
| 253 | 253 | |
| 254 | 254 | (def video-container :legacy) |
| 255 | 255 | |
| 256 | +(defn write-video-frame! | |
| 257 | + "Write an ALREADY-ENCODED frame to a media producer, from foreign memory. | |
| 258 | + | |
| 259 | + The `write-frame!` above takes a jolt string, which is fine for a test | |
| 260 | + payload and wrong for H.264. This takes [pointer length] and copies the | |
| 261 | + bytes straight into the buffer, so an encoded frame goes from the | |
| 262 | + encoder's output to the wire without becoming a jolt value." | |
| 263 | + [producer ptr len timestamp-us] | |
| 264 | + (let [h (uniffi/with-out-status #(raw/clone-moqmediaproducer producer %))] | |
| 265 | + (lowered [[:bytes [ptr len]] [:u64 timestamp-us]] | |
| 266 | + (fn [buf] | |
| 267 | + (uniffi/with-out-status | |
| 268 | + #(raw/method-moqmediaproducer-write-frame h buf %))))) | |
| 269 | + nil) | |
| 270 | + | |
| 256 | 271 | (defn lift-media-frame |
| 257 | 272 | "Read an Optional<MoqMediaFrame>, handing the payload to `use-payload` as a |
| 258 | 273 | BORROWED [pointer length] span. |
| @@ -253,6 +253,21 @@ | |||
| 253 | 253 | ||
| 254 | (def video-container :legacy) | 254 | (def video-container :legacy) |
| 255 | 255 | ||
| 256 | +(defn write-video-frame! | ||
| 257 | + "Write an ALREADY-ENCODED frame to a media producer, from foreign memory. | ||
| 258 | + | ||
| 259 | + The `write-frame!` above takes a jolt string, which is fine for a test | ||
| 260 | + payload and wrong for H.264. This takes [pointer length] and copies the | ||
| 261 | + bytes straight into the buffer, so an encoded frame goes from the | ||
| 262 | + encoder's output to the wire without becoming a jolt value." | ||
| 263 | + [producer ptr len timestamp-us] | ||
| 264 | + (let [h (uniffi/with-out-status #(raw/clone-moqmediaproducer producer %))] | ||
| 265 | + (lowered [[:bytes [ptr len]] [:u64 timestamp-us]] | ||
| 266 | + (fn [buf] | ||
| 267 | + (uniffi/with-out-status | ||
| 268 | + #(raw/method-moqmediaproducer-write-frame h buf %))))) | ||
| 269 | + nil) | ||
| 270 | + | ||
| 256 | (defn lift-media-frame | 271 | (defn lift-media-frame |
| 257 | "Read an Optional<MoqMediaFrame>, handing the payload to `use-payload` as a | 272 | "Read an Optional<MoqMediaFrame>, handing the payload to `use-payload` as a |
| 258 | BORROWED [pointer length] span. | 273 | BORROWED [pointer length] span. |
modified
src/frq/moq/smoke.clj +58 -1 | @@ -40,6 +40,7 @@ | ||
| 40 | 40 | [frq.codec.h264 :as h264] |
| 41 | 41 | [frq.capture.v4l2 :as v4l2] |
| 42 | 42 | [frq.capture.alsa :as alsa] |
| 43 | + [frq.av.plane :as plane] | |
| 43 | 44 | [jolt.ffi :as ffi])) |
| 44 | 45 | |
| 45 | 46 | (defn- check-contract [] |
| @@ -428,6 +429,61 @@ | ||
| 428 | 429 | {:device bad})))) |
| 429 | 430 | true)) |
| 430 | 431 | |
| 432 | +(defn- check-plane | |
| 433 | + "A picture end to end through frq.av.plane: source -> H.264 -> MoQ -> RGBA. | |
| 434 | + | |
| 435 | + This is the whole point of the port in one test. The source is a thunk | |
| 436 | + answering a synthetic I420 frame, which is what a camera will be; the frame | |
| 437 | + goes out through the encoder onto a real MoQ media track, comes back in | |
| 438 | + through a real subscribe, and is decoded to RGBA ready for | |
| 439 | + vidya/frame-rgba!. Nothing is faked in between. | |
| 440 | + | |
| 441 | + The frame is not flat this time. A left half at luma 0x40 and a right half | |
| 442 | + at 0xC0 means a decode that produced a uniform picture — the failure a gray | |
| 443 | + test frame cannot distinguish from success — shows up as two equal halves. | |
| 444 | + | |
| 445 | + It is pumped rather than awaited, on purpose: pump! is what glimmer's timer | |
| 446 | + will call, so driving it in a loop here is the same code path the app takes, | |
| 447 | + including the several pumps a subscribe and a first keyframe take to | |
| 448 | + settle." | |
| 449 | + [] | |
| 450 | + (ffi/with-arena [a] | |
| 451 | + (let [w 64 h 64 | |
| 452 | + n (h264/i420-size w h) | |
| 453 | + px (ffi/alloc a n)] | |
| 454 | + ;; Left half dark, right half bright; chroma neutral. | |
| 455 | + (dotimes [y h] | |
| 456 | + (dotimes [x w] | |
| 457 | + (ffi/write (+ px (* y w) x) :uint8 (if (< x (quot w 2)) 0x40 0xC0)))) | |
| 458 | + (dotimes [i (* 2 (quot (* w h) 4))] | |
| 459 | + (ffi/write (+ px (* w h) i) :uint8 0x80)) | |
| 460 | + (let [origin (media/new-origin)] | |
| 461 | + (plane/start! {:origin origin :path "/plane" :source (fn [] [px n]) | |
| 462 | + :width w :height h :fps 30 :bitrate 200000}) | |
| 463 | + (try | |
| 464 | + (let [deadline (+ (System/currentTimeMillis) 20000)] | |
| 465 | + (loop [pumps 0] | |
| 466 | + (plane/pump!) | |
| 467 | + (if-let [f (plane/poll-frame!)] | |
| 468 | + (do | |
| 469 | + (println " frame after" pumps "pumps:" (:w f) "x" (:h f) (pr-str (:key f))) | |
| 470 | + (when-not (and (= w (:w f)) (= h (:h f))) | |
| 471 | + (throw (ex-info "decoded size is wrong" {:got [(:w f) (:h f)]}))) | |
| 472 | + (let [at (fn [x y] (ffi/read (+ (:rgba f) (* 4 (+ (* y (:w f)) x))) :uint8)) | |
| 473 | + left (at 8 32) | |
| 474 | + right (at 56 32)] | |
| 475 | + (println " left" left "right" right) | |
| 476 | + (when-not (< left right) | |
| 477 | + (throw (ex-info "the two halves came back the same — the picture did not survive" | |
| 478 | + {:left left :right right}))) | |
| 479 | + (when (< (- right left) 40) | |
| 480 | + (throw (ex-info "contrast collapsed" {:left left :right right})))) | |
| 481 | + true) | |
| 482 | + (if (> (System/currentTimeMillis) deadline) | |
| 483 | + (throw (ex-info "no frame came back through the plane" {:pumps pumps})) | |
| 484 | + (do (Thread/sleep 10) (recur (inc pumps))))))) | |
| 485 | + (finally (plane/stop!))))))) | |
| 486 | + | |
| 431 | 487 | (defn -main [& _] |
| 432 | 488 | (println "libmoq_ffi smoke test") |
| 433 | 489 | (let [steps [["contract" check-contract] |
| @@ -439,7 +495,8 @@ | ||
| 439 | 495 | ["h264" check-h264] |
| 440 | 496 | ["v4l2" check-v4l2-layouts] |
| 441 | 497 | ["alsa" check-alsa] |
| 442 | - ["devices" check-enumeration]]] | |
| 498 | + ["devices" check-enumeration] | |
| 499 | + ["plane" check-plane]]] | |
| 443 | 500 | (doseq [[name f] steps] |
| 444 | 501 | (println (str name ":")) |
| 445 | 502 | (f)) |
| @@ -40,6 +40,7 @@ | |||
| 40 | [frq.codec.h264 :as h264] | 40 | [frq.codec.h264 :as h264] |
| 41 | [frq.capture.v4l2 :as v4l2] | 41 | [frq.capture.v4l2 :as v4l2] |
| 42 | [frq.capture.alsa :as alsa] | 42 | [frq.capture.alsa :as alsa] |
| 43 | + [frq.av.plane :as plane] | ||
| 43 | [jolt.ffi :as ffi])) | 44 | [jolt.ffi :as ffi])) |
| 44 | 45 | ||
| 45 | (defn- check-contract [] | 46 | (defn- check-contract [] |
| @@ -428,6 +429,61 @@ | |||
| 428 | {:device bad})))) | 429 | {:device bad})))) |
| 429 | true)) | 430 | true)) |
| 430 | 431 | ||
| 432 | +(defn- check-plane | ||
| 433 | + "A picture end to end through frq.av.plane: source -> H.264 -> MoQ -> RGBA. | ||
| 434 | + | ||
| 435 | + This is the whole point of the port in one test. The source is a thunk | ||
| 436 | + answering a synthetic I420 frame, which is what a camera will be; the frame | ||
| 437 | + goes out through the encoder onto a real MoQ media track, comes back in | ||
| 438 | + through a real subscribe, and is decoded to RGBA ready for | ||
| 439 | + vidya/frame-rgba!. Nothing is faked in between. | ||
| 440 | + | ||
| 441 | + The frame is not flat this time. A left half at luma 0x40 and a right half | ||
| 442 | + at 0xC0 means a decode that produced a uniform picture — the failure a gray | ||
| 443 | + test frame cannot distinguish from success — shows up as two equal halves. | ||
| 444 | + | ||
| 445 | + It is pumped rather than awaited, on purpose: pump! is what glimmer's timer | ||
| 446 | + will call, so driving it in a loop here is the same code path the app takes, | ||
| 447 | + including the several pumps a subscribe and a first keyframe take to | ||
| 448 | + settle." | ||
| 449 | + [] | ||
| 450 | + (ffi/with-arena [a] | ||
| 451 | + (let [w 64 h 64 | ||
| 452 | + n (h264/i420-size w h) | ||
| 453 | + px (ffi/alloc a n)] | ||
| 454 | + ;; Left half dark, right half bright; chroma neutral. | ||
| 455 | + (dotimes [y h] | ||
| 456 | + (dotimes [x w] | ||
| 457 | + (ffi/write (+ px (* y w) x) :uint8 (if (< x (quot w 2)) 0x40 0xC0)))) | ||
| 458 | + (dotimes [i (* 2 (quot (* w h) 4))] | ||
| 459 | + (ffi/write (+ px (* w h) i) :uint8 0x80)) | ||
| 460 | + (let [origin (media/new-origin)] | ||
| 461 | + (plane/start! {:origin origin :path "/plane" :source (fn [] [px n]) | ||
| 462 | + :width w :height h :fps 30 :bitrate 200000}) | ||
| 463 | + (try | ||
| 464 | + (let [deadline (+ (System/currentTimeMillis) 20000)] | ||
| 465 | + (loop [pumps 0] | ||
| 466 | + (plane/pump!) | ||
| 467 | + (if-let [f (plane/poll-frame!)] | ||
| 468 | + (do | ||
| 469 | + (println " frame after" pumps "pumps:" (:w f) "x" (:h f) (pr-str (:key f))) | ||
| 470 | + (when-not (and (= w (:w f)) (= h (:h f))) | ||
| 471 | + (throw (ex-info "decoded size is wrong" {:got [(:w f) (:h f)]}))) | ||
| 472 | + (let [at (fn [x y] (ffi/read (+ (:rgba f) (* 4 (+ (* y (:w f)) x))) :uint8)) | ||
| 473 | + left (at 8 32) | ||
| 474 | + right (at 56 32)] | ||
| 475 | + (println " left" left "right" right) | ||
| 476 | + (when-not (< left right) | ||
| 477 | + (throw (ex-info "the two halves came back the same — the picture did not survive" | ||
| 478 | + {:left left :right right}))) | ||
| 479 | + (when (< (- right left) 40) | ||
| 480 | + (throw (ex-info "contrast collapsed" {:left left :right right})))) | ||
| 481 | + true) | ||
| 482 | + (if (> (System/currentTimeMillis) deadline) | ||
| 483 | + (throw (ex-info "no frame came back through the plane" {:pumps pumps})) | ||
| 484 | + (do (Thread/sleep 10) (recur (inc pumps))))))) | ||
| 485 | + (finally (plane/stop!))))))) | ||
| 486 | + | ||
| 431 | (defn -main [& _] | 487 | (defn -main [& _] |
| 432 | (println "libmoq_ffi smoke test") | 488 | (println "libmoq_ffi smoke test") |
| 433 | (let [steps [["contract" check-contract] | 489 | (let [steps [["contract" check-contract] |
| @@ -439,7 +495,8 @@ | |||
| 439 | ["h264" check-h264] | 495 | ["h264" check-h264] |
| 440 | ["v4l2" check-v4l2-layouts] | 496 | ["v4l2" check-v4l2-layouts] |
| 441 | ["alsa" check-alsa] | 497 | ["alsa" check-alsa] |
| 442 | - ["devices" check-enumeration]]] | 498 | + ["devices" check-enumeration] |
| 499 | + ["plane" check-plane]]] | ||
| 443 | (doseq [[name f] steps] | 500 | (doseq [[name f] steps] |
| 444 | (println (str name ":")) | 501 | (println (str name ":")) |
| 445 | (f)) | 502 | (f)) |