Merge branch 'master' of https://github.com/DWJ01/owt-server

Author: DWJ01

README.md

@@ -34,7 +34,7 @@ The ${webrtc-javascript-sdk-sample-conference-dist} is built from owt-javascript
 If "--archive ${name}" option is appended to the pack command, a "Release-${name}.tgz" file will be generated in root folder. For other options, run the scripts with "--help" option.
 
 ## Quick Start
-In the repository root, run the following commands to start the media server on a single machine:
+In dist folder, run the following commands to start the media server on a single machine:
 1. `./bin/init-all.sh --deps`
 2. `./bin/start-all.sh`
 3. Open https://localhost:3004 to visit the web sample page. Due to the test certificate, you may need confirm this unsafe access.

source/agent/analytics/analytics-agent.js

@@ -116,6 +116,13 @@ createInternalConnection(connectionId, direction, internalOpt) {
           this.onStreamDestroyed(options.controller, newStreamId);
       }
       this.inputs[connectionId] = true;
+
+      var notifyStatus = this.onStatus;
+      this.engine.addEventListener('fatal', function (error) {
+          log.error('GStreamer pipeline error:', error);
+          notifyStatus(options.controller, connectionId, 'out', {type: 'failed', reason: 'Analytics error: ' + error});
+      });
+
       return Promise.resolve();
     }
     return Promise.reject('Connection already exist: ' + connectionId);

source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.cc

@@ -79,7 +79,8 @@ static void dump(void* index, uint8_t* buf, int len)
     }
 }
 
-VideoGstAnalyzer::VideoGstAnalyzer() {
+VideoGstAnalyzer::VideoGstAnalyzer(EventRegistry *handle) : m_asyncHandle(handle)
+{
     ELOG_INFO("Init");
     sourceid = 0;
     sink = NULL;
@@ -94,78 +95,104 @@ VideoGstAnalyzer::VideoGstAnalyzer() {
     }
 }
 
-VideoGstAnalyzer::~VideoGstAnalyzer() {
+VideoGstAnalyzer::~VideoGstAnalyzer()
+{
     ELOG_DEBUG("Closed all media in this Analyzer");
-    if (pipeline_ != nullptr && pipelineHandle != nullptr) {
-         destroyPlugin(pipeline_);
-         dlclose(pipelineHandle);
+    destroyPipeline();
+}
+
+bool VideoGstAnalyzer::notifyAsyncEvent(const std::string& event, const std::string& data)
+{
+    if (m_asyncHandle) {
+        return m_asyncHandle->notifyAsyncEvent(event, data);
+    } else {
+        return false;
     }
-    stopLoop();
 }
 
-gboolean VideoGstAnalyzer::StreamEventCallBack(GstBus *bus, GstMessage *message, gpointer data)
-    {
-        ELOG_DEBUG("Got %s message\n", GST_MESSAGE_TYPE_NAME(message));
- 
-        VideoGstAnalyzer* pStreamObj = static_cast<VideoGstAnalyzer*>(data);
- 
-        switch (GST_MESSAGE_TYPE(message)) {
-        case GST_MESSAGE_ERROR: {
-            GError *err;
-            gchar *debug;
-            gst_message_parse_error(message, &err, &debug);
-            ELOG_ERROR("Error: %s\n", err->message);
-            g_error_free(err);
-            g_free(debug);
-            g_main_loop_quit(pStreamObj->loop);
-            break;
-        }
-        case GST_MESSAGE_EOS:
-            /* end-of-stream */
-            ELOG_ERROR("End of stream\n");
-            g_main_loop_quit(pStreamObj->loop);
-            break;
-        case GST_MESSAGE_TAG:{
-            /* end-of-stream */
-            GstTagList *tags = NULL;
-            gst_message_parse_tag (message, &tags);
-
-            ELOG_DEBUG("Got tags from element %s:\n", GST_OBJECT_NAME (message->src));
-            gst_tag_list_unref (tags);
-            break;
-        }
-        case GST_MESSAGE_QOS:{
-            /* end-of-stream */
-            ELOG_DEBUG("Got QOS message from %s \n",message->src->name);
-            break;
-        }
-        case GST_MESSAGE_STATE_CHANGED:{
-            GstState old_state, new_state, pending_state;
-            gst_message_parse_state_changed(message, &old_state, &new_state, &pending_state);
-            ELOG_DEBUG("State change from %d to %d, play:%d \n",old_state, new_state, GST_STATE_PAUSED);
-            break;
-        }
-        default:
-            /* unhandled message */
-            break;
+bool VideoGstAnalyzer::notifyAsyncEventInEmergency(const std::string& event, const std::string& data)
+{
+    if (m_asyncHandle) {
+            return m_asyncHandle->notifyAsyncEventInEmergency(event, data);
+        } else {
+            return false;
         }
-        return true;
+}
+
+gboolean VideoGstAnalyzer::StreamEventCallBack(GstBus *bus, GstMessage *message, gpointer data)
+{
+    ELOG_DEBUG("Got %s message\n", GST_MESSAGE_TYPE_NAME(message));
+
+    VideoGstAnalyzer* pStreamObj = static_cast<VideoGstAnalyzer*>(data);
+
+    switch (GST_MESSAGE_TYPE(message)) {
+    case GST_MESSAGE_ERROR: {
+        GError *err;
+        gchar *debug;
+        gst_message_parse_error(message, &err, &debug);
+        ELOG_ERROR("Error: %s\n", err->message);
+        g_error_free(err);
+        g_free(debug);
+        pStreamObj->notifyAsyncEvent("fatal", "GStreamer pipeline error");
+        break;
+    }
+    case GST_MESSAGE_EOS:
+        /* end-of-stream */
+        ELOG_ERROR("End of stream\n");
+        g_main_loop_quit(pStreamObj->loop);
+        break;
+    case GST_MESSAGE_TAG:{
+        /* end-of-stream */
+        GstTagList *tags = NULL;
+        gst_message_parse_tag (message, &tags);
+
+        ELOG_DEBUG("Got tags from element %s:\n", GST_OBJECT_NAME (message->src));
+        gst_tag_list_unref (tags);
+        break;
+    }
+    case GST_MESSAGE_QOS:{
+        /* end-of-stream */
+        ELOG_DEBUG("Got QOS message from %s \n",message->src->name);
+        break;
     }
+    case GST_MESSAGE_STATE_CHANGED:{
+        GstState old_state, new_state, pending_state;
+        gst_message_parse_state_changed(message, &old_state, &new_state, &pending_state);
+        ELOG_DEBUG("State change from %d to %d, play:%d \n",old_state, new_state, GST_STATE_PAUSED);
+        break;
+    }
+    default:
+        /* unhandled message */
+        break;
+    }
+    return true;
+}
 
 void VideoGstAnalyzer::clearPipeline()
-    {
-        if (pipeline != nullptr){
-            gst_element_set_state(pipeline, GST_STATE_NULL);
-            gst_object_unref(GST_OBJECT(pipeline));
-            g_source_remove(m_bus_watch_id);
-            g_main_loop_unref(loop);
-            gst_object_unref(m_bus);
-        }
- 
+{
+    if (pipeline != nullptr){
+        gst_element_set_state(pipeline, GST_STATE_NULL);
+        gst_object_unref(GST_OBJECT(pipeline));
+        g_source_remove(m_bus_watch_id);
+        g_main_loop_unref(loop);
+        gst_object_unref(m_bus);
     }
 
-int VideoGstAnalyzer::createPipeline() {
+}
+
+void VideoGstAnalyzer::destroyPipeline()
+{
+    ELOG_DEBUG("Closed all media in this Analyzer");
+    setState(GST_STATE_NULL);
+    if (pipeline_ != nullptr && pipelineHandle != nullptr) {
+         destroyPlugin(pipeline_);
+         dlclose(pipelineHandle);
+         stopLoop();
+    }
+}
 
+int VideoGstAnalyzer::createPipeline()
+{
     pipelineHandle = dlopen(libraryName.c_str(), RTLD_LAZY);
     if (pipelineHandle == nullptr) {
         ELOG_ERROR_T("Failed to open the plugin.(%s)", libraryName.c_str());
@@ -263,7 +290,8 @@ void VideoGstAnalyzer::new_sample_from_sink (GstElement * source, gpointer data)
     gst_sample_unref(sample);
 }
 
-int VideoGstAnalyzer::addElementMany() {
+int VideoGstAnalyzer::addElementMany()
+{
     if(pipeline_){
         rvaStatus status = pipeline_->LinkElements();
         if(status != RVA_ERR_OK) {
@@ -290,20 +318,22 @@ int VideoGstAnalyzer::addElementMany() {
 }
 
 
-void VideoGstAnalyzer::stopLoop(){
+void VideoGstAnalyzer::stopLoop()
+{
     if(loop){
         ELOG_DEBUG("main loop quit\n");
         g_main_loop_quit(loop);
     }
-    g_thread_join(m_thread);
 }
 
-void VideoGstAnalyzer::main_loop_thread(gpointer data){
+void VideoGstAnalyzer::main_loop_thread(gpointer data)
+{
     g_main_loop_run(loop);
     g_thread_exit(0);
 }
 
-void VideoGstAnalyzer::setState(GstState newstate) {
+void VideoGstAnalyzer::setState(GstState newstate)
+{
     ret = gst_element_set_state(pipeline, newstate);
     if (ret == GST_STATE_CHANGE_FAILURE) {
         ELOG_ERROR("Unable to set the pipeline to the PLAYING state.\n");
@@ -312,7 +342,8 @@ void VideoGstAnalyzer::setState(GstState newstate) {
 } 
 
 
-int VideoGstAnalyzer::setPlaying() {
+int VideoGstAnalyzer::setPlaying()
+{
 
     setState(GST_STATE_PLAYING);
 
@@ -321,12 +352,14 @@ int VideoGstAnalyzer::setPlaying() {
     return 0;
 }
 
-void VideoGstAnalyzer::emitListenTo(int minPort, int maxPort) {
+void VideoGstAnalyzer::emitListenTo(int minPort, int maxPort)
+{
     ELOG_DEBUG("Listening\n");
     m_internalin.reset(new GstInternalIn((GstAppSrc*)source, minPort, maxPort));  
 }
 
-void VideoGstAnalyzer::addOutput(int connectionID, owt_base::FrameDestination* out) {
+void VideoGstAnalyzer::addOutput(int connectionID, owt_base::FrameDestination* out)
+{
     ELOG_DEBUG("Add analyzed stream back to OWT\n");
     if (sink != nullptr){
 
@@ -338,8 +371,7 @@ void VideoGstAnalyzer::addOutput(int connectionID, owt_base::FrameDestination* o
             }
             m_gstinternalout->setPad(encoder_pad);
         }
-        //gst_pad_send_event(encoder_pad, gst_event_new_custom( GST_EVENT_CUSTOM_UPSTREAM, gst_structure_new( "GstForceKeyUnit", "all-headers", G_TYPE_BOOLEAN, TRUE, NULL)));
-        //m_internalout.push_back(out);
+
         m_gstinternalout->addVideoDestination(out);
         if(!addlistener) {
             g_object_set (G_OBJECT (sink), "emit-signals", TRUE, "sync", FALSE, NULL);
@@ -352,22 +384,23 @@ void VideoGstAnalyzer::addOutput(int connectionID, owt_base::FrameDestination* o
 
 }
 
-void VideoGstAnalyzer::disconnect(owt_base::FrameDestination* out){
+void VideoGstAnalyzer::disconnect(owt_base::FrameDestination* out)
+{
     ELOG_DEBUG("Disconnect remote connection\n");
-    //m_internalout.remove(out);
     m_gstinternalout->removeVideoDestination(out);
 }
 
-int VideoGstAnalyzer::getListeningPort() {
+int VideoGstAnalyzer::getListeningPort()
+{
     int listeningPort; 
     listeningPort = m_internalin->getListeningPort();
     ELOG_DEBUG(">>>>>Listen port is :%d\n", listeningPort);
     return listeningPort; 
 }
 
 void VideoGstAnalyzer::setOutputParam(std::string codec, int width, int height, 
-    int framerate, int bitrate, int kfi, std::string algo, std::string libraryName){
-
+    int framerate, int bitrate, int kfi, std::string algo, std::string libraryName)
+{
     this->codec = codec;
     this->width = width;
     this->height = height;

source/agent/analytics/videoGstPipeline/VideoGstAnalyzer.h

@@ -17,16 +17,18 @@
 #include "AnalyticsPipeline.h"
 #include <stdio.h>
 #include "MediaFramePipeline.h"
+#include <EventRegistry.h>
 
 namespace mcu {
 
-class VideoGstAnalyzer{
+class VideoGstAnalyzer : public EventRegistry {
     DECLARE_LOGGER();
 public:
-    VideoGstAnalyzer();
-    virtual ~VideoGstAnalyzer();
+    VideoGstAnalyzer(EventRegistry* handle);
+    ~VideoGstAnalyzer();
     int createPipeline();
     void clearPipeline();
+    void destroyPipeline();
     int getListeningPort();
     void emitListenTo(int minPort,int maxPort);
     int setPlaying();
@@ -55,6 +57,10 @@ class VideoGstAnalyzer{
     static GMainLoop *loop;
     static gboolean StreamEventCallBack(GstBus *bus, GstMessage *message, gpointer data);
     void setState(GstState newstate);
+    // EventRegistry
+    bool notifyAsyncEvent(const std::string& event, const std::string& data) override;
+    bool notifyAsyncEventInEmergency(const std::string& event, const std::string& data) override;
+
     boost::scoped_ptr<GstInternalIn> m_internalin;
     boost::scoped_ptr<GstInternalOut> m_gstinternalout;
     //std::list<owt_base::InternalOut*> m_internalout;
@@ -67,6 +73,7 @@ class VideoGstAnalyzer{
     rvaPipeline* pipeline_;
     rva_create_t* createPlugin;
     rva_destroy_t* destroyPlugin;
+    EventRegistry *m_asyncHandle;
 
     GstStateChangeReturn ret;