rsinter: add 'event-types' property to intersink

The event-types property defaults to Eos.  Setting an
array of additional, serialized event types results in
calling the producer.set_forward_events with those types
so that the events will be forwarded to any consumers.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1875>

docs/plugins/gst_plugins_cache.json

@@ -7716,6 +7716,17 @@
                     }
                 },
                 "properties": {
+                    "event-types": {
+                        "blurb": "Forward Event Types (default EOS)",
+                        "conditionally-available": false,
+                        "construct": false,
+                        "construct-only": false,
+                        "controllable": false,
+                        "mutable": "ready",
+                        "readable": true,
+                        "type": "GstValueArray",
+                        "writable": true
+                    },
                     "producer-name": {
                         "blurb": "Producer Name to use",
                         "conditionally-available": false,

generic/inter/src/sink/imp.rs

@@ -15,12 +15,14 @@ const DEFAULT_PRODUCER_NAME: &str = "default";
 #[derive(Debug)]
 struct Settings {
     producer_name: String,
+    event_types: Vec<gst::EventType>,
 }
 
 impl Default for Settings {
     fn default() -> Self {
         Settings {
             producer_name: DEFAULT_PRODUCER_NAME.to_string(),
+            event_types: vec![gst::EventType::Eos],
         }
     }
 }
@@ -41,9 +43,9 @@ impl InterSink {
         let settings = self.settings.lock().unwrap();
         let state = self.state.lock().unwrap();
 
-        InterStreamProducer::acquire(&settings.producer_name, &state.appsink)?;
-
-        Ok(())
+        InterStreamProducer::acquire(&settings.producer_name, &state.appsink).map(|producer| {
+            producer.set_forward_events(settings.event_types.clone());
+        })
     }
 
     fn unprepare(&self) {
@@ -83,12 +85,28 @@ impl ObjectSubclass for InterSink {
 impl ObjectImpl for InterSink {
     fn properties() -> &'static [glib::ParamSpec] {
         static PROPERTIES: LazyLock<Vec<glib::ParamSpec>> = LazyLock::new(|| {
-            vec![glib::ParamSpecString::builder("producer-name")
-                .nick("Producer Name")
-                .blurb("Producer Name to use")
-                .doc_show_default()
-                .mutable_playing()
-                .build()]
+            vec![
+                glib::ParamSpecString::builder("producer-name")
+                    .nick("Producer Name")
+                    .blurb("Producer Name to use")
+                    .doc_show_default()
+                    .mutable_playing()
+                    .build(),
+                gst::ParamSpecArray::builder("event-types")
+                    .element_spec(
+                        &glib::ParamSpecEnum::builder_with_default(
+                            "event-type",
+                            gst::EventType::Eos,
+                        )
+                        .nick("Event Type")
+                        .blurb("Event Type")
+                        .build(),
+                    )
+                    .nick("Forwarded Event Types")
+                    .blurb("Forward Event Types (default EOS)")
+                    .mutable_ready()
+                    .build(),
+            ]
         });
 
         PROPERTIES.as_ref()
@@ -123,6 +141,16 @@ impl ObjectImpl for InterSink {
                     }
                 }
             }
+            "event-types" => {
+                let mut settings = self.settings.lock().unwrap();
+                let types = value
+                    .get::<gst::Array>()
+                    .expect("type checked upstream")
+                    .iter()
+                    .map(|v| v.get::<gst::EventType>().expect("type checked upstream"))
+                    .collect::<Vec<_>>();
+                settings.event_types = types;
+            }
             _ => unimplemented!(),
         };
     }
@@ -133,6 +161,15 @@ impl ObjectImpl for InterSink {
                 let settings = self.settings.lock().unwrap();
                 settings.producer_name.to_value()
             }
+            "event-types" => {
+                let settings = self.settings.lock().unwrap();
+                settings
+                    .event_types
+                    .iter()
+                    .map(|x| x.to_send_value())
+                    .collect::<gst::Array>()
+                    .to_value()
+            }
             _ => unimplemented!(),
         }
     }

generic/inter/tests/inter.rs

@@ -136,3 +136,51 @@ fn test_change_producer_name() {
     element1.set_state(gst::State::Null).unwrap();
     element2.set_state(gst::State::Null).unwrap();
 }
+
+#[test]
+#[serial]
+fn test_event_forwarding() {
+    init();
+
+    let mut hc = start_consumer("p");
+    let (srcpad, intersink) = start_producer("p");
+
+    intersink.set_state(gst::State::Null).unwrap();
+    intersink.set_property(
+        "event-types",
+        gst::Array::new(vec![gst::EventType::Eos, gst::EventType::CustomDownstream]),
+    );
+    intersink.set_state(gst::State::Playing).unwrap();
+
+    // FYI necessary push b/c:
+    // https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1875#note_2630453
+    push_one(&srcpad, gst::ClockTime::from_nseconds(1));
+
+    let s = gst::Structure::builder("MyEvent")
+        .field("unsigned", 100u64)
+        .build();
+    assert!(srcpad.push_event(gst::event::CustomDownstream::new(s)));
+    assert!(srcpad.push_event(gst::event::Eos::new()));
+
+    let mut found = false;
+    loop {
+        use gst::EventView;
+
+        if let Ok(event) = hc.pull_event() {
+            match event.view() {
+                EventView::CustomDownstream(e) => {
+                    let v = e.structure().unwrap().get::<u64>("unsigned");
+                    assert!(v.is_ok_and(|v| v == 100u64));
+                    found = true;
+                    break;
+                }
+                EventView::Eos(..) => {
+                    break;
+                }
+                _ => (),
+            };
+        }
+    }
+    intersink.set_state(gst::State::Null).unwrap();
+    assert!(found);
+}