ruststream/
lib.rs

1//! Rust core of the [`RustStream`](https://github.com/powersemmi/ruststream) messaging
2//! framework: broker-agnostic traits, message types, codecs, router runtime, and a
3//! conformance harness for broker authors.
4//!
5//! # Cargo features
6//!
7//! The core traits, the [`runtime::RustStream`] application object, middleware, and dispatch are
8//! always present. The rest is additive and opt-in. Codec features are mutually compatible and
9//! enable only the deserializers you need.
10//!
11//! * `json` (default): [`codec::JsonCodec`].
12//! * `msgpack`: [`codec::MsgpackCodec`].
13//! * `cbor`: [`codec::CborCodec`].
14//! * `memory`: [`memory::MemoryBroker`], an in-process broker usable in applications, prototypes
15//!   and tests.
16//! * `macros`: the `#[subscriber]`, [`#[ruststream::app]`](macro@app), and
17//!   [`#[derive(Message)]`](macro@Message) macros.
18//! * `asyncapi`: `AsyncAPI` document generation and the HTML viewer.
19//! * `metrics`: Prometheus metrics middleware and exporter.
20//! * `logging`: colored, `RUST_LOG`-driven console logging via `tracing-subscriber`
21//!   ([`logging::init`]). The generated `cli` `run` command installs it automatically.
22//! * `otel`: OpenTelemetry SDK integration: OTLP export for traces and metrics via
23//!   [`otel::OtelBuilder::init`], plus per-handler dispatch metrics middleware and W3C
24//!   trace-context propagation ([`otel::propagation`]).
25//! * `conformance`: the [`conformance::harness`] contract suite, per-capability suites in
26//!   [`conformance::capabilities`], and broker-agnostic [`conformance::helpers`] for application
27//!   tests. Generic over any broker's [`testing::TestableBroker`], so it pulls in no concrete broker
28//!   (enable `memory` too to run it against [`memory::MemoryBroker`]).
29//! * `testing`: the [`testing::TestApp`] in-process harness for application unit tests.
30//! * `cli`: the `ruststream` binary (`run`, `asyncapi gen`, `new`).
31//!
32//! Disable defaults (`default-features = false`) to drop the bundled JSON codec; the core traits,
33//! runtime, and dispatch remain. Add back only what you need.
34
35#![forbid(unsafe_code)]
36
37mod broker;
38mod buffered;
39mod capability;
40mod error;
41mod field;
42mod headers;
43mod message;
44mod publisher;
45mod schema;
46mod subscriber;
47mod subscription;
48pub mod testing;
49
50/// Re-exported for the [`register_testable_broker!`] macro's expansion; not a stable API.
51#[cfg(feature = "testing")]
52#[doc(hidden)]
53pub use inventory;
54
55pub use broker::{Broker, Connected, ConnectedBroker};
56pub use buffered::{Buffered, BufferedSubscriber};
57pub use capability::{
58    ApiKeyLocation, BatchSubscriber, DescribeServer, HttpApiKeyLocation, OwnedTransactions,
59    Partitioned, Positioned, RequestReply, SecurityScheme, Seekable, Seeker, ServerSpec, Subscribe,
60    Transaction, TransactionalPublisher,
61};
62pub use error::AckError;
63pub use field::{BuildContext, ContextField, Field, FieldMut};
64pub use headers::Headers;
65pub use message::{IncomingMessage, OutgoingMessage, RawMessage};
66pub use publisher::{DefaultPublish, PairError, PublishPolicy, Publisher};
67pub use schema::Message;
68pub use subscriber::Subscriber;
69pub use subscription::{
70    Name, SeekerPendingError, SeekerToken, StartAt, SubscriptionSource, WithSeeker,
71};
72
73pub mod codec;
74
75#[cfg(feature = "memory")]
76pub mod memory;
77
78pub mod runtime;
79
80pub use runtime::RustStream;
81
82/// Attribute macro that turns an `async fn` into a mountable subscriber definition.
83///
84/// Available with the `macros` feature. See [`ruststream_macros::subscriber`].
85#[cfg(feature = "macros")]
86pub use ruststream_macros::subscriber;
87
88/// Attribute macro that generates a `main` entry point from a `RustStream` builder function.
89///
90/// Available with the `macros` feature. See [`ruststream_macros::app`] and
91/// [`runtime::cli`].
92#[cfg(feature = "macros")]
93pub use ruststream_macros::app;
94
95/// Derive macro for [`Message`] metadata (type name + doc description).
96///
97/// Available with the `macros` feature.
98#[cfg(feature = "macros")]
99pub use ruststream_macros::Message;
100
101/// Derive macro that implements [`FromRef`](runtime::FromRef) for each field of an application
102/// state, so handlers can inject any field with [`State<T>`](runtime::State).
103///
104/// Available with the `macros` feature.
105#[cfg(feature = "macros")]
106pub use ruststream_macros::FromRef;
107
108#[cfg(feature = "conformance")]
109pub mod conformance;
110
111#[cfg(feature = "asyncapi")]
112pub mod asyncapi;
113
114/// Re-export of [`schemars`] so message types can derive `JsonSchema` without a direct dependency.
115///
116/// Derive it on a message type (`#[derive(ruststream::schemars::JsonSchema)]`) and its payload
117/// schema is emitted into the generated [`AsyncAPI`](asyncapi) document. Available with the
118/// `asyncapi` feature.
119#[cfg(feature = "asyncapi")]
120pub use schemars;
121
122#[cfg(feature = "metrics")]
123pub mod metrics;
124
125#[cfg(feature = "logging")]
126pub mod logging;
127
128#[cfg(feature = "otel")]
129pub mod otel;
130
131/// Implementation detail used by the `#[subscriber]` macro to capture a payload's JSON Schema.
132///
133/// Not part of the public API; no stability guarantees.
134#[doc(hidden)]
135pub mod __private {
136    use core::marker::PhantomData;
137
138    /// A type-carrying probe the macro reads a payload schema off.
139    ///
140    /// Schema selection uses inherent-vs-trait specialization (a stable-Rust trick): the schema
141    /// path is an inherent method on `Probe<T>` bounded by `T: JsonSchema`, and
142    /// [`NoSchemaProbe::schema_json`] is the trait fallback. Inherent methods win when present, so
143    /// `Probe::<T>::new().schema_json()` returns the schema for a concrete `T: JsonSchema` and
144    /// `None` otherwise - without forcing the bound onto every message type. The inherent method
145    /// exists only with the `asyncapi` feature.
146    #[derive(Debug)]
147    pub struct Probe<T>(pub PhantomData<T>);
148
149    impl<T> Probe<T> {
150        /// Constructs a probe for `T`.
151        #[must_use]
152        pub const fn new() -> Self {
153            Self(PhantomData)
154        }
155    }
156
157    impl<T> Default for Probe<T> {
158        fn default() -> Self {
159            Self::new()
160        }
161    }
162
163    /// The trait fallback: chosen for any `T` the inherent schema method does not cover.
164    pub trait NoSchemaProbe {
165        /// Returns `None` (no schema available for the probed type).
166        fn schema_json(&self) -> Option<String>;
167    }
168
169    impl<T> NoSchemaProbe for Probe<T> {
170        fn schema_json(&self) -> Option<String> {
171            None
172        }
173    }
174
175    #[cfg(feature = "asyncapi")]
176    impl<T: schemars::JsonSchema> Probe<T> {
177        /// Returns the serialized JSON Schema for `T` (inherent; preferred over the trait fallback).
178        #[must_use]
179        pub fn schema_json(&self) -> Option<String> {
180            serde_json::to_string(&schemars::schema_for!(T)).ok()
181        }
182    }
183
184    /// The trait fallback for [`Message`](crate::Message) metadata: chosen for any `T` the
185    /// inherent methods below do not cover.
186    pub trait NoMessageProbe {
187        /// Returns `None` (the probed type does not implement `Message`).
188        fn message_name(&self) -> Option<&'static str>;
189        /// Returns `None` (the probed type does not implement `Message`).
190        fn message_description(&self) -> Option<&'static str>;
191    }
192
193    impl<T> NoMessageProbe for Probe<T> {
194        fn message_name(&self) -> Option<&'static str> {
195            None
196        }
197
198        fn message_description(&self) -> Option<&'static str> {
199            None
200        }
201    }
202
203    impl<T: crate::Message> Probe<T> {
204        /// Returns [`Message::NAME`](crate::Message::NAME) for `T` (inherent; preferred over the
205        /// trait fallback).
206        #[must_use]
207        pub fn message_name(&self) -> Option<&'static str> {
208            Some(T::NAME)
209        }
210
211        /// Returns [`Message::DESCRIPTION`](crate::Message::DESCRIPTION) for `T` (inherent;
212        /// preferred over the trait fallback).
213        #[must_use]
214        pub fn message_description(&self) -> Option<&'static str> {
215            T::DESCRIPTION
216        }
217    }
218}
219
220/// Builds a [`NonZero`](core::num::NonZero) integer from a literal, rejecting zero at compile
221/// time.
222///
223/// The expansion is an inline `const` block, so `nonzero!(0)` fails the build instead of
224/// panicking at runtime, and the `NonZero` width is inferred from the call site - the same
225/// literal works for [`Buffered::max_size`](crate::Buffered::max_size) (`NonZeroUsize`) and any
226/// other `NonZero` parameter.
227///
228/// # Examples
229///
230/// ```
231/// use ruststream::{Buffered, Name, nonzero};
232///
233/// let source = Buffered::new(Name::new("orders")).max_size(nonzero!(128));
234/// # let _ = source;
235/// ```
236///
237/// Zero does not compile:
238///
239/// ```compile_fail
240/// let _: core::num::NonZeroUsize = ruststream::nonzero!(0);
241/// ```
242#[macro_export]
243macro_rules! nonzero {
244    ($value:expr) => {
245        const {
246            match ::core::num::NonZero::new($value) {
247                ::core::option::Option::Some(value) => value,
248                ::core::option::Option::None => panic!("nonzero!(..) requires a non-zero value"),
249            }
250        }
251    };
252}
ruststream/lib.rs

ruststream/
lib.rs
