threadshare: Use appsink callbacks instead of signals in the tests

5 files changed, 143 insertions, 108 deletions

diff --git a/generic/threadshare/tests/jitterbuffer.rs b/generic/threadshare/tests/jitterbuffer.rs
index 9c3ca9f84..3c77f1331 100644
--- a/generic/threadshare/tests/jitterbuffer.rs
+++ b/generic/threadshare/tests/jitterbuffer.rs
@@ -77,17 +77,21 @@ fn jb_pipeline() {
 
     let appsink = sink.dynamic_cast::<gst_app::AppSink>().unwrap();
     let (sender, receiver) = mpsc::channel();
-    appsink.connect_new_sample(move |appsink| {
-        let _sample = appsink
-            .emit_by_name("pull-sample", &[])
-            .unwrap()
-            .unwrap()
-            .get::<gst::Sample>()
-            .unwrap();
-
-        sender.send(()).unwrap();
-        Ok(gst::FlowSuccess::Ok)
-    });
+    appsink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample(move |appsink| {
+                let _sample = appsink
+                    .emit_by_name("pull-sample", &[])
+                    .unwrap()
+                    .unwrap()
+                    .get::<gst::Sample>()
+                    .unwrap();
+
+                sender.send(()).unwrap();
+                Ok(gst::FlowSuccess::Ok)
+            })
+            .build(),
+    );
 
     pipeline.set_state(gst::State::Playing).unwrap();
 
@@ -143,17 +147,21 @@ fn jb_ts_pipeline() {
 
     let appsink = sink.dynamic_cast::<gst_app::AppSink>().unwrap();
     let (sender, receiver) = mpsc::channel();
-    appsink.connect_new_sample(move |appsink| {
-        let _sample = appsink
-            .emit_by_name("pull-sample", &[])
-            .unwrap()
-            .unwrap()
-            .get::<gst::Sample>()
-            .unwrap();
-
-        sender.send(()).unwrap();
-        Ok(gst::FlowSuccess::Ok)
-    });
+    appsink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample(move |appsink| {
+                let _sample = appsink
+                    .emit_by_name("pull-sample", &[])
+                    .unwrap()
+                    .unwrap()
+                    .get::<gst::Sample>()
+                    .unwrap();
+
+                sender.send(()).unwrap();
+                Ok(gst::FlowSuccess::Ok)
+            })
+            .build(),
+    );
 
     pipeline.set_state(gst::State::Playing).unwrap();
 
diff --git a/generic/threadshare/tests/pipeline.rs b/generic/threadshare/tests/pipeline.rs
index b1c568f11..30b91520a 100644
--- a/generic/threadshare/tests/pipeline.rs
+++ b/generic/threadshare/tests/pipeline.rs
@@ -86,17 +86,21 @@ fn multiple_contexts_queue() {
 
         let appsink = sink.dynamic_cast::<gst_app::AppSink>().unwrap();
         let sender_clone = sender.clone();
-        appsink.connect_new_sample(move |appsink| {
-            let _sample = appsink
-                .emit_by_name("pull-sample", &[])
-                .unwrap()
-                .unwrap()
-                .get::<gst::Sample>()
-                .unwrap();
-
-            sender_clone.send(()).unwrap();
-            Ok(gst::FlowSuccess::Ok)
-        });
+        appsink.set_callbacks(
+            gst_app::AppSinkCallbacks::builder()
+                .new_sample(move |appsink| {
+                    let _sample = appsink
+                        .emit_by_name("pull-sample", &[])
+                        .unwrap()
+                        .unwrap()
+                        .get::<gst::Sample>()
+                        .unwrap();
+
+                    sender_clone.send(()).unwrap();
+                    Ok(gst::FlowSuccess::Ok)
+                })
+                .build(),
+        );
     }
 
     let pipeline_clone = pipeline.clone();
@@ -244,17 +248,21 @@ fn multiple_contexts_proxy() {
 
         let appsink = sink.dynamic_cast::<gst_app::AppSink>().unwrap();
         let sender_clone = sender.clone();
-        appsink.connect_new_sample(move |appsink| {
-            let _sample = appsink
-                .emit_by_name("pull-sample", &[])
-                .unwrap()
-                .unwrap()
-                .get::<gst::Sample>()
-                .unwrap();
-
-            sender_clone.send(()).unwrap();
-            Ok(gst::FlowSuccess::Ok)
-        });
+        appsink.set_callbacks(
+            gst_app::AppSinkCallbacks::builder()
+                .new_sample(move |appsink| {
+                    let _sample = appsink
+                        .emit_by_name("pull-sample", &[])
+                        .unwrap()
+                        .unwrap()
+                        .get::<gst::Sample>()
+                        .unwrap();
+
+                    sender_clone.send(()).unwrap();
+                    Ok(gst::FlowSuccess::Ok)
+                })
+                .build(),
+        );
     }
 
     let pipeline_clone = pipeline.clone();
@@ -361,21 +369,24 @@ fn eos() {
     let (sample_notifier, sample_notif_rcv) = mpsc::channel();
     let (eos_notifier, eos_notif_rcv) = mpsc::channel();
     let appsink = appsink.dynamic_cast::<gst_app::AppSink>().unwrap();
-    appsink.connect_new_sample(move |appsink| {
-        gst_debug!(CAT, obj: appsink, "eos: pulling sample");
-        let _ = appsink
-            .emit_by_name("pull-sample", &[])
-            .unwrap()
-            .unwrap()
-            .get::<gst::Sample>()
-            .unwrap();
-
-        sample_notifier.send(()).unwrap();
-
-        Ok(gst::FlowSuccess::Ok)
-    });
-
-    appsink.connect_eos(move |_appsink| eos_notifier.send(()).unwrap());
+    appsink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample(move |appsink| {
+                gst_debug!(CAT, obj: appsink, "eos: pulling sample");
+                let _ = appsink
+                    .emit_by_name("pull-sample", &[])
+                    .unwrap()
+                    .unwrap()
+                    .get::<gst::Sample>()
+                    .unwrap();
+
+                sample_notifier.send(()).unwrap();
+
+                Ok(gst::FlowSuccess::Ok)
+            })
+            .eos(move |_appsink| eos_notifier.send(()).unwrap())
+            .build(),
+    );
 
     fn push_buffer(src: &gst::Element) -> bool {
         gst_debug!(CAT, obj: src, "eos: pushing buffer");
@@ -504,19 +515,23 @@ fn premature_shutdown() {
     let (appsink_sender, appsink_receiver) = mpsc::channel();
 
     let appsink = appsink.dynamic_cast::<gst_app::AppSink>().unwrap();
-    appsink.connect_new_sample(move |appsink| {
-        gst_debug!(CAT, obj: appsink, "premature_shutdown: pulling sample");
-        let _sample = appsink
-            .emit_by_name("pull-sample", &[])
-            .unwrap()
-            .unwrap()
-            .get::<gst::Sample>()
-            .unwrap();
-
-        appsink_sender.send(()).unwrap();
-
-        Ok(gst::FlowSuccess::Ok)
-    });
+    appsink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample(move |appsink| {
+                gst_debug!(CAT, obj: appsink, "premature_shutdown: pulling sample");
+                let _sample = appsink
+                    .emit_by_name("pull-sample", &[])
+                    .unwrap()
+                    .unwrap()
+                    .get::<gst::Sample>()
+                    .unwrap();
+
+                appsink_sender.send(()).unwrap();
+
+                Ok(gst::FlowSuccess::Ok)
+            })
+            .build(),
+    );
 
     fn push_buffer(src: &gst::Element, intent: &str) -> bool {
         gst_debug!(
diff --git a/generic/threadshare/tests/proxy.rs b/generic/threadshare/tests/proxy.rs
index 68577f7d9..319e812ee 100644
--- a/generic/threadshare/tests/proxy.rs
+++ b/generic/threadshare/tests/proxy.rs
@@ -56,18 +56,22 @@ fn test_push() {
 
     let appsink = appsink.dynamic_cast::<gst_app::AppSink>().unwrap();
     let samples_clone = samples.clone();
-    appsink.connect_new_sample(move |appsink| {
-        let sample = appsink
-            .emit_by_name("pull-sample", &[])
-            .unwrap()
-            .unwrap()
-            .get::<gst::Sample>()
-            .unwrap();
-
-        samples_clone.lock().unwrap().push(sample);
-
-        Ok(gst::FlowSuccess::Ok)
-    });
+    appsink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample(move |appsink| {
+                let sample = appsink
+                    .emit_by_name("pull-sample", &[])
+                    .unwrap()
+                    .unwrap()
+                    .get::<gst::Sample>()
+                    .unwrap();
+
+                samples_clone.lock().unwrap().push(sample);
+
+                Ok(gst::FlowSuccess::Ok)
+            })
+            .build(),
+    );
 
     pipeline.set_state(gst::State::Playing).unwrap();
 
diff --git a/generic/threadshare/tests/queue.rs b/generic/threadshare/tests/queue.rs
index e217c1cb4..bf0ce17bf 100644
--- a/generic/threadshare/tests/queue.rs
+++ b/generic/threadshare/tests/queue.rs
@@ -51,18 +51,22 @@ fn test_push() {
 
     let appsink = appsink.dynamic_cast::<gst_app::AppSink>().unwrap();
     let samples_clone = samples.clone();
-    appsink.connect_new_sample(move |appsink| {
-        let sample = appsink
-            .emit_by_name("pull-sample", &[])
-            .unwrap()
-            .unwrap()
-            .get::<gst::Sample>()
-            .unwrap();
-
-        samples_clone.lock().unwrap().push(sample);
-
-        Ok(gst::FlowSuccess::Ok)
-    });
+    appsink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample(move |appsink| {
+                let sample = appsink
+                    .emit_by_name("pull-sample", &[])
+                    .unwrap()
+                    .unwrap()
+                    .get::<gst::Sample>()
+                    .unwrap();
+
+                samples_clone.lock().unwrap().push(sample);
+
+                Ok(gst::FlowSuccess::Ok)
+            })
+            .build(),
+    );
 
     pipeline.set_state(gst::State::Playing).unwrap();
 
diff --git a/generic/threadshare/tests/tcpclientsrc.rs b/generic/threadshare/tests/tcpclientsrc.rs
index 02a03e649..55f2df1de 100644
--- a/generic/threadshare/tests/tcpclientsrc.rs
+++ b/generic/threadshare/tests/tcpclientsrc.rs
@@ -73,18 +73,22 @@ fn test_push() {
 
     let appsink = appsink.dynamic_cast::<gst_app::AppSink>().unwrap();
     let samples_clone = samples.clone();
-    appsink.connect_new_sample(move |appsink| {
-        let sample = appsink
-            .emit_by_name("pull-sample", &[])
-            .unwrap()
-            .unwrap()
-            .get::<gst::Sample>()
-            .unwrap();
-
-        let mut samples = samples_clone.lock().unwrap();
-        samples.push(sample);
-        Ok(gst::FlowSuccess::Ok)
-    });
+    appsink.set_callbacks(
+        gst_app::AppSinkCallbacks::builder()
+            .new_sample(move |appsink| {
+                let sample = appsink
+                    .emit_by_name("pull-sample", &[])
+                    .unwrap()
+                    .unwrap()
+                    .get::<gst::Sample>()
+                    .unwrap();
+
+                let mut samples = samples_clone.lock().unwrap();
+                samples.push(sample);
+                Ok(gst::FlowSuccess::Ok)
+            })
+            .build(),
+    );
 
     // Wait for the server to listen
     listening_rx.recv().unwrap();