nandi/frqpublic Fork 0
7751d4f
Commits
Clone
git clone https://git.rickub.com/nandi/frq.git
git clone ssh://git@rickub.com/nandi/frq.git

Host key fingerprint (ed25519): SHA256:iycHnxEyq0Q7uyVpB7JlznP0G7JrTPXLYRcAU5CSLhc — verify it before your first connect.

Put a frame into a broadcast and take the same frame out

frq.moq.media is origins, broadcasts and tracks, and frq.moq.uniffi grows the
other half of the buffer codec: UniFFI serialises records, enums and optionals
into a RustBuffer big-endian, and until now this side could only read one.

An origin producer is a plain constructor rather than something a session
hands out, which is what makes any of this testable without a relay: create an
origin, create a broadcast under it, publish a track, and `consume` gives you
the subscriber's side of the same broadcast. No QUIC, no second process, and
the shapes over a session are identical -- only where the origin comes from
changes.

Two tracks in the smoke test, because they prove different things.
subscribe_media is the one frq.av will use and is checked as far as it can be
here: a media track declares a codec, and avc3 means the container really does
try to read Annex B out of whatever arrives, so a frame carrying the word
hello is never coming back out of it. The opaque track beside it parses
nothing, which is what lets the payload round trip be an assertion.

Four found by running it, none of them visible in a signature:

A top-level String argument is BARE UTF-8 -- the RustBuffer's own len
delimits it -- while a string inside a record, enum or optional carries an i32
byte count, because there the buffer's length no longer says where it ends.
Both are String on the Rust side. Backwards is not an ABI fault: the four
prefix bytes join the track name, and the object answers `not found` for a
track that is plainly there.

Each argument is lowered into a buffer of its OWN. The concatenation a
record's fields go through stops at the record boundary, so subscribe_media
takes three buffers and not one.

write_frame takes MoqFrame, not MoqMediaFrame -- one word and one field apart,
and the producer's has no keyframe flag because the container works that out
from the bitstream. The extra byte is not rejected; it is lifted, found left
over, and panicked on.

And complete! handed back a pointer into an arena it had already closed. That
one has been there since the futures were written and never fired, because a
:u64 future does not take that path. An :rb future now requires a lift
function and calls it while the cell is still alive.

Still outstanding: a payload comes back as a jolt string, which survives only
while it is text. frq.av's rule is that a video frame goes from the decoder's
buffer to the texture as a pointer and never becomes a jolt value at all, and
honouring that is the next thing this namespace owes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
nandi committed 2026-09-10T00:10:57-07:00 Browse files
7751d4f parent: 9e3ecaa
added src/frq/moq/media.clj +241 -0
new file mode 100644
@@ -0,0 +1,241 @@
1+(ns frq.moq.media
2+ "Broadcasts, media tracks, and the frames that cross them.
3+
4+ An origin is the piece that makes this testable without a network. A
5+ `MoqOriginProducer` is a plain constructor, not something a session hands
6+ out: create one, create a broadcast under it, publish a media track, and
7+ `consume` gives you the subscriber's side of the very same broadcast. No
8+ QUIC, no relay, no second process — which is how `frq.moq.smoke` can put a
9+ frame in and take the same frame out.
10+
11+ Over a session the shapes are identical; only where the origin comes from
12+ changes, so what is exercised locally is what runs over the wire.
13+
14+ ON FRAME PAYLOADS. `frame-payload` answers a jolt string, and that is a
15+ DEBUGGING affordance, not the path a real frame should take. UniFFI gives
16+ Bytes and String the same wire shape — an i32 length and that many bytes —
17+ so a payload survives the trip only while it happens to be text. H.264 is
18+ not text. `frq.av`'s rule is that a video frame goes from the decoder's
19+ buffer to the texture as a pointer and never becomes a jolt value at all,
20+ and honouring that here means reading the payload's address out of the
21+ buffer and handing it on — which is what the real subscribe path will do,
22+ and what this namespace does not do yet."
23+ (:require [frq.moq.uniffi :as uniffi]
24+ [frq.moq.raw :as raw]
25+ [jolt.ffi :as ffi]))
26+
27+(defn- lowered
28+ "Run `f` with a RustBuffer holding `ops`."
29+ [ops f]
30+ (ffi/with-arena [a]
31+ (let [buf (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
32+ (uniffi/lower-buffer buf ops)
33+ (f buf))))
34+
35+;; --- containers --------------------------------------------------------------
36+
37+(def containers
38+ "MoqContainer, as UniFFI numbers it. LEGACY and LOC carry nothing; CMAF
39+ carries its init segment, which `container-ops` does not build yet."
40+ {:legacy 1 :cmaf 2 :loc 3})
41+
42+(defn container-ops [kind]
43+ (let [v (or (containers kind)
44+ (throw (ex-info "unknown MoqContainer" {:kind kind
45+ :known (keys containers)})))]
46+ (when (= kind :cmaf)
47+ (throw (ex-info "CMAF needs its init segment, which is not built here" {})))
48+ [[:i32 v]]))
49+
50+;; --- origins and broadcasts --------------------------------------------------
51+
52+(defn new-origin
53+ "A MoqOriginProducer with default options (no cache cap)."
54+ []
55+ (lowered [[:u8 0]] ; MoqOriginOptions{cache_capacity_bytes: None}
56+ (fn [buf]
57+ (uniffi/with-out-status
58+ #(raw/constructor-moqoriginproducer-new buf %)))))
59+
60+(defn create-broadcast!
61+ "A MoqBroadcastProducer at `path` under this origin.
62+
63+ `path` goes through `lower-string`, not the record encoder: see the note on
64+ `subscribe-media!` about which strings carry a length and which do not."
65+ [origin path]
66+ (let [h (uniffi/with-out-status #(raw/clone-moqoriginproducer origin %))]
67+ (ffi/with-arena [a]
68+ (let [buf (uniffi/lower-string
69+ (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) path)]
70+ (uniffi/with-out-status
71+ #(raw/method-moqoriginproducer-create-broadcast h buf %))))))
72+
73+(defn broadcast-consumer
74+ "The subscriber's side of a broadcast we produce."
75+ [broadcast]
76+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))]
77+ (uniffi/with-out-status #(raw/method-moqbroadcastproducer-consume h %))))
78+
79+;; --- publishing --------------------------------------------------------------
80+
81+(defn publish-media!
82+ "Publish a media track and answer its MoqMediaProducer.
83+
84+ `init` is a MoqInit: a format string, an init blob, and an optional video
85+ hint. The blob goes out through the :string op because UniFFI lowers Bytes
86+ and String identically — an i32 length and that many bytes."
87+ ([broadcast format] (publish-media! broadcast format ""))
88+ ([broadcast format init-data]
89+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))]
90+ (lowered [[:string format] [:string init-data] [:u8 0]]
91+ (fn [buf]
92+ (uniffi/with-out-status
93+ #(raw/method-moqbroadcastproducer-publish-media h buf %)))))))
94+
95+(defn producer-name
96+ "The track name the object chose for this producer — what a subscriber asks
97+ for by name."
98+ [producer]
99+ (let [h (uniffi/with-out-status #(raw/clone-moqmediaproducer producer %))]
100+ (ffi/with-arena [a]
101+ (let [out (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
102+ (uniffi/with-out-status #(raw/method-moqmediaproducer-name out h %))
103+ (uniffi/lift-string out)))))
104+
105+(defn write-frame!
106+ "Write one MoqFrame to a media producer: a payload and a microsecond stamp.
107+
108+ MoqFrame, NOT MoqMediaFrame. The two names differ by one word and by one
109+ field — what a consumer hands back carries a `keyframe` flag, and what a
110+ producer takes does not, because the container works that out from the
111+ bitstream. Sending the extra byte is not an ABI error: the object lifts the
112+ record, finds a byte left over, and panics with `junk data left in buffer`."
113+ [producer payload timestamp-us]
114+ (let [h (uniffi/with-out-status #(raw/clone-moqmediaproducer producer %))]
115+ (lowered [[:string payload] [:u64 timestamp-us]]
116+ (fn [buf]
117+ (uniffi/with-out-status
118+ #(raw/method-moqmediaproducer-write-frame h buf %)))))
119+ nil)
120+
121+;; --- opaque tracks -----------------------------------------------------------
122+;; The same broadcast, without a codec in the way. `publish_track` and
123+;; `subscribe_track` move plain byte payloads and parse nothing, which is what
124+;; a round trip can actually be checked against: a media track declares a
125+;; format, and `avc3` means the container really does try to read Annex B out
126+;; of whatever is handed to it.
127+
128+(defn publish-track!
129+ "Publish an opaque track by name; answers a MoqTrackProducer."
130+ [broadcast name]
131+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))]
132+ (ffi/with-arena [a]
133+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
134+ (uniffi/with-out-status
135+ #(raw/method-moqbroadcastproducer-publish-track
136+ h
137+ (uniffi/lower-string (cell) name)
138+ (uniffi/lower-buffer (cell) [[:u8 0]]) ; Optional<MoqTrackInfo>
139+ %))))))
140+
141+(defn write-track-frame!
142+ "Write one MoqFrame to an opaque track producer."
143+ [producer payload timestamp-us]
144+ (let [h (uniffi/with-out-status #(raw/clone-moqtrackproducer producer %))]
145+ (lowered [[:string payload] [:u64 timestamp-us]]
146+ (fn [buf]
147+ (uniffi/with-out-status
148+ #(raw/method-moqtrackproducer-write-frame h buf %)))))
149+ nil)
150+
151+(defn subscribe-track!
152+ "Subscribe to an opaque track; answers a future settling to a
153+ MoqTrackConsumer."
154+ [broadcast-consumer name]
155+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastconsumer broadcast-consumer %))]
156+ (ffi/with-arena [a]
157+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
158+ (-> (raw/method-moqbroadcastconsumer-subscribe-track
159+ h
160+ (uniffi/lower-string (cell) name)
161+ (uniffi/lower-buffer (cell) [[:u8 0]])) ; Optional<MoqSubscription>
162+ (uniffi/start-future :u64))))))
163+
164+(defn read-frame!
165+ "Ask an opaque track for its next frame; answers an :rb future."
166+ [consumer]
167+ (let [h (uniffi/with-out-status #(raw/clone-moqtrackconsumer consumer %))]
168+ (-> (raw/method-moqtrackconsumer-read-frame h)
169+ (uniffi/start-future :rb))))
170+
171+(defn lift-plain-frame
172+ "Read an Optional<MoqFrame> out of a settled :rb buffer, and free it."
173+ [rb-ptr]
174+ (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len])
175+ data (ffi/read-field rb-ptr uniffi/rust-buffer [:data])
176+ v (when (and (pos? len) (not (ffi/null? data)))
177+ (let [c (uniffi/reader data len)]
178+ (uniffi/r-optional!
179+ c (fn [c]
180+ {:payload (uniffi/r-string! c)
181+ :timestamp-us (uniffi/r-u64! c)}))))]
182+ (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %))
183+ v))
184+
185+;; --- subscribing -------------------------------------------------------------
186+
187+(defn subscribe-media!
188+ "Subscribe to `name`; answers a future that settles to a MoqMediaConsumer.
189+
190+ THREE buffers, not one. Each argument of a UniFFI method is lowered into a
191+ RustBuffer of its own — the concatenation that a record's fields go through
192+ is a shape that stops at the record boundary.
193+
194+ And the name is lowered by `lower-string`, NOT as [[:string name]]. Where a
195+ string sits decides whether it carries its own length:
196+
197+ * a TOP-LEVEL string argument is bare UTF-8, and the RustBuffer's own
198+ `len` is the length;
199+ * a string INSIDE a record, enum or optional is prefixed with an i32 byte
200+ count, because the buffer's length no longer delimits it.
201+
202+ Both are `String` on the Rust side and the difference is invisible in the
203+ signature. Getting it backwards does not fail at the ABI — the four prefix
204+ bytes simply become part of the name, and the object answers `not found`
205+ for a track that is plainly there."
206+ [broadcast-consumer name container]
207+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastconsumer broadcast-consumer %))]
208+ (ffi/with-arena [a]
209+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
210+ (-> (raw/method-moqbroadcastconsumer-subscribe-media
211+ h
212+ (uniffi/lower-string (cell) name)
213+ (uniffi/lower-buffer (cell) (container-ops container))
214+ (uniffi/lower-buffer (cell) [[:u8 0]])) ; Optional::None
215+ (uniffi/start-future :u64))))))
216+
217+(defn next-frame!
218+ "Ask for the next frame; answers a future.
219+
220+ It settles to a RustBuffer holding an Optional<MoqMediaFrame> — absent when
221+ the track has ended — so it is an :rb future, not a :u64 one."
222+ [consumer]
223+ (let [h (uniffi/with-out-status #(raw/clone-moqmediaconsumer consumer %))]
224+ (-> (raw/method-moqmediaconsumer-next h)
225+ (uniffi/start-future :rb))))
226+
227+(defn lift-frame
228+ "Read an Optional<MoqMediaFrame> out of a settled :rb future's buffer, and
229+ free the buffer. nil means the track ended."
230+ [rb-ptr]
231+ (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len])
232+ data (ffi/read-field rb-ptr uniffi/rust-buffer [:data])
233+ v (when (and (pos? len) (not (ffi/null? data)))
234+ (let [c (uniffi/reader data len)]
235+ (uniffi/r-optional!
236+ c (fn [c]
237+ {:payload (uniffi/r-string! c)
238+ :timestamp-us (uniffi/r-u64! c)
239+ :keyframe (uniffi/r-bool! c)}))))]
240+ (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %))
241+ v))
new file mode 100644
@@ -0,0 +1,241 @@
1+(ns frq.moq.media
2+ "Broadcasts, media tracks, and the frames that cross them.
3+
4+ An origin is the piece that makes this testable without a network. A
5+ `MoqOriginProducer` is a plain constructor, not something a session hands
6+ out: create one, create a broadcast under it, publish a media track, and
7+ `consume` gives you the subscriber's side of the very same broadcast. No
8+ QUIC, no relay, no second process — which is how `frq.moq.smoke` can put a
9+ frame in and take the same frame out.
10+
11+ Over a session the shapes are identical; only where the origin comes from
12+ changes, so what is exercised locally is what runs over the wire.
13+
14+ ON FRAME PAYLOADS. `frame-payload` answers a jolt string, and that is a
15+ DEBUGGING affordance, not the path a real frame should take. UniFFI gives
16+ Bytes and String the same wire shape — an i32 length and that many bytes —
17+ so a payload survives the trip only while it happens to be text. H.264 is
18+ not text. `frq.av`'s rule is that a video frame goes from the decoder's
19+ buffer to the texture as a pointer and never becomes a jolt value at all,
20+ and honouring that here means reading the payload's address out of the
21+ buffer and handing it on — which is what the real subscribe path will do,
22+ and what this namespace does not do yet."
23+ (:require [frq.moq.uniffi :as uniffi]
24+ [frq.moq.raw :as raw]
25+ [jolt.ffi :as ffi]))
26+
27+(defn- lowered
28+ "Run `f` with a RustBuffer holding `ops`."
29+ [ops f]
30+ (ffi/with-arena [a]
31+ (let [buf (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
32+ (uniffi/lower-buffer buf ops)
33+ (f buf))))
34+
35+;; --- containers --------------------------------------------------------------
36+
37+(def containers
38+ "MoqContainer, as UniFFI numbers it. LEGACY and LOC carry nothing; CMAF
39+ carries its init segment, which `container-ops` does not build yet."
40+ {:legacy 1 :cmaf 2 :loc 3})
41+
42+(defn container-ops [kind]
43+ (let [v (or (containers kind)
44+ (throw (ex-info "unknown MoqContainer" {:kind kind
45+ :known (keys containers)})))]
46+ (when (= kind :cmaf)
47+ (throw (ex-info "CMAF needs its init segment, which is not built here" {})))
48+ [[:i32 v]]))
49+
50+;; --- origins and broadcasts --------------------------------------------------
51+
52+(defn new-origin
53+ "A MoqOriginProducer with default options (no cache cap)."
54+ []
55+ (lowered [[:u8 0]] ; MoqOriginOptions{cache_capacity_bytes: None}
56+ (fn [buf]
57+ (uniffi/with-out-status
58+ #(raw/constructor-moqoriginproducer-new buf %)))))
59+
60+(defn create-broadcast!
61+ "A MoqBroadcastProducer at `path` under this origin.
62+
63+ `path` goes through `lower-string`, not the record encoder: see the note on
64+ `subscribe-media!` about which strings carry a length and which do not."
65+ [origin path]
66+ (let [h (uniffi/with-out-status #(raw/clone-moqoriginproducer origin %))]
67+ (ffi/with-arena [a]
68+ (let [buf (uniffi/lower-string
69+ (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) path)]
70+ (uniffi/with-out-status
71+ #(raw/method-moqoriginproducer-create-broadcast h buf %))))))
72+
73+(defn broadcast-consumer
74+ "The subscriber's side of a broadcast we produce."
75+ [broadcast]
76+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))]
77+ (uniffi/with-out-status #(raw/method-moqbroadcastproducer-consume h %))))
78+
79+;; --- publishing --------------------------------------------------------------
80+
81+(defn publish-media!
82+ "Publish a media track and answer its MoqMediaProducer.
83+
84+ `init` is a MoqInit: a format string, an init blob, and an optional video
85+ hint. The blob goes out through the :string op because UniFFI lowers Bytes
86+ and String identically — an i32 length and that many bytes."
87+ ([broadcast format] (publish-media! broadcast format ""))
88+ ([broadcast format init-data]
89+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))]
90+ (lowered [[:string format] [:string init-data] [:u8 0]]
91+ (fn [buf]
92+ (uniffi/with-out-status
93+ #(raw/method-moqbroadcastproducer-publish-media h buf %)))))))
94+
95+(defn producer-name
96+ "The track name the object chose for this producer — what a subscriber asks
97+ for by name."
98+ [producer]
99+ (let [h (uniffi/with-out-status #(raw/clone-moqmediaproducer producer %))]
100+ (ffi/with-arena [a]
101+ (let [out (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
102+ (uniffi/with-out-status #(raw/method-moqmediaproducer-name out h %))
103+ (uniffi/lift-string out)))))
104+
105+(defn write-frame!
106+ "Write one MoqFrame to a media producer: a payload and a microsecond stamp.
107+
108+ MoqFrame, NOT MoqMediaFrame. The two names differ by one word and by one
109+ field — what a consumer hands back carries a `keyframe` flag, and what a
110+ producer takes does not, because the container works that out from the
111+ bitstream. Sending the extra byte is not an ABI error: the object lifts the
112+ record, finds a byte left over, and panics with `junk data left in buffer`."
113+ [producer payload timestamp-us]
114+ (let [h (uniffi/with-out-status #(raw/clone-moqmediaproducer producer %))]
115+ (lowered [[:string payload] [:u64 timestamp-us]]
116+ (fn [buf]
117+ (uniffi/with-out-status
118+ #(raw/method-moqmediaproducer-write-frame h buf %)))))
119+ nil)
120+
121+;; --- opaque tracks -----------------------------------------------------------
122+;; The same broadcast, without a codec in the way. `publish_track` and
123+;; `subscribe_track` move plain byte payloads and parse nothing, which is what
124+;; a round trip can actually be checked against: a media track declares a
125+;; format, and `avc3` means the container really does try to read Annex B out
126+;; of whatever is handed to it.
127+
128+(defn publish-track!
129+ "Publish an opaque track by name; answers a MoqTrackProducer."
130+ [broadcast name]
131+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastproducer broadcast %))]
132+ (ffi/with-arena [a]
133+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
134+ (uniffi/with-out-status
135+ #(raw/method-moqbroadcastproducer-publish-track
136+ h
137+ (uniffi/lower-string (cell) name)
138+ (uniffi/lower-buffer (cell) [[:u8 0]]) ; Optional<MoqTrackInfo>
139+ %))))))
140+
141+(defn write-track-frame!
142+ "Write one MoqFrame to an opaque track producer."
143+ [producer payload timestamp-us]
144+ (let [h (uniffi/with-out-status #(raw/clone-moqtrackproducer producer %))]
145+ (lowered [[:string payload] [:u64 timestamp-us]]
146+ (fn [buf]
147+ (uniffi/with-out-status
148+ #(raw/method-moqtrackproducer-write-frame h buf %)))))
149+ nil)
150+
151+(defn subscribe-track!
152+ "Subscribe to an opaque track; answers a future settling to a
153+ MoqTrackConsumer."
154+ [broadcast-consumer name]
155+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastconsumer broadcast-consumer %))]
156+ (ffi/with-arena [a]
157+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
158+ (-> (raw/method-moqbroadcastconsumer-subscribe-track
159+ h
160+ (uniffi/lower-string (cell) name)
161+ (uniffi/lower-buffer (cell) [[:u8 0]])) ; Optional<MoqSubscription>
162+ (uniffi/start-future :u64))))))
163+
164+(defn read-frame!
165+ "Ask an opaque track for its next frame; answers an :rb future."
166+ [consumer]
167+ (let [h (uniffi/with-out-status #(raw/clone-moqtrackconsumer consumer %))]
168+ (-> (raw/method-moqtrackconsumer-read-frame h)
169+ (uniffi/start-future :rb))))
170+
171+(defn lift-plain-frame
172+ "Read an Optional<MoqFrame> out of a settled :rb buffer, and free it."
173+ [rb-ptr]
174+ (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len])
175+ data (ffi/read-field rb-ptr uniffi/rust-buffer [:data])
176+ v (when (and (pos? len) (not (ffi/null? data)))
177+ (let [c (uniffi/reader data len)]
178+ (uniffi/r-optional!
179+ c (fn [c]
180+ {:payload (uniffi/r-string! c)
181+ :timestamp-us (uniffi/r-u64! c)}))))]
182+ (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %))
183+ v))
184+
185+;; --- subscribing -------------------------------------------------------------
186+
187+(defn subscribe-media!
188+ "Subscribe to `name`; answers a future that settles to a MoqMediaConsumer.
189+
190+ THREE buffers, not one. Each argument of a UniFFI method is lowered into a
191+ RustBuffer of its own — the concatenation that a record's fields go through
192+ is a shape that stops at the record boundary.
193+
194+ And the name is lowered by `lower-string`, NOT as [[:string name]]. Where a
195+ string sits decides whether it carries its own length:
196+
197+ * a TOP-LEVEL string argument is bare UTF-8, and the RustBuffer's own
198+ `len` is the length;
199+ * a string INSIDE a record, enum or optional is prefixed with an i32 byte
200+ count, because the buffer's length no longer delimits it.
201+
202+ Both are `String` on the Rust side and the difference is invisible in the
203+ signature. Getting it backwards does not fail at the ABI — the four prefix
204+ bytes simply become part of the name, and the object answers `not found`
205+ for a track that is plainly there."
206+ [broadcast-consumer name container]
207+ (let [h (uniffi/with-out-status #(raw/clone-moqbroadcastconsumer broadcast-consumer %))]
208+ (ffi/with-arena [a]
209+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
210+ (-> (raw/method-moqbroadcastconsumer-subscribe-media
211+ h
212+ (uniffi/lower-string (cell) name)
213+ (uniffi/lower-buffer (cell) (container-ops container))
214+ (uniffi/lower-buffer (cell) [[:u8 0]])) ; Optional::None
215+ (uniffi/start-future :u64))))))
216+
217+(defn next-frame!
218+ "Ask for the next frame; answers a future.
219+
220+ It settles to a RustBuffer holding an Optional<MoqMediaFrame> — absent when
221+ the track has ended — so it is an :rb future, not a :u64 one."
222+ [consumer]
223+ (let [h (uniffi/with-out-status #(raw/clone-moqmediaconsumer consumer %))]
224+ (-> (raw/method-moqmediaconsumer-next h)
225+ (uniffi/start-future :rb))))
226+
227+(defn lift-frame
228+ "Read an Optional<MoqMediaFrame> out of a settled :rb future's buffer, and
229+ free the buffer. nil means the track ended."
230+ [rb-ptr]
231+ (let [len (ffi/read-field rb-ptr uniffi/rust-buffer [:len])
232+ data (ffi/read-field rb-ptr uniffi/rust-buffer [:data])
233+ v (when (and (pos? len) (not (ffi/null? data)))
234+ (let [c (uniffi/reader data len)]
235+ (uniffi/r-optional!
236+ c (fn [c]
237+ {:payload (uniffi/r-string! c)
238+ :timestamp-us (uniffi/r-u64! c)
239+ :keyframe (uniffi/r-bool! c)}))))]
240+ (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %))
241+ v))
modified src/frq/moq/smoke.clj +68 -1
@@ -35,6 +35,7 @@
3535 (:require [frq.moq.uniffi :as uniffi]
3636 [frq.moq.raw :as raw]
3737 [frq.moq.client :as client]
38+ [frq.moq.media :as media]
3839 [jolt.ffi :as ffi]))
3940
4041 (defn- check-contract []
@@ -128,12 +129,78 @@
128129 (client/free-client! c)
129130 (println " free_moqclient ok")))))
130131
132+(defn- settle!
133+ "Poll `fut` until it settles, or give up. Returns what it settled to.
134+
135+ The polling is the point: every nil here has issued another poll, and this
136+ is what a caller on the loop thread would be doing from a timer instead of
137+ from a loop like this one."
138+ ([fut what ms] (settle! fut what ms nil))
139+ ([fut what ms lift]
140+ (let [deadline (+ (System/currentTimeMillis) ms)]
141+ (loop []
142+ (cond
143+ (uniffi/settled? fut) (if lift
144+ (uniffi/complete! fut lift)
145+ (uniffi/complete! fut))
146+ (> (System/currentTimeMillis) deadline)
147+ (throw (ex-info (str what ": future never settled") {:after-ms ms}))
148+ :else (do (Thread/sleep 10) (recur)))))))
149+
150+(defn- check-media
151+ "Put a payload into a broadcast and take the same payload out of it.
152+
153+ Entirely in-process: an origin producer is a plain constructor, so the
154+ broadcast a subscriber consumes here is the one the producer writes to, with
155+ no QUIC in between. That exercises argument lowering (a top-level string, an
156+ enum, an optional, a record), both subscribe paths, and the frame decode —
157+ everything except the wire itself.
158+
159+ TWO tracks, because they prove different things. `subscribe_media` is the
160+ one frq.av will use, and it is checked as far as it can be checked here: a
161+ media track declares a codec, and avc3 means the container really does try
162+ to read Annex B out of whatever arrives, so a frame carrying the word hello
163+ is never going to come back out of it. The opaque track beside it parses
164+ nothing, which is what lets the payload round trip be an actual assertion
165+ rather than a hope."
166+ []
167+ (let [origin (media/new-origin)
168+ broadcast (media/create-broadcast! origin "/smoke")
169+ producer (media/publish-media! broadcast "avc3")
170+ track (media/producer-name producer)
171+ consumer (media/broadcast-consumer broadcast)]
172+ (println " origin, broadcast, media track:" (pr-str track))
173+
174+ ;; The media path: subscribing is the assertion.
175+ (let [mc (settle! (media/subscribe-media! consumer track :loc) "subscribe_media" 10000)]
176+ (println " subscribe_media ->" mc)
177+ (when (zero? mc)
178+ (throw (ex-info "subscribe_media answered a null handle" {}))))
179+
180+ ;; The opaque path: the payload is the assertion.
181+ (let [tp (media/publish-track! broadcast "data")
182+ tc (settle! (media/subscribe-track! consumer "data") "subscribe_track" 10000)]
183+ (println " subscribe_track ->" tc)
184+ (media/write-track-frame! tp "hello-from-a-frame" 1234567)
185+ (println " wrote a frame")
186+ (let [frame (settle! (media/read-frame! tc) "read_frame" 10000
187+ media/lift-plain-frame)]
188+ (println " got frame:" (pr-str frame))
189+ (when-not frame
190+ (throw (ex-info "track ended instead of delivering a frame" {})))
191+ (when-not (= "hello-from-a-frame" (:payload frame))
192+ (throw (ex-info "payload did not survive" {:frame frame})))
193+ (when-not (= 1234567 (:timestamp-us frame))
194+ (throw (ex-info "timestamp did not survive" {:frame frame})))
195+ true))))
196+
131197 (defn -main [& _]
132198 (println "libmoq_ffi smoke test")
133199 (let [steps [["contract" check-contract]
134200 ["handle" check-handle]
135201 ["string" check-string]
136- ["connect" check-connect]]]
202+ ["connect" check-connect]
203+ ["media" check-media]]]
137204 (doseq [[name f] steps]
138205 (println (str name ":"))
139206 (f))
@@ -35,6 +35,7 @@
35 (:require [frq.moq.uniffi :as uniffi]35 (:require [frq.moq.uniffi :as uniffi]
36 [frq.moq.raw :as raw]36 [frq.moq.raw :as raw]
37 [frq.moq.client :as client]37 [frq.moq.client :as client]
38+ [frq.moq.media :as media]
38 [jolt.ffi :as ffi]))39 [jolt.ffi :as ffi]))
39 40
40 (defn- check-contract []41 (defn- check-contract []
@@ -128,12 +129,78 @@
128 (client/free-client! c)129 (client/free-client! c)
129 (println " free_moqclient ok")))))130 (println " free_moqclient ok")))))
130 131
132+(defn- settle!
133+ "Poll `fut` until it settles, or give up. Returns what it settled to.
134+
135+ The polling is the point: every nil here has issued another poll, and this
136+ is what a caller on the loop thread would be doing from a timer instead of
137+ from a loop like this one."
138+ ([fut what ms] (settle! fut what ms nil))
139+ ([fut what ms lift]
140+ (let [deadline (+ (System/currentTimeMillis) ms)]
141+ (loop []
142+ (cond
143+ (uniffi/settled? fut) (if lift
144+ (uniffi/complete! fut lift)
145+ (uniffi/complete! fut))
146+ (> (System/currentTimeMillis) deadline)
147+ (throw (ex-info (str what ": future never settled") {:after-ms ms}))
148+ :else (do (Thread/sleep 10) (recur)))))))
149+
150+(defn- check-media
151+ "Put a payload into a broadcast and take the same payload out of it.
152+
153+ Entirely in-process: an origin producer is a plain constructor, so the
154+ broadcast a subscriber consumes here is the one the producer writes to, with
155+ no QUIC in between. That exercises argument lowering (a top-level string, an
156+ enum, an optional, a record), both subscribe paths, and the frame decode —
157+ everything except the wire itself.
158+
159+ TWO tracks, because they prove different things. `subscribe_media` is the
160+ one frq.av will use, and it is checked as far as it can be checked here: a
161+ media track declares a codec, and avc3 means the container really does try
162+ to read Annex B out of whatever arrives, so a frame carrying the word hello
163+ is never going to come back out of it. The opaque track beside it parses
164+ nothing, which is what lets the payload round trip be an actual assertion
165+ rather than a hope."
166+ []
167+ (let [origin (media/new-origin)
168+ broadcast (media/create-broadcast! origin "/smoke")
169+ producer (media/publish-media! broadcast "avc3")
170+ track (media/producer-name producer)
171+ consumer (media/broadcast-consumer broadcast)]
172+ (println " origin, broadcast, media track:" (pr-str track))
173+
174+ ;; The media path: subscribing is the assertion.
175+ (let [mc (settle! (media/subscribe-media! consumer track :loc) "subscribe_media" 10000)]
176+ (println " subscribe_media ->" mc)
177+ (when (zero? mc)
178+ (throw (ex-info "subscribe_media answered a null handle" {}))))
179+
180+ ;; The opaque path: the payload is the assertion.
181+ (let [tp (media/publish-track! broadcast "data")
182+ tc (settle! (media/subscribe-track! consumer "data") "subscribe_track" 10000)]
183+ (println " subscribe_track ->" tc)
184+ (media/write-track-frame! tp "hello-from-a-frame" 1234567)
185+ (println " wrote a frame")
186+ (let [frame (settle! (media/read-frame! tc) "read_frame" 10000
187+ media/lift-plain-frame)]
188+ (println " got frame:" (pr-str frame))
189+ (when-not frame
190+ (throw (ex-info "track ended instead of delivering a frame" {})))
191+ (when-not (= "hello-from-a-frame" (:payload frame))
192+ (throw (ex-info "payload did not survive" {:frame frame})))
193+ (when-not (= 1234567 (:timestamp-us frame))
194+ (throw (ex-info "timestamp did not survive" {:frame frame})))
195+ true))))
196+
131 (defn -main [& _]197 (defn -main [& _]
132 (println "libmoq_ffi smoke test")198 (println "libmoq_ffi smoke test")
133 (let [steps [["contract" check-contract]199 (let [steps [["contract" check-contract]
134 ["handle" check-handle]200 ["handle" check-handle]
135 ["string" check-string]201 ["string" check-string]
136- ["connect" check-connect]]]202+ ["connect" check-connect]
203+ ["media" check-media]]]
137 (doseq [[name f] steps]204 (doseq [[name f] steps]
138 (println (str name ":"))205 (println (str name ":"))
139 (f))206 (f))
modified src/frq/moq/uniffi.clj +130 -15
@@ -220,6 +220,108 @@
220220 (with-out-status #(raw/rustbuffer-free rb-ptr %))
221221 s))
222222
223+;; --- the buffer codec -------------------------------------------------------
224+;; Anything that is not a scalar or a handle crosses as a RustBuffer holding
225+;; UniFFI's own serialisation: BIG-ENDIAN fixed-width integers, a bool as one
226+;; byte, bytes and strings as an i32 length followed by that many bytes, and an
227+;; Optional as a 0/1 flag byte followed by the value when present.
228+;;
229+;; Both directions are written out by hand here rather than reusing `read`
230+;; and `write`: jolt's integer types are NATIVE-endian, and this format is not.
231+
232+(defn- be-u64 [p off]
233+ (let [b #(ffi/read (+ p off %) :uint8)]
234+ (loop [i 0 acc 0]
235+ (if (= i 8) acc (recur (inc i) (+ (* acc 256) (b i)))))))
236+
237+(defn reader
238+ "A cursor over a RustBuffer's bytes."
239+ [data len]
240+ (atom {:data data :len len :off 0}))
241+
242+(defn- take! [c n]
243+ (let [{:keys [off len]} @c]
244+ (when (> (+ off n) len)
245+ (throw (ex-info "read past the end of a RustBuffer"
246+ {:off off :want n :len len})))
247+ (swap! c update :off + n)
248+ (+ (:data @c) off)))
249+
250+(defn r-u8! [c] (ffi/read (take! c 1) :uint8))
251+(defn r-bool! [c] (not (zero? (r-u8! c))))
252+(defn r-i32! [c] (let [p (take! c 4)] (be-u32 p 0)))
253+(defn r-u64! [c] (let [p (take! c 8)] (be-u64 p 0)))
254+
255+(defn r-string!
256+ "An i32 byte length, then that many UTF-8 bytes."
257+ [c]
258+ (let [n (r-i32! c)]
259+ (if (zero? n) "" (ffi/read-bytes (take! c n) n))))
260+
261+(defn r-optional!
262+ "A flag byte, then `f` when it is set."
263+ [c f]
264+ (when (= 1 (r-u8! c)) (f c)))
265+
266+;; -- writing --
267+
268+(defn lower-buffer
269+ "Serialise `ops` and hand the bytes to Rust as a RustBuffer in `dest`.
270+
271+ `ops` is a sequence of [type value] pairs in wire order, e.g.
272+
273+ [[:i32 3] [:u8 0]] ; MoqContainer::LOC, then Optional::None
274+
275+ Strings are materialised FIRST, in the arena, because their wire length is a
276+ UTF-8 byte count and jolt will only tell us one by encoding the string — so
277+ the total size is not known until every string has been. `string->ptr`
278+ records the size it allocated, which counts the NUL it appends; the wire
279+ length is one less, since a length-counted string carries no terminator.
280+
281+ `from_bytes` copies, so the scratch this builds in may die with the call."
282+ [dest ops]
283+ (ffi/with-arena [a]
284+ (let [;; [op value encoded-pointer byte-count]
285+ prepared (mapv (fn [[op v]]
286+ (if (= op :string)
287+ (let [p (ffi/string->ptr a v)]
288+ [op v p (max 0 (dec (ffi/size p)))])
289+ [op v nil nil]))
290+ ops)
291+ n (reduce (fn [n [op _ _ k]]
292+ (+ n (case op
293+ (:u8 :bool) 1
294+ :i32 4
295+ :u64 8
296+ :string (+ 4 k))))
297+ 0 prepared)
298+ buf (ffi/alloc a (max n 1))
299+ fbs (ffi/alloc a (ffi/layout-size foreign-bytes))
300+ put-int! (fn [off width v]
301+ (dotimes [i width]
302+ (ffi/write (+ buf off i) :uint8
303+ (bit-and (bit-shift-right v (* 8 (- width 1 i)))
304+ 255))))]
305+ (loop [off 0 todo (seq prepared)]
306+ (when todo
307+ (let [[op v p k] (first todo)]
308+ (recur
309+ (case op
310+ (:u8 :bool)
311+ (do (ffi/write (+ buf off) :uint8
312+ (if (= op :bool) (if v 1 0) v))
313+ (inc off))
314+
315+ :i32 (do (put-int! off 4 v) (+ off 4))
316+ :u64 (do (put-int! off 8 v) (+ off 8))
317+ :string (do (put-int! off 4 k)
318+ (when (pos? k) (ffi/copy p (+ buf off 4) k))
319+ (+ off 4 k)))
320+ (next todo)))))
321+ (ffi/write fbs foreign-bytes {:len n :data buf})
322+ (with-out-status #(raw/rustbuffer-from-bytes dest fbs %))
323+ dest)))
324+
223325 ;; --- futures -----------------------------------------------------------------
224326
225327 (defn- continuation
@@ -280,21 +382,34 @@
280382 Only valid once `settled?` has answered true — completing early is what
281383 blocks the calling thread, which on the loop thread is the freeze this whole
282384 polling shape exists to avoid. The future handle is freed either way, so a
283- raising `complete` still does not leak one."
284- [{:keys [handle kind]}]
285- (try
286- (case kind
287- :u64 (with-out-status #(raw/rust-future-complete-u64 handle %))
288- :void (do (with-out-status #(raw/rust-future-complete-void handle %)) nil)
289- :rb (ffi/with-arena [a]
290- (let [out (ffi/alloc a (ffi/layout-size rust-buffer))]
291- (with-out-status #(raw/rust-future-complete-rust-buffer out handle %))
292- out)))
293- (finally
294- (case kind
295- :u64 (raw/rust-future-free-u64 handle)
296- :void (raw/rust-future-free-void handle)
297- :rb (raw/rust-future-free-rust-buffer handle)))))
385+ raising `complete` still does not leak one.
386+
387+ An :rb future REQUIRES a `lift` function, and it is called while the buffer
388+ cell is still alive. The cell is arena memory whose lifetime is this call:
389+ handing the pointer back to a caller to read afterwards reads memory that
390+ has already been released, and what comes back is not a fault but a
391+ RustBuffer whose length no longer matches its capacity — which the object
392+ then panics on, some way from the mistake."
393+ ([fut]
394+ (let [{:keys [kind]} fut]
395+ (when (= kind :rb)
396+ (throw (ex-info "an :rb future needs a lift function — see complete!"
397+ {:kind kind})))
398+ (complete! fut nil)))
399+ ([{:keys [handle kind]} lift]
400+ (try
401+ (case kind
402+ :u64 (with-out-status #(raw/rust-future-complete-u64 handle %))
403+ :void (do (with-out-status #(raw/rust-future-complete-void handle %)) nil)
404+ :rb (ffi/with-arena [a]
405+ (let [out (ffi/alloc a (ffi/layout-size rust-buffer))]
406+ (with-out-status #(raw/rust-future-complete-rust-buffer out handle %))
407+ (lift out))))
408+ (finally
409+ (case kind
410+ :u64 (raw/rust-future-free-u64 handle)
411+ :void (raw/rust-future-free-void handle)
412+ :rb (raw/rust-future-free-rust-buffer handle))))))
298413
299414 ;; --- the version guard -------------------------------------------------------
300415
@@ -220,6 +220,108 @@
220 (with-out-status #(raw/rustbuffer-free rb-ptr %))220 (with-out-status #(raw/rustbuffer-free rb-ptr %))
221 s))221 s))
222 222
223+;; --- the buffer codec -------------------------------------------------------
224+;; Anything that is not a scalar or a handle crosses as a RustBuffer holding
225+;; UniFFI's own serialisation: BIG-ENDIAN fixed-width integers, a bool as one
226+;; byte, bytes and strings as an i32 length followed by that many bytes, and an
227+;; Optional as a 0/1 flag byte followed by the value when present.
228+;;
229+;; Both directions are written out by hand here rather than reusing `read`
230+;; and `write`: jolt's integer types are NATIVE-endian, and this format is not.
231+
232+(defn- be-u64 [p off]
233+ (let [b #(ffi/read (+ p off %) :uint8)]
234+ (loop [i 0 acc 0]
235+ (if (= i 8) acc (recur (inc i) (+ (* acc 256) (b i)))))))
236+
237+(defn reader
238+ "A cursor over a RustBuffer's bytes."
239+ [data len]
240+ (atom {:data data :len len :off 0}))
241+
242+(defn- take! [c n]
243+ (let [{:keys [off len]} @c]
244+ (when (> (+ off n) len)
245+ (throw (ex-info "read past the end of a RustBuffer"
246+ {:off off :want n :len len})))
247+ (swap! c update :off + n)
248+ (+ (:data @c) off)))
249+
250+(defn r-u8! [c] (ffi/read (take! c 1) :uint8))
251+(defn r-bool! [c] (not (zero? (r-u8! c))))
252+(defn r-i32! [c] (let [p (take! c 4)] (be-u32 p 0)))
253+(defn r-u64! [c] (let [p (take! c 8)] (be-u64 p 0)))
254+
255+(defn r-string!
256+ "An i32 byte length, then that many UTF-8 bytes."
257+ [c]
258+ (let [n (r-i32! c)]
259+ (if (zero? n) "" (ffi/read-bytes (take! c n) n))))
260+
261+(defn r-optional!
262+ "A flag byte, then `f` when it is set."
263+ [c f]
264+ (when (= 1 (r-u8! c)) (f c)))
265+
266+;; -- writing --
267+
268+(defn lower-buffer
269+ "Serialise `ops` and hand the bytes to Rust as a RustBuffer in `dest`.
270+
271+ `ops` is a sequence of [type value] pairs in wire order, e.g.
272+
273+ [[:i32 3] [:u8 0]] ; MoqContainer::LOC, then Optional::None
274+
275+ Strings are materialised FIRST, in the arena, because their wire length is a
276+ UTF-8 byte count and jolt will only tell us one by encoding the string — so
277+ the total size is not known until every string has been. `string->ptr`
278+ records the size it allocated, which counts the NUL it appends; the wire
279+ length is one less, since a length-counted string carries no terminator.
280+
281+ `from_bytes` copies, so the scratch this builds in may die with the call."
282+ [dest ops]
283+ (ffi/with-arena [a]
284+ (let [;; [op value encoded-pointer byte-count]
285+ prepared (mapv (fn [[op v]]
286+ (if (= op :string)
287+ (let [p (ffi/string->ptr a v)]
288+ [op v p (max 0 (dec (ffi/size p)))])
289+ [op v nil nil]))
290+ ops)
291+ n (reduce (fn [n [op _ _ k]]
292+ (+ n (case op
293+ (:u8 :bool) 1
294+ :i32 4
295+ :u64 8
296+ :string (+ 4 k))))
297+ 0 prepared)
298+ buf (ffi/alloc a (max n 1))
299+ fbs (ffi/alloc a (ffi/layout-size foreign-bytes))
300+ put-int! (fn [off width v]
301+ (dotimes [i width]
302+ (ffi/write (+ buf off i) :uint8
303+ (bit-and (bit-shift-right v (* 8 (- width 1 i)))
304+ 255))))]
305+ (loop [off 0 todo (seq prepared)]
306+ (when todo
307+ (let [[op v p k] (first todo)]
308+ (recur
309+ (case op
310+ (:u8 :bool)
311+ (do (ffi/write (+ buf off) :uint8
312+ (if (= op :bool) (if v 1 0) v))
313+ (inc off))
314+
315+ :i32 (do (put-int! off 4 v) (+ off 4))
316+ :u64 (do (put-int! off 8 v) (+ off 8))
317+ :string (do (put-int! off 4 k)
318+ (when (pos? k) (ffi/copy p (+ buf off 4) k))
319+ (+ off 4 k)))
320+ (next todo)))))
321+ (ffi/write fbs foreign-bytes {:len n :data buf})
322+ (with-out-status #(raw/rustbuffer-from-bytes dest fbs %))
323+ dest)))
324+
223 ;; --- futures -----------------------------------------------------------------325 ;; --- futures -----------------------------------------------------------------
224 326
225 (defn- continuation327 (defn- continuation
@@ -280,21 +382,34 @@
280 Only valid once `settled?` has answered true — completing early is what382 Only valid once `settled?` has answered true — completing early is what
281 blocks the calling thread, which on the loop thread is the freeze this whole383 blocks the calling thread, which on the loop thread is the freeze this whole
282 polling shape exists to avoid. The future handle is freed either way, so a384 polling shape exists to avoid. The future handle is freed either way, so a
283- raising `complete` still does not leak one."385+ raising `complete` still does not leak one.
284- [{:keys [handle kind]}]386+
285- (try387+ An :rb future REQUIRES a `lift` function, and it is called while the buffer
286- (case kind388+ cell is still alive. The cell is arena memory whose lifetime is this call:
287- :u64 (with-out-status #(raw/rust-future-complete-u64 handle %))389+ handing the pointer back to a caller to read afterwards reads memory that
288- :void (do (with-out-status #(raw/rust-future-complete-void handle %)) nil)390+ has already been released, and what comes back is not a fault but a
289- :rb (ffi/with-arena [a]391+ RustBuffer whose length no longer matches its capacity — which the object
290- (let [out (ffi/alloc a (ffi/layout-size rust-buffer))]392+ then panics on, some way from the mistake."
291- (with-out-status #(raw/rust-future-complete-rust-buffer out handle %))393+ ([fut]
292- out)))394+ (let [{:keys [kind]} fut]
293- (finally395+ (when (= kind :rb)
294- (case kind396+ (throw (ex-info "an :rb future needs a lift function — see complete!"
295- :u64 (raw/rust-future-free-u64 handle)397+ {:kind kind})))
296- :void (raw/rust-future-free-void handle)398+ (complete! fut nil)))
297- :rb (raw/rust-future-free-rust-buffer handle)))))399+ ([{:keys [handle kind]} lift]
400+ (try
401+ (case kind
402+ :u64 (with-out-status #(raw/rust-future-complete-u64 handle %))
403+ :void (do (with-out-status #(raw/rust-future-complete-void handle %)) nil)
404+ :rb (ffi/with-arena [a]
405+ (let [out (ffi/alloc a (ffi/layout-size rust-buffer))]
406+ (with-out-status #(raw/rust-future-complete-rust-buffer out handle %))
407+ (lift out))))
408+ (finally
409+ (case kind
410+ :u64 (raw/rust-future-free-u64 handle)
411+ :void (raw/rust-future-free-void handle)
412+ :rb (raw/rust-future-free-rust-buffer handle))))))
298 413
299 ;; --- the version guard -------------------------------------------------------414 ;; --- the version guard -------------------------------------------------------
300 415