Enable data flow from QUIC agent to audio agent. (#1102)

* Enable data flow from QUIC agent to audio agent.

* Determine a track's kind by its ID.

* Set track kind for data tracks.

* Timestamps for audio frames are set in QUIC agent.

Author: jianjunz

source/agent/addons/quic/QuicFactory.cc

@@ -6,6 +6,7 @@
 
 #include "QuicFactory.h"
 #include "owt/quic/web_transport_factory.h"
+#include "owt/quic/logging.h"
 #include <mutex>
 
 DEFINE_LOGGER(QuicFactory, "QuicFactory");

source/agent/addons/quic/QuicTransportStream.cc

@@ -36,12 +36,14 @@ QuicTransportStream::QuicTransportStream(owt::quic::WebTransportStreamInterface*
     , m_hasSink(false)
     , m_buffer(nullptr)
     , m_bufferSize(0)
-    , m_isMedia(false)
+    , m_trackKind("unknown")
+    , m_frameFormat(owt_base::FRAME_FORMAT_UNKNOWN)
     , m_readingFrameSize(false)
     , m_frameSizeOffset(0)
     , m_frameSizeArray(new uint8_t[frameHeaderSize])
     , m_currentFrameSize(0)
     , m_receivedFrameOffset(0)
+    , m_audioTimeStamp(0)
 {
 }
 
@@ -123,7 +125,7 @@ NAN_MODULE_INIT(QuicTransportStream::init)
     Nan::SetPrototypeMethod(tpl, "write", write);
     Nan::SetPrototypeMethod(tpl, "readTrackId", readTrackId);
     Nan::SetPrototypeMethod(tpl, "addDestination", addDestination);
-    Nan::SetAccessor(instanceTpl, Nan::New("isMedia").ToLocalChecked(), isMediaGetter, isMediaSetter);
+    Nan::SetAccessor(instanceTpl, Nan::New("trackKind").ToLocalChecked(), trackKindGetter, trackKindSetter);
     Nan::SetAccessor(instanceTpl, Nan::New("ondata").ToLocalChecked(), onDataGetter, onDataSetter);
 
     s_constructor.Reset(Nan::GetFunction(tpl).ToLocalChecked());
@@ -212,15 +214,16 @@ void QuicTransportStream::CheckReadableData()
     }
 }
 
-NAN_GETTER(QuicTransportStream::isMediaGetter){
+NAN_GETTER(QuicTransportStream::trackKindGetter){
     QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
-    info.GetReturnValue().Set(Nan::New(obj->m_isMedia));
+    info.GetReturnValue().Set(Nan::New(obj->m_trackKind).ToLocalChecked());
 }
 
-NAN_SETTER(QuicTransportStream::isMediaSetter)
+NAN_SETTER(QuicTransportStream::trackKindSetter)
 {
     QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
-    obj->m_isMedia = value->ToBoolean(Nan::GetCurrentContext()).ToLocalChecked()->Value();
+    Nan::Utf8String trackKind(Nan::To<v8::String>(value).ToLocalChecked());
+    obj->m_trackKind = std::string(*trackKind);
 }
 
 NAN_GETTER(QuicTransportStream::onDataGetter)
@@ -365,10 +368,14 @@ void QuicTransportStream::SignalOnData()
         return;
     }
 
+    if (m_trackKind == "unknown") {
+        return;
+    }
+
     while (m_stream->ReadableBytes() > 0) {
         auto readableBytes = m_stream->ReadableBytes();
         // Future check if it's an audio stream or video stream. Audio is not supported at this time.
-        if (m_isMedia) {
+        if (m_trackKind == "audio" || m_trackKind == "video") {
             // A new frame.
             if (m_currentFrameSize == 0 && m_receivedFrameOffset == 0 && !m_readingFrameSize) {
                 m_readingFrameSize = true;
@@ -403,16 +410,29 @@ void QuicTransportStream::SignalOnData()
             // Complete frame.
             if (m_receivedFrameOffset == m_currentFrameSize) {
                 owt_base::Frame frame;
-                frame.format = owt_base::FRAME_FORMAT_I420;
+                if (m_trackKind == "audio") {
+                    frame.format = owt_base::FRAME_FORMAT_OPUS;
+                    frame.timeStamp = m_audioTimeStamp;
+                    // TODO: Fill a correct timestamp and check overflow.
+                    m_audioTimeStamp += 10 * 1000;
+                    // TODO: Get format from signaling message.
+                    frame.additionalInfo.audio.isRtpPacket = false;
+                    frame.additionalInfo.audio.sampleRate = 48000;
+                    frame.additionalInfo.audio.channels = 2;
+                } else if (m_trackKind == "video") {
+                    frame.format = owt_base::FRAME_FORMAT_H264;
+                    // 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";
+                } else {
+                    ELOG_ERROR("Unexpected track kind: %s.", m_trackKind);
+                }
                 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_receivedFrameOffset = 0;
             }
-        } else {
+        } else if (m_trackKind == "data") {
             if (readableBytes > m_bufferSize) {
                 ReallocateBuffer(readableBytes);
             }

source/agent/addons/quic/QuicTransportStream.h

@@ -40,8 +40,8 @@ class QuicTransportStream : public owt_base::FrameSource, public owt_base::Frame
     // TODO: Make this as an async method when it's supported.
     static NAN_METHOD(readTrackId);
 
-    static NAN_GETTER(isMediaGetter);
-    static NAN_SETTER(isMediaSetter);
+    static NAN_GETTER(trackKindGetter);
+    static NAN_SETTER(trackKindSetter);
     static NAN_GETTER(onDataGetter);
     static NAN_SETTER(onDataSetter);
 
@@ -98,15 +98,18 @@ 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;
+    // Indicates the kind of this stream, could be 'audio', 'video', 'data'. If this is not a data track, it can only be piped to another QUIC agent.
+    std::string m_trackKind;
+    owt_base::FrameFormat m_frameFormat;
     Nan::Persistent<v8::Value> m_onDataCallback;
 
     size_t m_readingFrameSize;
     size_t m_frameSizeOffset;
     uint8_t* m_frameSizeArray;
     size_t m_currentFrameSize;
     size_t m_receivedFrameOffset;
+    // TODO: Using wall clock timestamps seems not working. Using an increasing sequence instead. Fix it later.
+    uint32_t m_audioTimeStamp;
 
     uv_async_t m_asyncOnContentSessionId;
     uv_async_t m_asyncOnTrackId;

source/agent/addons/quic/quic_sdk_url

@@ -1 +1 @@
-https://api.github.com/repos/open-webrtc-toolkit/owt-sdk-quic/actions/artifacts/84639537/zip
+https://api.github.com/repos/open-webrtc-toolkit/owt-sdk-quic/actions/artifacts/97822442/zip

source/agent/quic/index.js

@@ -24,6 +24,8 @@ const addon = require('./build/Release/quic');
 const cipher = require('../cipher');
 const path = require('path');
 const {InternalConnectionRouter} = require('./internalConnectionRouter');
+const audioTrackId = '00000000000000000000000000000001';
+const videoTrackId = '00000000000000000000000000000002';
 
 log.info('QUIC transport node.')
 
@@ -105,7 +107,6 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         log.debug(
             'A stream with session ID ' + stream.contentSessionId +
             ' is added.');
-        // Set isMedia=true if the session is for media.
         if (outgoingStreamPipelines.has(stream.contentSessionId)) {
           const pipeline = outgoingStreamPipelines.get(stream.contentSessionId);
           pipeline.quicStream(stream);
@@ -119,9 +120,9 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
           const options = publicationOptions.get(stream.contentSessionId);
           // Only publications for media have tracks.
           if (options.tracks && options.tracks.length) {
-            stream.isMedia = true;
             stream.readTrackId();
           } else {
+            stream.trackKind = 'data';
             frameSourceMap.get(stream.contentSessionId).addInputStream(stream);
           }
         } else {
@@ -133,6 +134,16 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         }
       });
       quicTransportServer.on('trackid', (stream) => {
+        // TODO: Defined track ID and get track kind based on ID. Currently it
+        // is hard coded.
+        if (stream.trackId === audioTrackId) {
+          stream.trackKind = 'audio';
+        } else if (stream.trackId === videoTrackId) {
+          stream.trackKind = 'video';
+        } else {
+          log.warn('Unexpected track ID: ' + stream.trackId);
+          return;
+        }
         if (frameSourceMap.has(stream.contentSessionId)) {
           frameSourceMap.get(stream.contentSessionId).addInputStream(stream);
         }