alopex_chirps_core/
backend.rs

1use crate::error::TransportError;
2use alopex_chirps_wire::frame::Frame;
3use alopex_chirps_wire::node_id::NodeId;
4use async_trait::async_trait;
5use serde::{Deserialize, Serialize};
6use std::net::SocketAddr;
7use tokio::sync::mpsc;
8
9/// Delivery semantics requested from a message backend.
10#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
11pub enum BackendProfile {
12    Control,
13    Ephemeral,
14    Durable,
15}
16
17/// Profiles a backend can implement without fallback.
18#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
19pub struct BackendCapabilities {
20    pub control: bool,
21    pub ephemeral: bool,
22    pub durable: bool,
23}
24
25impl Default for BackendCapabilities {
26    fn default() -> Self {
27        Self {
28            control: true,
29            ephemeral: true,
30            durable: false,
31        }
32    }
33}
34
35impl BackendCapabilities {
36    pub fn supports(self, profile: BackendProfile) -> bool {
37        match profile {
38            BackendProfile::Control => self.control,
39            BackendProfile::Ephemeral => self.ephemeral,
40            BackendProfile::Durable => self.durable,
41        }
42    }
43}
44
45/// Additive metadata reserved for durable acknowledgement/replay backends.
46#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
47pub struct EnvelopeMetadata {
48    pub message_id: Option<[u8; 16]>,
49    pub sequence: Option<u64>,
50    pub partition: Option<u64>,
51    pub acknowledgement: Option<u64>,
52    pub replay: bool,
53    pub checkpoint: Option<u64>,
54    pub offset: Option<u64>,
55}
56
57/// Abstraction over transport backends (QUIC, mock, etc.).
58#[async_trait]
59pub trait MessageBackend: Send + Sync {
60    /// Reports profile support. Durable is opt-in and false by default.
61    fn capabilities(&self) -> BackendCapabilities {
62        BackendCapabilities::default()
63    }
64
65    /// Sends a message to a specific target node.
66    async fn send(&self, target: NodeId, frame: Frame) -> Result<(), TransportError>;
67
68    /// Profile-aware extension point. Durable backends must override this method;
69    /// the default implementation never falls back to the control path.
70    async fn send_with_profile(
71        &self,
72        target: NodeId,
73        frame: Frame,
74        profile: BackendProfile,
75        _metadata: EnvelopeMetadata,
76    ) -> Result<(), TransportError> {
77        if !self.capabilities().supports(profile) || profile == BackendProfile::Durable {
78            return Err(TransportError::NotImplemented(
79                "requested message profile is not supported by this backend",
80            ));
81        }
82        self.send(target, frame).await
83    }
84
85    /// Broadcasts a message to all connected peers.
86    /// Returns the number of peers the message was sent to.
87    async fn broadcast(&self, frame: Frame) -> Result<usize, TransportError>;
88
89    /// Broadcast counterpart of `send_with_profile`.
90    async fn broadcast_with_profile(
91        &self,
92        frame: Frame,
93        profile: BackendProfile,
94        _metadata: EnvelopeMetadata,
95    ) -> Result<usize, TransportError> {
96        if !self.capabilities().supports(profile) || profile == BackendProfile::Durable {
97            return Err(TransportError::NotImplemented(
98                "requested message profile is not supported by this backend",
99            ));
100        }
101        self.broadcast(frame).await
102    }
103
104    /// Subscribes to incoming messages.
105    ///
106    /// Returns a channel receiver that will be sent tuples of `(sender_node_id, message)`.
107    async fn subscribe(&self) -> Result<mpsc::Receiver<(NodeId, Frame)>, TransportError>;
108
109    /// Closes the transport.
110    async fn close(&self) -> Result<(), TransportError>;
111
112    /// Returns a list of currently connected peers.
113    fn connected_peers(&self) -> Vec<(NodeId, SocketAddr)>;
114}
alopex_chirps_core/backend.rs

alopex_chirps_core/
backend.rs
