Transfork - Major rewrite (#117)

everything be changed

lib/common/async.ts

@@ -35,7 +35,7 @@ export class Watch<T> {
 
 	update(v: T | ((v: T) => T)) {
 		if (!this.#next.pending) {
-			throw new Error("already closed")
+			throw new Error("closed")
 		}
 
 		// If we're given a function, call it with the current value
@@ -53,6 +53,10 @@ export class Watch<T> {
 		this.#current[1] = undefined
 		this.#next.resolve(this.#current)
 	}
+
+	closed() {
+		return !this.#next.pending
+	}
 }
 
 // Wakes up a multiple consumers.
@@ -88,6 +92,7 @@ export class Queue<T> {
 	}
 
 	async push(v: T) {
+		if (this.#closed) throw new Error("closed")
 		const w = this.#stream.writable.getWriter()
 		await w.write(v)
 		w.releaseLock()

lib/common/hex.ts

@@ -0,0 +1,11 @@
+export function decode(str: string): Uint8Array {
+	const bytes = new Uint8Array(str.length / 2)
+	for (let i = 0; i < bytes.length; i += 1) {
+		bytes[i] = parseInt(str.slice(2 * i, 2 * i + 2), 16)
+	}
+	return bytes
+}
+
+export function encode(_bytes: Uint8Array): string {
+	throw "todo"
+}

lib/contribute/audio.ts

@@ -1,15 +1,67 @@
+import { Deferred } from "../common/async"
+import { Frame } from "../karp/frame"
+import { Group, Track } from "../transfork"
+import { Closed } from "../transfork/error"
+
 const SUPPORTED = [
 	// TODO support AAC
 	// "mp4a"
 	"Opus",
 ]
 
+export class Packer {
+	#source: MediaStreamTrackProcessor<AudioData>
+	#encoder: Encoder
+
+	#data: Track
+	#current?: Group
+
+	constructor(track: MediaStreamAudioTrack, encoder: Encoder, data: Track) {
+		this.#source = new MediaStreamTrackProcessor({ track })
+		this.#encoder = encoder
+		this.#data = data
+	}
+
+	async run() {
+		const output = new WritableStream({
+			write: (chunk) => this.#write(chunk),
+			close: () => this.#close(),
+			abort: (e) => this.#close(e),
+		})
+
+		return this.#source.readable.pipeThrough(this.#encoder.frames).pipeTo(output)
+	}
+
+	#write(frame: Frame) {
+		// TODO use a fixed interval instead of keyframes (audio)
+		// TODO actually just align with video
+		if (!this.#current || frame.type === "key") {
+			if (this.#current) {
+				this.#current.close()
+			}
+
+			this.#current = this.#data.appendGroup()
+		}
+
+		this.#current.writeFrame(frame.data)
+	}
+
+	#close(err?: any) {
+		const closed = Closed.from(err)
+		if (this.#current) {
+			this.#current.close(closed)
+		}
+
+		this.#data.close(closed)
+	}
+}
+
 export class Encoder {
 	#encoder!: AudioEncoder
 	#encoderConfig: AudioEncoderConfig
-	#decoderConfig?: AudioDecoderConfig
+	#decoderConfig = new Deferred<AudioDecoderConfig>()
 
-	frames: TransformStream<AudioData, AudioDecoderConfig | EncodedAudioChunk>
+	frames: TransformStream<AudioData, EncodedAudioChunk>
 
 	constructor(config: AudioEncoderConfig) {
 		this.#encoderConfig = config
@@ -21,7 +73,7 @@ export class Encoder {
 		})
 	}
 
-	#start(controller: TransformStreamDefaultController<AudioDecoderConfig | EncodedAudioChunk>) {
+	#start(controller: TransformStreamDefaultController<EncodedAudioChunk>) {
 		this.#encoder = new AudioEncoder({
 			output: (frame, metadata) => {
 				this.#enqueue(controller, frame, metadata)
@@ -40,17 +92,16 @@ export class Encoder {
 	}
 
 	#enqueue(
-		controller: TransformStreamDefaultController<AudioDecoderConfig | EncodedAudioChunk>,
+		controller: TransformStreamDefaultController<EncodedAudioChunk>,
 		frame: EncodedAudioChunk,
 		metadata?: EncodedAudioChunkMetadata,
 	) {
 		const config = metadata?.decoderConfig
-		if (config && !this.#decoderConfig) {
+		if (config && !this.#decoderConfig.pending) {
 			const config = metadata.decoderConfig
 			if (!config) throw new Error("missing decoder config")
 
-			controller.enqueue(config)
-			this.#decoderConfig = config
+			this.#decoderConfig.resolve(config)
 		}
 
 		controller.enqueue(frame)
@@ -72,4 +123,8 @@ export class Encoder {
 	get config() {
 		return this.#encoderConfig
 	}
+
+	async decoderConfig(): Promise<AudioDecoderConfig> {
+		return await this.#decoderConfig.promise
+	}
 }