rskit-messaging 0.1.0-alpha.2

Message broker abstractions, registry, middleware, and in-memory adapter
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

use std::time::Duration;

use async_trait::async_trait;
use rskit_errors::{AppError, AppResult, ErrorCode};

use crate::event::Event;
use crate::message::Message;

fn unsupported(capability: &str) -> AppResult<()> {
    Err(AppError::new(
        ErrorCode::InvalidInput,
        format!("messaging capability {capability} is not supported by this consumer"),
    ))
}

// ── Messaging traits ─────────────────────────────────────────────────────────

/// A producer that sends messages to a broker.
#[async_trait]
pub trait MessageProducer<T: Send + Sync>: Send + Sync {
    /// Send a single message.
    async fn send(&self, msg: Message<T>) -> AppResult<()>;

    /// Send a batch of messages.
    async fn send_batch(&self, msgs: Vec<Message<T>>) -> AppResult<()>;

    /// Flush pending messages within the given timeout.
    async fn flush(&self, timeout: Duration) -> AppResult<()>;

    /// Close or shut down the producer. Implementations with no persistent resources may no-op.
    async fn close(&self) -> AppResult<()> {
        Ok(())
    }
}

/// A consumer that receives messages from a broker.
#[async_trait]
pub trait MessageConsumer<T: Send + Sync>: Send + Sync {
    /// Subscribe to one or more topics.
    async fn subscribe(&self, topics: &[&str]) -> AppResult<()>;

    /// Receive the next message. Blocks until a message is available.
    async fn recv(&self) -> AppResult<Message<T>>;

    /// Close or shut down the consumer. Implementations with no persistent resources may no-op.
    async fn close(&self) -> AppResult<()> {
        Ok(())
    }

    /// Pause message delivery when the adapter supports it.
    async fn pause(&self) -> AppResult<()> {
        unsupported("pause")
    }

    /// Resume paused message delivery when the adapter supports it.
    async fn resume(&self) -> AppResult<()> {
        unsupported("resume")
    }
}

/// A producer that publishes structured [`Event`]s to topics.
#[async_trait]
pub trait EventProducer: Send + Sync {
    /// Publish a single event to the given topic.
    async fn publish(&self, topic: &str, event: Event) -> AppResult<()>;

    /// Publish a batch of events to the given topic.
    async fn publish_batch(&self, topic: &str, events: Vec<Event>) -> AppResult<()>;
}

/// A consumer that receives structured [`Event`]s from topics.
#[async_trait]
pub trait EventConsumer: Send + Sync {
    /// Subscribe to one or more topics.
    async fn subscribe(&self, topics: &[&str]) -> AppResult<()>;

    /// Receive the next event. Blocks until an event is available.
    async fn recv_event(&self) -> AppResult<Event>;
}

// ── Broker lifecycle trait ───────────────────────────────────────────────────

/// Lifecycle management for a message-broker connection.
///
/// This is intentionally simpler than the bootstrap `Component` trait — it
/// captures the start / stop / health contract that every broker adapter
/// needs without pulling in the full component registry.  Implementations
/// can bridge to the bootstrap `Component` trait where needed.
#[async_trait]
pub trait BrokerComponent: Send + Sync {
    /// Establish the broker connection and perform any setup.
    async fn start(&self) -> AppResult<()>;
    /// Gracefully disconnect from the broker.
    async fn stop(&self) -> AppResult<()>;
    /// Instant health check — returns `true` when the broker connection is
    /// usable.
    fn is_healthy(&self) -> bool;
}