Move RTP processing from sample to SDK.

Author: jianjunz

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

@@ -63,16 +63,6 @@ function updateConferenceStatus(message) {
       ('<p>' + message + '</p>');
 }
 
-function initWorker() {
-  dataWorker = new Worker('./scripts/data-worker.js');
-  dataWorker.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 => {
@@ -85,7 +75,6 @@ function joinConference() {
           }
         }
         updateConferenceStatus('Connected to conference server.');
-        initWorker();
         resolve();
       });
     }, 'http://jianjunz-nuc-ubuntu.sh.intel.com:3001');
@@ -153,7 +142,6 @@ async function writeData() {
 
 window.addEventListener('load', () => {
   windowOnLoad();
-  fetchWasm();
 });
 
 document.getElementById('start-sending').addEventListener('click', async () => {
@@ -190,58 +178,12 @@ document.getElementById('stop-sending').addEventListener('click', () => {
 
 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);
-        dataWorker.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);
-  }
+  document.getElementById('remote-video').srcObject = subscription.stream;
 }

src/sdk/conference/client.js

@@ -161,8 +161,7 @@ export const ConferenceClient = function(config, workerDir, signalingImpl) {
     if (notification === 'soac' || notification === 'progress') {
       if (channels.has(data.id)) {
         channels.get(data.id).onMessage(notification, data);
-      } else if (quicTransportChannel && quicTransportChannel
-          .hasContentSessionId(data.id)) {
+      } else if (quicTransportChannel) {
         quicTransportChannel.onMessage(notification, data);
       } else {
         Logger.warning('Cannot find a channel for incoming data.');

src/sdk/conference/webtransport/connection.js

@@ -42,9 +42,13 @@ export class QuicConnection extends EventDispatcher {
     this._subscribeOptions = new Map(); // Key is subscription ID.
     this._subscriptionInfoReady =
         new Map(); // Key is subscription ID, value is a promise.
+    // Key is subscription ID, value is an object with audio and video RTP
+    // configs.
+    this._rtpConfigs = new Map();
     this._transportId = this._token.transportId;
     this._initReceiveStreamReader();
     this._worker = new Worker(workerDir + '/media-worker.js', {type: 'module'});
+    this._initHandlersForWorker();
     // Key is subscription ID, value is a MediaStreamTrackGenerator writer.
     this._mstVideoGeneratorWriters = new Map();
     this._initRtpModule();
@@ -68,6 +72,8 @@ export class QuicConnection extends EventDispatcher {
           this._readyHandler();
         } else if (message.status === 'error') {
           this._errorHandler(message.data);
+        } else if (message.status === 'rtp'){
+          this._rtpHandler(message);
         }
         break;
       case 'stream':
@@ -507,17 +513,24 @@ export class QuicConnection extends EventDispatcher {
               // A MediaStream is associated with a subscription for media.
               // Media packets are received over WebTransport datagram.
               const generators = [];
-              for (const track of mediaOptions) {
+              for (const track of mediaOptions.tracks) {
                 const generator =
                     new MediaStreamTrackGenerator({kind: track.type});
                 generators.push(generator);
                 // TODO: Update key with the correct SSRC.
                 this._mstVideoGeneratorWriters.set(
-                    '0', generator.writable.getWriter());
+                    data.id, generator.writable.getWriter());
               }
               const mediaStream = new MediaStream(generators);
               const subscription =
                   this._createSubscription(data.id, mediaStream);
+              this._worker.postMessage([
+                'add-subscription',
+                [
+                  subscription.id, options,
+                  this._rtpConfigs.get(subscription.id)
+                ]
+              ]);
               resolve(subscription);
             }
             if (this._subscriptionInfoReady.has(data.id)) {
@@ -582,17 +595,21 @@ export class QuicConnection extends EventDispatcher {
     // its own status. Do nothing here.
   }
 
+  _rtpHandler(message) {
+    Logger.debug(`RTP config: ${JSON.stringify(message.data)}.`);
+    this._rtpConfigs.set(message.id, message.data);
+  }
+
   datagramReader() {
     return this._quicTransport.datagrams.readable.getReader();
   }
 
-  initHandlersForWorker() {
+  _initHandlersForWorker() {
     this._worker.onmessage = ((e) => {
       const [command, args] = e.data;
       switch (command) {
         case 'video-frame':
-          // TODO: Use actual subscription ID.
-          this._mstVideoGeneratorWriters.get('0').getWriter.write(args);
+          this._mstVideoGeneratorWriters.get(args[0]).write(args[1]);
           break;
         default:
           Logger.warn('Unrecognized command ' + command);

src/sdk/conference/webtransport/media-worker.js

@@ -19,7 +19,8 @@ const keyFrameRequested = new Map();
 
 let wasmModule;
 let mediaSession;
-let rtpReceiver;
+// Key is SSRC, value is an RTP receiver.
+let rtpReceivers = new Map();
 let frameBuffer;
 let videoDecoder;
 // 4 bytes for frame size before each frame. The 1st byte is reserved, always 0.
@@ -46,6 +47,9 @@ onmessage = async (e) => {
     case 'rtp-packet':
       await handleRtpPacket(args);
       break;
+    case 'add-subscription':
+      addNewSubscription(...args);
+      break;
     default:
       console.warn('Unrecognized command ' + command);
   }
@@ -69,14 +73,8 @@ async function initMediaSender(
 }
 
 async function initRtpModule() {
-  initVideoDecoder();
   wasmModule = await fetchWasm();
   mediaSession = new wasmModule.MediaSession();
-  rtpReceiver = mediaSession.createRtpVideoReceiver();
-  rtpReceiver.setCompleteFrameCallback((frame) => {
-    videoDecoder.decode(new EncodedVideoChunk(
-        {timestamp: Date.now(), data: frame, type: 'key'}));
-  });
 }
 
 async function fetchWasm() {
@@ -157,17 +155,17 @@ function initVideoEncoder(config, writer) {
   return videoEncoder;
 }
 
-function initVideoDecoder() {
+function initVideoDecoder(subscriptionId) {
   videoDecoder = new VideoDecoder({
-    output: videoFrameOutputCallback,
+    output: videoFrameOutputCallback.bind(null, subscriptionId),
     error: webCodecsErrorCallback,
   });
   videoDecoder.configure({codec: 'avc1.42400a', optimizeForLatency: true});
 }
 
-function videoFrameOutputCallback(frame) {
+function videoFrameOutputCallback(subscriptionId, frame) {
   // eslint-disable-next-line no-undef
-  postMessage(['video-frame', frame], [frame]);
+  postMessage(['video-frame', [subscriptionId, frame]], [frame]);
   frame.close();
 }
 
@@ -196,8 +194,29 @@ async function readMediaData(trackReadable, encoder, publicationId) {
   }
 }
 
+function getSsrc(packet) {
+  // SSRC starts from the 65th bit, in network order.
+  return new DataView(packet.buffer).getUint32(8, false);
+}
+
 async function handleRtpPacket(packet) {
+  const ssrc = getSsrc(packet);
   const buffer = wasmModule._malloc(packet.byteLength);
   wasmModule.writeArrayToMemory(packet, buffer);
-  rtpReceiver.onRtpPacket(buffer, packet.byteLength);
+  rtpReceivers.get(ssrc).onRtpPacket(buffer, packet.byteLength);
 }
+
+function addNewSubscription(subscriptionId, subscribeOptions, rtpConfig) {
+  // TODO: Audio is not supported yet, ignore the audio part.
+  initVideoDecoder(subscriptionId);
+  const videoSsrc = rtpConfig.video.ssrc;
+  if (rtpReceivers.has(videoSsrc)) {
+    console.error(`RTP receiver for SSRC ${videoSsrc} exits.`);
+  }
+  const rtpReceiver = mediaSession.createRtpVideoReceiver(videoSsrc);
+  rtpReceivers.set(videoSsrc, rtpReceiver);
+  rtpReceiver.setCompleteFrameCallback((frame) => {
+    videoDecoder.decode(new EncodedVideoChunk(
+        {timestamp: Date.now(), data: frame, type: 'key'}));
+  });
+}