Struct ChannelEventBus 

pub struct ChannelEventBus {
    sender: UnboundedSender<EngineMessage>,
    receiver: Arc<RwLock<Option<UnboundedReceiver<EngineMessage>>>>,
    subscribers: Arc<RwLock<HashMap<String, Vec<UnboundedSender<EngineMessage>>>>>,
    subscribers_with_offsets: Arc<RwLock<HashMap<String, Vec<UnboundedSender<LegacyMessageEnvelope>>>>>,
    subscribers_with_sequence: Arc<RwLock<HashMap<String, Vec<UnboundedSender<MessageEnvelope>>>>>,
    sequence: Arc<AtomicU64>,
    replay_error: Arc<RwLock<Option<String>>>,
}

In-process channel-based event bus.

Fields

sender: UnboundedSender<EngineMessage>

receiver: Arc<RwLock<Option<UnboundedReceiver<EngineMessage>>>>

subscribers: Arc<RwLock<HashMap<String, Vec<UnboundedSender<EngineMessage>>>>>

subscribers_with_offsets: Arc<RwLock<HashMap<String, Vec<UnboundedSender<LegacyMessageEnvelope>>>>>

subscribers_with_sequence: Arc<RwLock<HashMap<String, Vec<UnboundedSender<MessageEnvelope>>>>>

sequence: Arc<AtomicU64>

Monotonic sequence counter, incremented on each event dispatch.

replay_error: Arc<RwLock<Option<String>>>

When set, replay_from_offsets returns this error instead of Ok. Used to test graceful handling of replay failures.

Implementations

impl ChannelEventBus

pub fn new() -> Result<Self, String>

pub async fn clear_subscribers(&self)

Clear all subscribers, simulating a SIGKILL that severs the event bus pipeline. After this call, events sent via get_sender() will still be accepted but silently dropped (no subscriber receives them).

pub async fn set_replay_error(&self, error: Option<String>)

Configure replay_from_offsets to return an error, simulating replay failures.

fn get_topic_for_event(event: &EngineMessage) -> &'static str

Trait Implementations

impl EventBusTrait for ChannelEventBus

fn get_sender(&self) -> UnboundedSender<EngineMessage>

Get a sender to publish events

fn subscribe<'life0, 'async_trait>( &'life0 self, topics: Vec<String>, ) -> Pin<Box<dyn Future<Output = Result<UnboundedReceiver<EngineMessage>, String>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to specific topics and get a receiver for events

fn subscribe_with_offsets<'life0, 'async_trait>( &'life0 self, topics: Vec<String>, ) -> Pin<Box<dyn Future<Output = Result<UnboundedReceiver<LegacyMessageEnvelope>, String>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to specific topics and get a receiver for events WITH legacy offset metadata. Use this when you need to track exactly which offsets you’ve processed (for snapshots).

fn subscribe_with_sequence<'life0, 'async_trait>( &'life0 self, topics: Vec<String>, ) -> Pin<Box<dyn Future<Output = Result<UnboundedReceiver<MessageEnvelope>, String>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to specific topics and get a receiver for events WITH sequence metadata. Use this when you need monotonic ordering.

fn start_processing<'async_trait>( self: Arc<Self>, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait,

Start processing events (forwarding between channels)

fn replay_from_offsets<'life0, 'async_trait>( &'life0 self, offsets: TopicOffsets, _handler: ReplayHandler, ) -> Pin<Box<dyn Future<Output = Result<TopicOffsets, String>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Replay events from specific offsets until caught up to the current head.

fn get_current_offsets(&self) -> TopicOffsets

Get the current consumer offsets for all topics.

fn get_current_sequence(&self) -> u64

Get the current sequence number (highest dispatched).