Maybe JS joining fetch.

Author: kixelated

js/moq/src/ietf/connection.ts

@@ -5,7 +5,7 @@ import * as Path from "../path.js";
 import { type Reader, Readers, type Stream } from "../stream.ts";
 import { unreachable } from "../util/index.ts";
 import * as Control from "./control.ts";
-import { Fetch, FetchCancel, FetchError, FetchOk } from "./fetch.ts";
+import { Fetch, FetchCancel, FetchError, FetchHeader, FetchOk } from "./fetch.ts";
 import { GoAway } from "./goaway.ts";
 import { GroupHeader } from "./group.ts";
 import { Publish, PublishError, PublishOk } from "./publish.ts";
@@ -183,13 +183,13 @@ export class Connection implements Established {
 				} else if (msg instanceof PublishError) {
 					throw new Error("PUBLISH_ERROR messages are not supported");
 				} else if (msg instanceof Fetch) {
-					throw new Error("FETCH messages are not supported");
+					await this.#publisher.handleFetch(msg);
 				} else if (msg instanceof FetchOk) {
-					throw new Error("FETCH_OK messages are not supported");
+					this.#subscriber.handleFetchOk(msg);
 				} else if (msg instanceof FetchError) {
-					throw new Error("FETCH_ERROR messages are not supported");
+					this.#subscriber.handleFetchError(msg);
 				} else if (msg instanceof FetchCancel) {
-					throw new Error("FETCH_CANCEL messages are not supported");
+					this.#publisher.handleFetchCancel(msg);
 				} else if (msg instanceof MaxRequestId) {
 					this.#control.maxRequestId(msg.requestId);
 				} else if (msg instanceof RequestsBlocked) {
@@ -262,10 +262,17 @@ export class Connection implements Established {
 	 */
 	async #runObjectStream(stream: Reader) {
 		try {
-			// we don't support other stream types yet
-			const header = await GroupHeader.decode(stream);
-			console.debug("received group header", header);
-			await this.#subscriber.handleGroup(header, stream);
+			// TODO support varints for stream type; this is not correct
+			const typ = (await stream.peek(1))[0];
+			if (typ === FetchHeader.id) {
+				const fetch = await FetchHeader.decode(stream);
+				console.debug("received fetch header", fetch);
+				await this.#subscriber.handleFetch(fetch, stream);
+			} else {
+				const header = await GroupHeader.decode(stream);
+				console.debug("received group header", header);
+				await this.#subscriber.handleGroup(header, stream);
+			}
 		} catch (err) {
 			console.error("error processing object stream", err);
 		}

js/moq/src/ietf/fetch.ts

@@ -1,10 +1,13 @@
 import type * as Path from "../path.ts";
 import type { Reader, Writer } from "../stream.ts";
+import { GroupOrder } from "./group.ts";
 import { Location } from "./location.js";
 import * as Message from "./message.ts";
 import * as Namespace from "./namespace.ts";
 import { Parameters } from "./parameters.ts";
 
+const FETCH_END = 0x03;
+
 export const FetchType = {
 	Standalone: 0x1,
 	Relative: 0x2,
@@ -35,10 +38,10 @@ export class Fetch {
 
 	requestId: bigint;
 	subscriberPriority: number;
-	groupOrder: number;
+	groupOrder: GroupOrder;
 	fetchType: FetchType;
 
-	constructor(requestId: bigint, subscriberPriority: number, groupOrder: number, fetchType: FetchType) {
+	constructor(requestId: bigint, subscriberPriority: number, groupOrder: GroupOrder, fetchType: FetchType) {
 		this.requestId = requestId;
 		this.subscriberPriority = subscriberPriority;
 		this.groupOrder = groupOrder;
@@ -48,7 +51,7 @@ export class Fetch {
 	async #encode(w: Writer): Promise<void> {
 		await w.u62(this.requestId);
 		await w.u8(this.subscriberPriority);
-		await w.u8(this.groupOrder);
+		await this.groupOrder.encode(w);
 		await w.u53(this.fetchType.type);
 		if (this.fetchType.type === FetchType.Standalone) {
 			await Namespace.encode(w, this.fetchType.namespace);
@@ -79,7 +82,7 @@ export class Fetch {
 	static async #decode(r: Reader): Promise<Fetch> {
 		const requestId = await r.u62();
 		const subscriberPriority = await r.u8();
-		const groupOrder = await r.u8();
+		const groupOrder = await GroupOrder.decode(r);
 		const fetchType = await r.u53();
 
 		if (fetchType === FetchType.Standalone) {
@@ -127,11 +130,11 @@ export class FetchOk {
 	static id = 0x18;
 
 	requestId: bigint;
-	groupOrder: number;
+	groupOrder: GroupOrder;
 	endOfTrack: boolean;
 	endLocation: Location;
 
-	constructor(requestId: bigint, groupOrder: number, endOfTrack: boolean, endLocation: Location) {
+	constructor(requestId: bigint, groupOrder: GroupOrder, endOfTrack: boolean, endLocation: Location) {
 		this.requestId = requestId;
 		this.groupOrder = groupOrder;
 		this.endOfTrack = endOfTrack;
@@ -140,7 +143,7 @@ export class FetchOk {
 
 	async #encode(w: Writer): Promise<void> {
 		await w.u62(this.requestId);
-		await w.u8(this.groupOrder);
+		await this.groupOrder.encode(w);
 		await w.bool(this.endOfTrack);
 		this.endLocation.encode(w);
 		await w.u53(0); // no parameters
@@ -156,7 +159,7 @@ export class FetchOk {
 
 	static async #decode(r: Reader): Promise<FetchOk> {
 		const requestId = await r.u62();
-		const groupOrder = await r.u8();
+		const groupOrder = await GroupOrder.decode(r);
 		const endOfTrack = await r.bool();
 		const endLocation = await Location.decode(r);
 		await Parameters.decode(r); // ignore parameters
@@ -225,3 +228,88 @@ export class FetchCancel {
 		return new FetchCancel(requestId);
 	}
 }
+
+export class FetchHeader {
+	static id = 0x5;
+
+	requestId: bigint;
+
+	constructor(requestId: bigint) {
+		this.requestId = requestId;
+	}
+
+	async encode(w: Writer): Promise<void> {
+		await w.u62(this.requestId);
+	}
+
+	static async decode(r: Reader): Promise<FetchHeader> {
+		const requestId = await r.u62();
+		return new FetchHeader(requestId);
+	}
+}
+
+export class FetchObject {
+	groupId: number;
+	subgroupId: number;
+	objectId: number;
+	publisherPriority: number;
+	payload?: Uint8Array;
+
+	constructor(
+		groupId: number,
+		subgroupId: number,
+		objectId: number,
+		publisherPriority: number,
+		payload?: Uint8Array,
+	) {
+		this.groupId = groupId;
+		this.subgroupId = subgroupId;
+		this.objectId = objectId;
+		this.publisherPriority = publisherPriority;
+		this.payload = payload;
+	}
+
+	async encode(w: Writer): Promise<void> {
+		await w.u53(this.groupId);
+		await w.u53(this.subgroupId);
+		await w.u53(this.objectId);
+		await w.u8(this.publisherPriority);
+		await w.u53(0); // no extension headers
+
+		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); // no payload, length = 0
+			await w.u53(FETCH_END); // no payload, status = end
+		}
+	}
+
+	static async decode(r: Reader): Promise<FetchObject> {
+		const groupId = await r.u53();
+		const subgroupId = await r.u53();
+		const objectId = await r.u53();
+		const publisherPriority = await r.u8();
+		const payloadLength = await r.u53();
+
+		let payload: Uint8Array | undefined;
+		if (payloadLength === 0) {
+			const status = await r.u53();
+			if (status === 0) {
+				payload = new Uint8Array(0);
+			} else if (status === FETCH_END) {
+				payload = undefined;
+			} else {
+				throw new Error(`unexpected status: ${status}`);
+			}
+		} else {
+			payload = await r.read(payloadLength);
+		}
+
+		return new FetchObject(groupId, subgroupId, objectId, publisherPriority, payload);
+	}
+}

js/moq/src/ietf/frame.ts

@@ -1,10 +1,34 @@
 import type { Reader, Writer } from "../stream";
 
-export const GroupOrder = {
-	Any: 0x0,
-	Ascending: 0x1,
-	Descending: 0x2,
-} as const;
+export class GroupOrder {
+	#value: number;
+
+	private constructor(value: number) {
+		this.#value = value;
+	}
+
+	static readonly Any = new GroupOrder(0x0);
+	static readonly Ascending = new GroupOrder(0x1);
+	static readonly Descending = new GroupOrder(0x2);
+
+	async encode(w: Writer): Promise<void> {
+		await w.u8(this.#value);
+	}
+
+	static async decode(r: Reader): Promise<GroupOrder> {
+		const value = await r.u8();
+		switch (value) {
+			case 0x0:
+				return GroupOrder.Any;
+			case 0x1:
+				return GroupOrder.Ascending;
+			case 0x2:
+				return GroupOrder.Descending;
+			default:
+				throw new Error(`Invalid GroupOrder: ${value}`);
+		}
+	}
+}
 
 export interface GroupFlags {
 	hasExtensions: boolean;
@@ -80,3 +104,67 @@ export class GroupHeader {
 		return new GroupHeader(trackAlias, groupId, subGroupId, publisherPriority, flags);
 	}
 }
+
+const GROUP_END = 0x03;
+
+export class GroupObject {
+	id_delta: number;
+
+	// undefined means end of group
+	payload?: Uint8Array;
+
+	constructor(id_delta: number, payload?: Uint8Array) {
+		this.id_delta = id_delta;
+		this.payload = payload;
+	}
+
+	async encode(w: Writer, flags: GroupFlags): Promise<void> {
+		await w.u53(this.id_delta);
+
+		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<GroupObject> {
+		const delta = await r.u53();
+
+		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 GroupObject(delta, payload);
+		}
+
+		const status = await r.u53();
+
+		if (status === 0) {
+			return new GroupObject(delta, new Uint8Array(0));
+		}
+
+		if (!flags.hasEnd && status === 3) {
+			return new GroupObject(delta);
+		}
+
+		throw new Error(`Unsupported object status: ${status}`);
+	}
+}

js/moq/src/ietf/group.ts

@@ -1,6 +1,5 @@
 export * from "./connection.ts";
 export * from "./control.ts";
-export * from "./frame.ts";
 export * from "./goaway.ts";
 export * from "./parameters.ts";
 export * from "./publish.ts";