1use crate::origin;
2use crate::{
3 ALPN_14, ALPN_15, ALPN_16, ALPN_17, ALPN_18, ALPN_19, ALPN_LITE, ALPN_LITE_03, ALPN_LITE_04, ALPN_LITE_05,
4 ALPN_LITE_06_WIP, Consume, Driver, Error, NEGOTIATED, Role, Session, Version, Versions,
5 coding::{Decode, Encode, Reader, Stream},
6 ietf, lite, setup, stats,
7};
8
9#[derive(Default, Clone)]
11pub struct Server {
12 publish: Option<origin::Consumer>,
13 subscribe: Option<origin::Producer>,
14 stats: stats::Session,
15 versions: Versions,
16}
17
18impl Server {
19 pub fn new() -> Self {
21 Default::default()
22 }
23
24 pub fn with_publisher(mut self, publish: impl Consume<origin::Consumer>) -> Self {
29 self.publish = Some(publish.consume());
30 self
31 }
32
33 pub fn with_subscriber(mut self, subscribe: origin::Producer) -> Self {
36 self.subscribe = Some(subscribe);
37 self
38 }
39
40 pub fn with_stats(mut self, stats: stats::Session) -> Self {
45 self.stats = stats;
46 self
47 }
48
49 pub fn with_origin(self, origin: origin::Producer) -> Self {
51 self.with_publisher(&origin).with_subscriber(origin)
52 }
53
54 pub fn with_versions(mut self, versions: Versions) -> Self {
57 self.versions = versions;
58 self
59 }
60
61 pub async fn accept<S: web_transport_trait::Session>(&self, session: S) -> Result<(Session, Driver), Error> {
68 self.accept_request(session).await?.ok().await
69 }
70
71 pub async fn accept_request<S: web_transport_trait::Session>(&self, session: S) -> Result<Request<S>, Error> {
82 let deferred = |handshake| Request {
85 path: None,
86 role: None,
87 inner: Some(RequestInner {
88 server: self.clone(),
89 handshake,
90 }),
91 };
92
93 let (encoding, supported) = match session.protocol() {
94 Some(ALPN_19) => {
95 self.versions
96 .select(Version::Ietf(ietf::Version::Draft19))
97 .ok_or(Error::Version)?;
98 return self.accept_ietf_modern(session, ietf::Version::Draft19).await;
99 }
100 Some(ALPN_18) => {
101 self.versions
102 .select(Version::Ietf(ietf::Version::Draft18))
103 .ok_or(Error::Version)?;
104 return self.accept_ietf_modern(session, ietf::Version::Draft18).await;
105 }
106 Some(ALPN_17) => {
107 self.versions
108 .select(Version::Ietf(ietf::Version::Draft17))
109 .ok_or(Error::Version)?;
110 return self.accept_ietf_modern(session, ietf::Version::Draft17).await;
111 }
112 Some(ALPN_16) => {
113 let v = self
114 .versions
115 .select(Version::Ietf(ietf::Version::Draft16))
116 .ok_or(Error::Version)?;
117 (v, v.into())
118 }
119 Some(ALPN_15) => {
120 let v = self
121 .versions
122 .select(Version::Ietf(ietf::Version::Draft15))
123 .ok_or(Error::Version)?;
124 (v, v.into())
125 }
126 Some(ALPN_14) => {
127 let v = self
128 .versions
129 .select(Version::Ietf(ietf::Version::Draft14))
130 .ok_or(Error::Version)?;
131 (v, v.into())
132 }
133 Some(alpn @ (ALPN_LITE_05 | ALPN_LITE_06_WIP)) => {
134 let version = match alpn {
135 ALPN_LITE_06_WIP => lite::Version::Lite06Wip,
136 _ => lite::Version::Lite05,
137 };
138 self.versions.select(Version::Lite(version)).ok_or(Error::Version)?;
139
140 let client_setup = lite::accept_setup(&session, version).await?;
144 return Ok(Request {
145 path: client_setup.path.clone(),
146 role: client_setup.role,
147 inner: Some(RequestInner {
148 server: self.clone(),
149 handshake: Handshake::LiteSetup {
150 session,
151 version,
152 client_setup,
153 },
154 }),
155 });
156 }
157 Some(ALPN_LITE_04) => {
158 self.versions
159 .select(Version::Lite(lite::Version::Lite04))
160 .ok_or(Error::Version)?;
161 return Ok(deferred(Handshake::LiteBare {
162 session,
163 version: lite::Version::Lite04,
164 }));
165 }
166 Some(ALPN_LITE_03) => {
167 self.versions
168 .select(Version::Lite(lite::Version::Lite03))
169 .ok_or(Error::Version)?;
170 return Ok(deferred(Handshake::LiteBare {
171 session,
172 version: lite::Version::Lite03,
173 }));
174 }
175 Some(ALPN_LITE) | None => {
176 let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?;
177 (Version::Ietf(ietf::Version::Draft14), supported)
178 }
179 Some(p) => return Err(Error::UnknownAlpn(p.to_string())),
180 };
181
182 let mut stream = Stream::accept(&session, encoding).await?;
185 let mut client: setup::Client = stream.reader.decode().await?;
186
187 let version = client
188 .versions
189 .iter()
190 .flat_map(|v| Version::try_from(*v).ok())
191 .find(|v| supported.contains(v))
192 .ok_or(Error::Version)?;
193
194 let (path, request_id_max) = match version {
198 Version::Ietf(v) => {
199 let params = ietf::Parameters::decode(&mut client.parameters, v)?;
200 let path = match params.get_bytes(ietf::ParameterBytes::Path) {
201 Some(bytes) => Some(
202 std::str::from_utf8(bytes)
203 .map_err(|_| Error::Decode(crate::DecodeError::InvalidValue))?
204 .to_owned(),
205 ),
206 None => None,
207 };
208 let request_id_max = params
209 .get_varint(ietf::ParameterVarInt::MaxRequestId)
210 .map(ietf::RequestId);
211 (path, request_id_max)
212 }
213 Version::Lite(_) => (None, None),
214 };
215
216 Ok(Request {
217 path,
218 role: None,
219 inner: Some(RequestInner {
220 server: self.clone(),
221 handshake: Handshake::Legacy {
222 session,
223 stream,
224 version,
225 request_id_max,
226 },
227 }),
228 })
229 }
230
231 async fn accept_ietf_modern<S: web_transport_trait::Session>(
234 &self,
235 session: S,
236 version: ietf::Version,
237 ) -> Result<Request<S>, Error> {
238 let (peer_setup, path) = ietf::accept_setup(&session, version).await?;
239 Ok(Request {
240 path,
241 role: None,
242 inner: Some(RequestInner {
243 server: self.clone(),
244 handshake: Handshake::IetfModern {
245 session,
246 version,
247 peer_setup,
248 },
249 }),
250 })
251 }
252}
253
254pub struct Request<S: web_transport_trait::Session> {
262 path: Option<String>,
263 role: Option<Role>,
264 inner: Option<RequestInner<S>>,
266}
267
268struct RequestInner<S: web_transport_trait::Session> {
270 server: Server,
271 handshake: Handshake<S>,
272}
273
274enum Handshake<S: web_transport_trait::Session> {
277 IetfModern {
281 session: S,
282 version: ietf::Version,
283 peer_setup: Reader<S::RecvStream, Version>,
284 },
285 LiteBare { session: S, version: lite::Version },
287 Legacy {
291 session: S,
292 stream: Stream<S, Version>,
293 version: Version,
294 request_id_max: Option<ietf::RequestId>,
295 },
296 LiteSetup {
299 session: S,
300 version: lite::Version,
301 client_setup: lite::Setup,
302 },
303}
304
305impl<S: web_transport_trait::Session> Request<S> {
306 pub fn path(&self) -> &str {
313 self.path.as_deref().unwrap_or("")
314 }
315
316 pub fn role(&self) -> Option<Role> {
325 self.role
326 }
327
328 pub fn with_publisher(mut self, publish: impl Consume<origin::Consumer>) -> Self {
331 self.inner_mut().server.publish = Some(publish.consume());
332 self
333 }
334
335 pub fn with_subscriber(mut self, subscribe: origin::Producer) -> Self {
337 self.inner_mut().server.subscribe = Some(subscribe);
338 self
339 }
340
341 pub fn with_stats(mut self, stats: stats::Session) -> Self {
344 self.inner_mut().server.stats = stats;
345 self
346 }
347
348 fn inner_mut(&mut self) -> &mut RequestInner<S> {
349 self.inner.as_mut().expect("request already responded")
350 }
351
352 pub async fn ok(mut self) -> Result<(Session, Driver), Error> {
355 let RequestInner { server, handshake } = self.inner.take().expect("request already responded");
356
357 let publish = server.publish.map(|origin| origin.with_stats(server.stats.clone()));
361 let subscribe = server.subscribe.map(|origin| origin.with_stats(server.stats.clone()));
362
363 let (session, mut stream, version, request_id_max) = match handshake {
364 Handshake::IetfModern {
365 session,
366 version,
367 peer_setup,
368 } => {
369 let protocol = ietf::start(
372 session.clone(),
373 None,
374 None,
375 false,
376 publish,
377 subscribe,
378 version,
379 None,
380 Some(peer_setup),
381 )?;
382 tracing::debug!(?version, "connected");
383 return Ok(Session::new(session, version.into(), None, protocol));
384 }
385 Handshake::LiteBare { session, version } => {
386 let start = lite::start(
387 session.clone(),
388 None,
389 publish,
390 subscribe,
391 version,
392 lite::Setup::default(),
393 None,
394 )?;
395 return Ok(Session::new(
396 session,
397 version.into(),
398 start.recv_bandwidth,
399 start.driver,
400 ));
401 }
402 Handshake::LiteSetup {
403 session,
404 version,
405 client_setup,
406 } => {
407 let our_setup = lite::Setup {
409 probe: lite::ProbeLevel::Report,
410 path: None,
411 role: None,
412 cost: None,
414 };
415 let start = lite::start(
416 session.clone(),
417 None,
418 publish,
419 subscribe,
420 version,
421 our_setup,
422 Some(client_setup),
423 )?;
424 return Ok(Session::new(
425 session,
426 version.into(),
427 start.recv_bandwidth,
428 start.driver,
429 ));
430 }
431 Handshake::Legacy {
432 session,
433 stream,
434 version,
435 request_id_max,
436 } => (session, stream, version, request_id_max),
437 };
438
439 let parameters = match version {
441 Version::Ietf(v) => {
442 let mut parameters = ietf::Parameters::default();
443 parameters.set_varint(ietf::ParameterVarInt::MaxRequestId, u32::MAX as u64);
444 parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec());
445 parameters.encode_bytes(v)?
446 }
447 Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?,
448 };
449
450 let server_setup = setup::Server {
451 version: version.into(),
452 parameters,
453 };
454 stream.writer.encode(&server_setup).await?;
455
456 let (recv_bw, protocol) = match version {
457 Version::Lite(v) => {
458 let stream = stream.with_version(v);
459 let start = lite::start(
461 session.clone(),
462 Some(stream),
463 publish,
464 subscribe,
465 v,
466 lite::Setup::default(),
467 None,
468 )?;
469 (start.recv_bandwidth, start.driver)
470 }
471 Version::Ietf(v) => {
472 let stream = stream.with_version(v);
473 let protocol = ietf::start(
475 session.clone(),
476 Some(stream),
477 request_id_max,
478 false,
479 publish,
480 subscribe,
481 v,
482 None,
483 None,
484 )?;
485 (None, protocol)
486 }
487 };
488
489 Ok(Session::new(session, version, recv_bw, protocol))
490 }
491
492 pub fn close(mut self, err: Error) {
494 let inner = self.inner.take().expect("request already responded");
495 inner.close(err);
496 }
497}
498
499impl<S: web_transport_trait::Session> RequestInner<S> {
500 fn close(self, err: Error) {
501 let session = match self.handshake {
502 Handshake::IetfModern { session, .. } => session,
503 Handshake::LiteBare { session, .. } => session,
504 Handshake::Legacy { session, .. } => session,
505 Handshake::LiteSetup { session, .. } => session,
506 };
507 session.close(err.to_code(), &err.to_string());
508 }
509}
510
511impl<S: web_transport_trait::Session> Drop for Request<S> {
512 fn drop(&mut self) {
515 if let Some(inner) = self.inner.take() {
516 tracing::warn!("Request dropped without ok() or close(); rejecting the session");
517 inner.close(Error::Cancel);
518 }
519 }
520}
521
522#[cfg(test)]
523mod tests {
524 use super::*;
525 use std::{
526 collections::VecDeque,
527 sync::{Arc, Mutex},
528 };
529
530 use crate::ALPN_LITE_05;
531 use bytes::Bytes;
532
533 #[derive(Debug, Clone, Default)]
534 struct FakeError;
535 impl std::fmt::Display for FakeError {
536 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
537 write!(f, "fake transport error")
538 }
539 }
540 impl std::error::Error for FakeError {}
541 impl web_transport_trait::Error for FakeError {
542 fn session_error(&self) -> Option<(u32, String)> {
543 Some((0, "closed".to_string()))
544 }
545 }
546
547 #[derive(Clone)]
550 struct FakeSession {
551 protocol: Option<&'static str>,
552 uni: Arc<Mutex<VecDeque<Vec<u8>>>>,
553 }
554
555 impl FakeSession {
556 fn new(protocol: &'static str, uni: impl IntoIterator<Item = Vec<u8>>) -> Self {
557 Self {
558 protocol: Some(protocol),
559 uni: Arc::new(Mutex::new(uni.into_iter().collect())),
560 }
561 }
562 }
563
564 impl web_transport_trait::Session for FakeSession {
565 type SendStream = FakeSend;
566 type RecvStream = FakeRecv;
567 type Error = FakeError;
568
569 async fn accept_uni(&self) -> Result<Self::RecvStream, Self::Error> {
570 let data = self.uni.lock().unwrap().pop_front();
572 match data {
573 Some(data) => Ok(FakeRecv { data: data.into() }),
574 None => std::future::pending().await,
575 }
576 }
577 async fn accept_bi(&self) -> Result<(Self::SendStream, Self::RecvStream), Self::Error> {
578 std::future::pending().await
579 }
580 async fn open_bi(&self) -> Result<(Self::SendStream, Self::RecvStream), Self::Error> {
581 std::future::pending().await
582 }
583 async fn open_uni(&self) -> Result<Self::SendStream, Self::Error> {
584 std::future::pending().await
585 }
586 fn send_datagram(&self, _payload: Bytes) -> Result<(), Self::Error> {
587 Ok(())
588 }
589 async fn recv_datagram(&self) -> Result<Bytes, Self::Error> {
590 std::future::pending().await
591 }
592 fn max_datagram_size(&self) -> usize {
593 1200
594 }
595 fn protocol(&self) -> Option<&str> {
596 self.protocol
597 }
598 fn close(&self, _code: u32, _reason: &str) {}
599 async fn closed(&self) -> Self::Error {
600 std::future::pending().await
601 }
602 }
603
604 #[derive(Clone, Default)]
605 struct FakeSend;
606 impl web_transport_trait::SendStream for FakeSend {
607 type Error = FakeError;
608 async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
609 Ok(buf.len())
610 }
611 fn set_priority(&mut self, _order: u8) {}
612 fn finish(&mut self) -> Result<(), Self::Error> {
613 Ok(())
614 }
615 fn reset(&mut self, _code: u32) {}
616 async fn closed(&mut self) -> Result<(), Self::Error> {
617 Ok(())
618 }
619 }
620
621 struct FakeRecv {
622 data: VecDeque<u8>,
623 }
624 impl web_transport_trait::RecvStream for FakeRecv {
625 type Error = FakeError;
626 async fn read(&mut self, dst: &mut [u8]) -> Result<Option<usize>, Self::Error> {
627 if self.data.is_empty() {
628 return Ok(None);
629 }
630 let size = dst.len().min(self.data.len());
631 for slot in dst.iter_mut().take(size) {
632 *slot = self.data.pop_front().unwrap();
633 }
634 Ok(Some(size))
635 }
636 fn stop(&mut self, _code: u32) {}
637 async fn closed(&mut self) -> Result<(), Self::Error> {
638 Ok(())
639 }
640 }
641
642 fn lite05_setup(path: Option<&str>, role: Option<Role>) -> Vec<u8> {
644 let v = lite::Version::Lite05;
645 let mut buf = Vec::new();
646 lite::DataType::Setup.encode(&mut buf, v).unwrap();
647 lite::Setup {
648 probe: lite::ProbeLevel::None,
649 path: path.map(str::to_string),
650 role,
651 cost: None,
652 }
653 .encode(&mut buf, v)
654 .unwrap();
655 buf
656 }
657
658 fn ietf_setup(version: ietf::Version, path: Option<&str>) -> Vec<u8> {
661 let mut params = ietf::Parameters::default();
662 if let Some(path) = path {
663 params.set_bytes(ietf::ParameterBytes::Path, path.as_bytes().to_vec());
664 }
665 let parameters = params.encode_bytes(version).unwrap();
666
667 let mut buf = Vec::new();
668 setup::Setup { parameters }
669 .encode(&mut buf, crate::Version::Ietf(version))
670 .unwrap();
671 buf
672 }
673
674 #[tokio::test(start_paused = true)]
675 async fn accept_request_reads_ietf_path() {
676 for (alpn, version) in [
679 (ALPN_17, ietf::Version::Draft17),
680 (ALPN_18, ietf::Version::Draft18),
681 (ALPN_19, ietf::Version::Draft19),
682 ] {
683 let session = FakeSession::new(alpn, [ietf_setup(version, Some("/team/room"))]);
684 let request = Server::new().accept_request(session).await.unwrap();
685 assert_eq!(request.path(), "/team/room", "{alpn}");
686 }
687 }
688
689 #[tokio::test(start_paused = true)]
690 async fn accept_request_ietf_without_path_is_empty() {
691 let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, None)]);
692 let request = Server::new().accept_request(session).await.unwrap();
693 assert_eq!(request.path(), "");
694 }
695
696 #[tokio::test(start_paused = true)]
697 async fn accept_request_ietf_empty_path_is_accepted() {
698 let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, Some(""))]);
699 let request = Server::new().accept_request(session).await.unwrap();
700 assert_eq!(request.path(), "");
701 }
702
703 fn lite05_group() -> Vec<u8> {
705 let mut buf = Vec::new();
706 lite::DataType::Group.encode(&mut buf, lite::Version::Lite05).unwrap();
707 buf
708 }
709
710 #[tokio::test(start_paused = true)]
711 async fn accept_request_reads_lite05_path() {
712 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/team/room"), None)]);
713 let request = Server::new().accept_request(session).await.unwrap();
714 assert_eq!(request.path(), "/team/room");
715 assert_eq!(request.role(), None, "a client that omits the role is bidirectional");
716 }
717
718 #[tokio::test(start_paused = true)]
719 async fn accept_request_lite05_without_path_is_empty() {
720 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None)]);
721 let request = Server::new().accept_request(session).await.unwrap();
722 assert_eq!(request.path(), "");
723 }
724
725 #[tokio::test(start_paused = true)]
726 async fn accept_request_lite05_empty_path_is_accepted() {
727 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some(""), None)]);
730 let request = Server::new().accept_request(session).await.unwrap();
731 assert_eq!(request.path(), "");
732 }
733
734 #[tokio::test(start_paused = true)]
735 async fn accept_request_reads_lite05_role() {
736 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/team/room"), Some(Role::Publisher))]);
737 let request = Server::new().accept_request(session).await.unwrap();
738 assert_eq!(request.role(), Some(Role::Publisher));
739 }
740
741 #[tokio::test(start_paused = true)]
742 async fn accept_request_skips_uni_stream_before_setup() {
743 let session = FakeSession::new(ALPN_LITE_05, [lite05_group(), lite05_setup(Some("/team/room"), None)]);
746 let request = Server::new().accept_request(session).await.unwrap();
747 assert_eq!(request.path(), "/team/room");
748 }
749}