Find the other people in the call, and decode all of them
Multi-peer video. Nothing about the far side is configured any more: the plane watches the origin for ANNOUNCEMENTS, and each one turns up a peer's broadcast. Its catalog names the video track and says which container it is in, and from there it is subscribe, decode, RGBA, per peer. The catalog is worth the parsing it took. It carries the container per track, which is the LOC-versus-LEGACY question that answers `mux: loc: malformed loc properties` when guessed wrong -- so video-container stops being a constant this namespace knows and becomes a thing the peer says. Reading it meant maps, f64 and MoqVideo/MoqAudio/MoqDimensions in the buffer codec; every field is parsed even where unused, because they are positional and skipping one makes the next come out as nonsense. EVERY PEER KEEPS ITS OWN DECODER. That is what makes several possible at all: a decoder's output buffer is overwritten by its next decode, so one shared between peers would hand out the same pixels under every key. One decode per peer per pump, and each pointer stays good until that peer decodes again -- which is why poll-frames! can answer several at once where poll-frame! could only answer one. A peer advances at most one step per pump -- catalog subscribe, catalog read, track subscribe, frame -- so no peer arriving can hold the loop thread while it negotiates. The self-view is the same path as everyone else's, and finding it took a fix. An origin announces `us` for a broadcast created as `/us`: the leading slash is ours, and announcements come back relative to the origin root. Comparing them verbatim does not fail loudly -- it puts your own face in the grid under a peer's name. Normalised now, and keyed __local__ to match what frq.av already calls it. Two peers, two different pictures, checked apart rather than together: each frame has its halves the other way round, so a second peer confused for the first one counted twice would show up as identical contrast rather than passing. Still no audio, and still Android-less. frq.av is still untouched. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
8c28876 parent: 5db1a03 modified
src/frq/av/plane.clj +166 -60 | @@ -10,7 +10,7 @@ | ||
| 10 | 10 | |
| 11 | 11 | start! stop! live? joltmoq_start / _stop / _is_live |
| 12 | 12 | poll-status! joltmoq_poll_status + _status_text |
| 13 | - poll-frame! joltmoq_frame_poll + _frame_rgba | |
| 13 | + poll-frames! joltmoq_frame_poll + _frame_rgba | |
| 14 | 14 | |
| 15 | 15 | WHY IT IS PUMPED AND NOT THREADED. jolt has fibers, but a fiber is bound to |
| 16 | 16 | its carrier for life and a blocking foreign call pins that carrier and |
| @@ -21,15 +21,23 @@ | ||
| 21 | 21 | decoded per poll, so the pointer handed out stays valid until the next one, |
| 22 | 22 | which is precisely what `joltmoq_frame_rgba` promised. |
| 23 | 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: | |
| 24 | + WHAT IS HERE. Video, for as many peers as announce themselves. Outbound is | |
| 25 | + capture → H.264 → a MoQ media track. Inbound is DISCOVERED rather than | |
| 26 | + configured: an announcement watch on the origin turns up each peer's | |
| 27 | + broadcast, its catalog names the video track and says which container it is | |
| 28 | + in, and from there it is subscribe → H.264 → RGBA per peer. | |
| 28 | 29 | |
| 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 | |
| 30 | + Each peer keeps its OWN decoder, which is what makes several of them | |
| 31 | + possible at all: a decoder's output buffer is overwritten by its next | |
| 32 | + decode, so one shared between peers would hand out the same pixels for all | |
| 33 | + of them. One decode per peer per pump, and the pointers stay good until | |
| 34 | + that peer's next. | |
| 35 | + | |
| 36 | + What is still not here: | |
| 37 | + | |
| 38 | + * audio — Opus and ALSA are bound, but mixing several peers into one | |
| 39 | + playback stream needs a jitter buffer and a resampler for clock drift, | |
| 40 | + and a bad one is worse than none | |
| 33 | 41 | * Android, where neither V4L2 nor ALSA exists |
| 34 | 42 | |
| 35 | 43 | A SOURCE IS A FUNCTION, not a camera. `start!` takes `:source`, a thunk |
| @@ -38,7 +46,8 @@ | ||
| 38 | 46 | is what lets the plane be exercised on a machine with no camera, and it is |
| 39 | 47 | also how the phone will pass a Camera2 buffer in later without this |
| 40 | 48 | namespace learning about JNI." |
| 41 | - (:require [frq.moq.media :as media] | |
| 49 | + (:require [clojure.string :as str] | |
| 50 | + [frq.moq.media :as media] | |
| 42 | 51 | [frq.moq.uniffi :as uniffi] |
| 43 | 52 | [frq.codec.h264 :as h264] |
| 44 | 53 | [jolt.ffi :as ffi])) |
| @@ -76,17 +85,20 @@ | ||
| 76 | 85 | {:broadcast broadcast |
| 77 | 86 | :producer producer |
| 78 | 87 | :track track |
| 79 | - :consumer consumer | |
| 80 | - :subscribe (media/subscribe-media! consumer track media/video-container) | |
| 81 | - :media nil | |
| 82 | - :pending nil | |
| 88 | + :path path | |
| 89 | + ;; The announcement watch is the whole of peer discovery. An | |
| 90 | + ;; empty prefix takes everything on the origin, because in a | |
| 91 | + ;; call every participant is a broadcast and none of their | |
| 92 | + ;; paths are known in advance. | |
| 93 | + :announced (media/announced! (media/origin-consumer origin) "") | |
| 94 | + :announce nil | |
| 95 | + :peers {} | |
| 83 | 96 | :encoder (h264/encoder {:width width :height height |
| 84 | 97 | :fps fps :bitrate bitrate}) |
| 85 | - :decoder (h264/decoder) | |
| 86 | 98 | :source source |
| 87 | 99 | :size [width height] |
| 88 | 100 | :camera? camera? |
| 89 | - :frame nil | |
| 101 | + :frames [] | |
| 90 | 102 | :status (atom []) |
| 91 | 103 | :pts (atom 0) |
| 92 | 104 | :fps fps}) |
| @@ -97,7 +109,8 @@ | ||
| 97 | 109 | [] |
| 98 | 110 | (when-let [p @plane] |
| 99 | 111 | (try (h264/close! (:encoder p)) (catch Exception _ nil)) |
| 100 | - (try (h264/close-decoder! (:decoder p)) (catch Exception _ nil)) | |
| 112 | + (doseq [[_ peer] (:peers p)] | |
| 113 | + (try (h264/close-decoder! (:decoder peer)) (catch Exception _ nil))) | |
| 101 | 114 | (reset! plane nil)) |
| 102 | 115 | nil) |
| 103 | 116 | |
| @@ -137,7 +150,7 @@ | ||
| 137 | 150 | (when (pos? len) |
| 138 | 151 | (media/write-video-frame! producer p len us)))))))) |
| 139 | 152 | |
| 140 | -;; --- the inbound half -------------------------------------------------------- | |
| 153 | +;; --- the inbound half ------------------------------------------------------- | |
| 141 | 154 | |
| 142 | 155 | (defn- settle |
| 143 | 156 | "Answer a settled future's value, or nil while it has not settled. |
| @@ -148,44 +161,126 @@ | ||
| 148 | 161 | (when (and fut (uniffi/settled? fut)) |
| 149 | 162 | (if lift (uniffi/complete! fut lift) (uniffi/complete! fut)))) |
| 150 | 163 | |
| 151 | -(defn- pump-in! | |
| 152 | - "Advance the subscribe, then take at most ONE frame and decode it. | |
| 164 | +(defn- normalise-path | |
| 165 | + "An origin announces `us` for a broadcast created as `/us`. | |
| 166 | + | |
| 167 | + The leading slash is ours, not the origin's: `create-broadcast!` takes the | |
| 168 | + path we hand it and announcements come back relative to the origin root. | |
| 169 | + Comparing the two verbatim is how the self-view stops being recognised as | |
| 170 | + ours — which does not fail loudly, it just puts your own face in the grid | |
| 171 | + under a peer's name." | |
| 172 | + [path] | |
| 173 | + (when path (str/replace path #"^/+" ""))) | |
| 174 | + | |
| 175 | +(defn- pump-announce! | |
| 176 | + "Advance the announcement watch; add a peer for anything new. | |
| 177 | + | |
| 178 | + Our own broadcast is announced back to us like anyone else's, and it is | |
| 179 | + taken as the self-view rather than filtered out — `frq.av` already has a | |
| 180 | + key for that and a self-view is a picture the person expects to see." | |
| 181 | + [p] | |
| 182 | + (let [p (if (:announce p) p (assoc p :announce (media/next-announcement! (:announced p))))] | |
| 183 | + (if-let [ann (settle (:announce p) media/lift-announcement)] | |
| 184 | + (let [path (media/announcement-path ann) | |
| 185 | + p (assoc p :announce nil)] | |
| 186 | + (if (contains? (:peers p) path) | |
| 187 | + p | |
| 188 | + (assoc-in p [:peers path] | |
| 189 | + {:broadcast (media/announcement-broadcast ann) | |
| 190 | + :catalog nil | |
| 191 | + :catalog-pending nil | |
| 192 | + :subscribe nil | |
| 193 | + :media nil | |
| 194 | + :pending nil | |
| 195 | + :decoder (h264/decoder) | |
| 196 | + :self? (= (normalise-path path) | |
| 197 | + (normalise-path (:path p)))}))) | |
| 198 | + p))) | |
| 199 | + | |
| 200 | +(defn- pump-peer! | |
| 201 | + "Walk one peer from announced to a decoded picture. | |
| 153 | 202 | |
| 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." | |
| 203 | + Four states, advanced at most one step per pump so that no peer can hold | |
| 204 | + the loop thread: subscribe the catalog, read it for a video track name and | |
| 205 | + its container, subscribe that track, then decode a frame from it." | |
| 206 | + [peer] | |
| 207 | + (cond | |
| 208 | + ;; 1. The catalog: what tracks this peer has, and in which container. | |
| 209 | + (nil? (:catalog peer)) | |
| 210 | + (let [peer (if (:catalog-pending peer) | |
| 211 | + peer | |
| 212 | + (assoc peer :catalog-pending | |
| 213 | + {:sub (media/subscribe-catalog! (:broadcast peer))})) | |
| 214 | + sub (:catalog-pending peer)] | |
| 215 | + (cond | |
| 216 | + (:consumer sub) | |
| 217 | + (let [cp (or (:next sub) (media/next-catalog! (:consumer sub)))] | |
| 218 | + (if-let [cat (settle cp media/lift-catalog)] | |
| 219 | + (if-let [[track video] (first (:video cat))] | |
| 220 | + (assoc peer :catalog {:track track :container (:container video)} | |
| 221 | + :catalog-pending nil) | |
| 222 | + ;; A catalog with no video is a peer who is not sending a | |
| 223 | + ;; picture — audio only, or not yet publishing. Wait rather | |
| 224 | + ;; than treat it as an error. | |
| 225 | + (assoc peer :catalog-pending {:consumer (:consumer sub) :next nil})) | |
| 226 | + (assoc peer :catalog-pending {:consumer (:consumer sub) :next cp}))) | |
| 227 | + | |
| 228 | + :else | |
| 229 | + (if-let [cc (settle (:sub sub) nil)] | |
| 230 | + (assoc peer :catalog-pending {:consumer cc :next nil}) | |
| 231 | + peer))) | |
| 232 | + | |
| 233 | + ;; 2. Subscribe to the track the catalog named. | |
| 234 | + (nil? (:media peer)) | |
| 235 | + (let [peer (if (:subscribe peer) | |
| 236 | + peer | |
| 237 | + (assoc peer :subscribe | |
| 238 | + (media/subscribe-media! (:broadcast peer) | |
| 239 | + (get-in peer [:catalog :track]) | |
| 240 | + (get-in peer [:catalog :container]))))] | |
| 241 | + (if-let [mc (settle (:subscribe peer) nil)] | |
| 242 | + (assoc peer :media mc :subscribe nil) | |
| 243 | + peer)) | |
| 244 | + | |
| 245 | + ;; 3. A frame. | |
| 246 | + :else | |
| 247 | + (let [peer (if (:pending peer) peer (assoc peer :pending (media/next-frame! (:media peer)))) | |
| 248 | + ;; The decode happens INSIDE the lift, while the RustBuffer the | |
| 249 | + ;; payload points into is still alive. Lifting the span out and | |
| 250 | + ;; decoding afterwards reads a buffer that has already been freed, | |
| 251 | + ;; and what comes back from that is not a fault but plausible | |
| 252 | + ;; rubbish. | |
| 253 | + ;; | |
| 254 | + ;; The RGBA pointer it answers belongs to this peer's DECODER, not | |
| 255 | + ;; to the RustBuffer, so it outlives the lift and is good until | |
| 256 | + ;; this peer decodes again. | |
| 257 | + decoded (settle (:pending peer) | |
| 258 | + #(media/lift-media-frame | |
| 259 | + % | |
| 260 | + (fn [ptr len] | |
| 261 | + (when (pos? len) | |
| 262 | + (h264/decode! | |
| 263 | + (:decoder peer) ptr len | |
| 264 | + (fn [rgba w h] | |
| 265 | + (when-not (or (ffi/null? rgba) (zero? w)) | |
| 266 | + {:w w :h h :rgba rgba})))))))] | |
| 267 | + (if decoded | |
| 268 | + (assoc peer :pending nil :frame (:payload decoded)) | |
| 269 | + (assoc peer :frame nil))))) | |
| 270 | + | |
| 271 | +(defn- pump-in! | |
| 272 | + "Discover peers, then advance every one of them." | |
| 158 | 273 | [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)))))) | |
| 274 | + (let [p (pump-announce! p) | |
| 275 | + peers (reduce-kv (fn [m path peer] (assoc m path (pump-peer! peer))) | |
| 276 | + {} (:peers p))] | |
| 277 | + (assoc p | |
| 278 | + :peers peers | |
| 279 | + :frames (into [] | |
| 280 | + (keep (fn [[path peer]] | |
| 281 | + (when-let [f (:frame peer)] | |
| 282 | + (assoc f :key (if (:self? peer) "__local__" path))))) | |
| 283 | + peers)))) | |
| 189 | 284 | |
| 190 | 285 | ;; --- the pump ---------------------------------------------------------------- |
| 191 | 286 | |
| @@ -197,14 +292,25 @@ | ||
| 197 | 292 | (reset! plane (pump-in! p))) |
| 198 | 293 | nil) |
| 199 | 294 | |
| 200 | -(defn poll-frame! | |
| 201 | - "The frame decoded by the last `pump!`, or nil. | |
| 295 | +(defn poll-frames! | |
| 296 | + "Every frame decoded by the last `pump!`, one per peer at most. | |
| 297 | + | |
| 298 | + Each is {:key :w :h :rgba} with `:key` the peer's broadcast path, or | |
| 299 | + \"__local__\" for our own. `:rgba` is BORROWED — it is that peer's decoder | |
| 300 | + buffer and that peer's next decode overwrites it — so hand each to | |
| 301 | + `vidya/frame-rgba!` and let it go. Copying is the one copy this whole path | |
| 302 | + exists to avoid. | |
| 303 | + | |
| 304 | + Several frames at once is safe precisely because the decoders are | |
| 305 | + per-peer: one shared decoder would make every pointer here alias the last | |
| 306 | + picture decoded." | |
| 307 | + [] | |
| 308 | + (:frames @plane)) | |
| 202 | 309 | |
| 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." | |
| 310 | +(defn peers | |
| 311 | + "The broadcast paths currently known, self included." | |
| 206 | 312 | [] |
| 207 | - (:frame @plane)) | |
| 313 | + (some-> @plane :peers keys vec)) | |
| 208 | 314 | |
| 209 | 315 | (defn poll-status! |
| 210 | 316 | "Drain what the plane has learned, as [code text] pairs, oldest first." |
| @@ -10,7 +10,7 @@ | |||
| 10 | 10 | ||
| 11 | start! stop! live? joltmoq_start / _stop / _is_live | 11 | start! stop! live? joltmoq_start / _stop / _is_live |
| 12 | poll-status! joltmoq_poll_status + _status_text | 12 | poll-status! joltmoq_poll_status + _status_text |
| 13 | - poll-frame! joltmoq_frame_poll + _frame_rgba | 13 | + poll-frames! joltmoq_frame_poll + _frame_rgba |
| 14 | 14 | ||
| 15 | WHY IT IS PUMPED AND NOT THREADED. jolt has fibers, but a fiber is bound to | 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 | 16 | its carrier for life and a blocking foreign call pins that carrier and |
| @@ -21,15 +21,23 @@ | |||
| 21 | decoded per poll, so the pointer handed out stays valid until the next one, | 21 | decoded per poll, so the pointer handed out stays valid until the next one, |
| 22 | which is precisely what `joltmoq_frame_rgba` promised. | 22 | which is precisely what `joltmoq_frame_rgba` promised. |
| 23 | 23 | ||
| 24 | - WHAT IS HERE. One peer, video only. Outbound is capture → H.264 → a MoQ | 24 | + WHAT IS HERE. Video, for as many peers as announce themselves. Outbound is |
| 25 | - media track; inbound is that track → H.264 → RGBA. That is enough to carry | 25 | + capture → H.264 → a MoQ media track. Inbound is DISCOVERED rather than |
| 26 | - a picture end to end and it is deliberately the smallest thing that can be, | 26 | + configured: an announcement watch on the origin turns up each peer's |
| 27 | - because the parts that are NOT here are the ones worth doing carefully: | 27 | + broadcast, its catalog names the video track and says which container it is |
| 28 | + in, and from there it is subscribe → H.264 → RGBA per peer. | ||
| 28 | 29 | ||
| 29 | - * more than one peer, and the announce/catalog handling that finds them | 30 | + Each peer keeps its OWN decoder, which is what makes several of them |
| 30 | - * audio at all — Opus and ALSA are bound, but mixing several peers into | 31 | + possible at all: a decoder's output buffer is overwritten by its next |
| 31 | - one playback stream needs a jitter buffer and a resampler for clock | 32 | + decode, so one shared between peers would hand out the same pixels for all |
| 32 | - drift, and a bad one is worse than none | 33 | + of them. One decode per peer per pump, and the pointers stay good until |
| 34 | + that peer's next. | ||
| 35 | + | ||
| 36 | + What is still not here: | ||
| 37 | + | ||
| 38 | + * audio — Opus and ALSA are bound, but mixing several peers into one | ||
| 39 | + playback stream needs a jitter buffer and a resampler for clock drift, | ||
| 40 | + and a bad one is worse than none | ||
| 33 | * Android, where neither V4L2 nor ALSA exists | 41 | * Android, where neither V4L2 nor ALSA exists |
| 34 | 42 | ||
| 35 | A SOURCE IS A FUNCTION, not a camera. `start!` takes `:source`, a thunk | 43 | A SOURCE IS A FUNCTION, not a camera. `start!` takes `:source`, a thunk |
| @@ -38,7 +46,8 @@ | |||
| 38 | is what lets the plane be exercised on a machine with no camera, and it is | 46 | 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 | 47 | also how the phone will pass a Camera2 buffer in later without this |
| 40 | namespace learning about JNI." | 48 | namespace learning about JNI." |
| 41 | - (:require [frq.moq.media :as media] | 49 | + (:require [clojure.string :as str] |
| 50 | + [frq.moq.media :as media] | ||
| 42 | [frq.moq.uniffi :as uniffi] | 51 | [frq.moq.uniffi :as uniffi] |
| 43 | [frq.codec.h264 :as h264] | 52 | [frq.codec.h264 :as h264] |
| 44 | [jolt.ffi :as ffi])) | 53 | [jolt.ffi :as ffi])) |
| @@ -76,17 +85,20 @@ | |||
| 76 | {:broadcast broadcast | 85 | {:broadcast broadcast |
| 77 | :producer producer | 86 | :producer producer |
| 78 | :track track | 87 | :track track |
| 79 | - :consumer consumer | 88 | + :path path |
| 80 | - :subscribe (media/subscribe-media! consumer track media/video-container) | 89 | + ;; The announcement watch is the whole of peer discovery. An |
| 81 | - :media nil | 90 | + ;; empty prefix takes everything on the origin, because in a |
| 82 | - :pending nil | 91 | + ;; call every participant is a broadcast and none of their |
| 92 | + ;; paths are known in advance. | ||
| 93 | + :announced (media/announced! (media/origin-consumer origin) "") | ||
| 94 | + :announce nil | ||
| 95 | + :peers {} | ||
| 83 | :encoder (h264/encoder {:width width :height height | 96 | :encoder (h264/encoder {:width width :height height |
| 84 | :fps fps :bitrate bitrate}) | 97 | :fps fps :bitrate bitrate}) |
| 85 | - :decoder (h264/decoder) | ||
| 86 | :source source | 98 | :source source |
| 87 | :size [width height] | 99 | :size [width height] |
| 88 | :camera? camera? | 100 | :camera? camera? |
| 89 | - :frame nil | 101 | + :frames [] |
| 90 | :status (atom []) | 102 | :status (atom []) |
| 91 | :pts (atom 0) | 103 | :pts (atom 0) |
| 92 | :fps fps}) | 104 | :fps fps}) |
| @@ -97,7 +109,8 @@ | |||
| 97 | [] | 109 | [] |
| 98 | (when-let [p @plane] | 110 | (when-let [p @plane] |
| 99 | (try (h264/close! (:encoder p)) (catch Exception _ nil)) | 111 | (try (h264/close! (:encoder p)) (catch Exception _ nil)) |
| 100 | - (try (h264/close-decoder! (:decoder p)) (catch Exception _ nil)) | 112 | + (doseq [[_ peer] (:peers p)] |
| 113 | + (try (h264/close-decoder! (:decoder peer)) (catch Exception _ nil))) | ||
| 101 | (reset! plane nil)) | 114 | (reset! plane nil)) |
| 102 | nil) | 115 | nil) |
| 103 | 116 | ||
| @@ -137,7 +150,7 @@ | |||
| 137 | (when (pos? len) | 150 | (when (pos? len) |
| 138 | (media/write-video-frame! producer p len us)))))))) | 151 | (media/write-video-frame! producer p len us)))))))) |
| 139 | 152 | ||
| 140 | -;; --- the inbound half -------------------------------------------------------- | 153 | +;; --- the inbound half ------------------------------------------------------- |
| 141 | 154 | ||
| 142 | (defn- settle | 155 | (defn- settle |
| 143 | "Answer a settled future's value, or nil while it has not settled. | 156 | "Answer a settled future's value, or nil while it has not settled. |
| @@ -148,44 +161,126 @@ | |||
| 148 | (when (and fut (uniffi/settled? fut)) | 161 | (when (and fut (uniffi/settled? fut)) |
| 149 | (if lift (uniffi/complete! fut lift) (uniffi/complete! fut)))) | 162 | (if lift (uniffi/complete! fut lift) (uniffi/complete! fut)))) |
| 150 | 163 | ||
| 151 | -(defn- pump-in! | 164 | +(defn- normalise-path |
| 152 | - "Advance the subscribe, then take at most ONE frame and decode it. | 165 | + "An origin announces `us` for a broadcast created as `/us`. |
| 166 | + | ||
| 167 | + The leading slash is ours, not the origin's: `create-broadcast!` takes the | ||
| 168 | + path we hand it and announcements come back relative to the origin root. | ||
| 169 | + Comparing the two verbatim is how the self-view stops being recognised as | ||
| 170 | + ours — which does not fail loudly, it just puts your own face in the grid | ||
| 171 | + under a peer's name." | ||
| 172 | + [path] | ||
| 173 | + (when path (str/replace path #"^/+" ""))) | ||
| 174 | + | ||
| 175 | +(defn- pump-announce! | ||
| 176 | + "Advance the announcement watch; add a peer for anything new. | ||
| 177 | + | ||
| 178 | + Our own broadcast is announced back to us like anyone else's, and it is | ||
| 179 | + taken as the self-view rather than filtered out — `frq.av` already has a | ||
| 180 | + key for that and a self-view is a picture the person expects to see." | ||
| 181 | + [p] | ||
| 182 | + (let [p (if (:announce p) p (assoc p :announce (media/next-announcement! (:announced p))))] | ||
| 183 | + (if-let [ann (settle (:announce p) media/lift-announcement)] | ||
| 184 | + (let [path (media/announcement-path ann) | ||
| 185 | + p (assoc p :announce nil)] | ||
| 186 | + (if (contains? (:peers p) path) | ||
| 187 | + p | ||
| 188 | + (assoc-in p [:peers path] | ||
| 189 | + {:broadcast (media/announcement-broadcast ann) | ||
| 190 | + :catalog nil | ||
| 191 | + :catalog-pending nil | ||
| 192 | + :subscribe nil | ||
| 193 | + :media nil | ||
| 194 | + :pending nil | ||
| 195 | + :decoder (h264/decoder) | ||
| 196 | + :self? (= (normalise-path path) | ||
| 197 | + (normalise-path (:path p)))}))) | ||
| 198 | + p))) | ||
| 199 | + | ||
| 200 | +(defn- pump-peer! | ||
| 201 | + "Walk one peer from announced to a decoded picture. | ||
| 153 | 202 | ||
| 154 | - One, not all of them: the decoder answers a pointer into its own buffer and | 203 | + Four states, advanced at most one step per pump so that no peer can hold |
| 155 | - the next decode overwrites it, so draining the queue here would hand out | 204 | + the loop thread: subscribe the catalog, read it for a video track name and |
| 156 | - three pointers to the same pixels. joltmoq had the same rule and stated it | 205 | + its container, subscribe that track, then decode a frame from it." |
| 157 | - the same way." | 206 | + [peer] |
| 207 | + (cond | ||
| 208 | + ;; 1. The catalog: what tracks this peer has, and in which container. | ||
| 209 | + (nil? (:catalog peer)) | ||
| 210 | + (let [peer (if (:catalog-pending peer) | ||
| 211 | + peer | ||
| 212 | + (assoc peer :catalog-pending | ||
| 213 | + {:sub (media/subscribe-catalog! (:broadcast peer))})) | ||
| 214 | + sub (:catalog-pending peer)] | ||
| 215 | + (cond | ||
| 216 | + (:consumer sub) | ||
| 217 | + (let [cp (or (:next sub) (media/next-catalog! (:consumer sub)))] | ||
| 218 | + (if-let [cat (settle cp media/lift-catalog)] | ||
| 219 | + (if-let [[track video] (first (:video cat))] | ||
| 220 | + (assoc peer :catalog {:track track :container (:container video)} | ||
| 221 | + :catalog-pending nil) | ||
| 222 | + ;; A catalog with no video is a peer who is not sending a | ||
| 223 | + ;; picture — audio only, or not yet publishing. Wait rather | ||
| 224 | + ;; than treat it as an error. | ||
| 225 | + (assoc peer :catalog-pending {:consumer (:consumer sub) :next nil})) | ||
| 226 | + (assoc peer :catalog-pending {:consumer (:consumer sub) :next cp}))) | ||
| 227 | + | ||
| 228 | + :else | ||
| 229 | + (if-let [cc (settle (:sub sub) nil)] | ||
| 230 | + (assoc peer :catalog-pending {:consumer cc :next nil}) | ||
| 231 | + peer))) | ||
| 232 | + | ||
| 233 | + ;; 2. Subscribe to the track the catalog named. | ||
| 234 | + (nil? (:media peer)) | ||
| 235 | + (let [peer (if (:subscribe peer) | ||
| 236 | + peer | ||
| 237 | + (assoc peer :subscribe | ||
| 238 | + (media/subscribe-media! (:broadcast peer) | ||
| 239 | + (get-in peer [:catalog :track]) | ||
| 240 | + (get-in peer [:catalog :container]))))] | ||
| 241 | + (if-let [mc (settle (:subscribe peer) nil)] | ||
| 242 | + (assoc peer :media mc :subscribe nil) | ||
| 243 | + peer)) | ||
| 244 | + | ||
| 245 | + ;; 3. A frame. | ||
| 246 | + :else | ||
| 247 | + (let [peer (if (:pending peer) peer (assoc peer :pending (media/next-frame! (:media peer)))) | ||
| 248 | + ;; The decode happens INSIDE the lift, while the RustBuffer the | ||
| 249 | + ;; payload points into is still alive. Lifting the span out and | ||
| 250 | + ;; decoding afterwards reads a buffer that has already been freed, | ||
| 251 | + ;; and what comes back from that is not a fault but plausible | ||
| 252 | + ;; rubbish. | ||
| 253 | + ;; | ||
| 254 | + ;; The RGBA pointer it answers belongs to this peer's DECODER, not | ||
| 255 | + ;; to the RustBuffer, so it outlives the lift and is good until | ||
| 256 | + ;; this peer decodes again. | ||
| 257 | + decoded (settle (:pending peer) | ||
| 258 | + #(media/lift-media-frame | ||
| 259 | + % | ||
| 260 | + (fn [ptr len] | ||
| 261 | + (when (pos? len) | ||
| 262 | + (h264/decode! | ||
| 263 | + (:decoder peer) ptr len | ||
| 264 | + (fn [rgba w h] | ||
| 265 | + (when-not (or (ffi/null? rgba) (zero? w)) | ||
| 266 | + {:w w :h h :rgba rgba})))))))] | ||
| 267 | + (if decoded | ||
| 268 | + (assoc peer :pending nil :frame (:payload decoded)) | ||
| 269 | + (assoc peer :frame nil))))) | ||
| 270 | + | ||
| 271 | +(defn- pump-in! | ||
| 272 | + "Discover peers, then advance every one of them." | ||
| 158 | [p] | 273 | [p] |
| 159 | - (let [p (if (and (:subscribe p) (nil? (:media p))) | 274 | + (let [p (pump-announce! p) |
| 160 | - (if-let [mc (settle (:subscribe p) nil)] | 275 | + peers (reduce-kv (fn [m path peer] (assoc m path (pump-peer! peer))) |
| 161 | - (assoc p :media mc :subscribe nil) | 276 | + {} (:peers p))] |
| 162 | - p) | 277 | + (assoc p |
| 163 | - p)] | 278 | + :peers peers |
| 164 | - (if-not (:media p) | 279 | + :frames (into [] |
| 165 | - p | 280 | + (keep (fn [[path peer]] |
| 166 | - (let [p (if (:pending p) p (assoc p :pending (media/next-frame! (:media p)))) | 281 | + (when-let [f (:frame peer)] |
| 167 | - ;; The decode happens INSIDE the lift, while the RustBuffer the | 282 | + (assoc f :key (if (:self? peer) "__local__" path))))) |
| 168 | - ;; payload points into is still alive. Lifting a span out and | 283 | + peers)))) |
| 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 | 284 | ||
| 190 | ;; --- the pump ---------------------------------------------------------------- | 285 | ;; --- the pump ---------------------------------------------------------------- |
| 191 | 286 | ||
| @@ -197,14 +292,25 @@ | |||
| 197 | (reset! plane (pump-in! p))) | 292 | (reset! plane (pump-in! p))) |
| 198 | nil) | 293 | nil) |
| 199 | 294 | ||
| 200 | -(defn poll-frame! | 295 | +(defn poll-frames! |
| 201 | - "The frame decoded by the last `pump!`, or nil. | 296 | + "Every frame decoded by the last `pump!`, one per peer at most. |
| 297 | + | ||
| 298 | + Each is {:key :w :h :rgba} with `:key` the peer's broadcast path, or | ||
| 299 | + \"__local__\" for our own. `:rgba` is BORROWED — it is that peer's decoder | ||
| 300 | + buffer and that peer's next decode overwrites it — so hand each to | ||
| 301 | + `vidya/frame-rgba!` and let it go. Copying is the one copy this whole path | ||
| 302 | + exists to avoid. | ||
| 303 | + | ||
| 304 | + Several frames at once is safe precisely because the decoders are | ||
| 305 | + per-peer: one shared decoder would make every pointer here alias the last | ||
| 306 | + picture decoded." | ||
| 307 | + [] | ||
| 308 | + (:frames @plane)) | ||
| 202 | 309 | ||
| 203 | - {:key :w :h :rgba}, where `:rgba` is BORROWED — it is the decoder's own | 310 | +(defn peers |
| 204 | - buffer and the next `pump!` overwrites it. Hand it to `vidya/frame-rgba!` | 311 | + "The broadcast paths currently known, self included." |
| 205 | - and let it go; copying it is the one copy this whole path exists to avoid." | ||
| 206 | [] | 312 | [] |
| 207 | - (:frame @plane)) | 313 | + (some-> @plane :peers keys vec)) |
| 208 | 314 | ||
| 209 | (defn poll-status! | 315 | (defn poll-status! |
| 210 | "Drain what the plane has learned, as [code text] pairs, oldest first." | 316 | "Drain what the plane has learned, as [code text] pairs, oldest first." |
modified
src/frq/moq/media.clj +139 -0 | @@ -76,6 +76,145 @@ | ||
| 76 | 76 | (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))] |
| 77 | 77 | (uniffi/with-out-status #(raw/method-moqbroadcastproducer-consume h %)))) |
| 78 | 78 | |
| 79 | +;; --- discovery -------------------------------------------------------------- | |
| 80 | +;; How a peer is FOUND, rather than known. `subscribe-media!` needs a path and | |
| 81 | +;; a track name; announcements are where those come from when the other side | |
| 82 | +;; of the call is somebody else's client rather than a broadcast we made. | |
| 83 | + | |
| 84 | +(defn origin-consumer | |
| 85 | + "The subscriber's side of an origin." | |
| 86 | + [origin] | |
| 87 | + (let [h (uniffi/with-out-status #(raw/clone-moqoriginproducer origin %))] | |
| 88 | + (uniffi/with-out-status #(raw/method-moqoriginproducer-consume h %)))) | |
| 89 | + | |
| 90 | +(defn announced! | |
| 91 | + "Watch for broadcasts whose path starts with `prefix`; answers a MoqAnnounced. | |
| 92 | + | |
| 93 | + An empty prefix watches everything on the origin, which in a call is what | |
| 94 | + you want: every participant is a broadcast and none of their paths are | |
| 95 | + known in advance." | |
| 96 | + [origin-consumer prefix] | |
| 97 | + (let [h (uniffi/with-out-status #(raw/clone-moqoriginconsumer origin-consumer %))] | |
| 98 | + (ffi/with-arena [a] | |
| 99 | + (let [buf (uniffi/lower-string | |
| 100 | + (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) prefix)] | |
| 101 | + (uniffi/with-out-status | |
| 102 | + #(raw/method-moqoriginconsumer-announced h buf %)))))) | |
| 103 | + | |
| 104 | +(defn next-announcement! | |
| 105 | + "Ask for the next announcement; answers an :rb future. | |
| 106 | + | |
| 107 | + It settles to an Optional<MoqAnnouncement> — absent when the origin has | |
| 108 | + closed, which is the end of the watch rather than an error." | |
| 109 | + [announced] | |
| 110 | + (let [h (uniffi/with-out-status #(raw/clone-moqannounced announced %))] | |
| 111 | + (-> (raw/method-moqannounced-next h) | |
| 112 | + (uniffi/start-future :rb)))) | |
| 113 | + | |
| 114 | +(defn lift-announcement | |
| 115 | + "Read an Optional<MoqAnnouncement> from a settled :rb buffer. | |
| 116 | + | |
| 117 | + Answers the announcement's HANDLE, which outlives the buffer — an interface | |
| 118 | + crosses as a u64 the far side has already cloned for us, so freeing the | |
| 119 | + buffer it arrived in does not touch it." | |
| 120 | + [rb-ptr] | |
| 121 | + (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len]) | |
| 122 | + data (ffi/read-field rb-ptr uniffi/rust-buffer [:data]) | |
| 123 | + v (when (and (pos? len) (not (ffi/null? data))) | |
| 124 | + (let [c (uniffi/reader data len)] | |
| 125 | + (uniffi/r-optional! c uniffi/r-u64!)))] | |
| 126 | + (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %)) | |
| 127 | + v)) | |
| 128 | + | |
| 129 | +(defn announcement-path | |
| 130 | + [ann] | |
| 131 | + (let [h (uniffi/with-out-status #(raw/clone-moqannouncement ann %))] | |
| 132 | + (ffi/with-arena [a] | |
| 133 | + (let [out (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))] | |
| 134 | + (uniffi/with-out-status #(raw/method-moqannouncement-path out h %)) | |
| 135 | + (uniffi/lift-string out))))) | |
| 136 | + | |
| 137 | +(defn announcement-broadcast | |
| 138 | + "The MoqBroadcastConsumer this announcement is for." | |
| 139 | + [ann] | |
| 140 | + (let [h (uniffi/with-out-status #(raw/clone-moqannouncement ann %))] | |
| 141 | + (uniffi/with-out-status #(raw/method-moqannouncement-broadcast h %)))) | |
| 142 | + | |
| 143 | +;; --- the catalog ------------------------------------------------------------ | |
| 144 | +;; What a peer's broadcast says it contains. Two things come out of it that | |
| 145 | +;; cannot be guessed from the outside: the track NAMES, and each track's | |
| 146 | +;; CONTAINER — which is the LOC-versus-LEGACY question that answers | |
| 147 | +;; `mux: loc: malformed loc properties` when got wrong. | |
| 148 | + | |
| 149 | +(def ^:private container-names | |
| 150 | + (into {} (map (fn [[k v]] [v k])) containers)) | |
| 151 | + | |
| 152 | +(defn- r-dimensions [c] | |
| 153 | + {:width (uniffi/r-i32! c) :height (uniffi/r-i32! c)}) | |
| 154 | + | |
| 155 | +(defn- r-container [c] | |
| 156 | + (let [v (uniffi/r-i32! c)] | |
| 157 | + (when (= v (containers :cmaf)) (uniffi/r-bytes-span! c)) ; init blob, skipped | |
| 158 | + (container-names v :unknown))) | |
| 159 | + | |
| 160 | +(defn- r-video [c] | |
| 161 | + {:codec (uniffi/r-string! c) | |
| 162 | + :description (uniffi/r-optional! c uniffi/r-bytes-span!) | |
| 163 | + :coded (uniffi/r-optional! c r-dimensions) | |
| 164 | + :aspect (uniffi/r-optional! c r-dimensions) | |
| 165 | + :bitrate (uniffi/r-optional! c uniffi/r-u64!) | |
| 166 | + :stalled (uniffi/r-bool! c) | |
| 167 | + :framerate (uniffi/r-optional! c uniffi/r-f64!) | |
| 168 | + :container (r-container c)}) | |
| 169 | + | |
| 170 | +(defn- r-audio [c] | |
| 171 | + {:codec (uniffi/r-string! c) | |
| 172 | + :description (uniffi/r-optional! c uniffi/r-bytes-span!) | |
| 173 | + :sample-rate (uniffi/r-i32! c) | |
| 174 | + :channels (uniffi/r-i32! c) | |
| 175 | + :bitrate (uniffi/r-optional! c uniffi/r-u64!) | |
| 176 | + :container (r-container c)}) | |
| 177 | + | |
| 178 | +(defn next-catalog! | |
| 179 | + "Ask for the next catalog update; answers an :rb future." | |
| 180 | + [catalog-consumer] | |
| 181 | + (let [h (uniffi/with-out-status #(raw/clone-moqcatalogconsumer catalog-consumer %))] | |
| 182 | + (-> (raw/method-moqcatalogconsumer-next h) | |
| 183 | + (uniffi/start-future :rb)))) | |
| 184 | + | |
| 185 | +(defn lift-catalog | |
| 186 | + "Read an Optional<MoqCatalog> from a settled :rb buffer. | |
| 187 | + | |
| 188 | + Every field is read even though only `:video` is used yet: the fields are | |
| 189 | + positional in the buffer, so skipping one means parsing it anyway, and | |
| 190 | + half-parsing a record is how the next field comes out as nonsense." | |
| 191 | + [rb-ptr] | |
| 192 | + (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len]) | |
| 193 | + data (ffi/read-field rb-ptr uniffi/rust-buffer [:data]) | |
| 194 | + v (when (and (pos? len) (not (ffi/null? data))) | |
| 195 | + (let [c (uniffi/reader data len)] | |
| 196 | + (uniffi/r-optional! | |
| 197 | + c (fn [c] | |
| 198 | + {:video (uniffi/r-map! c r-video) | |
| 199 | + :audio (uniffi/r-map! c r-audio) | |
| 200 | + :display (uniffi/r-optional! c r-dimensions) | |
| 201 | + :rotation (uniffi/r-optional! c uniffi/r-f64!) | |
| 202 | + :flip (uniffi/r-optional! c uniffi/r-bool!) | |
| 203 | + :sections (uniffi/r-map! c uniffi/r-string!)}))))] | |
| 204 | + (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %)) | |
| 205 | + v)) | |
| 206 | + | |
| 207 | +(defn subscribe-catalog! | |
| 208 | + "Watch a broadcast's catalog; answers a future settling to a | |
| 209 | + MoqCatalogConsumer. | |
| 210 | + | |
| 211 | + What names a peer's tracks. `subscribe-media!` wants a track name, and on a | |
| 212 | + broadcast we did not publish there is nothing else to learn it from." | |
| 213 | + [broadcast-consumer] | |
| 214 | + (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastconsumer broadcast-consumer %))] | |
| 215 | + (-> (raw/method-moqbroadcastconsumer-subscribe-catalog h) | |
| 216 | + (uniffi/start-future :u64)))) | |
| 217 | + | |
| 79 | 218 | ;; --- publishing -------------------------------------------------------------- |
| 80 | 219 | |
| 81 | 220 | (defn publish-media! |
| @@ -76,6 +76,145 @@ | |||
| 76 | (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))] | 76 | (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))] |
| 77 | (uniffi/with-out-status #(raw/method-moqbroadcastproducer-consume h %)))) | 77 | (uniffi/with-out-status #(raw/method-moqbroadcastproducer-consume h %)))) |
| 78 | 78 | ||
| 79 | +;; --- discovery -------------------------------------------------------------- | ||
| 80 | +;; How a peer is FOUND, rather than known. `subscribe-media!` needs a path and | ||
| 81 | +;; a track name; announcements are where those come from when the other side | ||
| 82 | +;; of the call is somebody else's client rather than a broadcast we made. | ||
| 83 | + | ||
| 84 | +(defn origin-consumer | ||
| 85 | + "The subscriber's side of an origin." | ||
| 86 | + [origin] | ||
| 87 | + (let [h (uniffi/with-out-status #(raw/clone-moqoriginproducer origin %))] | ||
| 88 | + (uniffi/with-out-status #(raw/method-moqoriginproducer-consume h %)))) | ||
| 89 | + | ||
| 90 | +(defn announced! | ||
| 91 | + "Watch for broadcasts whose path starts with `prefix`; answers a MoqAnnounced. | ||
| 92 | + | ||
| 93 | + An empty prefix watches everything on the origin, which in a call is what | ||
| 94 | + you want: every participant is a broadcast and none of their paths are | ||
| 95 | + known in advance." | ||
| 96 | + [origin-consumer prefix] | ||
| 97 | + (let [h (uniffi/with-out-status #(raw/clone-moqoriginconsumer origin-consumer %))] | ||
| 98 | + (ffi/with-arena [a] | ||
| 99 | + (let [buf (uniffi/lower-string | ||
| 100 | + (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) prefix)] | ||
| 101 | + (uniffi/with-out-status | ||
| 102 | + #(raw/method-moqoriginconsumer-announced h buf %)))))) | ||
| 103 | + | ||
| 104 | +(defn next-announcement! | ||
| 105 | + "Ask for the next announcement; answers an :rb future. | ||
| 106 | + | ||
| 107 | + It settles to an Optional<MoqAnnouncement> — absent when the origin has | ||
| 108 | + closed, which is the end of the watch rather than an error." | ||
| 109 | + [announced] | ||
| 110 | + (let [h (uniffi/with-out-status #(raw/clone-moqannounced announced %))] | ||
| 111 | + (-> (raw/method-moqannounced-next h) | ||
| 112 | + (uniffi/start-future :rb)))) | ||
| 113 | + | ||
| 114 | +(defn lift-announcement | ||
| 115 | + "Read an Optional<MoqAnnouncement> from a settled :rb buffer. | ||
| 116 | + | ||
| 117 | + Answers the announcement's HANDLE, which outlives the buffer — an interface | ||
| 118 | + crosses as a u64 the far side has already cloned for us, so freeing the | ||
| 119 | + buffer it arrived in does not touch it." | ||
| 120 | + [rb-ptr] | ||
| 121 | + (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len]) | ||
| 122 | + data (ffi/read-field rb-ptr uniffi/rust-buffer [:data]) | ||
| 123 | + v (when (and (pos? len) (not (ffi/null? data))) | ||
| 124 | + (let [c (uniffi/reader data len)] | ||
| 125 | + (uniffi/r-optional! c uniffi/r-u64!)))] | ||
| 126 | + (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %)) | ||
| 127 | + v)) | ||
| 128 | + | ||
| 129 | +(defn announcement-path | ||
| 130 | + [ann] | ||
| 131 | + (let [h (uniffi/with-out-status #(raw/clone-moqannouncement ann %))] | ||
| 132 | + (ffi/with-arena [a] | ||
| 133 | + (let [out (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))] | ||
| 134 | + (uniffi/with-out-status #(raw/method-moqannouncement-path out h %)) | ||
| 135 | + (uniffi/lift-string out))))) | ||
| 136 | + | ||
| 137 | +(defn announcement-broadcast | ||
| 138 | + "The MoqBroadcastConsumer this announcement is for." | ||
| 139 | + [ann] | ||
| 140 | + (let [h (uniffi/with-out-status #(raw/clone-moqannouncement ann %))] | ||
| 141 | + (uniffi/with-out-status #(raw/method-moqannouncement-broadcast h %)))) | ||
| 142 | + | ||
| 143 | +;; --- the catalog ------------------------------------------------------------ | ||
| 144 | +;; What a peer's broadcast says it contains. Two things come out of it that | ||
| 145 | +;; cannot be guessed from the outside: the track NAMES, and each track's | ||
| 146 | +;; CONTAINER — which is the LOC-versus-LEGACY question that answers | ||
| 147 | +;; `mux: loc: malformed loc properties` when got wrong. | ||
| 148 | + | ||
| 149 | +(def ^:private container-names | ||
| 150 | + (into {} (map (fn [[k v]] [v k])) containers)) | ||
| 151 | + | ||
| 152 | +(defn- r-dimensions [c] | ||
| 153 | + {:width (uniffi/r-i32! c) :height (uniffi/r-i32! c)}) | ||
| 154 | + | ||
| 155 | +(defn- r-container [c] | ||
| 156 | + (let [v (uniffi/r-i32! c)] | ||
| 157 | + (when (= v (containers :cmaf)) (uniffi/r-bytes-span! c)) ; init blob, skipped | ||
| 158 | + (container-names v :unknown))) | ||
| 159 | + | ||
| 160 | +(defn- r-video [c] | ||
| 161 | + {:codec (uniffi/r-string! c) | ||
| 162 | + :description (uniffi/r-optional! c uniffi/r-bytes-span!) | ||
| 163 | + :coded (uniffi/r-optional! c r-dimensions) | ||
| 164 | + :aspect (uniffi/r-optional! c r-dimensions) | ||
| 165 | + :bitrate (uniffi/r-optional! c uniffi/r-u64!) | ||
| 166 | + :stalled (uniffi/r-bool! c) | ||
| 167 | + :framerate (uniffi/r-optional! c uniffi/r-f64!) | ||
| 168 | + :container (r-container c)}) | ||
| 169 | + | ||
| 170 | +(defn- r-audio [c] | ||
| 171 | + {:codec (uniffi/r-string! c) | ||
| 172 | + :description (uniffi/r-optional! c uniffi/r-bytes-span!) | ||
| 173 | + :sample-rate (uniffi/r-i32! c) | ||
| 174 | + :channels (uniffi/r-i32! c) | ||
| 175 | + :bitrate (uniffi/r-optional! c uniffi/r-u64!) | ||
| 176 | + :container (r-container c)}) | ||
| 177 | + | ||
| 178 | +(defn next-catalog! | ||
| 179 | + "Ask for the next catalog update; answers an :rb future." | ||
| 180 | + [catalog-consumer] | ||
| 181 | + (let [h (uniffi/with-out-status #(raw/clone-moqcatalogconsumer catalog-consumer %))] | ||
| 182 | + (-> (raw/method-moqcatalogconsumer-next h) | ||
| 183 | + (uniffi/start-future :rb)))) | ||
| 184 | + | ||
| 185 | +(defn lift-catalog | ||
| 186 | + "Read an Optional<MoqCatalog> from a settled :rb buffer. | ||
| 187 | + | ||
| 188 | + Every field is read even though only `:video` is used yet: the fields are | ||
| 189 | + positional in the buffer, so skipping one means parsing it anyway, and | ||
| 190 | + half-parsing a record is how the next field comes out as nonsense." | ||
| 191 | + [rb-ptr] | ||
| 192 | + (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len]) | ||
| 193 | + data (ffi/read-field rb-ptr uniffi/rust-buffer [:data]) | ||
| 194 | + v (when (and (pos? len) (not (ffi/null? data))) | ||
| 195 | + (let [c (uniffi/reader data len)] | ||
| 196 | + (uniffi/r-optional! | ||
| 197 | + c (fn [c] | ||
| 198 | + {:video (uniffi/r-map! c r-video) | ||
| 199 | + :audio (uniffi/r-map! c r-audio) | ||
| 200 | + :display (uniffi/r-optional! c r-dimensions) | ||
| 201 | + :rotation (uniffi/r-optional! c uniffi/r-f64!) | ||
| 202 | + :flip (uniffi/r-optional! c uniffi/r-bool!) | ||
| 203 | + :sections (uniffi/r-map! c uniffi/r-string!)}))))] | ||
| 204 | + (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %)) | ||
| 205 | + v)) | ||
| 206 | + | ||
| 207 | +(defn subscribe-catalog! | ||
| 208 | + "Watch a broadcast's catalog; answers a future settling to a | ||
| 209 | + MoqCatalogConsumer. | ||
| 210 | + | ||
| 211 | + What names a peer's tracks. `subscribe-media!` wants a track name, and on a | ||
| 212 | + broadcast we did not publish there is nothing else to learn it from." | ||
| 213 | + [broadcast-consumer] | ||
| 214 | + (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastconsumer broadcast-consumer %))] | ||
| 215 | + (-> (raw/method-moqbroadcastconsumer-subscribe-catalog h) | ||
| 216 | + (uniffi/start-future :u64)))) | ||
| 217 | + | ||
| 79 | ;; --- publishing -------------------------------------------------------------- | 218 | ;; --- publishing -------------------------------------------------------------- |
| 80 | 219 | ||
| 81 | (defn publish-media! | 220 | (defn publish-media! |
modified
src/frq/moq/smoke.clj +85 -48 | @@ -429,60 +429,97 @@ | ||
| 429 | 429 | {:device bad})))) |
| 430 | 430 | true)) |
| 431 | 431 | |
| 432 | +(defn- i420-halves | |
| 433 | + "An I420 frame whose left half is `lo` and right half `hi`. | |
| 434 | + | |
| 435 | + Flat frames are no use as a test: a decode that produced a uniform picture | |
| 436 | + would pass every check a flat frame can make. Two halves also give each | |
| 437 | + peer a distinguishable picture, which is what makes multi-peer testable — | |
| 438 | + if the frames were identical there would be no way to tell a second peer | |
| 439 | + from the first one counted twice." | |
| 440 | + [a w h lo hi] | |
| 441 | + (let [n (h264/i420-size w h) | |
| 442 | + p (ffi/alloc a n)] | |
| 443 | + (dotimes [y h] | |
| 444 | + (dotimes [x w] | |
| 445 | + (ffi/write (+ p (* y w) x) :uint8 (if (< x (quot w 2)) lo hi)))) | |
| 446 | + (dotimes [i (* 2 (quot (* w h) 4))] | |
| 447 | + (ffi/write (+ p (* w h) i) :uint8 0x80)) | |
| 448 | + [p n])) | |
| 449 | + | |
| 432 | 450 | (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." | |
| 451 | + "Two peers' video, end to end through frq.av.plane. | |
| 452 | + | |
| 453 | + Nothing is configured: the plane watches the origin for ANNOUNCEMENTS, | |
| 454 | + reads each peer's catalog for its video track name and container, and | |
| 455 | + subscribes. Our own broadcast is announced back like anyone else's and | |
| 456 | + becomes the self-view. | |
| 457 | + | |
| 458 | + The second broadcast stands in for another participant. It is published | |
| 459 | + the same way a remote client would publish it and discovered the same way | |
| 460 | + — the only thing it does not cross is the wire, which the connect check | |
| 461 | + already covers. | |
| 462 | + | |
| 463 | + Each picture has a different contrast so a peer cannot be confused for the | |
| 464 | + other, and the assertion is per peer: the right key, the right size, and | |
| 465 | + the halves the right way round." | |
| 449 | 466 | [] |
| 450 | 467 | (ffi/with-arena [a] |
| 451 | 468 | (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}) | |
| 469 | + [ours _] (i420-halves a w h 0x40 0xC0) | |
| 470 | + [theirs _] (i420-halves a w h 0xC0 0x40) | |
| 471 | + origin (media/new-origin)] | |
| 472 | + (plane/start! {:origin origin :path "/us" :source (fn [] [ours nil]) | |
| 473 | + :width w :height h :fps 30 :bitrate 200000}) | |
| 474 | + ;; A second participant on the same origin, published exactly as a | |
| 475 | + ;; remote one would be. | |
| 476 | + (let [b2 (media/create-broadcast! origin "/them") | |
| 477 | + p2 (media/publish-media! b2 "avc3") | |
| 478 | + enc (h264/encoder {:width w :height h :fps 30 :bitrate 200000})] | |
| 463 | 479 | (try |
| 464 | - (let [deadline (+ (System/currentTimeMillis) 20000)] | |
| 465 | - (loop [pumps 0] | |
| 480 | + (let [deadline (+ (System/currentTimeMillis) 25000)] | |
| 481 | + (loop [pumps 0 seen {}] | |
| 482 | + ;; Keep the other peer publishing: a subscriber that arrives | |
| 483 | + ;; after the first keyframe needs another one. | |
| 484 | + (h264/force-keyframe! enc) | |
| 485 | + (h264/encode! enc theirs (* pumps 33333) | |
| 486 | + (fn [p len _] | |
| 487 | + (when (pos? len) | |
| 488 | + (media/write-video-frame! p2 p len (* pumps 33333))))) | |
| 466 | 489 | (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!))))))) | |
| 490 | + (let [seen (reduce (fn [m f] | |
| 491 | + (if (contains? m (:key f)) | |
| 492 | + m | |
| 493 | + (let [at #(ffi/read (+ (:rgba f) (* 4 (+ (* 32 (:w f)) %))) :uint8)] | |
| 494 | + (assoc m (:key f) | |
| 495 | + {:w (:w f) :h (:h f) | |
| 496 | + :left (at 8) :right (at 56)})))) | |
| 497 | + seen (plane/poll-frames!))] | |
| 498 | + (cond | |
| 499 | + (>= (count seen) 2) | |
| 500 | + (do | |
| 501 | + (doseq [[k v] seen] | |
| 502 | + (println " " k (str (:w v) "x" (:h v)) | |
| 503 | + "left" (:left v) "right" (:right v))) | |
| 504 | + (when-not (contains? seen "__local__") | |
| 505 | + (throw (ex-info "no self-view" {:keys (keys seen)}))) | |
| 506 | + (let [local (get seen "__local__") | |
| 507 | + other (val (first (dissoc seen "__local__")))] | |
| 508 | + (when-not (< (:left local) (:right local)) | |
| 509 | + (throw (ex-info "self-view halves are the wrong way round" | |
| 510 | + {:frame local}))) | |
| 511 | + (when-not (> (:left other) (:right other)) | |
| 512 | + (throw (ex-info "peer halves are the wrong way round — the feeds are crossed" | |
| 513 | + {:frame other})))) | |
| 514 | + true) | |
| 515 | + | |
| 516 | + (> (System/currentTimeMillis) deadline) | |
| 517 | + (throw (ex-info "did not see two peers" {:pumps pumps :seen (keys seen)})) | |
| 518 | + | |
| 519 | + :else (do (Thread/sleep 10) (recur (inc pumps) seen)))))) | |
| 520 | + (finally | |
| 521 | + (h264/close! enc) | |
| 522 | + (plane/stop!))))))) | |
| 486 | 523 | |
| 487 | 524 | (defn -main [& _] |
| 488 | 525 | (println "libmoq_ffi smoke test") |
| @@ -429,60 +429,97 @@ | |||
| 429 | {:device bad})))) | 429 | {:device bad})))) |
| 430 | true)) | 430 | true)) |
| 431 | 431 | ||
| 432 | +(defn- i420-halves | ||
| 433 | + "An I420 frame whose left half is `lo` and right half `hi`. | ||
| 434 | + | ||
| 435 | + Flat frames are no use as a test: a decode that produced a uniform picture | ||
| 436 | + would pass every check a flat frame can make. Two halves also give each | ||
| 437 | + peer a distinguishable picture, which is what makes multi-peer testable — | ||
| 438 | + if the frames were identical there would be no way to tell a second peer | ||
| 439 | + from the first one counted twice." | ||
| 440 | + [a w h lo hi] | ||
| 441 | + (let [n (h264/i420-size w h) | ||
| 442 | + p (ffi/alloc a n)] | ||
| 443 | + (dotimes [y h] | ||
| 444 | + (dotimes [x w] | ||
| 445 | + (ffi/write (+ p (* y w) x) :uint8 (if (< x (quot w 2)) lo hi)))) | ||
| 446 | + (dotimes [i (* 2 (quot (* w h) 4))] | ||
| 447 | + (ffi/write (+ p (* w h) i) :uint8 0x80)) | ||
| 448 | + [p n])) | ||
| 449 | + | ||
| 432 | (defn- check-plane | 450 | (defn- check-plane |
| 433 | - "A picture end to end through frq.av.plane: source -> H.264 -> MoQ -> RGBA. | 451 | + "Two peers' video, end to end through frq.av.plane. |
| 434 | - | 452 | + |
| 435 | - This is the whole point of the port in one test. The source is a thunk | 453 | + Nothing is configured: the plane watches the origin for ANNOUNCEMENTS, |
| 436 | - answering a synthetic I420 frame, which is what a camera will be; the frame | 454 | + reads each peer's catalog for its video track name and container, and |
| 437 | - goes out through the encoder onto a real MoQ media track, comes back in | 455 | + subscribes. Our own broadcast is announced back like anyone else's and |
| 438 | - through a real subscribe, and is decoded to RGBA ready for | 456 | + becomes the self-view. |
| 439 | - vidya/frame-rgba!. Nothing is faked in between. | 457 | + |
| 440 | - | 458 | + The second broadcast stands in for another participant. It is published |
| 441 | - The frame is not flat this time. A left half at luma 0x40 and a right half | 459 | + the same way a remote client would publish it and discovered the same way |
| 442 | - at 0xC0 means a decode that produced a uniform picture — the failure a gray | 460 | + — the only thing it does not cross is the wire, which the connect check |
| 443 | - test frame cannot distinguish from success — shows up as two equal halves. | 461 | + already covers. |
| 444 | - | 462 | + |
| 445 | - It is pumped rather than awaited, on purpose: pump! is what glimmer's timer | 463 | + Each picture has a different contrast so a peer cannot be confused for the |
| 446 | - will call, so driving it in a loop here is the same code path the app takes, | 464 | + other, and the assertion is per peer: the right key, the right size, and |
| 447 | - including the several pumps a subscribe and a first keyframe take to | 465 | + the halves the right way round." |
| 448 | - settle." | ||
| 449 | [] | 466 | [] |
| 450 | (ffi/with-arena [a] | 467 | (ffi/with-arena [a] |
| 451 | (let [w 64 h 64 | 468 | (let [w 64 h 64 |
| 452 | - n (h264/i420-size w h) | 469 | + [ours _] (i420-halves a w h 0x40 0xC0) |
| 453 | - px (ffi/alloc a n)] | 470 | + [theirs _] (i420-halves a w h 0xC0 0x40) |
| 454 | - ;; Left half dark, right half bright; chroma neutral. | 471 | + origin (media/new-origin)] |
| 455 | - (dotimes [y h] | 472 | + (plane/start! {:origin origin :path "/us" :source (fn [] [ours nil]) |
| 456 | - (dotimes [x w] | 473 | + :width w :height h :fps 30 :bitrate 200000}) |
| 457 | - (ffi/write (+ px (* y w) x) :uint8 (if (< x (quot w 2)) 0x40 0xC0)))) | 474 | + ;; A second participant on the same origin, published exactly as a |
| 458 | - (dotimes [i (* 2 (quot (* w h) 4))] | 475 | + ;; remote one would be. |
| 459 | - (ffi/write (+ px (* w h) i) :uint8 0x80)) | 476 | + (let [b2 (media/create-broadcast! origin "/them") |
| 460 | - (let [origin (media/new-origin)] | 477 | + p2 (media/publish-media! b2 "avc3") |
| 461 | - (plane/start! {:origin origin :path "/plane" :source (fn [] [px n]) | 478 | + enc (h264/encoder {:width w :height h :fps 30 :bitrate 200000})] |
| 462 | - :width w :height h :fps 30 :bitrate 200000}) | ||
| 463 | (try | 479 | (try |
| 464 | - (let [deadline (+ (System/currentTimeMillis) 20000)] | 480 | + (let [deadline (+ (System/currentTimeMillis) 25000)] |
| 465 | - (loop [pumps 0] | 481 | + (loop [pumps 0 seen {}] |
| 482 | + ;; Keep the other peer publishing: a subscriber that arrives | ||
| 483 | + ;; after the first keyframe needs another one. | ||
| 484 | + (h264/force-keyframe! enc) | ||
| 485 | + (h264/encode! enc theirs (* pumps 33333) | ||
| 486 | + (fn [p len _] | ||
| 487 | + (when (pos? len) | ||
| 488 | + (media/write-video-frame! p2 p len (* pumps 33333))))) | ||
| 466 | (plane/pump!) | 489 | (plane/pump!) |
| 467 | - (if-let [f (plane/poll-frame!)] | 490 | + (let [seen (reduce (fn [m f] |
| 468 | - (do | 491 | + (if (contains? m (:key f)) |
| 469 | - (println " frame after" pumps "pumps:" (:w f) "x" (:h f) (pr-str (:key f))) | 492 | + m |
| 470 | - (when-not (and (= w (:w f)) (= h (:h f))) | 493 | + (let [at #(ffi/read (+ (:rgba f) (* 4 (+ (* 32 (:w f)) %))) :uint8)] |
| 471 | - (throw (ex-info "decoded size is wrong" {:got [(:w f) (:h f)]}))) | 494 | + (assoc m (:key f) |
| 472 | - (let [at (fn [x y] (ffi/read (+ (:rgba f) (* 4 (+ (* y (:w f)) x))) :uint8)) | 495 | + {:w (:w f) :h (:h f) |
| 473 | - left (at 8 32) | 496 | + :left (at 8) :right (at 56)})))) |
| 474 | - right (at 56 32)] | 497 | + seen (plane/poll-frames!))] |
| 475 | - (println " left" left "right" right) | 498 | + (cond |
| 476 | - (when-not (< left right) | 499 | + (>= (count seen) 2) |
| 477 | - (throw (ex-info "the two halves came back the same — the picture did not survive" | 500 | + (do |
| 478 | - {:left left :right right}))) | 501 | + (doseq [[k v] seen] |
| 479 | - (when (< (- right left) 40) | 502 | + (println " " k (str (:w v) "x" (:h v)) |
| 480 | - (throw (ex-info "contrast collapsed" {:left left :right right})))) | 503 | + "left" (:left v) "right" (:right v))) |
| 481 | - true) | 504 | + (when-not (contains? seen "__local__") |
| 482 | - (if (> (System/currentTimeMillis) deadline) | 505 | + (throw (ex-info "no self-view" {:keys (keys seen)}))) |
| 483 | - (throw (ex-info "no frame came back through the plane" {:pumps pumps})) | 506 | + (let [local (get seen "__local__") |
| 484 | - (do (Thread/sleep 10) (recur (inc pumps))))))) | 507 | + other (val (first (dissoc seen "__local__")))] |
| 485 | - (finally (plane/stop!))))))) | 508 | + (when-not (< (:left local) (:right local)) |
| 509 | + (throw (ex-info "self-view halves are the wrong way round" | ||
| 510 | + {:frame local}))) | ||
| 511 | + (when-not (> (:left other) (:right other)) | ||
| 512 | + (throw (ex-info "peer halves are the wrong way round — the feeds are crossed" | ||
| 513 | + {:frame other})))) | ||
| 514 | + true) | ||
| 515 | + | ||
| 516 | + (> (System/currentTimeMillis) deadline) | ||
| 517 | + (throw (ex-info "did not see two peers" {:pumps pumps :seen (keys seen)})) | ||
| 518 | + | ||
| 519 | + :else (do (Thread/sleep 10) (recur (inc pumps) seen)))))) | ||
| 520 | + (finally | ||
| 521 | + (h264/close! enc) | ||
| 522 | + (plane/stop!))))))) | ||
| 486 | 523 | ||
| 487 | (defn -main [& _] | 524 | (defn -main [& _] |
| 488 | (println "libmoq_ffi smoke test") | 525 | (println "libmoq_ffi smoke test") |
modified
src/frq/moq/uniffi.clj +18 -0 | @@ -269,6 +269,24 @@ | ||
| 269 | 269 | (let [n (r-i32! c)] |
| 270 | 270 | {:ptr (take! c n) :len n})) |
| 271 | 271 | |
| 272 | +(defn r-f64! | |
| 273 | + "An IEEE 754 double, big-endian like everything else in a RustBuffer." | |
| 274 | + [c] | |
| 275 | + (let [bits (r-u64! c)] | |
| 276 | + (Double/longBitsToDouble (if (>= bits 9223372036854775808) | |
| 277 | + (- bits 18446744073709551616) | |
| 278 | + bits)))) | |
| 279 | + | |
| 280 | +(defn r-map! | |
| 281 | + "An i32 count, then that many key/value pairs. Keys are strings." | |
| 282 | + [c read-value] | |
| 283 | + (let [n (r-i32! c)] | |
| 284 | + (loop [i 0 acc {}] | |
| 285 | + (if (= i n) | |
| 286 | + acc | |
| 287 | + (let [k (r-string! c)] | |
| 288 | + (recur (inc i) (assoc acc k (read-value c)))))))) | |
| 289 | + | |
| 272 | 290 | (defn r-optional! |
| 273 | 291 | "A flag byte, then `f` when it is set." |
| 274 | 292 | [c f] |
| @@ -269,6 +269,24 @@ | |||
| 269 | (let [n (r-i32! c)] | 269 | (let [n (r-i32! c)] |
| 270 | {:ptr (take! c n) :len n})) | 270 | {:ptr (take! c n) :len n})) |
| 271 | 271 | ||
| 272 | +(defn r-f64! | ||
| 273 | + "An IEEE 754 double, big-endian like everything else in a RustBuffer." | ||
| 274 | + [c] | ||
| 275 | + (let [bits (r-u64! c)] | ||
| 276 | + (Double/longBitsToDouble (if (>= bits 9223372036854775808) | ||
| 277 | + (- bits 18446744073709551616) | ||
| 278 | + bits)))) | ||
| 279 | + | ||
| 280 | +(defn r-map! | ||
| 281 | + "An i32 count, then that many key/value pairs. Keys are strings." | ||
| 282 | + [c read-value] | ||
| 283 | + (let [n (r-i32! c)] | ||
| 284 | + (loop [i 0 acc {}] | ||
| 285 | + (if (= i n) | ||
| 286 | + acc | ||
| 287 | + (let [k (r-string! c)] | ||
| 288 | + (recur (inc i) (assoc acc k (read-value c)))))))) | ||
| 289 | + | ||
| 272 | (defn r-optional! | 290 | (defn r-optional! |
| 273 | "A flag byte, then `f` when it is set." | 291 | "A flag byte, then `f` when it is set." |
| 274 | [c f] | 292 | [c f] |