etdl_core/
publisher.rs

1//! Publisher abstraction for generated ETDL handlers.
2//!
3//! Generated code emits Consequence `send` operations as calls on a
4//! [`Publisher`] supplied by the caller. This keeps generated handlers free of
5//! any concrete transport (Kafka, NATS, HTTP, in-memory, ...) so they remain
6//! pure, deterministic, and testable — while the application wires a real
7//! transport at the boundary.
8//!
9//! The reference implementation ships [`NoopPublisher`] (discards with a log) and
10//! [`ChannelCapturingPublisher`] (records `(channel, payload)` pairs for tests).
11//! Applications implement [`Publisher`] for their own infrastructure and, per
12//! ETDL §9.2, SHOULD inject the W3C `traceparent` (see
13//! [`crate::telemetry::inject_traceparent`]) into every outbound message.
14
15use std::sync::{Arc, Mutex};
16
17/// An error produced while publishing a message to a channel.
18#[derive(Debug, Clone)]
19pub struct PublishError(pub String);
20
21impl std::fmt::Display for PublishError {
22    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
23        write!(f, "{}", self.0)
24    }
25}
26
27impl std::error::Error for PublishError {}
28
29impl From<String> for PublishError {
30    fn from(s: String) -> Self {
31        PublishError(s)
32    }
33}
34
35/// A transport-agnostic channel publisher.
36///
37/// `payload` is the message serialized to a JSON [`serde_json::Value`]. The
38/// concrete AsyncAPI message type is serialized by generated code before the
39/// call so the trait stays free of generics and remains object-safe.
40pub trait Publisher: Send + Sync {
41    /// Publish `payload` to `channel`.
42    fn publish(&self, channel: &str, payload: &serde_json::Value) -> Result<(), PublishError>;
43}
44
45/// A [`Publisher`] that logs and discards every message.
46///
47/// Useful as a default in tests or during bring-up when no transport exists yet.
48#[derive(Debug, Default, Clone)]
49pub struct NoopPublisher;
50
51impl Publisher for NoopPublisher {
52    fn publish(&self, channel: &str, payload: &serde_json::Value) -> Result<(), PublishError> {
53        eprintln!(
54            "[etdl.publisher] noop: channel={} payload={}",
55            channel, payload
56        );
57        Ok(())
58    }
59}
60
61/// A [`Publisher`] that records `(channel, payload)` pairs for assertions.
62#[derive(Debug, Default, Clone)]
63pub struct ChannelCapturingPublisher {
64    sent: Arc<Mutex<Vec<(String, serde_json::Value)>>>,
65}
66
67impl ChannelCapturingPublisher {
68    /// Create a new empty capturing publisher.
69    pub fn new() -> Self {
70        Self::default()
71    }
72
73    /// The recorded `(channel, payload)` pairs in publish order.
74    pub fn sent(&self) -> Vec<(String, serde_json::Value)> {
75        self.sent.lock().map(|g| g.clone()).unwrap_or_default()
76    }
77
78    /// True if any message was published to `channel`.
79    pub fn published_to(&self, channel: &str) -> bool {
80        self.sent().iter().any(|(c, _)| c == channel)
81    }
82}
83
84impl Publisher for ChannelCapturingPublisher {
85    fn publish(&self, channel: &str, payload: &serde_json::Value) -> Result<(), PublishError> {
86        if let Ok(mut g) = self.sent.lock() {
87            g.push((channel.to_string(), payload.clone()));
88        }
89        Ok(())
90    }
91}
92
93#[cfg(test)]
94mod tests {
95    use super::*;
96
97    #[test]
98    fn noop_publisher_accepts_all() {
99        let p = NoopPublisher;
100        assert!(p.publish("ch", &serde_json::json!({"a": 1})).is_ok());
101    }
102
103    #[test]
104    fn capturing_publisher_records_ordered() {
105        let p = ChannelCapturingPublisher::new();
106        p.publish("a", &serde_json::json!(1)).unwrap();
107        p.publish("b", &serde_json::json!({"k": "v"})).unwrap();
108        let sent = p.sent();
109        assert_eq!(sent.len(), 2);
110        assert_eq!(sent[0].0, "a");
111        assert_eq!(sent[1].0, "b");
112        assert!(p.published_to("b"));
113        assert!(!p.published_to("c"));
114    }
115}
etdl_core/publisher.rs

etdl_core/
publisher.rs
