dittolive-ditto-core 4.8.0-experimental-rust.2

//! This file MUST be kept in sync with dittolive-ditto-sys/src/bus.rs under penalty of
//! ABI-incompatibility!
//!
//! This is the FFI for the Bus Streams API.

use safer_ffi::{
    derive_ReprC,
    dyn_traits::VirtualPtr,
    layout::{CLayoutOf, ReprC, __HasNiche__},
    option::TaggedOption,
};

use crate::peer_pubkey::PeerPubkey;

pub type Callback<T> = safer_ffi::closure::BoxDynFnMut1<(), T>;
pub type Continuation<'a, T> = safer_ffi::closure::BoxDynFnMut1<(), T>;

/// The Ditto Bus API uses [`Bytes`](safer_ffi::bytes::Bytes) to transfer data between
pub type Payload = safer_ffi::bytes::Bytes<'static>;

/// The Bus itself. It allows listening for and opening connections (and might setup a default
/// one for single messages too).
#[derive_ReprC]
#[repr(C)]
pub struct Bus {
    pub inner: VirtualPtr<dyn IBus + Send + Sync>,
}
/// The Bus itself. It allows listening for and opening connections (and might setup a default
/// one for single messages too).
#[derive_ReprC(dyn, Clone)]
pub trait IBus {
    /// Adds an acceptor to the acceptor queue.
    ///
    /// If the `topic` was already in use, `None` is returned.
    extern "C" fn add_acceptor(
        &self,
        topic: safer_ffi::bytes::Bytes<'static>,
        on_accept: Callback<Stream>,
    ) -> Option<AcceptorId>;

    /// Removes an acceptor from the acceptor queue, based on the ID returned when it was added.
    extern "C" fn delete_acceptor(&self, id: Option<AcceptorId>);

    /// Attempts to establish a connection, yielding a connection result once done.
    extern "C" fn connect(
        &self,
        with: StreamInfo<'_>,
        then: Continuation<'static, ConnectionResult>,
    );

    /// DEPRECATED
    /// Sends a message on the single_message channel.
    extern "C" fn send_single_message(
        &self,
        to: PeerPubkey,
        data: Payload,
        reliability: ReliabilityMode,
    ) -> SendHandle;

    /// DEPRECATED
    /// Replaces the current single message callback by the one provided.
    ///
    /// Replacing the callback by `None` is valid.
    ///
    /// The previous callback will be returned by this method.
    extern "C" fn set_recv_single_message_callback(
        &self,
        callback: TaggedOption<Callback<SingleMessage>>,
    ) -> TaggedOption<Callback<SingleMessage>>;
}

/// The identifier of an acceptor in the [`Bus`].
#[repr(transparent)]
pub struct AcceptorId(pub core::num::NonZeroU32);
unsafe impl ReprC for AcceptorId {
    type CLayout = u32;
    fn is_valid(it: &'_ Self::CLayout) -> bool {
        *it != 0
    }
}
unsafe impl __HasNiche__ for AcceptorId {
    fn is_niche(it: &'_ <Self as ReprC>::CLayout) -> bool {
        *it == 0
    }
}
#[test]
fn acceptor_id_layout() {
    assert!(AcceptorId::is_niche(&0));
    assert!(!AcceptorId::is_valid(&0));
    assert!(!AcceptorId::is_niche(&1));
    assert!(AcceptorId::is_valid(&1));
    assert!(!AcceptorId::is_niche(&u32::MAX));
    assert!(AcceptorId::is_valid(&u32::MAX));
}

/// DEPRECATED
/// Requested reliability level for a message to be transmitted to another peer.
#[derive_ReprC]
#[repr(u8)]
pub enum ReliabilityMode {
    /// No guarantees of successful delivery, ordering, or once-only delivery
    Unreliable,
    /// Messages will be delivered at most once, in the same order that they are sent,
    /// but there may be gaps.
    UnreliableSequenced,
    /// Every message will be delivered in order or else the connection fails
    Reliable,
}

/// DEPRECATED
/// A message received from the single message API.
#[derive_ReprC]
#[repr(C)]
pub struct SingleMessage {
    /// The source of the message.
    pub source: PeerPubkey,
    /// The payload of the message.
    pub payload: Payload,
}

/// An open, bidirectional stream.
#[derive_ReprC]
#[repr(C)]
#[must_use = "Dropping a stream closes the connection."]
pub struct Stream {
    pub inner: VirtualPtr<dyn IStream + Send + Sync>,
}
#[derive_ReprC(dyn)]
/// An open, bidirectional stream.
pub trait IStream {
    /// The stream's information.
    extern "C" fn info(&self) -> StreamInfo<'_>;

    /// Send a payload.
    ///
    /// The returned SendHandle can be used to cancel the send (provided it hasn't already been
    /// delivered). `finished` is called once the send is considered complete
    /// (immediately for unreliable streams, upon ACK otherwise). If the send failed
    /// (only allowed for cancelations), the payload
    extern "C" fn send(&self, payload: Payload) -> SendHandle;

    /// Replaces the current recv callback.
    ///
    /// This can also be used to remove the existing one or extract it to concatenate callbacks.
    extern "C" fn set_recv_callback(
        &self,
        on_recv: TaggedOption<Callback<Payload>>,
    ) -> TaggedOption<Callback<Payload>>;

    /// Returns the stream's closer if the stream was indeed closed.
    extern "C" fn is_closed(&self) -> TaggedOption<StreamClosedBy>;

    /// Adds a `continuation` on the stream's closure.
    ///
    /// All continuations added to the stream's closure will be called once it is closed.
    ///
    /// The `continuation` is guaranteed to be called if the stream was already closed.
    extern "C" fn add_on_close(&self, continuation: Continuation<'static, StreamClosedBy>);
}

/// A handle on an send operation, allowing to await its completion or attempt to cancel it.
///
/// If dropped, the handle MUST neither `cancel` nor await the operation, but simply detach.
#[derive_ReprC]
#[repr(transparent)]
pub struct SendHandle(pub VirtualPtr<dyn ISendHandle + Send + Sync>);
/// A handle on an send operation, allowing to await its completion or attempt to cancel it.
///
/// If dropped, the handle MUST neither `cancel` nor await the operation, but simply detach.
#[derive_ReprC(dyn, Clone)]
pub trait ISendHandle {
    /// Attempts to cancel the send operation, returning the payload of the message once the
    /// cancellation succeeds.
    ///
    /// This cancellation may fail to cancel the operation, even if the message is still in a
    /// send queue.
    ///
    /// If a continuation has been passed to `Self::then`, and hasn't been called yet, it _must_ be
    /// called with the current result of `Self::poll`.
    extern "C" fn cancel(&self) -> CancellationResult;

    /// Returns the current status of the operation.
    ///
    /// Note that consistency isn't guaranteed: calling `poll` after `then` may yield a
    /// different result than that yielded to the continuation. In such a case, the
    /// continuation's argument is considered "canon".
    extern "C" fn poll(&self) -> SendStatus;

    /// Calls the `continuation` when the send status changes.
    ///
    /// Previously set callbacks MUST:
    /// - Never be called upon a change in the [`SendStatus`] if `keep_previous == false` once this
    ///   method returns.
    /// - Also be called upon a change in the [`SendStatus`] if `keep_previous == true`, without any
    ///   call order being specified.
    extern "C" fn set_on_change(&self, callback: Callback<SendStatus>, keep_previous: bool);
}

/// The informations about a stream
#[derive_ReprC]
#[repr(transparent)]
pub struct StreamInfo<'a>(pub VirtualPtr<dyn IStreamInfo + 'a>);
/// The informations about a stream (or stream candidate)
#[derive_ReprC(dyn)]
pub trait IStreamInfo {
    /// The peer with which the stream is/will be connected.
    extern "C" fn peer_pubkey(&self) -> PeerPubkey;

    /// The topic of the stream.
    ///
    /// The return value _may_ alias `self`. You can ensure it's fully owned by calling
    /// [`Bytes::upgrade`](safer_ffi::bytes::Bytes::upgrade) which will copy only if necessary.
    extern "C" fn topic(&self) -> safer_ffi::bytes::Bytes<'_>;
}

/// Identifies whether the stream was closed locally or remotely.
#[derive_ReprC]
#[repr(u8)]
#[derive(Clone, Copy, PartialEq, Eq)]
pub enum StreamClosedBy {
    /// The stream was closed by the remote peer.
    Remote,
    /// The stream was closed by the locally.
    Local,
}

/// The result of attempting to connect.
#[repr(C, u8)]
pub enum ConnectionResult {
    /// Upon success, the stream is yielded.
    Accepted(Stream) = 1,
    /// Upon failure, the reason for that failureis yielded.
    Rejected(ConnectionError) = 0,
}

mod seal {
    use super::*;
    /// A safer-ffi friendly equivalent layout to [`ConnectionResult`].
    #[derive_ReprC]
    #[repr(C)]
    pub struct ConnectionResult_Layout {
        /// `true` if accepted, `false` if rejected.
        pub accepted: bool,
        /// Initialized iff `accepted` is `true`
        /// if not, must be a valid [`ConnectionError`] instead
        pub stream: Stream,
    }
    unsafe impl ReprC for ConnectionResult {
        type CLayout = CLayoutOf<ConnectionResult_Layout>;
        fn is_valid(it: &'_ Self::CLayout) -> bool {
            ConnectionResult_Layout::is_valid(it)
                || unsafe {
                    bool::is_valid(&it.accepted)
                        && !core::mem::transmute::<_, bool>(it.accepted)
                        && ConnectionError::is_valid(core::mem::transmute::<
                            &Stream_Layout,
                            &ConnectionError_Layout,
                        >(&it.stream))
                }
        }
    }
    #[test]
    fn connection_result_layout() {
        for expected in [
            ConnectionResult::Rejected(ConnectionError::PeerNotFound),
            ConnectionResult::Rejected(ConnectionError::TopicRejected),
        ] {
            assert!(ConnectionResult::is_valid(&unsafe {
                core::mem::transmute(expected)
            }))
        }
    }
    /// A safer-ffi friendly equivalent layout to [`CancellationResult`].
    #[derive_ReprC]
    #[repr(C)]
    pub struct CancellationResult_Layout {
        /// `true` if cancelling succeeded, `false` otherwise.
        pub success: bool,
        /// Initialized iff `accepted` is `true`,
        /// if not, must be a valid [`CancellationError`] instead
        pub error_reason: CancellationError,
    }
    unsafe impl ReprC for CancellationResult {
        type CLayout = CLayoutOf<CancellationResult_Layout>;
        fn is_valid(it: &'_ Self::CLayout) -> bool {
            CancellationResult_Layout::is_valid(it)
                || unsafe {
                    bool::is_valid(&it.success) && core::mem::transmute::<_, bool>(it.success)
                }
        }
    }
    #[test]
    fn cancellation_result_layout() {
        for expected in [
            CancellationResult::Ok,
            CancellationResult::Err(CancellationError::CancellationFailed),
        ] {
            assert!(CancellationResult::is_valid(&unsafe {
                core::mem::transmute(expected)
            }))
        }
        assert!(CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[bool; 2], CancellationResult_Layout_Layout>([false, false])
        }));
        assert!(CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[bool; 2], CancellationResult_Layout_Layout>([true, false])
        }));
        assert!(!CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[u8; 2], CancellationResult_Layout_Layout>([2, 0])
        }));
        assert!(!CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[u8; 2], CancellationResult_Layout_Layout>([2, 1])
        }));
        assert!(!CancellationResult::is_valid(&unsafe {
            core::mem::transmute::<[u8; 2], CancellationResult_Layout_Layout>([0, 1])
        }));
    }
}

/// The reason for a connection attempt's failure.
#[derive_ReprC]
#[repr(u8)]
#[derive(thiserror::Error, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Clone, Copy)]
pub enum ConnectionError {
    /// The peer you attempted to connect to was not found or failed to respond.
    #[error("The peer you attempted to connect to was not found or failed to respond in time.")]
    PeerNotFound,
    /// The peer you attempted to connect to was found, but refused your topic.
    #[error("The peer you attempted to connect to was found, but refused your topic.")]
    TopicRejected,
}

/// The reason for a connection attempt's failure.
#[derive_ReprC]
#[repr(u8)]
#[derive(thiserror::Error, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, Clone, Copy)]
pub enum CancellationError {
    /// Cancelling the send operation failed.
    #[error("Cancelling the send operation failed.")]
    CancellationFailed,
}
/// An FFI-safe `Result<Payload, CancellationError>`
#[repr(C, u8)]
pub enum CancellationResult {
    /// Cancelling the send operation was successful.
    ///
    /// The send operation's payload is returned if possible.
    Ok = 1,
    /// Cancelling the send operation was unsuccessful.
    Err(CancellationError) = 0,
}
impl From<Result<(), CancellationError>> for CancellationResult {
    fn from(value: Result<(), CancellationError>) -> Self {
        match value {
            Ok(()) => Self::Ok,
            Err(error) => Self::Err(error),
        }
    }
}
impl From<CancellationResult> for Result<(), CancellationError> {
    fn from(value: CancellationResult) -> Self {
        match value {
            CancellationResult::Ok => Ok(()),
            CancellationResult::Err(error) => Err(error),
        }
    }
}

/// Indicates the status of a given send operation.
///
/// A [`SendHandle`]'s state may become [`Unknown`](SendStatus::Unknown) at any time,
/// but may never return to another state after that.
///
/// Transition details:
/// - `Pending -> Sent`: Trivial, each state is technically allowed to represent the other.
/// - `Pending -> Failed`: This likely indicates that the stream has been closed before the message
///   could be sent.
#[derive_ReprC]
#[derive(Copy, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
#[repr(u8)]
pub enum SendStatus {
    /// The send status couldn't be recovered.
    ///
    /// Transition to this state may happen from any state (including "final" states),
    /// as the status may stop being tracked.
    Unknown = 0,
    /// The send operation is still pending.
    ///
    /// Note this doesn't guarantee that the message _hasn't_ been sent or acknowledged yet,
    /// just that the handle wasn't able to confirm that either of these states has been
    /// reached.
    Pending = 1,
    /// The send operation has been executed, but no acknowledgement has been received by the
    /// handle.
    ///
    /// Note that an acknowledgement might still be on the way, or that the handle is still
    /// expected to switch to [`SentNeverAck`](SendStatus::SentNeverAck) if no acknowledgement
    /// is expected.
    Sent = 2,
    /// The send was identified as having failed.
    ///
    /// This typically would happen if the target disconnected before reaching one of the other end
    /// states.
    Failed = 3,
    /// The send was cancelled.
    ///
    /// Note that this does not guarantee that the message won't be delivered, even if it hadn't
    /// been sent yet.
    Cancelled = 4,
}

impl ISendHandle for SendStatus {
    extern "C" fn cancel(&self) -> CancellationResult {
        CancellationResult::Err(CancellationError::CancellationFailed)
    }
    extern "C" fn poll(&self) -> SendStatus {
        *self
    }
    extern "C" fn set_on_change(
        &self,
        mut continuation: Callback<SendStatus>,
        _keep_previous: bool,
    ) {
        continuation.call(*self)
    }
}

impl core::fmt::Debug for SendStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match self {
            SendStatus::Unknown => "SendStatus::Unknown",
            SendStatus::Pending => "SendStatus::Pending",
            SendStatus::Sent => "SendStatus::Sent",
            SendStatus::Failed => "SendStatus::Failed",
            SendStatus::Cancelled => "SendStatus::Cancelled",
        };
        f.write_str(s)
    }
}
impl core::fmt::Display for SendStatus {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let s = match self {
            SendStatus::Unknown => "Unknown",
            SendStatus::Pending => "Pending",
            SendStatus::Sent => "Sent",
            SendStatus::Failed => "Failed",
            SendStatus::Cancelled => "Cancelled",
        };
        f.write_str(s)
    }
}