From d890aa00113d10a41b09ddcfdc14518b82a1fa44 Mon Sep 17 00:00:00 2001
From: Wanjohi <71614375+wanjohiryan@users.noreply.github.com>
Date: Fri, 15 Dec 2023 16:40:41 +0300
Subject: [PATCH] Add async

---
 gst-warp-sink/src/lib.rs         | 16 ++++++++++++--
 gst-warp-sink/src/moqsink/imp.rs | 36 +++++++++++++++++++++++++++++++-
 2 files changed, 49 insertions(+), 3 deletions(-)

diff --git a/gst-warp-sink/src/lib.rs b/gst-warp-sink/src/lib.rs
index 4eb4a7e..57514ef 100644
--- a/gst-warp-sink/src/lib.rs
+++ b/gst-warp-sink/src/lib.rs
@@ -1,6 +1,9 @@
+use gst::glib;
+use gst::glib::once_cell::sync::Lazy;
+use tokio::runtime;
+
 mod moqsink;
 mod relayurl;
-use gst::glib;
 
 fn plugin_init(plugin: &gst::Plugin) -> Result<(), glib::BoolError> {
     moqsink::register(plugin)?;
@@ -17,4 +20,13 @@ gst::plugin_define!(
     env!("CARGO_PKG_NAME"),
     env!("CARGO_PKG_REPOSITORY"),
     env!("BUILD_REL_DATE")
-);
\ No newline at end of file
+);
+
+//TODO: https://github.com/sdroege/gst-plugin-rs/blob/1faef49b51746a82c5b99a09a1034220cf92af9d/net/webrtc/src/lib.rs#L44C1-L50C4
+pub static RUNTIME: Lazy<runtime::Runtime> = Lazy::new(|| {
+    runtime::Builder::new_multi_thread()
+        .enable_all()
+        .worker_threads(1)
+        .build()
+        .unwrap()
+});
\ No newline at end of file
diff --git a/gst-warp-sink/src/moqsink/imp.rs b/gst-warp-sink/src/moqsink/imp.rs
index 9610a81..3fa8f27 100644
--- a/gst-warp-sink/src/moqsink/imp.rs
+++ b/gst-warp-sink/src/moqsink/imp.rs
@@ -8,6 +8,8 @@ use std::sync::Mutex;
 
 use crate::relayurl::*;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
+use tokio::runtime::Runtime;
+use tokio::sync::mpsc;
 
 use moq_transport::cache::{broadcast, fragment, segment, track};
 use url::Url;
@@ -117,6 +119,38 @@ impl MoqSink {
             }
         };
 
+        // // Initialize shared state and channels
+        // let (sender, receiver) = mpsc::channel(32);
+        // // self.sender = Some(sender);
+
+        // // Spawn a new thread to run the Tokio runtime
+        // std::thread::spawn(move || {
+        //     // Create a new Tokio runtime
+        //     let rt = Runtime::new().unwrap();
+
+        //     // Block on the async code within the Tokio runtime
+        //     rt.block_on(async move {
+        //         // Set up your async tasks here
+        //         let session_task = session.run();
+        //         let media_task = media.run();
+
+        //         tokio::select! {
+        //             res = session_task => {
+        //                 // Handle session result
+        //                 // You might want to send a message to the GStreamer element using the channel
+        //             },
+        //             res = media_task => {
+        //                 // Handle media result
+        //                 // You might want to send a message to the GStreamer element using the channel
+        //             },
+        //             msg = receiver.recv() => {
+        //                 // Handle messages sent from the GStreamer element to the Tokio runtime
+        //                 // These might be control commands or data to process
+        //             }
+        //         }
+        //     });
+        // });
+
         Ok(())
     }
 
@@ -305,4 +339,4 @@ impl BaseSinkImpl for MoqSink {
     fn start(&self) -> Result<(), gst::ErrorMessage> {
         self.start()
     }
-}
+}
\ No newline at end of file