Create message handling

CMakeLists.txt

@@ -41,9 +41,12 @@ else()
 endif()
 
 include_directories(/usr/lib/x86_64-linux-gnu/glib-2.0/include)
+include_directories(/opt/nvidia/deepstream/deepstream-7.1/sources/includes)
 include_directories(/usr/include/gstreamer-1.0)
 include_directories(/usr/include/nlohmann)
 include_directories(/usr/local/cuda/include)
+link_directories(/opt/nvidia/deepstream/deepstream/lib/)
+link_directories(/opt/nvidia/deepstream/deepstream/lib/gst-plugins)
 link_directories(/usr/local/cuda/lib64/)
 link_directories(/usr/local/cuda/targets/x86_64-linux/lib/)
 include_directories(${PROJECT_SOURCE_DIR}/camera_manager.hpp)
@@ -57,12 +60,12 @@ include_directories(${PROJECT_SOURCE_DIR}/nv_osd_manager.hpp)
 include_directories(${PROJECT_SOURCE_DIR}/queue_manager.hpp)
 include_directories(${PROJECT_SOURCE_DIR}/nv_ds_logger_manager.hpp)
 include_directories(${PROJECT_SOURCE_DIR}/sink_manager.hpp)
-
+include_directories(${PROJECT_SOURCE_DIR}/message_handling.hpp)
 
 set(SOURCES src/main.cpp  src/camera_manager.cpp src/pipeline_manager.cpp src/streammux_manager.cpp 
             src/source_bin.cpp src/gstds_example_manager.cpp src/tiler_manager.cpp
             src/nv_video_convert_manager.cpp src/nv_osd_manager.cpp src/queue_manager.cpp 
-            src/nv_ds_logger_manager.cpp src/sink_manager.cpp)
+            src/nv_ds_logger_manager.cpp src/sink_manager.cpp src/message_handling.cpp)
 # Create the executable
 add_executable(${PROJECT_NAME}  ${SOURCES})
 
@@ -86,4 +89,6 @@ target_include_directories(${PROJECT_NAME} PRIVATE ${GLIB_INCLUDE_DIRS})
 target_link_libraries(${PROJECT_NAME} ${GLIB_LIBRARIES})
 
 target_link_libraries(${PROJECT_NAME} gstbase-1.0 gstreamer-1.0 gstrtp-1.0 gstvideo-1.0)
-target_link_libraries(${PROJECT_NAME} cudart cuda)
+target_link_libraries(${PROJECT_NAME} cudart cuda)
+target_link_libraries(${PROJECT_NAME} nvdsgst_infer nvds_meta nvds_inferutils
+    nvds_utils nvdsgst_helper)#nvdsgst_metnvdsa

src/message_handling.cpp

@@ -0,0 +1,104 @@
+#include "message_handling.hpp"
+
+int MessageHandling::counter_total = 0;
+int MessageHandling::counter_eos = 0;
+int MessageHandling::counter_warning = 0;
+int MessageHandling::counter_error = 0;
+int MessageHandling::counter_element = 0;
+bool MessageHandling::pipeline_is_run = false;
+
+MessageHandling::MessageHandling() {}
+
+// Definition of static function
+gboolean MessageHandling::bus_call(GstBus *bus, GstMessage *msg,
+                                   gpointer user_data) {
+    (void)bus;  // This explicitly marks it as unused
+    StreamData *data = static_cast<StreamData *>(user_data);
+    GMainLoop *loop = data->loop;
+    // GMainLoop *loop= (GMainLoop *) data;
+    g_print("Bus_call \n");
+    gchar *debug;
+    GError *error;
+    counter_total++;
+    switch (GST_MESSAGE_TYPE(msg)) {
+        case GST_MESSAGE_EOS:
+            counter_eos++;
+            g_print("GST_MESSAGE_EOS \n");
+            if (data->g_run_forever == FALSE) {
+                g_print("End of stream  \n");
+                pipeline_is_run = false;
+                g_main_loop_quit(loop);
+            }
+            break;
+        case GST_MESSAGE_WARNING:
+            counter_warning++;
+            g_print("GST_MESSAGE_WARNING \n");
+            gst_message_parse_warning(msg, &error, &debug);
+            g_printerr("WARNING from element %s: %s\n",
+                       GST_OBJECT_NAME(msg->src), error->message);
+            g_free(debug);
+            g_printerr("Warning: %s\n", error->message);
+            g_error_free(error);
+            break;
+        case GST_MESSAGE_ERROR:
+            counter_error++;
+            g_print("GST_MESSAGE_ERROR \n");
+            gst_message_parse_error(msg, &error, &debug);
+            g_printerr("ERROR from element %s: %s\n", GST_OBJECT_NAME(msg->src),
+                       error->message);
+            if (debug) g_printerr("Error details: %s\n", debug);
+            g_free(debug);
+            g_error_free(error);
+            g_main_loop_quit(loop);
+            break;
+        case GST_MESSAGE_ELEMENT:
+            counter_element++;
+            g_print("GST_MESSAGE_ELEMENT \n");
+            if (gst_nvmessage_is_stream_eos(msg)) {
+                guint stream_id;
+                if (gst_nvmessage_parse_stream_eos(msg, &stream_id)) {
+                    //                g_print ("Got EOS from stream %d\n",
+                    //                stream_id);
+                    g_print("Got EOS from stream %d \n", stream_id);
+                    // g_mutex_lock (&eos_lock);
+                    // g_eos_list[stream_id] = TRUE;
+                    // g_mutex_unlock (&eos_lock);
+                    //                g_timeout_add_seconds (10, add_sources,
+                    //                (gpointer) g_source_bin_list);
+                    //                add_sources((gpointer) g_source_bin_list);
+                    // g_print ("camera_list.at(%d).connection_status is "
+                    //          "%d \n",
+                    //          stream_id);
+                }
+            }
+            break;
+        default:
+            // g_print ("GST_MESSAGE_TYPE (msg) is %d  ", GST_MESSAGE_TYPE
+            // (msg));
+            g_message("Received message of type: %s",
+                      gst_message_type_get_name(GST_MESSAGE_TYPE(msg)));
+
+            // g_print (gst_message_type_get_name(GST_MESSAGE_TYPE(msg)));
+            g_print("%s\n", gst_message_type_get_name(GST_MESSAGE_TYPE(msg)));
+
+            // g_print (GST_MESSAGE_TYPE (msg));
+
+            break;
+    }
+    g_print(
+        "counter_eos is %d, counter_warning is %d, counter_error is %d, "
+        "counter_element is %d, counter_total is %d \n",
+        counter_eos, counter_warning, counter_error, counter_element,
+        counter_total);
+    return TRUE;
+}
+
+void MessageHandling::create_message_handler(GstElement *pipeline,
+                                             bool g_run_forever) {
+    /* add a message handler */
+    bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
+    StreamData *stream_data = new StreamData{g_run_forever, loop};
+    bus_watch_id = gst_bus_add_watch(bus, bus_call, stream_data);  // loop
+
+    gst_object_unref(bus);
+}

src/message_handling.hpp

@@ -0,0 +1,30 @@
+#include <glib.h>
+#include <gst-nvmessage.h>
+#include <gst/gst.h>
+
+#include <fstream>
+#include <iostream>
+
+class MessageHandling {
+   private:
+    typedef struct {
+        bool g_run_forever;
+        GMainLoop *loop;
+    } StreamData;
+
+   public:
+    static gboolean bus_call(GstBus *, GstMessage *, gpointer);
+    void create_message_handler(GstElement *, bool);
+    GstBus *bus = NULL;
+    GMainLoop *loop = NULL;
+    static int counter_total;
+    static int counter_eos;
+    static bool pipeline_is_run;
+    static int counter_warning;
+    static GMutex eos_lock;
+    static int counter_error;
+    static int counter_element;
+    guint bus_watch_id;
+    MessageHandling();
+    ~MessageHandling();
+};

src/pipeline_manager.cpp

@@ -91,5 +91,7 @@ bool PipelineManager::create_pipeline_elements(int num_sources,
     nv_ds_logger_manager->create_nv_ds_logger();
     sink_manager->create_sink(prop);
 
+    message_handling->create_message_handler(pipeline, g_run_forever);
+
     return true;
 }

src/pipeline_manager.hpp

@@ -3,6 +3,7 @@
 
 #include "cuda_runtime_api.h"
 #include "gstds_example_manager.hpp"
+#include "message_handling.hpp"
 #include "nv_ds_logger_manager.hpp"
 #include "nv_osd_manager.hpp"
 #include "nv_video_convert_manager.hpp"
@@ -26,6 +27,7 @@ class PipelineManager {
     NvOsdManager *nv_osd_manager = new NvOsdManager();
     NvDsLoggerManager *nv_ds_logger_manager = new NvDsLoggerManager();
     SinkManager *sink_manager = new SinkManager();
+    MessageHandling *message_handling = new MessageHandling();
 
    public:
     int current_device = -1;