Fix the audio catalog, but disable audio for now.

.dockerignore

@@ -1,3 +1,2 @@
 target
 dev
-*.mp4

.github/workflows/main.yml

@@ -6,8 +6,7 @@ on:
 
 env:
   REGISTRY: docker.io
-  IMAGE: kixelated/moq-rs
+  IMAGE: ${{ github.repository }}
-  IMAGE-PUB: kixelated/moq-pub
   SERVICE: api # Restart the API service TODO and relays
 
 jobs:
@@ -43,16 +42,6 @@ jobs:
           tags: ${{env.REGISTRY}}/${{env.IMAGE}}
           platforms: linux/amd64,linux/arm64
 
-      # Same, but include ffmpeg for publishing BBB
-      - uses: depot/build-push-action@v1
-        with:
-          project: r257ctfqm6
-          context: .
-          push: true
-          target: moq-pub # instead of the default target
-          tags: ${{env.REGISTRY}}/${{env.IMAGE-PUB}}
-          platforms: linux/amd64,linux/arm64
-
       # Log in to GCP
       - uses: google-github-actions/auth@v1
         with:

.gitignore

@@ -1,4 +1,3 @@
 .DS_Store
 target/
 logs/
-*.mp4

Cargo.lock

@@ -1066,7 +1066,6 @@ dependencies = [
  "rfc6381-codec",
  "rustls",
  "rustls-native-certs",
- "rustls-pemfile",
  "serde_json",
  "tokio",
  "tracing",
@@ -1107,11 +1106,9 @@ dependencies = [
 name = "moq-transport"
 version = "0.2.0"
 dependencies = [
- "async-trait",
  "bytes",
  "indexmap 2.0.0",
  "log",
- "paste",
  "quinn",
  "thiserror",
  "tokio",
@@ -1322,12 +1319,6 @@ dependencies = [
  "windows-targets",
 ]
 
-[[package]]
-name = "paste"
-version = "1.0.14"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "de3145af08024dea9fa9914f381a17b8fc6034dfb00f3a84013f7ff43f29ed4c"
-
 [[package]]
 name = "percent-encoding"
 version = "2.3.0"

Dockerfile

@@ -12,28 +12,14 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
     --mount=type=cache,target=/build/target \
     cargo build --release && cp /build/target/release/moq-* /usr/local/cargo/bin
 
-# Special image for moq-pub with ffmpeg and a publish script included.
+# Runtime image
-FROM rust:latest as moq-pub
+FROM rust:latest
-
-# Install required utilities and ffmpeg
-RUN apt-get update && \
-    apt-get install -y ffmpeg wget
-
-# Copy the publish script into the image
-COPY deploy/publish.sh /usr/local/bin/publish
-
-# Copy the compiled binary
-COPY --from=builder /usr/local/cargo/bin/moq-pub /usr/local/cargo/bin/moq-pub
-CMD [ "publish" ]
-
-# moq-rs image with just the binaries
-FROM rust:latest as moq-rs
 
 LABEL org.opencontainers.image.source=https://github.com/kixelated/moq-rs
 LABEL org.opencontainers.image.licenses="MIT OR Apache-2.0"
 
 # Fly.io entrypoint
-ADD deploy/fly-relay.sh .
+ADD fly-relay.sh .
 
 # Copy the compiled binaries
 COPY --from=builder /usr/local/cargo/bin /usr/local/cargo/bin

HACKATHON.md

@@ -1,53 +0,0 @@
-# Hackathon
-
-IETF Prague 118
-
-## MoqTransport
-
-Reference libraries are available at [moq-rs](https://github.com/kixelated/moq-rs) and [moq-js](https://github.com/kixelated/moq-js). The Rust library is [well documented](https://docs.rs/moq-transport/latest/moq_transport/) but the web library, not so much.
-
-**TODO** Update both to draft-01.
-**TODO** Switch any remaining forks over to extensions. ex: track_id in SUBSCRIBE
-
-The stream mapping right now is quite rigid: `stream == group == object`.
-
-**TODO** Support multiple objects per group. They MUST NOT use different priorities, different tracks, or out-of-order sequences.
-
-The API and cache aren't designed to send/receive arbitrary objects over arbitrary streams as specified in the draft. I don't think it should, and it wouldn't be possible to implement in time for the hackathon anyway.
-
-**TODO** Make an extension to enforce this stream mapping?
-
-## Generic Relay
-
-I'm hosting a simple CDN at: `relay.quic.video`
-
-The traffic is sharded based on the WebTransport path to avoid namespace collisions. Think of it like a customer ID, although it's completely unauthenticated for now. Use your username or whatever string you want: `CONNECT https://relay.quic.video/alan`.
-
-**TODO** Currently, it performs an implicit `ANNOUNCE ""` when `role=publisher`. This means there can only be a single publisher per shard and `role=both` is not supported. I should have explicit `ANNOUNCE` messages supported before the hackathon to remove this limitation.
-
-**TODO** I don't know if I will have subscribe hints fully working in time. They will be parsed but might be ignored.
-
-## CMAF Media
-
-You can [publish](https://quic.video/publish) and [watch](https://quic.video/watch) broadcasts.
-There's a [24/7 bunny stream](https://quic.video/watch/bbb) or you can publish your own using [moq-pub](https://github.com/kixelated/moq-rs/tree/main/moq-pub).
-
-If you want to fetch from the relay directly, the name of the broadcast is the path. For example, `https://quic.video/watch/bbb` can be accessed at `relay.quic.video/bbb`. 
-
-The namespace is empty and the catalog track is `.catalog`. I'm currently using simple JSON catalog with no support for delta updates.
-
-**TODO** update to the proposed [Warp catalog](https://datatracker.ietf.org/doc/draft-wilaw-moq-catalogformat/).
-
-The media tracks uses a single (unbounded) object per group. Video groups are per GoP, while audio groups are per frame. There's also an init track containing information required to initialize the decoder.
-
-**TODO** Base64 encode the init track in the catalog.
-
-
-## Clock
-
-**TODO** Host a clock demo that sends a group per second:
-
-```
-GROUP: YYYY-MM-DD HH:MM
-OBJECT: SS
-```

deploy/fly.toml

@@ -1,20 +0,0 @@
-app = "englishm-moq-relay"
-kill_signal = "SIGINT"
-kill_timeout = 5
-
-[env]
-PORT = "4443"
-
-[experimental]
-cmd = "./fly-relay.sh"
-
-[[services]]
-internal_port = 4443
-protocol = "udp"
-
-[services.concurrency]
-hard_limit = 25
-soft_limit = 20
-
-[[services.ports]]
-port = "4443"

deploy/publish.sh

@@ -1,41 +0,0 @@
-#!/bin/bash
-set -euo pipefail
-
-ADDR=${ADDR:-"https://relay.quic.video"}
-NAME=${NAME:-"bbb"}
-URL=${URL:-"http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4"}
-
-# Download the funny bunny
-wget -nv "${URL}" -O "${NAME}.mp4"
-
-# ffmpeg
-#   -hide_banner: Hide the banner
-#   -v quiet: and any other output
-#   -stats: But we still want some stats on stderr
-#   -stream_loop -1: Loop the broadcast an infinite number of times
-#   -re: Output in real-time
-#   -i "${INPUT}": Read from a file on disk
-#   -vf "drawtext": Render the current time in the corner of the video
-#   -an: Disable audio for now
-#   -b:v 3M: Output video at 3Mbps
-#   -preset ultrafast: Don't use much CPU at the cost of quality
-#   -tune zerolatency: Optimize for latency at the cost of quality
-#   -f mp4: Output to mp4 format
-#   -movflags: Build a fMP4 file with a frame per fragment
-# - | moq-pub: Output to stdout and moq-pub to publish
-
-# Run ffmpeg
-ffmpeg \
-	-stream_loop -1 \
-	-hide_banner \
-	-v quiet \
-	-re \
-    -i "${NAME}.mp4" \
-	-vf "drawtext=fontfile=/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf:text='%{gmtime\: %H\\\\\:%M\\\\\:%S.%3N}':x=(W-tw)-24:y=24:fontsize=48:fontcolor=white:box=1:boxcolor=black@0.5" \
-	-an \
-	-b:v 3M \
-	-preset ultrafast \
-	-tune zerolatency \
-    -f mp4 \
-	-movflags empty_moov+frag_every_frame+separate_moof+omit_tfhd_offset \
-	- | moq-pub "${ADDR}/${NAME}"

fly.toml

@@ -0,0 +1,19 @@
+app = "englishm-moq-relay"
+kill_signal = "SIGINT"
+kill_timeout = 5
+
+[env]
+  PORT = "4443"
+
+[experimental]
+  cmd = "./fly-relay.sh"
+
+[[services]]
+  internal_port = 4443
+  protocol = "udp"
+  [services.concurrency]
+    hard_limit = 25
+    soft_limit = 20
+
+  [[services.ports]]
+    port = "4443"

moq-pub/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "moq-pub"
 description = "Media over QUIC"
-authors = ["Mike English", "Luke Curley"]
+authors = ["Mike English"]
 repository = "https://github.com/kixelated/moq-rs"
 license = "MIT OR Apache-2.0"
 
@@ -23,9 +23,8 @@ webtransport-quinn = "0.6"
 url = "2"
 
 # Crypto
-rustls = { version = "0.21", features = ["dangerous_configuration"] }
+rustls = "0.21"
 rustls-native-certs = "0.6"
-rustls-pemfile = "1"
 
 # Async stuff
 tokio = { version = "1", features = ["full"] }

moq-pub/README.md

@@ -5,7 +5,7 @@ A command line tool for publishing media via Media over QUIC (MoQ).
 Expects to receive fragmented MP4 via standard input and connect to a MOQT relay.
 
 ```
-ffmpeg ... - | moq-pub https://localhost:4443
+ffmpeg ... - | moq-pub -i - --host localhost:4443
 ```
 
 ### Invoking `moq-pub`:
@@ -13,7 +13,7 @@ ffmpeg ... - | moq-pub https://localhost:4443
 Here's how I'm currently testing things, with a local copy of Big Buck Bunny named `bbb_source.mp4`:
 
 ```
-$ ffmpeg -hide_banner -v quiet -stream_loop -1 -re -i bbb_source.mp4 -an -f mp4 -movflags empty_moov+frag_every_frame+separate_moof+omit_tfhd_offset - | RUST_LOG=moq_pub=info moq-pub https://localhost:4443
+$ ffmpeg -hide_banner -v quiet -stream_loop -1 -re -i bbb_source.mp4 -an -f mp4 -movflags empty_moov+frag_every_frame+separate_moof+omit_tfhd_offset - | RUST_LOG=moq_pub=info moq-pub -i -
 ```
 
 This relies on having `moq-relay` (the relay server) already running locally in another shell.

moq-pub/src/cli.rs

@@ -1,5 +1,5 @@
 use clap::Parser;
-use std::{net, path};
+use std::net;
 use url::Url;
 
 #[derive(Parser, Clone, Debug)]
@@ -21,19 +21,6 @@ pub struct Config {
 	/// Connect to the given URL starting with https://
 	#[arg(value_parser = moq_url)]
 	pub url: Url,
-
-	/// Use the TLS root CA at this path, encoded as PEM.
-	///
-	/// This value can be provided multiple times for multiple roots.
-	/// If this is empty, system roots will be used instead
-	#[arg(long)]
-	pub tls_root: Vec<path::PathBuf>,
-
-	/// Danger: Disable TLS certificate verification.
-	///
-	/// Fine for local development, but should be used in caution in production.
-	#[arg(long)]
-	pub tls_disable_verify: bool,
 }
 
 fn moq_url(s: &str) -> Result<Url, String> {

moq-pub/src/main.rs

@@ -1,5 +1,3 @@
-use std::{fs, io, sync::Arc, time};
-
 use anyhow::Context;
 use clap::Parser;
 
@@ -28,28 +26,10 @@ async fn main() -> anyhow::Result<()> {
 	let (publisher, subscriber) = broadcast::new("");
 	let mut media = Media::new(&config, publisher).await?;
 
-	// Create a list of acceptable root certificates.
+	// Ugh, just let me use my native root certs already
 	let mut roots = rustls::RootCertStore::empty();
-
+	for cert in rustls_native_certs::load_native_certs().expect("could not load platform certs") {
-	if config.tls_root.is_empty() {
+		roots.add(&rustls::Certificate(cert.0)).unwrap();
-		// Add the platform's native root certificates.
-		for cert in rustls_native_certs::load_native_certs().context("could not load platform certs")? {
-			roots
-				.add(&rustls::Certificate(cert.0))
-				.context("failed to add root cert")?;
-		}
-	} else {
-		// Add the specified root certificates.
-		for root in &config.tls_root {
-			let root = fs::File::open(root).context("failed to open root cert file")?;
-			let mut root = io::BufReader::new(root);
-
-			let root = rustls_pemfile::certs(&mut root).context("failed to read root cert")?;
-			anyhow::ensure!(root.len() == 1, "expected a single root cert");
-			let root = rustls::Certificate(root[0].to_owned());
-
-			roots.add(&root).context("failed to add root cert")?;
-		}
 	}
 
 	let mut tls_config = rustls::ClientConfig::builder()
@@ -57,12 +37,6 @@ async fn main() -> anyhow::Result<()> {
 		.with_root_certificates(roots)
 		.with_no_client_auth();
 
-	// Allow disabling TLS verification altogether.
-	if config.tls_disable_verify {
-		let noop = NoCertificateVerification {};
-		tls_config.dangerous().set_certificate_verifier(Arc::new(noop));
-	}
-
 	tls_config.alpn_protocols = vec![webtransport_quinn::ALPN.to_vec()]; // this one is important
 
 	let arc_tls_config = std::sync::Arc::new(tls_config);
@@ -89,19 +63,3 @@ async fn main() -> anyhow::Result<()> {
 
 	Ok(())
 }
-
-pub struct NoCertificateVerification {}
-
-impl rustls::client::ServerCertVerifier for NoCertificateVerification {
-	fn verify_server_cert(
-		&self,
-		_end_entity: &rustls::Certificate,
-		_intermediates: &[rustls::Certificate],
-		_server_name: &rustls::ServerName,
-		_scts: &mut dyn Iterator<Item = &[u8]>,
-		_ocsp_response: &[u8],
-		_now: time::SystemTime,
-	) -> Result<rustls::client::ServerCertVerified, rustls::Error> {
-		Ok(rustls::client::ServerCertVerified::assertion())
-	}
-}

moq-pub/src/media.rs

@@ -1,6 +1,6 @@
 use crate::cli::Config;
 use anyhow::{self, Context};
-use moq_transport::cache::{broadcast, fragment, segment, track};
+use moq_transport::cache::{broadcast, segment, track};
 use moq_transport::VarInt;
 use mp4::{self, ReadBox};
 use serde_json::json;
@@ -44,17 +44,11 @@ impl Media {
 		let mut init_track = broadcast.create_track("0.mp4")?;
 		let mut init_segment = init_track.create_segment(segment::Info {
 			sequence: VarInt::ZERO,
-			priority: 0,
+			priority: i32::MAX,
 			expires: None,
 		})?;
 
-		// Create a single fragment, optionally setting the size
+		init_segment.write_chunk(init.into())?;
-		let mut init_fragment = init_segment.create_fragment(fragment::Info {
-			sequence: VarInt::ZERO,
-			size: None, // size is only needed when we have multiple fragments.
-		})?;
-
-		init_fragment.write_chunk(init.into())?;
 
 		let mut tracks = HashMap::new();
 
@@ -134,7 +128,7 @@ impl Media {
 	) -> Result<(), anyhow::Error> {
 		let mut segment = track.create_segment(segment::Info {
 			sequence: VarInt::ZERO,
-			priority: 0,
+			priority: i32::MAX,
 			expires: None,
 		})?;
 
@@ -217,14 +211,8 @@ impl Media {
 		let catalog_str = serde_json::to_string_pretty(&catalog)?;
 		log::info!("catalog: {}", catalog_str);
 
-		// Create a single fragment for the segment.
-		let mut fragment = segment.create_fragment(fragment::Info {
-			sequence: VarInt::ZERO,
-			size: None, // Size is only needed when we have multiple fragments.
-		})?;
-
 		// Add the segment and add the fragment.
-		fragment.write_chunk(catalog_str.into())?;
+		segment.write_chunk(catalog_str.into())?;
 
 		Ok(())
 	}
@@ -272,7 +260,7 @@ struct Track {
 	track: track::Publisher,
 
 	// The current segment
-	current: Option<fragment::Publisher>,
+	segment: Option<segment::Publisher>,
 
 	// The number of units per second.
 	timescale: u64,
@@ -286,16 +274,16 @@ impl Track {
 		Self {
 			track,
 			sequence: 0,
-			current: None,
+			segment: None,
 			timescale,
 		}
 	}
 
 	pub fn header(&mut self, raw: Vec<u8>, fragment: Fragment) -> anyhow::Result<()> {
-		if let Some(current) = self.current.as_mut() {
+		if let Some(segment) = self.segment.as_mut() {
 			if !fragment.keyframe {
 				// Use the existing segment
-				current.write_chunk(raw.into())?;
+				segment.write_chunk(raw.into())?;
 				return Ok(());
 			}
 		}
@@ -304,7 +292,7 @@ impl Track {
 
 		// Compute the timestamp in milliseconds.
 		// Overflows after 583 million years, so we're fine.
-		let timestamp: u32 = fragment
+		let timestamp: i32 = fragment
 			.timestamp(self.timescale)
 			.as_millis()
 			.try_into()
@@ -313,34 +301,26 @@ impl Track {
 		// Create a new segment.
 		let mut segment = self.track.create_segment(segment::Info {
 			sequence: VarInt::try_from(self.sequence).context("sequence too large")?,
-
+			priority: timestamp, // newer segments are higher priority
-			// Newer segments are higher priority
-			priority: u32::MAX.checked_sub(timestamp).context("priority too large")?,
 
 			// Delete segments after 10s.
 			expires: Some(time::Duration::from_secs(10)),
 		})?;
 
-		// Create a single fragment for the segment that we will keep appending.
-		let mut fragment = segment.create_fragment(fragment::Info {
-			sequence: VarInt::ZERO,
-			size: None,
-		})?;
-
 		self.sequence += 1;
 
 		// Insert the raw atom into the segment.
-		fragment.write_chunk(raw.into())?;
+		segment.write_chunk(raw.into())?;
 
 		// Save for the next iteration
-		self.current = Some(fragment);
+		self.segment = Some(segment);
 
 		Ok(())
 	}
 
 	pub fn data(&mut self, raw: Vec<u8>) -> anyhow::Result<()> {
-		let fragment = self.current.as_mut().context("missing current fragment")?;
+		let segment = self.segment.as_mut().context("missing segment")?;
-		fragment.write_chunk(raw.into())?;
+		segment.write_chunk(raw.into())?;
 
 		Ok(())
 	}

moq-relay/Cargo.toml

@@ -23,7 +23,7 @@ url = "2"
 
 # Crypto
 ring = "0.16"
-rustls = { version = "0.21", features = ["dangerous_configuration"] }
+rustls = "0.21"
 rustls-pemfile = "1"
 rustls-native-certs = "0.6"
 webpki = "0.22"

moq-relay/src/config.rs

@@ -31,12 +31,6 @@ pub struct Config {
 	#[arg(long)]
 	pub tls_root: Vec<path::PathBuf>,
 
-	/// Danger: Disable TLS certificate verification.
-	///
-	/// Fine for local development and between relays, but should be used in caution in production.
-	#[arg(long)]
-	pub tls_disable_verify: bool,
-
 	/// Optional: Use the moq-api via HTTP to store origin information.
 	#[arg(long)]
 	pub api: Option<Url>,

moq-relay/src/error.rs

@@ -37,15 +37,15 @@ impl moq_transport::MoqError for RelayError {
 		}
 	}
 
-	fn reason(&self) -> String {
+	fn reason(&self) -> &str {
 		match self {
-			Self::Transport(err) => format!("transport error: {}", err.reason()),
+			Self::Transport(err) => err.reason(),
-			Self::Cache(err) => format!("cache error: {}", err.reason()),
+			Self::Cache(err) => err.reason(),
-			Self::MoqApi(err) => format!("api error: {}", err),
+			Self::MoqApi(_err) => "api error",
-			Self::Url(err) => format!("url error: {}", err),
+			Self::Url(_) => "url error",
-			Self::MissingNode => "missing node".to_owned(),
+			Self::MissingNode => "missing node",
-			Self::WebTransportServer(err) => format!("upstream server error: {}", err),
+			Self::WebTransportServer(_) => "server error",
-			Self::WebTransportClient(err) => format!("upstream client error: {}", err),
+			Self::WebTransportClient(_) => "upstream error",
 		}
 	}
 }

moq-relay/src/quic.rs

@@ -19,8 +19,8 @@ pub struct Quic {
 impl Quic {
 	// Create a QUIC endpoint that can be used for both clients and servers.
 	pub async fn new(config: Config, tls: Tls) -> anyhow::Result<Self> {
-		let mut client_config = tls.client.clone();
+		let mut client_config = tls.client();
-		let mut server_config = tls.server.clone();
+		let mut server_config = tls.server();
 		client_config.alpn_protocols = vec![webtransport_quinn::ALPN.to_vec()];
 		server_config.alpn_protocols = vec![webtransport_quinn::ALPN.to_vec()];
 

moq-relay/src/tls.rs

@@ -3,19 +3,23 @@ use ring::digest::{digest, SHA256};
 use rustls::server::{ClientHello, ResolvesServerCert};
 use rustls::sign::CertifiedKey;
 use rustls::{Certificate, PrivateKey, RootCertStore};
+use rustls::{ClientConfig, ServerConfig};
+use std::fs;
 use std::io::{self, Cursor, Read};
 use std::path;
 use std::sync::Arc;
-use std::{fs, time};
 use webpki::{DnsNameRef, EndEntityCert};
 
 use crate::Config;
 
 #[derive(Clone)]
 pub struct Tls {
-	pub server: rustls::ServerConfig,
+	// Support serving multiple certificates, choosing one that looks valid for the given SNI.
-	pub client: rustls::ClientConfig,
+	// We store the parsed certificate, and the certified cert/key that rustls expects
-	pub fingerprints: Vec<String>,
+	serve: Arc<ServeCerts>,
+
+	// Accept any cert that is trusted by the system's native trust store.
+	accept: Arc<RootCertStore>,
 }
 
 impl Tls {
@@ -52,34 +56,32 @@ impl Tls {
 			}
 		}
 
-		// Create the TLS configuration we'll use as a client (relay -> relay)
-		let mut client = rustls::ClientConfig::builder()
-			.with_safe_defaults()
-			.with_root_certificates(roots)
-			.with_no_client_auth();
-
-		// Allow disabling TLS verification altogether.
-		if config.tls_disable_verify {
-			let noop = NoCertificateVerification {};
-			client.dangerous().set_certificate_verifier(Arc::new(noop));
-		}
-
-		let fingerprints = serve.fingerprints();
-
-		// Create the TLS configuration we'll use as a server (relay <- browser)
-		let server = rustls::ServerConfig::builder()
-			.with_safe_defaults()
-			.with_no_client_auth()
-			.with_cert_resolver(Arc::new(serve));
-
 		let certs = Self {
-			server,
+			serve: Arc::new(serve),
-			client,
+			accept: Arc::new(roots),
-			fingerprints,
 		};
 
 		Ok(certs)
 	}
+
+	pub fn client(&self) -> ClientConfig {
+		rustls::ClientConfig::builder()
+			.with_safe_defaults()
+			.with_root_certificates(self.accept.clone())
+			.with_no_client_auth()
+	}
+
+	pub fn server(&self) -> ServerConfig {
+		rustls::ServerConfig::builder()
+			.with_safe_defaults()
+			.with_no_client_auth()
+			.with_cert_resolver(self.serve.clone())
+	}
+
+	// Return the SHA256 fingerprint of our certificates.
+	pub fn fingerprints(&self) -> Vec<String> {
+		self.serve.fingerprints()
+	}
 }
 
 #[derive(Default)]
@@ -164,19 +166,3 @@ impl ResolvesServerCert for ServeCerts {
 		self.list.last().cloned()
 	}
 }
-
-pub struct NoCertificateVerification {}
-
-impl rustls::client::ServerCertVerifier for NoCertificateVerification {
-	fn verify_server_cert(
-		&self,
-		_end_entity: &rustls::Certificate,
-		_intermediates: &[rustls::Certificate],
-		_server_name: &rustls::ServerName,
-		_scts: &mut dyn Iterator<Item = &[u8]>,
-		_ocsp_response: &[u8],
-		_now: time::SystemTime,
-	) -> Result<rustls::client::ServerCertVerified, rustls::Error> {
-		Ok(rustls::client::ServerCertVerified::assertion())
-	}
-}

moq-relay/src/web.rs

@@ -17,9 +17,9 @@ impl Web {
 	pub fn new(config: Config, tls: Tls) -> Self {
 		// Get the first certificate's fingerprint.
 		// TODO serve all of them so we can support multiple signature algorithms.
-		let fingerprint = tls.fingerprints.first().expect("missing certificate").clone();
+		let fingerprint = tls.fingerprints().first().expect("missing certificate").clone();
 
-		let mut tls_config = tls.server.clone();
+		let mut tls_config = tls.server();
 		tls_config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
 		let tls_config = axum_server::tls_rustls::RustlsConfig::from_config(Arc::new(tls_config));
 

moq-transport/Cargo.toml

@@ -24,6 +24,3 @@ indexmap = "2"
 quinn = "0.10"
 webtransport-quinn = "0.6"
 #webtransport-quinn = { path = "../../webtransport-rs/webtransport-quinn" }
-
-async-trait = "0.1"
-paste = "1"

moq-transport/src/cache/broadcast.rs

@@ -136,12 +136,12 @@ impl Publisher {
 	}
 
 	/// Block until the next track requested by a subscriber.
-	pub async fn next_track(&mut self) -> Result<track::Publisher, CacheError> {
+	pub async fn next_track(&mut self) -> Result<Option<track::Publisher>, CacheError> {
 		loop {
 			let notify = {
 				let state = self.state.lock();
 				if state.has_next()? {
-					return Ok(state.into_mut().next());
+					return Ok(Some(state.into_mut().next()));
 				}
 
 				state.changed()

moq-transport/src/cache/error.rs

@@ -39,13 +39,13 @@ impl MoqError for CacheError {
 	}
 
 	/// A reason that is sent over the wire.
-	fn reason(&self) -> String {
+	fn reason(&self) -> &str {
 		match self {
-			Self::Closed => "closed".to_owned(),
+			Self::Closed => "closed",
-			Self::Reset(code) => format!("reset code: {}", code),
+			Self::Reset(_) => "reset",
-			Self::Stop => "stop".to_owned(),
+			Self::Stop => "stop",
-			Self::NotFound => "not found".to_owned(),
+			Self::NotFound => "not found",
-			Self::Duplicate => "duplicate".to_owned(),
+			Self::Duplicate => "duplicate",
 		}
 	}
 }

moq-transport/src/cache/fragment.rs

@@ -1,216 +0,0 @@
-//! A fragment is a stream of bytes with a header, split into a [Publisher] and [Subscriber] handle.
-//!
-//! A [Publisher] writes an ordered stream of bytes in chunks.
-//! There's no framing, so these chunks can be of any size or position, and won't be maintained over the network.
-//!
-//! A [Subscriber] reads an ordered stream of bytes in chunks.
-//! These chunks are returned directly from the QUIC connection, so they may be of any size or position.
-//! You can clone the [Subscriber] and each will read a copy of of all future chunks. (fanout)
-//!
-//! The fragment is closed with [CacheError::Closed] when all publishers or subscribers are dropped.
-use core::fmt;
-use std::{ops::Deref, sync::Arc};
-
-use crate::VarInt;
-use bytes::Bytes;
-
-use super::{CacheError, Watch};
-
-/// Create a new segment with the given info.
-pub fn new(info: Info) -> (Publisher, Subscriber) {
-	let state = Watch::new(State::default());
-	let info = Arc::new(info);
-
-	let publisher = Publisher::new(state.clone(), info.clone());
-	let subscriber = Subscriber::new(state, info);
-
-	(publisher, subscriber)
-}
-
-/// Static information about the segment.
-#[derive(Debug)]
-pub struct Info {
-	// The sequence number of the fragment within the segment.
-	// NOTE: These may be received out of order or with gaps.
-	pub sequence: VarInt,
-
-	// The size of the fragment, optionally None if this is the last fragment in a segment.
-	// TODO enforce this size.
-	pub size: Option<VarInt>,
-}
-
-struct State {
-	// The data that has been received thus far.
-	chunks: Vec<Bytes>,
-
-	// Set when the publisher is dropped.
-	closed: Result<(), CacheError>,
-}
-
-impl State {
-	pub fn close(&mut self, err: CacheError) -> Result<(), CacheError> {
-		self.closed.clone()?;
-		self.closed = Err(err);
-		Ok(())
-	}
-
-	pub fn bytes(&self) -> usize {
-		self.chunks.iter().map(|f| f.len()).sum::<usize>()
-	}
-}
-
-impl Default for State {
-	fn default() -> Self {
-		Self {
-			chunks: Vec::new(),
-			closed: Ok(()),
-		}
-	}
-}
-
-impl fmt::Debug for State {
-	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-		// We don't want to print out the contents, so summarize.
-		f.debug_struct("State")
-			.field("chunks", &self.chunks.len().to_string())
-			.field("bytes", &self.bytes().to_string())
-			.field("closed", &self.closed)
-			.finish()
-	}
-}
-
-/// Used to write data to a segment and notify subscribers.
-pub struct Publisher {
-	// Mutable segment state.
-	state: Watch<State>,
-
-	// Immutable segment state.
-	info: Arc<Info>,
-
-	// Closes the segment when all Publishers are dropped.
-	_dropped: Arc<Dropped>,
-}
-
-impl Publisher {
-	fn new(state: Watch<State>, info: Arc<Info>) -> Self {
-		let _dropped = Arc::new(Dropped::new(state.clone()));
-		Self { state, info, _dropped }
-	}
-
-	/// Write a new chunk of bytes.
-	pub fn write_chunk(&mut self, chunk: Bytes) -> Result<(), CacheError> {
-		let mut state = self.state.lock_mut();
-		state.closed.clone()?;
-		state.chunks.push(chunk);
-		Ok(())
-	}
-
-	/// Close the segment with an error.
-	pub fn close(self, err: CacheError) -> Result<(), CacheError> {
-		self.state.lock_mut().close(err)
-	}
-}
-
-impl Deref for Publisher {
-	type Target = Info;
-
-	fn deref(&self) -> &Self::Target {
-		&self.info
-	}
-}
-
-impl fmt::Debug for Publisher {
-	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-		f.debug_struct("Publisher")
-			.field("state", &self.state)
-			.field("info", &self.info)
-			.finish()
-	}
-}
-
-/// Notified when a segment has new data available.
-#[derive(Clone)]
-pub struct Subscriber {
-	// Modify the segment state.
-	state: Watch<State>,
-
-	// Immutable segment state.
-	info: Arc<Info>,
-
-	// The number of chunks that we've read.
-	// NOTE: Cloned subscribers inherit this index, but then run in parallel.
-	index: usize,
-
-	// Dropped when all Subscribers are dropped.
-	_dropped: Arc<Dropped>,
-}
-
-impl Subscriber {
-	fn new(state: Watch<State>, info: Arc<Info>) -> Self {
-		let _dropped = Arc::new(Dropped::new(state.clone()));
-
-		Self {
-			state,
-			info,
-			index: 0,
-			_dropped,
-		}
-	}
-
-	/// Block until the next chunk of bytes is available.
-	pub async fn read_chunk(&mut self) -> Result<Option<Bytes>, CacheError> {
-		loop {
-			let notify = {
-				let state = self.state.lock();
-				if self.index < state.chunks.len() {
-					let chunk = state.chunks[self.index].clone();
-					self.index += 1;
-					return Ok(Some(chunk));
-				}
-
-				match &state.closed {
-					Err(CacheError::Closed) => return Ok(None),
-					Err(err) => return Err(err.clone()),
-					Ok(()) => state.changed(),
-				}
-			};
-
-			notify.await; // Try again when the state changes
-		}
-	}
-}
-
-impl Deref for Subscriber {
-	type Target = Info;
-
-	fn deref(&self) -> &Self::Target {
-		&self.info
-	}
-}
-
-impl fmt::Debug for Subscriber {
-	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-		f.debug_struct("Subscriber")
-			.field("state", &self.state)
-			.field("info", &self.info)
-			.field("index", &self.index)
-			.finish()
-	}
-}
-
-struct Dropped {
-	// Modify the segment state.
-	state: Watch<State>,
-}
-
-impl Dropped {
-	fn new(state: Watch<State>) -> Self {
-		Self { state }
-	}
-}
-
-impl Drop for Dropped {
-	fn drop(&mut self) {
-		self.state.lock_mut().close(CacheError::Closed).ok();
-	}
-}

moq-transport/src/cache/mod.rs

@@ -1,17 +1,10 @@
 //! Allows a publisher to push updates, automatically caching and fanning it out to any subscribers.
 //!
-//! The hierarchy is: [broadcast] -> [track] -> [segment] -> [fragment] -> [Bytes](bytes::Bytes)
+//! The naming scheme doesn't match the spec because it's vague and confusing.
-//!
+//! The hierarchy is: [broadcast] -> [track] -> [segment] -> [Bytes](bytes::Bytes)
-//! The naming scheme doesn't match the spec because it's more strict, and bikeshedding of course:
-//!
-//! - [broadcast] is kinda like "track namespace"
-//! - [track] is "track"
-//! - [segment] is "group" but MUST use a single stream.
-//! - [fragment] is "object" but MUST have the same properties as the segment.
 
 pub mod broadcast;
 mod error;
-pub mod fragment;
 pub mod segment;
 pub mod track;
 

moq-transport/src/cache/segment.rs

@@ -1,18 +1,20 @@
-//! A segment is a stream of fragments with a header, split into a [Publisher] and [Subscriber] handle.
+//! A segment is a stream of bytes with a header, split into a [Publisher] and [Subscriber] handle.
 //!
-//! A [Publisher] writes an ordered stream of fragments.
+//! A [Publisher] writes an ordered stream of bytes in chunks.
-//! Each fragment can have a sequence number, allowing the subscriber to detect gaps fragments.
+//! There's no framing, so these chunks can be of any size or position, and won't be maintained over the network.
 //!
-//! A [Subscriber] reads an ordered stream of fragments.
+//! A [Subscriber] reads an ordered stream of bytes in chunks.
-//! The subscriber can be cloned, in which case each subscriber receives a copy of each fragment. (fanout)
+//! These chunks are returned directly from the QUIC connection, so they may be of any size or position.
+//! A closed [Subscriber] will receive a copy of all future chunks. (fanout)
 //!
 //! The segment is closed with [CacheError::Closed] when all publishers or subscribers are dropped.
 use core::fmt;
 use std::{ops::Deref, sync::Arc, time};
 
 use crate::VarInt;
+use bytes::Bytes;
 
-use super::{fragment, CacheError, Watch};
+use super::{CacheError, Watch};
 
 /// Create a new segment with the given info.
 pub fn new(info: Info) -> (Publisher, Subscriber) {
@@ -29,11 +31,10 @@ pub fn new(info: Info) -> (Publisher, Subscriber) {
 #[derive(Debug)]
 pub struct Info {
 	// The sequence number of the segment within the track.
-	// NOTE: These may be received out of order or with gaps.
 	pub sequence: VarInt,
 
 	// The priority of the segment within the BROADCAST.
-	pub priority: u32,
+	pub priority: i32,
 
 	// Cache the segment for at most this long.
 	pub expires: Option<time::Duration>,
@@ -41,7 +42,7 @@ pub struct Info {
 
 struct State {
 	// The data that has been received thus far.
-	fragments: Vec<fragment::Subscriber>,
+	data: Vec<Bytes>,
 
 	// Set when the publisher is dropped.
 	closed: Result<(), CacheError>,
@@ -58,7 +59,7 @@ impl State {
 impl Default for State {
 	fn default() -> Self {
 		Self {
-			fragments: Vec::new(),
+			data: Vec::new(),
 			closed: Ok(()),
 		}
 	}
@@ -66,8 +67,12 @@ impl Default for State {
 
 impl fmt::Debug for State {
 	fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+		// We don't want to print out the contents, so summarize.
+		let size = self.data.iter().map(|chunk| chunk.len()).sum::<usize>();
+		let data = format!("size={} chunks={}", size, self.data.len());
+
 		f.debug_struct("State")
-			.field("fragments", &self.fragments)
+			.field("data", &data)
 			.field("closed", &self.closed)
 			.finish()
 	}
@@ -91,20 +96,14 @@ impl Publisher {
 		Self { state, info, _dropped }
 	}
 
-	/// Write a fragment
+	/// Write a new chunk of bytes.
-	pub fn push_fragment(&mut self, fragment: fragment::Subscriber) -> Result<(), CacheError> {
+	pub fn write_chunk(&mut self, data: Bytes) -> Result<(), CacheError> {
 		let mut state = self.state.lock_mut();
 		state.closed.clone()?;
-		state.fragments.push(fragment);
+		state.data.push(data);
 		Ok(())
 	}
 
-	pub fn create_fragment(&mut self, fragment: fragment::Info) -> Result<fragment::Publisher, CacheError> {
-		let (publisher, subscriber) = fragment::new(fragment);
-		self.push_fragment(subscriber)?;
-		Ok(publisher)
-	}
-
 	/// Close the segment with an error.
 	pub fn close(self, err: CacheError) -> Result<(), CacheError> {
 		self.state.lock_mut().close(err)
@@ -158,14 +157,14 @@ impl Subscriber {
 	}
 
 	/// Block until the next chunk of bytes is available.
-	pub async fn next_fragment(&mut self) -> Result<Option<fragment::Subscriber>, CacheError> {
+	pub async fn read_chunk(&mut self) -> Result<Option<Bytes>, CacheError> {
 		loop {
 			let notify = {
 				let state = self.state.lock();
-				if self.index < state.fragments.len() {
+				if self.index < state.data.len() {
-					let fragment = state.fragments[self.index].clone();
+					let chunk = state.data[self.index].clone();
 					self.index += 1;
-					return Ok(Some(fragment));
+					return Ok(Some(chunk));
 				}
 
 				match &state.closed {

moq-transport/src/cache/track.rs

@@ -206,7 +206,7 @@ impl Subscriber {
 		}
 	}
 
-	/// Block until the next segment arrives
+	/// Block until the next segment arrives, or return None if the track is [CacheError::Closed].
 	pub async fn next_segment(&mut self) -> Result<Option<segment::Subscriber>, CacheError> {
 		loop {
 			let notify = {

moq-transport/src/coding/decode.rs

@@ -1,5 +1,5 @@
 use super::{BoundsExceeded, VarInt};
-use std::{io, str};
+use std::str;
 
 use thiserror::Error;
 
@@ -7,13 +7,6 @@ use thiserror::Error;
 // TODO Use trait aliases when they're stable, or add these bounds to every method.
 pub trait AsyncRead: tokio::io::AsyncRead + Unpin + Send {}
 impl AsyncRead for webtransport_quinn::RecvStream {}
-impl<T> AsyncRead for tokio::io::Take<&mut T> where T: AsyncRead {}
-impl<T: AsRef<[u8]> + Unpin + Send> AsyncRead for io::Cursor<T> {}
-
-#[async_trait::async_trait]
-pub trait Decode: Sized {
-	async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError>;
-}
 
 /// A decode error.
 #[derive(Error, Debug)]
@@ -24,32 +17,12 @@ pub enum DecodeError {
 	#[error("invalid string")]
 	InvalidString(#[from] str::Utf8Error),
 
-	#[error("invalid message: {0:?}")]
+	#[error("invalid type: {0:?}")]
-	InvalidMessage(VarInt),
+	InvalidType(VarInt),
-
-	#[error("invalid role: {0:?}")]
-	InvalidRole(VarInt),
-
-	#[error("invalid subscribe location")]
-	InvalidSubscribeLocation,
 
 	#[error("varint bounds exceeded")]
 	BoundsExceeded(#[from] BoundsExceeded),
 
-	// TODO move these to ParamError
-	#[error("duplicate parameter")]
-	DupliateParameter,
-
-	#[error("missing parameter")]
-	MissingParameter,
-
-	#[error("invalid parameter")]
-	InvalidParameter,
-
 	#[error("io error: {0}")]
 	IoError(#[from] std::io::Error),
-
-	// Used to signal that the stream has ended.
-	#[error("no more messages")]
-	Final,
 }

moq-transport/src/coding/encode.rs

@@ -6,12 +6,6 @@ use thiserror::Error;
 // TODO Use trait aliases when they're stable, or add these bounds to every method.
 pub trait AsyncWrite: tokio::io::AsyncWrite + Unpin + Send {}
 impl AsyncWrite for webtransport_quinn::SendStream {}
-impl AsyncWrite for Vec<u8> {}
-
-#[async_trait::async_trait]
-pub trait Encode: Sized {
-	async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError>;
-}
 
 /// An encode error.
 #[derive(Error, Debug)]

moq-transport/src/coding/mod.rs

@@ -1,11 +1,9 @@
 mod decode;
 mod encode;
-mod params;
 mod string;
 mod varint;
 
 pub use decode::*;
 pub use encode::*;
-pub use params::*;
 pub use string::*;
 pub use varint::*;

moq-transport/src/coding/params.rs

@@ -1,85 +1,69 @@
-use std::io::Cursor;
+use std::cmp::min;
-use std::{cmp::max, collections::HashMap};
 
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use crate::VarInt;
 
-use crate::coding::{AsyncRead, AsyncWrite, Decode, Encode};
+use super::{AsyncRead, AsyncWrite, DecodeError, EncodeError};
+use tokio::io::AsyncReadExt;
 
-use crate::{
+// I hate this parameter encoding so much.
-	coding::{DecodeError, EncodeError},
+// i hate it i hate it i hate it
-	VarInt,
-};
 
-#[derive(Default, Debug, Clone)]
+// TODO Use #[async_trait] so we can do Param<VarInt> instead.
-pub struct Params(pub HashMap<VarInt, Vec<u8>>);
+pub struct ParamInt(pub VarInt);
 
-#[async_trait::async_trait]
+impl ParamInt {
-impl Decode for Params {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-	async fn decode<R: AsyncRead>(mut r: &mut R) -> Result<Self, DecodeError> {
+		// Why do we have a redundant size in front of each VarInt?
-		let mut params = HashMap::new();
+		let size = VarInt::decode(r).await?;
+		let mut take = r.take(size.into_inner());
+		let value = VarInt::decode(&mut take).await?;
 
-		// I hate this shit so much; let me encode my role and get on with my life.
+		// Like seriously why do I have to check if the VarInt length mismatches.
-		let count = VarInt::decode(r).await?;
+		if take.limit() != 0 {
-		for _ in 0..count.into_inner() {
+			return Err(DecodeError::InvalidSize);
-			let kind = VarInt::decode(r).await?;
-			if params.contains_key(&kind) {
-				return Err(DecodeError::DupliateParameter);
-			}
-
-			let size = VarInt::decode(r).await?;
-
-			// Don't allocate the entire requested size to avoid a possible attack
-			// Instead, we allocate up to 1024 and keep appending as we read further.
-			let mut pr = r.take(size.into_inner());
-			let mut buf = Vec::with_capacity(max(1024, pr.limit() as usize));
-			pr.read_to_end(&mut buf).await?;
-			params.insert(kind, buf);
-
-			r = pr.into_inner();
 		}
 
-		Ok(Params(params))
+		Ok(Self(value))
 	}
-}
 
-#[async_trait::async_trait]
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-impl Encode for Params {
+		// Seriously why do I have to compute the size.
-	async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
+		let size = self.0.size();
-		VarInt::try_from(self.0.len())?.encode(w).await?;
+		VarInt::try_from(size)?.encode(w).await?;
 
-		for (kind, value) in self.0.iter() {
+		self.0.encode(w).await?;
-			kind.encode(w).await?;
-			VarInt::try_from(value.len())?.encode(w).await?;
-			w.write_all(value).await?;
-		}
 
 		Ok(())
 	}
 }
 
-impl Params {
+pub struct ParamBytes(pub Vec<u8>);
-	pub fn new() -> Self {
+
-		Self::default()
+impl ParamBytes {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
+		let size = VarInt::decode(r).await?;
+		let mut take = r.take(size.into_inner());
+		let mut buf = Vec::with_capacity(min(take.limit() as usize, 1024));
+		take.read_to_end(&mut buf).await?;
+
+		Ok(Self(buf))
 	}
 
-	pub async fn set<P: Encode>(&mut self, kind: VarInt, p: P) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		let mut value = Vec::new();
+		let size = VarInt::try_from(self.0.len())?;
-		p.encode(&mut value).await?;
+		size.encode(w).await?;
-		self.0.insert(kind, value);
+		w.write_all(&self.0).await?;
 
 		Ok(())
 	}
+}
 
-	pub fn has(&self, kind: VarInt) -> bool {
+pub struct ParamUnknown {}
-		self.0.contains_key(&kind)
-	}
 
-	pub async fn get<P: Decode>(&mut self, kind: VarInt) -> Result<Option<P>, DecodeError> {
+impl ParamUnknown {
-		if let Some(value) = self.0.remove(&kind) {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<(), DecodeError> {
-			let mut cursor = Cursor::new(value);
+		// Really? Is there no way to advance without reading?
-			Ok(Some(P::decode(&mut cursor).await?))
+		ParamBytes::decode(r).await?;
-		} else {
+		Ok(())
-			Ok(None)
-		}
 	}
 }

moq-transport/src/coding/string.rs

@@ -5,25 +5,20 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
 use crate::VarInt;
 
-use super::{Decode, DecodeError, Encode, EncodeError};
+use super::{DecodeError, EncodeError};
 
-#[async_trait::async_trait]
+/// Encode a string with a varint length prefix.
-impl Encode for String {
+pub async fn encode_string<W: AsyncWrite>(s: &str, w: &mut W) -> Result<(), EncodeError> {
-	async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
+	let size = VarInt::try_from(s.len())?;
-		let size = VarInt::try_from(self.len())?;
+	size.encode(w).await?;
-		size.encode(w).await?;
+	w.write_all(s.as_ref()).await?;
-		w.write_all(self.as_ref()).await?;
+	Ok(())
-		Ok(())
-	}
 }
 
-#[async_trait::async_trait]
+/// Decode a string with a varint length prefix.
-impl Decode for String {
+pub async fn decode_string<R: AsyncRead>(r: &mut R) -> Result<String, DecodeError> {
-	/// Decode a string with a varint length prefix.
+	let size = VarInt::decode(r).await?.into_inner();
-	async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
+	let mut str = String::with_capacity(min(1024, size) as usize);
-		let size = VarInt::decode(r).await?.into_inner();
+	r.take(size).read_to_string(&mut str).await?;
-		let mut str = String::with_capacity(min(1024, size) as usize);
+	Ok(str)
-		r.take(size).read_to_string(&mut str).await?;
-		Ok(str)
-	}
 }

moq-transport/src/coding/varint.rs

@@ -9,7 +9,7 @@ use crate::coding::{AsyncRead, AsyncWrite};
 use thiserror::Error;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
-use super::{Decode, DecodeError, Encode, EncodeError};
+use super::{DecodeError, EncodeError};
 
 #[derive(Debug, Copy, Clone, Eq, PartialEq, Error)]
 #[error("value out of range")]
@@ -164,23 +164,14 @@ impl fmt::Display for VarInt {
 	}
 }
 
-#[async_trait::async_trait]
-impl Decode for VarInt {
-	/// Decode a varint from the given reader.
-	async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-		let b = r.read_u8().await?;
-		Self::decode_byte(b, r).await
-	}
-}
-
 impl VarInt {
-	/// Decode a varint given the first byte, reading the rest as needed.
+	/// Decode a varint from the given reader.
-	/// This is silly but useful for determining if the stream has ended.
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-	pub async fn decode_byte<R: AsyncRead>(b: u8, r: &mut R) -> Result<Self, DecodeError> {
-		let tag = b >> 6;
-
 		let mut buf = [0u8; 8];
-		buf[0] = b & 0b0011_1111;
+		r.read_exact(buf[0..1].as_mut()).await?;
+
+		let tag = buf[0] >> 6;
+		buf[0] &= 0b0011_1111;
 
 		let x = match tag {
 			0b00 => u64::from(buf[0]),
@@ -201,12 +192,9 @@ impl VarInt {
 
 		Ok(Self(x))
 	}
-}
 
-#[async_trait::async_trait]
-impl Encode for VarInt {
 	/// Encode a varint to the given writer.
-	async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
 		let x = self.0;
 		if x < 2u64.pow(6) {
 			w.write_u8(x as u8).await?;

moq-transport/src/error.rs

@@ -1,7 +1,5 @@
 pub trait MoqError {
 	/// An integer code that is sent over the wire.
 	fn code(&self) -> u32;
-
+	fn reason(&self) -> &str;
-	/// An optional reason sometimes sent over the wire.
-	fn reason(&self) -> String;
 }

moq-transport/src/lib.rs

@@ -5,7 +5,9 @@
 //! The specification is a work in progress and will change.
 //! See the [specification](https://datatracker.ietf.org/doc/draft-ietf-moq-transport/) and [github](https://github.com/moq-wg/moq-transport) for any updates.
 //!
-//! This implementation has some required extensions until the draft stablizes. See: [Extensions](crate::setup::Extensions)
+//! **FORKED**: This is implementation makes extensive changes to the protocol.
+//! See [KIXEL_00](crate::setup::Version::KIXEL_00) for a list of differences.
+//! Many of these will get merged into the specification, so don't panic.
 mod coding;
 mod error;
 

moq-transport/src/message/announce.rs

@@ -1,30 +1,22 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, Params};
+use crate::coding::{decode_string, encode_string, DecodeError, EncodeError};
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
 
 /// Sent by the publisher to announce the availability of a group of tracks.
 #[derive(Clone, Debug)]
 pub struct Announce {
-	/// The track namespace
+	// The track namespace
 	pub namespace: String,
-
-	/// Optional parameters
-	pub params: Params,
 }
 
 impl Announce {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-		let namespace = String::decode(r).await?;
+		let namespace = decode_string(r).await?;
-		let params = Params::decode(r).await?;
+		Ok(Self { namespace })
-
-		Ok(Self { namespace, params })
 	}
 
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		self.namespace.encode(w).await?;
+		encode_string(&self.namespace, w).await?;
-		self.params.encode(w).await?;
-
 		Ok(())
 	}
 }

moq-transport/src/message/announce_ok.rs

@@ -1,7 +1,4 @@
-use crate::{
+use crate::coding::{decode_string, encode_string, AsyncRead, AsyncWrite, DecodeError, EncodeError};
-	coding::{AsyncRead, AsyncWrite, Decode, DecodeError, Encode, EncodeError},
-	setup::Extensions,
-};
 
 /// Sent by the subscriber to accept an Announce.
 #[derive(Clone, Debug)]
@@ -12,12 +9,12 @@ pub struct AnnounceOk {
 }
 
 impl AnnounceOk {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-		let namespace = String::decode(r).await?;
+		let namespace = decode_string(r).await?;
 		Ok(Self { namespace })
 	}
 
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		self.namespace.encode(w).await
+		encode_string(&self.namespace, w).await
 	}
 }

moq-transport/src/message/announce_reset.rs

@@ -1,11 +1,10 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
+use crate::coding::{decode_string, encode_string, DecodeError, EncodeError, VarInt};
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
 
 /// Sent by the subscriber to reject an Announce.
 #[derive(Clone, Debug)]
-pub struct AnnounceError {
+pub struct AnnounceReset {
 	// Echo back the namespace that was reset
 	pub namespace: String,
 
@@ -16,11 +15,11 @@ pub struct AnnounceError {
 	pub reason: String,
 }
 
-impl AnnounceError {
+impl AnnounceReset {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-		let namespace = String::decode(r).await?;
+		let namespace = decode_string(r).await?;
 		let code = VarInt::decode(r).await?.try_into()?;
-		let reason = String::decode(r).await?;
+		let reason = decode_string(r).await?;
 
 		Ok(Self {
 			namespace,
@@ -29,10 +28,10 @@ impl AnnounceError {
 		})
 	}
 
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		self.namespace.encode(w).await?;
+		encode_string(&self.namespace, w).await?;
 		VarInt::from_u32(self.code).encode(w).await?;
-		self.reason.encode(w).await?;
+		encode_string(&self.reason, w).await?;
 
 		Ok(())
 	}

moq-transport/src/message/announce_stop.rs

@@ -0,0 +1,24 @@
+use crate::coding::{decode_string, encode_string, DecodeError, EncodeError};
+
+use crate::coding::{AsyncRead, AsyncWrite};
+
+/// Sent by the publisher to terminate an Announce.
+#[derive(Clone, Debug)]
+pub struct AnnounceStop {
+	// Echo back the namespace that was reset
+	pub namespace: String,
+}
+
+impl AnnounceStop {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
+		let namespace = decode_string(r).await?;
+
+		Ok(Self { namespace })
+	}
+
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
+		encode_string(&self.namespace, w).await?;
+
+		Ok(())
+	}
+}

moq-transport/src/message/go_away.rs

@@ -1,7 +1,6 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError};
+use crate::coding::{decode_string, encode_string, DecodeError, EncodeError};
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
 
 /// Sent by the server to indicate that the client should connect to a different server.
 #[derive(Clone, Debug)]
@@ -10,12 +9,12 @@ pub struct GoAway {
 }
 
 impl GoAway {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-		let url = String::decode(r).await?;
+		let url = decode_string(r).await?;
 		Ok(Self { url })
 	}
 
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		self.url.encode(w).await
+		encode_string(&self.url, w).await
 	}
 }

moq-transport/src/message/mod.rs

@@ -6,17 +6,16 @@
 //!
 //! Messages sent by the publisher:
 //! - [Announce]
-//! - [Unannounce]
+//! - [AnnounceReset]
 //! - [SubscribeOk]
-//! - [SubscribeError]
 //! - [SubscribeReset]
 //! - [Object]
 //!
 //! Messages sent by the subscriber:
 //! - [Subscribe]
-//! - [Unsubscribe]
+//! - [SubscribeStop]
 //! - [AnnounceOk]
-//! - [AnnounceError]
+//! - [AnnounceStop]
 //!
 //! Example flow:
 //! ```test
@@ -33,35 +32,30 @@
 mod announce;
 mod announce_ok;
 mod announce_reset;
+mod announce_stop;
 mod go_away;
 mod object;
 mod subscribe;
-mod subscribe_error;
-mod subscribe_fin;
 mod subscribe_ok;
 mod subscribe_reset;
-mod unannounce;
+mod subscribe_stop;
-mod unsubscribe;
 
 pub use announce::*;
 pub use announce_ok::*;
 pub use announce_reset::*;
+pub use announce_stop::*;
 pub use go_away::*;
 pub use object::*;
 pub use subscribe::*;
-pub use subscribe_error::*;
-pub use subscribe_fin::*;
 pub use subscribe_ok::*;
 pub use subscribe_reset::*;
-pub use unannounce::*;
+pub use subscribe_stop::*;
-pub use unsubscribe::*;
 
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
+use crate::coding::{DecodeError, EncodeError, VarInt};
 
 use std::fmt;
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
 
 // Use a macro to generate the message types rather than copy-paste.
 // This implements a decode/encode method that uses the specified type.
@@ -74,23 +68,23 @@ macro_rules! message_types {
 		}
 
 		impl Message {
-			pub async fn decode<R: AsyncRead>(r: &mut R, ext: &Extensions) -> Result<Self, DecodeError> {
+			pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 				let t = VarInt::decode(r).await?;
 
 				match t.into_inner() {
 					$($val => {
-						let msg = $name::decode(r, ext).await?;
+						let msg = $name::decode(r).await?;
 						Ok(Self::$name(msg))
 					})*
-					_ => Err(DecodeError::InvalidMessage(t)),
+					_ => Err(DecodeError::InvalidType(t)),
 				}
 			}
 
-			pub async fn encode<W: AsyncWrite>(&self, w: &mut W, ext: &Extensions) -> Result<(), EncodeError> {
+			pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
 				match self {
 					$(Self::$name(ref m) => {
 						VarInt::from_u32($val).encode(w).await?;
-						m.encode(w, ext).await
+						m.encode(w).await
 					},)*
 				}
 			}
@@ -133,28 +127,15 @@ macro_rules! message_types {
 message_types! {
 	// NOTE: Object and Setup are in other modules.
 	// Object = 0x0
-	// ObjectUnbounded = 0x2
+	// SetupClient = 0x1
-	// SetupClient = 0x40
+	// SetupServer = 0x2
-	// SetupServer = 0x41
-
-	// SUBSCRIBE family, sent by subscriber
 	Subscribe = 0x3,
-	Unsubscribe = 0xa,
-
-	// SUBSCRIBE family, sent by publisher
 	SubscribeOk = 0x4,
-	SubscribeError = 0x5,
+	SubscribeReset = 0x5,
-	SubscribeFin = 0xb,
+	SubscribeStop = 0x15,
-	SubscribeReset = 0xc,
-
-	// ANNOUNCE family, sent by publisher
 	Announce = 0x6,
-	Unannounce = 0x9,
-
-	// ANNOUNCE family, sent by subscriber
 	AnnounceOk = 0x7,
-	AnnounceError = 0x8,
+	AnnounceReset = 0x8,
-
+	AnnounceStop = 0x18,
-	// Misc
 	GoAway = 0x10,
 }

moq-transport/src/message/object.rs

@@ -1,10 +1,9 @@
-use std::{io, time};
+use std::time;
 
-use tokio::io::AsyncReadExt;
+use crate::coding::{DecodeError, EncodeError, VarInt};
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use crate::setup;
 
 /// Sent by the publisher as the header of each data stream.
 #[derive(Clone, Debug)]
@@ -14,78 +13,47 @@ pub struct Object {
 	pub track: VarInt,
 
 	// The sequence number within the track.
-	pub group: VarInt,
-
-	// The sequence number within the group.
 	pub sequence: VarInt,
 
-	// The priority, where **smaller** values are sent first.
+	// The priority, where **larger** values are sent first.
-	pub priority: u32,
+	// Proposal: int32 instead of a varint.
+	pub priority: i32,
 
 	// Cache the object for at most this many seconds.
 	// Zero means never expire.
 	pub expires: Option<time::Duration>,
-
-	/// An optional size, allowing multiple OBJECTs on the same stream.
-	pub size: Option<VarInt>,
 }
 
 impl Object {
-	pub async fn decode<R: AsyncRead>(r: &mut R, extensions: &setup::Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-		// Try reading the first byte, returning a special error if the stream naturally ended.
+		let typ = VarInt::decode(r).await?;
-		let typ = match r.read_u8().await {
+		if typ.into_inner() != 0 {
-			Ok(b) => VarInt::decode_byte(b, r).await?,
+			return Err(DecodeError::InvalidType(typ));
-			Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Err(DecodeError::Final),
+		}
-			Err(e) => return Err(e.into()),
-		};
 
-		let size_present = match typ.into_inner() {
+		// NOTE: size has been omitted
-			0 => false,
-			2 => true,
-			_ => return Err(DecodeError::InvalidMessage(typ)),
-		};
 
 		let track = VarInt::decode(r).await?;
-		let group = VarInt::decode(r).await?;
 		let sequence = VarInt::decode(r).await?;
-		let priority = VarInt::decode(r).await?.try_into()?;
+		let priority = r.read_i32().await?; // big-endian
-
+		let expires = match VarInt::decode(r).await?.into_inner() {
-		let expires = match extensions.object_expires {
+			0 => None,
-			true => match VarInt::decode(r).await?.into_inner() {
+			secs => Some(time::Duration::from_secs(secs)),
-				0 => None,
-				secs => Some(time::Duration::from_secs(secs)),
-			},
-			false => None,
-		};
-
-		// The presence of the size field depends on the type.
-		let size = match size_present {
-			true => Some(VarInt::decode(r).await?),
-			false => None,
 		};
 
 		Ok(Self {
 			track,
-			group,
 			sequence,
 			priority,
 			expires,
-			size,
 		})
 	}
 
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, extensions: &setup::Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		// The kind changes based on the presence of the size.
+		VarInt::ZERO.encode(w).await?;
-		let kind = match self.size {
-			Some(_) => VarInt::from_u32(2),
-			None => VarInt::ZERO,
-		};
-
-		kind.encode(w).await?;
 		self.track.encode(w).await?;
-		self.group.encode(w).await?;
 		self.sequence.encode(w).await?;
-		VarInt::from_u32(self.priority).encode(w).await?;
+		w.write_i32(self.priority).await?;
 
 		// Round up if there's any decimal points.
 		let expires = match self.expires {
@@ -95,13 +63,7 @@ impl Object {
 			Some(expires) => expires.as_secs(),
 		};
 
-		if extensions.object_expires {
+		VarInt::try_from(expires)?.encode(w).await?;
-			VarInt::try_from(expires)?.encode(w).await?;
-		}
-
-		if let Some(size) = self.size {
-			size.encode(w).await?;
-		}
 
 		Ok(())
 	}

moq-transport/src/message/subscribe.rs

@@ -1,141 +1,38 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, Params, VarInt};
+use crate::coding::{decode_string, encode_string, DecodeError, EncodeError, VarInt};
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
 
 /// Sent by the subscriber to request all future objects for the given track.
 ///
 /// Objects will use the provided ID instead of the full track name, to save bytes.
 #[derive(Clone, Debug)]
 pub struct Subscribe {
-	/// An ID we choose so we can map to the track_name.
+	// An ID we choose so we can map to the track_name.
 	// Proposal: https://github.com/moq-wg/moq-transport/issues/209
 	pub id: VarInt,
 
-	/// The track namespace.
+	// The track namespace.
-	///
+	pub namespace: String,
-	/// Must be None if `extensions.subscribe_split` is false.
-	pub namespace: Option<String>,
 
-	/// The track name.
+	// The track name.
 	pub name: String,
-
-	/// The start/end group/object.
-	pub start_group: SubscribeLocation,
-	pub start_object: SubscribeLocation,
-	pub end_group: SubscribeLocation,
-	pub end_object: SubscribeLocation,
-
-	/// Optional parameters
-	pub params: Params,
 }
 
 impl Subscribe {
-	pub async fn decode<R: AsyncRead>(r: &mut R, ext: &Extensions) -> Result<Self, DecodeError> {
-		let id = VarInt::decode(r).await?;
-
-		let namespace = match ext.subscribe_split {
-			true => Some(String::decode(r).await?),
-			false => None,
-		};
-
-		let name = String::decode(r).await?;
-
-		let start_group = SubscribeLocation::decode(r).await?;
-		let start_object = SubscribeLocation::decode(r).await?;
-		let end_group = SubscribeLocation::decode(r).await?;
-		let end_object = SubscribeLocation::decode(r).await?;
-
-		// You can't have a start object without a start group.
-		if start_group == SubscribeLocation::None && start_object != SubscribeLocation::None {
-			return Err(DecodeError::InvalidSubscribeLocation);
-		}
-
-		// You can't have an end object without an end group.
-		if end_group == SubscribeLocation::None && end_object != SubscribeLocation::None {
-			return Err(DecodeError::InvalidSubscribeLocation);
-		}
-
-		// NOTE: There's some more location restrictions in the draft, but they're enforced at a higher level.
-
-		let params = Params::decode(r).await?;
-
-		Ok(Self {
-			id,
-			namespace,
-			name,
-			start_group,
-			start_object,
-			end_group,
-			end_object,
-			params,
-		})
-	}
-
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, ext: &Extensions) -> Result<(), EncodeError> {
-		self.id.encode(w).await?;
-
-		if self.namespace.is_some() != ext.subscribe_split {
-			panic!("namespace must be None if subscribe_split is false");
-		}
-
-		if ext.subscribe_split {
-			self.namespace.as_ref().unwrap().encode(w).await?;
-		}
-
-		self.name.encode(w).await?;
-
-		self.start_group.encode(w).await?;
-		self.start_object.encode(w).await?;
-		self.end_group.encode(w).await?;
-		self.end_object.encode(w).await?;
-
-		self.params.encode(w).await?;
-
-		Ok(())
-	}
-}
-
-/// Signal where the subscription should begin, relative to the current cache.
-#[derive(Clone, Debug, PartialEq)]
-pub enum SubscribeLocation {
-	None,
-	Absolute(VarInt),
-	Latest(VarInt),
-	Future(VarInt),
-}
-
-impl SubscribeLocation {
 	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
-		let kind = VarInt::decode(r).await?;
+		let id = VarInt::decode(r).await?;
+		let namespace = decode_string(r).await?;
+		let name = decode_string(r).await?;
 
-		match kind.into_inner() {
+		Ok(Self { id, namespace, name })
-			0 => Ok(Self::None),
-			1 => Ok(Self::Absolute(VarInt::decode(r).await?)),
-			2 => Ok(Self::Latest(VarInt::decode(r).await?)),
-			3 => Ok(Self::Future(VarInt::decode(r).await?)),
-			_ => Err(DecodeError::InvalidSubscribeLocation),
-		}
 	}
+}
 
+impl Subscribe {
 	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		match self {
+		self.id.encode(w).await?;
-			Self::None => {
+		encode_string(&self.namespace, w).await?;
-				VarInt::from_u32(0).encode(w).await?;
+		encode_string(&self.name, w).await?;
-			}
-			Self::Absolute(val) => {
-				VarInt::from_u32(1).encode(w).await?;
-				val.encode(w).await?;
-			}
-			Self::Latest(val) => {
-				VarInt::from_u32(2).encode(w).await?;
-				val.encode(w).await?;
-			}
-			Self::Future(val) => {
-				VarInt::from_u32(3).encode(w).await?;
-				val.encode(w).await?;
-			}
-		}
 
 		Ok(())
 	}

moq-transport/src/message/subscribe_error.rs

@@ -1,36 +0,0 @@
-use crate::coding::{AsyncRead, AsyncWrite};
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
-use crate::setup::Extensions;
-
-/// Sent by the publisher to reject a Subscribe.
-#[derive(Clone, Debug)]
-pub struct SubscribeError {
-	// NOTE: No full track name because of this proposal: https://github.com/moq-wg/moq-transport/issues/209
-
-	// The ID for this subscription.
-	pub id: VarInt,
-
-	// An error code.
-	pub code: u32,
-
-	// An optional, human-readable reason.
-	pub reason: String,
-}
-
-impl SubscribeError {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
-		let id = VarInt::decode(r).await?;
-		let code = VarInt::decode(r).await?.try_into()?;
-		let reason = String::decode(r).await?;
-
-		Ok(Self { id, code, reason })
-	}
-
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
-		self.id.encode(w).await?;
-		VarInt::from_u32(self.code).encode(w).await?;
-		self.reason.encode(w).await?;
-
-		Ok(())
-	}
-}

moq-transport/src/message/subscribe_fin.rs

@@ -1,37 +0,0 @@
-use crate::coding::{AsyncRead, AsyncWrite};
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
-use crate::setup::Extensions;
-
-/// Sent by the publisher to cleanly terminate a Subscribe.
-#[derive(Clone, Debug)]
-pub struct SubscribeFin {
-	// NOTE: No full track name because of this proposal: https://github.com/moq-wg/moq-transport/issues/209
-	/// The ID for this subscription.
-	pub id: VarInt,
-
-	/// The final group/object sent on this subscription.
-	pub final_group: VarInt,
-	pub final_object: VarInt,
-}
-
-impl SubscribeFin {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
-		let id = VarInt::decode(r).await?;
-		let final_group = VarInt::decode(r).await?;
-		let final_object = VarInt::decode(r).await?;
-
-		Ok(Self {
-			id,
-			final_group,
-			final_object,
-		})
-	}
-
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
-		self.id.encode(w).await?;
-		self.final_group.encode(w).await?;
-		self.final_object.encode(w).await?;
-
-		Ok(())
-	}
-}

moq-transport/src/message/subscribe_ok.rs

@@ -1,31 +1,26 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
+use crate::coding::{DecodeError, EncodeError, VarInt};
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
 
 /// Sent by the publisher to accept a Subscribe.
 #[derive(Clone, Debug)]
 pub struct SubscribeOk {
 	// NOTE: No full track name because of this proposal: https://github.com/moq-wg/moq-transport/issues/209
-	/// The ID for this track.
-	pub id: VarInt,
 
-	/// The subscription will expire in this many milliseconds.
+	// The ID for this track.
-	pub expires: VarInt,
+	pub id: VarInt,
 }
 
 impl SubscribeOk {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 		let id = VarInt::decode(r).await?;
-		let expires = VarInt::decode(r).await?;
+		Ok(Self { id })
-		Ok(Self { id, expires })
 	}
 }
 
 impl SubscribeOk {
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
 		self.id.encode(w).await?;
-		self.expires.encode(w).await?;
 		Ok(())
 	}
 }

moq-transport/src/message/subscribe_reset.rs

@@ -1,49 +1,35 @@
-use crate::coding::{AsyncRead, AsyncWrite};
+use crate::coding::{decode_string, encode_string, DecodeError, EncodeError, VarInt};
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
-use crate::setup::Extensions;
 
-/// Sent by the publisher to terminate a Subscribe.
+use crate::coding::{AsyncRead, AsyncWrite};
+
+/// Sent by the publisher to reject a Subscribe.
 #[derive(Clone, Debug)]
 pub struct SubscribeReset {
 	// NOTE: No full track name because of this proposal: https://github.com/moq-wg/moq-transport/issues/209
-	/// The ID for this subscription.
+
+	// The ID for this subscription.
 	pub id: VarInt,
 
-	/// An error code.
+	// An error code.
 	pub code: u32,
 
-	/// An optional, human-readable reason.
+	// An optional, human-readable reason.
 	pub reason: String,
-
-	/// The final group/object sent on this subscription.
-	pub final_group: VarInt,
-	pub final_object: VarInt,
 }
 
 impl SubscribeReset {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 		let id = VarInt::decode(r).await?;
 		let code = VarInt::decode(r).await?.try_into()?;
-		let reason = String::decode(r).await?;
+		let reason = decode_string(r).await?;
-		let final_group = VarInt::decode(r).await?;
-		let final_object = VarInt::decode(r).await?;
 
-		Ok(Self {
+		Ok(Self { id, code, reason })
-			id,
-			code,
-			reason,
-			final_group,
-			final_object,
-		})
 	}
 
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
 		self.id.encode(w).await?;
 		VarInt::from_u32(self.code).encode(w).await?;
-		self.reason.encode(w).await?;
+		encode_string(&self.reason, w).await?;
-
-		self.final_group.encode(w).await?;
-		self.final_object.encode(w).await?;
 
 		Ok(())
 	}

moq-transport/src/message/subscribe_stop.rs

@@ -1,26 +1,25 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
+use crate::coding::{DecodeError, EncodeError, VarInt};
 
 use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
 
 /// Sent by the subscriber to terminate a Subscribe.
 #[derive(Clone, Debug)]
-pub struct Unsubscribe {
+pub struct SubscribeStop {
 	// NOTE: No full track name because of this proposal: https://github.com/moq-wg/moq-transport/issues/209
 
 	// The ID for this subscription.
 	pub id: VarInt,
 }
 
-impl Unsubscribe {
+impl SubscribeStop {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 		let id = VarInt::decode(r).await?;
 		Ok(Self { id })
 	}
 }
 
-impl Unsubscribe {
+impl SubscribeStop {
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
 		self.id.encode(w).await?;
 		Ok(())
 	}

moq-transport/src/message/unannounce.rs

@@ -1,25 +0,0 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError};
-
-use crate::coding::{AsyncRead, AsyncWrite};
-use crate::setup::Extensions;
-
-/// Sent by the publisher to terminate an Announce.
-#[derive(Clone, Debug)]
-pub struct Unannounce {
-	// Echo back the namespace that was reset
-	pub namespace: String,
-}
-
-impl Unannounce {
-	pub async fn decode<R: AsyncRead>(r: &mut R, _ext: &Extensions) -> Result<Self, DecodeError> {
-		let namespace = String::decode(r).await?;
-
-		Ok(Self { namespace })
-	}
-
-	pub async fn encode<W: AsyncWrite>(&self, w: &mut W, _ext: &Extensions) -> Result<(), EncodeError> {
-		self.namespace.encode(w).await?;
-
-		Ok(())
-	}
-}

moq-transport/src/session/client.rs

@@ -1,6 +1,6 @@
-use super::{Control, Publisher, SessionError, Subscriber};
+use super::{Publisher, SessionError, Subscriber};
 use crate::{cache::broadcast, setup};
-use webtransport_quinn::Session;
+use webtransport_quinn::{RecvStream, SendStream, Session};
 
 /// An endpoint that connects to a URL to publish and/or consume live streams.
 pub struct Client {}
@@ -9,6 +9,7 @@ impl Client {
 	/// Connect using an established WebTransport session, performing the MoQ handshake as a publisher.
 	pub async fn publisher(session: Session, source: broadcast::Subscriber) -> Result<Publisher, SessionError> {
 		let control = Self::send_setup(&session, setup::Role::Publisher).await?;
+
 		let publisher = Publisher::new(session, control, source);
 		Ok(publisher)
 	}
@@ -16,6 +17,7 @@ impl Client {
 	/// Connect using an established WebTransport session, performing the MoQ handshake as a subscriber.
 	pub async fn subscriber(session: Session, source: broadcast::Publisher) -> Result<Subscriber, SessionError> {
 		let control = Self::send_setup(&session, setup::Role::Subscriber).await?;
+
 		let subscriber = Subscriber::new(session, control, source);
 		Ok(subscriber)
 	}
@@ -27,46 +29,26 @@ impl Client {
 		}
 	*/
 
-	async fn send_setup(session: &Session, role: setup::Role) -> Result<Control, SessionError> {
+	async fn send_setup(session: &Session, role: setup::Role) -> Result<(SendStream, RecvStream), SessionError> {
 		let mut control = session.open_bi().await?;
 
-		let versions: setup::Versions = [setup::Version::DRAFT_01, setup::Version::KIXEL_01].into();
-
 		let client = setup::Client {
 			role,
-			versions: versions.clone(),
+			versions: vec![setup::Version::KIXEL_00].into(),
-			params: Default::default(),
-
-			// Offer all extensions
-			extensions: setup::Extensions {
-				object_expires: true,
-				subscriber_id: true,
-				subscribe_split: true,
-			},
 		};
 
 		client.encode(&mut control.0).await?;
 
-		let mut server = setup::Server::decode(&mut control.1).await?;
+		let server = setup::Server::decode(&mut control.1).await?;
 
-		match server.version {
+		if server.version != setup::Version::KIXEL_00 {
-			setup::Version::DRAFT_01 => {
+			return Err(SessionError::Version(Some(server.version)));
-				// We always require this extension
-				server.extensions.require_subscriber_id()?;
-
-				if server.role.is_publisher() {
-					// We only require object expires if we're a subscriber, so we don't cache objects indefinitely.
-					server.extensions.require_object_expires()?;
-				}
-			}
-			setup::Version::KIXEL_01 => {
-				// KIXEL_01 didn't support extensions; all were enabled.
-				server.extensions = client.extensions.clone()
-			}
-			_ => return Err(SessionError::Version(versions, [server.version].into())),
 		}
 
-		let control = Control::new(control.0, control.1, server.extensions);
+		// Make sure the server replied with the
+		if !client.role.is_compatible(server.role) {
+			return Err(SessionError::RoleIncompatible(client.role, server.role));
+		}
 
 		Ok(control)
 	}

moq-transport/src/session/control.rs

@@ -6,21 +6,19 @@ use tokio::sync::Mutex;
 use webtransport_quinn::{RecvStream, SendStream};
 
 use super::SessionError;
-use crate::{message::Message, setup::Extensions};
+use crate::message::Message;
 
 #[derive(Debug, Clone)]
 pub(crate) struct Control {
 	send: Arc<Mutex<SendStream>>,
 	recv: Arc<Mutex<RecvStream>>,
-	pub ext: Extensions,
 }
 
 impl Control {
-	pub fn new(send: SendStream, recv: RecvStream, ext: Extensions) -> Self {
+	pub fn new(send: SendStream, recv: RecvStream) -> Self {
 		Self {
 			send: Arc::new(Mutex::new(send)),
 			recv: Arc::new(Mutex::new(recv)),
-			ext,
 		}
 	}
 
@@ -28,7 +26,7 @@ impl Control {
 		let mut stream = self.send.lock().await;
 		log::info!("sending message: {:?}", msg);
 		msg.into()
-			.encode(&mut *stream, &self.ext)
+			.encode(&mut *stream)
 			.await
 			.map_err(|e| SessionError::Unknown(e.to_string()))?;
 		Ok(())
@@ -37,7 +35,7 @@ impl Control {
 	// It's likely a mistake to call this from two different tasks, but it's easier to just support it.
 	pub async fn recv(&self) -> Result<Message, SessionError> {
 		let mut stream = self.recv.lock().await;
-		let msg = Message::decode(&mut *stream, &self.ext)
+		let msg = Message::decode(&mut *stream)
 			.await
 			.map_err(|e| SessionError::Unknown(e.to_string()))?;
 		Ok(msg)

moq-transport/src/session/error.rs

@@ -14,8 +14,8 @@ pub enum SessionError {
 	#[error("decode error: {0}")]
 	Decode(#[from] coding::DecodeError),
 
-	#[error("unsupported versions: client={0:?} server={1:?}")]
+	#[error("unsupported version: {0:?}")]
-	Version(setup::Versions, setup::Versions),
+	Version(Option<setup::Version>),
 
 	#[error("incompatible roles: client={0:?} server={1:?}")]
 	RoleIncompatible(setup::Role, setup::Role),
@@ -32,22 +32,6 @@ pub enum SessionError {
 	#[error("role violation: msg={0}")]
 	RoleViolation(VarInt),
 
-	/// Our enforced stream mapping was disrespected.
-	#[error("stream mapping conflict")]
-	StreamMapping,
-
-	/// The priority was invalid.
-	#[error("invalid priority: {0}")]
-	InvalidPriority(VarInt),
-
-	/// The size was invalid.
-	#[error("invalid size: {0}")]
-	InvalidSize(VarInt),
-
-	/// A required extension was not offered.
-	#[error("required extension not offered: {0:?}")]
-	RequiredExtension(VarInt),
-
 	/// An unclassified error because I'm lazy. TODO classify these errors
 	#[error("unknown error: {0}")]
 	Unknown(String),
@@ -60,42 +44,29 @@ impl MoqError for SessionError {
 			Self::Cache(err) => err.code(),
 			Self::RoleIncompatible(..) => 406,
 			Self::RoleViolation(..) => 405,
-			Self::StreamMapping => 409,
 			Self::Unknown(_) => 500,
 			Self::Write(_) => 501,
 			Self::Read(_) => 502,
 			Self::Session(_) => 503,
-			Self::Version(..) => 406,
+			Self::Version(_) => 406,
 			Self::Encode(_) => 500,
 			Self::Decode(_) => 500,
-			Self::InvalidPriority(_) => 400,
-			Self::InvalidSize(_) => 400,
-			Self::RequiredExtension(_) => 426,
 		}
 	}
 
 	/// A reason that is sent over the wire.
-	fn reason(&self) -> String {
+	fn reason(&self) -> &str {
 		match self {
 			Self::Cache(err) => err.reason(),
-			Self::RoleViolation(kind) => format!("role violation for message type {:?}", kind),
+			Self::RoleViolation(_) => "role violation",
-			Self::RoleIncompatible(client, server) => {
+			Self::RoleIncompatible(..) => "role incompatible",
-				format!(
+			Self::Read(_) => "read error",
-					"role incompatible: client wanted {:?} but server wanted {:?}",
+			Self::Write(_) => "write error",
-					client, server
+			Self::Session(_) => "session error",
-				)
+			Self::Unknown(_) => "unknown",
-			}
+			Self::Version(_) => "unsupported version",
-			Self::Read(err) => format!("read error: {}", err),
+			Self::Encode(_) => "encode error",
-			Self::Write(err) => format!("write error: {}", err),
+			Self::Decode(_) => "decode error",
-			Self::Session(err) => format!("session error: {}", err),
-			Self::Unknown(err) => format!("unknown error: {}", err),
-			Self::Version(client, server) => format!("unsupported versions: client={:?} server={:?}", client, server),
-			Self::Encode(err) => format!("encode error: {}", err),
-			Self::Decode(err) => format!("decode error: {}", err),
-			Self::StreamMapping => "streaming mapping conflict".to_owned(),
-			Self::InvalidPriority(priority) => format!("invalid priority: {}", priority),
-			Self::InvalidSize(size) => format!("invalid size: {}", size),
-			Self::RequiredExtension(id) => format!("required extension was missing: {:?}", id),
 		}
 	}
 }

moq-transport/src/session/publisher.rs

@@ -4,7 +4,7 @@ use std::{
 };
 
 use tokio::task::AbortHandle;
-use webtransport_quinn::Session;
+use webtransport_quinn::{RecvStream, SendStream, Session};
 
 use crate::{
 	cache::{broadcast, segment, track, CacheError},
@@ -27,11 +27,13 @@ pub struct Publisher {
 }
 
 impl Publisher {
-	pub(crate) fn new(webtransport: Session, control: Control, source: broadcast::Subscriber) -> Self {
+	pub(crate) fn new(webtransport: Session, control: (SendStream, RecvStream), source: broadcast::Subscriber) -> Self {
+		let control = Control::new(control.0, control.1);
+
 		Self {
 			webtransport,
-			control,
 			subscribes: Default::default(),
+			control,
 			source,
 		}
 	}
@@ -83,9 +85,9 @@ impl Publisher {
 	async fn recv_message(&mut self, msg: &Message) -> Result<(), SessionError> {
 		match msg {
 			Message::AnnounceOk(msg) => self.recv_announce_ok(msg).await,
-			Message::AnnounceError(msg) => self.recv_announce_error(msg).await,
+			Message::AnnounceStop(msg) => self.recv_announce_stop(msg).await,
 			Message::Subscribe(msg) => self.recv_subscribe(msg).await,
-			Message::Unsubscribe(msg) => self.recv_unsubscribe(msg).await,
+			Message::SubscribeStop(msg) => self.recv_subscribe_stop(msg).await,
 			_ => Err(SessionError::RoleViolation(msg.id())),
 		}
 	}
@@ -95,7 +97,7 @@ impl Publisher {
 		Err(CacheError::NotFound.into())
 	}
 
-	async fn recv_announce_error(&mut self, _msg: &message::AnnounceError) -> Result<(), SessionError> {
+	async fn recv_announce_stop(&mut self, _msg: &message::AnnounceStop) -> Result<(), SessionError> {
 		// We didn't send an announce.
 		Err(CacheError::NotFound.into())
 	}
@@ -113,24 +115,14 @@ impl Publisher {
 			hash_map::Entry::Vacant(entry) => entry.insert(abort),
 		};
 
-		self.control
+		self.control.send(message::SubscribeOk { id: msg.id }).await
-			.send(message::SubscribeOk {
-				id: msg.id,
-				expires: VarInt::ZERO,
-			})
-			.await
 	}
 
 	async fn reset_subscribe<E: MoqError>(&mut self, id: VarInt, err: E) -> Result<(), SessionError> {
 		let msg = message::SubscribeReset {
 			id,
 			code: err.code(),
-			reason: err.reason(),
+			reason: err.reason().to_string(),
-
-			// TODO properly populate these
-			// But first: https://github.com/moq-wg/moq-transport/issues/313
-			final_group: VarInt::ZERO,
-			final_object: VarInt::ZERO,
 		};
 
 		self.control.send(msg).await
@@ -138,8 +130,7 @@ impl Publisher {
 
 	fn start_subscribe(&mut self, msg: message::Subscribe) -> Result<AbortHandle, SessionError> {
 		// We currently don't use the namespace field in SUBSCRIBE
-		// Make sure the namespace is empty if it's provided.
+		if !msg.namespace.is_empty() {
-		if msg.namespace.as_ref().map_or(false, |namespace| !namespace.is_empty()) {
 			return Err(CacheError::NotFound.into());
 		}
 
@@ -185,42 +176,31 @@ impl Publisher {
 	}
 
 	async fn run_segment(&self, id: VarInt, segment: &mut segment::Subscriber) -> Result<(), SessionError> {
-		log::trace!("serving group: {:?}", segment);
+		let object = message::Object {
+			track: id,
+			sequence: segment.sequence,
+			priority: segment.priority,
+			expires: segment.expires,
+		};
+
+		log::trace!("serving object: {:?}", object);
 
 		let mut stream = self.webtransport.open_uni().await?;
+		stream.set_priority(object.priority).ok();
 
-		// Convert the u32 to a i32, since the Quinn set_priority is signed.
+		object
-		let priority = (segment.priority as i64 - i32::MAX as i64) as i32;
+			.encode(&mut stream)
-		stream.set_priority(priority).ok();
+			.await
+			.map_err(|e| SessionError::Unknown(e.to_string()))?;
 
-		while let Some(mut fragment) = segment.next_fragment().await? {
+		while let Some(data) = segment.read_chunk().await? {
-			let object = message::Object {
+			stream.write_chunk(data).await?;
-				track: id,
-
-				// Properties of the segment
-				group: segment.sequence,
-				priority: segment.priority,
-				expires: segment.expires,
-
-				// Properties of the fragment
-				sequence: fragment.sequence,
-				size: fragment.size,
-			};
-
-			object
-				.encode(&mut stream, &self.control.ext)
-				.await
-				.map_err(|e| SessionError::Unknown(e.to_string()))?;
-
-			while let Some(chunk) = fragment.read_chunk().await? {
-				stream.write_all(&chunk).await?;
-			}
 		}
 
 		Ok(())
 	}
 
-	async fn recv_unsubscribe(&mut self, msg: &message::Unsubscribe) -> Result<(), SessionError> {
+	async fn recv_subscribe_stop(&mut self, msg: &message::SubscribeStop) -> Result<(), SessionError> {
 		let abort = self
 			.subscribes
 			.lock()

moq-transport/src/session/server.rs

@@ -1,4 +1,4 @@
-use super::{Control, Publisher, SessionError, Subscriber};
+use super::{Publisher, SessionError, Subscriber};
 use crate::{cache::broadcast, setup};
 
 use webtransport_quinn::{RecvStream, SendStream, Session};
@@ -13,32 +13,13 @@ impl Server {
 	pub async fn accept(session: Session) -> Result<Request, SessionError> {
 		let mut control = session.accept_bi().await?;
 
-		let mut client = setup::Client::decode(&mut control.1).await?;
+		let client = setup::Client::decode(&mut control.1).await?;
 
-		if client.versions.contains(&setup::Version::DRAFT_01) {
+		client
-			// We always require subscriber ID.
+			.versions
-			client.extensions.require_subscriber_id()?;
+			.iter()
-
+			.find(|version| **version == setup::Version::KIXEL_00)
-			// We require OBJECT_EXPIRES for publishers only.
+			.ok_or_else(|| SessionError::Version(client.versions.last().cloned()))?;
-			if client.role.is_publisher() {
-				client.extensions.require_object_expires()?;
-			}
-
-		// We don't require SUBSCRIBE_SPLIT since it's easy enough to support, but it's clearly an oversight.
-		// client.extensions.require(&Extension::SUBSCRIBE_SPLIT)?;
-		} else if client.versions.contains(&setup::Version::KIXEL_01) {
-			// Extensions didn't exist in KIXEL_01, so we set them manually.
-			client.extensions = setup::Extensions {
-				object_expires: true,
-				subscriber_id: true,
-				subscribe_split: true,
-			};
-		} else {
-			return Err(SessionError::Version(
-				client.versions,
-				[setup::Version::DRAFT_01, setup::Version::KIXEL_01].into(),
-			));
-		}
 
 		Ok(Request {
 			session,
@@ -58,21 +39,17 @@ pub struct Request {
 impl Request {
 	/// Accept the session as a publisher, using the provided broadcast to serve subscriptions.
 	pub async fn publisher(mut self, source: broadcast::Subscriber) -> Result<Publisher, SessionError> {
-		let setup = self.setup(setup::Role::Publisher)?;
+		self.send_setup(setup::Role::Publisher).await?;
-		setup.encode(&mut self.control.0).await?;
 
-		let control = Control::new(self.control.0, self.control.1, setup.extensions);
+		let publisher = Publisher::new(self.session, self.control, source);
-		let publisher = Publisher::new(self.session, control, source);
 		Ok(publisher)
 	}
 
 	/// Accept the session as a subscriber only.
 	pub async fn subscriber(mut self, source: broadcast::Publisher) -> Result<Subscriber, SessionError> {
-		let setup = self.setup(setup::Role::Subscriber)?;
+		self.send_setup(setup::Role::Subscriber).await?;
-		setup.encode(&mut self.control.0).await?;
 
-		let control = Control::new(self.control.0, self.control.1, setup.extensions);
+		let subscriber = Subscriber::new(self.session, self.control, source);
-		let subscriber = Subscriber::new(self.session, control, source);
 		Ok(subscriber)
 	}
 
@@ -83,12 +60,10 @@ impl Request {
 	}
 	*/
 
-	fn setup(&mut self, role: setup::Role) -> Result<setup::Server, SessionError> {
+	async fn send_setup(&mut self, role: setup::Role) -> Result<(), SessionError> {
 		let server = setup::Server {
 			role,
-			version: setup::Version::DRAFT_01,
+			version: setup::Version::KIXEL_00,
-			extensions: self.client.extensions.clone(),
-			params: Default::default(),
 		};
 
 		// We need to sure we support the opposite of the client's role.
@@ -97,7 +72,9 @@ impl Request {
 			return Err(SessionError::RoleIncompatible(self.client.role, server.role));
 		}
 
-		Ok(server)
+		server.encode(&mut self.control.0).await?;
+
+		Ok(())
 	}
 
 	/// Reject the request, closing the Webtransport session.

moq-transport/src/session/subscriber.rs

@@ -1,4 +1,4 @@
-use webtransport_quinn::{RecvStream, Session};
+use webtransport_quinn::{RecvStream, SendStream, Session};
 
 use std::{
 	collections::HashMap,
@@ -6,8 +6,7 @@ use std::{
 };
 
 use crate::{
-	cache::{broadcast, fragment, segment, track, CacheError},
+	cache::{broadcast, segment, track, CacheError},
-	coding::DecodeError,
 	message,
 	message::Message,
 	session::{Control, SessionError},
@@ -35,7 +34,9 @@ pub struct Subscriber {
 }
 
 impl Subscriber {
-	pub(crate) fn new(webtransport: Session, control: Control, source: broadcast::Publisher) -> Self {
+	pub(crate) fn new(webtransport: Session, control: (SendStream, RecvStream), source: broadcast::Publisher) -> Self {
+		let control = Control::new(control.0, control.1);
+
 		Self {
 			webtransport,
 			subscribes: Default::default(),
@@ -63,28 +64,28 @@ impl Subscriber {
 			let msg = self.control.recv().await?;
 
 			log::info!("message received: {:?}", msg);
-			if let Err(err) = self.recv_message(&msg) {
+			if let Err(err) = self.recv_message(&msg).await {
 				log::warn!("message error: {:?} {:?}", err, msg);
 			}
 		}
 	}
 
-	fn recv_message(&mut self, msg: &Message) -> Result<(), SessionError> {
+	async fn recv_message(&mut self, msg: &Message) -> Result<(), SessionError> {
 		match msg {
-			Message::Announce(_) => Ok(()),       // don't care
+			Message::Announce(_) => Ok(()),      // don't care
-			Message::Unannounce(_) => Ok(()),     // also don't care
+			Message::AnnounceReset(_) => Ok(()), // also don't care
-			Message::SubscribeOk(_msg) => Ok(()), // don't care
+			Message::SubscribeOk(_) => Ok(()),   // guess what, don't care
-			Message::SubscribeReset(msg) => self.recv_subscribe_error(msg.id, CacheError::Reset(msg.code)),
+			Message::SubscribeReset(msg) => self.recv_subscribe_reset(msg).await,
-			Message::SubscribeFin(msg) => self.recv_subscribe_error(msg.id, CacheError::Closed),
-			Message::SubscribeError(msg) => self.recv_subscribe_error(msg.id, CacheError::Reset(msg.code)),
 			Message::GoAway(_msg) => unimplemented!("GOAWAY"),
 			_ => Err(SessionError::RoleViolation(msg.id())),
 		}
 	}
 
-	fn recv_subscribe_error(&mut self, id: VarInt, err: CacheError) -> Result<(), SessionError> {
+	async fn recv_subscribe_reset(&mut self, msg: &message::SubscribeReset) -> Result<(), SessionError> {
+		let err = CacheError::Reset(msg.code);
+
 		let mut subscribes = self.subscribes.lock().unwrap();
-		let subscribe = subscribes.remove(&id).ok_or(CacheError::NotFound)?;
+		let subscribe = subscribes.remove(&msg.id).ok_or(CacheError::NotFound)?;
 		subscribe.close(err)?;
 
 		Ok(())
@@ -106,82 +107,36 @@ impl Subscriber {
 
 	async fn run_stream(self, mut stream: RecvStream) -> Result<(), SessionError> {
 		// Decode the object on the data stream.
-		let mut object = message::Object::decode(&mut stream, &self.control.ext)
+		let object = message::Object::decode(&mut stream)
 			.await
 			.map_err(|e| SessionError::Unknown(e.to_string()))?;
 
 		log::trace!("received object: {:?}", object);
 
 		// A new scope is needed because the async compiler is dumb
-		let mut segment = {
+		let mut publisher = {
 			let mut subscribes = self.subscribes.lock().unwrap();
 			let track = subscribes.get_mut(&object.track).ok_or(CacheError::NotFound)?;
 
 			track.create_segment(segment::Info {
-				sequence: object.group,
+				sequence: object.sequence,
 				priority: object.priority,
 				expires: object.expires,
 			})?
 		};
 
-		// Create the first fragment
+		while let Some(data) = stream.read_chunk(usize::MAX, true).await? {
-		let mut fragment = segment.create_fragment(fragment::Info {
+			// NOTE: This does not make a copy!
-			sequence: object.sequence,
+			// Bytes are immutable and ref counted.
-			size: object.size,
+			publisher.write_chunk(data.bytes)?;
-		})?;
-
-		let mut remain = object.size.map(usize::from);
-
-		loop {
-			if let Some(0) = remain {
-				// Decode the next object from the stream.
-				let next = match message::Object::decode(&mut stream, &self.control.ext).await {
-					Ok(next) => next,
-
-					// No more objects
-					Err(DecodeError::Final) => break,
-
-					// Unknown error
-					Err(err) => return Err(err.into()),
-				};
-
-				// NOTE: This is a custom restriction; not part of the moq-transport draft.
-				// We require every OBJECT to contain the same priority since prioritization is done per-stream.
-				// We also require every OBJECT to contain the same group so we know when the group ends, and can detect gaps.
-				if next.priority != object.priority && next.group != object.group {
-					return Err(SessionError::StreamMapping);
-				}
-
-				// Create a new object.
-				fragment = segment.create_fragment(fragment::Info {
-					sequence: object.sequence,
-					size: object.size,
-				})?;
-
-				object = next;
-				remain = object.size.map(usize::from);
-			}
-
-			match stream.read_chunk(remain.unwrap_or(usize::MAX), true).await? {
-				// Unbounded object has ended
-				None if remain.is_none() => break,
-
-				// Bounded object ended early, oops.
-				None => return Err(DecodeError::UnexpectedEnd.into()),
-
-				// NOTE: This does not make a copy!
-				// Bytes are immutable and ref counted.
-				Some(data) => fragment.write_chunk(data.bytes)?,
-			}
 		}
 
 		Ok(())
 	}
 
 	async fn run_source(mut self) -> Result<(), SessionError> {
-		loop {
+		// NOTE: This returns Closed when the source is closed.
-			// NOTE: This returns Closed when the source is closed.
+		while let Some(track) = self.source.next_track().await? {
-			let track = self.source.next_track().await?;
 			let name = track.name.clone();
 
 			let id = VarInt::from_u32(self.next.fetch_add(1, atomic::Ordering::SeqCst));
@@ -189,19 +144,13 @@ impl Subscriber {
 
 			let msg = message::Subscribe {
 				id,
-				namespace: self.control.ext.subscribe_split.then(|| "".to_string()),
+				namespace: "".to_string(),
 				name,
-
-				// TODO correctly support these
-				start_group: message::SubscribeLocation::Latest(VarInt::ZERO),
-				start_object: message::SubscribeLocation::Absolute(VarInt::ZERO),
-				end_group: message::SubscribeLocation::None,
-				end_object: message::SubscribeLocation::None,
-
-				params: Default::default(),
 			};
 
 			self.control.send(msg).await?;
 		}
+
+		Ok(())
 	}
 }

moq-transport/src/setup/client.rs

@@ -1,6 +1,6 @@
-use super::{Extensions, Role, Versions};
+use super::{Role, Versions};
 use crate::{
-	coding::{Decode, DecodeError, Encode, EncodeError, Params},
+	coding::{DecodeError, EncodeError},
 	VarInt,
 };
 
@@ -15,57 +15,29 @@ pub struct Client {
 	pub versions: Versions,
 
 	/// Indicate if the client is a publisher, a subscriber, or both.
+	// Proposal: moq-wg/moq-transport#151
 	pub role: Role,
-
-	/// A list of known/offered extensions.
-	pub extensions: Extensions,
-
-	/// Unknown parameters.
-	pub params: Params,
 }
 
 impl Client {
 	/// Decode a client setup message.
 	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 		let typ = VarInt::decode(r).await?;
-		if typ.into_inner() != 0x40 {
+		if typ.into_inner() != 1 {
-			return Err(DecodeError::InvalidMessage(typ));
+			return Err(DecodeError::InvalidType(typ));
 		}
 
 		let versions = Versions::decode(r).await?;
-		let mut params = Params::decode(r).await?;
+		let role = Role::decode(r).await?;
 
-		let role = params
+		Ok(Self { versions, role })
-			.get::<Role>(VarInt::from_u32(0))
-			.await?
-			.ok_or(DecodeError::MissingParameter)?;
-
-		// Make sure the PATH parameter isn't used
-		// TODO: This assumes WebTransport support only
-		if params.has(VarInt::from_u32(1)) {
-			return Err(DecodeError::InvalidParameter);
-		}
-
-		let extensions = Extensions::load(&mut params).await?;
-
-		Ok(Self {
-			versions,
-			role,
-			extensions,
-			params,
-		})
 	}
 
 	/// Encode a server setup message.
 	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		VarInt::from_u32(0x40).encode(w).await?;
+		VarInt::from_u32(1).encode(w).await?;
 		self.versions.encode(w).await?;
-
+		self.role.encode(w).await?;
-		let mut params = self.params.clone();
-		params.set(VarInt::from_u32(0), self.role).await?;
-		self.extensions.store(&mut params).await?;
-
-		params.encode(w).await?;
 
 		Ok(())
 	}

moq-transport/src/setup/extension.rs

@@ -1,84 +0,0 @@
-use tokio::io::{AsyncRead, AsyncWrite};
-
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, Params};
-use crate::session::SessionError;
-use crate::VarInt;
-use paste::paste;
-
-/// This is a custom extension scheme to allow/require draft PRs.
-///
-/// By convention, the extension number is the PR number + 0xe0000.
-
-macro_rules! extensions {
-    {$($name:ident = $val:expr,)*} => {
-		#[derive(Clone, Default, Debug)]
-		pub struct Extensions {
-			$(
-				pub $name: bool,
-			)*
-		}
-
-		impl Extensions {
-			pub async fn load(params: &mut Params) -> Result<Self, DecodeError> {
-				let mut extensions = Self::default();
-
-				$(
-					if let Some(_) = params.get::<ExtensionExists>(VarInt::from_u32($val)).await? {
-						extensions.$name = true
-					}
-				)*
-
-				Ok(extensions)
-			}
-
-			pub async fn store(&self, params: &mut Params) -> Result<(), EncodeError> {
-				$(
-					if self.$name {
-						params.set(VarInt::from_u32($val), ExtensionExists{}).await?;
-					}
-				)*
-
-				Ok(())
-			}
-
-			paste! {
-				$(
-					pub fn [<require_ $name>](&self) -> Result<(), SessionError> {
-						match self.$name {
-							true => Ok(()),
-							false => Err(SessionError::RequiredExtension(VarInt::from_u32($val))),
-						}
-					}
-				)*
-			}
-		}
-	}
-}
-
-struct ExtensionExists;
-
-#[async_trait::async_trait]
-impl Decode for ExtensionExists {
-	async fn decode<R: AsyncRead>(_r: &mut R) -> Result<Self, DecodeError> {
-		Ok(ExtensionExists {})
-	}
-}
-
-#[async_trait::async_trait]
-impl Encode for ExtensionExists {
-	async fn encode<W: AsyncWrite>(&self, _w: &mut W) -> Result<(), EncodeError> {
-		Ok(())
-	}
-}
-
-extensions! {
-	// required for publishers: OBJECT contains expires VarInt in seconds: https://github.com/moq-wg/moq-transport/issues/249
-	// TODO write up a PR
-	object_expires = 0xe00f9,
-
-	// required: SUBSCRIBE chooses track ID: https://github.com/moq-wg/moq-transport/pull/258
-	subscriber_id = 0xe0102,
-
-	// optional: SUBSCRIBE contains namespace/name tuple: https://github.com/moq-wg/moq-transport/pull/277
-	subscribe_split = 0xe0115,
-}

moq-transport/src/setup/mod.rs

@@ -5,13 +5,11 @@
 //! Both sides negotate the [Version] and [Role].
 
 mod client;
-mod extension;
 mod role;
 mod server;
 mod version;
 
 pub use client::*;
-pub use extension::*;
 pub use role::*;
 pub use server::*;
 pub use version::*;

moq-transport/src/setup/role.rs

@@ -1,6 +1,6 @@
 use crate::coding::{AsyncRead, AsyncWrite};
 
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
+use crate::coding::{DecodeError, EncodeError, VarInt};
 
 /// Indicates the endpoint is a publisher, subscriber, or both.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -36,9 +36,9 @@ impl Role {
 impl From<Role> for VarInt {
 	fn from(r: Role) -> Self {
 		VarInt::from_u32(match r {
-			Role::Publisher => 0x1,
+			Role::Publisher => 0x0,
-			Role::Subscriber => 0x2,
+			Role::Subscriber => 0x1,
-			Role::Both => 0x3,
+			Role::Both => 0x2,
 		})
 	}
 }
@@ -48,27 +48,23 @@ impl TryFrom<VarInt> for Role {
 
 	fn try_from(v: VarInt) -> Result<Self, Self::Error> {
 		match v.into_inner() {
-			0x1 => Ok(Self::Publisher),
+			0x0 => Ok(Self::Publisher),
-			0x2 => Ok(Self::Subscriber),
+			0x1 => Ok(Self::Subscriber),
-			0x3 => Ok(Self::Both),
+			0x2 => Ok(Self::Both),
-			_ => Err(DecodeError::InvalidRole(v)),
+			_ => Err(DecodeError::InvalidType(v)),
 		}
 	}
 }
 
-#[async_trait::async_trait]
+impl Role {
-impl Decode for Role {
 	/// Decode the role.
-	async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 		let v = VarInt::decode(r).await?;
 		v.try_into()
 	}
-}
 
-#[async_trait::async_trait]
-impl Encode for Role {
 	/// Encode the role.
-	async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
 		VarInt::from(*self).encode(w).await
 	}
 }

moq-transport/src/setup/server.rs

@@ -1,6 +1,6 @@
-use super::{Extensions, Role, Version};
+use super::{Role, Version};
 use crate::{
-	coding::{Decode, DecodeError, Encode, EncodeError, Params},
+	coding::{DecodeError, EncodeError},
 	VarInt,
 };
 
@@ -17,54 +17,27 @@ pub struct Server {
 	/// Indicate if the server is a publisher, a subscriber, or both.
 	// Proposal: moq-wg/moq-transport#151
 	pub role: Role,
-
-	/// Custom extensions.
-	pub extensions: Extensions,
-
-	/// Unknown parameters.
-	pub params: Params,
 }
 
 impl Server {
 	/// Decode the server setup.
 	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 		let typ = VarInt::decode(r).await?;
-		if typ.into_inner() != 0x41 {
+		if typ.into_inner() != 2 {
-			return Err(DecodeError::InvalidMessage(typ));
+			return Err(DecodeError::InvalidType(typ));
 		}
 
 		let version = Version::decode(r).await?;
-		let mut params = Params::decode(r).await?;
+		let role = Role::decode(r).await?;
 
-		let role = params
+		Ok(Self { version, role })
-			.get::<Role>(VarInt::from_u32(0))
-			.await?
-			.ok_or(DecodeError::MissingParameter)?;
-
-		// Make sure the PATH parameter isn't used
-		if params.has(VarInt::from_u32(1)) {
-			return Err(DecodeError::InvalidParameter);
-		}
-
-		let extensions = Extensions::load(&mut params).await?;
-
-		Ok(Self {
-			version,
-			role,
-			extensions,
-			params,
-		})
 	}
 
 	/// Encode the server setup.
 	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
-		VarInt::from_u32(0x41).encode(w).await?;
+		VarInt::from_u32(2).encode(w).await?;
 		self.version.encode(w).await?;
-
+		self.role.encode(w).await?;
-		let mut params = self.params.clone();
-		params.set(VarInt::from_u32(0), self.role).await?;
-		self.extensions.store(&mut params).await?;
-		params.encode(w).await?;
 
 		Ok(())
 	}

moq-transport/src/setup/version.rs

@@ -1,4 +1,4 @@
-use crate::coding::{Decode, DecodeError, Encode, EncodeError, VarInt};
+use crate::coding::{DecodeError, EncodeError, VarInt};
 
 use crate::coding::{AsyncRead, AsyncWrite};
 
@@ -9,12 +9,9 @@ use std::ops::Deref;
 pub struct Version(pub VarInt);
 
 impl Version {
-	/// https://www.ietf.org/archive/id/draft-ietf-moq-transport-00.html
+	/// <https://www.ietf.org/archive/id/draft-ietf-moq-transport-00.html>
 	pub const DRAFT_00: Version = Version(VarInt::from_u32(0xff00));
 
-	/// https://www.ietf.org/archive/id/draft-ietf-moq-transport-01.html
-	pub const DRAFT_01: Version = Version(VarInt::from_u32(0xff01));
-
 	/// Fork of draft-ietf-moq-transport-00.
 	///
 	/// Rough list of differences:
@@ -59,18 +56,6 @@ impl Version {
 	/// # GROUP
 	/// - GROUP concept was removed, replaced with OBJECT as a QUIC stream.
 	pub const KIXEL_00: Version = Version(VarInt::from_u32(0xbad00));
-
-	/// Fork of draft-ietf-moq-transport-01.
-	///
-	/// Most of the KIXEL_00 changes made it into the draft, or were reverted.
-	/// This was only used for a short time until extensions were created.
-	///
-	/// - SUBSCRIBE contains a separate track namespace and track name field (accidental revert). [#277](https://github.com/moq-wg/moq-transport/pull/277)
-	/// - SUBSCRIBE contains the `track_id` instead of SUBSCRIBE_OK. [#145](https://github.com/moq-wg/moq-transport/issues/145)
-	/// - SUBSCRIBE_* reference `track_id` the instead of the `track_full_name`. [#145](https://github.com/moq-wg/moq-transport/issues/145)
-	/// - OBJECT `priority` is still a VarInt, but the max value is a u32 (implementation reasons)
-	/// - OBJECT messages within the same `group` MUST be on the same QUIC stream.
-	pub const KIXEL_01: Version = Version(VarInt::from_u32(0xbad01));
 }
 
 impl From<VarInt> for Version {
@@ -103,10 +88,9 @@ impl Version {
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct Versions(Vec<Version>);
 
-#[async_trait::async_trait]
+impl Versions {
-impl Decode for Versions {
 	/// Decode the version list.
-	async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
+	pub async fn decode<R: AsyncRead>(r: &mut R) -> Result<Self, DecodeError> {
 		let count = VarInt::decode(r).await?.into_inner();
 		let mut vs = Vec::new();
 
@@ -117,12 +101,9 @@ impl Decode for Versions {
 
 		Ok(Self(vs))
 	}
-}
 
-#[async_trait::async_trait]
-impl Encode for Versions {
 	/// Encode the version list.
-	async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
+	pub async fn encode<W: AsyncWrite>(&self, w: &mut W) -> Result<(), EncodeError> {
 		let size: VarInt = self.0.len().try_into()?;
 		size.encode(w).await?;
 
@@ -147,9 +128,3 @@ impl From<Vec<Version>> for Versions {
 		Self(vs)
 	}
 }
-
-impl<const N: usize> From<[Version; N]> for Versions {
-	fn from(vs: [Version; N]) -> Self {
-		Self(vs.to_vec())
-	}
-}