Merge pull request #721 from jianjunz/quic

Reland "Add an agent for WebTransport Support."

Author: taste1981

doc/servermd/WebTransportGuide.md

@@ -0,0 +1,23 @@
+# Open WebRTC Toolkit WebTransport Support
+
+## Overview
+
+Open WebRTC Toolkit (OWT) added [WebTransport](https://w3c.github.io/webtransport/) over [QUIC](https://tools.ietf.org/html/draft-ietf-quic-transport-32) support since 5.0. At this moment, it only supports simple data forwarding.
+
+WebTransport over other protocols are not support.
+
+## Streams
+
+Similar to media streams, a stream for arbitrary data also supports `publish` and `subscribe`. A new property `data` is added to indicate if a stream is for arbitrary data. Media features like mixing and transcoding are not supported. If you develop client applications without OWT Client SDKs, please refer to [PR 125](https://github.com/open-webrtc-toolkit/owt-server/pull/125) for protocol changes and payload format. OWT JavaScript SDK and C++ SDK support streams for arbitrary data, please see API references of these two SDKs for newly added APIs. Basically, client side creates data streams, then `publish` and `subscribe` just as media streams. `streamadded` event also fires for data streams.
+
+## QUIC agent
+
+QUIC agent handles WebTransport connections from clients. It depends on [OWT QUIC SDK](https://github.com/open-webrtc-toolkit/owt-deps-quic). By default, it listens on 7700 port. Only one instance is running on a single machine or a docker environment.
+
+## Roadmap
+
+We are planning to support follow features, but we don't have an ETA so far.
+
+- Signaling over QUIC. Instead of creating a WebSocket connection between a client and portal, we could use a WebTransport connection for both signaling and other data.
+- Media over QUIC. Sending and receiving audio and video data over QUIC instead of WebRTC.
+- Datagram support. Datagram support with FEC and NACK may help us to reduce the latency when sending media over QUIC.

scripts/build.js

@@ -170,4 +170,4 @@ Promise.all(works)
       let checkOutput = execSync(`node ${moduleTestScript} ${runtimeAddonDir}`).toString();
       console.log(checkOutput);
     }
-  });
+  });

scripts/build.json

@@ -49,6 +49,12 @@
             "webrtc-frame"
         ]
     },
+    "mcu-quic": {
+        "include":[
+            "mcu",
+            "quic"
+        ]
+    },
     "video-mixer-sw" : {
         "path" : "source/agent/video/videoMixer/videoMixer_sw",
         "gyp" : "binding.sw.gyp"
@@ -111,5 +117,9 @@
     },
     "resource-util" : {
         "path" : "source/agent/addons/resourceUtil"
+    },
+    "quic" : {
+        "path" : "source/agent/addons/quic",
+        "description": "QUIC agent supports WebTransport - QuicTransport connections. Experimental feature."
     }
 }

scripts/pack.js

@@ -40,6 +40,8 @@ const originCwd = cwd();
 const osScript = path.join(rootDir, 'scripts/detectOS.sh');
 const osType = execSync(`bash ${osScript}`).toString().toLowerCase();
 
+const experimentalTargets = ['quic-agent'];
+
 var allTargets = [];
 
 if (options.full) {
@@ -131,7 +133,8 @@ function getPackList(targets) {
   }
 
   var packList = targets.filter((element) => {
-    if (options.target.includes('all')) return true;
+    // Don't include QUIC agent by default until CI is added for QUIC SDK.
+    if (options.target.includes('all') && !experimentalTargets.includes(element.rules.name)) return true;
     return options.target.includes(element.rules.name);
   });
   if (packList.length === 0) {
@@ -598,6 +601,9 @@ function packScripts() {
   }
   scriptItems.push('app');
   scriptItems.forEach((m) => {
+    if (experimentalTargets.includes(m)) {
+      return;
+    }
     startCommands += '${bin}/daemon.sh start ' + m + ' $1\n';
     stopCommands += '${bin}/daemon.sh stop ' + m + '\n';
   });

scripts/release/daemon-bin.sh

@@ -169,6 +169,13 @@ case $startStop in
           > "${stdout}" 2>&1 </dev/null &
         echo $! > ${pid}
         ;;
+      quic-agent )
+        cd ${OWT_HOME}/quic_agent
+        export LD_LIBRARY_PATH=./lib:${LD_LIBRARY_PATH}
+        nohup nice -n ${OWT_NICENESS} ./OWT-MCU-Agent . -U quic\
+          > "${stdout}" 2>&1 </dev/null &
+        echo $! > ${pid}
+        ;;
       management-console )
         cd ${OWT_HOME}/management_console
         nohup nice -n ${OWT_NICENESS} node . \

scripts/release/daemon-mcu.sh

@@ -169,6 +169,13 @@ case $startStop in
           > "${stdout}" 2>&1 </dev/null &
         echo $! > ${pid}
         ;;
+      quic-agent )
+        cd ${OWT_HOME}/quic_agent
+        export LD_LIBRARY_PATH=./lib:${LD_LIBRARY_PATH}
+        nohup nice -n ${OWT_NICENESS} node . -U quic\
+          > "${stdout}" 2>&1 </dev/null &
+        echo $! > ${pid}
+        ;;
       analytics-agent )
         cd ${OWT_HOME}/analytics_agent
         export LD_LIBRARY_PATH=./lib:${LD_LIBRARY_PATH}

source/agent/addons/quic/QuicFactory.cc

@@ -0,0 +1,29 @@
+/*
+ * Copyright (C) 2019 Intel Corporation
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "QuicFactory.h"
+#include "owt/quic/quic_transport_factory.h"
+#include <mutex>
+
+DEFINE_LOGGER(QuicFactory, "QuicFactory");
+
+std::once_flag getQuicFactoryOnce;
+
+std::shared_ptr<QuicFactory> QuicFactory::s_quicFactory = nullptr;
+
+QuicFactory::QuicFactory()
+    : m_quicTransportFactory(std::shared_ptr<owt::quic::QuicTransportFactory>(owt::quic::QuicTransportFactory::Create()))
+{
+}
+
+std::shared_ptr<owt::quic::QuicTransportFactory> QuicFactory::getQuicTransportFactory()
+{
+    std::call_once(getQuicFactoryOnce, []() {
+        QuicFactory* factory = new QuicFactory();
+        s_quicFactory = std::shared_ptr<QuicFactory>(factory);
+    });
+    return s_quicFactory->m_quicTransportFactory;
+}

source/agent/addons/quic/QuicFactory.h

@@ -0,0 +1,32 @@
+/*
+ * Copyright (C) 2019 Intel Corporation
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef QUIC_QUICFACTORY_H_
+#define QUIC_QUICFACTORY_H_
+
+#include <memory>
+#include <logger.h>
+
+namespace owt {
+namespace quic {
+    class QuicTransportFactory;
+}
+}
+
+class QuicFactory {
+public:
+    DECLARE_LOGGER();
+    virtual ~QuicFactory() = default;
+    static std::shared_ptr<owt::quic::QuicTransportFactory> getQuicTransportFactory();
+
+private:
+    explicit QuicFactory();
+
+    static std::shared_ptr<QuicFactory> s_quicFactory;
+    std::shared_ptr<owt::quic::QuicTransportFactory> m_quicTransportFactory;
+};
+
+#endif

source/agent/addons/quic/QuicTransportConnection.cc

@@ -0,0 +1,124 @@
+/*
+ * Copyright (C) 2020 Intel Corporation
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "QuicTransportConnection.h"
+#include "QuicTransportStream.h"
+
+using v8::Function;
+using v8::FunctionTemplate;
+using v8::Local;
+using v8::Object;
+using v8::ObjectTemplate;
+using v8::Value;
+
+Nan::Persistent<v8::Function> QuicTransportConnection::s_constructor;
+
+DEFINE_LOGGER(QuicTransportConnection, "QuicTransportConnection");
+
+QuicTransportConnection::QuicTransportConnection()
+    : m_session(nullptr)
+    , m_visitor(nullptr)
+{
+}
+
+QuicTransportConnection::~QuicTransportConnection()
+{
+    if (!uv_is_closing(reinterpret_cast<uv_handle_t*>(&m_asyncOnStream))) {
+        uv_close(reinterpret_cast<uv_handle_t*>(&m_asyncOnStream), NULL);
+    }
+    m_session->SetVisitor(nullptr);
+}
+
+NAN_MODULE_INIT(QuicTransportConnection::init)
+{
+    Local<FunctionTemplate> tpl = Nan::New<FunctionTemplate>(newInstance);
+    tpl->SetClassName(Nan::New("QuicTransportConnection").ToLocalChecked());
+    Local<ObjectTemplate> instanceTpl = tpl->InstanceTemplate();
+    instanceTpl->SetInternalFieldCount(1);
+
+    Nan::SetPrototypeMethod(tpl, "createBidirectionalStream", createBidirectionalStream);
+
+    s_constructor.Reset(Nan::GetFunction(tpl).ToLocalChecked());
+    Nan::Set(target, Nan::New("QuicTransportConnection").ToLocalChecked(), Nan::GetFunction(tpl).ToLocalChecked());
+}
+
+void QuicTransportConnection::setVisitor(Visitor* visitor)
+{
+    m_visitor = visitor;
+}
+
+void QuicTransportConnection::OnIncomingStream(owt::quic::QuicTransportStreamInterface* stream)
+{
+    ELOG_DEBUG("OnIncomingStream.");
+    {
+        std::lock_guard<std::mutex> lock(m_streamQueueMutex);
+        m_streamsToBeNotified.push(stream);
+    }
+    m_asyncOnStream.data = this;
+    uv_async_send(&m_asyncOnStream);
+}
+
+void QuicTransportConnection::onEnded()
+{
+}
+
+NAN_METHOD(QuicTransportConnection::newInstance)
+{
+    if (!info.IsConstructCall()) {
+        ELOG_DEBUG("Not construct call.");
+        return;
+    }
+    QuicTransportConnection* obj = new QuicTransportConnection();
+    obj->Wrap(info.This());
+    uv_async_init(uv_default_loop(), &obj->m_asyncOnStream, &QuicTransportConnection::onStreamCallback);
+    info.GetReturnValue().Set(info.This());
+}
+
+v8::Local<v8::Object> QuicTransportConnection::newInstance(owt::quic::QuicTransportSessionInterface* session)
+{
+    Local<Object> connectionObject = Nan::NewInstance(Nan::New(QuicTransportConnection::s_constructor)).ToLocalChecked();
+    QuicTransportConnection* obj = Nan::ObjectWrap::Unwrap<QuicTransportConnection>(connectionObject);
+    obj->m_session = session;
+    return connectionObject;
+}
+
+NAUV_WORK_CB(QuicTransportConnection::onStreamCallback)
+{
+    ELOG_DEBUG("OnStreamCallback.");
+    Nan::HandleScope scope;
+    QuicTransportConnection* obj = reinterpret_cast<QuicTransportConnection*>(async->data);
+    if (obj == nullptr || obj->m_streamsToBeNotified.empty()) {
+        return;
+    }
+    while (!obj->m_streamsToBeNotified.empty()) {
+        obj->m_streamQueueMutex.lock();
+        auto quicStream=obj->m_streamsToBeNotified.front();
+        obj->m_streamsToBeNotified.pop();
+        obj->m_streamQueueMutex.unlock();
+        v8::Local<v8::Object> streamObject = QuicTransportStream::newInstance(quicStream);
+        QuicTransportStream* stream = Nan::ObjectWrap::Unwrap<QuicTransportStream>(streamObject);
+        quicStream->SetVisitor(stream);
+        Nan::MaybeLocal<v8::Value> onEvent = Nan::Get(obj->handle(), Nan::New<v8::String>("onincomingstream").ToLocalChecked());
+        if (!onEvent.IsEmpty()) {
+            v8::Local<v8::Value> onEventLocal = onEvent.ToLocalChecked();
+            if (onEventLocal->IsFunction()) {
+                v8::Local<v8::Function> eventCallback = onEventLocal.As<Function>();
+                Nan::AsyncResource* resource = new Nan::AsyncResource(Nan::New<v8::String>("onincomingstream").ToLocalChecked());
+                Local<Value> args[] = { streamObject };
+                resource->runInAsyncScope(Nan::GetCurrentContext()->Global(), eventCallback, 1, args);
+            }
+        } else {
+            ELOG_DEBUG("onEvent is empty");
+        }
+    }
+}
+
+NAN_METHOD(QuicTransportConnection::createBidirectionalStream){
+    QuicTransportConnection* obj = Nan::ObjectWrap::Unwrap<QuicTransportConnection>(info.Holder());
+    auto stream=obj->m_session->CreateBidirectionalStream();
+    v8::Local<v8::Object> streamObject = QuicTransportStream::newInstance(stream);
+    info.GetReturnValue().Set(streamObject);
+}

source/agent/addons/quic/QuicTransportConnection.h

@@ -0,0 +1,64 @@
+/*
+ * Copyright (C) 2020 Intel Corporation
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef QUIC_QUICTRANSPORTCONNECTION_H_
+#define QUIC_QUICTRANSPORTCONNECTION_H_
+
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+#include <vector>
+
+#include <logger.h>
+#include <nan.h>
+
+#include "QuicTransportStream.h"
+#include "owt/quic/quic_transport_session_interface.h"
+
+class QuicTransportConnection : public Nan::ObjectWrap, public owt::quic::QuicTransportSessionInterface::Visitor, QuicTransportStream::Visitor {
+    DECLARE_LOGGER();
+
+public:
+    class Visitor {
+        // Authentication passed with user ID `id`.
+        virtual void onAuthentication(const std::string& id) = 0;
+        // Connection is closed.
+        virtual void onClose() = 0;
+    };
+    explicit QuicTransportConnection();
+    ~QuicTransportConnection();
+    void setVisitor(Visitor* visitor);
+    static v8::Local<v8::Object> newInstance(owt::quic::QuicTransportSessionInterface* session);
+
+    static NAN_MODULE_INIT(init);
+    static NAN_METHOD(newInstance);
+    static NAN_METHOD(createBidirectionalStream);
+    static NAUV_WORK_CB(onStreamCallback);
+
+    static Nan::Persistent<v8::Function> s_constructor;
+
+protected:
+    // Overrides owt::quic::QuicTransportSessionInterface::Visitor.
+    void OnIncomingStream(owt::quic::QuicTransportStreamInterface*) override;
+    void OnCanCreateNewOutgoingStream(bool unidirectional) override { }
+
+    // Overrides QuicTransportStream::Visitor.
+    void onEnded() override;
+
+private:
+    owt::quic::QuicTransportSessionInterface* m_session;
+    Visitor* m_visitor;
+    // Key is content session ID, i.e.: publication ID, subscription ID.
+    std::unordered_map<std::string, std::unique_ptr<QuicTransportStream>> m_streams;
+    // Move to `m_streams` after associate with a publication or subscription.
+    std::vector<std::unique_ptr<QuicTransportStream>> m_unassociatedStreams;
+
+    uv_async_t m_asyncOnStream;
+    std::mutex m_streamQueueMutex;
+    std::queue<owt::quic::QuicTransportStreamInterface*> m_streamsToBeNotified;
+};
+
+#endif