ruststream_macros/lib.rs
1//! Procedural macros for [RustStream](https://github.com/powersemmi/ruststream).
2//!
3//! Re-exported from the `ruststream` crate under the `macros` feature; depend on that rather than
4//! on this crate directly.
5
6#![forbid(unsafe_code)]
7
8mod expand;
9mod from_ref;
10mod parse;
11
12use proc_macro::TokenStream;
13use proc_macro2::TokenStream as TokenStream2;
14use quote::quote;
15use syn::{DeriveInput, ItemFn, parse_macro_input};
16
17use parse::{SubscriberArgs, doc_description};
18
19/// Turns an `async fn` handler into a mountable subscriber definition.
20///
21/// ```ignore
22/// /// Processes incoming orders.
23/// #[subscriber("orders")]
24/// async fn handle(order: &Order) -> HandlerResult { HandlerResult::Ack }
25/// // later: broker_scope.include(handle);
26///
27/// // reply form: the return value is encoded and published to "responses" through the
28/// // TypedPublisher (broker + reply codec) passed at wiring time.
29/// #[subscriber("requests", publish("responses"))]
30/// async fn reply(req: &Request) -> Response { /* ... */ }
31/// // later: broker_scope.include_publishing(reply, typed_publisher);
32///
33/// // reply form with explicit ack control: `Ok` publishes the reply, `Err` skips it and the
34/// // dispatcher acts on the returned HandlerResult.
35/// #[subscriber("requests", publish("responses"))]
36/// async fn confirm(req: &Request) -> Result<Response, HandlerResult> { /* ... */ }
37///
38/// // batch form: the handler takes the whole decoded batch as a slice; the source's
39/// // subscriber must implement BatchSubscriber. Mounted with include_batch.
40/// #[subscriber(batch("orders"))]
41/// async fn bill(orders: &[Order]) -> HandlerResult { /* settles the whole batch */ }
42/// ```
43///
44/// Without `publish(..)` the handler returns any `Into<Settle>` (a `Settle`, a `HandlerResult`,
45/// `()`, or `Result<_, E>`). Attach a post-settle continuation with `HandlerResult::ack().and_after`
46/// (any outcome works), which runs after the message is settled. With `publish(..)` it returns the
47/// reply value to publish, or `Result<Reply, HandlerResult>` to control acknowledgement:
48/// `Err(result)` publishes nothing and returns `result` to the dispatcher. The `Result` form is
49/// detected syntactically, so spell it out in the signature (a type alias is treated as a plain
50/// reply type).
51///
52/// Wrapping the source in `batch(..)` switches the definition to a `BatchDef`: the handler takes
53/// `&[T]` and runs once per batch pulled from the broker's `BatchSubscriber` (use the `Buffered`
54/// adapter for brokers without native batching). It returns any `IntoBatchResult` - one outcome
55/// for the whole batch (`HandlerResult`, `()`, `Result<_, E>`), or a per-element vector
56/// (`Vec<Settle>`, or `Vec<HandlerResult>`) to settle element `i` of the slice with outcome `i`,
57/// each element carrying its own optional `and_after` continuation. The source type is recovered
58/// from the constructor path, so a generic source spells its parameters:
59/// `batch(Buffered::<Name>::new(Name::new("orders")))`.
60///
61/// Combining `batch(..)` with `publish(..)` produces a `BatchPublishingDef` (mounted with
62/// `include_batch_publishing`): the handler returns `Vec<Reply>` (or
63/// `Result<Vec<Reply>, HandlerResult>` for explicit ack control, all-or-nothing - selective
64/// outcomes do not compose with a transaction), every reply is published to the reply name, and
65/// the whole batch is acked after. Hand the mount a `TypedPublisher` for independent reply
66/// publishes, or `.transactional()` for one transaction per batch.
67///
68/// A `workers(n)` clause processes up to `n` deliveries (or batches) of this subscriber
69/// concurrently, each in its own task; global processing order is lost by design, and
70/// back-pressure holds at `n` in-flight deliveries. `workers(n, by_key)` switches to `n`
71/// sequential lanes keyed by the message's partition key, preserving per-key ordering
72/// (single-message forms only). The default is the sequential loop.
73///
74/// In both forms the handler may declare an optional second parameter, the per-delivery
75/// `&mut Context`, to read app state or publish manually. Any further parameter is an extractor: its
76/// type must implement
77/// [`FromContext`](../ruststream/runtime/trait.FromContext.html), and the generated handler resolves
78/// it from the delivery context (in declaration order) before the body runs, so dependencies arrive
79/// as arguments. A failed extraction settles the delivery by the rejection's `HandlerResult` without
80/// running the body.
81///
82/// ```ignore
83/// // `State<Db>` is resolved from the application state before the body runs.
84/// #[subscriber("orders")]
85/// async fn handle(order: &Order, State(db): State<Db>) -> HandlerResult { /* ... */ }
86/// ```
87#[proc_macro_attribute]
88pub fn subscriber(attr: TokenStream, item: TokenStream) -> TokenStream {
89 let args = parse_macro_input!(attr as SubscriberArgs);
90 let func = parse_macro_input!(item as ItemFn);
91 expand::subscriber(&args, &func).unwrap_or_else(|err| err.to_compile_error().into())
92}
93
94/// Generates a `main` entry point for a `RustStream` service.
95///
96/// Place it on a synchronous, argument-free function that builds and returns an application -
97/// `impl App` (the recommended form, hiding the composed type parameters) or a concrete
98/// `RustStream<_>`. The expansion keeps the function and adds a `main` that hands it to
99/// `ruststream::runtime::cli::run_main`, producing a binary that understands the `run` and
100/// `asyncapi gen` commands with no hand-written runtime boilerplate.
101///
102/// ```ignore
103/// #[ruststream::app]
104/// fn app() -> impl App {
105/// RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new())
106/// }
107/// ```
108#[proc_macro_attribute]
109pub fn app(attr: TokenStream, item: TokenStream) -> TokenStream {
110 let func = parse_macro_input!(item as ItemFn);
111 expand_app(&attr.into(), &func).unwrap_or_else(|err| err.to_compile_error().into())
112}
113
114fn expand_app(attr: &TokenStream2, func: &ItemFn) -> syn::Result<TokenStream> {
115 if !attr.is_empty() {
116 return Err(syn::Error::new_spanned(
117 attr,
118 "#[ruststream::app] takes no arguments",
119 ));
120 }
121 if let Some(asyncness) = func.sig.asyncness {
122 return Err(syn::Error::new_spanned(
123 asyncness,
124 "#[ruststream::app] requires a synchronous builder returning `impl App` or `RustStream`",
125 ));
126 }
127 if !func.sig.inputs.is_empty() {
128 return Err(syn::Error::new_spanned(
129 &func.sig.inputs,
130 "#[ruststream::app] builder must take no arguments",
131 ));
132 }
133 let name = &func.sig.ident;
134 Ok(quote! {
135 #func
136
137 fn main() -> ::std::process::ExitCode {
138 ::ruststream::runtime::cli::run_main(#name)
139 }
140 }
141 .into())
142}
143
144/// Derives [`Message`](../ruststream/trait.Message.html) metadata: the type name and its doc
145/// comment.
146///
147/// ```ignore
148/// /// An order placed by a customer.
149/// #[derive(Message)]
150/// struct Order { id: u32 }
151/// // Order::NAME == "Order", Order::DESCRIPTION == Some("An order placed by a customer.")
152/// ```
153#[proc_macro_derive(Message)]
154pub fn derive_message(item: TokenStream) -> TokenStream {
155 let input = parse_macro_input!(item as DeriveInput);
156 let name = &input.ident;
157 let name_str = name.to_string();
158 let description = doc_description(&input.attrs);
159 let (impl_generics, ty_generics, where_clause) = input.generics.split_for_impl();
160
161 quote! {
162 impl #impl_generics ::ruststream::Message for #name #ty_generics #where_clause {
163 const NAME: &'static str = #name_str;
164 const DESCRIPTION: ::core::option::Option<&'static str> = #description;
165 }
166 }
167 .into()
168}
169
170/// Derives [`FromRef`](../ruststream/runtime/trait.FromRef.html) for each field of an
171/// application-state struct, so `#[subscriber]` handlers can inject any field with
172/// `State<FieldType>` without a hand-written impl.
173///
174/// Each field gets a `FromRef` impl that clones it out of the state. Because the generated impl
175/// carries no generic parameter, it is legal even for fields whose type comes from another crate (a
176/// broker publisher, a client pool). A field that another field's type already claims, or that
177/// should not be injectable, opts out with `#[from_ref(skip)]`; two fields may not share a type
178/// (injection by type would be ambiguous).
179///
180/// ```ignore
181/// #[derive(FromRef)]
182/// struct AppState {
183/// orders: OrderService, // handlers can now take `State<OrderService>`
184/// #[from_ref(skip)]
185/// config: Config,
186/// }
187/// ```
188#[proc_macro_derive(FromRef, attributes(from_ref))]
189pub fn derive_from_ref(item: TokenStream) -> TokenStream {
190 let input = parse_macro_input!(item as DeriveInput);
191 from_ref::expand(&input)
192 .unwrap_or_else(syn::Error::into_compile_error)
193 .into()
194}