Merge pull request #40 from montevideo-tech/feature/fetch-control-messages

fetch control message

Author: englishm

lib/transport/client.ts

@@ -48,7 +48,7 @@ export class Client {
 
 		// Send the setup message.
 		await setup.send.client({
-			versions: [Setup.Version.DRAFT_06],
+			versions: [Setup.Version.DRAFT_07],
 			role: this.config.role,
 		})
 

lib/transport/control.ts

@@ -3,7 +3,7 @@ import { Reader, Writer } from "./stream"
 export type Message = Subscriber | Publisher
 
 // Sent by subscriber
-export type Subscriber = Subscribe | Unsubscribe | AnnounceOk | AnnounceError
+export type Subscriber = Subscribe | Unsubscribe | AnnounceOk | AnnounceError | Fetch | FetchCancel
 
 export function isSubscriber(m: Message): m is Subscriber {
 	return (
@@ -12,7 +12,7 @@ export function isSubscriber(m: Message): m is Subscriber {
 }
 
 // Sent by publisher
-export type Publisher = SubscribeOk | SubscribeError | SubscribeDone | Announce | Unannounce
+export type Publisher = SubscribeOk | SubscribeError | SubscribeDone | Announce | Unannounce | FetchOk | FetchError
 
 export function isPublisher(m: Message): m is Publisher {
 	return (
@@ -39,6 +39,10 @@ export enum Msg {
 	AnnounceError = "announce_error",
 	Unannounce = "unannounce",
 	GoAway = "go_away",
+	Fetch = "fetch",
+	FetchCancel = "fetch_cancel",
+	FetchOk = "fetch_ok",
+	FetchError = "fetch_error",
 }
 
 enum Id {
@@ -56,6 +60,10 @@ enum Id {
 	AnnounceError = 0x8,
 	Unannounce = 0x9,
 	GoAway = 0x10,
+	Fetch = 0x16,
+	FetchCancel = 0x17,
+	FetchOk = 0x18,
+	FetchError = 0x19,
 }
 
 export interface Subscribe {
@@ -158,6 +166,42 @@ export interface Unannounce {
 	namespace: string[]
 }
 
+export interface Fetch {
+	kind: Msg.Fetch
+	id: bigint
+	namespace: string[]
+	name: string
+	subscriber_priority: number
+	group_order: GroupOrder
+	start_group: number
+	start_object: number
+	end_group: number
+	end_object: number
+	params?: Parameters
+}
+
+export interface FetchOk {
+	kind: Msg.FetchOk
+	id: bigint
+	group_order: number
+	end_of_track: number
+	largest_group_id: bigint
+	largest_object_id: bigint
+	params?: Parameters
+}
+
+export interface FetchError {
+	kind: Msg.FetchError
+	id: bigint
+	code: bigint
+	reason: string
+}
+
+export interface FetchCancel {
+	kind: Msg.FetchCancel
+	id: bigint
+}
+
 export class Stream {
 	private decoder: Decoder
 	private encoder: Encoder
@@ -244,9 +288,15 @@ export class Decoder {
 				return Msg.Unannounce
 			case Id.GoAway:
 				return Msg.GoAway
+			case Id.Fetch:
+				return Msg.Fetch
+			case Id.FetchCancel:
+				return Msg.FetchCancel
+			case Id.FetchOk:
+				return Msg.FetchOk
+			case Id.FetchError:
+				return Msg.FetchError
 		}
-
-		throw new Error(`unknown control message type: ${t}`)
 	}
 
 	async message(): Promise<Message> {
@@ -259,7 +309,6 @@ export class Decoder {
 			case Msg.SubscribeError:
 				return this.subscribe_error()
 			case Msg.SubscribeDone:
-				return this.subscribe_done()
 			case Msg.Unsubscribe:
 				return this.unsubscribe()
 			case Msg.Announce:
@@ -272,6 +321,14 @@ export class Decoder {
 				return this.announce_error()
 			case Msg.GoAway:
 				throw new Error("TODO: implement go away")
+			case Msg.Fetch:
+				return this.fetch()
+			case Msg.FetchCancel:
+				return this.fetchCancel()
+			case Msg.FetchOk:
+				return this.fetchOk()
+			case Msg.FetchError:
+				return this.fetchError()
 		}
 	}
 
@@ -450,6 +507,50 @@ export class Decoder {
 			namespace: await this.r.tuple(),
 		}
 	}
+
+	private async fetch(): Promise<Fetch> {
+		return {
+			kind: Msg.Fetch,
+			id: await this.r.u62(),
+			namespace: await this.r.tuple(),
+			name: await this.r.string(),
+			subscriber_priority: await this.r.u8(),
+			group_order: await this.decodeGroupOrder(),
+			start_group: await this.r.u53(),
+			start_object: await this.r.u53(),
+			end_group: await this.r.u53(),
+			end_object: await this.r.u53(),
+			params: await this.parameters(),
+		}
+	}
+
+	private async fetchCancel(): Promise<FetchCancel> {
+		return {
+			kind: Msg.FetchCancel,
+			id: await this.r.u62(),
+		}
+	}
+
+	private async fetchOk(): Promise<FetchOk> {
+		return {
+			kind: Msg.FetchOk,
+			id: await this.r.u62(),
+			group_order: await this.r.u8(),
+			end_of_track: await this.r.u8(),
+			largest_group_id: await this.r.u62(),
+			largest_object_id: await this.r.u62(),
+			params: await this.parameters(),
+		}
+	}
+
+	private async fetchError(): Promise<FetchError> {
+		return {
+			kind: Msg.FetchError,
+			id: await this.r.u62(),
+			code: await this.r.u62(),
+			reason: await this.r.string(),
+		}
+	}
 }
 
 export class Encoder {
@@ -479,6 +580,14 @@ export class Encoder {
 				return this.announce_error(m)
 			case Msg.Unannounce:
 				return this.unannounce(m)
+			case Msg.Fetch:
+				return this.fetch(m)
+			case Msg.FetchCancel:
+				return this.fetchCancel(m)
+			case Msg.FetchOk:
+				return this.fetchOk(m)
+			case Msg.FetchError:
+				return this.fetchError(m)
 		}
 	}
 
@@ -669,4 +778,78 @@ export class Encoder {
 
 		return this.w.concatBuffer(paramFields)
 	}
+
+	async fetch(f: Fetch) {
+		const buffer = new Uint8Array(8)
+
+		const msgData = this.w.concatBuffer([
+			this.w.setVint62(buffer, f.id),
+			this.w.encodeTuple(buffer, f.namespace),
+			this.w.encodeString(buffer, f.name),
+			this.w.setUint8(buffer, f.subscriber_priority),
+			this.w.setUint8(buffer, f.group_order),
+			this.w.setVint53(buffer, f.start_group),
+			this.w.setVint53(buffer, f.start_object),
+			this.w.setVint53(buffer, f.end_group),
+			this.w.setVint53(buffer, f.end_object),
+			this.encodeParameters(buffer, f.params),
+		])
+
+		const messageType = this.w.setVint53(buffer, Id.Fetch)
+		const messageLength = this.w.setVint53(buffer, msgData.length)
+
+		for (const elem of [messageType, messageLength, msgData]) {
+			await this.w.write(elem)
+		}
+	}
+
+	async fetchCancel(fc: FetchCancel) {
+		const buffer = new Uint8Array(8)
+
+		const msgData = this.w.concatBuffer([this.w.setVint62(buffer, fc.id)])
+
+		const messageType = this.w.setVint53(buffer, Id.FetchCancel)
+		const messageLength = this.w.setVint53(buffer, msgData.length)
+
+		for (const elem of [messageType, messageLength, msgData]) {
+			await this.w.write(elem)
+		}
+	}
+
+	async fetchOk(fo: FetchOk) {
+		const buffer = new Uint8Array(8)
+
+		const msgData = this.w.concatBuffer([
+			this.w.setVint62(buffer, fo.id),
+			this.w.setUint8(buffer, fo.group_order),
+			this.w.setUint8(buffer, fo.end_of_track),
+			this.w.setVint62(buffer, fo.largest_group_id),
+			this.w.setVint62(buffer, fo.largest_object_id),
+			this.encodeParameters(buffer, fo.params),
+		])
+
+		const messageType = this.w.setVint53(buffer, Id.FetchOk)
+		const messageLength = this.w.setVint53(buffer, msgData.length)
+
+		for (const elem of [messageType, messageLength, msgData]) {
+			await this.w.write(elem)
+		}
+	}
+
+	async fetchError(fe: FetchError) {
+		const buffer = new Uint8Array(8)
+
+		const msgData = this.w.concatBuffer([
+			this.w.setVint62(buffer, fe.id),
+			this.w.setVint62(buffer, fe.code),
+			this.w.encodeString(buffer, fe.reason),
+		])
+
+		const messageType = this.w.setVint53(buffer, Id.FetchError)
+		const messageLength = this.w.setVint53(buffer, msgData.length)
+
+		for (const elem of [messageType, messageLength, msgData]) {
+			await this.w.write(elem)
+		}
+	}
 }

lib/transport/setup.ts

@@ -11,6 +11,7 @@ export enum Version {
 	DRAFT_04 = 0xff000004,
 	DRAFT_05 = 0xff000005,
 	DRAFT_06 = 0xff000006,
+	DRAFT_07 = 0xff000007,
 	KIXEL_00 = 0xbad00,
 	KIXEL_01 = 0xbad01,
 }