pamoja_ffi/

transport.rs

//! The C ABI for composing transports.
//!
//! A ladder rung, a fault injector, and a degraded link all take "some
//! transport", which in Rust is any `impl Transport`. A C ABI has no generics,
//! so this module carries one tagged handle that holds whichever transport was
//! built and dispatches to it. One handle keeps the composing calls to a single
//! function each, rather than one per transport kind, and lets the set of kinds
//! grow without reshaping the surface.
//!
//! A transport built here is separate from a client handle such as
//! [`PamojaMqttClient`](crate::mqtt::PamojaMqttClient). A client is for driving
//! a link directly; a transport is for composition, and is consumed by whatever
//! it is composed into. Keeping them apart means nothing has to move out of a
//! live, shared handle. A link the host implements itself enters through
//! [`pamoja_transport_from_callbacks`](crate::host::pamoja_transport_from_callbacks)
//! and is one of these handles from then on.

use std::ffi::CString;
use std::future::Future;
use std::pin::Pin;
use std::ptr;

use pamoja_core::{Message, Receive, Result, Transport};

use crate::{read_bytes, set_last_error, PamojaStatus};

/// Object-safe erasure of a transport, so a wrapper can hold any of them.
///
/// The core trait returns `impl Future`, which is not dyn-compatible; this one
/// boxes the future so a wrapping kind can hold a transport without naming its
/// concrete type. That is what keeps the union below from naming itself: a
/// nested transport is reached through this trait, whose futures are already a
/// type the compiler can name.
trait DynTransport: Send {
    /// Connects the erased transport.
    fn connect(&mut self) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>>;

    /// Sends a payload over the erased transport.
    fn send<'a>(
        &'a mut self,
        topic: &'a str,
        payload: &'a [u8],
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;

    /// Subscribes the erased transport to a topic.
    fn subscribe<'a>(
        &'a mut self,
        topic: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;

    /// Awaits the next message the erased transport delivers.
    fn recv(&mut self) -> Pin<Box<dyn Future<Output = Result<Option<Message>>> + Send + '_>>;
}

/// Newtype carrying one concrete transport behind [`DynTransport`].
struct Erased<T>(T);

impl<T: Transport + Receive + Send> DynTransport for Erased<T> {
    fn connect(&mut self) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
        Box::pin(Transport::connect(&mut self.0))
    }

    fn send<'a>(
        &'a mut self,
        topic: &'a str,
        payload: &'a [u8],
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>> {
        Box::pin(Transport::send(&mut self.0, topic, payload))
    }

    fn subscribe<'a>(
        &'a mut self,
        topic: &'a str,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>> {
        Box::pin(Transport::subscribe(&mut self.0, topic))
    }

    fn recv(&mut self) -> Pin<Box<dyn Future<Output = Result<Option<Message>>> + Send + '_>> {
        Box::pin(Receive::recv(&mut self.0))
    }
}

/// A transport of any kind, ready to be nested inside a wrapper.
pub(crate) struct AnyTransport(Box<dyn DynTransport>);

impl AnyTransport {
    /// Erases one transport so a wrapper can hold it.
    fn new(transport: Kind) -> Self {
        Self(Box::new(Erased(transport)))
    }
}

impl Transport for AnyTransport {
    async fn connect(&mut self) -> Result<()> {
        self.0.connect().await
    }

    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
        self.0.send(topic, payload).await
    }

    async fn subscribe(&mut self, topic: &str) -> Result<()> {
        self.0.subscribe(topic).await
    }
}

impl Receive for AnyTransport {
    async fn recv(&mut self) -> Result<Option<Message>> {
        self.0.recv().await
    }
}

/// One transport, whichever kind it was built as.
///
/// A wrapping kind holds its inner transport erased rather than as this enum, so
/// a faulty link can wrap a degraded one to any depth without the enum naming
/// itself. Naming itself would make the hidden type of each method depend on
/// knowing that same type, which is a cycle rather than recursion.
pub(crate) enum Kind {
    /// An MQTT broker connection.
    #[cfg(feature = "mqtt")]
    Mqtt(pamoja_mqtt::MqttTransport),
    /// A CoAP endpoint.
    #[cfg(feature = "coap")]
    Coap(pamoja_coap::CoapTransport),
    /// An in-process link to a loopback broker.
    #[cfg(feature = "loopback")]
    Loopback(pamoja_loopback::LoopbackTransport),
    /// Another transport with a set number of sends made to fail.
    #[cfg(feature = "loopback")]
    Faulty(pamoja_loopback::Faulty<AnyTransport>),
    /// Another transport carrying loss and outages.
    #[cfg(feature = "sim")]
    Degraded(pamoja_sim::DegradedLink<AnyTransport>),
    /// A link the host implements through callbacks.
    Host(crate::host::CallbackTransport),
}

impl Kind {
    /// Whether the transport delivers messages, so a ladder listens on it rather
    /// than treating it as an uplink.
    pub(crate) fn listens(&self) -> bool {
        match self {
            Kind::Host(inner) => inner.listens(),
            _ => true,
        }
    }
}

impl Transport for Kind {
    async fn connect(&mut self) -> Result<()> {
        match self {
            #[cfg(feature = "mqtt")]
            Kind::Mqtt(inner) => inner.connect().await,
            #[cfg(feature = "coap")]
            Kind::Coap(inner) => inner.connect().await,
            #[cfg(feature = "loopback")]
            Kind::Loopback(inner) => inner.connect().await,
            #[cfg(feature = "loopback")]
            Kind::Faulty(inner) => inner.connect().await,
            #[cfg(feature = "sim")]
            Kind::Degraded(inner) => inner.connect().await,
            Kind::Host(inner) => inner.connect().await,
        }
    }

    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
        match self {
            #[cfg(feature = "mqtt")]
            Kind::Mqtt(inner) => inner.send(topic, payload).await,
            #[cfg(feature = "coap")]
            Kind::Coap(inner) => inner.send(topic, payload).await,
            #[cfg(feature = "loopback")]
            Kind::Loopback(inner) => inner.send(topic, payload).await,
            #[cfg(feature = "loopback")]
            Kind::Faulty(inner) => inner.send(topic, payload).await,
            #[cfg(feature = "sim")]
            Kind::Degraded(inner) => inner.send(topic, payload).await,
            Kind::Host(inner) => inner.send(topic, payload).await,
        }
    }

    async fn subscribe(&mut self, topic: &str) -> Result<()> {
        match self {
            #[cfg(feature = "mqtt")]
            Kind::Mqtt(inner) => inner.subscribe(topic).await,
            #[cfg(feature = "coap")]
            Kind::Coap(inner) => inner.subscribe(topic).await,
            #[cfg(feature = "loopback")]
            Kind::Loopback(inner) => inner.subscribe(topic).await,
            #[cfg(feature = "loopback")]
            Kind::Faulty(inner) => inner.subscribe(topic).await,
            #[cfg(feature = "sim")]
            Kind::Degraded(inner) => inner.subscribe(topic).await,
            Kind::Host(inner) => inner.subscribe(topic).await,
        }
    }
}

impl Receive for Kind {
    async fn recv(&mut self) -> Result<Option<Message>> {
        match self {
            #[cfg(feature = "mqtt")]
            Kind::Mqtt(inner) => inner.recv().await,
            #[cfg(feature = "coap")]
            Kind::Coap(inner) => inner.recv().await,
            #[cfg(feature = "loopback")]
            Kind::Loopback(inner) => inner.recv().await,
            #[cfg(feature = "loopback")]
            Kind::Faulty(inner) => inner.recv().await,
            #[cfg(feature = "sim")]
            Kind::Degraded(inner) => inner.recv().await,
            Kind::Host(inner) => inner.recv().await,
        }
    }
}

/// An opaque handle to one message that arrived on a subscribed topic.
///
/// CoAP and the loopback broker both hand back a topic and a payload, so one
/// handle serves them rather than a near-identical type per transport.
pub struct PamojaMessage {
    topic: CString,
    payload: Vec<u8>,
}

impl PamojaMessage {
    /// Wraps a received message in a handle the caller owns.
    ///
    /// A topic carrying an interior null cannot cross as a C string, so it is
    /// replaced with an empty one rather than truncated at the null, which would
    /// hand the caller a plausible-looking wrong topic.
    pub(crate) fn into_raw(topic: String, payload: Vec<u8>) -> *mut Self {
        let topic = CString::new(topic).unwrap_or_default();
        Box::into_raw(Box::new(Self { topic, payload }))
    }

    /// Takes the message back into the core type, for a message a host built.
    pub(crate) fn into_message(self) -> Message {
        Message::new(self.topic.into_string().unwrap_or_default(), self.payload)
    }
}

/// Returns the topic a message arrived on.
///
/// # Arguments
///
/// * `message` - the message.
///
/// # Returns
///
/// A null-terminated UTF-8 string owned by the message and valid until it is
/// freed, or null if `message` is null.
///
/// # Safety
///
/// `message` must be a live handle from a call that produced one, or null.
#[no_mangle]
pub unsafe extern "C" fn pamoja_message_topic(
    message: *const PamojaMessage,
) -> *const std::ffi::c_char {
    if message.is_null() {
        return ptr::null();
    }
    (*message).topic.as_ptr()
}

/// Returns a pointer to a message payload.
///
/// # Arguments
///
/// * `message` - the message.
///
/// # Returns
///
/// A pointer to the bytes, valid until the message is freed, or null if
/// `message` is null.
///
/// # Safety
///
/// `message` must be a live handle from a call that produced one, or null.
#[no_mangle]
pub unsafe extern "C" fn pamoja_message_payload(message: *const PamojaMessage) -> *const u8 {
    if message.is_null() {
        return ptr::null();
    }
    (*message).payload.as_ptr()
}

/// Returns the length in bytes of a message payload.
///
/// # Arguments
///
/// * `message` - the message.
///
/// # Returns
///
/// The length, or 0 if `message` is null.
///
/// # Safety
///
/// `message` must be a live handle from a call that produced one, or null.
#[no_mangle]
pub unsafe extern "C" fn pamoja_message_payload_len(message: *const PamojaMessage) -> usize {
    if message.is_null() {
        return 0;
    }
    (*message).payload.len()
}

/// Releases a message handle.
///
/// Passing null is a no-op.
///
/// # Safety
///
/// `message` must be a handle from a call that produced one and that has not
/// already been freed, or null. After this call it must not be used again.
#[no_mangle]
pub unsafe extern "C" fn pamoja_message_free(message: *mut PamojaMessage) {
    if !message.is_null() {
        drop(Box::from_raw(message));
    }
}

/// An opaque handle to one transport, ready to drive or to compose.
///
/// Release it with [`pamoja_transport_free`] unless it has been consumed by a
/// call that takes ownership, such as adding it to a ladder.
pub struct PamojaTransport {
    pub(crate) kind: Kind,
}

impl PamojaTransport {
    /// Wraps a transport kind in a handle the caller owns.
    pub(crate) fn into_raw(kind: Kind) -> *mut Self {
        Box::into_raw(Box::new(Self { kind }))
    }
}

/// Creates an MQTT transport from broker settings.
///
/// # Arguments
///
/// * `config` - the broker settings, read the same way a client reads them.
///
/// # Returns
///
/// A handle the caller must release with [`pamoja_transport_free`] or hand to a
/// call that consumes it, or null on failure.
///
/// # Safety
///
/// `config` must point to a valid config whose strings are valid
/// null-terminated UTF-8 for the duration of the call.
#[cfg(feature = "mqtt")]
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_mqtt(
    config: *const crate::mqtt::PamojaMqttConfig,
) -> *mut PamojaTransport {
    let Some(settings) = crate::mqtt::mqtt_settings(config) else {
        return ptr::null_mut();
    };
    PamojaTransport::into_raw(Kind::Mqtt(pamoja_mqtt::MqttTransport::new(settings)))
}

/// Wraps a transport so a set number of its next sends fail.
///
/// This is how a caller checks that a ladder falls through to its next rung, or
/// that a buffer fills, without unplugging anything.
///
/// # Arguments
///
/// * `transport` - the transport to wrap, consumed by this call.
/// * `failures` - how many upcoming sends to fail.
///
/// # Returns
///
/// A new handle owning the wrapped transport, or null if `transport` is null.
///
/// # Safety
///
/// `transport` must be a live handle that has not been freed or consumed. After
/// this call it must not be used again, whatever the result.
#[cfg(feature = "loopback")]
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_faulty(
    transport: *mut PamojaTransport,
    failures: usize,
) -> *mut PamojaTransport {
    let Some(inner) = take_transport(transport) else {
        return ptr::null_mut();
    };
    PamojaTransport::into_raw(Kind::Faulty(pamoja_loopback::Faulty::new(
        AnyTransport::new(inner),
        failures,
    )))
}

/// Wraps a transport in a link that loses packets and goes down.
///
/// # Arguments
///
/// * `transport` - the transport to wrap, consumed by this call.
/// * `drop_every` - lose one send in every this many, or 0 to lose none.
/// * `up` - how many sends the link stays up for, or 0 to never go down.
/// * `down` - how many sends it then stays down for.
///
/// # Returns
///
/// A new handle owning the wrapped transport, or null if `transport` is null.
///
/// # Safety
///
/// `transport` must be a live handle that has not been freed or consumed. After
/// this call it must not be used again, whatever the result.
#[cfg(feature = "sim")]
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_degraded(
    transport: *mut PamojaTransport,
    drop_every: u32,
    up: u32,
    down: u32,
) -> *mut PamojaTransport {
    let Some(inner) = take_transport(transport) else {
        return ptr::null_mut();
    };
    let mut link = pamoja_sim::DegradedLink::new(AnyTransport::new(inner));
    if drop_every != 0 {
        link = link.drop_every(drop_every);
    }
    if up != 0 {
        link = link.intermittent(up, down);
    }
    PamojaTransport::into_raw(Kind::Degraded(link))
}

/// Connects a transport.
///
/// # Arguments
///
/// * `transport` - the transport to connect.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] once connected.
///
/// # Safety
///
/// `transport` must be a live handle that has not been freed or consumed.
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_connect(transport: *mut PamojaTransport) -> PamojaStatus {
    let Some(transport) = transport_handle(transport) else {
        return PamojaStatus::InvalidArgument;
    };
    status(crate::runtime().block_on(transport.kind.connect()))
}

/// Sends a payload to a topic over a transport.
///
/// # Arguments
///
/// * `transport` - the transport to send over.
/// * `topic` - the destination topic, as null-terminated UTF-8.
/// * `payload` - the bytes to send.
/// * `payload_len` - the length of `payload`.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] once the transport has taken the payload.
///
/// # Safety
///
/// `transport` must be a live handle, `topic` a valid null-terminated UTF-8
/// string, and `payload` must point to at least `payload_len` readable bytes or
/// be null when that length is 0.
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_send(
    transport: *mut PamojaTransport,
    topic: *const std::ffi::c_char,
    payload: *const u8,
    payload_len: usize,
) -> PamojaStatus {
    let Some(transport) = transport_handle(transport) else {
        return PamojaStatus::InvalidArgument;
    };
    let Some(topic) = crate::read_str(topic, "topic") else {
        return PamojaStatus::InvalidArgument;
    };
    let payload = match read_bytes(payload, payload_len) {
        Ok(payload) => payload,
        Err(status) => return status,
    };
    status(crate::runtime().block_on(transport.kind.send(topic, &payload)))
}

/// Subscribes a transport to a topic.
///
/// # Arguments
///
/// * `transport` - the transport to subscribe.
/// * `topic` - the topic to subscribe to, as null-terminated UTF-8.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] once subscribed.
///
/// # Safety
///
/// `transport` must be a live handle and `topic` a valid null-terminated UTF-8
/// string.
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_subscribe(
    transport: *mut PamojaTransport,
    topic: *const std::ffi::c_char,
) -> PamojaStatus {
    let Some(transport) = transport_handle(transport) else {
        return PamojaStatus::InvalidArgument;
    };
    let Some(topic) = crate::read_str(topic, "topic") else {
        return PamojaStatus::InvalidArgument;
    };
    status(crate::runtime().block_on(transport.kind.subscribe(topic)))
}

/// Waits for the next message a transport delivers on a subscribed topic.
///
/// # Arguments
///
/// * `transport` - the transport to receive from.
/// * `out_message` - receives a message handle to release with
///   [`pamoja_message_free`], or null once the link has ended.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] with a message or with null once the link has ended, or
/// [`PamojaStatus::Closed`] if the transport is not connected.
///
/// # Safety
///
/// `transport` must be a live handle and `out_message` must be writable.
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_recv(
    transport: *mut PamojaTransport,
    out_message: *mut *mut PamojaMessage,
) -> PamojaStatus {
    let Some(transport) = transport_handle(transport) else {
        return PamojaStatus::InvalidArgument;
    };
    if out_message.is_null() {
        set_last_error("out_message must not be null".to_owned());
        return PamojaStatus::InvalidArgument;
    }
    *out_message = ptr::null_mut();
    match crate::runtime().block_on(transport.kind.recv()) {
        Ok(Some(message)) => {
            *out_message = PamojaMessage::into_raw(message.topic, message.payload);
            PamojaStatus::Ok
        }
        Ok(None) => PamojaStatus::Ok,
        Err(error) => status(Err(error)),
    }
}

/// Waits a limited time for the next message a transport delivers.
///
/// Running out of time loses nothing: a message that arrives afterwards waits for
/// the next receive. This is the call to use rather than abandoning a receive that
/// is still waiting, which would take that message instead.
///
/// # Arguments
///
/// * `transport` - the transport to receive from.
/// * `timeout_ms` - how long to wait, in milliseconds.
/// * `out_message` - receives a message handle to release with
///   [`pamoja_message_free`], or null when the time ran out or the link has ended.
/// * `out_timed_out` - receives whether the time ran out before a message arrived.
///
/// # Returns
///
/// [`PamojaStatus::Ok`] with a message, with the time run out, or with null once
/// the link has ended, or [`PamojaStatus::Closed`] if the transport is not
/// connected.
///
/// # Safety
///
/// `transport` must be a live handle, and `out_message` and `out_timed_out` must
/// be writable.
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_recv_within(
    transport: *mut PamojaTransport,
    timeout_ms: u64,
    out_message: *mut *mut PamojaMessage,
    out_timed_out: *mut bool,
) -> PamojaStatus {
    let Some(transport) = transport_handle(transport) else {
        return PamojaStatus::InvalidArgument;
    };
    receive_within(timeout_ms, out_message, out_timed_out, || {
        transport.kind.recv()
    })
}

/// Releases a transport handle.
///
/// Passing null is a no-op.
///
/// # Safety
///
/// `transport` must be a handle that has not already been freed or consumed by
/// a call that takes ownership, or null. After this call it must not be used
/// again.
#[no_mangle]
pub unsafe extern "C" fn pamoja_transport_free(transport: *mut PamojaTransport) {
    if !transport.is_null() {
        drop(Box::from_raw(transport));
    }
}

/// Borrows a transport handle, rejecting a null pointer.
///
/// # Safety
///
/// `transport` must be a live handle from a call that produced one, or null.
unsafe fn transport_handle<'a>(transport: *mut PamojaTransport) -> Option<&'a mut PamojaTransport> {
    if transport.is_null() {
        set_last_error("transport must not be null".to_owned());
        return None;
    }
    Some(&mut *transport)
}

/// Takes ownership of a transport handle, leaving the caller nothing to free.
///
/// # Safety
///
/// `transport` must be a live handle that has not been freed or consumed, or
/// null. After this call the caller must not use it again.
pub(crate) unsafe fn take_transport(transport: *mut PamojaTransport) -> Option<Kind> {
    if transport.is_null() {
        set_last_error("transport must not be null".to_owned());
        return None;
    }
    Some(Box::from_raw(transport).kind)
}

/// Runs a receive against a time limit and writes what came of it through the
/// out-parameters every timed receive in this crate shares.
///
/// A receive is cancel-safe, so the one dropped when the time runs out leaves the
/// next message queued rather than taking it.
///
/// # Safety
///
/// `out_message` and `out_timed_out` must be writable, or null to be refused.
pub(crate) unsafe fn receive_within<F>(
    timeout_ms: u64,
    out_message: *mut *mut PamojaMessage,
    out_timed_out: *mut bool,
    receive: impl FnOnce() -> F,
) -> PamojaStatus
where
    F: Future<Output = Result<Option<Message>>>,
{
    if out_message.is_null() || out_timed_out.is_null() {
        set_last_error("out_message and out_timed_out must not be null".to_owned());
        return PamojaStatus::InvalidArgument;
    }
    *out_message = ptr::null_mut();
    *out_timed_out = false;
    let limit = std::time::Duration::from_millis(timeout_ms);
    let receive = receive();
    match crate::runtime().block_on(async move { tokio::time::timeout(limit, receive).await }) {
        Err(_) => {
            *out_timed_out = true;
            PamojaStatus::Ok
        }
        Ok(Ok(Some(message))) => {
            *out_message = PamojaMessage::into_raw(message.topic, message.payload);
            PamojaStatus::Ok
        }
        Ok(Ok(None)) => PamojaStatus::Ok,
        Ok(Err(error)) => status(Err(error)),
    }
}

/// Maps a transport result onto a status, recording any failure.
pub(crate) fn status(result: Result<()>) -> PamojaStatus {
    match result {
        Ok(()) => PamojaStatus::Ok,
        Err(error) => {
            let status = PamojaStatus::from_error(&error);
            set_last_error(error.to_string());
            status
        }
    }
}