Managed/mount push_data parity + JVM panic-safety fold-ins (chunk #2)

CHANGELOG.md

@@ -16,14 +16,15 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   + `streamDescriptorsForData(dataIndex, descriptors)`, the
   `DataStreamHandle` record, the offline `Muxer.pushData` / `pushDataTo`
   pair plus the `dataHandles()` / `dataStreamHandle(index)` accessors,
-  `MuxerFileSink.pushData` / `pushDataTo`, and `pushData` / `pushDataTo` /
-  `dataHandle()` on the srt and rtp `MuxSender`s. Pass-through / PTS /
-  size-ceiling semantics are those of the Rust `push_data` family (see the
-  muxer + pipeline entry below). Not yet exposed (recorded follow-ups): the
-  rtp `MountHandle` and srt `ManagedMuxSender` data push families, and
+  `MuxerFileSink.pushData` / `pushDataTo`, `pushData` / `pushDataTo` /
+  `dataHandle()` on the srt and rtp `MuxSender`s, the srt
+  `ManagedMuxSender.pushData` / `pushDataTo` / `dataHandle()` pair, and the
+  rtp `MountHandle.pushData` / `pushDataTo` / `dataHandle()` / `dataHandles()`
+  family. Pass-through / PTS / size-ceiling semantics are those of the Rust
+  `push_data` family (see the muxer + pipeline entry below). Not yet exposed:
   `data_handles_for_program` (the JVM `MuxerConfig` is single-program).
 
-### Added — C data-stream surface (ABI minor 12)
+### Added — C data-stream surface (ABI minor 12 → 13)
 
 - **tst-c** gains the data-stream mux surface, binding the Rust
   muxer/pipeline family below: `tst_data_stream_handle_t` plus seven new
@@ -36,6 +37,11 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   those of the Rust `push_data` family (see the muxer + pipeline entry
   below). ABI minor 11 → 12; additive — no symbol removed, no signature or
   struct layout changed.
+- **tst-c** further gains the managed-sender and RTSP-mount data push pairs:
+  `tst_managed_mux_sender_send_data` / `_to` (behind `TST_HAS_SRT`) and
+  `tst_rtsp_mount_push_data` / `_to` (behind `TST_HAS_RTP`), mirroring the
+  `send_klv` / `push_klv` families. ABI minor 12 → 13; additive — no symbol
+  removed, no signature or struct layout changed.
 
 ### Added — Python data-stream surface (`tstrans.mpegts` + srt/rtp `MuxSender`)
 
@@ -48,6 +54,9 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 - **`tstrans.srt.MuxSender` / `tstrans.rtp.MuxSender`** gain `push_data` /
   `push_data_to` / `data_handle()` — push raw private-PES payloads through a
   live sender, following the `push_klv` family shape.
+- **`tstrans.srt.ManagedMuxSender`** and **`tstrans.rtp.MountHandle`** gain the
+  same `push_data` / `push_data_to` / `data_handle()` trio, completing data
+  parity across every live sender / mount shell.
 
 ### Changed — `tio.transmux` passes private/application data streams through
 
@@ -60,6 +69,18 @@ Versioning: [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
   always carry PTS and the demuxer substitutes 0 for a PTS-less source PES,
   so a source sample with no PTS re-emerges with a literal PTS of 0.
 
+### Changed — tst-jni panic safety + handle-decode hardening (`org.tstrans`)
+
+- An unexpected Rust panic inside any `org.tstrans` JVM native now surfaces as
+  a `java.lang.RuntimeException` instead of aborting the JVM process — every
+  native is wrapped in a `jni_catch` panic boundary (the JVM twin of the C
+  binding's `ffi_catch`).
+- `Muxer.pull`'s JNI-bridge failure cases now throw an unchecked
+  `RuntimeException` rather than a checked `MuxException` (the hot drain path
+  keeps no `throws` clause), and every handle-targeted `*To` push native now
+  rejects out-of-range / negative stream handles instead of silently
+  truncating them to a valid handle.
+
 ### Added — private/application data streams (muxer + pipeline)
 
 Arbitrary private PES carriage: declare a data elementary stream with an

bindings/c/core/src/lib.rs

@@ -13,7 +13,7 @@
 //! transport surfaces are gated on the `rtp` cargo feature. The
 //! offline byte-feeding `tst_demuxer_*` surface is unconditional (no
 //! feature gate), as is the offline `tst_muxer_*` surface (un-gated from
-//! `srt` in ABI 0.9). ABI minor is `0.12` (see [`TST_ABI_VERSION_MINOR`]).
+//! `srt` in ABI 0.9). ABI minor is `0.13` (see [`TST_ABI_VERSION_MINOR`]).
 
 #![cfg_attr(not(feature = "std"), no_std)]
 #![allow(clippy::missing_safety_doc)] // every extern "C" fn has a /// header documenting the contract
@@ -186,7 +186,7 @@ pub const TST_ABI_VERSION_MAJOR: crate::c_types::c_int = 0;
 /// Minor version of the C ABI contract. See [`TST_ABI_VERSION_MAJOR`]
 /// for the bump policy.
 ///
-/// Cbindgen emits this as `#define TST_ABI_VERSION_MINOR 12` in the
+/// Cbindgen emits this as `#define TST_ABI_VERSION_MINOR 13` in the
 /// generated header. Runtime accessor: [`tst_get_abi_version_minor`].
 ///
 /// History (additive bumps only — major stays at 0 pre-1.0):
@@ -253,7 +253,14 @@ pub const TST_ABI_VERSION_MAJOR: crate::c_types::c_int = 0;
 ///   alongside video/KLV, mirroring the Rust `push_data` family.
 ///   Additive — no symbol removed, no signature or struct layout
 ///   changed.
-pub const TST_ABI_VERSION_MINOR: crate::c_types::c_int = 12;
+/// - `13` — private-data push through the managed-sender and RTSP-mount
+///   pipeline shells: the srt-gated pair `tst_managed_mux_sender_send_data`
+///   / `tst_managed_mux_sender_send_data_to` (behind `TST_HAS_SRT`) and the
+///   rtp-gated pair `tst_rtsp_mount_push_data` / `tst_rtsp_mount_push_data_to`
+///   (behind `TST_HAS_RTP`). Completes the data-stream surface parity with the
+///   video/klv/audio/subtitle push families on both shells. Additive — no
+///   symbol removed, no signature or struct layout changed.
+pub const TST_ABI_VERSION_MINOR: crate::c_types::c_int = 13;
 
 // =========================================================================
 // Runtime version accessors

bindings/c/core/src/rtsp/server/mount.rs

@@ -46,15 +46,16 @@ use std::sync::atomic::{AtomicBool, Ordering};
 
 use tst_core::mpegts::common::Pts90khz;
 use tst_core::mpegts::mux::{
-    AudioStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
+    AudioStreamHandle, DataStreamHandle, KlvStreamHandle, SubtitleStreamHandle, VideoStreamHandle,
 };
 
 use crate::config::TstMuxConfig;
 use crate::error::{
     TstError, mount_error_to_code, record_mux_error, rtsp_server_error_to_code, set_last_error,
 };
 use crate::handle::{
-    TstAudioStreamHandle, TstKlvStreamHandle, TstSubtitleStreamHandle, TstVideoStreamHandle,
+    TstAudioStreamHandle, TstDataStreamHandle, TstKlvStreamHandle, TstSubtitleStreamHandle,
+    TstVideoStreamHandle,
 };
 use crate::panic::ffi_catch;
 use crate::rtsp::server::types::{TstRtspMountHandle, TstRtspServer};
@@ -544,6 +545,53 @@ pub unsafe extern "C" fn tst_rtsp_mount_push_subtitle(
     })
 }
 
+/// Push one data payload through the mount's single data stream and out the
+/// RTSP broadcast fanout (single-stream shorthand).
+///
+/// Pass-through: `data` lands verbatim as one PES packet on `stream_id`
+/// 0xBD. PTS is written only for `carries_pts = true` streams.
+///
+/// Returns `0` on success, `TST_E_CLOSED` after `tst_rtsp_mount_cancel`,
+/// `TST_E_RTSP_MOUNT` on muxer or mount errors, `TST_E_INVALID_CONFIG` if
+/// `handle` is null.
+///
+/// # Safety
+///
+/// `handle` must be a valid non-freed `*mut tst_rtsp_mount_handle_t`.
+/// `data` must be readable for `len` bytes.
+#[cfg(feature = "rtp")]
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_rtsp_mount_push_data(
+    handle: *mut TstRtspMountHandle,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    ffi_catch(TstError::Internal as i32, || {
+        let Some(h) = (unsafe { handle.as_ref() }) else {
+            set_last_error(TstError::InvalidConfig, "null mount handle pointer");
+            return TstError::InvalidConfig as i32;
+        };
+        if h.cancelled.load(Ordering::Acquire) {
+            set_last_error(TstError::Closed, "mount handle has been cancelled");
+            return TstError::Closed as i32;
+        }
+        let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+            Ok(s) => s,
+            Err(code) => return code,
+        };
+        let pts = Pts90khz::new(pts_90khz);
+        match h.inner.push_data(slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                let code = mount_error_to_code(&e);
+                set_last_error(code, &format!("push_data failed: {e}"));
+                code as i32
+            }
+        }
+    })
+}
+
 // ---------------------------------------------------------------------------
 // Push — multi-stream (_to) variants
 // ---------------------------------------------------------------------------
@@ -764,6 +812,56 @@ pub unsafe extern "C" fn tst_rtsp_mount_push_subtitle_to(
     })
 }
 
+/// Push one data payload targeting a specific data elementary stream.
+///
+/// On a single-stream mount, prefer `tst_rtsp_mount_push_data`.
+///
+/// # Safety
+///
+/// `handle` must be a valid non-freed `*mut tst_rtsp_mount_handle_t`.
+/// `data` must be readable for `len` bytes.
+#[cfg(feature = "rtp")]
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_rtsp_mount_push_data_to(
+    handle: *mut TstRtspMountHandle,
+    stream_handle: TstDataStreamHandle,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    ffi_catch(TstError::Internal as i32, || {
+        let Some(h) = (unsafe { handle.as_ref() }) else {
+            set_last_error(TstError::InvalidConfig, "null mount handle pointer");
+            return TstError::InvalidConfig as i32;
+        };
+        if h.cancelled.load(Ordering::Acquire) {
+            set_last_error(TstError::Closed, "mount handle has been cancelled");
+            return TstError::Closed as i32;
+        }
+        let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+            Ok(s) => s,
+            Err(code) => return code,
+        };
+        // Trust-boundary validation — see push_video_to rationale above.
+        let stream = match DataStreamHandle::try_from_raw(stream_handle) {
+            Ok(s) => s,
+            Err(e) => {
+                crate::error::record_mux_error(&e);
+                return unsafe { crate::error::tst_get_last_error() };
+            }
+        };
+        let pts = Pts90khz::new(pts_90khz);
+        match h.inner.push_data_to(stream, slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                let code = mount_error_to_code(&e);
+                set_last_error(code, &format!("push_data_to failed: {e}"));
+                code as i32
+            }
+        }
+    })
+}
+
 // ---------------------------------------------------------------------------
 // Lifecycle helpers — flush, cancel, reset_stats
 // ---------------------------------------------------------------------------
@@ -1041,6 +1139,20 @@ mod tests {
         assert_eq!(rc, crate::error::TstError::InvalidConfig as i32);
     }
 
+    #[test]
+    fn push_data_null_handle_returns_invalid_config() {
+        let rc = unsafe { tst_rtsp_mount_push_data(std::ptr::null_mut(), [0u8; 4].as_ptr(), 4, 0) };
+        assert_eq!(rc, crate::error::TstError::InvalidConfig as i32);
+    }
+
+    #[test]
+    fn push_data_to_null_handle_returns_invalid_config() {
+        let rc = unsafe {
+            tst_rtsp_mount_push_data_to(std::ptr::null_mut(), 0, [0u8; 4].as_ptr(), 4, 0)
+        };
+        assert_eq!(rc, crate::error::TstError::InvalidConfig as i32);
+    }
+
     #[test]
     fn flush_null_handle_returns_invalid_config() {
         let rc = unsafe { tst_rtsp_mount_flush(std::ptr::null_mut()) };

bindings/c/core/src/sender/mux_sender.rs

@@ -1237,6 +1237,92 @@ pub unsafe extern "C" fn tst_managed_mux_sender_send_subtitle_to(
         })
 }
 
+/// Send one data payload through the managed mux sender's single data
+/// stream and out the underlying reconnecting transport.
+///
+/// Pass-through contract identical to `tst_mux_sender_send_data`: `data`
+/// lands verbatim as one PES packet on `stream_id` 0xBD; PTS is written
+/// only for `carries_pts = true` streams.
+///
+/// Single-stream form: see `tst_managed_mux_sender_send_data_to` for the
+/// multi-stream variant.
+///
+/// # Errors
+///
+/// Routed through `tst_get_last_error()`. Same code set as
+/// `tst_mux_sender_send_data` plus `TST_E_NOT_AVAILABLE` (transport
+/// mid-reconnect; transient).
+///
+/// # C ABI
+///
+/// `tst_managed_mux_sender_send_data` — see `bindings/c/include/tstrans.h`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_managed_mux_sender_send_data(
+    p: *mut TstManagedMuxSender,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    let Some(handle) = (unsafe { p.as_ref() }) else {
+        set_last_error(TstError::InvalidConfig, "null sender pointer");
+        return TstError::InvalidConfig as i32;
+    };
+    let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+        Ok(s) => s,
+        Err(code) => return code,
+    };
+    let pts = Pts90khz::new(pts_90khz);
+    handle
+        .inner
+        .with_inner_ref(|s| match s.send_data(slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                record_shell_error(&e);
+                unsafe { tst_get_last_error() }
+            }
+        })
+}
+
+/// Send one data payload targeting a specific data elementary stream on a
+/// managed (auto-reconnecting) sender. `stream_handle` is stable across
+/// reconnects. Out-of-range handles surface as `TST_E_INVALID_USAGE`.
+///
+/// On a single-stream sender, prefer `tst_managed_mux_sender_send_data`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn tst_managed_mux_sender_send_data_to(
+    p: *mut TstManagedMuxSender,
+    stream_handle: TstDataStreamHandle,
+    data: *const u8,
+    len: usize,
+    pts_90khz: i64,
+) -> libc::c_int {
+    let Some(wrapper) = (unsafe { p.as_ref() }) else {
+        set_last_error(TstError::InvalidConfig, "null sender pointer");
+        return TstError::InvalidConfig as i32;
+    };
+    let slice = match unsafe { crate::ffi_slice::ffi_slice(data, len, "data") } {
+        Ok(s) => s,
+        Err(code) => return code,
+    };
+    let stream = match DataStreamHandle::try_from_raw(stream_handle) {
+        Ok(h) => h,
+        Err(e) => {
+            crate::error::record_mux_error(&e);
+            return unsafe { tst_get_last_error() };
+        }
+    };
+    let pts = Pts90khz::new(pts_90khz);
+    wrapper
+        .inner
+        .with_inner_ref(|s| match s.send_data_to(stream, slice, pts) {
+            Ok(()) => 0,
+            Err(e) => {
+                record_shell_error(&e);
+                unsafe { tst_get_last_error() }
+            }
+        })
+}
+
 /// Snapshot stats for a `tst_managed_mux_sender_t` into `*out`.
 ///
 /// Returns 0 on success, `TST_E_INVALID_CONFIG` if either pointer is