cyclonedds 3.0.2

use crate::DdsResult;
use cyclonedds_rust_sys::*;
use std::ffi::c_void;
use std::sync::Arc;

// Existing callback types (4)
type DataAvailableCb = Arc<dyn Fn(i32) + Send + Sync>;
type PublicationMatchedCb = Arc<dyn Fn(i32, dds_publication_matched_status_t) + Send + Sync>;
type SubscriptionMatchedCb = Arc<dyn Fn(i32, dds_subscription_matched_status_t) + Send + Sync>;
type LivelinessChangedCb = Arc<dyn Fn(i32, dds_liveliness_changed_status_t) + Send + Sync>;

// New callback types (9 with status, 1 without)
type InconsistentTopicCb = Arc<dyn Fn(i32, dds_inconsistent_topic_status_t) + Send + Sync>;
type LivelinessLostCb = Arc<dyn Fn(i32, dds_liveliness_lost_status_t) + Send + Sync>;
type OfferedDeadlineMissedCb = Arc<dyn Fn(i32, dds_offered_deadline_missed_status_t) + Send + Sync>;
type OfferedIncompatibleQosCb =
    Arc<dyn Fn(i32, dds_offered_incompatible_qos_status_t) + Send + Sync>;
type DataOnReadersCb = Arc<dyn Fn(i32) + Send + Sync>;
type SampleLostCb = Arc<dyn Fn(i32, dds_sample_lost_status_t) + Send + Sync>;
type SampleRejectedCb = Arc<dyn Fn(i32, dds_sample_rejected_status_t) + Send + Sync>;
type RequestedDeadlineMissedCb =
    Arc<dyn Fn(i32, dds_requested_deadline_missed_status_t) + Send + Sync>;
type RequestedIncompatibleQosCb =
    Arc<dyn Fn(i32, dds_requested_incompatible_qos_status_t) + Send + Sync>;

struct ListenerClosures {
    // Existing
    on_data_available: Option<DataAvailableCb>,
    on_publication_matched: Option<PublicationMatchedCb>,
    on_subscription_matched: Option<SubscriptionMatchedCb>,
    on_liveliness_changed: Option<LivelinessChangedCb>,
    // New
    on_inconsistent_topic: Option<InconsistentTopicCb>,
    on_liveliness_lost: Option<LivelinessLostCb>,
    on_offered_deadline_missed: Option<OfferedDeadlineMissedCb>,
    on_offered_incompatible_qos: Option<OfferedIncompatibleQosCb>,
    on_data_on_readers: Option<DataOnReadersCb>,
    on_sample_lost: Option<SampleLostCb>,
    on_sample_rejected: Option<SampleRejectedCb>,
    on_requested_deadline_missed: Option<RequestedDeadlineMissedCb>,
    on_requested_incompatible_qos: Option<RequestedIncompatibleQosCb>,
}

struct ListenerInner {
    ptr: *mut dds_listener_t,
    _closures: Box<ListenerClosures>,
}

unsafe impl Send for ListenerInner {}
unsafe impl Sync for ListenerInner {}

#[derive(Clone)]
pub struct Listener {
    inner: Arc<ListenerInner>,
}

// ── Trampolines ─────────────────────────────────────────────────────
//
// CycloneDDS invokes these from its own threads (typically `recv`, while
// holding internal ddsi locks). A Rust panic escaping an `extern "C"` frame
// aborts the process on Rust >= 1.81 (and was outright UB before), so every
// callback must be wrapped in a panic barrier: a single `unwrap()` in a user
// closure would otherwise take down the whole application.
//
// The barrier lives in `dispatch` and the trampolines are generated by the two
// macros below, so a newly added listener callback cannot silently forget it.
// The same rule is already applied in `waitset.rs::trampoline_qc_filter`.

/// Resolve the user-data pointer and run `f` behind a panic barrier.
///
/// # Safety
/// `arg` must be either null or the `ListenerClosures` pointer registered with
/// `dds_create_listener`, which stays alive as long as any entity holds a
/// clone of the owning `Listener`.
#[inline]
unsafe fn dispatch<F>(arg: *mut c_void, f: F)
where
    F: FnOnce(&ListenerClosures),
{
    if arg.is_null() {
        return;
    }
    let closures = &*(arg as *const ListenerClosures);
    if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(closures))).is_err() {
        eprintln!("cyclonedds: listener callback panicked; panic contained at the FFI boundary");
    }
}

/// Trampoline for callbacks shaped `(entity, status, arg)`.
macro_rules! status_trampoline {
    ($name:ident, $field:ident, $status:ty) => {
        unsafe extern "C" fn $name(entity: dds_entity_t, status: $status, arg: *mut c_void) {
            dispatch(arg, |closures| {
                if let Some(ref cb) = closures.$field {
                    cb(entity, status);
                }
            });
        }
    };
}

/// Trampoline for callbacks shaped `(entity, arg)`.
macro_rules! plain_trampoline {
    ($name:ident, $field:ident) => {
        unsafe extern "C" fn $name(entity: dds_entity_t, arg: *mut c_void) {
            dispatch(arg, |closures| {
                if let Some(ref cb) = closures.$field {
                    cb(entity);
                }
            });
        }
    };
}

plain_trampoline!(trampoline_data_available, on_data_available);
status_trampoline!(
    trampoline_publication_matched,
    on_publication_matched,
    dds_publication_matched_status_t
);
status_trampoline!(
    trampoline_subscription_matched,
    on_subscription_matched,
    dds_subscription_matched_status_t
);
status_trampoline!(
    trampoline_liveliness_changed,
    on_liveliness_changed,
    dds_liveliness_changed_status_t
);

status_trampoline!(
    trampoline_inconsistent_topic,
    on_inconsistent_topic,
    dds_inconsistent_topic_status_t
);
status_trampoline!(
    trampoline_liveliness_lost,
    on_liveliness_lost,
    dds_liveliness_lost_status_t
);
status_trampoline!(
    trampoline_offered_deadline_missed,
    on_offered_deadline_missed,
    dds_offered_deadline_missed_status_t
);
status_trampoline!(
    trampoline_offered_incompatible_qos,
    on_offered_incompatible_qos,
    dds_offered_incompatible_qos_status_t
);
plain_trampoline!(trampoline_data_on_readers, on_data_on_readers);
status_trampoline!(
    trampoline_sample_lost,
    on_sample_lost,
    dds_sample_lost_status_t
);
status_trampoline!(
    trampoline_sample_rejected,
    on_sample_rejected,
    dds_sample_rejected_status_t
);
status_trampoline!(
    trampoline_requested_deadline_missed,
    on_requested_deadline_missed,
    dds_requested_deadline_missed_status_t
);
status_trampoline!(
    trampoline_requested_incompatible_qos,
    on_requested_incompatible_qos,
    dds_requested_incompatible_qos_status_t
);

// ── ListenerBuilder ─────────────────────────────────────────────────

pub struct ListenerBuilder {
    // Existing
    on_data_available: Option<DataAvailableCb>,
    on_publication_matched: Option<PublicationMatchedCb>,
    on_subscription_matched: Option<SubscriptionMatchedCb>,
    on_liveliness_changed: Option<LivelinessChangedCb>,
    // New
    on_inconsistent_topic: Option<InconsistentTopicCb>,
    on_liveliness_lost: Option<LivelinessLostCb>,
    on_offered_deadline_missed: Option<OfferedDeadlineMissedCb>,
    on_offered_incompatible_qos: Option<OfferedIncompatibleQosCb>,
    on_data_on_readers: Option<DataOnReadersCb>,
    on_sample_lost: Option<SampleLostCb>,
    on_sample_rejected: Option<SampleRejectedCb>,
    on_requested_deadline_missed: Option<RequestedDeadlineMissedCb>,
    on_requested_incompatible_qos: Option<RequestedIncompatibleQosCb>,
}

impl ListenerBuilder {
    pub fn new() -> Self {
        ListenerBuilder {
            on_data_available: None,
            on_publication_matched: None,
            on_subscription_matched: None,
            on_liveliness_changed: None,
            on_inconsistent_topic: None,
            on_liveliness_lost: None,
            on_offered_deadline_missed: None,
            on_offered_incompatible_qos: None,
            on_data_on_readers: None,
            on_sample_lost: None,
            on_sample_rejected: None,
            on_requested_deadline_missed: None,
            on_requested_incompatible_qos: None,
        }
    }

    // ── Existing builder methods ────────────────────────────────────

    pub fn on_data_available(mut self, cb: impl Fn(i32) + Send + Sync + 'static) -> Self {
        self.on_data_available = Some(Arc::new(cb));
        self
    }

    pub fn on_publication_matched(
        mut self,
        cb: impl Fn(i32, dds_publication_matched_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_publication_matched = Some(Arc::new(cb));
        self
    }

    pub fn on_subscription_matched(
        mut self,
        cb: impl Fn(i32, dds_subscription_matched_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_subscription_matched = Some(Arc::new(cb));
        self
    }

    pub fn on_liveliness_changed(
        mut self,
        cb: impl Fn(i32, dds_liveliness_changed_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_liveliness_changed = Some(Arc::new(cb));
        self
    }

    // ── New builder methods ─────────────────────────────────────────

    pub fn on_inconsistent_topic(
        mut self,
        cb: impl Fn(i32, dds_inconsistent_topic_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_inconsistent_topic = Some(Arc::new(cb));
        self
    }

    pub fn on_liveliness_lost(
        mut self,
        cb: impl Fn(i32, dds_liveliness_lost_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_liveliness_lost = Some(Arc::new(cb));
        self
    }

    pub fn on_offered_deadline_missed(
        mut self,
        cb: impl Fn(i32, dds_offered_deadline_missed_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_offered_deadline_missed = Some(Arc::new(cb));
        self
    }

    pub fn on_offered_incompatible_qos(
        mut self,
        cb: impl Fn(i32, dds_offered_incompatible_qos_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_offered_incompatible_qos = Some(Arc::new(cb));
        self
    }

    pub fn on_data_on_readers(mut self, cb: impl Fn(i32) + Send + Sync + 'static) -> Self {
        self.on_data_on_readers = Some(Arc::new(cb));
        self
    }

    pub fn on_sample_lost(
        mut self,
        cb: impl Fn(i32, dds_sample_lost_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_sample_lost = Some(Arc::new(cb));
        self
    }

    pub fn on_sample_rejected(
        mut self,
        cb: impl Fn(i32, dds_sample_rejected_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_sample_rejected = Some(Arc::new(cb));
        self
    }

    pub fn on_requested_deadline_missed(
        mut self,
        cb: impl Fn(i32, dds_requested_deadline_missed_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_requested_deadline_missed = Some(Arc::new(cb));
        self
    }

    pub fn on_requested_incompatible_qos(
        mut self,
        cb: impl Fn(i32, dds_requested_incompatible_qos_status_t) + Send + Sync + 'static,
    ) -> Self {
        self.on_requested_incompatible_qos = Some(Arc::new(cb));
        self
    }

    // ── Build ───────────────────────────────────────────────────────

    pub fn build(self) -> DdsResult<Listener> {
        let closures = Box::new(ListenerClosures {
            on_data_available: self.on_data_available,
            on_publication_matched: self.on_publication_matched,
            on_subscription_matched: self.on_subscription_matched,
            on_liveliness_changed: self.on_liveliness_changed,
            on_inconsistent_topic: self.on_inconsistent_topic,
            on_liveliness_lost: self.on_liveliness_lost,
            on_offered_deadline_missed: self.on_offered_deadline_missed,
            on_offered_incompatible_qos: self.on_offered_incompatible_qos,
            on_data_on_readers: self.on_data_on_readers,
            on_sample_lost: self.on_sample_lost,
            on_sample_rejected: self.on_sample_rejected,
            on_requested_deadline_missed: self.on_requested_deadline_missed,
            on_requested_incompatible_qos: self.on_requested_incompatible_qos,
        });

        let arg_ptr = &*closures as *const ListenerClosures as *mut c_void;
        let ptr = unsafe { dds_create_listener(arg_ptr) };
        if ptr.is_null() {
            return Err(crate::DdsError::OutOfResources);
        }

        unsafe {
            // Existing
            if closures.on_data_available.is_some() {
                dds_lset_data_available(ptr, Some(trampoline_data_available));
            }
            if closures.on_publication_matched.is_some() {
                dds_lset_publication_matched(ptr, Some(trampoline_publication_matched));
            }
            if closures.on_subscription_matched.is_some() {
                dds_lset_subscription_matched(ptr, Some(trampoline_subscription_matched));
            }
            if closures.on_liveliness_changed.is_some() {
                dds_lset_liveliness_changed(ptr, Some(trampoline_liveliness_changed));
            }
            // New
            if closures.on_inconsistent_topic.is_some() {
                dds_lset_inconsistent_topic(ptr, Some(trampoline_inconsistent_topic));
            }
            if closures.on_liveliness_lost.is_some() {
                dds_lset_liveliness_lost(ptr, Some(trampoline_liveliness_lost));
            }
            if closures.on_offered_deadline_missed.is_some() {
                dds_lset_offered_deadline_missed(ptr, Some(trampoline_offered_deadline_missed));
            }
            if closures.on_offered_incompatible_qos.is_some() {
                dds_lset_offered_incompatible_qos(ptr, Some(trampoline_offered_incompatible_qos));
            }
            if closures.on_data_on_readers.is_some() {
                dds_lset_data_on_readers(ptr, Some(trampoline_data_on_readers));
            }
            if closures.on_sample_lost.is_some() {
                dds_lset_sample_lost(ptr, Some(trampoline_sample_lost));
            }
            if closures.on_sample_rejected.is_some() {
                dds_lset_sample_rejected(ptr, Some(trampoline_sample_rejected));
            }
            if closures.on_requested_deadline_missed.is_some() {
                dds_lset_requested_deadline_missed(ptr, Some(trampoline_requested_deadline_missed));
            }
            if closures.on_requested_incompatible_qos.is_some() {
                dds_lset_requested_incompatible_qos(
                    ptr,
                    Some(trampoline_requested_incompatible_qos),
                );
            }
        }

        Ok(Listener {
            inner: Arc::new(ListenerInner {
                ptr,
                _closures: closures,
            }),
        })
    }
}

impl Default for ListenerBuilder {
    fn default() -> Self {
        Self::new()
    }
}

impl Listener {
    pub fn builder() -> ListenerBuilder {
        ListenerBuilder::new()
    }

    pub fn as_ptr(&self) -> *mut dds_listener_t {
        self.inner.ptr
    }

    #[cfg(test)]
    pub(crate) fn reference_count(&self) -> usize {
        Arc::strong_count(&self.inner)
    }
}

impl Drop for ListenerInner {
    fn drop(&mut self) {
        unsafe {
            dds_delete_listener(self.ptr);
        }
    }
}