Set src element on inference bus messages (#54)

The video inference element posted its Element messages with
gst::message::Element::new, which leaves the message src unset. Bus
consumers rely on src to attribute each message to the element that
produced it, so unattributed messages can't be matched back to this
element on shared buses with multiple inference variants.

Build the messages with Element::builder(s).src(&*self.obj()).build()
at every post site in the video element (inference results, error,
visual-anomaly, object-tracking, and classification fallbacks) so each
message carries its producing element.

Add an e2e regression test asserting that inference element messages
expose a non-empty src.

Author: ferjm

src/video/imp.rs

@@ -898,7 +898,9 @@ impl BaseTransformImpl for EdgeImpulseVideoInfer {
                         inbuf.pts().unwrap_or(gst::ClockTime::ZERO),
                         e.to_string(),
                     );
-                    let _ = self.obj().post_message(gst::message::Element::new(s));
+                    let _ = self
+                        .obj()
+                        .post_message(gst::message::Element::builder(s).src(&*self.obj()).build());
                     // Put the model back in the state
                     let mut state = self.state.lock().unwrap();
                     state.model = Some(model);
@@ -1061,7 +1063,9 @@ impl BaseTransformImpl for EdgeImpulseVideoInfer {
                         elapsed.as_millis() as u32,
                         resize_time_ms,
                     );
-                    let _ = self.obj().post_message(gst::message::Element::new(s));
+                    let _ = self
+                        .obj()
+                        .post_message(gst::message::Element::builder(s).src(&*self.obj()).build());
                 }
             }
 
@@ -1148,7 +1152,9 @@ impl BaseTransformImpl for EdgeImpulseVideoInfer {
                         elapsed.as_millis() as u32,
                         resize_time_ms,
                     );
-                    let _ = self.obj().post_message(gst::message::Element::new(s));
+                    let _ = self
+                        .obj()
+                        .post_message(gst::message::Element::builder(s).src(&*self.obj()).build());
                 }
             }
 
@@ -1229,7 +1235,9 @@ impl BaseTransformImpl for EdgeImpulseVideoInfer {
                         elapsed.as_millis() as u32,
                         resize_time_ms,
                     );
-                    let _ = self.obj().post_message(gst::message::Element::new(s));
+                    let _ = self
+                        .obj()
+                        .post_message(gst::message::Element::builder(s).src(&*self.obj()).build());
                 } else {
                     // Classification fallback if bounding_boxes is empty
                     if let Some(classification) = result_value
@@ -1270,7 +1278,9 @@ impl BaseTransformImpl for EdgeImpulseVideoInfer {
                         elapsed.as_millis() as u32,
                         resize_time_ms,
                     );
-                    let _ = self.obj().post_message(gst::message::Element::new(s));
+                    let _ = self
+                        .obj()
+                        .post_message(gst::message::Element::builder(s).src(&*self.obj()).build());
                 }
             } else {
                 // No bounding_boxes field, treat as classification
@@ -1312,7 +1322,9 @@ impl BaseTransformImpl for EdgeImpulseVideoInfer {
                     elapsed.as_millis() as u32,
                     resize_time_ms,
                 );
-                let _ = self.obj().post_message(gst::message::Element::new(s));
+                let _ = self
+                    .obj()
+                    .post_message(gst::message::Element::builder(s).src(&*self.obj()).build());
             }
 
             // Put the model back in the state

tests/e2e.rs

@@ -82,6 +82,7 @@ fn run_pipeline_for(pipeline_str: &str, duration: Duration) -> TestResults {
                                 "result": serde_json::from_str::<serde_json::Value>(&result_json)
                                     .unwrap_or(serde_json::Value::Null),
                                 "timing_ms": timing_ms,
+                                "src": msg.src().map(|s| s.name().to_string()),
                             });
                             res.inference_messages.push(entry);
                             res.buffer_count += 1;
@@ -200,6 +201,32 @@ fn test_video_inference_message_structure() {
     );
 }
 
+/// Regression: inference element messages must carry their `src` element.
+///
+/// GStreamer element messages should identify the element that posted them so
+/// bus consumers can attribute each message to its producing element. The
+/// inference result messages are posted via `gst::message::Element` and must be
+/// built with `src` set to this element; posting without a `src` leaves the
+/// message unattributed on the bus.
+#[test]
+fn test_inference_messages_carry_src_element() {
+    let src = video_test_source(5);
+    let pipeline = format!("{src} ! edgeimpulsevideoinfer ! fakesink");
+
+    let results = run_pipeline_for(&pipeline, Duration::from_secs(10));
+
+    assert!(
+        !results.inference_messages.is_empty(),
+        "Expected at least one inference bus message"
+    );
+    for msg in &results.inference_messages {
+        assert!(
+            msg.get("src").and_then(|v| v.as_str()).is_some(),
+            "Inference element message must carry its src element. Message: {msg}"
+        );
+    }
+}
+
 #[test]
 fn test_overlay_does_not_crash() {
     let src = video_test_source(5);