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 origin: None,
88 inner: Some(RequestInner {
89 server: self.clone(),
90 handshake,
91 }),
92 };
93
94 let (encoding, supported) = match session.protocol() {
95 Some(ALPN_19) => {
96 self.versions
97 .select(Version::Ietf(ietf::Version::Draft19))
98 .ok_or(Error::Version)?;
99 return self.accept_ietf_modern(session, ietf::Version::Draft19).await;
100 }
101 Some(ALPN_18) => {
102 self.versions
103 .select(Version::Ietf(ietf::Version::Draft18))
104 .ok_or(Error::Version)?;
105 return self.accept_ietf_modern(session, ietf::Version::Draft18).await;
106 }
107 Some(ALPN_17) => {
108 self.versions
109 .select(Version::Ietf(ietf::Version::Draft17))
110 .ok_or(Error::Version)?;
111 return self.accept_ietf_modern(session, ietf::Version::Draft17).await;
112 }
113 Some(ALPN_16) => {
114 let v = self
115 .versions
116 .select(Version::Ietf(ietf::Version::Draft16))
117 .ok_or(Error::Version)?;
118 (v, v.into())
119 }
120 Some(ALPN_15) => {
121 let v = self
122 .versions
123 .select(Version::Ietf(ietf::Version::Draft15))
124 .ok_or(Error::Version)?;
125 (v, v.into())
126 }
127 Some(ALPN_14) => {
128 let v = self
129 .versions
130 .select(Version::Ietf(ietf::Version::Draft14))
131 .ok_or(Error::Version)?;
132 (v, v.into())
133 }
134 Some(alpn @ (ALPN_LITE_05 | ALPN_LITE_06_WIP)) => {
135 let version = match alpn {
136 ALPN_LITE_06_WIP => lite::Version::Lite06Wip,
137 _ => lite::Version::Lite05,
138 };
139 self.versions.select(Version::Lite(version)).ok_or(Error::Version)?;
140
141 let client_setup = lite::accept_setup(&session, version).await?;
145 return Ok(Request {
146 path: client_setup.path.clone(),
147 role: client_setup.role,
148 origin: client_setup.origin,
149 inner: Some(RequestInner {
150 server: self.clone(),
151 handshake: Handshake::LiteSetup {
152 session,
153 version,
154 client_setup,
155 },
156 }),
157 });
158 }
159 Some(ALPN_LITE_04) => {
160 self.versions
161 .select(Version::Lite(lite::Version::Lite04))
162 .ok_or(Error::Version)?;
163 return Ok(deferred(Handshake::LiteBare {
164 session,
165 version: lite::Version::Lite04,
166 }));
167 }
168 Some(ALPN_LITE_03) => {
169 self.versions
170 .select(Version::Lite(lite::Version::Lite03))
171 .ok_or(Error::Version)?;
172 return Ok(deferred(Handshake::LiteBare {
173 session,
174 version: lite::Version::Lite03,
175 }));
176 }
177 Some(ALPN_LITE) | None => {
178 let supported = self.versions.filter(&NEGOTIATED.into()).ok_or(Error::Version)?;
179 (Version::Ietf(ietf::Version::Draft14), supported)
180 }
181 Some(p) => return Err(Error::UnknownAlpn(p.to_string())),
182 };
183
184 let mut stream = Stream::accept(&session, encoding).await?;
187 let mut client: setup::Client = stream.reader.decode().await?;
188
189 let version = client
190 .versions
191 .iter()
192 .flat_map(|v| Version::try_from(*v).ok())
193 .find(|v| supported.contains(v))
194 .ok_or(Error::Version)?;
195
196 let (path, request_id_max) = match version {
200 Version::Ietf(v) => {
201 let params = ietf::Parameters::decode(&mut client.parameters, v)?;
202 let path = match params.get_bytes(ietf::ParameterBytes::Path) {
203 Some(bytes) => Some(
204 std::str::from_utf8(bytes)
205 .map_err(|_| Error::Decode(crate::DecodeError::InvalidValue))?
206 .to_owned(),
207 ),
208 None => None,
209 };
210 let request_id_max = params
211 .get_varint(ietf::ParameterVarInt::MaxRequestId)
212 .map(ietf::RequestId);
213 (path, request_id_max)
214 }
215 Version::Lite(_) => (None, None),
216 };
217
218 Ok(Request {
219 path,
220 role: None,
221 origin: None,
222 inner: Some(RequestInner {
223 server: self.clone(),
224 handshake: Handshake::Legacy {
225 session,
226 stream,
227 version,
228 request_id_max,
229 },
230 }),
231 })
232 }
233
234 async fn accept_ietf_modern<S: web_transport_trait::Session>(
237 &self,
238 session: S,
239 version: ietf::Version,
240 ) -> Result<Request<S>, Error> {
241 let (peer_setup, path) = ietf::accept_setup(&session, version).await?;
242 Ok(Request {
243 path,
244 role: None,
245 origin: None,
246 inner: Some(RequestInner {
247 server: self.clone(),
248 handshake: Handshake::IetfModern {
249 session,
250 version,
251 peer_setup,
252 },
253 }),
254 })
255 }
256}
257
258pub struct Request<S: web_transport_trait::Session> {
266 path: Option<String>,
267 role: Option<Role>,
268 origin: Option<crate::Origin>,
269 inner: Option<RequestInner<S>>,
271}
272
273struct RequestInner<S: web_transport_trait::Session> {
275 server: Server,
276 handshake: Handshake<S>,
277}
278
279enum Handshake<S: web_transport_trait::Session> {
282 IetfModern {
286 session: S,
287 version: ietf::Version,
288 peer_setup: Reader<S::RecvStream, Version>,
289 },
290 LiteBare { session: S, version: lite::Version },
292 Legacy {
296 session: S,
297 stream: Stream<S, Version>,
298 version: Version,
299 request_id_max: Option<ietf::RequestId>,
300 },
301 LiteSetup {
304 session: S,
305 version: lite::Version,
306 client_setup: lite::Setup,
307 },
308}
309
310impl<S: web_transport_trait::Session> Request<S> {
311 pub fn path(&self) -> &str {
318 self.path.as_deref().unwrap_or("")
319 }
320
321 pub fn role(&self) -> Option<Role> {
330 self.role
331 }
332
333 pub fn peer_origin(&self) -> Option<crate::Origin> {
341 self.origin
342 }
343
344 pub fn with_publisher(mut self, publish: impl Consume<origin::Consumer>) -> Self {
347 self.inner_mut().server.publish = Some(publish.consume());
348 self
349 }
350
351 pub fn with_subscriber(mut self, subscribe: origin::Producer) -> Self {
353 self.inner_mut().server.subscribe = Some(subscribe);
354 self
355 }
356
357 pub fn with_stats(mut self, stats: stats::Session) -> Self {
360 self.inner_mut().server.stats = stats;
361 self
362 }
363
364 fn inner_mut(&mut self) -> &mut RequestInner<S> {
365 self.inner.as_mut().expect("request already responded")
366 }
367
368 pub async fn ok(mut self) -> Result<(Session, Driver), Error> {
371 let RequestInner { server, handshake } = self.inner.take().expect("request already responded");
372
373 let publish = server.publish.map(|origin| origin.with_stats(server.stats.clone()));
377 let subscribe = server.subscribe.map(|origin| origin.with_stats(server.stats.clone()));
378
379 let (session, mut stream, version, request_id_max) = match handshake {
380 Handshake::IetfModern {
381 session,
382 version,
383 peer_setup,
384 } => {
385 let protocol = ietf::start(
388 session.clone(),
389 None,
390 None,
391 false,
392 publish,
393 subscribe,
394 version,
395 None,
396 Some(peer_setup),
397 )?;
398 tracing::debug!(?version, "connected");
399 return Ok(Session::new(session, version.into(), None, protocol));
400 }
401 Handshake::LiteBare { session, version } => {
402 let start = lite::start(
403 session.clone(),
404 None,
405 publish,
406 subscribe,
407 version,
408 lite::Setup::default(),
409 None,
410 )?;
411 return Ok(Session::new(
412 session,
413 version.into(),
414 start.recv_bandwidth,
415 start.driver,
416 ));
417 }
418 Handshake::LiteSetup {
419 session,
420 version,
421 client_setup,
422 } => {
423 let our_setup = lite::Setup {
425 probe: lite::ProbeLevel::Report,
426 path: None,
427 role: None,
428 cost: None,
430 origin: None,
432 };
433 let start = lite::start(
434 session.clone(),
435 None,
436 publish,
437 subscribe,
438 version,
439 our_setup,
440 Some(client_setup),
441 )?;
442 return Ok(Session::new(
443 session,
444 version.into(),
445 start.recv_bandwidth,
446 start.driver,
447 ));
448 }
449 Handshake::Legacy {
450 session,
451 stream,
452 version,
453 request_id_max,
454 } => (session, stream, version, request_id_max),
455 };
456
457 let parameters = match version {
459 Version::Ietf(v) => {
460 let mut parameters = ietf::Parameters::default();
461 parameters.set_varint(ietf::ParameterVarInt::MaxRequestId, u32::MAX as u64);
462 parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec());
463 parameters.encode_bytes(v)?
464 }
465 Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?,
466 };
467
468 let server_setup = setup::Server {
469 version: version.into(),
470 parameters,
471 };
472 stream.writer.encode(&server_setup).await?;
473
474 let (recv_bw, protocol) = match version {
475 Version::Lite(v) => {
476 let stream = stream.with_version(v);
477 let start = lite::start(
479 session.clone(),
480 Some(stream),
481 publish,
482 subscribe,
483 v,
484 lite::Setup::default(),
485 None,
486 )?;
487 (start.recv_bandwidth, start.driver)
488 }
489 Version::Ietf(v) => {
490 let stream = stream.with_version(v);
491 let protocol = ietf::start(
493 session.clone(),
494 Some(stream),
495 request_id_max,
496 false,
497 publish,
498 subscribe,
499 v,
500 None,
501 None,
502 )?;
503 (None, protocol)
504 }
505 };
506
507 Ok(Session::new(session, version, recv_bw, protocol))
508 }
509
510 pub fn close(mut self, err: Error) {
512 let inner = self.inner.take().expect("request already responded");
513 inner.close(err);
514 }
515}
516
517impl<S: web_transport_trait::Session> RequestInner<S> {
518 fn close(self, err: Error) {
519 let session = match self.handshake {
520 Handshake::IetfModern { session, .. } => session,
521 Handshake::LiteBare { session, .. } => session,
522 Handshake::Legacy { session, .. } => session,
523 Handshake::LiteSetup { session, .. } => session,
524 };
525 session.close(err.to_code(), &err.to_string());
526 }
527}
528
529impl<S: web_transport_trait::Session> Drop for Request<S> {
530 fn drop(&mut self) {
533 if let Some(inner) = self.inner.take() {
534 tracing::warn!("Request dropped without ok() or close(); rejecting the session");
535 inner.close(Error::Cancel);
536 }
537 }
538}
539
540#[cfg(test)]
541mod tests {
542 use super::*;
543 use crate::Origin;
544 use std::{
545 collections::VecDeque,
546 sync::{Arc, Mutex},
547 };
548
549 use crate::ALPN_LITE_05;
550 use bytes::Bytes;
551
552 #[derive(Debug, Clone, Default)]
553 struct FakeError;
554 impl std::fmt::Display for FakeError {
555 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
556 write!(f, "fake transport error")
557 }
558 }
559 impl std::error::Error for FakeError {}
560 impl web_transport_trait::Error for FakeError {
561 fn session_error(&self) -> Option<(u32, String)> {
562 Some((0, "closed".to_string()))
563 }
564 }
565
566 #[derive(Clone)]
569 struct FakeSession {
570 protocol: Option<&'static str>,
571 uni: Arc<Mutex<VecDeque<Vec<u8>>>>,
572 }
573
574 impl FakeSession {
575 fn new(protocol: &'static str, uni: impl IntoIterator<Item = Vec<u8>>) -> Self {
576 Self {
577 protocol: Some(protocol),
578 uni: Arc::new(Mutex::new(uni.into_iter().collect())),
579 }
580 }
581 }
582
583 impl web_transport_trait::Session for FakeSession {
584 type SendStream = FakeSend;
585 type RecvStream = FakeRecv;
586 type Error = FakeError;
587
588 async fn accept_uni(&self) -> Result<Self::RecvStream, Self::Error> {
589 let data = self.uni.lock().unwrap().pop_front();
591 match data {
592 Some(data) => Ok(FakeRecv { data: data.into() }),
593 None => std::future::pending().await,
594 }
595 }
596 async fn accept_bi(&self) -> Result<(Self::SendStream, Self::RecvStream), Self::Error> {
597 std::future::pending().await
598 }
599 async fn open_bi(&self) -> Result<(Self::SendStream, Self::RecvStream), Self::Error> {
600 std::future::pending().await
601 }
602 async fn open_uni(&self) -> Result<Self::SendStream, Self::Error> {
603 std::future::pending().await
604 }
605 fn send_datagram(&self, _payload: Bytes) -> Result<(), Self::Error> {
606 Ok(())
607 }
608 async fn recv_datagram(&self) -> Result<Bytes, Self::Error> {
609 std::future::pending().await
610 }
611 fn max_datagram_size(&self) -> usize {
612 1200
613 }
614 fn protocol(&self) -> Option<&str> {
615 self.protocol
616 }
617 fn close(&self, _code: u32, _reason: &str) {}
618 async fn closed(&self) -> Self::Error {
619 std::future::pending().await
620 }
621 }
622
623 #[derive(Clone, Default)]
624 struct FakeSend;
625 impl web_transport_trait::SendStream for FakeSend {
626 type Error = FakeError;
627 async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
628 Ok(buf.len())
629 }
630 fn set_priority(&mut self, _order: u8) {}
631 fn finish(&mut self) -> Result<(), Self::Error> {
632 Ok(())
633 }
634 fn reset(&mut self, _code: u32) {}
635 async fn closed(&mut self) -> Result<(), Self::Error> {
636 Ok(())
637 }
638 }
639
640 struct FakeRecv {
641 data: VecDeque<u8>,
642 }
643 impl web_transport_trait::RecvStream for FakeRecv {
644 type Error = FakeError;
645 async fn read(&mut self, dst: &mut [u8]) -> Result<Option<usize>, Self::Error> {
646 if self.data.is_empty() {
647 return Ok(None);
648 }
649 let size = dst.len().min(self.data.len());
650 for slot in dst.iter_mut().take(size) {
651 *slot = self.data.pop_front().unwrap();
652 }
653 Ok(Some(size))
654 }
655 fn stop(&mut self, _code: u32) {}
656 async fn closed(&mut self) -> Result<(), Self::Error> {
657 Ok(())
658 }
659 }
660
661 fn lite05_setup(path: Option<&str>, role: Option<Role>, origin: Option<Origin>) -> Vec<u8> {
663 let v = lite::Version::Lite05;
664 let mut buf = Vec::new();
665 lite::DataType::Setup.encode(&mut buf, v).unwrap();
666 lite::Setup {
667 probe: lite::ProbeLevel::None,
668 path: path.map(str::to_string),
669 role,
670 cost: None,
671 origin,
672 }
673 .encode(&mut buf, v)
674 .unwrap();
675 buf
676 }
677
678 fn ietf_setup(version: ietf::Version, path: Option<&str>) -> Vec<u8> {
681 let mut params = ietf::Parameters::default();
682 if let Some(path) = path {
683 params.set_bytes(ietf::ParameterBytes::Path, path.as_bytes().to_vec());
684 }
685 let parameters = params.encode_bytes(version).unwrap();
686
687 let mut buf = Vec::new();
688 setup::Setup { parameters }
689 .encode(&mut buf, crate::Version::Ietf(version))
690 .unwrap();
691 buf
692 }
693
694 #[tokio::test(start_paused = true)]
695 async fn accept_request_reads_ietf_path() {
696 for (alpn, version) in [
699 (ALPN_17, ietf::Version::Draft17),
700 (ALPN_18, ietf::Version::Draft18),
701 (ALPN_19, ietf::Version::Draft19),
702 ] {
703 let session = FakeSession::new(alpn, [ietf_setup(version, Some("/team/room"))]);
704 let request = Server::new().accept_request(session).await.unwrap();
705 assert_eq!(request.path(), "/team/room", "{alpn}");
706 }
707 }
708
709 #[tokio::test(start_paused = true)]
710 async fn accept_request_ietf_without_path_is_empty() {
711 let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, None)]);
712 let request = Server::new().accept_request(session).await.unwrap();
713 assert_eq!(request.path(), "");
714 }
715
716 #[tokio::test(start_paused = true)]
717 async fn accept_request_ietf_empty_path_is_accepted() {
718 let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, Some(""))]);
719 let request = Server::new().accept_request(session).await.unwrap();
720 assert_eq!(request.path(), "");
721 }
722
723 fn lite05_group() -> Vec<u8> {
725 let mut buf = Vec::new();
726 lite::DataType::Group.encode(&mut buf, lite::Version::Lite05).unwrap();
727 buf
728 }
729
730 #[tokio::test(start_paused = true)]
731 async fn accept_request_reads_lite05_path() {
732 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/team/room"), None, None)]);
733 let request = Server::new().accept_request(session).await.unwrap();
734 assert_eq!(request.path(), "/team/room");
735 assert_eq!(request.role(), None, "a client that omits the role is bidirectional");
736 }
737
738 #[tokio::test(start_paused = true)]
739 async fn accept_request_lite05_without_path_is_empty() {
740 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, None)]);
741 let request = Server::new().accept_request(session).await.unwrap();
742 assert_eq!(request.path(), "");
743 }
744
745 #[tokio::test(start_paused = true)]
746 async fn accept_request_lite05_empty_path_is_accepted() {
747 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some(""), None, None)]);
750 let request = Server::new().accept_request(session).await.unwrap();
751 assert_eq!(request.path(), "");
752 }
753
754 #[tokio::test(start_paused = true)]
755 async fn accept_request_reads_lite05_role() {
756 let session = FakeSession::new(
757 ALPN_LITE_05,
758 [lite05_setup(Some("/team/room"), Some(Role::Publisher), None)],
759 );
760 let request = Server::new().accept_request(session).await.unwrap();
761 assert_eq!(request.role(), Some(Role::Publisher));
762 }
763
764 #[tokio::test(start_paused = true)]
765 async fn accept_request_skips_uni_stream_before_setup() {
766 let session = FakeSession::new(
769 ALPN_LITE_05,
770 [lite05_group(), lite05_setup(Some("/team/room"), None, None)],
771 );
772 let request = Server::new().accept_request(session).await.unwrap();
773 assert_eq!(request.path(), "/team/room");
774 }
775
776 #[tokio::test(start_paused = true)]
777 async fn accept_request_reads_lite05_peer_origin() {
778 let origin = Origin::new(42).unwrap();
779 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None, Some(origin))]);
780 let request = Server::new().accept_request(session).await.unwrap();
781 assert_eq!(request.peer_origin(), Some(origin));
782 }
783}