Receive mixed stream with WebTransport datagram.

After RTP packets are received, it uses WebRTC RTP depacketizer for
depacketizing, WebCodecs for decoding and MediaStreamGenerator for
rendering.

Author: jianjunz

src/samples/conference/public/quic.html

@@ -30,13 +30,19 @@ <h2>Sample of QuicTransport</h2>
   <div class="operations">
     <button id="start-sending">Start sending</button>
     <button id="stop-sending">Stop sending</button>
+    <button id="start-receiving">Start receiving</button>
+    <button id="stop-receiving">Stop receiving</button>
+  </div>
+  <div id="videocontainer">
+    <video id="remote-video" playsinline muted autoplay controls></video>
   </div>
   <div id="conference-status"></div>
   <div class="stats">
   </div>
-  <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/3.1.2/socket.io.js" type="text/javascript"></script>
+
+  <script src="scripts/socket.io.js" type="text/javascript"></script>
   <script src="scripts/rest-sample.js" type="text/javascript"></script>
-  <script src="scripts/owt.js" type="text/javascript"></script>
+  <script src="../../../../dist/sdk-debug/owt.js" type="text/javascript"></script>
   <script src="scripts/quic.js" type="text/javascript"></script>
 </body>
 

src/samples/conference/public/scripts/media-worker.js

@@ -3,9 +3,10 @@
 // SPDX-License-Identifier: Apache-2.0
 
 /* eslint-disable require-jsdoc */
-/* global VideoEncoder */
+/* global VideoEncoder, VideoDecoder, EncodedVideoChunk */
 
-let bidirectionalStreamWritable, videoEncoder, frameBuffer, sendStreamWriter;
+let bidirectionalStreamWritable, videoEncoder, frameBuffer, sendStreamWriter,
+    mediaSession, datagramReceiver, videoDecoder;
 // 4 bytes for frame size before each frame. The 1st byte is reserved, always 0.
 const sizePrefix = 4;
 
@@ -16,7 +17,15 @@ onmessage = (e) => {
     bidirectionalStreamWritable = e.data[1];
     sendStreamWriter = bidirectionalStreamWritable.getWriter();
     writeTrackId();
-    initVideoEncoder();
+    // initVideoEncoder();
+  } else if (e.data[0] === 'datagram-receiver') {
+    datagramReceiver = e.data[1];
+  } else if (e.data[0] === 'encoded-video-frame') {
+    if (videoDecoder.state === 'closed') {
+      return;
+    }
+    videoDecoder.decode(new EncodedVideoChunk(
+        {timestamp: Date.now(), data: e.data[1], type: 'key'}));
   }
 };
 
@@ -60,6 +69,22 @@ function initVideoEncoder() {
   });
 }
 
+function initVideoDecoder() {
+  videoDecoder = new VideoDecoder({
+    output: videoFrameOutputCallback,
+    error: webCodecsErrorCallback,
+  });
+  videoDecoder.configure({codec: 'avc1.42400a', optimizeForLatency: true});
+}
+
+function videoFrameOutputCallback(frame) {
+  postMessage(['video-frame', frame], [frame]);
+}
+
+function webCodecsErrorCallback(error) {
+  console.log('error: ' + error.message);
+}
+
 // Read data from video track.
 async function readVideoData(readable) {
   const reader = readable.getReader();
@@ -72,4 +97,21 @@ async function readVideoData(readable) {
     videoEncoder.encode(value);
     value.close();
   }
-}
+}
+
+async function fetchWasm() {
+  const Module={};
+  Module['instantiateWasm'] = async (imports, successCallback) => {
+    const response = await fetch('./owt.wasm');
+    const buffer = await response.arrayBuffer();
+    const module=new WebAssembly.Module(buffer);
+    const instance = await WebAssembly.instantiate(module, imports);
+    successCallback(instance, module);
+    return {};
+  };
+  // Module['wasmModule']=new WebAssembly.Module(buffer);
+  importScripts('./owt.js');
+  console.log('Got wasm binary.');
+}
+
+initVideoDecoder();

src/samples/conference/public/scripts/quic.js

@@ -8,15 +8,25 @@
 
 let quicChannel = null;
 let bidirectionalStream = null;
-let writeTask, mediaStream, mediaWorker, conferenceId, myId;
+let writeTask, mediaStream, mediaWorker, conferenceId, myId, mixedStream, generatorWriter;
+
+window.Module={};
 
 const conference = new Owt.Conference.ConferenceClient({
   webTransportConfiguration: {
     serverCertificateFingerprints: [{
       value:
-          'DD:A8:11:FD:A1:08:17:41:36:CD:1A:33:1E:CF:AE:0D:46:3D:15:16:2C:67:C5:A2:06:35:C2:0E:88:A1:9E:C6',
+          '59:74:C6:C5:2C:D8:E8:18:A9:D2:14:77:ED:94:89:87:DF:83:BA:B3:96:4C:4C:0B:B8:D3:22:58:11:55:67:1A',
+      algorithm: 'sha-256',
+    }],
+    serverCertificateHashes: [{
+      value: new Uint8Array([
+        0x59, 0x74, 0xC6, 0xC5, 0x2C, 0xD8, 0xE8, 0x18, 0xA9, 0xD2, 0x14,
+        0x77, 0xED, 0x94, 0x89, 0x87, 0xDF, 0x83, 0xBA, 0xB3, 0x96, 0x4C,
+        0x4C, 0x0B, 0xB8, 0xD3, 0x22, 0x58, 0x11, 0x55, 0x67, 0x1A
+      ]),
       algorithm: 'sha-256',
-    }]
+    }],
   }
 });
 conference.addEventListener('streamadded', async (event) => {
@@ -26,35 +36,51 @@ conference.addEventListener('streamadded', async (event) => {
         conferenceId, event.stream.id, 'common',
         'http://jianjunz-nuc-ubuntu.sh.intel.com:3001');
   }
-  if (event.stream.source.data || event.stream.source.video) {
-    const subscription = await conference.subscribe(
-        event.stream,
-        {audio: false, video: {codecs: ['h264']}, transport: {type: 'quic'}});
-    const reader = subscription.stream.readable.getReader();
-    while (true) {
-      const {value, done} = await reader.read();
-      if (done) {
-        console.log('Subscription ends.');
-        break;
-      }
-      console.log('Received data: ' + value);
-    }
-  }
+  // if (event.stream.source.data) {
+  //   const subscription = await conference.subscribe(
+  //       event.stream,
+  //       // {transport:{type: 'quic'}});
+  //       {audio: false, video: {codecs: ['h264']}, transport: {type: 'quic'}});
+  //   const reader = subscription.stream.readable.getReader();
+  //   while (true) {
+  //     const {value, done} = await reader.read();
+  //     if (done) {
+  //       console.log('Subscription ends.');
+  //       break;
+  //     }
+  //     //console.log('Received data: ' + value);
+  //   }
+  // }
 });
 
 function updateConferenceStatus(message) {
   document.getElementById('conference-status').innerHTML +=
       ('<p>' + message + '</p>');
 }
 
+function initWorker() {
+  mediaWorker = new Worker('./scripts/media-worker.js');
+  mediaWorker.onmessage=((e) => {
+    if (e.data[0] === 'video-frame') {
+      generatorWriter.write(e.data[1]);
+      //console.log(e.data[1]);
+    }
+  });
+}
 
 function joinConference() {
   return new Promise((resolve, reject) => {
     createToken(undefined, 'user', 'presenter', token => {
       conference.join(token).then((info) => {
         conferenceId = info.id;
         myId = info.self.id;
+        for (const stream of info.remoteStreams) {
+          if (stream.source.video === 'mixed') {
+            mixedStream = stream;
+          }
+        }
         updateConferenceStatus('Connected to conference server.');
+        initWorker();
         resolve();
       });
     }, 'http://jianjunz-nuc-ubuntu.sh.intel.com:3001');
@@ -89,17 +115,17 @@ async function attachReader(stream) {
 
 async function createSendChannel() {
   bidirectionalStream = await conference.createSendStream();
-  const localStream = new Owt.Base.LocalStream(
-      bidirectionalStream,
-      new Owt.Base.StreamSourceInfo(undefined, 'camera', undefined));
-  attachReader(bidirectionalStream);
-  const publication = await conference.publish(
-      localStream, {video: {codec: 'h264'}, transport: {type: 'quic'}});
   // const localStream = new Owt.Base.LocalStream(
   //     bidirectionalStream,
-  //     new Owt.Base.StreamSourceInfo(undefined, undefined, true));
+  //     new Owt.Base.StreamSourceInfo(undefined, 'camera', undefined));
+  // attachReader(bidirectionalStream);
   // const publication = await conference.publish(
-  //     localStream, {transport: {type: 'quic'}});
+  //     localStream, {video: {codec: 'h264'}, transport: {type: 'quic'}});
+  const localStream = new Owt.Base.LocalStream(
+      bidirectionalStream,
+      new Owt.Base.StreamSourceInfo(undefined, undefined, true));
+  const publication = await conference.publish(
+      localStream, {transport: {type: 'quic'}});
   console.log(publication);
   updateConferenceStatus('Created send channel.');
 }
@@ -123,7 +149,6 @@ async function writeUuid() {
 async function writeVideoData() {
   mediaStream = await navigator.mediaDevices.getUserMedia({video: true});
   const track = new MediaStreamTrackProcessor(mediaStream.getVideoTracks()[0]);
-  mediaWorker = new Worker('./scripts/media-worker.js');
   mediaWorker.postMessage(['video-source', track.readable], [track.readable]);
   mediaWorker.postMessage(
       ['send-stream', bidirectionalStream.writable],
@@ -135,26 +160,87 @@ async function writeData() {
   const encoded = encoder.encode('message', {stream: true});
   const writer = bidirectionalStream.writable.getWriter();
   await writer.ready;
-  await writer.write(new ArrayBuffer(2));
+  const ab=new Uint8Array(10000);
+  ab.fill(1, 0);
+  await writer.write(ab);
   writer.releaseLock();
   return;
 }
 
 window.addEventListener('load', () => {
   windowOnLoad();
+  fetchWasm();
 });
 
 document.getElementById('start-sending').addEventListener('click', async () => {
   if (!bidirectionalStream) {
     updateConferenceStatus('Stream is not created.');
     return;
   }
-  writeVideoData();
-  // writeTask = setInterval(writeData, 2000);
+  //writeVideoData();
+  writeTask = setInterval(writeData, 200);
   updateConferenceStatus('Started sending.');
 });
 
 document.getElementById('stop-sending').addEventListener('click', () => {
   clearInterval(writeTask);
   updateConferenceStatus('Stopped sending.');
 });
+
+document.getElementById('start-receiving')
+    .addEventListener('click', async () => {
+      const video=document.getElementById('remote-video');
+      const generator = new MediaStreamTrackGenerator({kind: 'video'});
+      generatorWriter=generator.writable.getWriter();
+      video.srcObject = new MediaStream([generator]);
+      const reader = conference.datagramReader();
+      const ms = new Module.MediaSession();
+      const receiver = ms.createRtpVideoReceiver();
+      receiver.setCompleteFrameCallback((frame) => {
+        const copiedFrame = frame.slice(0);
+        mediaWorker.postMessage(
+            ['encoded-video-frame', copiedFrame], [copiedFrame.buffer]);
+      });
+      subscribeMixedStream();
+      while (true) {
+        const received = await reader.read();
+        const buffer = Module._malloc(received.value.byteLength);
+        Module.writeArrayToMemory(received.value, buffer);
+        receiver.onRtpPacket(buffer, received.value.byteLength);
+      }
+    });
+
+async function fetchWasm() {
+  Module['instantiateWasm'] = async (imports, successCallback) => {
+    const response = await fetch('scripts/owt.wasm');
+    const buffer = await response.arrayBuffer();
+    const module=await WebAssembly.compile(buffer);
+    const instance = await WebAssembly.instantiate(module, imports);
+    successCallback(instance, module);
+    return {};
+  };
+  const scriptPromise = new Promise((resolve, reject) => {
+    const script = document.createElement('script');
+    document.body.appendChild(script);
+    script.onload = resolve;
+    script.onerror = reject;
+    script.async = true;
+    script.src = 'scripts/owt.js';
+  });
+  await scriptPromise;
+}
+
+async function subscribeMixedStream() {
+  const subscription = await conference.subscribe(
+      mixedStream,
+      {audio: false, video: {codecs: ['h264']}, transport: {type: 'quic'}});
+  const reader = subscription.stream.readable.getReader();
+  while (true) {
+    const {value, done} = await reader.read();
+    if (done) {
+      console.log('Subscription ends.');
+      break;
+    }
+    // console.log('Received data: ' + value);
+  }
+}

src/sdk/conference/client.js

@@ -558,5 +558,9 @@ export const ConferenceClient = function(config, signalingImpl) {
       }
       return quicTransportChannel.createSendStream();
     };
+
+    this.datagramReader = function() {
+      return quicTransportChannel.datagramReader();
+    };
   }
 };

src/sdk/conference/webtransport/connection.js

@@ -40,6 +40,7 @@ export class QuicConnection extends EventDispatcher {
         new Map();  // Key is subscription ID, value is a promise.
     this._transportId = this._token.transportId;
     this._initReceiveStreamReader();
+    //this._initDatagrams();
   }
 
   /**
@@ -73,6 +74,14 @@ export class QuicConnection extends EventDispatcher {
     await this._authenticate(this._tokenString);
   }
 
+  async _initDatagrams() {
+    const datagramReader = this._quicTransport.datagrams.readable.getReader();
+    while (true) {
+      const value = await datagramReader.read();
+      console.log(value);
+    }
+  }
+
   async _initReceiveStreamReader() {
     const receiveStreamReader =
         this._quicTransport.incomingBidirectionalStreams.getReader();
@@ -275,12 +284,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) {
@@ -416,4 +425,8 @@ export class QuicConnection extends EventDispatcher {
     // Ready message from server is useless for QuicStream since QuicStream has
     // its own status. Do nothing here.
   }
+
+  datagramReader() {
+    return this._quicTransport.datagrams.readable.getReader();
+  }
 }