anycms_event/
transport.rs

1//! Transport abstraction for distributed event bus.
2//!
3//! Allows plugging in different transport backends (Redis, Kafka, NATS, etc.)
4//!
5//! # Example
6//!
7//! ```ignore
8//! use anycms_event::transport::Transport;
9//!
10//! // Implement Transport for your backend
11//! struct KafkaTransport { /* ... */ }
12//!
13//! impl Transport for KafkaTransport {
14//!     fn publish(&self, event_name: &str, payload: &str) -> TransportFuture<'_> {
15//!         Box::pin(async move { /* publish to Kafka */ })
16//!     }
17//!     fn clone_box(&self) -> Box<dyn Transport> {
18//!         Box::new(self.clone())
19//!     }
20//! }
21//! ```
22
23use std::fmt::Debug;
24use std::future::Future;
25use std::pin::Pin;
26use std::sync::Arc;
27
28/// Type alias for the boxed future returned by Transport methods.
29pub type TransportFuture<'a> = Pin<Box<dyn Future<Output = Result<(), TransportError>> + Send + 'a>>;
30
31/// A message received from the transport layer.
32#[derive(Debug, Clone)]
33pub struct TransportMessage {
34    /// Event name / type identifier.
35    pub event_name: String,
36    /// Serialized event payload (JSON string).
37    pub payload: String,
38}
39
40/// Callback invoked when a message is received from the transport.
41pub type TransportMessageCallback = Arc<dyn Fn(TransportMessage) + Send + Sync>;
42
43/// Handle to an active subscription on a transport backend.
44///
45/// Implement this to provide lifecycle control over subscriptions.
46pub trait TransportSubscription: Send + Sync {
47    /// Stop receiving messages and clean up resources.
48    fn stop(&self);
49    /// Check if the subscription is still active.
50    fn is_active(&self) -> bool;
51}
52
53/// Error type for transport operations.
54#[derive(Debug, thiserror::Error)]
55pub enum TransportError {
56    /// Failed to connect to the transport backend.
57    #[error("Connection error: {0}")]
58    Connection(String),
59
60    /// Failed to publish a message.
61    #[error("Publish error: {0}")]
62    Publish(String),
63
64    /// Failed to subscribe to a channel.
65    #[error("Subscribe error: {0}")]
66    Subscribe(String),
67
68    /// Failed to serialize/deserialize a message.
69    #[error("Serialization error: {0}")]
70    Serialization(String),
71}
72
73/// Abstraction for transport backends that bridge EventBus instances across processes.
74///
75/// Implement this trait to add support for new messaging backends (Kafka, NATS, etc.)
76///
77/// # Trait Object Safety
78///
79/// This trait is object-safe and can be used as `dyn Transport` or `Box<dyn Transport>`.
80pub trait Transport: Send + Sync {
81    /// Publish a serialized event payload to the transport.
82    ///
83    /// # Arguments
84    /// * `event_name` - The event type identifier, used for routing
85    /// * `payload` - Pre-serialized JSON string of the event
86    fn publish(&self, event_name: &str, payload: &str) -> TransportFuture<'_>;
87
88    /// Subscribe to events matching `event_pattern` from the transport.
89    ///
90    /// The `callback` is invoked for each received message. Returns a
91    /// [`TransportSubscription`] handle for lifecycle control.
92    fn subscribe(
93        &self,
94        event_pattern: &str,
95        callback: TransportMessageCallback,
96    ) -> Result<Box<dyn TransportSubscription>, TransportError>;
97
98    /// Clone this transport into a boxed trait object.
99    ///
100    /// Required because trait objects cannot implement `Clone` directly.
101    /// Implement as `Box::new(self.clone())`.
102    fn clone_box(&self) -> Box<dyn Transport>;
103}
104
105/// Handle to a background forwarder task.
106///
107/// Implementors should provide a way to stop the forwarder and check its status.
108pub trait ForwarderHandle: Send + Sync {
109    /// Stop the forwarder task.
110    fn stop(&self);
111
112    /// Check if the forwarder has finished.
113    fn is_finished(&self) -> bool;
114}
115
116impl Debug for Box<dyn Transport> {
117    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
118        f.debug_struct("Transport").finish()
119    }
120}
121
122impl Debug for Box<dyn TransportSubscription> {
123    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
124        f.debug_struct("TransportSubscription").finish()
125    }
126}
anycms_event/transport.rs

anycms_event/
transport.rs
