Skip to content
This repository has been archived by the owner on Oct 25, 2024. It is now read-only.

Commit

Permalink
Support receive MediaStream over WebTransport datagrams.
Browse files Browse the repository at this point in the history
  • Loading branch information
@jianjunz
jianjunz committed Nov 26, 2021
1 parent b80c458 commit 6e5fddb
Show file tree
Hide file tree
Showing 2 changed files with 237 additions and 43 deletions.
172 changes: 153 additions & 19 deletions src/sdk/conference/webtransport/connection.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@

/* eslint-disable require-jsdoc */
/* global Promise, Map, WebTransport, WebTransportBidirectionalStream,
Uint8Array, Uint32Array, TextEncoder, Worker, MediaStreamTrackProcessor */
Uint8Array, Uint32Array, TextEncoder, Worker, MediaStreamTrackProcessor,
proto */

'use strict';

Expand Down Expand Up @@ -44,6 +45,10 @@ export class QuicConnection extends EventDispatcher {
this._transportId = this._token.transportId;
this._initReceiveStreamReader();
this._worker = new Worker(workerDir + '/media-worker.js', {type: 'module'});
// Key is SSRC, value is a MediaStreamTrackGenerator writer.
this._mstGeneratorWriters = new Map();
this._initRtpModule();
this._initDatagramReader();
}

/**
Expand Down Expand Up @@ -77,6 +82,10 @@ export class QuicConnection extends EventDispatcher {
await this._authenticate(this._tokenString);
}

async _initRtpModule() {
this._worker.postMessage(['init-rtp']);
}

async _initReceiveStreamReader() {
const receiveStreamReader =
this._quicTransport.incomingBidirectionalStreams.getReader();
Expand Down Expand Up @@ -173,6 +182,20 @@ export class QuicConnection extends EventDispatcher {
}
}

async _initDatagramReader() {
const datagramReader = this._quicTransport.datagrams.readable.getReader();
console.log('Get datagram reader.');
let receivingDone = false;
while (!receivingDone) {
const {value: datagram, done: readingDatagramsDone} =
await datagramReader.read();
this._worker.postMessage(['rtp-packet', datagram]);
if (readingDatagramsDone) {
receivingDone = true;
}
}
}

_createSubscription(id, receiveStream) {
// TODO: Incomplete subscription.
const subscription = new Subscription(id, () => {
Expand Down Expand Up @@ -207,6 +230,94 @@ export class QuicConnection extends EventDispatcher {
return quicStream;
}

async bindFeedbackReader(stream, publicationId) {
// The receiver side of a publication stream starts with a UUID of
// publication ID, then each feedback message has a 4 bytes header indicates
// its length, and followed by protobuf encoded body.
const feedbackChunkReader = stream.readable.getReader();
let feedbackChunksDone = false;
let publicationIdOffset = 0;
const headerSize=4;
const header = new Uint8Array(headerSize);
let headerOffset = 0;
let bodySize = 0;
let bodyOffset = 0;
let bodyBytes;
while (!feedbackChunksDone) {
let valueOffset=0;
const {value, done} = await feedbackChunkReader.read();
Logger.debug(value);
while (valueOffset < value.byteLength) {
if (publicationIdOffset < uuidByteLength) {
// TODO: Check publication ID matches. For now, we just skip this ID.
const readLength =
Math.min(uuidByteLength - publicationIdOffset, value.byteLength);
valueOffset += readLength;
publicationIdOffset += readLength;
}
if (headerOffset < headerSize) {
// Read header.
const copyLength = Math.min(
headerSize - headerOffset, value.byteLength - valueOffset);
if (copyLength === 0) {
continue;
}
header.set(
value.subarray(valueOffset, valueOffset + copyLength),
headerOffset);
headerOffset += copyLength;
valueOffset += copyLength;
if (headerOffset < headerSize) {
continue;
}
bodySize = 0;
bodyOffset = 0;
for (let i = 0; i < headerSize; i++) {
bodySize += (header[i] << ((headerSize - 1 - i) * 8));
}
bodyBytes = new Uint8Array(bodySize);
Logger.debug('Body size ' + bodySize);
}
if (bodyOffset < bodySize) {
const copyLength =
Math.min(bodySize - bodyOffset, value.byteLength - valueOffset);
if (copyLength === 0) {
continue;
}
Logger.debug('Bytes for body: '+copyLength);
bodyBytes.set(
value.subarray(valueOffset, valueOffset + copyLength),
bodyOffset);
bodyOffset += copyLength;
valueOffset += copyLength;
if (valueOffset < bodySize) {
continue;
}
// Decode body.
const feedback = proto.owt.protobuf.Feedback.deserializeBinary(bodyBytes);
this.handleFeedback(feedback, publicationId);
}
}
if (done) {
feedbackChunksDone = true;
break;
}
}
}

async handleFeedback(feedback, publicationId) {
Logger.debug(
'Key frame request type: ' +
proto.owt.protobuf.Feedback.Type.KEY_FRAME_REQUEST);
if (feedback.getType() ===
proto.owt.protobuf.Feedback.Type.KEY_FRAME_REQUEST) {
this._worker.postMessage(
['rtcp-feedback', ['key-frame-request', publicationId]]);
} else {
Logger.warning('Unrecognized feedback type ' + feedback.getType());
}
}

async publish(stream, options) {
// TODO: Avoid a stream to be published twice. The first 16 bit data send to
// server must be it's publication ID.
Expand All @@ -225,6 +336,7 @@ export class QuicConnection extends EventDispatcher {
for (const track of stream.stream.getTracks()) {
const quicStream =
await this._quicTransport.createBidirectionalStream();
this.bindFeedbackReader(quicStream, publicationId);
this._quicMediaStreamTracks.set(track.id, quicStream);
quicStreams.push(quicStream);
}
Expand Down Expand Up @@ -262,6 +374,7 @@ export class QuicConnection extends EventDispatcher {
[
'media-sender',
[
publicationId,
track.id,
track.kind,
processor.readable,
Expand Down Expand Up @@ -317,12 +430,12 @@ export class QuicConnection extends EventDispatcher {
if (typeof options !== 'object') {
return Promise.reject(new TypeError('Options should be an object.'));
}
// if (options.audio === undefined) {
// options.audio = !!stream.settings.audio;
// }
// if (options.video === undefined) {
// options.video = !!stream.settings.video;
// }
if (options.audio === undefined) {
options.audio = !!stream.settings.audio;
}
if (options.video === undefined) {
options.video = !!stream.settings.video;
}
let mediaOptions;
let dataOptions;
if (options.audio || options.video) {
Expand Down Expand Up @@ -375,19 +488,40 @@ export class QuicConnection extends EventDispatcher {
})
.then((data) => {
this._subscribeOptions.set(data.id, options);
Logger.debug('Subscribe info is set.');
if (this._quicDataStreams.has(data.id)) {
// QUIC stream created before signaling returns.
// TODO: Update subscription to accept list of QUIC streams.
const subscription = this._createSubscription(
data.id, this._quicDataStreams.get(data.id)[0]);
resolve(subscription);
if (dataOptions) {
// A WebTransport stream is associated with a subscription for
// data.
if (this._quicDataStreams.has(data.id)) {
// QUIC stream created before signaling returns.
// TODO: Update subscription to accept list of QUIC streams.
const subscription = this._createSubscription(
data.id, this._quicDataStreams.get(data.id)[0]);
resolve(subscription);
} else {
this._quicDataStreams.set(data.id, null);
// QUIC stream is not created yet, resolve promise after getting
// QUIC stream.
this._subscribePromises.set(
data.id, {resolve: resolve, reject: reject});
}
} else {
this._quicDataStreams.set(data.id, null);
// QUIC stream is not created yet, resolve promise after getting
// QUIC stream.
this._subscribePromises.set(
data.id, {resolve: resolve, reject: reject});
// A MediaStream is associated with a subscription for media.
// Media packets are received over WebTransport datagram.
const ms = new Module.MediaSession();
console.log('Created media session.');
const generators = [];
for (const track of mediaOptions){
const generator =
new MediaStreamTrackGenerator({kind: track.type});
generators.push(generator);
// TODO: Update key with the correct SSRC.
this._mstGeneratorWriters.set(
0, generator.writable.getWriter());
}
const mediaStream = new MediaStream(generators);
const subscription =
this._createSubscription(data.id, mediaStream);
resolve(subscription);
}
if (this._subscriptionInfoReady.has(data.id)) {
this._subscriptionInfoReady.get(data.id)();
Expand Down
Loading

0 comments on commit 6e5fddb

Please sign in to comment.