rob
/
moq-rs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Making good progress on WebCodecs.

This commit is contained in:
Luke Curley 2023-03-28 07:54:41 +09:00
parent cc00a79881
commit 805f6ca392
22 changed files with 1159 additions and 5120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4235
player/package-lock.json generated
View File

File diff suppressed because it is too large Load Diff

5
player/package.json
View File

@ -7,7 +7,8 @@
},
"devDependencies": {
"@parcel/validator-typescript": "^2.6.0",
"parcel": "^2.6.0",
"typescript": ">=3.0.0"
"parcel": "^2.8.0",
"typescript": ">=3.0.0",
"@types/dom-webcodecs": "^0.1.6"
}
}

120
player/src/audio/decoder.ts Normal file
View File

@ -0,0 +1,120 @@
import * as Message from "./message";
import { InitParser } from "../mp4/init";
import { Renderer } from "./renderer"
import { MP4New, MP4Sample, MP4ArrayBuffer } from "../mp4/index"
export class Decoder {
tracks: Map<string, InitParser>;
renderer: Renderer;
constructor(renderer: Renderer) {
this.tracks = new Map();
this.renderer = renderer;
}
async init(msg: Message.Init) {
let track = this.tracks.get(msg.track);
if (!track) {
track = new InitParser()
this.tracks.set(msg.track, track)
}
while (1) {
const data = await msg.stream.read()
if (!data) break
track.init(data)
}
// TODO this will hang on incomplete data
const init = await track.ready;
const info = init.info;
if (info.audioTracks.length != 1 || info.videoTracks.length != 0) {
throw new Error("expected a single audio track")
}
}
async decode(msg: Message.Segment) {
let track = this.tracks.get(msg.track);
if (!track) {
track = new InitParser()
this.tracks.set(msg.track, track)
}
// Wait for the init segment to be fully received and parsed
const init = await track.ready;
const info = init.info;
const video = info.videoTracks[0]
const decoder = new AudioDecoder({
output: (frame: AudioFrame) => {
this.renderer.push(frame)
},
error: (err: Error) => {
console.warn(err)
}
});
decoder.configure({
codec: info.mime,
// TODO what else?
// optimizeForLatency: true
})
const input = MP4New();
input.onSamples = (id: number, user: any, samples: MP4Sample[]) => {
for (let sample of samples) {
const timestamp = sample.dts / (1000 / info.timescale) // milliseconds
decoder.decode(new EncodedAudioChunk({
data: sample.data,
duration: sample.duration,
timestamp: timestamp,
}))
}
}
input.onReady = (info: any) => {
input.setExtractionOptions(info.tracks[0].id, {}, { nbSamples: 1 });
input.start();
}
let offset = 0
// MP4box requires us to reparse the init segment unfortunately
for (let raw of track.raw) {
offset = input.appendBuffer(raw)
}
/* TODO I'm not actually sure why this code doesn't work; something trips up the MP4 parser
while (1) {
const data = await stream.read()
if (!data) break
input.appendBuffer(data)
input.flush()
}
*/
// One day I'll figure it out; until then read one top-level atom at a time
while (!await msg.stream.done()) {
const raw = await msg.stream.peek(4)
const size = new DataView(raw.buffer, raw.byteOffset, raw.byteLength).getUint32(0)
const atom = await msg.stream.bytes(size)
// Make a copy of the atom because mp4box only accepts an ArrayBuffer unfortunately
let box = new Uint8Array(atom.byteLength);
box.set(atom)
// and for some reason we need to modify the underlying ArrayBuffer with offset
let buffer = box.buffer as MP4ArrayBuffer
buffer.fileStart = offset
// Parse the data
offset = input.appendBuffer(buffer)
input.flush()
}
}
}

20
player/src/audio/index.ts
View File

@ -1,3 +1,19 @@
self.addEventListener('message', (e: Event) => {
import * as Message from "./message"
})
// Wrapper around the WebWorker API
export default class Audio {
worker: Worker;
constructor(config: Message.Config) {
this.worker = new Worker(new URL('worker.ts', import.meta.url), { type: "module" })
this.worker.postMessage({ config }, [ ])
}
init(init: Message.Init) {
this.worker.postMessage({ init }, [ init.stream.buffer, init.stream.reader ])
}
segment(segment: Message.Segment) {
this.worker.postMessage({ segment }, [ segment.stream.buffer, segment.stream.reader ])
}
}

15
player/src/audio/message.ts Normal file
View File

@ -0,0 +1,15 @@
import Reader from "../stream/reader";
export interface Config {
canvas: OffscreenCanvas;
}
export interface Init {
track: string;
stream: Reader;
}
export interface Segment {
track: string;
stream: Reader;
}

12
player/src/audio/renderer.ts Normal file
View File

@ -0,0 +1,12 @@
import * as Message from "./message";
export class Renderer {
render: number; // non-zero if requestAnimationFrame has been called
sync: DOMHighResTimeStamp; // the wall clock value for timestamp 0
last?: number; // the timestamp of the last rendered frame
constructor(config: Message.Config) {
this.render = 0;
this.sync = 0;
}
}

21
player/src/audio/worker.ts Normal file
View File

@ -0,0 +1,21 @@
import { Renderer } from "./renderer"
import { Decoder } from "./decoder"
import * as Message from "./message"
let decoder: Decoder;
let renderer: Renderer;
self.addEventListener('message', async (e: MessageEvent) => {
if (e.data.config) {
const config = e.data.config as Message.Config
renderer = new Renderer(config)
decoder = new Decoder(renderer)
}
if (e.data.segment) {
const segment = e.data.segment as Message.Segment
await decoder.decode(segment)
}
})

30
player/src/mp4/mp4.ts → player/src/mp4/index.ts
View File

@ -1,13 +1,13 @@
// Wrapper around MP4Box to play nicely with MP4Box.
// I tried getting a mp4box.all.d.ts file to work but just couldn't figure it out
import { createFile, ISOFile, DataStream, BoxParser } from "./mp4box.all.js"
import { createFile, ISOFile, DataStream, BoxParser } from "./mp4box.all"
// Rename some stuff so it's on brand.
export { createFile as MP4New, ISOFile as MP4File, DataStream as MP4Stream, BoxParser as MP4Parser }
export { createFile as New, ISOFile as File, DataStream as Stream, BoxParser as Parser }
export type MP4ArrayBuffer = ArrayBuffer & {fileStart: number};
export type ArrayBufferOffset = ArrayBuffer & {fileStart: number};
export interface MP4MediaTrack {
export interface MediaTrack {
id: number;
created: Date;
modified: Date;
@ -25,13 +25,13 @@ export interface MP4MediaTrack {
nb_samples: number;
}
export interface MP4VideoData {
export interface VideoData {
width: number;
height: number;
}
export interface MP4VideoTrack extends MP4MediaTrack {
video: MP4VideoData;
export interface VideoTrack extends MediaTrack {
video: VideoData;
}
export interface MP4AudioData {
@ -40,13 +40,13 @@ export interface MP4AudioData {
sample_size: number;
}
export interface MP4AudioTrack extends MP4MediaTrack {
export interface AudioTrack extends MediaTrack {
audio: MP4AudioData;
}
export type MP4Track = MP4VideoTrack | MP4AudioTrack;
export type Track = VideoTrack | AudioTrack;
export interface MP4Info {
export interface Info {
duration: number;
timescale: number;
fragment_duration: number;
@ -56,13 +56,13 @@ export interface MP4Info {
brands: string[];
created: Date;
modified: Date;
tracks: MP4Track[];
tracks: Track[];
mime: string;
videoTracks: MP4Track[];
audioTracks: MP4Track[];
videoTracks: Track[];
audioTracks: Track[];
}
export interface MP4Sample {
export interface Sample {
number: number;
track_id: number;
timescale: number;
@ -83,3 +83,5 @@ export interface MP4Sample {
offset: number;
subsamples: any;
}
export { Init, InitParser } from "./init"

34
player/src/player/init.ts → player/src/mp4/init.ts
View File

@ -1,24 +1,28 @@
import { MP4New, MP4File, MP4ArrayBuffer, MP4Info } from "../mp4/mp4"
import * as MP4 from "./index"
export interface Init {
raw: MP4.ArrayBufferOffset;
info: MP4.Info;
}
export class InitParser {
mp4box: MP4File;
mp4box: MP4.File;
offset: number;
raw: MP4ArrayBuffer[];
ready: Promise<Init>;
raw: MP4.ArrayBufferOffset[];
info: Promise<MP4.Info>;
constructor() {
this.mp4box = MP4New()
this.mp4box = MP4.New()
this.raw = []
this.offset = 0
// Create a promise that gets resolved once the init segment has been parsed.
this.ready = new Promise((resolve, reject) => {
this.info = new Promise((resolve, reject) => {
this.mp4box.onError = reject
// https://github.com/gpac/mp4box.js#onreadyinfo
this.mp4box.onReady = (info: MP4Info) => {
this.mp4box.onReady = (info: MP4.Info) => {
if (!info.isFragmented) {
reject("expected a fragmented mp4")
}
@ -27,10 +31,7 @@ export class InitParser {
reject("expected a single track")
}
resolve({
info: info,
raw: this.raw,
})
resolve(info)
}
})
}
@ -41,7 +42,7 @@ export class InitParser {
box.set(data)
// and for some reason we need to modify the underlying ArrayBuffer with fileStart
let buffer = box.buffer as MP4ArrayBuffer
let buffer = box.buffer as MP4.ArrayBufferOffset
buffer.fileStart = this.offset
// Parse the data
@ -51,9 +52,4 @@ export class InitParser {
// Add the box to our queue of chunks
this.raw.push(buffer)
}
}
export interface Init {
raw: MP4ArrayBuffer[];
info: MP4Info;
}
}

2
player/src/stream/index.ts Normal file
View File

@ -0,0 +1,2 @@
export { default as Reader } from "./reader"
export { default as Writer } from "./writer"

29
player/src/stream/reader.ts
View File

@ -10,19 +10,26 @@ export default class Reader {
// Returns any number of bytes
async read(): Promise<Uint8Array | undefined> {
if (this.buffer.byteLength) {
const buffer = this.buffer;
this.buffer = new Uint8Array()
return buffer
}
const result = await this.reader.getReader().read()
const r = this.reader.getReader()
const result = await r.read()
r.releaseLock()
return result.value
}
async readAll(): Promise<Uint8Array> {
const r = this.reader.getReader()
while (1) {
const result = await this.reader.getReader().read()
const result = await r.read()
if (result.done) {
break
}
@ -42,12 +49,16 @@ export default class Reader {
const result = this.buffer
this.buffer = new Uint8Array()
r.releaseLock()
return result
}
async bytes(size: number): Promise<Uint8Array> {
const r = this.reader.getReader()
while (this.buffer.byteLength < size) {
const result = await this.reader.getReader().read()
const result = await r.read()
if (result.done) {
throw "short buffer"
}
@ -67,12 +78,16 @@ export default class Reader {
const result = new Uint8Array(this.buffer.buffer, this.buffer.byteOffset, size)
this.buffer = new Uint8Array(this.buffer.buffer, this.buffer.byteOffset + size)
r.releaseLock()
return result
}
async peek(size: number): Promise<Uint8Array> {
const r = this.reader.getReader()
while (this.buffer.byteLength < size) {
const result = await this.reader.getReader().read()
const result = await r.read()
if (result.done) {
throw "short buffer"
}
@ -89,7 +104,11 @@ export default class Reader {
}
}
return new Uint8Array(this.buffer.buffer, this.buffer.byteOffset, size)
const result = new Uint8Array(this.buffer.buffer, this.buffer.byteOffset, size)
r.releaseLock()
return result
}
async view(size: number): Promise<DataView> {

69
player/src/transport/index.ts
View File

@ -1,6 +1,7 @@
import * as Message from "./message"
import Reader from "../stream/reader"
import Writer from "../stream/writer"
import * as Stream from "../stream"
import * as MP4 from "../mp4"
import Video from "../video/index"
///<reference path="./types/webtransport.d.ts"/>
@ -13,11 +14,14 @@ export interface PlayerInit {
export class Player {
quic: Promise<WebTransport>;
api: Promise<WritableStream>;
tracks: Map<string, MP4.InitParser>
//audio: Worker;
video: Video;
constructor(props: PlayerInit) {
this.tracks = new Map();
//this.audio = new Worker("../audio")
this.video = new Video({
canvas: props.canvas.transferControlToOffscreen(),
@ -76,7 +80,7 @@ export class Player {
const stream = await this.api
const writer = new Writer(stream)
const writer = new Stream.Writer(stream)
await writer.uint32(size)
await writer.string("warp")
await writer.string(payload)
@ -97,7 +101,7 @@ export class Player {
}
async handleStream(stream: ReadableStream) {
let r = new Reader(stream.getReader())
let r = new Stream.Reader(stream)
while (!await r.done()) {
const size = await r.uint32();
@ -117,19 +121,52 @@ export class Player {
}
}
async handleInit(stream: Reader, msg: Message.Init) {
// TODO properly determine if audio or video
this.video.init({
track: msg.id,
stream: stream,
})
async handleInit(stream: Stream.Reader, msg: Message.Init) {
let track = this.tracks.get(msg.id);
if (!track) {
track = new MP4.InitParser()
this.tracks.set(msg.id, track)
}
while (1) {
const data = await stream.read()
if (!data) break
track.push(data)
}
const info = await track.info
if (info.audioTracks.length + info.videoTracks.length != 1) {
throw new Error("expected a single track")
}
if (info.videoTracks.length) {
this.video.init({
track: msg.id,
info: info,
raw: track.raw,
})
}
}
async handleSegment(stream: Reader, msg: Message.Segment) {
// TODO properly determine if audio or video
this.video.segment({
track: msg.init,
stream: stream,
})
async handleSegment(stream: Stream.Reader, msg: Message.Segment) {
let track = this.tracks.get(msg.init);
if (!track) {
track = new MP4.InitParser()
this.tracks.set(msg.init, track)
}
const info = await track.info
// Wait until we learn if this is an audio or video track
if (info.videoTracks.length) {
this.video.segment({
track: msg.init,
buffer: stream.buffer,
reader: stream.reader,
})
}
}
}

16
player/src/util/deferred.ts Normal file
View File

@ -0,0 +1,16 @@
export default class Deferred<T> {
promise: Promise<T>
resolve: (value: T | PromiseLike<T>) => void
reject: (value: T | PromiseLike<T>) => void
constructor() {
// Set initial values so TS stops being annoying.
this.resolve = (value: T | PromiseLike<T>) => {};
this.reject = (value: T | PromiseLike<T>) => {};
this.promise = new Promise((resolve, reject) => {
this.resolve = resolve
this.reject = reject
})
}
}

1
player/src/util/index.ts Normal file
View File

@ -0,0 +1 @@
export { default as Deferred } from "./deferred"

65
player/src/video/decoder.ts
View File

@ -1,10 +1,13 @@
import * as Message from "./message";
import { Track } from "./track";
import * as MP4 from "../mp4"
import * as Stream from "../stream"
import * as Util from "../util"
import { Renderer } from "./renderer"
import { MP4New, MP4Sample, MP4ArrayBuffer } from "../mp4/mp4"
export class Decoder {
tracks: Map<string, Track>;
// Store the init message for each track
tracks: Map<string, Util.Deferred<Message.Init>>
renderer: Renderer;
constructor(renderer: Renderer) {
@ -13,37 +16,24 @@ export class Decoder {
}
async init(msg: Message.Init) {
let track = this.tracks.get(msg.track);
if (!track) {
track = new Track()
this.tracks.set(msg.track, track)
}
let track = this.tracks.get(msg.track);
if (!track) {
track = new Util.Deferred()
this.tracks.set(msg.track, track)
}
while (1) {
const data = await msg.stream.read()
if (!data) break
track.init(data)
}
// TODO this will hang on incomplete data
const init = await track.ready;
const info = init.info;
if (info.audioTracks.length + info.videoTracks.length != 1) {
throw new Error("expected a single track")
}
track.resolve(msg)
}
async decode(msg: Message.Segment) {
let track = this.tracks.get(msg.track);
if (!track) {
track = new Track()
track = new Util.Deferred()
this.tracks.set(msg.track, track)
}
// Wait for the init segment to be fully received and parsed
const init = await track.ready;
const init = await track.promise;
const info = init.info;
const video = info.videoTracks[0]
@ -57,17 +47,18 @@ export class Decoder {
});
decoder.configure({
codec: info.mime,
codec: video.codec,
codedHeight: video.track_height,
codedWidth: video.track_width,
// optimizeForLatency: true
})
const input = MP4New();
const input = MP4.New();
input.onSamples = (id: number, user: any, samples: MP4Sample[]) => {
input.onSamples = (id: number, user: any, samples: MP4.Sample[]) => {
for (let sample of samples) {
const timestamp = sample.dts / (1000 / info.timescale) // milliseconds
console.log(sample)
decoder.decode(new EncodedVideoChunk({
data: sample.data,
@ -83,13 +74,16 @@ export class Decoder {
input.start();
}
let offset = 0
// MP4box requires us to reparse the init segment unfortunately
for (let raw of track.raw) {
offset = input.appendBuffer(raw)
let offset = 0;
for (let raw of init.raw) {
raw.fileStart = offset
input.appendBuffer(raw)
}
const stream = new Stream.Reader(msg.reader, msg.buffer)
/* TODO I'm not actually sure why this code doesn't work; something trips up the MP4 parser
while (1) {
const data = await stream.read()
@ -101,23 +95,22 @@ export class Decoder {
*/
// One day I'll figure it out; until then read one top-level atom at a time
while (!await msg.stream.done()) {
const raw = await msg.stream.peek(4)
while (!await stream.done()) {
const raw = await stream.peek(4)
const size = new DataView(raw.buffer, raw.byteOffset, raw.byteLength).getUint32(0)
const atom = await msg.stream.bytes(size)
const atom = await stream.bytes(size)
// Make a copy of the atom because mp4box only accepts an ArrayBuffer unfortunately
let box = new Uint8Array(atom.byteLength);
box.set(atom)
// and for some reason we need to modify the underlying ArrayBuffer with offset
let buffer = box.buffer as MP4ArrayBuffer
let buffer = box.buffer as MP4.ArrayBufferOffset
buffer.fileStart = offset
// Parse the data
offset = input.appendBuffer(buffer)
input.flush()
}
}
}

4
player/src/video/index.ts
View File

@ -10,10 +10,10 @@ export default class Video {
}
init(init: Message.Init) {
this.worker.postMessage({ init }, [ init.stream.buffer, init.stream.reader ])
this.worker.postMessage({ init }) // note: we copy the raw init bytes each time
}
segment(segment: Message.Segment) {
this.worker.postMessage({ segment }, [ segment.stream.buffer, segment.stream.reader ])
this.worker.postMessage({ segment }, [ segment.buffer.buffer, segment.reader ])
}
}

8
player/src/video/message.ts
View File

@ -1,4 +1,4 @@
import Reader from "../stream/reader";
import * as MP4 from "../mp4"
export interface Config {
canvas: OffscreenCanvas;
@ -6,10 +6,12 @@ export interface Config {
export interface Init {
track: string;
stream: Reader;
info: MP4.Info;
raw: MP4.ArrayBufferOffset[];
}
export interface Segment {
track: string;
stream: Reader;
buffer: Uint8Array; // unread buffered data
reader: ReadableStream; // unread unbuffered data
}

58
player/src/video/track.ts
View File

@ -1,58 +0,0 @@
import { MP4New, MP4File, MP4ArrayBuffer, MP4Info } from "../mp4/mp4"
export interface Init {
raw: MP4ArrayBuffer[];
info: MP4Info;
}
export class Track {
mp4box: MP4File;
offset: number;
raw: MP4ArrayBuffer[];
ready: Promise<Init>;
constructor() {
this.mp4box = MP4New()
this.raw = []
this.offset = 0
// Create a promise that gets resolved once the init segment has been parsed.
this.ready = new Promise((resolve, reject) => {
this.mp4box.onError = reject
// https://github.com/gpac/mp4box.js#onreadyinfo
this.mp4box.onReady = (info: MP4Info) => {
if (!info.isFragmented) {
reject("expected a fragmented mp4")
}
if (info.tracks.length != 1) {
reject("expected a single track")
}
resolve({
raw: this.raw,
info,
})
}
})
}
init(data: Uint8Array) {
// Make a copy of the atom because mp4box only accepts an ArrayBuffer unfortunately
let box = new Uint8Array(data.byteLength);
box.set(data)
// and for some reason we need to modify the underlying ArrayBuffer with fileStart
let buffer = box.buffer as MP4ArrayBuffer
buffer.fileStart = this.offset
// Parse the data
this.offset = this.mp4box.appendBuffer(buffer)
this.mp4box.flush()
// Add the box to our queue of chunks
this.raw.push(buffer)
}
}

8
player/src/video/worker.ts
View File

@ -11,11 +11,11 @@ self.addEventListener('message', async (e: MessageEvent) => {
renderer = new Renderer(config)
decoder = new Decoder(renderer)
}
if (e.data.segment) {
} else if (e.data.init) {
const init = e.data.init as Message.Init
await decoder.init(init)
} else if (e.data.segment) {
const segment = e.data.segment as Message.Segment
await decoder.decode(segment)
}
})

2
player/tsconfig.json
View File

@ -4,6 +4,8 @@
],
"compilerOptions": {
"target": "es2022",
"module": "es2022",
"moduleResolution": "node",
"strict": true,
"typeRoots": [
"src/types"

1515
player/yarn.lock
View File

File diff suppressed because it is too large Load Diff

10
server/internal/warp/session.go
View File

@ -200,6 +200,11 @@ func (s *Session) writeInit(ctx context.Context, init *MediaInit) (err error) {
return fmt.Errorf("failed to write init data: %w", err)
}
err = stream.Close()
if err != nil {
return fmt.Errorf("failed to close init stream: %w", err)
}
return nil
}
@ -210,6 +215,11 @@ func (s *Session) writeSegment(ctx context.Context, segment *MediaSegment) (err
return fmt.Errorf("failed to create stream: %w", err)
}
if temp == nil {
// Not sure when this happens, perhaps when closing a connection?
return fmt.Errorf("received a nil stream from quic-go: %w", err)
}
// Wrap the stream in an object that buffers writes instead of blocking.
stream := NewStream(temp)
s.streams.Add(stream.Run)