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 origin: None,
416 };
417 let start = lite::start(
418 session.clone(),
419 None,
420 publish,
421 subscribe,
422 version,
423 our_setup,
424 Some(client_setup),
425 )?;
426 return Ok(Session::new(
427 session,
428 version.into(),
429 start.recv_bandwidth,
430 start.driver,
431 ));
432 }
433 Handshake::Legacy {
434 session,
435 stream,
436 version,
437 request_id_max,
438 } => (session, stream, version, request_id_max),
439 };
440
441 let parameters = match version {
443 Version::Ietf(v) => {
444 let mut parameters = ietf::Parameters::default();
445 parameters.set_varint(ietf::ParameterVarInt::MaxRequestId, u32::MAX as u64);
446 parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec());
447 parameters.encode_bytes(v)?
448 }
449 Version::Lite(v) => lite::Parameters::default().encode_bytes(v)?,
450 };
451
452 let server_setup = setup::Server {
453 version: version.into(),
454 parameters,
455 };
456 stream.writer.encode(&server_setup).await?;
457
458 let (recv_bw, protocol) = match version {
459 Version::Lite(v) => {
460 let stream = stream.with_version(v);
461 let start = lite::start(
463 session.clone(),
464 Some(stream),
465 publish,
466 subscribe,
467 v,
468 lite::Setup::default(),
469 None,
470 )?;
471 (start.recv_bandwidth, start.driver)
472 }
473 Version::Ietf(v) => {
474 let stream = stream.with_version(v);
475 let protocol = ietf::start(
477 session.clone(),
478 Some(stream),
479 request_id_max,
480 false,
481 publish,
482 subscribe,
483 v,
484 None,
485 None,
486 )?;
487 (None, protocol)
488 }
489 };
490
491 Ok(Session::new(session, version, recv_bw, protocol))
492 }
493
494 pub fn close(mut self, err: Error) {
496 let inner = self.inner.take().expect("request already responded");
497 inner.close(err);
498 }
499}
500
501impl<S: web_transport_trait::Session> RequestInner<S> {
502 fn close(self, err: Error) {
503 let session = match self.handshake {
504 Handshake::IetfModern { session, .. } => session,
505 Handshake::LiteBare { session, .. } => session,
506 Handshake::Legacy { session, .. } => session,
507 Handshake::LiteSetup { session, .. } => session,
508 };
509 session.close(err.to_code(), &err.to_string());
510 }
511}
512
513impl<S: web_transport_trait::Session> Drop for Request<S> {
514 fn drop(&mut self) {
517 if let Some(inner) = self.inner.take() {
518 tracing::warn!("Request dropped without ok() or close(); rejecting the session");
519 inner.close(Error::Cancel);
520 }
521 }
522}
523
524#[cfg(test)]
525mod tests {
526 use super::*;
527 use std::{
528 collections::VecDeque,
529 sync::{Arc, Mutex},
530 };
531
532 use crate::ALPN_LITE_05;
533 use bytes::Bytes;
534
535 #[derive(Debug, Clone, Default)]
536 struct FakeError;
537 impl std::fmt::Display for FakeError {
538 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
539 write!(f, "fake transport error")
540 }
541 }
542 impl std::error::Error for FakeError {}
543 impl web_transport_trait::Error for FakeError {
544 fn session_error(&self) -> Option<(u32, String)> {
545 Some((0, "closed".to_string()))
546 }
547 }
548
549 #[derive(Clone)]
552 struct FakeSession {
553 protocol: Option<&'static str>,
554 uni: Arc<Mutex<VecDeque<Vec<u8>>>>,
555 }
556
557 impl FakeSession {
558 fn new(protocol: &'static str, uni: impl IntoIterator<Item = Vec<u8>>) -> Self {
559 Self {
560 protocol: Some(protocol),
561 uni: Arc::new(Mutex::new(uni.into_iter().collect())),
562 }
563 }
564 }
565
566 impl web_transport_trait::Session for FakeSession {
567 type SendStream = FakeSend;
568 type RecvStream = FakeRecv;
569 type Error = FakeError;
570
571 async fn accept_uni(&self) -> Result<Self::RecvStream, Self::Error> {
572 let data = self.uni.lock().unwrap().pop_front();
574 match data {
575 Some(data) => Ok(FakeRecv { data: data.into() }),
576 None => std::future::pending().await,
577 }
578 }
579 async fn accept_bi(&self) -> Result<(Self::SendStream, Self::RecvStream), Self::Error> {
580 std::future::pending().await
581 }
582 async fn open_bi(&self) -> Result<(Self::SendStream, Self::RecvStream), Self::Error> {
583 std::future::pending().await
584 }
585 async fn open_uni(&self) -> Result<Self::SendStream, Self::Error> {
586 std::future::pending().await
587 }
588 fn send_datagram(&self, _payload: Bytes) -> Result<(), Self::Error> {
589 Ok(())
590 }
591 async fn recv_datagram(&self) -> Result<Bytes, Self::Error> {
592 std::future::pending().await
593 }
594 fn max_datagram_size(&self) -> usize {
595 1200
596 }
597 fn protocol(&self) -> Option<&str> {
598 self.protocol
599 }
600 fn close(&self, _code: u32, _reason: &str) {}
601 async fn closed(&self) -> Self::Error {
602 std::future::pending().await
603 }
604 }
605
606 #[derive(Clone, Default)]
607 struct FakeSend;
608 impl web_transport_trait::SendStream for FakeSend {
609 type Error = FakeError;
610 async fn write(&mut self, buf: &[u8]) -> Result<usize, Self::Error> {
611 Ok(buf.len())
612 }
613 fn set_priority(&mut self, _order: u8) {}
614 fn finish(&mut self) -> Result<(), Self::Error> {
615 Ok(())
616 }
617 fn reset(&mut self, _code: u32) {}
618 async fn closed(&mut self) -> Result<(), Self::Error> {
619 Ok(())
620 }
621 }
622
623 struct FakeRecv {
624 data: VecDeque<u8>,
625 }
626 impl web_transport_trait::RecvStream for FakeRecv {
627 type Error = FakeError;
628 async fn read(&mut self, dst: &mut [u8]) -> Result<Option<usize>, Self::Error> {
629 if self.data.is_empty() {
630 return Ok(None);
631 }
632 let size = dst.len().min(self.data.len());
633 for slot in dst.iter_mut().take(size) {
634 *slot = self.data.pop_front().unwrap();
635 }
636 Ok(Some(size))
637 }
638 fn stop(&mut self, _code: u32) {}
639 async fn closed(&mut self) -> Result<(), Self::Error> {
640 Ok(())
641 }
642 }
643
644 fn lite05_setup(path: Option<&str>, role: Option<Role>) -> Vec<u8> {
646 let v = lite::Version::Lite05;
647 let mut buf = Vec::new();
648 lite::DataType::Setup.encode(&mut buf, v).unwrap();
649 lite::Setup {
650 probe: lite::ProbeLevel::None,
651 path: path.map(str::to_string),
652 role,
653 cost: None,
654 origin: None,
655 }
656 .encode(&mut buf, v)
657 .unwrap();
658 buf
659 }
660
661 fn ietf_setup(version: ietf::Version, path: Option<&str>) -> Vec<u8> {
664 let mut params = ietf::Parameters::default();
665 if let Some(path) = path {
666 params.set_bytes(ietf::ParameterBytes::Path, path.as_bytes().to_vec());
667 }
668 let parameters = params.encode_bytes(version).unwrap();
669
670 let mut buf = Vec::new();
671 setup::Setup { parameters }
672 .encode(&mut buf, crate::Version::Ietf(version))
673 .unwrap();
674 buf
675 }
676
677 #[tokio::test(start_paused = true)]
678 async fn accept_request_reads_ietf_path() {
679 for (alpn, version) in [
682 (ALPN_17, ietf::Version::Draft17),
683 (ALPN_18, ietf::Version::Draft18),
684 (ALPN_19, ietf::Version::Draft19),
685 ] {
686 let session = FakeSession::new(alpn, [ietf_setup(version, Some("/team/room"))]);
687 let request = Server::new().accept_request(session).await.unwrap();
688 assert_eq!(request.path(), "/team/room", "{alpn}");
689 }
690 }
691
692 #[tokio::test(start_paused = true)]
693 async fn accept_request_ietf_without_path_is_empty() {
694 let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, None)]);
695 let request = Server::new().accept_request(session).await.unwrap();
696 assert_eq!(request.path(), "");
697 }
698
699 #[tokio::test(start_paused = true)]
700 async fn accept_request_ietf_empty_path_is_accepted() {
701 let session = FakeSession::new(ALPN_19, [ietf_setup(ietf::Version::Draft19, Some(""))]);
702 let request = Server::new().accept_request(session).await.unwrap();
703 assert_eq!(request.path(), "");
704 }
705
706 fn lite05_group() -> Vec<u8> {
708 let mut buf = Vec::new();
709 lite::DataType::Group.encode(&mut buf, lite::Version::Lite05).unwrap();
710 buf
711 }
712
713 #[tokio::test(start_paused = true)]
714 async fn accept_request_reads_lite05_path() {
715 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/team/room"), None)]);
716 let request = Server::new().accept_request(session).await.unwrap();
717 assert_eq!(request.path(), "/team/room");
718 assert_eq!(request.role(), None, "a client that omits the role is bidirectional");
719 }
720
721 #[tokio::test(start_paused = true)]
722 async fn accept_request_lite05_without_path_is_empty() {
723 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(None, None)]);
724 let request = Server::new().accept_request(session).await.unwrap();
725 assert_eq!(request.path(), "");
726 }
727
728 #[tokio::test(start_paused = true)]
729 async fn accept_request_lite05_empty_path_is_accepted() {
730 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some(""), None)]);
733 let request = Server::new().accept_request(session).await.unwrap();
734 assert_eq!(request.path(), "");
735 }
736
737 #[tokio::test(start_paused = true)]
738 async fn accept_request_reads_lite05_role() {
739 let session = FakeSession::new(ALPN_LITE_05, [lite05_setup(Some("/team/room"), Some(Role::Publisher))]);
740 let request = Server::new().accept_request(session).await.unwrap();
741 assert_eq!(request.role(), Some(Role::Publisher));
742 }
743
744 #[tokio::test(start_paused = true)]
745 async fn accept_request_skips_uni_stream_before_setup() {
746 let session = FakeSession::new(ALPN_LITE_05, [lite05_group(), lite05_setup(Some("/team/room"), None)]);
749 let request = Server::new().accept_request(session).await.unwrap();
750 assert_eq!(request.path(), "/team/room");
751 }
752}