Not working the linking of appsrc

Cargo.lock

@@ -743,6 +743,7 @@ name = "webvtt"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "glib",
  "gst-plugin-closedcaption",
  "gstreamer",
  "gstreamer-app",

Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 rust-version = "1.56"
 
 [dependencies]
+glib = { git = "https://github.com/gtk-rs/gtk-rs-core" }
 gst = { package = "gstreamer", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs", features = ["v1_18"] }
 gst-app = { package = "gstreamer-app", git = "https://gitlab.freedesktop.org/gstreamer/gstreamer-rs" }
 gst-plugin-closedcaption = { path = "../../gst-plugins-rs/video/closedcaption" }  # git = "https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs" }

src/main.rs

@@ -9,30 +9,81 @@ fn main() -> Result<(), Error>  {
     // TODO: create a video source with X buffers and burn the captions in the video feed using the captions tooling
 
     let pipeline = gst::parse_launch(
-        "videotestsrc num-buffers=1800 ! video/x-raw,width=1280,height=720,framerate=30/1 ! glimagesink"
-        // "cccombiner name=ccc ! cea608overlay ! autovideosink \
-        //  videotestsrc num-buffers=1800 ! video/x-raw,width=1280,height=720,framerate=30/1 ! queue ! ccc.sink \
-        //  appsrc name=text-src ! tttocea608 ! queue ! ccc.caption",
+        "cccombiner name=ccc ! cea608overlay ! autovideosink \
+         videotestsrc num-buffers=1800 ! video/x-raw,width=1280,height=720,framerate=30/1 ! queue ! ccc.sink \
+         tttocea608 name=converter ! queue ! ccc.caption",
     )?
         .downcast::<gst::Pipeline>()
         .expect("Expected a gst::Pipeline");
 
-    // let text_src = pipeline.by_name("text-src")
-    //     .unwrap()
-    //     .downcast::<gst_app::AppSrc>()
-    //     .unwrap();
-    // text_src.set_is_live(true);
-    // text_src.set_caps(Some(
-    //     &gst::Caps::builder("text/x-raw")
-    //     .field("format", "utf8")
-    //     .build(),
-    // ));
+    let text_src = gst::ElementFactory::make("appsrc", Some("text-src")).unwrap();
+    text_src.set_property("is-live", true);
+    text_src.set_property("caps",
+        &gst::Caps::builder("text/x-raw")
+        .field("format", "utf8")
+        .build(),
+    );
+    pipeline.add(&text_src).expect("Failed to add elements to pipeline");
+
+    let converter = pipeline.by_name("converter").unwrap();
+    text_src.link_pads(Some("src"), &converter, Some("sink")).expect("Failed to link elements");
 
     println!("Starting pipeline..");
 
     main_loop(pipeline)
 }
 
+fn main_loop(pipeline: gst::Pipeline) -> Result<(), Error> {
+    let main_loop = glib::MainLoop::new(None, false);
+
+    pipeline.set_state(gst::State::Playing)?;
+
+    let bus = pipeline.bus().unwrap();
+    bus.add_watch({
+        let main_loop = main_loop.clone();
+        move |_, msg| {
+            use gst::MessageView;
+
+            let main_loop = &main_loop;
+            match msg.view() {
+                MessageView::Eos(..) => main_loop.quit(),
+                MessageView::Error(err) => {
+                    println!(
+                        "Error from {:?}: {} ({:?})",
+                        err.src().map(|s| s.path_string()),
+                        err.error(),
+                        err.debug()
+                    );
+                    main_loop.quit();
+                }
+                _ => (),
+            };
+
+            glib::Continue(true)
+        }
+    })
+        .expect("Failed to add bus watch");
+
+    let pipeline_weak = pipeline.downgrade();
+    let timeout_id = glib::timeout_add_local(std::time::Duration::from_secs(10), move || {
+        let pipeline = match pipeline_weak.upgrade() {
+            Some(pipeline) => pipeline,
+            None => return glib::Continue(true),
+        };
+        push_buffer(&pipeline);
+        glib::Continue(true)
+    });
+
+    main_loop.run();
+
+    pipeline.set_state(gst::State::Null)?;
+
+    timeout_id.remove();
+    bus.remove_watch().unwrap();
+
+    Ok(())
+}
+
 fn push_buffer(pipeline: &gst::Pipeline) {
     let src = pipeline
         .by_name("text-src")
@@ -40,44 +91,16 @@ fn push_buffer(pipeline: &gst::Pipeline) {
         .downcast::<gst_app::AppSrc>()
         .unwrap();
 
-    let timestamp = pipeline.current_clock_time().unwrap() + gst::ClockTime::from_seconds(1);
+    let timestamp = pipeline.query_position::<gst::ClockTime>().unwrap();
+
+    println!("Trying to publish text buffer NOW! - {}", timestamp.display());
 
     let text = "Nice story, bro!1".to_string();
-    let mut buffer = gst::Buffer::from_slice(text);
+    let mut buffer = gst::Buffer::from_mut_slice(text.into_bytes());
     {
         let buffer = buffer.get_mut().unwrap();
         buffer.set_pts(timestamp);
         buffer.set_duration(gst::ClockTime::from_seconds(10));
-        buffer.set_flags(gst::BufferFlags::DELTA_UNIT);
     }
     src.push_buffer(buffer).unwrap();
-}
-
-fn main_loop(pipeline: gst::Pipeline) -> Result<(), Error> {
-    // let main_loop = glib::MainLoop::new(None, false);
-
-    pipeline.set_state(gst::State::Playing)?;
-
-    let bus = pipeline.bus().unwrap();
-    for msg in bus.iter_timed(gst::ClockTime::NONE) {
-        use gst::MessageView;
-
-        match msg.view() {
-            MessageView::Eos(..) => break,
-            MessageView::Error(err) => {
-                println!(
-                    "Error from {:?}: {} ({:?})",
-                    msg.src().map(|s| s.path_string()),
-                    err.error(),
-                    err.debug()
-                );
-                break;
-            }
-            _ => (),
-        }
-    }
-
-    pipeline.set_state(gst::State::Null)?;
-
-    Ok(())
 }