Set src element on inference bus messages

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);