Start implementing joining fetch.

Author: kixelated

js/moq/src/ietf/connection.ts

@@ -7,7 +7,7 @@ import { unreachable } from "../util/index.ts";
 import * as Control from "./control.ts";
 import { Fetch, FetchCancel, FetchError, FetchOk } from "./fetch.ts";
 import { GoAway } from "./goaway.ts";
-import { Group } from "./object.ts";
+import { GroupHeader } from "./group.ts";
 import { Publish, PublishError, PublishOk } from "./publish.ts";
 import {
 	PublishNamespace,
@@ -263,7 +263,7 @@ export class Connection implements Established {
 	async #runObjectStream(stream: Reader) {
 		try {
 			// we don't support other stream types yet
-			const header = await Group.decode(stream);
+			const header = await GroupHeader.decode(stream);
 			console.debug("received group header", header);
 			await this.#subscriber.handleGroup(header, stream);
 		} catch (err) {

js/moq/src/ietf/fetch.ts

@@ -1,44 +1,71 @@
 import type * as Path from "../path.ts";
 import type { Reader, Writer } from "../stream.ts";
+import { Location } from "./location.js";
 import * as Message from "./message.ts";
+import * as Namespace from "./namespace.ts";
+import { Parameters } from "./parameters.ts";
+
+export const FetchType = {
+	Standalone: 0x1,
+	Relative: 0x2,
+	Absolute: 0x3,
+} as const;
+
+export type FetchType =
+	| {
+			type: typeof FetchType.Standalone;
+			namespace: Path.Valid;
+			track: string;
+			start: Location;
+			end: Location;
+	  }
+	| {
+			type: typeof FetchType.Relative;
+			subscribeId: bigint;
+			groupOffset: number;
+	  }
+	| {
+			type: typeof FetchType.Absolute;
+			subscribeId: bigint;
+			groupId: number;
+	  };
 
 export class Fetch {
 	static id = 0x16;
 
 	requestId: bigint;
-	trackNamespace: Path.Valid;
-	trackName: string;
 	subscriberPriority: number;
 	groupOrder: number;
-	startGroup: bigint;
-	startObject: bigint;
-	endGroup: bigint;
-	endObject: bigint;
-
-	constructor(
-		requestId: bigint,
-		trackNamespace: Path.Valid,
-		trackName: string,
-		subscriberPriority: number,
-		groupOrder: number,
-		startGroup: bigint,
-		startObject: bigint,
-		endGroup: bigint,
-		endObject: bigint,
-	) {
+	fetchType: FetchType;
+
+	constructor(requestId: bigint, subscriberPriority: number, groupOrder: number, fetchType: FetchType) {
 		this.requestId = requestId;
-		this.trackNamespace = trackNamespace;
-		this.trackName = trackName;
 		this.subscriberPriority = subscriberPriority;
 		this.groupOrder = groupOrder;
-		this.startGroup = startGroup;
-		this.startObject = startObject;
-		this.endGroup = endGroup;
-		this.endObject = endObject;
-	}
-
-	async #encode(_w: Writer): Promise<void> {
-		throw new Error("FETCH messages are not supported");
+		this.fetchType = fetchType;
+	}
+
+	async #encode(w: Writer): Promise<void> {
+		await w.u62(this.requestId);
+		await w.u8(this.subscriberPriority);
+		await w.u8(this.groupOrder);
+		await w.u53(this.fetchType.type);
+		if (this.fetchType.type === FetchType.Standalone) {
+			await Namespace.encode(w, this.fetchType.namespace);
+			await w.string(this.fetchType.track);
+			this.fetchType.start.encode(w);
+			this.fetchType.end.encode(w);
+		} else if (this.fetchType.type === FetchType.Relative) {
+			await w.u62(this.fetchType.subscribeId);
+			await w.u53(this.fetchType.groupOffset);
+		} else if (this.fetchType.type === FetchType.Absolute) {
+			await w.u62(this.fetchType.subscribeId);
+			await w.u53(this.fetchType.groupId);
+		} else {
+			const fetchType: never = this.fetchType;
+			throw new Error(`unknown fetch type: ${fetchType}`);
+		}
+		await w.u53(0); // no parameters
 	}
 
 	async encode(w: Writer): Promise<void> {
@@ -49,22 +76,74 @@ export class Fetch {
 		return Message.decode(r, Fetch.#decode);
 	}
 
-	static async #decode(_r: Reader): Promise<Fetch> {
-		throw new Error("FETCH messages are not supported");
+	static async #decode(r: Reader): Promise<Fetch> {
+		const requestId = await r.u62();
+		const subscriberPriority = await r.u8();
+		const groupOrder = await r.u8();
+		const fetchType = await r.u53();
+
+		if (fetchType === FetchType.Standalone) {
+			const namespace = await Namespace.decode(r);
+			const track = await r.string();
+			const start = await Location.decode(r);
+			const end = await Location.decode(r);
+			await Parameters.decode(r); // ignore parameters
+			return new Fetch(requestId, subscriberPriority, groupOrder, {
+				type: FetchType.Standalone,
+				namespace,
+				track,
+				start,
+				end,
+			});
+		}
+
+		if (fetchType === FetchType.Relative) {
+			const subscribeId = await r.u62();
+			const groupOffset = await r.u53();
+			await Parameters.decode(r); // ignore parameters
+			return new Fetch(requestId, subscriberPriority, groupOrder, {
+				type: FetchType.Relative,
+				subscribeId,
+				groupOffset,
+			});
+		}
+
+		if (fetchType === FetchType.Absolute) {
+			const subscribeId = await r.u62();
+			const groupId = await r.u53();
+			await Parameters.decode(r); // ignore parameters
+			return new Fetch(requestId, subscriberPriority, groupOrder, {
+				type: FetchType.Absolute,
+				subscribeId,
+				groupId,
+			});
+		}
+
+		throw new Error(`unknown fetch type: ${fetchType}`);
 	}
 }
 
 export class FetchOk {
 	static id = 0x18;
 
 	requestId: bigint;
+	groupOrder: number;
+	endOfTrack: boolean;
+	endLocation: Location;
 
-	constructor(requestId: bigint) {
+	constructor(requestId: bigint, groupOrder: number, endOfTrack: boolean, endLocation: Location) {
 		this.requestId = requestId;
+		this.groupOrder = groupOrder;
+		this.endOfTrack = endOfTrack;
+		this.endLocation = endLocation;
 	}
 
-	async #encode(_w: Writer): Promise<void> {
-		throw new Error("FETCH_OK messages are not supported");
+	async #encode(w: Writer): Promise<void> {
+		await w.u62(this.requestId);
+		await w.u8(this.groupOrder);
+		await w.bool(this.endOfTrack);
+		this.endLocation.encode(w);
+		await w.u53(0); // no parameters
 	}
 
 	async encode(w: Writer): Promise<void> {
@@ -75,8 +154,13 @@ export class FetchOk {
 		return Message.decode(r, FetchOk.#decode);
 	}
 
-	static async #decode(_r: Reader): Promise<FetchOk> {
-		throw new Error("FETCH_OK messages are not supported");
+	static async #decode(r: Reader): Promise<FetchOk> {
+		const requestId = await r.u62();
+		const groupOrder = await r.u8();
+		const endOfTrack = await r.bool();
+		const endLocation = await Location.decode(r);
+		await Parameters.decode(r); // ignore parameters
+		return new FetchOk(requestId, groupOrder, endOfTrack, endLocation);
 	}
 }
 
@@ -93,8 +177,10 @@ export class FetchError {
 		this.reasonPhrase = reasonPhrase;
 	}
 
-	async #encode(_w: Writer): Promise<void> {
-		throw new Error("FETCH_ERROR messages are not supported");
+	async #encode(w: Writer): Promise<void> {
+		await w.u62(this.requestId);
+		await w.u53(this.errorCode);
+		await w.string(this.reasonPhrase);
 	}
 
 	async encode(w: Writer): Promise<void> {
@@ -105,8 +191,11 @@ export class FetchError {
 		return Message.decode(r, FetchError.#decode);
 	}
 
-	static async #decode(_r: Reader): Promise<FetchError> {
-		throw new Error("FETCH_ERROR messages are not supported");
+	static async #decode(r: Reader): Promise<FetchError> {
+		const requestId = await r.u62();
+		const errorCode = await r.u53();
+		const reasonPhrase = await r.string();
+		return new FetchError(requestId, errorCode, reasonPhrase);
 	}
 }
 
@@ -119,8 +208,8 @@ export class FetchCancel {
 		this.requestId = requestId;
 	}
 
-	async #encode(_w: Writer): Promise<void> {
-		throw new Error("FETCH_CANCEL messages are not supported");
+	async #encode(w: Writer): Promise<void> {
+		await w.u62(this.requestId);
 	}
 
 	async encode(w: Writer): Promise<void> {
@@ -131,7 +220,8 @@ export class FetchCancel {
 		return Message.decode(r, FetchCancel.#decode);
 	}
 
-	static async #decode(_r: Reader): Promise<FetchCancel> {
-		throw new Error("FETCH_CANCEL messages are not supported");
+	static async #decode(r: Reader): Promise<FetchCancel> {
+		const requestId = await r.u62();
+		return new FetchCancel(requestId);
 	}
 }

js/moq/src/ietf/frame.ts

@@ -0,0 +1,68 @@
+import type { Reader, Writer } from "../stream.ts";
+import type { GroupFlags } from "./group.ts";
+
+const GROUP_END = 0x03;
+
+export class Frame {
+	// undefined means end of group
+	payload?: Uint8Array;
+
+	constructor(payload?: Uint8Array) {
+		this.payload = payload;
+	}
+
+	async encode(w: Writer, flags: GroupFlags): Promise<void> {
+		await w.u53(0); // id_delta = 0
+
+		if (flags.hasExtensions) {
+			await w.u53(0); // extensions length = 0
+		}
+
+		if (this.payload !== undefined) {
+			await w.u53(this.payload.byteLength);
+
+			if (this.payload.byteLength === 0) {
+				await w.u53(0); // status = normal
+			} else {
+				await w.write(this.payload);
+			}
+		} else {
+			await w.u53(0); // length = 0
+			await w.u53(GROUP_END);
+		}
+	}
+
+	static async decode(r: Reader, flags: GroupFlags): Promise<Frame> {
+		console.debug("reading frame delta");
+		const delta = await r.u53();
+		console.debug("read frame delta", delta);
+		if (delta !== 0) {
+			console.warn(`object ID delta is not supported, ignoring: ${delta}`);
+		}
+
+		if (flags.hasExtensions) {
+			const extensionsLength = await r.u53();
+			// We don't care about extensions
+			await r.read(extensionsLength);
+		}
+
+		const payloadLength = await r.u53();
+
+		if (payloadLength > 0) {
+			const payload = await r.read(payloadLength);
+			return new Frame(payload);
+		}
+
+		const status = await r.u53();
+
+		if (flags.hasEnd) {
+			// Empty frame
+			if (status === 0) return new Frame(new Uint8Array(0));
+		} else if (status === 0 || status === GROUP_END) {
+			// TODO status === 0 should be an empty frame, but moq-rs seems to be sending it incorrectly on group end.
+			return new Frame();
+		}
+
+		throw new Error(`Unsupported object status: ${status}`);
+	}
+}