Allow data flow between QUIC agent and video agent.

Author: jianjunz

doc/Client-Portal Protocol.md

@@ -362,14 +362,14 @@ This a format for client reconnects.
 ```
   object(PublicationRequest)::
     {
-     media: object(WebRTCMediaOptions) | null,
+     media: object(WebRTCMediaOptions) | object(WebCodecsMediaOptions) | null,
      data: true | false,
      transport: object(TransportOptions),
      attributes: object(ClientDefinedAttributes) | null
     }
 ```
 
-A publication can send either media or data, but a QUIC *transport* channel can support multiple stream for both media and data. Setting `media:null` and `data:false` is meaningless, so it should be rejected by server. Protocol itself doesn't forbit to create WebRTC connection for data. However, SCTP data channel is not implemented at server side, so currently `data:true` is only support by QUIC transport channels. 
+A publication can send either media or data. Setting `media:null` and `data:false` is meaningless, so it should be rejected by server. Protocol itself doesn't forbit to create WebRTC connection for data. However, SCTP data channel is not implemented at server side, so currently `data:true` is only support by WebTransport channels. When transport's type is "webrtc", `media` should be an object of `WebRTCMediaOptions`. When transport's type is "quic", `media` should be an object of `WebCodecsMediaOptions` or `null`.
 
 ```
   object(WebRTCMediaOptions)::
@@ -385,6 +385,20 @@ A publication can send either media or data, but a QUIC *transport* channel can
     }
 ```
 
+```
+  object(WebCodecsMediaOptions)::
+      {
+        tracks: [
+          {
+            type: "audio" | "video",
+            source: "mic" | "screen-cast" | ... | "encoded-file",
+            format: object(AudioFormat) | object(VideoFormat)
+          }
+        ]
+      }
+    }
+```
+
 **ResponseData**: The PublicationResult object with following definition if **ResponseStatus** is “ok”:
 
   object(PublicationResult)::

doc/design/quic-transport-payload-format.md renamed to doc/design/web-transport-payload-format.md

@@ -39,6 +39,22 @@ After creating a WebTransport, a stream with session 0 should be created for aut
    +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 ```
 
+## Media Stream
+
+After sending 128 bit length session ID, a 128 bit length track ID is sent to remote side to indicates the track of a stream. Since audio track and video track of a single stream shares the same track ID at this time, track 1 is for audio and track 2 is for video.
+
+When a WebTransport stream is used for transmitting data of a media stream track (e.g.: H.264 bitstream), a 32 (8+24) bit length header is added to indicate frame size.
+
+```
+    0                   1                   2                   3
+    0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |     Reserved    |               Message length                |
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+   |                          Message                            ...
+   +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
+```
+
 ## Authentication
 
 If signaling messages are transmitted over WebTransport, authentication follows the regular process defined by [Client-Portal Protocol](https://github.com/open-webrtc-toolkit/owt-server/blob/master/doc/Client-Portal%20Protocol.md). Otherwise, client sends a token for WebTransport as a signaling message. WebTransport token is issued during joining a conference. If the token is valid, server sends a 128 bit length zeros to client.

source/agent/addons/internalIO/InternalClientWrapper.cc

@@ -88,23 +88,36 @@ NAN_METHOD(InternalClient::close) {
   obj->me = nullptr;
 }
 
-NAN_METHOD(InternalClient::addDestination) {
-  InternalClient* obj = ObjectWrap::Unwrap<InternalClient>(info.Holder());
-  owt_base::InternalClient* me = obj->me;
+NAN_METHOD(InternalClient::addDestination)
+{
+    InternalClient* obj = ObjectWrap::Unwrap<InternalClient>(info.Holder());
+    owt_base::InternalClient* me = obj->me;
 
-  Nan::Utf8String param0(Nan::To<v8::String>(info[0]).ToLocalChecked());
-  std::string track = std::string(*param0);
+    Nan::Utf8String param0(Nan::To<v8::String>(info[0]).ToLocalChecked());
+    std::string track = std::string(*param0);
 
-  FrameDestination* param =
-    ObjectWrap::Unwrap<FrameDestination>(
-      info[1]->ToObject(Nan::GetCurrentContext()).ToLocalChecked());
-  owt_base::FrameDestination* dest = param->dest;
+    bool isNanDestination(false);
+    if (info.Length() >= 3) {
+        isNanDestination = info[2]->ToBoolean(Nan::GetCurrentContext()).ToLocalChecked()->Value();
+    }
 
-  if (track == "audio") {
-    me->addAudioDestination(dest);
-  } else if (track == "video") {
-    me->addVideoDestination(dest);
-  }
+    owt_base::FrameDestination* dest(nullptr);
+    if (isNanDestination) {
+        NanFrameNode* param = Nan::ObjectWrap::Unwrap<NanFrameNode>(info[1]->ToObject());
+        dest = param->FrameDestination();
+    } else {
+        FrameDestination* param = ObjectWrap::Unwrap<FrameDestination>(
+            info[1]->ToObject(Nan::GetCurrentContext()).ToLocalChecked());
+        dest = param->dest;
+    }
+
+    if (track == "audio") {
+        me->addAudioDestination(dest);
+    } else if (track == "video") {
+        me->addVideoDestination(dest);
+    } else if (track == "data") {
+        me->addDataDestination(dest);
+    }
 }
 
 NAN_METHOD(InternalClient::removeDestination) {

source/agent/addons/internalIO/InternalServerWrapper.cc

@@ -99,10 +99,20 @@ NAN_METHOD(InternalServer::addSource) {
   Nan::Utf8String param0(Nan::To<v8::String>(info[0]).ToLocalChecked());
   std::string streamId = std::string(*param0);
 
-  FrameSource* param =
-    ObjectWrap::Unwrap<FrameSource>(
-      info[1]->ToObject(Nan::GetCurrentContext()).ToLocalChecked());
-  owt_base::FrameSource* src = param->src;
+  bool isNanSource(false);
+  if (info.Length() >= 3) {
+      isNanSource = info[2]->ToBoolean(Nan::GetCurrentContext()).ToLocalChecked()->Value();
+  }
+
+  owt_base::FrameSource* src(nullptr);
+  if (isNanSource) {
+      NanFrameNode* param = Nan::ObjectWrap::Unwrap<NanFrameNode>(info[1]->ToObject());
+      src = param->FrameSource();
+  } else {
+      FrameSource* param = ObjectWrap::Unwrap<FrameSource>(
+          info[1]->ToObject(Nan::GetCurrentContext()).ToLocalChecked());
+      src = param->src;
+  }
 
   me->addSource(streamId, src);
 }

source/agent/addons/internalIO/InternalServerWrapper.h

@@ -44,7 +44,10 @@ class InternalServer : public node::ObjectWrap,
     static NAN_METHOD(close);
 
     static NAN_METHOD(getListeningPort);
-
+    // Arguments:
+    // type: string, type of the source, "audio", "video" or "data".
+    // source: A node addon object or NAN object.
+    // isNanObject: indicates whether `source` is a NAN object.
     static NAN_METHOD(addSource);
 
     static NAN_METHOD(removeSource);

source/agent/addons/quic/MediaFramePacketizer.h

@@ -0,0 +1,4 @@
+#ifndef QUIC_ADDON_MEDIA_FRAME_PACKETIZER_H_
+#define QUIC_ADDON_MEDIA_FRAME_PACKETIZER_H_
+
+#endif

source/agent/addons/quic/QuicTransportStream.cc

@@ -5,6 +5,7 @@
  */
 
 #include "QuicTransportStream.h"
+#include "../common/MediaFramePipelineWrapper.h"
 
 using v8::Function;
 using v8::FunctionTemplate;
@@ -30,6 +31,9 @@ QuicTransportStream::QuicTransportStream(owt::quic::WebTransportStreamInterface*
     , m_isPiped(false)
     , m_buffer(nullptr)
     , m_bufferSize(0)
+    , m_isMedia(false)
+    , m_currentFrameSize(0)
+    , m_readFrameSize(0)
 {
 }
 
@@ -113,15 +117,32 @@ NAN_METHOD(QuicTransportStream::close)
 NAN_METHOD(QuicTransportStream::addDestination)
 {
     QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
-    if (info.Length() != 2) {
+    if (info.Length() > 3) {
         Nan::ThrowTypeError("Invalid argument length for addDestination.");
         return;
     }
-    // TODO: Check if info[0] is an Nan wrapped object.
-    auto framePtr = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info[1]->ToObject());
-    // void* ptr = info[0]->ToObject()->GetAlignedPointerFromInternalField(0);
-    // auto framePtr=static_cast<owt_base::FrameDestination*>(ptr);
-    obj->addDataDestination(framePtr);
+    Nan::Utf8String param0(Nan::To<v8::String>(info[0]).ToLocalChecked());
+    std::string track = std::string(*param0);
+    bool isNanDestination(false);
+    if (info.Length() == 3) {
+        isNanDestination = info[2]->ToBoolean(Nan::GetCurrentContext()).ToLocalChecked()->Value();
+    }
+    owt_base::FrameDestination* dest(nullptr);
+    if (isNanDestination) {
+        NanFrameNode* param = Nan::ObjectWrap::Unwrap<NanFrameNode>(info[1]->ToObject());
+        dest = param->FrameDestination();
+    } else {
+        ::FrameDestination* param = node::ObjectWrap::Unwrap<::FrameDestination>(
+            info[1]->ToObject(Nan::GetCurrentContext()).ToLocalChecked());
+        dest = param->dest;
+    }
+    if (track == "audio") {
+        obj->addAudioDestination(dest);
+    } else if (track == "video") {
+        obj->addVideoDestination(dest);
+    } else if (track == "data") {
+        obj->addDataDestination(dest);
+    }
     obj->m_isPiped = true;
 }
 
@@ -153,6 +174,13 @@ void QuicTransportStream::MaybeReadContentSessionId()
         m_receivedContentSessionId = true;
         m_asyncOnContentSessionId.data = this;
         uv_async_send(&m_asyncOnContentSessionId);
+        for (uint8_t d : m_contentSessionId) {
+            if (d != 0) {
+                m_isPiped = true;
+                m_isMedia = true;
+                break;
+            }
+        }
         if (m_stream->ReadableBytes() > 0) {
             SignalOnData();
         }
@@ -173,7 +201,6 @@ NAUV_WORK_CB(QuicTransportStream::onData)
             v8::Local<v8::Function> eventCallback = onEventLocal.As<Function>();
             Nan::AsyncResource* resource = new Nan::AsyncResource(Nan::New<v8::String>("ondata").ToLocalChecked());
             auto readableBytes = obj->m_stream->ReadableBytes();
-            ELOG_DEBUG("Readable bytes: %d", readableBytes);
             uint8_t* buffer = new uint8_t[readableBytes]; // Use a shared buffer instead to reduce performance cost on new.
             obj->m_stream->Read(buffer, readableBytes);
             Local<Value> args[] = { Nan::NewBuffer((char*)buffer, readableBytes).ToLocalChecked() };
@@ -211,15 +238,55 @@ void QuicTransportStream::SignalOnData()
 
     while (m_stream->ReadableBytes() > 0) {
         auto readableBytes = m_stream->ReadableBytes();
-        if (readableBytes > m_bufferSize) {
-            ReallocateBuffer(readableBytes);
+        ELOG_DEBUG("Readable bytes: %d", readableBytes);
+        if (m_isMedia) {
+            // A new frame.
+            if (m_currentFrameSize == 0 && m_readFrameSize == 0) {
+                if (readableBytes >= 2) {
+                    const int headerSize = 4;  // In bytes.
+                    uint8_t* frameSizeArray = new uint8_t[headerSize];
+                    memset(frameSizeArray, 0, headerSize * sizeof(uint8_t));
+                    m_stream->Read((uint8_t*)frameSizeArray, headerSize);
+                    // Usually only the last 2 bytes are used. The first two bits could be used for indicating frame size.
+                    for (int i = 0; i < headerSize; i++) {
+                        m_currentFrameSize <<= 8;
+                        m_currentFrameSize += frameSizeArray[i];
+                    }
+                    if (m_currentFrameSize > m_bufferSize) {
+                        ReallocateBuffer(m_currentFrameSize);
+                    }
+                }
+                continue;
+            }
+            if (m_readFrameSize < m_currentFrameSize) {
+                // Append data to current frame.
+                size_t readBytes = std::min(readableBytes, m_currentFrameSize - m_readFrameSize);
+                m_stream->Read(m_buffer + m_readFrameSize, readBytes);
+                m_readFrameSize += readBytes;
+            }
+            // Complete frame.
+            if (m_readFrameSize == m_currentFrameSize) {
+                owt_base::Frame frame;
+                frame.format = owt_base::FRAME_FORMAT_I420;
+                frame.length = m_currentFrameSize;
+                frame.payload = m_buffer;
+                // Transport layer doesn't know a frame's type. Video agent is able to parse the type of a frame from bistream. However, video agent doesn't feed the frame to decoder when a key frame is requested.
+                frame.additionalInfo.video.isKeyFrame = "key";
+                deliverFrame(frame);
+                m_currentFrameSize = 0;
+                m_readFrameSize = 0;
+            }
+        } else {
+            if (readableBytes > m_bufferSize) {
+                ReallocateBuffer(readableBytes);
+            }
+            owt_base::Frame frame;
+            frame.format = owt_base::FRAME_FORMAT_DATA;
+            frame.length = readableBytes;
+            frame.payload = m_buffer;
+            m_stream->Read(frame.payload, readableBytes);
+            deliverFrame(frame);
         }
-        owt_base::Frame frame;
-        frame.format = owt_base::FRAME_FORMAT_DATA;
-        frame.length = readableBytes;
-        frame.payload = m_buffer;
-        m_stream->Read(frame.payload, readableBytes);
-        deliverFrame(frame);
     }
 }
 

source/agent/addons/quic/QuicTransportStream.h

@@ -7,12 +7,12 @@
 #ifndef QUIC_QUICTRANSPORTSTREAM_H_
 #define QUIC_QUICTRANSPORTSTREAM_H_
 
-#include "owt/quic/web_transport_stream_interface.h"
 #include "../../core/owt_base/MediaFramePipeline.h"
 #include "../common/MediaFramePipelineWrapper.h"
+#include "owt/quic/web_transport_stream_interface.h"
 #include <logger.h>
-#include <nan.h>
 #include <mutex>
+#include <nan.h>
 #include <string>
 
 class QuicTransportStream : public owt_base::FrameSource, public owt_base::FrameDestination, public NanFrameNode, public owt::quic::WebTransportStreamInterface::Visitor {
@@ -37,7 +37,7 @@ class QuicTransportStream : public owt_base::FrameSource, public owt_base::Frame
     static NAN_METHOD(addDestination);
     static NAN_METHOD(removeDestination);
     static NAUV_WORK_CB(onContentSessionId);
-    static NAUV_WORK_CB(onData);  // TODO: Move to pipe.
+    static NAUV_WORK_CB(onData); // TODO: Move to pipe.
 
     // Overrides owt_base::FrameSource.
     void onFeedback(const owt_base::FeedbackMsg&) override;
@@ -72,6 +72,11 @@ class QuicTransportStream : public owt_base::FrameSource, public owt_base::Frame
     uint8_t* m_buffer;
     size_t m_bufferSize;
 
+    // Indicates whether this is a media stream. If this is not a media stream, it can only be piped to another QUIC agent.
+    bool m_isMedia;
+    size_t m_currentFrameSize;
+    size_t m_readFrameSize;
+
     uv_async_t m_asyncOnContentSessionId;
     uv_async_t m_asyncOnData;
 };

source/agent/conference/quicController.js

@@ -34,15 +34,20 @@ class Transport {
 
 class Operation {
   constructor(id, transport, direction, tracks, data) {
-    if(tracks){
-      throw new Error('QUIC agent does not support media stream tracks so far.');
-    }
     this.id = id;
     this.transport = transport;
     this.transportId = transport.id;
     this.direction = direction;
+    this.tracks = tracks;
     this.data = data;
     this.promise = Promise.resolve();
+    this.tracks = this.tracks ? this.tracks.map(t => {
+        if (t.type === 'video') {
+            t.format = { codec : 'h264', profile : 'B' };
+        }
+        return t;
+    })
+                              : undefined;
   }
 }
 
@@ -168,6 +173,7 @@ class QuicController extends EventEmitter {
 
   // Return Promise
   terminate(sessionId, direction, reason) {
+    console.trace();
     log.debug(`terminate, sessionId: ${sessionId} direction: ${direction}, ${reason}`);
 
     if (!this.operations.has(sessionId)) {

source/agent/connections.js

@@ -124,7 +124,11 @@ module.exports = function Connections () {
                 if (!dest) {
                     return Promise.reject({ type : 'failed', reason : 'Destination connection(' + name + ') is not ready' });
                 }
-                connections[from].connection.addDestination(name, dest);
+                let isNanObj=false;
+                if (dest.constructor.name === 'QuicTransportStream'){
+                    isNanObj=true;
+                }
+                connections[from].connection.addDestination(name, dest, isNanObj);
                 connections[connectionId][name + 'From'] = from;
             }
         }

source/agent/internalConnectionRouter.js

@@ -107,7 +107,11 @@ class StreamSource {
   addDestination(track, sink) {
     if (!this.dests[track].has(sink.id)) {
       this.dests[track].set(sink.id, sink);
-      this.conn.addDestination(track, sink.conn);
+      let isNanObject = false;
+      if (sink instanceof QuicTransportStreamPipeline) {
+          isNanObject = true;
+      }
+      this.conn.addDestination(track, sink.conn, isNanObject);
       sink._addSource(track, this);
     }
   }