From d39ed7cb02ba9cc120ba231cce3881e64de85daf Mon Sep 17 00:00:00 2001
From: Barzan Hayati <barzanhayati@gmail.com>
Date: Tue, 1 Jul 2025 16:22:51 +0000
Subject: [PATCH] Run pipeline

---
 data/addresses.txt       |  4 ++--
 data/configuration.json  |  2 +-
 src/message_handling.cpp | 11 +++++++----
 src/message_handling.hpp |  4 ++--
 src/pipeline_manager.cpp | 20 +++++++++++++++++++-
 5 files changed, 31 insertions(+), 10 deletions(-)

diff --git a/data/addresses.txt b/data/addresses.txt
index 01ffec9..52c5743 100644
--- a/data/addresses.txt
+++ b/data/addresses.txt
@@ -1,2 +1,2 @@
-/opt/nvidia/deepstream/deepstream-7.1/samples/streams/yoga.mp4
-/opt/nvidia/deepstream/deepstream-7.1/samples/streams/sample_720p.mp4
+file:///opt/nvidia/deepstream/deepstream-7.1/samples/streams/sample_720p.mp4
+file:///opt/nvidia/deepstream/deepstream-7.1/samples/streams/sample_720p.mp4
diff --git a/data/configuration.json b/data/configuration.json
index 5b1a7c6..8d40e62 100644
--- a/data/configuration.json
+++ b/data/configuration.json
@@ -2,7 +2,7 @@
     "MUXER_OUTPUT_HEIGHT": 1080,
     "MUXER_OUTPUT_WIDTH": 1920,
     "output_video_path": "test.mkv",
-    "display_output": 1,
+    "display_output": 2,
     "codec_rtsp_out": "H264",
     "mount_address": "/rtsp-output",
     "udp_buffer_size": 524288,
diff --git a/src/message_handling.cpp b/src/message_handling.cpp
index 2596ca6..79d3029 100644
--- a/src/message_handling.cpp
+++ b/src/message_handling.cpp
@@ -9,12 +9,14 @@ bool MessageHandling::pipeline_is_run = false;
 
 MessageHandling::MessageHandling() {}
 
+void MessageHandling::source_remove() { g_source_remove(bus_watch_id); }
+
 // 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 *passed_loop = data->loop;
     // GMainLoop *loop= (GMainLoop *) data;
     g_print("Bus_call \n");
     gchar *debug;
@@ -27,7 +29,7 @@ gboolean MessageHandling::bus_call(GstBus *bus, GstMessage *msg,
             if (data->g_run_forever == FALSE) {
                 g_print("End of stream  \n");
                 pipeline_is_run = false;
-                g_main_loop_quit(loop);
+                g_main_loop_quit(passed_loop);
             }
             break;
         case GST_MESSAGE_WARNING:
@@ -49,7 +51,7 @@ gboolean MessageHandling::bus_call(GstBus *bus, GstMessage *msg,
             if (debug) g_printerr("Error details: %s\n", debug);
             g_free(debug);
             g_error_free(error);
-            g_main_loop_quit(loop);
+            g_main_loop_quit(passed_loop);
             break;
         case GST_MESSAGE_ELEMENT:
             counter_element++;
@@ -94,7 +96,8 @@ gboolean MessageHandling::bus_call(GstBus *bus, GstMessage *msg,
 }
 
 void MessageHandling::create_message_handler(GstElement *pipeline,
-                                             bool g_run_forever) {
+                                             bool g_run_forever,
+                                             GMainLoop *loop) {
     /* add a message handler */
     bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
     StreamData *stream_data = new StreamData{g_run_forever, loop};
diff --git a/src/message_handling.hpp b/src/message_handling.hpp
index 00039c9..3f14a51 100644
--- a/src/message_handling.hpp
+++ b/src/message_handling.hpp
@@ -14,9 +14,8 @@ class MessageHandling {
 
    public:
     static gboolean bus_call(GstBus *, GstMessage *, gpointer);
-    void create_message_handler(GstElement *, bool);
+    void create_message_handler(GstElement *, bool, GMainLoop *);
     GstBus *bus = NULL;
-    GMainLoop *loop = NULL;
     static int counter_total;
     static int counter_eos;
     static bool pipeline_is_run;
@@ -26,5 +25,6 @@ class MessageHandling {
     static int counter_element;
     guint bus_watch_id;
     MessageHandling();
+    void source_remove();
     ~MessageHandling();
 };
\ No newline at end of file
diff --git a/src/pipeline_manager.cpp b/src/pipeline_manager.cpp
index 0f35368..d0e51e9 100644
--- a/src/pipeline_manager.cpp
+++ b/src/pipeline_manager.cpp
@@ -179,7 +179,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);
+    message_handling->create_message_handler(pipeline, g_run_forever, loop);
     setup_pipeline();
 
     playing_pipeline(num_sources, url_camera);
@@ -192,5 +192,23 @@ bool PipelineManager::create_pipeline_elements(int num_sources,
     DataPointer* pointer_data = new DataPointer{tiler_manager};
     g_timeout_add(40, event_thread_func, pointer_data);  // NULL
 
+    message_handling->pipeline_is_run = true;
+
+    g_main_loop_run(loop);
+
+    /* Out of the main loop, clean up nicely */
+    g_print("Returned, stopping playback \n");
+    gst_element_set_state(pipeline, GST_STATE_NULL);
+    g_print("Deleting pipeline \n");
+    gst_object_unref(GST_OBJECT(pipeline));
+    // g_source_remove (bus_watch_id);
+    message_handling->source_remove();
+    g_main_loop_unref(loop);
+    gst_deinit();
+    // g_free (g_source_bin_list);
+    //    g_free (uri);
+    g_mutex_clear(&eos_lock);
+    rtsp_streaming_manager->destroy_sink_bin();
+
     return true;
 }
\ No newline at end of file