restitux/gamestream-webtransport-proxy
1
0
Fork 0
Code Issues Pull Requests 6 Actions Packages Projects Releases Wiki Activity

backend and frontend: support out of order chunks + now it's performant on chrome

Browse Source
This commit is contained in:
restitux
2025-08-12 02:20:46 -06:00
parent 7afd8db8d8
commit e80543144a
21 changed files with 876 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -680,6 +680,8 @@ dependencies = [
"directories",
"flatbuffers",
"getrandom 0.3.3",
"h3-datagram",
"h3-quinn",
"hex",
"hmac-sha256",
"http",
frontend/src/lib/proto/video-update.ts
+2 -1
View File
@@ -2,7 +2,8 @@
/* eslint-disable @typescript-eslint/no-unused-vars, @typescript-eslint/no-explicit-any, @typescript-eslint/no-non-null-assertion */
export { DecodeUnit } from './video-update/decode-unit.js';
export { DecodeUnitBuffer } from './video-update/decode-unit-buffer.js';
export { DecodeUnitStart } from './video-update/decode-unit-start.js';
export { FrameType } from './video-update/frame-type.js';
export { Setup } from './video-update/setup.js';
export { Update } from './video-update/update.js';
frontend/src/lib/proto/video-update/decode-unit-buffer.ts
+100
View File
@@ -0,0 +1,100 @@
// automatically generated by the FlatBuffers compiler, do not modify
/* eslint-disable @typescript-eslint/no-unused-vars, @typescript-eslint/no-explicit-any, @typescript-eslint/no-non-null-assertion */
import * as flatbuffers from 'flatbuffers';
export class DecodeUnitBuffer {
bb: flatbuffers.ByteBuffer|null = null;
bb_pos = 0;
__init(i:number, bb:flatbuffers.ByteBuffer):DecodeUnitBuffer {
this.bb_pos = i;
this.bb = bb;
return this;
}
static getRootAsDecodeUnitBuffer(bb:flatbuffers.ByteBuffer, obj?:DecodeUnitBuffer):DecodeUnitBuffer {
return (obj || new DecodeUnitBuffer()).__init(bb.readInt32(bb.position()) + bb.position(), bb);
}
static getSizePrefixedRootAsDecodeUnitBuffer(bb:flatbuffers.ByteBuffer, obj?:DecodeUnitBuffer):DecodeUnitBuffer {
bb.setPosition(bb.position() + flatbuffers.SIZE_PREFIX_LENGTH);
return (obj || new DecodeUnitBuffer()).__init(bb.readInt32(bb.position()) + bb.position(), bb);
}
frameNumber():bigint {
const offset = this.bb!.__offset(this.bb_pos, 4);
return offset ? this.bb!.readUint64(this.bb_pos + offset) : BigInt('0');
}
bufferIndex():bigint {
const offset = this.bb!.__offset(this.bb_pos, 6);
return offset ? this.bb!.readUint64(this.bb_pos + offset) : BigInt('0');
}
bufferOffset():bigint {
const offset = this.bb!.__offset(this.bb_pos, 8);
return offset ? this.bb!.readUint64(this.bb_pos + offset) : BigInt('0');
}
data(index: number):number|null {
const offset = this.bb!.__offset(this.bb_pos, 10);
return offset ? this.bb!.readUint8(this.bb!.__vector(this.bb_pos + offset) + index) : 0;
}
dataLength():number {
const offset = this.bb!.__offset(this.bb_pos, 10);
return offset ? this.bb!.__vector_len(this.bb_pos + offset) : 0;
}
dataArray():Uint8Array|null {
const offset = this.bb!.__offset(this.bb_pos, 10);
return offset ? new Uint8Array(this.bb!.bytes().buffer, this.bb!.bytes().byteOffset + this.bb!.__vector(this.bb_pos + offset), this.bb!.__vector_len(this.bb_pos + offset)) : null;
}
static startDecodeUnitBuffer(builder:flatbuffers.Builder) {
builder.startObject(4);
}
static addFrameNumber(builder:flatbuffers.Builder, frameNumber:bigint) {
builder.addFieldInt64(0, frameNumber, BigInt('0'));
}
static addBufferIndex(builder:flatbuffers.Builder, bufferIndex:bigint) {
builder.addFieldInt64(1, bufferIndex, BigInt('0'));
}
static addBufferOffset(builder:flatbuffers.Builder, bufferOffset:bigint) {
builder.addFieldInt64(2, bufferOffset, BigInt('0'));
}
static addData(builder:flatbuffers.Builder, dataOffset:flatbuffers.Offset) {
builder.addFieldOffset(3, dataOffset, 0);
}
static createDataVector(builder:flatbuffers.Builder, data:number[]|Uint8Array):flatbuffers.Offset {
builder.startVector(1, data.length, 1);
for (let i = data.length - 1; i >= 0; i--) {
builder.addInt8(data[i]!);
}
return builder.endVector();
}
static startDataVector(builder:flatbuffers.Builder, numElems:number) {
builder.startVector(1, numElems, 1);
}
static endDecodeUnitBuffer(builder:flatbuffers.Builder):flatbuffers.Offset {
const offset = builder.endObject();
return offset;
}
static createDecodeUnitBuffer(builder:flatbuffers.Builder, frameNumber:bigint, bufferIndex:bigint, bufferOffset:bigint, dataOffset:flatbuffers.Offset):flatbuffers.Offset {
DecodeUnitBuffer.startDecodeUnitBuffer(builder);
DecodeUnitBuffer.addFrameNumber(builder, frameNumber);
DecodeUnitBuffer.addBufferIndex(builder, bufferIndex);
DecodeUnitBuffer.addBufferOffset(builder, bufferOffset);
DecodeUnitBuffer.addData(builder, dataOffset);
return DecodeUnitBuffer.endDecodeUnitBuffer(builder);
}
}
frontend/src/lib/proto/video-update/decode-unit-start.ts
+91
View File
@@ -0,0 +1,91 @@
// automatically generated by the FlatBuffers compiler, do not modify
/* eslint-disable @typescript-eslint/no-unused-vars, @typescript-eslint/no-explicit-any, @typescript-eslint/no-non-null-assertion */
import * as flatbuffers from 'flatbuffers';
import { FrameType } from '../video-update/frame-type.js';
export class DecodeUnitStart {
bb: flatbuffers.ByteBuffer|null = null;
bb_pos = 0;
__init(i:number, bb:flatbuffers.ByteBuffer):DecodeUnitStart {
this.bb_pos = i;
this.bb = bb;
return this;
}
static getRootAsDecodeUnitStart(bb:flatbuffers.ByteBuffer, obj?:DecodeUnitStart):DecodeUnitStart {
return (obj || new DecodeUnitStart()).__init(bb.readInt32(bb.position()) + bb.position(), bb);
}
static getSizePrefixedRootAsDecodeUnitStart(bb:flatbuffers.ByteBuffer, obj?:DecodeUnitStart):DecodeUnitStart {
bb.setPosition(bb.position() + flatbuffers.SIZE_PREFIX_LENGTH);
return (obj || new DecodeUnitStart()).__init(bb.readInt32(bb.position()) + bb.position(), bb);
}
frameNumber():bigint {
const offset = this.bb!.__offset(this.bb_pos, 4);
return offset ? this.bb!.readUint64(this.bb_pos + offset) : BigInt('0');
}
frameType():FrameType {
const offset = this.bb!.__offset(this.bb_pos, 6);
return offset ? this.bb!.readInt8(this.bb_pos + offset) : FrameType.PFRAME;
}
numBuffers():bigint {
const offset = this.bb!.__offset(this.bb_pos, 8);
return offset ? this.bb!.readUint64(this.bb_pos + offset) : BigInt('0');
}
receiveTimeMs():number {
const offset = this.bb!.__offset(this.bb_pos, 10);
return offset ? this.bb!.readUint16(this.bb_pos + offset) : 0;
}
fullLength():bigint {
const offset = this.bb!.__offset(this.bb_pos, 12);
return offset ? this.bb!.readUint64(this.bb_pos + offset) : BigInt('0');
}
static startDecodeUnitStart(builder:flatbuffers.Builder) {
builder.startObject(5);
}
static addFrameNumber(builder:flatbuffers.Builder, frameNumber:bigint) {
builder.addFieldInt64(0, frameNumber, BigInt('0'));
}
static addFrameType(builder:flatbuffers.Builder, frameType:FrameType) {
builder.addFieldInt8(1, frameType, FrameType.PFRAME);
}
static addNumBuffers(builder:flatbuffers.Builder, numBuffers:bigint) {
builder.addFieldInt64(2, numBuffers, BigInt('0'));
}
static addReceiveTimeMs(builder:flatbuffers.Builder, receiveTimeMs:number) {
builder.addFieldInt16(3, receiveTimeMs, 0);
}
static addFullLength(builder:flatbuffers.Builder, fullLength:bigint) {
builder.addFieldInt64(4, fullLength, BigInt('0'));
}
static endDecodeUnitStart(builder:flatbuffers.Builder):flatbuffers.Offset {
const offset = builder.endObject();
return offset;
}
static createDecodeUnitStart(builder:flatbuffers.Builder, frameNumber:bigint, frameType:FrameType, numBuffers:bigint, receiveTimeMs:number, fullLength:bigint):flatbuffers.Offset {
DecodeUnitStart.startDecodeUnitStart(builder);
DecodeUnitStart.addFrameNumber(builder, frameNumber);
DecodeUnitStart.addFrameType(builder, frameType);
DecodeUnitStart.addNumBuffers(builder, numBuffers);
DecodeUnitStart.addReceiveTimeMs(builder, receiveTimeMs);
DecodeUnitStart.addFullLength(builder, fullLength);
return DecodeUnitStart.endDecodeUnitStart(builder);
}
}
frontend/src/lib/proto/video-update/update.ts
+12 -8
View File
@@ -2,37 +2,41 @@
/* eslint-disable @typescript-eslint/no-unused-vars, @typescript-eslint/no-explicit-any, @typescript-eslint/no-non-null-assertion */
import { DecodeUnit } from '../video-update/decode-unit.js';
import { DecodeUnitBuffer } from '../video-update/decode-unit-buffer.js';
import { DecodeUnitStart } from '../video-update/decode-unit-start.js';
import { Setup } from '../video-update/setup.js';
export enum Update {
NONE = 0,
Setup = 1,
DecodeUnit = 2
DecodeUnitStart = 2,
DecodeUnitBuffer = 3
}
export function unionToUpdate(
type: Update,
accessor: (obj:DecodeUnit|Setup) => DecodeUnit|Setup|null
): DecodeUnit|Setup|null {
accessor: (obj:DecodeUnitBuffer|DecodeUnitStart|Setup) => DecodeUnitBuffer|DecodeUnitStart|Setup|null
): DecodeUnitBuffer|DecodeUnitStart|Setup|null {
switch(Update[type]) {
case 'NONE': return null;
case 'Setup': return accessor(new Setup())! as Setup;
case 'DecodeUnit': return accessor(new DecodeUnit())! as DecodeUnit;
case 'DecodeUnitStart': return accessor(new DecodeUnitStart())! as DecodeUnitStart;
case 'DecodeUnitBuffer': return accessor(new DecodeUnitBuffer())! as DecodeUnitBuffer;
default: return null;
}
}
export function unionListToUpdate(
type: Update,
accessor: (index: number, obj:DecodeUnit|Setup) => DecodeUnit|Setup|null,
accessor: (index: number, obj:DecodeUnitBuffer|DecodeUnitStart|Setup) => DecodeUnitBuffer|DecodeUnitStart|Setup|null,
index: number
): DecodeUnit|Setup|null {
): DecodeUnitBuffer|DecodeUnitStart|Setup|null {
switch(Update[type]) {
case 'NONE': return null;
case 'Setup': return accessor(index, new Setup())! as Setup;
case 'DecodeUnit': return accessor(index, new DecodeUnit())! as DecodeUnit;
case 'DecodeUnitStart': return accessor(index, new DecodeUnitStart())! as DecodeUnitStart;
case 'DecodeUnitBuffer': return accessor(index, new DecodeUnitBuffer())! as DecodeUnitBuffer;
default: return null;
}
}
frontend/src/lib/video.ts
+142 -65
View File
@@ -1,107 +1,184 @@
import { VideoUpdate } from "$lib/proto/video";
import { ByteBuffer } from "flatbuffers";
import { DecodeUnitBuffer } from "./proto/video-update";
import { DoorClosed, Video } from "lucide-svelte";
function parseData(newBuffer: Uint8Array, oldBuffer: Uint8Array): [Array<VideoUpdate.VideoUpdate>, Uint8Array<ArrayBuffer>] {
let packets = new Array<VideoUpdate.VideoUpdate>();
let unparsedData = new Uint8Array();
let data = new Uint8Array([...oldBuffer, ...newBuffer]);
let index = 0;
while (true) {
if (index >= data.length) {
break
}
const view = new DataView(data.buffer.slice(index, index + 4));
const dataLength = view.getUint32(0, true);
const slice_start_index = index + 4;
const slice_end_index = index + 4 + dataLength;
if (data.length < slice_end_index) {
unparsedData = data.slice(index, data.length);
break;
}
const dataToParse = new ByteBuffer(data.slice(slice_start_index, slice_end_index));
const videoUpdate = VideoUpdate.VideoUpdate.getRootAsVideoUpdate(dataToParse);
packets.push(videoUpdate);
index += 4 + dataLength;
}
return [packets, unparsedData];
}
export async function streamVideoFromReader(reader: ReadableStreamDefaultReader, canvasElement: OffscreenCanvas) {
function getVideoDecoder(canvasElement: OffscreenCanvas): VideoDecoder {
const canvasCtx: OffscreenCanvasRenderingContext2D | null = canvasElement.getContext('2d');
if (canvasCtx == null) {
throw new Error(`Could not get 2d canvas context`);
}
try {
let unparsedData = new Uint8Array();
const videoDecoder = new VideoDecoder({
output: (frame) => {
canvasElement.width = frame.displayWidth;
canvasElement.height = frame.displayHeight;
//console.log(`rendering frame start: ${performance.now()}`);
//canvasElement.width = frame.displayWidth;
//canvasElement.height = frame.displayHeight;
//console.log(`rendering frame drawImage: ${performance.now()}`);
canvasCtx.drawImage(frame, 0, 0);
//console.log(`rendering frame end: ${performance.now()}`);
frame.close();
//console.log(`rendering frame close: ${performance.now()}`);
},
error: (e) => {
console.error('Decode error:', e);
}
});
let decodeUnit: VideoUpdate.DecodeUnit = new VideoUpdate.DecodeUnit();
while (true) {
const { value, done } = await reader.read();
if (done) break;
return videoDecoder;
}
let [packets, remainingData] = parseData(value, unparsedData);
unparsedData = remainingData;
for (let i = 0; i < packets.length; i++) {
if (packets[i].updateType() == VideoUpdate.Update.Setup) {
let setup = packets[i].update(new VideoUpdate.Setup());
async function configureDecoder(videoDecoder: VideoDecoder, videoFormat: string, width: number, height: number) {
let config: VideoDecoderConfig = {
codec: setup.videoFormat(),
codedWidth: setup.width(),
codedHeight: setup.height(),
codec: videoFormat,
codedWidth: width,
codedHeight: height,
optimizeForLatency: true,
//hardwareAcceleration: "prefer-hardware",
};
const codecSupport = await VideoDecoder.isConfigSupported(config);
console.log(codecSupport);
if (codecSupport.supported) {
videoDecoder.configure(config);
} else {
throw new Error(`Could not configure decoder`);
}
} else if (packets[i].updateType() == VideoUpdate.Update.DecodeUnit) {
packets[i].update(decodeUnit);
let frameType: EncodedAudioChunkType = "delta";
if (decodeUnit.frameType() == VideoUpdate.FrameType.IDR) {
console.log("GOT KEYFRAME");
frameType = "key";
}
class Decoder {
videoDecoder: VideoDecoder;
frameNumber: bigint | undefined;
frameType: EncodedAudioChunkType = "delta";
fullLength: bigint = 0n;
receiveTimeMs: number = 0;
//frameBroken: boolean = false;
//lastBufferIndex: bigint = 0n;
frameCollector: boolean[] = new Array<boolean>();
dataOffset: number = 0;
data: Uint8Array = new Uint8Array();
constructor(videoDecoder: VideoDecoder) {
this.videoDecoder = videoDecoder;
}
print() {
console.log(
`
frameNumber: ${this.frameNumber}
frameType: ${this.frameType}
fullLength: ${this.fullLength}
receiveTimeMs: ${this.receiveTimeMs}
frameCollector: ${this.frameCollector}
ts: ${performance.now()}
`
);
}
processStart(decodeUnitStart: VideoUpdate.DecodeUnitStart) {
//this.print();
const frameCompleted = !this.frameCollector.includes(false);
if (!frameCompleted) {
console.log(`Got setup packet for frame ${decodeUnitStart.frameNumber()} but the last frame has not been completed`);
}
this.frameNumber = decodeUnitStart.frameNumber();
this.frameType = "delta";
if (decodeUnitStart.frameType() == VideoUpdate.FrameType.IDR) {
this.frameType = "key";
}
this.fullLength = decodeUnitStart.fullLength();
this.receiveTimeMs = decodeUnitStart.receiveTimeMs();
//this.frameBroken = false;
//this.lastBufferIndex = -1n;
//this.dataOffset = 0;
this.frameCollector = new Array(Number(decodeUnitStart.numBuffers())).fill(false);
this.data = new Uint8Array(Number(this.fullLength));
//this.print();
//console.log(`start: `, this);
//console.log(performance.now());
}
processBuffer(decodeUnitBuffer: VideoUpdate.DecodeUnitBuffer) {
//console.log(`buffer: `, this);
//console.log(performance.now());
//this.print();
const frameNumber = decodeUnitBuffer.frameNumber();
if (this.frameNumber === undefined) {
console.log("frameNumber is undefined but we got a buffer, ignoring...");
return;
}
if (this.frameNumber != frameNumber) {
console.log(`Got buffer for frame ${frameNumber} but we are processing frame ${this.frameNumber}, ignoring...`);
return;
}
let offset = decodeUnitBuffer.bufferOffset();
for (var i = 0; i < decodeUnitBuffer.dataLength(); i++) {
this.data[Number(offset) + i] = decodeUnitBuffer.data(i)!;
}
this.frameCollector[Number(decodeUnitBuffer.bufferIndex())] = true;
const gotAllframes = !this.frameCollector.includes(false);
if (gotAllframes) {
const chunk = new EncodedVideoChunk({
timestamp: decodeUnit.receiveTimeMs(),
type: frameType,
data: decodeUnit.dataArray()!,
//timestamp: this.receiveTimeMs,
timestamp: 0,
type: this.frameType,
data: this.data,
});
videoDecoder.decode(chunk);
//console.log(`${ performance.now() }: Enqueing a new decode request, current queue size ${ this.videoDecoder.decodeQueueSize } `);
this.videoDecoder.decode(chunk);
}
//this.print();
}
}
export async function streamVideoFromReader(reader: ReadableStreamDefaultReader, canvasElement: OffscreenCanvas) {
const videoDecoder = getVideoDecoder(canvasElement);
try {
let decodeUnitBuffer: VideoUpdate.DecodeUnitBuffer = new VideoUpdate.DecodeUnitBuffer();
let decoder = new Decoder(videoDecoder);
while (true) {
const { value, done } = await reader.read();
if (done) break;
const dataToParse = new ByteBuffer(value);
const videoUpdate = VideoUpdate.VideoUpdate.getRootAsVideoUpdate(dataToParse);
if (videoUpdate.updateType() == VideoUpdate.Update.Setup) {
let setup = videoUpdate.update(new VideoUpdate.Setup());
await configureDecoder(videoDecoder, setup.videoFormat(), setup.width(), setup.height());
} else if (videoUpdate.updateType() == VideoUpdate.Update.DecodeUnitStart) {
let decodeUnitStart: VideoUpdate.DecodeUnitStart = new VideoUpdate.DecodeUnitStart();
videoUpdate.update(decodeUnitStart);
decoder.processStart(decodeUnitStart);
} else if (videoUpdate.updateType() == VideoUpdate.Update.DecodeUnitBuffer) {
videoUpdate.update(decodeUnitBuffer);
decoder.processBuffer(decodeUnitBuffer);
} else {
throw new Error(`Got packet of unknown type`);
}
}
}
} catch (e) {
var error = <Error>e;
console.error('Error connecting to stream:', error);
frontend/src/routes/Cover.svelte
+2
View File
@@ -23,6 +23,8 @@
let streamData = await getStreamData(app.id, server_name);
streamStore.Url = streamData.Url;
streamStore.CertHash = streamData.CertHash;
streamStore.Width = streamData.Width;
streamStore.Height = streamData.Height;
console.log(`Stream data retrieved. Navigating to /stream.`);
await goto('/stream');
frontend/src/routes/getStreamData.ts
+11 -32
View File
@@ -1,51 +1,30 @@
//Setup {
// video_format: VideoFormat,
// width: u64,
// height: u64,
// redraw_rate: u64,
// dr_flags: i32,
//},
//DecodeUnit {
// frame_number: u64,
// frame_type: FrameType,
// host_processing_latency: u16,
// receieve_time_ms: u64,
// enqueue_time_ms: u64,
// presentation_time: u64,
// full_length: usize,
// //buffers: Vec<Buffer>,
// buffer: Buffer,
// index: u64,
// hdr_active: bool,
// colorspace: u8,
//},
type StreamData = {
Url: string,
CertHash: Array<number>,
Width: number,
Height: number,
}
export async function getStreamData(appId: number, server_name: string): Promise<StreamData> {
try {
// Create the POST request payload
const width = 1920;
const height = 1080;
const payload = {
id: appId,
server: server_name,
server_mode: {
fps: 60,
width: 1920,
height: 1080,
width: width,
height: height,
},
stream_config: {
bitrate_kbps: 1024 * 10 * 2,
bitrate_kbps: 1024 * 10 * 5,
mode: {
fps: 60,
width: 1920,
height: 1080,
width: width,
height: height,
}
}
};
@@ -66,7 +45,7 @@ export async function getStreamData(appId: number, server_name: string): Promise
const streamDataResp = await response.json();
console.log('Stream started:', streamDataResp);
let streamData: StreamData = { Url: streamDataResp.url, CertHash: streamDataResp.cert_hash };
let streamData: StreamData = { Url: streamDataResp.url, CertHash: streamDataResp.cert_hash, Width: width, Height: height };
return streamData;
frontend/src/routes/stores/streamStore.svelte.ts
+2
View File
@@ -1,4 +1,6 @@
export const streamStore = $state({
Url: '',
CertHash: [0],
Width: 0,
Height: 0,
});
frontend/src/routes/stream/+page.svelte
+3 -1
View File
@@ -4,6 +4,8 @@
$: url = streamStore.Url;
$: certHash = streamStore.CertHash;
$: width = streamStore.Width;
$: height = streamStore.Height;
</script>
<svelte:head>
@@ -13,7 +15,7 @@
<!--<section>
</section>-->
<Stream {url} {certHash} />
<Stream {url} {certHash} {width} {height} />
<style>
section {
frontend/src/routes/stream/Stream.svelte
+12 -2
View File
@@ -6,16 +6,26 @@
interface Props {
url: string;
certHash: Array<number>;
width: number;
height: number;
}
let { url, certHash }: Props = $props();
let { url, certHash, width, height }: Props = $props();
let loading = $state(true);
let fullscreen = $state(false);
let gameplayView: HTMLDivElement;
let gameplayCanvas: HTMLCanvasElement;
async function startStream() {
await startWebtransportStream(url, certHash, gameplayCanvas, gameplayCanvas, gameplayCanvas);
await startWebtransportStream(
url,
certHash,
width,
height,
gameplayCanvas,
gameplayCanvas,
gameplayCanvas
);
}
async function requestFullscreen() {
frontend/src/routes/stream/stream.ts
+15 -5
View File
@@ -9,7 +9,7 @@ export async function getStreamTransport(url: string, certHash: Array<number>):
}
const transport = new WebTransport(url, {
congestionControl: "low-latency",
//congestionControl: "low-latency",
serverCertificateHashes: [
{
algorithm: "sha-256",
@@ -20,7 +20,7 @@ export async function getStreamTransport(url: string, certHash: Array<number>):
console.log('Connecting to WebTransport at ', url);
await transport.ready;
console.log('WebTransport connection established');
console.log(`WebTransport connection established`);
return transport;
}
@@ -42,19 +42,29 @@ export async function spawnWorker(gameplayCanvas: HTMLCanvasElement, reader: Rea
export async function startWebtransportStream(
url: string,
certHash: Array<number>,
width: number,
height: number,
gameplayCanvas: HTMLCanvasElement,
keyEventElement: HTMLElement,
mouseElement: HTMLElement,
) {
console.log(width, height);
gameplayCanvas.width = width;
gameplayCanvas.height = height;
console.log(`Connecting to stream at ${url} with cert_hash ${certHash}`);
const transport = await getStreamTransport(url, certHash);
const datagrams = transport.datagrams;
datagrams.incomingHighWaterMark = 20000;
const stream = await transport.createBidirectionalStream();
const reader = stream.readable
//const reader = stream.readable
const datagramReader = datagrams.readable
const writer = stream.writable.getWriter();
spawnWorker(gameplayCanvas, reader);
spawnWorker(gameplayCanvas, datagramReader);
keyEventElement.addEventListener("keydown", (event: KeyboardEvent) => { sendKeyboardEvent(writer, event, KeyAction.DOWN) });
keyEventElement.addEventListener("keyup", (event: KeyboardEvent) => { sendKeyboardEvent(writer, event, KeyAction.UP) });
@@ -66,7 +76,7 @@ export async function startWebtransportStream(
mouseElement.addEventListener("click", async () => {
console.log("Requesting pointer lock");
await mouseElement.requestPointerLock({
unadjustedMovement: true,
//unadjustedMovement: true,
});
console.log("Pointer lock aquired");
})
gamestream-webtransport-proxy/Cargo.toml
+2
View File
@@ -8,6 +8,8 @@ anyhow = "1.0.98"
directories = "6.0.0"
flatbuffers = "25.2.10"
getrandom = { version = "0.3.3", features = ["std"] }
h3-datagram = "0.0.2"
h3-quinn = "0.0.10"
hex = "0.4.3"
hmac-sha256 = "1.1.12"
http = "1.3.1"
gamestream-webtransport-proxy/src/certs.rs
+29 -3
View File
@@ -120,15 +120,41 @@ fn generate_http_cert_and_key(
cert_builder.set_version(2)?;
let serial = openssl::bn::BigNum::from_u32(1)?;
let asn_serial = openssl::asn1::Asn1Integer::from_bn(&serial)?;
cert_builder.set_serial_number(&asn_serial)?;
// Set subject (Distinguished Name)
let mut name_builder = X509NameBuilder::new()?;
name_builder.append_entry_by_text("CN", "mumble-web self-signed")?;
name_builder.append_entry_by_text("CN", "localhost")?;
let subject_name = name_builder.build();
cert_builder.set_subject_name(&subject_name)?;
// Set issuer (same as subject for self-signed)
cert_builder.set_issuer_name(&subject_name)?;
let context = cert_builder.x509v3_context(None, None);
let mut san = openssl::x509::extension::SubjectAlternativeName::new();
san.dns("localhost");
let san_extension = san.build(&context)?;
let key_usage = openssl::x509::extension::KeyUsage::new()
.digital_signature()
.key_encipherment()
.build()?;
let ext_key_usage = openssl::x509::extension::ExtendedKeyUsage::new()
.server_auth()
.build()?;
// Add Subject Key Identifier
let subject_key_id = openssl::x509::extension::SubjectKeyIdentifier::new().build(&context)?;
cert_builder.append_extension(san_extension)?;
cert_builder.append_extension(key_usage)?;
cert_builder.append_extension(ext_key_usage)?;
cert_builder.append_extension(subject_key_id)?;
cert_builder.set_not_before(&now)?;
cert_builder.set_not_after(&expiration_time)?;
cert_builder.set_pubkey(&key)?;
gamestream-webtransport-proxy/src/gamestream/decoder.rs
+60 -37
View File
@@ -119,23 +119,27 @@ pub enum RendererMessage {
redraw_rate: u64,
dr_flags: i32,
},
DecodeUnit {
DecodeUnitStart {
frame_number: u64,
frame_type: FrameType,
num_buffers: u64,
host_processing_latency: u16,
receive_time_ms: u64,
enqueue_time_ms: u64,
presentation_time: u64,
full_length: usize,
//buffers: Vec<Buffer>,
buffer: Buffer,
index: u64,
full_length: u64,
hdr_active: bool,
colorspace: u8,
},
DecodeUnitBuffer {
frame_number: u64,
buffer_index: u64,
buffer_offset: u64,
buffer: Buffer,
},
}
impl RendererMessage {
@@ -155,23 +159,43 @@ impl RendererMessage {
})
}
fn from_decode_unit(decode_unit: _DECODE_UNIT) -> Result<Self> {
//fn from_decode_unit(decode_unit: _DECODE_UNIT) -> Result<Vec<Self>> {
let mut buffer = Vec::new();
//let mut buffers = Vec::new();
fn from_decode_unit(decode_unit: _DECODE_UNIT) -> Result<Vec<Self>> {
let mut messages = Vec::new();
if decode_unit.bufferList.is_null() {
return Err(anyhow!("DecodeUnit bufferList is null"));
}
let frame_number = <u64>::try_from(decode_unit.frameNumber)?;
messages.push(RendererMessage::DecodeUnitStart {
frame_number,
frame_type: FrameType::try_from(decode_unit.frameType)?,
num_buffers: 0,
host_processing_latency: decode_unit.frameHostProcessingLatency,
receive_time_ms: decode_unit.receiveTimeMs,
enqueue_time_ms: decode_unit.enqueueTimeMs,
presentation_time: decode_unit.presentationTimeMs as u64,
full_length: <u64>::try_from(decode_unit.fullLength)?,
hdr_active: decode_unit.hdrActive,
colorspace: decode_unit.colorspace,
});
let mut next = unsafe { *decode_unit.bufferList };
let mut index = 0;
let mut offset = 0;
loop {
let mut b = Buffer::try_from(next)?;
buffer.append(&mut b.data);
let b = Buffer::try_from(next)?;
let buffer_len = b.data.len() as u64;
//buffers.push(msg);
messages.push(RendererMessage::DecodeUnitBuffer {
frame_number,
buffer_index: index,
buffer_offset: offset,
buffer: b,
});
offset = offset + buffer_len;
index = index + 1;
if next.next.is_null() {
break;
@@ -180,22 +204,15 @@ impl RendererMessage {
next = unsafe { *next.next };
}
Ok(RendererMessage::DecodeUnit {
frame_number: <u64>::try_from(decode_unit.frameNumber)?,
frame_type: FrameType::try_from(decode_unit.frameType)?,
host_processing_latency: decode_unit.frameHostProcessingLatency,
receive_time_ms: decode_unit.receiveTimeMs,
enqueue_time_ms: decode_unit.enqueueTimeMs,
presentation_time: decode_unit.presentationTimeMs as u64,
full_length: <usize>::try_from(decode_unit.fullLength)?,
buffer: Buffer {
data: buffer,
buffer_type: BufferType::PICDATA,
},
index,
hdr_active: decode_unit.hdrActive,
colorspace: decode_unit.colorspace,
})
if let RendererMessage::DecodeUnitStart {
ref mut num_buffers,
..
} = messages[0]
{
*num_buffers = index;
}
Ok(messages)
}
}
@@ -261,23 +278,29 @@ extern "C" fn submit_decode_unit_cb(decode_unit: PDECODE_UNIT) -> std::os::raw::
return -1;
}
let decode_unit = unsafe { *decode_unit };
//debug!("decode unit bytes: {}", decode_unit.fullLength);
let message = match RendererMessage::from_decode_unit(decode_unit) {
let messages = match RendererMessage::from_decode_unit(decode_unit) {
Ok(m) => m,
Err(e) => {
error!("Cannot construct RendererMessage: {e}");
return -1;
}
};
debug!(
"got decode unit with {} buffers: {:?}",
messages.len() - 1,
std::time::Instant::now()
);
send_message(message)
//for msg in messages {
// let ret = send_message(msg);
// if ret != 0 {
// return ret;
// }
//}
//0
for msg in messages {
let ret = send_message(msg);
if ret != 0 {
return ret;
}
}
//debug!("dispatched decode unit: {:?}", std::time::Instant::now());
0
}
pub fn decoder_callbacks() -> Result<(DECODER_RENDERER_CALLBACKS, mpsc::Receiver<RendererMessage>)>
gamestream-webtransport-proxy/src/proxy/handler.rs
+14 -3
View File
@@ -29,6 +29,12 @@ async fn setup_webtransport(
) -> Result<(
impl tokio::io::AsyncWrite + Send + Sync + 'static,
impl tokio::io::AsyncRead + Send + Sync + 'static,
h3_datagram::datagram_handler::DatagramSender<
<h3_quinn::Connection as h3_datagram::quic_traits::DatagramConnectionExt<
salvo::hyper::body::Bytes,
>>::SendDatagramHandler,
salvo::hyper::body::Bytes,
>,
//salvo::webtransport::stream::SendStream<
// impl salvo::proto::quic::SendStream<salvo::hyper::body::Bytes>,
// salvo::hyper::body::Bytes,
@@ -39,13 +45,17 @@ async fn setup_webtransport(
//>,
)> {
let session = req.web_transport_mut().await?;
let datagram_send = session.datagram_sender();
let bidirectional_stream = session
.accept_bi()
.await?
.ok_or(anyhow!("No bidirectional stream"))?;
if let webtransport::server::AcceptedBi::BidiStream(_, stream) = bidirectional_stream {
Ok(stream.split())
let (stream_send, stream_recv) = stream.split();
Ok((stream_send, stream_recv, datagram_send))
} else {
Err(anyhow!("bidirectional stream was of the wrong type"))
}
@@ -76,7 +86,8 @@ impl crate::proxy::Proxy {
});
info!("WebTransport connection initiated");
let (wt_send, wt_recv) = match setup_webtransport(req).await {
let (wt_stream_send, wt_stream_recv, wt_datagram_send) = match setup_webtransport(req).await
{
Ok(w) => w,
Err(e) => {
error!("Could not upgrade connection to WebTransport: {e}");
@@ -95,7 +106,7 @@ impl crate::proxy::Proxy {
}
};
match super::proxy_main(stream, wt_send, wt_recv).await {
match super::proxy_main(stream, wt_stream_send, wt_stream_recv, wt_datagram_send).await {
Ok(()) => Ok(()),
Err(e) => {
error!("Proxy main loop failed: {e}");
gamestream-webtransport-proxy/src/proxy/mod.rs
+7 -1
View File
@@ -34,6 +34,12 @@ async fn proxy_main(
stream: backend::Stream,
mut wt_send: impl tokio::io::AsyncWrite + Send + Sync + 'static + std::marker::Unpin,
mut wt_recv: impl tokio::io::AsyncRead + Send + Sync + 'static + std::marker::Unpin,
mut wt_datagram_send: h3_datagram::datagram_handler::DatagramSender<
<h3_quinn::Connection as h3_datagram::quic_traits::DatagramConnectionExt<
salvo::hyper::body::Bytes,
>>::SendDatagramHandler,
salvo::hyper::body::Bytes,
>,
) -> Result<()> {
debug!(
"Connecting to stream at address {} with stream config {:?}",
@@ -50,7 +56,7 @@ async fn proxy_main(
gamestream_packet = channels.gamestream_channels.decoder_rx.recv() => {
match gamestream_packet {
Some(frame) => {
video::send_video_update(&frame, &mut wt_send).await?;
video::send_video_update(&frame, &mut wt_datagram_send).await?;
}
None => {
error!("Decoder channel is None");
gamestream-webtransport-proxy/src/proxy/video/mod.rs
+90 -19
View File
@@ -1,5 +1,4 @@
use anyhow::Result;
use tokio::io::AsyncWriteExt;
use tracing::debug;
use crate::gamestream;
@@ -16,7 +15,7 @@ fn create_setup_videoupdate(
let mut builder = flatbuffers::FlatBufferBuilder::with_capacity(1024);
//TODO: this is hardcoded to h264 main profile
let video_format = Some(builder.create_string("avc1.4D002A"));
let video_format = Some(builder.create_string("avc1.4D401E"));
let setup = video_update::Setup::create(
&mut builder,
@@ -41,27 +40,60 @@ fn create_setup_videoupdate(
builder.finished_data().to_vec()
}
fn create_decodeunit_videoupdate(
fn create_decodeunitstart_videoupdate(
frame_number: u64,
frame_type: &gamestream::decoder::FrameType,
num_buffers: u64,
receive_time_ms: u64,
buffer: &gamestream::decoder::Buffer,
full_length: u64,
) -> Vec<u8> {
let mut builder = flatbuffers::FlatBufferBuilder::with_capacity(1024);
let data_vector = builder.create_vector(&buffer.data);
let frame_type_fb = match frame_type {
gamestream::decoder::FrameType::IDR => video_update::FrameType::IDR,
gamestream::decoder::FrameType::PFRAME => video_update::FrameType::PFRAME,
};
let decode_unit = video_update::DecodeUnit::create(
let decode_unit_start = video_update::DecodeUnitStart::create(
&mut builder,
&video_update::DecodeUnitArgs {
frame_number: frame_number as u16,
&video_update::DecodeUnitStartArgs {
frame_number,
frame_type: frame_type_fb,
num_buffers,
receive_time_ms: receive_time_ms as u16,
full_length,
},
);
let video_update = video_update::VideoUpdate::create(
&mut builder,
&video_update::VideoUpdateArgs {
update_type: video_update::Update::DecodeUnitStart,
update: Some(decode_unit_start.as_union_value()),
},
);
builder.finish(video_update, None);
builder.finished_data().to_vec()
}
fn create_decodeunitbuffer_videoupdate(
frame_number: u64,
buffer_index: u64,
buffer_offset: u64,
buffer: &gamestream::decoder::Buffer,
) -> Vec<u8> {
let mut builder = flatbuffers::FlatBufferBuilder::with_capacity(1024);
let data_vector = builder.create_vector(&buffer.data);
let decode_unit_buffer = video_update::DecodeUnitBuffer::create(
&mut builder,
&video_update::DecodeUnitBufferArgs {
frame_number,
buffer_index,
buffer_offset,
data: Some(data_vector),
},
);
@@ -69,8 +101,8 @@ fn create_decodeunit_videoupdate(
let video_update = video_update::VideoUpdate::create(
&mut builder,
&video_update::VideoUpdateArgs {
update_type: video_update::Update::DecodeUnit,
update: Some(decode_unit.as_union_value()),
update_type: video_update::Update::DecodeUnitBuffer,
update: Some(decode_unit_buffer.as_union_value()),
},
);
@@ -81,8 +113,14 @@ fn create_decodeunit_videoupdate(
pub async fn send_video_update(
frame: &gamestream::decoder::RendererMessage,
mut wt_send: impl tokio::io::AsyncWrite + Send + Sync + std::marker::Unpin,
wt_datagram_send: &mut h3_datagram::datagram_handler::DatagramSender<
<h3_quinn::Connection as h3_datagram::quic_traits::DatagramConnectionExt<
salvo::hyper::body::Bytes,
>>::SendDatagramHandler,
salvo::hyper::body::Bytes,
>,
) -> Result<()> {
let mut print_time = false;
let buffer = match frame {
gamestream::decoder::RendererMessage::Setup {
video_format,
@@ -91,24 +129,57 @@ pub async fn send_video_update(
redraw_rate,
dr_flags,
} => create_setup_videoupdate(video_format, *width, *height, *redraw_rate),
gamestream::decoder::RendererMessage::DecodeUnit {
gamestream::decoder::RendererMessage::DecodeUnitStart {
frame_number,
frame_type,
num_buffers,
host_processing_latency,
receive_time_ms,
enqueue_time_ms,
presentation_time,
full_length,
buffer,
index,
hdr_active,
colorspace,
} => create_decodeunit_videoupdate(*frame_number, frame_type, *receive_time_ms, buffer),
} => {
//debug!(
// "sending decodeunitstart {}: {:?}",
// *frame_number,
// std::time::Instant::now()
//);
create_decodeunitstart_videoupdate(
*frame_number,
frame_type,
*num_buffers,
*receive_time_ms,
*full_length,
)
}
gamestream::decoder::RendererMessage::DecodeUnitBuffer {
frame_number,
buffer_index,
buffer_offset,
buffer,
} => {
//debug!(
// "sending decodeunitbuffer {}/{}: {:?}",
// *frame_number,
// *buffer_index,
// std::time::Instant::now()
//);
create_decodeunitbuffer_videoupdate(
*frame_number,
*buffer_index,
*buffer_offset,
buffer,
)
}
};
let buffer_len = buffer.len() as u32;
let bytes = salvo::hyper::body::Bytes::copy_from_slice(&buffer);
wt_send.write_all(&buffer_len.to_le_bytes()).await?;
wt_send.write_all(&buffer).await?;
wt_datagram_send.send_datagram(bytes)?;
debug!("sent start: {:?}", std::time::Instant::now());
//if (print_time) {
//}
Ok(())
}
gamestream-webtransport-proxy/src/proxy/video/video_generated.rs
+246 -54
View File
@@ -106,13 +106,14 @@ impl flatbuffers::SimpleToVerifyInSlice for FrameType {}
#[deprecated(since = "2.0.0", note = "Use associated constants instead. This will no longer be generated in 2021.")]
pub const ENUM_MIN_UPDATE: u8 = 0;
#[deprecated(since = "2.0.0", note = "Use associated constants instead. This will no longer be generated in 2021.")]
pub const ENUM_MAX_UPDATE: u8 = 2;
pub const ENUM_MAX_UPDATE: u8 = 3;
#[deprecated(since = "2.0.0", note = "Use associated constants instead. This will no longer be generated in 2021.")]
#[allow(non_camel_case_types)]
pub const ENUM_VALUES_UPDATE: [Update; 3] = [
pub const ENUM_VALUES_UPDATE: [Update; 4] = [
Update::NONE,
Update::Setup,
Update::DecodeUnit,
Update::DecodeUnitStart,
Update::DecodeUnitBuffer,
];
#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
@@ -122,21 +123,24 @@ pub struct Update(pub u8);
impl Update {
pub const NONE: Self = Self(0);
pub const Setup: Self = Self(1);
pub const DecodeUnit: Self = Self(2);
pub const DecodeUnitStart: Self = Self(2);
pub const DecodeUnitBuffer: Self = Self(3);
pub const ENUM_MIN: u8 = 0;
pub const ENUM_MAX: u8 = 2;
pub const ENUM_MAX: u8 = 3;
pub const ENUM_VALUES: &'static [Self] = &[
Self::NONE,
Self::Setup,
Self::DecodeUnit,
Self::DecodeUnitStart,
Self::DecodeUnitBuffer,
];
/// Returns the variant's name or "" if unknown.
pub fn variant_name(self) -> Option<&'static str> {
match self {
Self::NONE => Some("NONE"),
Self::Setup => Some("Setup"),
Self::DecodeUnit => Some("DecodeUnit"),
Self::DecodeUnitStart => Some("DecodeUnitStart"),
Self::DecodeUnitBuffer => Some("DecodeUnitBuffer"),
_ => None,
}
}
@@ -342,150 +346,315 @@ impl core::fmt::Debug for Setup<'_> {
ds.finish()
}
}
pub enum DecodeUnitOffset {}
pub enum DecodeUnitStartOffset {}
#[derive(Copy, Clone, PartialEq)]
pub struct DecodeUnit<'a> {
pub struct DecodeUnitStart<'a> {
pub _tab: flatbuffers::Table<'a>,
}
impl<'a> flatbuffers::Follow<'a> for DecodeUnit<'a> {
type Inner = DecodeUnit<'a>;
impl<'a> flatbuffers::Follow<'a> for DecodeUnitStart<'a> {
type Inner = DecodeUnitStart<'a>;
#[inline]
unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
Self { _tab: flatbuffers::Table::new(buf, loc) }
}
}
impl<'a> DecodeUnit<'a> {
impl<'a> DecodeUnitStart<'a> {
pub const VT_FRAME_NUMBER: flatbuffers::VOffsetT = 4;
pub const VT_FRAME_TYPE: flatbuffers::VOffsetT = 6;
pub const VT_RECEIVE_TIME_MS: flatbuffers::VOffsetT = 8;
pub const VT_DATA: flatbuffers::VOffsetT = 10;
pub const VT_NUM_BUFFERS: flatbuffers::VOffsetT = 8;
pub const VT_RECEIVE_TIME_MS: flatbuffers::VOffsetT = 10;
pub const VT_FULL_LENGTH: flatbuffers::VOffsetT = 12;
#[inline]
pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
DecodeUnit { _tab: table }
DecodeUnitStart { _tab: table }
}
#[allow(unused_mut)]
pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>(
_fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>,
args: &'args DecodeUnitArgs<'args>
) -> flatbuffers::WIPOffset<DecodeUnit<'bldr>> {
let mut builder = DecodeUnitBuilder::new(_fbb);
if let Some(x) = args.data { builder.add_data(x); }
builder.add_receive_time_ms(args.receive_time_ms);
args: &'args DecodeUnitStartArgs
) -> flatbuffers::WIPOffset<DecodeUnitStart<'bldr>> {
let mut builder = DecodeUnitStartBuilder::new(_fbb);
builder.add_full_length(args.full_length);
builder.add_num_buffers(args.num_buffers);
builder.add_frame_number(args.frame_number);
builder.add_receive_time_ms(args.receive_time_ms);
builder.add_frame_type(args.frame_type);
builder.finish()
}
#[inline]
pub fn frame_number(&self) -> u16 {
pub fn frame_number(&self) -> u64 {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<u16>(DecodeUnit::VT_FRAME_NUMBER, Some(0)).unwrap()}
unsafe { self._tab.get::<u64>(DecodeUnitStart::VT_FRAME_NUMBER, Some(0)).unwrap()}
}
#[inline]
pub fn frame_type(&self) -> FrameType {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<FrameType>(DecodeUnit::VT_FRAME_TYPE, Some(FrameType::PFRAME)).unwrap()}
unsafe { self._tab.get::<FrameType>(DecodeUnitStart::VT_FRAME_TYPE, Some(FrameType::PFRAME)).unwrap()}
}
#[inline]
pub fn num_buffers(&self) -> u64 {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<u64>(DecodeUnitStart::VT_NUM_BUFFERS, Some(0)).unwrap()}
}
#[inline]
pub fn receive_time_ms(&self) -> u16 {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<u16>(DecodeUnit::VT_RECEIVE_TIME_MS, Some(0)).unwrap()}
unsafe { self._tab.get::<u16>(DecodeUnitStart::VT_RECEIVE_TIME_MS, Some(0)).unwrap()}
}
#[inline]
pub fn data(&self) -> Option<flatbuffers::Vector<'a, u8>> {
pub fn full_length(&self) -> u64 {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(DecodeUnit::VT_DATA, None)}
unsafe { self._tab.get::<u64>(DecodeUnitStart::VT_FULL_LENGTH, Some(0)).unwrap()}
}
}
impl flatbuffers::Verifiable for DecodeUnit<'_> {
impl flatbuffers::Verifiable for DecodeUnitStart<'_> {
#[inline]
fn run_verifier(
v: &mut flatbuffers::Verifier, pos: usize
) -> Result<(), flatbuffers::InvalidFlatbuffer> {
use self::flatbuffers::Verifiable;
v.visit_table(pos)?
.visit_field::<u16>("frame_number", Self::VT_FRAME_NUMBER, false)?
.visit_field::<u64>("frame_number", Self::VT_FRAME_NUMBER, false)?
.visit_field::<FrameType>("frame_type", Self::VT_FRAME_TYPE, false)?
.visit_field::<u64>("num_buffers", Self::VT_NUM_BUFFERS, false)?
.visit_field::<u16>("receive_time_ms", Self::VT_RECEIVE_TIME_MS, false)?
.visit_field::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'_, u8>>>("data", Self::VT_DATA, false)?
.visit_field::<u64>("full_length", Self::VT_FULL_LENGTH, false)?
.finish();
Ok(())
}
}
pub struct DecodeUnitArgs<'a> {
pub frame_number: u16,
pub struct DecodeUnitStartArgs {
pub frame_number: u64,
pub frame_type: FrameType,
pub num_buffers: u64,
pub receive_time_ms: u16,
pub data: Option<flatbuffers::WIPOffset<flatbuffers::Vector<'a, u8>>>,
pub full_length: u64,
}
impl<'a> Default for DecodeUnitArgs<'a> {
impl<'a> Default for DecodeUnitStartArgs {
#[inline]
fn default() -> Self {
DecodeUnitArgs {
DecodeUnitStartArgs {
frame_number: 0,
frame_type: FrameType::PFRAME,
num_buffers: 0,
receive_time_ms: 0,
data: None,
full_length: 0,
}
}
}
pub struct DecodeUnitBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> {
pub struct DecodeUnitStartBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> {
fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>,
start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
}
impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> DecodeUnitBuilder<'a, 'b, A> {
impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> DecodeUnitStartBuilder<'a, 'b, A> {
#[inline]
pub fn add_frame_number(&mut self, frame_number: u16) {
self.fbb_.push_slot::<u16>(DecodeUnit::VT_FRAME_NUMBER, frame_number, 0);
pub fn add_frame_number(&mut self, frame_number: u64) {
self.fbb_.push_slot::<u64>(DecodeUnitStart::VT_FRAME_NUMBER, frame_number, 0);
}
#[inline]
pub fn add_frame_type(&mut self, frame_type: FrameType) {
self.fbb_.push_slot::<FrameType>(DecodeUnit::VT_FRAME_TYPE, frame_type, FrameType::PFRAME);
self.fbb_.push_slot::<FrameType>(DecodeUnitStart::VT_FRAME_TYPE, frame_type, FrameType::PFRAME);
}
#[inline]
pub fn add_num_buffers(&mut self, num_buffers: u64) {
self.fbb_.push_slot::<u64>(DecodeUnitStart::VT_NUM_BUFFERS, num_buffers, 0);
}
#[inline]
pub fn add_receive_time_ms(&mut self, receive_time_ms: u16) {
self.fbb_.push_slot::<u16>(DecodeUnit::VT_RECEIVE_TIME_MS, receive_time_ms, 0);
self.fbb_.push_slot::<u16>(DecodeUnitStart::VT_RECEIVE_TIME_MS, receive_time_ms, 0);
}
#[inline]
pub fn add_data(&mut self, data: flatbuffers::WIPOffset<flatbuffers::Vector<'b , u8>>) {
self.fbb_.push_slot_always::<flatbuffers::WIPOffset<_>>(DecodeUnit::VT_DATA, data);
pub fn add_full_length(&mut self, full_length: u64) {
self.fbb_.push_slot::<u64>(DecodeUnitStart::VT_FULL_LENGTH, full_length, 0);
}
#[inline]
pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> DecodeUnitBuilder<'a, 'b, A> {
pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> DecodeUnitStartBuilder<'a, 'b, A> {
let start = _fbb.start_table();
DecodeUnitBuilder {
DecodeUnitStartBuilder {
fbb_: _fbb,
start_: start,
}
}
#[inline]
pub fn finish(self) -> flatbuffers::WIPOffset<DecodeUnit<'a>> {
pub fn finish(self) -> flatbuffers::WIPOffset<DecodeUnitStart<'a>> {
let o = self.fbb_.end_table(self.start_);
flatbuffers::WIPOffset::new(o.value())
}
}
impl core::fmt::Debug for DecodeUnit<'_> {
impl core::fmt::Debug for DecodeUnitStart<'_> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
let mut ds = f.debug_struct("DecodeUnit");
let mut ds = f.debug_struct("DecodeUnitStart");
ds.field("frame_number", &self.frame_number());
ds.field("frame_type", &self.frame_type());
ds.field("num_buffers", &self.num_buffers());
ds.field("receive_time_ms", &self.receive_time_ms());
ds.field("full_length", &self.full_length());
ds.finish()
}
}
pub enum DecodeUnitBufferOffset {}
#[derive(Copy, Clone, PartialEq)]
pub struct DecodeUnitBuffer<'a> {
pub _tab: flatbuffers::Table<'a>,
}
impl<'a> flatbuffers::Follow<'a> for DecodeUnitBuffer<'a> {
type Inner = DecodeUnitBuffer<'a>;
#[inline]
unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
Self { _tab: flatbuffers::Table::new(buf, loc) }
}
}
impl<'a> DecodeUnitBuffer<'a> {
pub const VT_FRAME_NUMBER: flatbuffers::VOffsetT = 4;
pub const VT_BUFFER_INDEX: flatbuffers::VOffsetT = 6;
pub const VT_BUFFER_OFFSET: flatbuffers::VOffsetT = 8;
pub const VT_DATA: flatbuffers::VOffsetT = 10;
#[inline]
pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self {
DecodeUnitBuffer { _tab: table }
}
#[allow(unused_mut)]
pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>(
_fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>,
args: &'args DecodeUnitBufferArgs<'args>
) -> flatbuffers::WIPOffset<DecodeUnitBuffer<'bldr>> {
let mut builder = DecodeUnitBufferBuilder::new(_fbb);
builder.add_buffer_offset(args.buffer_offset);
builder.add_buffer_index(args.buffer_index);
builder.add_frame_number(args.frame_number);
if let Some(x) = args.data { builder.add_data(x); }
builder.finish()
}
#[inline]
pub fn frame_number(&self) -> u64 {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<u64>(DecodeUnitBuffer::VT_FRAME_NUMBER, Some(0)).unwrap()}
}
#[inline]
pub fn buffer_index(&self) -> u64 {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<u64>(DecodeUnitBuffer::VT_BUFFER_INDEX, Some(0)).unwrap()}
}
#[inline]
pub fn buffer_offset(&self) -> u64 {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<u64>(DecodeUnitBuffer::VT_BUFFER_OFFSET, Some(0)).unwrap()}
}
#[inline]
pub fn data(&self) -> Option<flatbuffers::Vector<'a, u8>> {
// Safety:
// Created from valid Table for this object
// which contains a valid value in this slot
unsafe { self._tab.get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(DecodeUnitBuffer::VT_DATA, None)}
}
}
impl flatbuffers::Verifiable for DecodeUnitBuffer<'_> {
#[inline]
fn run_verifier(
v: &mut flatbuffers::Verifier, pos: usize
) -> Result<(), flatbuffers::InvalidFlatbuffer> {
use self::flatbuffers::Verifiable;
v.visit_table(pos)?
.visit_field::<u64>("frame_number", Self::VT_FRAME_NUMBER, false)?
.visit_field::<u64>("buffer_index", Self::VT_BUFFER_INDEX, false)?
.visit_field::<u64>("buffer_offset", Self::VT_BUFFER_OFFSET, false)?
.visit_field::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'_, u8>>>("data", Self::VT_DATA, false)?
.finish();
Ok(())
}
}
pub struct DecodeUnitBufferArgs<'a> {
pub frame_number: u64,
pub buffer_index: u64,
pub buffer_offset: u64,
pub data: Option<flatbuffers::WIPOffset<flatbuffers::Vector<'a, u8>>>,
}
impl<'a> Default for DecodeUnitBufferArgs<'a> {
#[inline]
fn default() -> Self {
DecodeUnitBufferArgs {
frame_number: 0,
buffer_index: 0,
buffer_offset: 0,
data: None,
}
}
}
pub struct DecodeUnitBufferBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> {
fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>,
start_: flatbuffers::WIPOffset<flatbuffers::TableUnfinishedWIPOffset>,
}
impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> DecodeUnitBufferBuilder<'a, 'b, A> {
#[inline]
pub fn add_frame_number(&mut self, frame_number: u64) {
self.fbb_.push_slot::<u64>(DecodeUnitBuffer::VT_FRAME_NUMBER, frame_number, 0);
}
#[inline]
pub fn add_buffer_index(&mut self, buffer_index: u64) {
self.fbb_.push_slot::<u64>(DecodeUnitBuffer::VT_BUFFER_INDEX, buffer_index, 0);
}
#[inline]
pub fn add_buffer_offset(&mut self, buffer_offset: u64) {
self.fbb_.push_slot::<u64>(DecodeUnitBuffer::VT_BUFFER_OFFSET, buffer_offset, 0);
}
#[inline]
pub fn add_data(&mut self, data: flatbuffers::WIPOffset<flatbuffers::Vector<'b , u8>>) {
self.fbb_.push_slot_always::<flatbuffers::WIPOffset<_>>(DecodeUnitBuffer::VT_DATA, data);
}
#[inline]
pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> DecodeUnitBufferBuilder<'a, 'b, A> {
let start = _fbb.start_table();
DecodeUnitBufferBuilder {
fbb_: _fbb,
start_: start,
}
}
#[inline]
pub fn finish(self) -> flatbuffers::WIPOffset<DecodeUnitBuffer<'a>> {
let o = self.fbb_.end_table(self.start_);
flatbuffers::WIPOffset::new(o.value())
}
}
impl core::fmt::Debug for DecodeUnitBuffer<'_> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
let mut ds = f.debug_struct("DecodeUnitBuffer");
ds.field("frame_number", &self.frame_number());
ds.field("buffer_index", &self.buffer_index());
ds.field("buffer_offset", &self.buffer_offset());
ds.field("data", &self.data());
ds.finish()
}
@@ -556,13 +725,28 @@ impl<'a> VideoUpdate<'a> {
#[inline]
#[allow(non_snake_case)]
pub fn update_as_decode_unit(&self) -> Option<DecodeUnit<'a>> {
if self.update_type() == Update::DecodeUnit {
pub fn update_as_decode_unit_start(&self) -> Option<DecodeUnitStart<'a>> {
if self.update_type() == Update::DecodeUnitStart {
self.update().map(|t| {
// Safety:
// Created from a valid Table for this object
// Which contains a valid union in this slot
unsafe { DecodeUnit::init_from_table(t) }
unsafe { DecodeUnitStart::init_from_table(t) }
})
} else {
None
}
}
#[inline]
#[allow(non_snake_case)]
pub fn update_as_decode_unit_buffer(&self) -> Option<DecodeUnitBuffer<'a>> {
if self.update_type() == Update::DecodeUnitBuffer {
self.update().map(|t| {
// Safety:
// Created from a valid Table for this object
// Which contains a valid union in this slot
unsafe { DecodeUnitBuffer::init_from_table(t) }
})
} else {
None
@@ -581,7 +765,8 @@ impl flatbuffers::Verifiable for VideoUpdate<'_> {
.visit_union::<Update, _>("update_type", Self::VT_UPDATE_TYPE, "update", Self::VT_UPDATE, false, |key, v, pos| {
match key {
Update::Setup => v.verify_union_variant::<flatbuffers::ForwardsUOffset<Setup>>("Update::Setup", pos),
Update::DecodeUnit => v.verify_union_variant::<flatbuffers::ForwardsUOffset<DecodeUnit>>("Update::DecodeUnit", pos),
Update::DecodeUnitStart => v.verify_union_variant::<flatbuffers::ForwardsUOffset<DecodeUnitStart>>("Update::DecodeUnitStart", pos),
Update::DecodeUnitBuffer => v.verify_union_variant::<flatbuffers::ForwardsUOffset<DecodeUnitBuffer>>("Update::DecodeUnitBuffer", pos),
_ => Ok(()),
}
})?
@@ -643,8 +828,15 @@ impl core::fmt::Debug for VideoUpdate<'_> {
ds.field("update", &"InvalidFlatbuffer: Union discriminant does not match value.")
}
},
Update::DecodeUnit => {
if let Some(x) = self.update_as_decode_unit() {
Update::DecodeUnitStart => {
if let Some(x) = self.update_as_decode_unit_start() {
ds.field("update", &x)
} else {
ds.field("update", &"InvalidFlatbuffer: Union discriminant does not match value.")
}
},
Update::DecodeUnitBuffer => {
if let Some(x) = self.update_as_decode_unit_buffer() {
ds.field("update", &x)
} else {
ds.field("update", &"InvalidFlatbuffer: Union discriminant does not match value.")
gamestream-webtransport-proxy/src/stream.rs
+4 -1
View File
@@ -315,7 +315,10 @@ impl crate::backend::Backend {
let webtransport_url = url_constructor::UrlConstructor::new()
.scheme("https")
.host(host)
// TODO: this is hardcoded to 127.0.0.1 to fix problems with
// tls certificates and IPv6 in chrome. This needs to eventually be fixed
// but I don't actually know what the fix is
.host("127.0.0.1")
.port(port)
.subdir("api/stream/connect")
.build();
schema/video.fbs
+12 -3
View File
@@ -13,18 +13,27 @@ enum FrameType: byte {
IDR,
}
table DecodeUnit {
frame_number: uint16;
table DecodeUnitStart {
frame_number: uint64;
frame_type: FrameType;
num_buffers: uint64;
receive_time_ms: uint16;
full_length: uint64;
}
table DecodeUnitBuffer {
frame_number: uint64;
buffer_index: uint64;
buffer_offset: uint64;
data: [ubyte];
}
union Update {
Setup:Setup,
DecodeUnit:DecodeUnit,
DecodeUnitStart:DecodeUnitStart,
DecodeUnitBuffer:DecodeUnitBuffer,
}
table VideoUpdate {