nandi/frqpublic Fork 0
f417261
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.

Run the plane over a real QUIC session

Publish and discover are two origins, not one. On a local origin they are
the same object and it makes no difference; over a session they are
publisher() and consumer(), and conflating them is how a client publishes
into the void or watches an origin nobody announces on. The plane takes them
separately now and defaults the local case to a one-liner.

The test stands a relay up in-process -- MoqServer, a self-signed
certificate, one origin wired as both what it publishes and what it
consumes, which is what makes it a relay rather than two unrelated halves --
and dials it over actual QUIC. A frame goes out, is announced back through
the relay, discovered, subscribed and decoded. Thirteen pumps.

Everything is polled, the relay's accept included: the client cannot finish
connecting until the relay accepts, and there is no other thread to do it
on. That is a fair model of the real thing, where glimmer's timer is the
only clock this code gets.

Two things the Python control taught before any of it was written.

listen() is ASYNC. It reads like a synchronous bind, and not awaiting it
leaves the server not listening -- reported later, and confusingly, as
`bind: not listening; call listen() first`.

And a session must be TOLD, not merely dropped. Freeing the handle without
shutdown() or cancel() panics the process: the drop tries to close the QUIC
connection from whatever thread got there, and outside a tokio worker there
is no reactor to do it on. Either call prevents it, and the plane now makes
one in a finally. That is a crash we would otherwise have found on the first
hung-up call rather than here.

frq.av is still untouched. What is left before it can switch: V4L2 and ALSA
wired in as the plane's actual source and sink rather than test thunks, the
status transitions frq.av reads out of poll_status, and Android.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
nandi committed 2026-09-10T04:40:51-07:00 Browse files
f417261 parent: 3541022
modified src/frq/av/plane.clj +19 -3
@@ -48,6 +48,7 @@
4848 namespace learning about JNI."
4949 (:require [clojure.string :as str]
5050 [frq.av.audio :as audio]
51+ [frq.moq.client :as client]
5152 [frq.moq.media :as media]
5253 [frq.moq.uniffi :as uniffi]
5354 [frq.codec.h264 :as h264]
@@ -76,8 +77,8 @@
7677 Everything that can fail does so HERE rather than at the first frame: the
7778 encoder validates its size, the decoder opens, and the subscribe settles,
7879 so a plane that comes up is one that can carry a picture."
79- [{:keys [origin path source mic width height fps bitrate camera? muted?
80- channels]
80+ [{:keys [origin discover session path source mic width height fps bitrate
81+ camera? muted? channels]
8182 :or {path "/frq" width 640 height 480 fps 30 bitrate 800000
8283 camera? true muted? false channels 1}}]
8384 (stop!)
@@ -99,11 +100,20 @@
99100 :producer producer
100101 :track track
101102 :path path
103+ :session session
102104 ;; The announcement watch is the whole of peer discovery. An
103105 ;; empty prefix takes everything on the origin, because in a
104106 ;; call every participant is a broadcast and none of their
105107 ;; paths are known in advance.
106- :announced (media/announced! (media/origin-consumer origin) "")
108+ ;;
109+ ;; PUBLISH AND DISCOVER ARE TWO ORIGINS, not one. On a local
110+ ;; origin they are the same object and it makes no difference;
111+ ;; over a session they are `publisher()` and `consumer()`, and
112+ ;; conflating them is how a client publishes into the void or
113+ ;; watches an origin nobody announces on. The default keeps the
114+ ;; local case a one-liner.
115+ :announced (media/announced!
116+ (or discover (media/origin-consumer origin)) "")
107117 :announce nil
108118 :peers {}
109119 :encoder (h264/encoder {:width width :height height
@@ -134,6 +144,12 @@
134144 (when-let [r (:ring peer)] (try (audio/close-ring! r) (catch Exception _ nil))))
135145 (when-let [e (:mic-encoder p)] (try (opus/free-encoder! e) (catch Exception _ nil)))
136146 (when-let [m (:mix p)] (try (ffi/free m) (catch Exception _ nil)))
147+ ;; The session has to be told, not merely dropped. Freeing its handle
148+ ;; without a shutdown panics the process — the drop tries to close the
149+ ;; QUIC connection from whatever thread got there, and outside a tokio
150+ ;; worker there is no reactor to do it on.
151+ (when-let [sess (:session p)]
152+ (try (client/shutdown! sess) (catch Exception _ nil)))
137153 (reset! plane nil))
138154 nil)
139155
@@ -48,6 +48,7 @@
48 namespace learning about JNI."48 namespace learning about JNI."
49 (:require [clojure.string :as str]49 (:require [clojure.string :as str]
50 [frq.av.audio :as audio]50 [frq.av.audio :as audio]
51+ [frq.moq.client :as client]
51 [frq.moq.media :as media]52 [frq.moq.media :as media]
52 [frq.moq.uniffi :as uniffi]53 [frq.moq.uniffi :as uniffi]
53 [frq.codec.h264 :as h264]54 [frq.codec.h264 :as h264]
@@ -76,8 +77,8 @@
76 Everything that can fail does so HERE rather than at the first frame: the77 Everything that can fail does so HERE rather than at the first frame: the
77 encoder validates its size, the decoder opens, and the subscribe settles,78 encoder validates its size, the decoder opens, and the subscribe settles,
78 so a plane that comes up is one that can carry a picture."79 so a plane that comes up is one that can carry a picture."
79- [{:keys [origin path source mic width height fps bitrate camera? muted?80+ [{:keys [origin discover session path source mic width height fps bitrate
80- channels]81+ camera? muted? channels]
81 :or {path "/frq" width 640 height 480 fps 30 bitrate 80000082 :or {path "/frq" width 640 height 480 fps 30 bitrate 800000
82 camera? true muted? false channels 1}}]83 camera? true muted? false channels 1}}]
83 (stop!)84 (stop!)
@@ -99,11 +100,20 @@
99 :producer producer100 :producer producer
100 :track track101 :track track
101 :path path102 :path path
103+ :session session
102 ;; The announcement watch is the whole of peer discovery. An104 ;; The announcement watch is the whole of peer discovery. An
103 ;; empty prefix takes everything on the origin, because in a105 ;; empty prefix takes everything on the origin, because in a
104 ;; call every participant is a broadcast and none of their106 ;; call every participant is a broadcast and none of their
105 ;; paths are known in advance.107 ;; paths are known in advance.
106- :announced (media/announced! (media/origin-consumer origin) "")108+ ;;
109+ ;; PUBLISH AND DISCOVER ARE TWO ORIGINS, not one. On a local
110+ ;; origin they are the same object and it makes no difference;
111+ ;; over a session they are `publisher()` and `consumer()`, and
112+ ;; conflating them is how a client publishes into the void or
113+ ;; watches an origin nobody announces on. The default keeps the
114+ ;; local case a one-liner.
115+ :announced (media/announced!
116+ (or discover (media/origin-consumer origin)) "")
107 :announce nil117 :announce nil
108 :peers {}118 :peers {}
109 :encoder (h264/encoder {:width width :height height119 :encoder (h264/encoder {:width width :height height
@@ -134,6 +144,12 @@
134 (when-let [r (:ring peer)] (try (audio/close-ring! r) (catch Exception _ nil))))144 (when-let [r (:ring peer)] (try (audio/close-ring! r) (catch Exception _ nil))))
135 (when-let [e (:mic-encoder p)] (try (opus/free-encoder! e) (catch Exception _ nil)))145 (when-let [e (:mic-encoder p)] (try (opus/free-encoder! e) (catch Exception _ nil)))
136 (when-let [m (:mix p)] (try (ffi/free m) (catch Exception _ nil)))146 (when-let [m (:mix p)] (try (ffi/free m) (catch Exception _ nil)))
147+ ;; The session has to be told, not merely dropped. Freeing its handle
148+ ;; without a shutdown panics the process — the drop tries to close the
149+ ;; QUIC connection from whatever thread got there, and outside a tokio
150+ ;; worker there is no reactor to do it on.
151+ (when-let [sess (:session p)]
152+ (try (client/shutdown! sess) (catch Exception _ nil)))
137 (reset! plane nil))153 (reset! plane nil))
138 nil)154 nil)
139 155
modified src/frq/moq/client.clj +126 -0
@@ -86,9 +86,135 @@
8686
8787 ;; --- the session -------------------------------------------------------------
8888
89+;; --- sessions ---------------------------------------------------------------
90+
91+(defn session-publisher
92+ "The origin this session publishes INTO. Broadcasts created here go out
93+ over the wire."
94+ [session]
95+ (uniffi/with-out-status
96+ #(raw/method-moqsession-publisher (clone-session session) %)))
97+
98+(defn session-consumer
99+ "The origin this session receives FROM. Peers' broadcasts are announced
100+ here."
101+ [session]
102+ (uniffi/with-out-status
103+ #(raw/method-moqsession-consumer (clone-session session) %)))
104+
105+(defn- lower-strings
106+ "A Sequence<String>: an i32 count, then each string length-prefixed."
107+ [dest xs]
108+ (uniffi/lower-buffer dest (into [[:i32 (count xs)]]
109+ (map (fn [x] [:string x])) xs)))
110+
111+(defn set-tls-fingerprints!
112+ "Trust exactly these certificate fingerprints.
113+
114+ What a self-signed relay needs: the certificate is not in any root store,
115+ and the alternative — disabling verification altogether — trusts whatever
116+ answers the address."
117+ [client fingerprints]
118+ (ffi/with-arena [a]
119+ (let [buf (lower-strings (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))
120+ fingerprints)]
121+ (uniffi/with-out-status
122+ #(raw/method-moqclient-set-tls-fingerprints (clone-client client) buf %))))
123+ nil)
124+
125+;; --- a relay ----------------------------------------------------------------
126+;; Not for production — for having something real to connect TO. A MoQ call
127+;; needs a relay in the middle, and standing one up in-process is what lets
128+;; the session path be exercised over actual QUIC rather than mocked.
129+
130+(defn- clone-server [h] (uniffi/with-out-status #(raw/clone-moqserver h %)))
131+(defn- clone-request [h] (uniffi/with-out-status #(raw/clone-moqrequest h %)))
132+
133+(defn new-server [] (uniffi/with-out-status #(raw/constructor-moqserver-new %)))
134+
135+(defn server-bind! [server addr]
136+ (ffi/with-arena [a]
137+ (let [buf (uniffi/lower-string (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) addr)]
138+ (uniffi/with-out-status #(raw/method-moqserver-set-bind (clone-server server) buf %))))
139+ nil)
140+
141+(defn server-tls-generate!
142+ "Make a self-signed certificate for these hostnames."
143+ [server hostnames]
144+ (ffi/with-arena [a]
145+ (let [buf (lower-strings (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) hostnames)]
146+ (uniffi/with-out-status
147+ #(raw/method-moqserver-set-tls-generate (clone-server server) buf %))))
148+ nil)
149+
150+(defn server-origin!
151+ "Wire one origin as both what the relay publishes and what it consumes.
152+
153+ That is what makes it a relay rather than two unrelated halves: a
154+ broadcast arriving from one session is announced to every other."
155+ [server origin]
156+ (ffi/with-arena [a]
157+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))
158+ ;; Optional<MoqOriginProducer>: present, then the handle.
159+ ops [[:u8 1] [:u64 origin]]]
160+ (uniffi/with-out-status
161+ #(raw/method-moqserver-set-publish
162+ (clone-server server) (uniffi/lower-buffer (cell) ops) %))
163+ (uniffi/with-out-status
164+ #(raw/method-moqserver-set-consume
165+ (clone-server server) (uniffi/lower-buffer (cell) ops) %))))
166+ nil)
167+
168+(defn server-listen!
169+ "Bind the socket; answers a future settling to the bound address.
170+
171+ Async, which is easy to miss — `listen` reads like a synchronous bind and
172+ answering it without awaiting leaves the server not listening, which the
173+ next call reports as `bind: not listening; call listen() first`."
174+ [server]
175+ (-> (raw/method-moqserver-listen (clone-server server))
176+ (uniffi/start-future :rb)))
177+
178+(defn server-fingerprints
179+ [server]
180+ (ffi/with-arena [a]
181+ (let [out (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
182+ (uniffi/with-out-status
183+ #(raw/method-moqserver-cert-fingerprints out (clone-server server) %))
184+ (let [len (ffi/read-field out uniffi/rust-buffer [:len])
185+ data (ffi/read-field out uniffi/rust-buffer [:data])
186+ v (if (pos? len)
187+ (uniffi/r-list! (uniffi/reader data len) uniffi/r-string!)
188+ [])]
189+ (uniffi/with-out-status #(raw/rustbuffer-free out %))
190+ v))))
191+
192+(defn server-accept!
193+ "Wait for the next incoming session; answers an :rb future settling to an
194+ Optional<MoqRequest> handle."
195+ [server]
196+ (-> (raw/method-moqserver-accept (clone-server server))
197+ (uniffi/start-future :rb)))
198+
199+(defn accept-request!
200+ "Accept an incoming session; answers a future settling to a MoqSession."
201+ [request]
202+ (-> (raw/method-moqrequest-accept (clone-request request))
203+ (uniffi/start-future :u64)))
204+
205+(defn server-cancel! [server]
206+ (uniffi/with-out-status #(raw/method-moqserver-cancel (clone-server server) %))
207+ nil)
208+
89209 (defn shutdown!
90210 "Graceful shutdown — equivalent to `(cancel! session 0)`.
91211
212+ NOT OPTIONAL BEFORE THE HANDLE GOES. A session dropped without one panics
213+ the process — `there is no reactor running, must be called from the
214+ context of a Tokio 1.x runtime` — because the drop tries to close the
215+ QUIC connection from whatever thread happened to free it. Either this or
216+ `cancel!` has to run first, and `frq.av.plane` does it in a finally.
217+
92218 Named as upstream names it: UniFFI's Kotlin generator already emits a
93219 `close()` that releases the FFI handle, so `close` would mean two different
94220 things depending on which side of the binding you were reading."
@@ -86,9 +86,135 @@
86 86
87 ;; --- the session -------------------------------------------------------------87 ;; --- the session -------------------------------------------------------------
88 88
89+;; --- sessions ---------------------------------------------------------------
90+
91+(defn session-publisher
92+ "The origin this session publishes INTO. Broadcasts created here go out
93+ over the wire."
94+ [session]
95+ (uniffi/with-out-status
96+ #(raw/method-moqsession-publisher (clone-session session) %)))
97+
98+(defn session-consumer
99+ "The origin this session receives FROM. Peers' broadcasts are announced
100+ here."
101+ [session]
102+ (uniffi/with-out-status
103+ #(raw/method-moqsession-consumer (clone-session session) %)))
104+
105+(defn- lower-strings
106+ "A Sequence<String>: an i32 count, then each string length-prefixed."
107+ [dest xs]
108+ (uniffi/lower-buffer dest (into [[:i32 (count xs)]]
109+ (map (fn [x] [:string x])) xs)))
110+
111+(defn set-tls-fingerprints!
112+ "Trust exactly these certificate fingerprints.
113+
114+ What a self-signed relay needs: the certificate is not in any root store,
115+ and the alternative — disabling verification altogether — trusts whatever
116+ answers the address."
117+ [client fingerprints]
118+ (ffi/with-arena [a]
119+ (let [buf (lower-strings (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))
120+ fingerprints)]
121+ (uniffi/with-out-status
122+ #(raw/method-moqclient-set-tls-fingerprints (clone-client client) buf %))))
123+ nil)
124+
125+;; --- a relay ----------------------------------------------------------------
126+;; Not for production — for having something real to connect TO. A MoQ call
127+;; needs a relay in the middle, and standing one up in-process is what lets
128+;; the session path be exercised over actual QUIC rather than mocked.
129+
130+(defn- clone-server [h] (uniffi/with-out-status #(raw/clone-moqserver h %)))
131+(defn- clone-request [h] (uniffi/with-out-status #(raw/clone-moqrequest h %)))
132+
133+(defn new-server [] (uniffi/with-out-status #(raw/constructor-moqserver-new %)))
134+
135+(defn server-bind! [server addr]
136+ (ffi/with-arena [a]
137+ (let [buf (uniffi/lower-string (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) addr)]
138+ (uniffi/with-out-status #(raw/method-moqserver-set-bind (clone-server server) buf %))))
139+ nil)
140+
141+(defn server-tls-generate!
142+ "Make a self-signed certificate for these hostnames."
143+ [server hostnames]
144+ (ffi/with-arena [a]
145+ (let [buf (lower-strings (ffi/alloc a (ffi/layout-size uniffi/rust-buffer)) hostnames)]
146+ (uniffi/with-out-status
147+ #(raw/method-moqserver-set-tls-generate (clone-server server) buf %))))
148+ nil)
149+
150+(defn server-origin!
151+ "Wire one origin as both what the relay publishes and what it consumes.
152+
153+ That is what makes it a relay rather than two unrelated halves: a
154+ broadcast arriving from one session is announced to every other."
155+ [server origin]
156+ (ffi/with-arena [a]
157+ (let [cell #(ffi/alloc a (ffi/layout-size uniffi/rust-buffer))
158+ ;; Optional<MoqOriginProducer>: present, then the handle.
159+ ops [[:u8 1] [:u64 origin]]]
160+ (uniffi/with-out-status
161+ #(raw/method-moqserver-set-publish
162+ (clone-server server) (uniffi/lower-buffer (cell) ops) %))
163+ (uniffi/with-out-status
164+ #(raw/method-moqserver-set-consume
165+ (clone-server server) (uniffi/lower-buffer (cell) ops) %))))
166+ nil)
167+
168+(defn server-listen!
169+ "Bind the socket; answers a future settling to the bound address.
170+
171+ Async, which is easy to miss — `listen` reads like a synchronous bind and
172+ answering it without awaiting leaves the server not listening, which the
173+ next call reports as `bind: not listening; call listen() first`."
174+ [server]
175+ (-> (raw/method-moqserver-listen (clone-server server))
176+ (uniffi/start-future :rb)))
177+
178+(defn server-fingerprints
179+ [server]
180+ (ffi/with-arena [a]
181+ (let [out (ffi/alloc a (ffi/layout-size uniffi/rust-buffer))]
182+ (uniffi/with-out-status
183+ #(raw/method-moqserver-cert-fingerprints out (clone-server server) %))
184+ (let [len (ffi/read-field out uniffi/rust-buffer [:len])
185+ data (ffi/read-field out uniffi/rust-buffer [:data])
186+ v (if (pos? len)
187+ (uniffi/r-list! (uniffi/reader data len) uniffi/r-string!)
188+ [])]
189+ (uniffi/with-out-status #(raw/rustbuffer-free out %))
190+ v))))
191+
192+(defn server-accept!
193+ "Wait for the next incoming session; answers an :rb future settling to an
194+ Optional<MoqRequest> handle."
195+ [server]
196+ (-> (raw/method-moqserver-accept (clone-server server))
197+ (uniffi/start-future :rb)))
198+
199+(defn accept-request!
200+ "Accept an incoming session; answers a future settling to a MoqSession."
201+ [request]
202+ (-> (raw/method-moqrequest-accept (clone-request request))
203+ (uniffi/start-future :u64)))
204+
205+(defn server-cancel! [server]
206+ (uniffi/with-out-status #(raw/method-moqserver-cancel (clone-server server) %))
207+ nil)
208+
89 (defn shutdown!209 (defn shutdown!
90 "Graceful shutdown — equivalent to `(cancel! session 0)`.210 "Graceful shutdown — equivalent to `(cancel! session 0)`.
91 211
212+ NOT OPTIONAL BEFORE THE HANDLE GOES. A session dropped without one panics
213+ the process — `there is no reactor running, must be called from the
214+ context of a Tokio 1.x runtime` — because the drop tries to close the
215+ QUIC connection from whatever thread happened to free it. Either this or
216+ `cancel!` has to run first, and `frq.av.plane` does it in a finally.
217+
92 Named as upstream names it: UniFFI's Kotlin generator already emits a218 Named as upstream names it: UniFFI's Kotlin generator already emits a
93 `close()` that releases the FFI handle, so `close` would mean two different219 `close()` that releases the FFI handle, so `close` would mean two different
94 things depending on which side of the binding you were reading."220 things depending on which side of the binding you were reading."
modified src/frq/moq/media.clj +14 -0
@@ -111,6 +111,20 @@
111111 (-> (raw/method-moqannounced-next h)
112112 (uniffi/start-future :rb))))
113113
114+(defn lift-optional-handle
115+ "Read an Optional<interface> from a settled :rb buffer as a handle.
116+
117+ An interface crosses as a u64 the far side has already cloned for us, so
118+ freeing the buffer it arrived in does not touch it. Announcements,
119+ requests and anything else optional-and-opaque come back this way."
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+ (uniffi/r-optional! (uniffi/reader data len) uniffi/r-u64!))]
125+ (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %))
126+ v))
127+
114128 (defn lift-announcement
115129 "Read an Optional<MoqAnnouncement> from a settled :rb buffer.
116130
@@ -111,6 +111,20 @@
111 (-> (raw/method-moqannounced-next h)111 (-> (raw/method-moqannounced-next h)
112 (uniffi/start-future :rb))))112 (uniffi/start-future :rb))))
113 113
114+(defn lift-optional-handle
115+ "Read an Optional<interface> from a settled :rb buffer as a handle.
116+
117+ An interface crosses as a u64 the far side has already cloned for us, so
118+ freeing the buffer it arrived in does not touch it. Announcements,
119+ requests and anything else optional-and-opaque come back this way."
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+ (uniffi/r-optional! (uniffi/reader data len) uniffi/r-u64!))]
125+ (uniffi/with-out-status #(raw/rustbuffer-free rb-ptr %))
126+ v))
127+
114 (defn lift-announcement128 (defn lift-announcement
115 "Read an Optional<MoqAnnouncement> from a settled :rb buffer.129 "Read an Optional<MoqAnnouncement> from a settled :rb buffer.
116 130
modified src/frq/moq/smoke.clj +84 -2
@@ -32,7 +32,8 @@
3232 had gone wrong yet.
3333
3434 just repl -m frq.moq.smoke"
35- (:require [frq.moq.uniffi :as uniffi]
35+ (:require [clojure.string :as str]
36+ [frq.moq.uniffi :as uniffi]
3637 [frq.moq.raw :as raw]
3738 [frq.moq.client :as client]
3839 [frq.moq.media :as media]
@@ -143,12 +144,14 @@
143144 from a loop like this one."
144145 ([fut what ms] (settle! fut what ms nil))
145146 ([fut what ms lift]
147+ ;; ms 0 means one look and no waiting — for a caller with its own loop.
146148 (let [deadline (+ (System/currentTimeMillis) ms)]
147149 (loop []
148150 (cond
149151 (uniffi/settled? fut) (if lift
150152 (uniffi/complete! fut lift)
151153 (uniffi/complete! fut))
154+ (zero? ms) nil
152155 (> (System/currentTimeMillis) deadline)
153156 (throw (ex-info (str what ": future never settled") {:after-ms ms}))
154157 :else (do (Thread/sleep 10) (recur)))))))
@@ -594,6 +597,84 @@
594597 (opus/free-encoder! enc)
595598 (plane/stop!)))))))
596599
600+(defn- check-session
601+ "The plane over a real QUIC session, not a local origin.
602+
603+ A relay is stood up in-process — MoqServer with a self-signed certificate
604+ and one origin wired as both what it publishes and what it consumes, which
605+ is what makes it a relay rather than two unrelated halves. A client dials
606+ it over actual QUIC, and the plane runs on that session's publisher() and
607+ consumer() rather than on an origin it made itself.
608+
609+ What this proves that the loopback checks cannot: that publish and
610+ discover being two DIFFERENT origins works, that a broadcast survives the
611+ wire, and that the announcement comes back through the relay rather than
612+ from an object we already had.
613+
614+ Everything is polled, the relay included. Its accept has to be driven from
615+ the same loop as the client's connect, because the client cannot finish
616+ connecting until the relay accepts and there is no other thread to do it
617+ on — which is a fair model of the real thing, where glimmer's timer is
618+ the only clock this code gets."
619+ []
620+ (ffi/with-arena [a]
621+ (let [w 64 h 64
622+ [px _] (i420-halves a w h 0x40 0xC0)
623+ relay (media/new-origin)
624+ server (client/new-server)]
625+ (client/server-bind! server "127.0.0.1:0")
626+ (client/server-tls-generate! server ["localhost"])
627+ (client/server-origin! server relay)
628+ (let [addr (settle! (client/server-listen! server) "listen" 10000
629+ uniffi/lift-string)
630+ port (last (str/split addr #":"))
631+ fps (client/server-fingerprints server)
632+ c (client/new-client)]
633+ (println " relay on" addr "fingerprint" (subs (first fps) 0 16))
634+ (client/set-tls-fingerprints! c fps)
635+ (let [connect (client/connect! c (str "https://localhost:" port "/room"))
636+ incoming (client/server-accept! server)
637+ deadline (+ (System/currentTimeMillis) 25000)]
638+ (loop [req nil accepted nil sess nil]
639+ (let [;; The relay side: an incoming request, then accept it.
640+ req (or req (settle! incoming "accept" 0 media/lift-optional-handle))
641+ accepted (or accepted (when req (client/accept-request! req)))
642+ _ (when accepted (settle! accepted "request accept" 0 nil))
643+ ;; The client side.
644+ sess (or sess (settle! connect "connect" 0 nil))]
645+ (cond
646+ sess
647+ (do
648+ (println " connected over QUIC")
649+ (try
650+ (plane/start! {:origin (client/session-publisher sess)
651+ :discover (client/session-consumer sess)
652+ :session sess
653+ :path "/us" :source (fn [] [px nil])
654+ :width w :height h :fps 30 :bitrate 200000})
655+ (let [d2 (+ (System/currentTimeMillis) 20000)]
656+ (loop [pumps 0]
657+ (plane/pump!)
658+ (if-let [f (first (plane/poll-frames!))]
659+ (do (println " frame back through the relay after" pumps
660+ "pumps:" (:w f) "x" (:h f) (pr-str (:key f)))
661+ (when-not (and (= w (:w f)) (= h (:h f)))
662+ (throw (ex-info "wrong size over the wire" {:frame f})))
663+ true)
664+ (if (> (System/currentTimeMillis) d2)
665+ (throw (ex-info "no frame came back over the session"
666+ {:pumps pumps}))
667+ (do (Thread/sleep 10) (recur (inc pumps)))))))
668+ (finally
669+ (plane/stop!)
670+ (client/server-cancel! server))))
671+
672+ (> (System/currentTimeMillis) deadline)
673+ (throw (ex-info "the session never came up"
674+ {:request req :accepted (some? accepted)}))
675+
676+ :else (do (Thread/sleep 10) (recur req accepted sess))))))))))
677+
597678 (defn -main [& _]
598679 (println "libmoq_ffi smoke test")
599680 (let [steps [["contract" check-contract]
@@ -607,7 +688,8 @@
607688 ["alsa" check-alsa]
608689 ["devices" check-enumeration]
609690 ["plane" check-plane]
610- ["audio" check-audio]]]
691+ ["audio" check-audio]
692+ ["session" check-session]]]
611693 (doseq [[name f] steps]
612694 (println (str name ":"))
613695 (f))
@@ -32,7 +32,8 @@
32 had gone wrong yet.32 had gone wrong yet.
33 33
34 just repl -m frq.moq.smoke"34 just repl -m frq.moq.smoke"
35- (:require [frq.moq.uniffi :as uniffi]35+ (:require [clojure.string :as str]
36+ [frq.moq.uniffi :as uniffi]
36 [frq.moq.raw :as raw]37 [frq.moq.raw :as raw]
37 [frq.moq.client :as client]38 [frq.moq.client :as client]
38 [frq.moq.media :as media]39 [frq.moq.media :as media]
@@ -143,12 +144,14 @@
143 from a loop like this one."144 from a loop like this one."
144 ([fut what ms] (settle! fut what ms nil))145 ([fut what ms] (settle! fut what ms nil))
145 ([fut what ms lift]146 ([fut what ms lift]
147+ ;; ms 0 means one look and no waiting — for a caller with its own loop.
146 (let [deadline (+ (System/currentTimeMillis) ms)]148 (let [deadline (+ (System/currentTimeMillis) ms)]
147 (loop []149 (loop []
148 (cond150 (cond
149 (uniffi/settled? fut) (if lift151 (uniffi/settled? fut) (if lift
150 (uniffi/complete! fut lift)152 (uniffi/complete! fut lift)
151 (uniffi/complete! fut))153 (uniffi/complete! fut))
154+ (zero? ms) nil
152 (> (System/currentTimeMillis) deadline)155 (> (System/currentTimeMillis) deadline)
153 (throw (ex-info (str what ": future never settled") {:after-ms ms}))156 (throw (ex-info (str what ": future never settled") {:after-ms ms}))
154 :else (do (Thread/sleep 10) (recur)))))))157 :else (do (Thread/sleep 10) (recur)))))))
@@ -594,6 +597,84 @@
594 (opus/free-encoder! enc)597 (opus/free-encoder! enc)
595 (plane/stop!)))))))598 (plane/stop!)))))))
596 599
600+(defn- check-session
601+ "The plane over a real QUIC session, not a local origin.
602+
603+ A relay is stood up in-process — MoqServer with a self-signed certificate
604+ and one origin wired as both what it publishes and what it consumes, which
605+ is what makes it a relay rather than two unrelated halves. A client dials
606+ it over actual QUIC, and the plane runs on that session's publisher() and
607+ consumer() rather than on an origin it made itself.
608+
609+ What this proves that the loopback checks cannot: that publish and
610+ discover being two DIFFERENT origins works, that a broadcast survives the
611+ wire, and that the announcement comes back through the relay rather than
612+ from an object we already had.
613+
614+ Everything is polled, the relay included. Its accept has to be driven from
615+ the same loop as the client's connect, because the client cannot finish
616+ connecting until the relay accepts and there is no other thread to do it
617+ on — which is a fair model of the real thing, where glimmer's timer is
618+ the only clock this code gets."
619+ []
620+ (ffi/with-arena [a]
621+ (let [w 64 h 64
622+ [px _] (i420-halves a w h 0x40 0xC0)
623+ relay (media/new-origin)
624+ server (client/new-server)]
625+ (client/server-bind! server "127.0.0.1:0")
626+ (client/server-tls-generate! server ["localhost"])
627+ (client/server-origin! server relay)
628+ (let [addr (settle! (client/server-listen! server) "listen" 10000
629+ uniffi/lift-string)
630+ port (last (str/split addr #":"))
631+ fps (client/server-fingerprints server)
632+ c (client/new-client)]
633+ (println " relay on" addr "fingerprint" (subs (first fps) 0 16))
634+ (client/set-tls-fingerprints! c fps)
635+ (let [connect (client/connect! c (str "https://localhost:" port "/room"))
636+ incoming (client/server-accept! server)
637+ deadline (+ (System/currentTimeMillis) 25000)]
638+ (loop [req nil accepted nil sess nil]
639+ (let [;; The relay side: an incoming request, then accept it.
640+ req (or req (settle! incoming "accept" 0 media/lift-optional-handle))
641+ accepted (or accepted (when req (client/accept-request! req)))
642+ _ (when accepted (settle! accepted "request accept" 0 nil))
643+ ;; The client side.
644+ sess (or sess (settle! connect "connect" 0 nil))]
645+ (cond
646+ sess
647+ (do
648+ (println " connected over QUIC")
649+ (try
650+ (plane/start! {:origin (client/session-publisher sess)
651+ :discover (client/session-consumer sess)
652+ :session sess
653+ :path "/us" :source (fn [] [px nil])
654+ :width w :height h :fps 30 :bitrate 200000})
655+ (let [d2 (+ (System/currentTimeMillis) 20000)]
656+ (loop [pumps 0]
657+ (plane/pump!)
658+ (if-let [f (first (plane/poll-frames!))]
659+ (do (println " frame back through the relay after" pumps
660+ "pumps:" (:w f) "x" (:h f) (pr-str (:key f)))
661+ (when-not (and (= w (:w f)) (= h (:h f)))
662+ (throw (ex-info "wrong size over the wire" {:frame f})))
663+ true)
664+ (if (> (System/currentTimeMillis) d2)
665+ (throw (ex-info "no frame came back over the session"
666+ {:pumps pumps}))
667+ (do (Thread/sleep 10) (recur (inc pumps)))))))
668+ (finally
669+ (plane/stop!)
670+ (client/server-cancel! server))))
671+
672+ (> (System/currentTimeMillis) deadline)
673+ (throw (ex-info "the session never came up"
674+ {:request req :accepted (some? accepted)}))
675+
676+ :else (do (Thread/sleep 10) (recur req accepted sess))))))))))
677+
597 (defn -main [& _]678 (defn -main [& _]
598 (println "libmoq_ffi smoke test")679 (println "libmoq_ffi smoke test")
599 (let [steps [["contract" check-contract]680 (let [steps [["contract" check-contract]
@@ -607,7 +688,8 @@
607 ["alsa" check-alsa]688 ["alsa" check-alsa]
608 ["devices" check-enumeration]689 ["devices" check-enumeration]
609 ["plane" check-plane]690 ["plane" check-plane]
610- ["audio" check-audio]]]691+ ["audio" check-audio]
692+ ["session" check-session]]]
611 (doseq [[name f] steps]693 (doseq [[name f] steps]
612 (println (str name ":"))694 (println (str name ":"))
613 (f))695 (f))
modified src/frq/moq/uniffi.clj +6 -0
@@ -287,6 +287,12 @@
287287 (let [k (r-string! c)]
288288 (recur (inc i) (assoc acc k (read-value c))))))))
289289
290+(defn r-list!
291+ "An i32 count, then that many items."
292+ [c read-item]
293+ (let [n (r-i32! c)]
294+ (mapv (fn [_] (read-item c)) (range n))))
295+
290296 (defn r-optional!
291297 "A flag byte, then `f` when it is set."
292298 [c f]
@@ -287,6 +287,12 @@
287 (let [k (r-string! c)]287 (let [k (r-string! c)]
288 (recur (inc i) (assoc acc k (read-value c))))))))288 (recur (inc i) (assoc acc k (read-value c))))))))
289 289
290+(defn r-list!
291+ "An i32 count, then that many items."
292+ [c read-item]
293+ (let [n (r-i32! c)]
294+ (mapv (fn [_] (read-item c)) (range n))))
295+
290 (defn r-optional!296 (defn r-optional!
291 "A flag byte, then `f` when it is set."297 "A flag byte, then `f` when it is set."
292 [c f]298 [c f]