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 lanes;
11mod outgoing;
12mod parse;
13
14use proc_macro::TokenStream;
15use proc_macro2::TokenStream as TokenStream2;
16use quote::{ToTokens, quote};
17use syn::parse::{Parse, ParseStream};
18use syn::punctuated::Punctuated;
19use syn::{
20 Attribute, DeriveInput, Ident, ItemFn, LitInt, LitStr, Meta, Token, Type, parse_macro_input,
21};
22
23use parse::{SubscriberArgs, doc_description};
24
25/// Turns an `async fn` handler into a mountable subscriber definition.
26///
27/// The attribute carries the declarative settings of the subscription; the *shape* of the
28/// handler - one message (`&T`) or a whole batch of them (`&[T]`) - is read off the signature,
29/// which is the only place it was ever really written. Which lane a type rides is the type's
30/// own business: a `serde` type decodes through the scope codec, a `#[derive(Deserialized)]`
31/// type constructs itself from the payload bytes, and a reply encodes through the reply codec
32/// unless its type is `#[derive(Serialized)]`, in which case its bytes leave as they are.
33///
34/// ```ignore
35/// /// Processes incoming orders.
36/// #[subscriber("orders")]
37/// async fn handle(order: &Order) -> HandlerOutcome { HandlerOutcome::ack() }
38/// // later: broker_scope.include(handle);
39///
40/// // reply form: the return value is encoded and published through the reply publisher
41/// // attached at the include site (the broker's default without it), at the destination its
42/// // own type declares.
43/// #[derive(Serialize, Outgoing)]
44/// #[outgoing(name = "responses")]
45/// struct Response { /* ... */ }
46///
47/// #[subscriber("requests", reply)]
48/// async fn reply(req: &Request) -> Response { /* ... */ }
49///
50/// // a reply type declaring no name takes the clause's:
51/// #[subscriber("requests", reply("responses"))]
52/// async fn reply_elsewhere(req: &Request) -> Receipt { /* ... */ }
53///
54/// // reply form with explicit ack control: `Ok` publishes the reply, `Err` skips it and the
55/// // dispatcher settles by the returned HandlerOutcome.
56/// #[subscriber("requests", reply("responses"))]
57/// async fn confirm(req: &Request) -> Result<Response, HandlerOutcome> { /* ... */ }
58///
59/// // batch form: the slice parameter says the handler takes the whole decoded batch; the
60/// // subscription's subscriber must implement BatchSubscriber (natively, or through the
61/// // buffer the mount site adds).
62/// #[subscriber("orders")]
63/// async fn bill(orders: &[Order]) -> HandlerOutcome { /* settles the whole batch */ }
64///
65/// // raw form: no codec, no serde - the input type deserializes itself from the payload
66/// // bytes. `Deserialize` means the framework's codec does it; `Deserialized` means it is
67/// // already done by the user's own type.
68/// #[derive(Deserialized)]
69/// struct Frame<'a>(&'a [u8]);
70///
71/// #[subscriber("frames")]
72/// async fn on_frame(frame: &Frame<'_>) -> HandlerOutcome { /* parse it yourself */ }
73///
74/// // raw batch: the batch shape on the same lane; the views borrow the batch's own messages
75/// // for the duration of the call.
76/// #[subscriber("frames")]
77/// async fn ingest(frames: &[Frame<'_>]) -> HandlerOutcome { /* parse them yourself */ }
78///
79/// // the settings the attribute leaves out are filled in at the mount site, through the same
80/// // builder the attribute expands into:
81/// #[subscriber]
82/// async fn audit(order: &Order) -> HandlerOutcome { HandlerOutcome::ack() }
83/// // later: broker_scope.include(audit.name(subject).workers(nonzero!(4)));
84///
85/// // raw reply: the reply type carries its own bytes, so they are published as they are -
86/// // no codec on the way out. `Serialize` means the framework's codec does it; `Serialized`
87/// // means it is already done by the user's own type. The same reply clause serves both
88/// // wires, and Result<Export, HandlerOutcome> gives the usual explicit ack control.
89/// #[derive(Outgoing, Serialized)]
90/// struct Export(Vec<u8>);
91///
92/// #[subscriber("frames", reply("frames-out"))]
93/// async fn mirror(frame: &Frame<'_>) -> Export { Export(frame.0.to_vec()) }
94///
95/// // the lanes mix freely (the gateway shape): the input decodes with the scope codec, the
96/// // returned bytes still go out unencoded - and the other diagonal (a Deserialized input
97/// // with an encoded serde reply) works the same way.
98/// #[subscriber("orders", reply("orders-wire"))]
99/// async fn encode(order: &Order) -> Export { /* your wire format */ }
100/// ```
101///
102/// A reply type derives `Outgoing`: `#[outgoing(name = "..")]` on it is the destination, and a
103/// derive without a name takes the clause's `reply("..")` as the destination instead. A declared
104/// name wins over the clause's, and startup logs a warning when the two differ.
105///
106/// Without a `reply` clause the handler returns any accepted outcome shape (a `HandlerOutcome`,
107/// `()`, or `Result<_, E>`). Attach a post-settle continuation with
108/// `HandlerOutcome::ack().and_after` (any outcome works), which runs after the message is
109/// settled. With one it returns the reply value to publish, or
110/// `Result<Reply, HandlerOutcome>` to control acknowledgement: `Err(outcome)` publishes nothing
111/// and settles by `outcome`. The `Result` form is
112/// detected syntactically, so spell it out in the signature (a type alias is treated as a plain
113/// reply type). A failed reply publish nacks the delivery with requeue on both wires, so a
114/// reply is retried rather than silently lost.
115///
116/// A `&[T]` message parameter makes the definition a batch one: the handler runs once per batch
117/// pulled from the subscription's `BatchSubscriber` (added by the mount site's buffer for
118/// subscriptions without native batching). It returns one outcome
119/// for the whole batch (`HandlerOutcome`, `()`, `Result<_, E>`), or a per-element
120/// `Vec<HandlerOutcome>` to settle element `i` of the slice with outcome `i`, each element
121/// carrying its own optional `and_after` continuation. A `&[Message<H, T>]` parameter pairs
122/// each element with its typed
123/// header contract: the core decodes both under the subscriber's decode policy, and the body
124/// reads `element.headers` next to `element.body` (the single-message counterpart stays the
125/// `Headers<T>` extractor parameter). A batch body may name the broker's subscription-scoped
126/// batch context (`ctx: &mut Context<'_, MemoryBatchContext>`-style) to reach handles every
127/// delivery of the subscription shares, like a seeker.
128///
129/// Combining a batch handler with `reply(..)` produces a batch reply definition: the handler
130/// returns `Vec<Reply>` (or
131/// `Result<Vec<Reply>, HandlerOutcome>` for explicit ack control, all-or-nothing - selective
132/// outcomes do not compose with a transaction), every reply is published to the reply name, and
133/// the whole batch is acked after. Name the reply's policy with `.out(Reply, ..)` at the mount
134/// site for independent reply publishes, and chain `.transactional()` for one transaction per
135/// batch.
136///
137/// A `workers(n)` clause processes up to `n` deliveries (or batches) of this subscriber
138/// concurrently on long-lived workers; global processing order is lost by design, and
139/// back-pressure holds at `n` in-flight deliveries. `workers(n, by_key)` switches to `n`
140/// sequential lanes keyed by the message's partition key, preserving per-key ordering
141/// (single-message forms only). The default is the sequential loop.
142///
143/// A `threads(n)` clause is the same position for a handler that computes: the subscription's
144/// deliveries (or batches) are handled on `n` dedicated threads of its own, each running a
145/// current-thread runtime, so the computation holds up none of the app runtime's threads.
146/// `threads(n, by_key)` keeps a key on one thread. A subscription names `workers(..)` or
147/// `threads(..)`, not both. What the delivery leaves behind (continuations, hooks, a retry timer)
148/// stays on its thread; the handler sends work to the app's runtime through the `spawn` and `run`
149/// methods of a `Ctx(main): Ctx<MainRuntime>` parameter.
150///
151/// ```ignore
152/// #[subscriber("images.resize", threads(8))]
153/// async fn resize(job: &Resize) -> HandlerOutcome {
154/// render(job);
155/// HandlerOutcome::ack()
156/// }
157/// ```
158///
159/// Clause values need not be literals: `workers(..)` and `threads(..)` take any `usize` expression (a constant,
160/// a static, a function call - an integer literal keeps the compile-time zero rejection, a
161/// runtime value of zero panics at registration), `reply(..)` takes a `&'static str`
162/// expression, and `on_failure(..)` keys accept a `FailurePolicy` expression next to the keyword
163/// vocabulary.
164///
165/// An `Out(out): Out<P>` parameter injects a live publisher, paired at startup from the
166/// policy attached at the include site (`b.include(f).out(marker, policy).build()`); its optional third
167/// position declares the message set the handler publishes (`Out<impl Publisher, Marker,
168/// (A, B)>` - a tuple, a single type, or a `#[derive(OutMessages)]` set enum), narrowing what
169/// the publish builder accepts on it, and the generated document reports that narrowed list as
170/// the handler's send operations. `Out` parameters combine freely in one handler: with each
171/// other, with a byte input, with a batch handler, and with every reply form (the two `reply`
172/// spellings and the batch publishing form). An `Out` parameter's attachment is
173/// required at the include site: one `.out(marker, policy)` per slot, next to the reply's own
174/// optional `.out(Reply, policy)`, and a `.build()` to commit. Repositioning a
175/// subscription from inside the body is not a parameter but a broker context field: the broker
176/// publishes its position and seek-handle keys, and the handler reads them with the `Ctx`
177/// extractor like any other broker field.
178///
179/// A `start_at(<position>)` clause opens the subscription at that position instead of the
180/// broker's default, seeking before the first delivery ("start from the latest on deploy",
181/// "replay the whole log"). The position is the broker's own type, named by its constructor
182/// (`MemoryPosition::start()`, a Kafka-style `latest()`); the source's subscriber must
183/// implement the `Seekable` capability, so the clause does not compile against a broker
184/// without a replayable log. The position is forced on every startup; without the clause the
185/// subscription simply opens at the broker's default, and conditional defaults stay on the
186/// broker's own subscription descriptor.
187///
188/// ```ignore
189/// // Opens at the start of the log: entries published before the service started are
190/// // replayed into the fresh subscription.
191/// #[subscriber("audit", start_at(MemoryPosition::start()))]
192/// async fn record(entry: &Entry) -> HandlerOutcome { /* ... */ }
193/// ```
194///
195/// In both forms the handler may declare an optional second parameter, the per-delivery
196/// `&mut Context`, to read app state or publish manually. Any further parameter is an extractor: its
197/// type must implement
198/// [`FromContext`](../ruststream/runtime/trait.FromContext.html), and the generated handler resolves
199/// it from the delivery context (in declaration order) before the body runs, so dependencies arrive
200/// as arguments. A failed extraction settles the delivery by the rejection's `HandlerOutcome`
201/// without running the body.
202///
203/// ```ignore
204/// // `State<Db>` is resolved from the application state before the body runs.
205/// #[subscriber("orders")]
206/// async fn handle(order: &Order, State(db): State<Db>) -> HandlerOutcome { /* ... */ }
207/// ```
208///
209/// The attribute expands into the definition and then the settings builder over it, so the
210/// clauses above are exactly the calls a user would write at the mount site (see
211/// `SubscriberSettings` in the core crate). A setting the attribute names is fixed in the
212/// definition's type and its builder method no longer applies, so the two can never disagree.
213#[proc_macro_attribute]
214pub fn subscriber(attr: TokenStream, item: TokenStream) -> TokenStream {
215 let args = parse_macro_input!(attr as SubscriberArgs);
216 let func = parse_macro_input!(item as ItemFn);
217 expand::subscriber(&args, &func).unwrap_or_else(|err| err.to_compile_error().into())
218}
219
220/// Generates a `main` entry point for a `RustStream` service.
221///
222/// Place it on a synchronous, argument-free function that builds and returns an application -
223/// `impl App` (the recommended form, hiding the composed type parameters) or a concrete
224/// `RustStream<_>`. The expansion keeps the function and adds a `main` that hands it to
225/// `ruststream::runtime::cli::run_main`, producing a binary that understands the `run` and
226/// `asyncapi gen` commands with no hand-written runtime boilerplate.
227///
228/// ```ignore
229/// #[ruststream::app]
230/// fn app() -> impl App {
231/// RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new())
232/// }
233/// ```
234///
235/// The attribute takes the runtime arguments of `#[tokio::main]`: `flavor = "current_thread"`
236/// runs the service on a current-thread runtime, `worker_threads = n` sizes the default
237/// multi-threaded one (`flavor = "multi_thread"` names the default). A service whose CPU-bound
238/// subscriptions run on dedicated threads (`threads(n)`) sizes the runtime for its I/O alone.
239///
240/// ```ignore
241/// #[ruststream::app(flavor = "current_thread")]
242/// fn app() -> impl App {
243/// RustStream::new(AppInfo::new("svc", "0.1.0")).register_broker(MemoryBroker::new())
244/// }
245/// ```
246#[proc_macro_attribute]
247pub fn app(attr: TokenStream, item: TokenStream) -> TokenStream {
248 let func = parse_macro_input!(item as ItemFn);
249 let runtime = parse_macro_input!(attr as AppRuntimeArgs);
250 expand_app(&runtime, &func).unwrap_or_else(|err| err.to_compile_error().into())
251}
252
253/// The runtime arguments of `#[ruststream::app]`, as `#[tokio::main]` spells them.
254struct AppRuntimeArgs {
255 flavor: Option<LitStr>,
256 worker_threads: Option<LitInt>,
257}
258
259impl Parse for AppRuntimeArgs {
260 fn parse(input: ParseStream) -> syn::Result<Self> {
261 let mut args = Self {
262 flavor: None,
263 worker_threads: None,
264 };
265 let metas = Punctuated::<Meta, Token![,]>::parse_terminated(input)?;
266 let mut pairs = Vec::with_capacity(metas.len());
267 for meta in metas {
268 let Meta::NameValue(pair) = meta else {
269 return Err(syn::Error::new_spanned(
270 meta,
271 "#[ruststream::app] takes `flavor = \"current_thread\"` or \
272 `flavor = \"multi_thread\"`, and `worker_threads = n`",
273 ));
274 };
275 pairs.push(pair);
276 }
277 for pair in pairs {
278 let syn::Expr::Lit(syn::ExprLit { lit, .. }) = &pair.value else {
279 return Err(syn::Error::new_spanned(
280 &pair.value,
281 "expected a literal, as `#[tokio::main]` takes",
282 ));
283 };
284 if pair.path.is_ident("flavor") {
285 let syn::Lit::Str(flavor) = lit else {
286 return Err(syn::Error::new_spanned(
287 lit,
288 "expected \"current_thread\" or \"multi_thread\"",
289 ));
290 };
291 if !matches!(flavor.value().as_str(), "current_thread" | "multi_thread") {
292 return Err(syn::Error::new_spanned(
293 flavor,
294 "unknown runtime flavor; expected \"current_thread\" or \"multi_thread\"",
295 ));
296 }
297 if args.flavor.replace(flavor.clone()).is_some() {
298 return Err(syn::Error::new_spanned(&pair.path, "duplicate `flavor`"));
299 }
300 } else if pair.path.is_ident("worker_threads") {
301 let syn::Lit::Int(count) = lit else {
302 return Err(syn::Error::new_spanned(lit, "expected a worker count"));
303 };
304 if count.base10_parse::<usize>()? == 0 {
305 return Err(syn::Error::new_spanned(
306 count,
307 "a runtime needs at least one worker",
308 ));
309 }
310 if args.worker_threads.replace(count.clone()).is_some() {
311 return Err(syn::Error::new_spanned(
312 &pair.path,
313 "duplicate `worker_threads`",
314 ));
315 }
316 } else {
317 return Err(syn::Error::new_spanned(
318 &pair.path,
319 "unknown argument; #[ruststream::app] takes `flavor` and `worker_threads`",
320 ));
321 }
322 }
323 if let (Some(flavor), Some(count)) = (&args.flavor, &args.worker_threads)
324 && flavor.value() == "current_thread"
325 {
326 return Err(syn::Error::new_spanned(
327 count,
328 "`worker_threads` sizes a multi-threaded runtime; a current-thread runtime runs \
329 on the thread that calls `main`",
330 ));
331 }
332 Ok(args)
333 }
334}
335
336impl AppRuntimeArgs {
337 /// The `AppRuntime` value the arguments name.
338 fn runtime(&self) -> TokenStream2 {
339 let runtime = quote!(::ruststream::runtime::cli::AppRuntime);
340 if self
341 .flavor
342 .as_ref()
343 .is_some_and(|flavor| flavor.value() == "current_thread")
344 {
345 return quote!(#runtime::CurrentThread);
346 }
347 let Some(count) = &self.worker_threads else {
348 return quote!(#runtime::MultiThread);
349 };
350 // The literal was checked to be non-zero, so the `None` arm is unreachable; MIN keeps
351 // the lowering panic-free.
352 quote! {
353 #runtime::MultiThreadWorkers(match ::core::num::NonZeroUsize::new(#count) {
354 ::core::option::Option::Some(count) => count,
355 ::core::option::Option::None => ::core::num::NonZeroUsize::MIN,
356 })
357 }
358 }
359}
360
361fn expand_app(runtime: &AppRuntimeArgs, func: &ItemFn) -> syn::Result<TokenStream> {
362 if let Some(asyncness) = func.sig.asyncness {
363 return Err(syn::Error::new_spanned(
364 asyncness,
365 "#[ruststream::app] requires a synchronous builder returning `impl App` or `RustStream`",
366 ));
367 }
368 if !func.sig.inputs.is_empty() {
369 return Err(syn::Error::new_spanned(
370 &func.sig.inputs,
371 "#[ruststream::app] builder must take no arguments",
372 ));
373 }
374 let name = &func.sig.ident;
375 let runtime = runtime.runtime();
376 Ok(quote! {
377 #func
378
379 fn main() -> ::std::process::ExitCode {
380 ::ruststream::runtime::cli::run_main_on(#runtime, #name)
381 }
382 }
383 .into())
384}
385
386/// Derives [`MessageInfo`](../ruststream/trait.MessageInfo.html) metadata: the type name, its doc
387/// comment, and its header contract.
388///
389/// The optional `#[message(headers(Meta))]` attribute declares the type's typed header
390/// contract (see `MessageHeaders` in the core crate): `Meta` is serialized into the header map
391/// on the typed publish path and lifted into the message's `AsyncAPI` headers schema. Without
392/// the attribute the type declares no typed headers.
393///
394/// The derive describes a message the service receives, or replies with. A message the handler
395/// sends through an `Out` slot declares where it goes as well, so it derives
396/// [`Outgoing`](derive.Outgoing.html) instead - only that derive makes the type a declared
397/// message set (see `OutMessages` in the core crate) the third `Out` argument can name.
398///
399/// ```ignore
400/// /// An order placed by a customer.
401/// #[derive(MessageInfo)]
402/// struct Order { id: u32 }
403/// // Order::NAME == "Order", Order::DESCRIPTION == Some("An order placed by a customer.")
404///
405/// #[derive(Message, Serialize)]
406/// #[message(headers(ChunkMeta))]
407/// struct ChunkDone { output_key: String }
408/// ```
409#[proc_macro_derive(MessageInfo, attributes(message))]
410pub fn derive_message(item: TokenStream) -> TokenStream {
411 let input = parse_macro_input!(item as DeriveInput);
412 let name = &input.ident;
413 let name_str = name.to_string();
414 let description = doc_description(&input.attrs);
415 let contract = match message_headers_contract(&input.attrs) {
416 Ok(Some(headers)) => quote!(::ruststream::WithHeaders<#headers>),
417 Ok(None) => quote!(::ruststream::NoHeaders),
418 Err(err) => return err.into_compile_error().into(),
419 };
420 let (impl_generics, ty_generics, where_clause) = input.generics.split_for_impl();
421
422 quote! {
423 impl #impl_generics ::ruststream::MessageInfo for #name #ty_generics #where_clause {
424 const NAME: &'static str = #name_str;
425 const DESCRIPTION: ::core::option::Option<&'static str> = #description;
426 }
427
428 impl #impl_generics ::ruststream::MessageHeaders for #name #ty_generics #where_clause {
429 type Contract = #contract;
430 }
431 }
432 .into()
433}
434
435/// Derives everything a message type declares about being sent, from one `key = value`
436/// attribute: its destination
437/// ([`OutgoingDestination`](../ruststream/trait.OutgoingDestination.html)), its optional typed
438/// header contract, and the [`MessageInfo`](../ruststream/trait.MessageInfo.html) metadata the
439/// generated document reads (the type's name and doc comment).
440///
441/// The destination decides which positions the publish builder demands at the call site:
442///
443/// * `#[outgoing(name = "orders.done")]` fixes it - `publisher.message(&done).publish()` names
444/// nothing.
445/// * `#[outgoing(name = "orders.{tenant}.v1")]` declares a space of names - `to()` opens one
446/// setter per placeholder, and the publish appears once the last one is bound.
447/// * the derive alone declares nothing - the call site names it: `.to("orders.archived")`.
448///
449/// The same declaration decides a reply's destination: `#[subscriber(.., reply)]` publishes the
450/// return value where its type says, and `reply("..")` supplies the destination of a reply type
451/// that declares none. A reply type derives this in every case; a name template does not resolve
452/// there, because the runtime has nothing to bind its placeholders with.
453///
454/// `headers = Meta` declares the typed header contract: `Meta` stays an ordinary serde struct
455/// the derive does not touch, and the publish builder then demands `with_headers(&meta)`.
456///
457/// Do not combine with `#[derive(MessageInfo)]` on the same type: this derive already provides the
458/// message metadata, and the two produce conflicting impls.
459///
460/// ```ignore
461/// /// A finished order.
462/// #[derive(Outgoing, Serialize)]
463/// #[outgoing(name = "orders.done", headers = OrderMeta)]
464/// struct OrderDone { id: u64 }
465///
466/// #[derive(Outgoing, Serialize)]
467/// #[outgoing(name = "orders.{tenant}.{region}.v1")]
468/// struct OrderPlaced { id: u64 }
469/// // out.message(&placed).to().tenant("acme").region("eu").publish().await?;
470///
471/// #[derive(Outgoing, Serialize)]
472/// struct OrderArchived { id: u64 }
473/// // out.message(&archived).to("orders.archived").publish().await?;
474/// ```
475#[proc_macro_derive(Outgoing, attributes(outgoing))]
476pub fn derive_outgoing(item: TokenStream) -> TokenStream {
477 let input = parse_macro_input!(item as DeriveInput);
478 outgoing::expand(&input)
479 .unwrap_or_else(syn::Error::into_compile_error)
480 .into()
481}
482
483/// Parses the optional `#[message(headers(Type))]` helper attribute of `#[derive(MessageInfo)]`.
484fn message_headers_contract(attrs: &[Attribute]) -> syn::Result<Option<Type>> {
485 let mut found: Option<Type> = None;
486 for attr in attrs.iter().filter(|attr| attr.path().is_ident("message")) {
487 let meta: Meta = attr.parse_args()?;
488 let Meta::List(list) = &meta else {
489 return Err(syn::Error::new_spanned(
490 meta,
491 "expected `headers(<Type>)` inside #[message(..)]",
492 ));
493 };
494 if !list.path.is_ident("headers") {
495 return Err(syn::Error::new_spanned(
496 &list.path,
497 "expected `headers(<Type>)` inside #[message(..)]",
498 ));
499 }
500 let headers: Type = list.parse_args()?;
501 if found.replace(headers).is_some() {
502 return Err(syn::Error::new_spanned(
503 attr,
504 "duplicate `headers(..)`: a message declares one header contract",
505 ));
506 }
507 }
508 Ok(found)
509}
510
511/// Derives [`FromRef`](../ruststream/runtime/trait.FromRef.html) for each field of an
512/// application-state struct, so `#[subscriber]` handlers can inject any field with
513/// `State<FieldType>` without a hand-written impl.
514///
515/// Each field gets a `FromRef` impl that clones it out of the state. Because the generated impl
516/// carries no generic parameter, it is legal even for fields whose type comes from another crate (a
517/// broker publisher, a client pool). A field that another field's type already claims, or that
518/// should not be injectable, opts out with `#[from_ref(skip)]`; two fields may not share a type
519/// (injection by type would be ambiguous).
520///
521/// ```ignore
522/// #[derive(FromRef)]
523/// struct AppState {
524/// orders: OrderService, // handlers can now take `State<OrderService>`
525/// #[from_ref(skip)]
526/// config: Config,
527/// }
528/// ```
529#[proc_macro_derive(FromRef, attributes(from_ref))]
530pub fn derive_from_ref(item: TokenStream) -> TokenStream {
531 let input = parse_macro_input!(item as DeriveInput);
532 from_ref::expand(&input)
533 .unwrap_or_else(syn::Error::into_compile_error)
534 .into()
535}
536
537/// Derives [`OutSlot`](../ruststream/runtime/trait.OutSlot.html) for a unit struct, making it a
538/// slot marker for `Out<impl Publisher, Marker>` handler parameters. The marker's
539/// [`NAME`](../ruststream/runtime/trait.OutSlot.html#associatedconstant.NAME) is the struct's
540/// identifier, used by startup errors and test assertions.
541///
542/// The optional `#[publishes(Type, ..)]` attribute lists the message types the slot may
543/// publish, which is what the generated document reports for a handler leaving its `Out`
544/// parameter unrestricted. The list is enforced, not only documented: publishing a type it does
545/// not name is a compile error (see `PublishedThrough` in the core crate), so the document cannot
546/// fall behind what handlers send. Each listed type declares its own destination with
547/// `#[derive(Outgoing)]`; listing one type twice is an error.
548///
549/// ```ignore
550/// #[derive(OutSlot)]
551/// struct Encoded;
552///
553/// #[derive(OutSlot)]
554/// #[publishes(ChunkDone, Progress)]
555/// struct Events;
556///
557/// #[subscriber("chunks")]
558/// async fn transcode(chunk: &[u8], Out(out): Out<impl Publisher, Encoded>) -> HandlerOutcome {
559/// /* ... */
560/// }
561/// // include site: b.include(transcode).out(Encoded, KafkaPublish::default());
562/// ```
563#[proc_macro_derive(OutSlot, attributes(publishes))]
564pub fn derive_out_slot(item: TokenStream) -> TokenStream {
565 let input = parse_macro_input!(item as DeriveInput);
566 let syn::Data::Struct(data) = &input.data else {
567 return syn::Error::new_spanned(&input.ident, "OutSlot is derived on a unit struct")
568 .into_compile_error()
569 .into();
570 };
571 if !matches!(data.fields, syn::Fields::Unit) {
572 return syn::Error::new_spanned(
573 &data.fields,
574 "an OutSlot marker is a unit struct: it identifies the slot and carries no data",
575 )
576 .into_compile_error()
577 .into();
578 }
579 if !input.generics.params.is_empty() {
580 return syn::Error::new_spanned(
581 &input.generics,
582 "an OutSlot marker takes no generic parameters",
583 )
584 .into_compile_error()
585 .into();
586 }
587 let dictionary = match publishes_dictionary(&input.attrs) {
588 Ok(dictionary) => dictionary,
589 Err(err) => return err.into_compile_error().into(),
590 };
591 let name = &input.ident;
592 let name_str = name.to_string();
593
594 let outgoing = slot_outgoing_metadata(name, &dictionary);
595 // The list is the whole membership: what the document reports as leaving the slot is exactly
596 // what the publish builder admits.
597 let memberships = dictionary.iter().map(|ty| {
598 quote! {
599 impl ::ruststream::runtime::PublishedThrough<#name> for #ty {}
600 }
601 });
602 let destination = slot_destination(&dictionary);
603
604 quote! {
605 impl ::ruststream::runtime::OutSlot for #name {
606 const NAME: &'static str = #name_str;
607
608 type Destination = #destination;
609
610 #outgoing
611 }
612
613 #(#memberships)*
614 }
615 .into()
616}
617
618/// What the marker offers a transform, folded out of its `#[publishes(..)]` list: the list as
619/// types, nested right and closed with `()`, projected through `ListOffer`. A marker with no list
620/// promises nothing, because it admits every declared message, so it offers `Reads`.
621fn slot_destination(dictionary: &[Type]) -> TokenStream2 {
622 if dictionary.is_empty() {
623 return quote!(::ruststream::runtime::Reads);
624 }
625 let list = dictionary
626 .iter()
627 .rev()
628 .fold(quote!(()), |tail, ty| quote!((#ty, #tail)));
629 quote!(<#list as ::ruststream::runtime::ListOffer>::Offer)
630}
631
632/// The marker's `outgoing()` override, reporting its `#[publishes(..)]` list as the document's
633/// outgoing messages: each listed type contributes what its own `#[derive(Outgoing)]` declares.
634/// An empty list keeps the trait's own default (nothing declared).
635fn slot_outgoing_metadata(name: &Ident, dictionary: &[Type]) -> TokenStream2 {
636 if dictionary.is_empty() {
637 return quote!();
638 }
639 let entries = dictionary.iter().map(|ty| {
640 quote! {
641 __rs_entries.extend(
642 <#ty as ::ruststream::runtime::OutMessages<#name>>::outgoing(),
643 );
644 }
645 });
646 quote! {
647 fn outgoing() -> ::std::vec::Vec<::ruststream::runtime::OutgoingMessageMetadata> {
648 let mut __rs_entries = ::std::vec::Vec::new();
649 #(#entries)*
650 __rs_entries
651 }
652 }
653}
654
655/// Derives a declared message set (`OutMessages` in the core crate) from an enum whose
656/// variants each wrap one message model. The enum is a type-level declaration: it is never
657/// constructed, its variants' inner types are the set.
658///
659/// Naming the enum as the third `Out` argument (`Out<impl Publisher, Marker, SendSet>`)
660/// declares that the handler publishes exactly those models: the publish builder compiles for
661/// each of them (and for nothing else), and the generated `AsyncAPI` document lists them as the
662/// handler's `send` operations. Each model declares where it goes with `#[derive(Outgoing)]`.
663///
664/// Do not combine with `#[derive(Outgoing)]` on the same enum: a type is a message or a set,
665/// and the two derives produce conflicting impls.
666///
667/// ```ignore
668/// #[derive(OutMessages)]
669/// enum ConvertSends {
670/// Progress(Progress),
671/// Done(ChunkDone),
672/// }
673///
674/// #[subscriber("chunks.raw")]
675/// async fn convert(
676/// chunk: &Chunk,
677/// Out(out): Out<impl Publisher, Events, ConvertSends>,
678/// ) -> HandlerOutcome { /* out.message(&Progress { .. }).publish() */ }
679/// ```
680#[proc_macro_derive(OutMessages)]
681pub fn derive_out_messages(item: TokenStream) -> TokenStream {
682 let input = parse_macro_input!(item as DeriveInput);
683 let syn::Data::Enum(data) = &input.data else {
684 return syn::Error::new_spanned(
685 &input.ident,
686 "OutMessages is derived on an enum whose variants each wrap one message model",
687 )
688 .into_compile_error()
689 .into();
690 };
691 if !input.generics.params.is_empty() {
692 return syn::Error::new_spanned(
693 &input.generics,
694 "an OutMessages set takes no generic parameters",
695 )
696 .into_compile_error()
697 .into();
698 }
699 let models = match out_messages_models(data) {
700 Ok(models) => models,
701 Err(err) => return err.into_compile_error().into(),
702 };
703 let name = &input.ident;
704 let memberships = models.iter().enumerate().map(|(index, model)| {
705 quote! {
706 impl ::ruststream::runtime::ContainsMessage<
707 #model,
708 ::ruststream::runtime::SlotPos<#index>,
709 > for #name
710 {
711 }
712 }
713 });
714 // Each model already declares itself as a one-element set; the enum's contribution is the
715 // concatenation, so a model's destination is described in exactly one place.
716 let entries = models.iter().map(|model| {
717 quote! {
718 __rs_entries.extend(
719 <#model as ::ruststream::runtime::OutMessages<__RsM>>::outgoing(),
720 );
721 }
722 });
723 quote! {
724 #(#memberships)*
725
726 impl<__RsM: ::ruststream::runtime::OutSlot> ::ruststream::runtime::OutMessages<__RsM>
727 for #name
728 where
729 #(#models: ::ruststream::runtime::OutMessages<__RsM>,)*
730 {
731 fn outgoing()
732 -> ::std::vec::Vec<::ruststream::runtime::OutgoingMessageMetadata>
733 {
734 let mut __rs_entries = ::std::vec::Vec::new();
735 #(#entries)*
736 __rs_entries
737 }
738 }
739 }
740 .into()
741}
742
743/// Derives a self-deserializing input type (`Deserialized` in the core crate): `Deserialize`
744/// means the framework's codec does it, `Deserialized` means it is already done by the user's
745/// own type.
746///
747/// Covers the obvious shape - a newtype or single-field struct over the payload view
748/// `&'a [u8]` - and emits the construction plus both `Input` spellings, so `&Frame<'_>` and
749/// `&[Frame<'_>]` bodies mount.
750///
751/// A type that deserializes *itself* - a generated Protobuf message, a flatbuffers root, a capnp
752/// reader - names its format's own reader with `#[wire(decode = <path>)]` instead, and then the
753/// shape of the type is its own business. The path is a function `fn(&[u8]) -> Result<Self, E>`
754/// (or `fn(&'a [u8]) -> Result<Self<'a>, E>` for a borrowing view); the derive calls it and
755/// nothing here depends on the format's crate. `#[wire(prost)]` is the shorthand for
756/// `prost::Message::decode`.
757///
758/// ```ignore
759/// #[derive(Deserialized)]
760/// struct Frame<'a>(&'a [u8]);
761///
762/// #[subscriber("frames")]
763/// async fn inspect(frame: &Frame<'_>) -> HandlerOutcome { /* parse it yourself */ }
764///
765/// // What a generator's config puts on every message it emits.
766/// #[derive(Clone, PartialEq, prost::Message, Deserialized)]
767/// #[wire(prost)]
768/// struct Order { #[prost(uint64, tag = "1")] id: u64 }
769/// ```
770#[proc_macro_derive(Deserialized, attributes(wire))]
771pub fn derive_deserialized(item: TokenStream) -> TokenStream {
772 let input = parse_macro_input!(item as DeriveInput);
773 lanes::derive_deserialized(&input)
774 .unwrap_or_else(syn::Error::into_compile_error)
775 .into()
776}
777
778/// Derives a self-serialized outgoing type (`Serialized` in the core crate): `Serialize` means
779/// the framework's codec does it, `Serialized` means it is already done by the user's own
780/// type.
781///
782/// Covers the obvious shape - a newtype or single-field struct over an owned byte buffer - and
783/// emits the bytes accessor plus the wire spellings that route the type onto the serialized
784/// wire (`MessageWire` for a typed publish, `ReplyShape` for the reply position), so a handler
785/// returning it - or passing it to `message(..)` - publishes those bytes as they are.
786///
787/// A type that serializes *itself* names its format's own writer with
788/// `#[wire(encode = <path>)]`, and then holds fields rather than bytes: the path is a function
789/// `fn(&Self, &mut impl BufMut)`, returning either nothing or `Result<(), E>`, and the derive
790/// calls it against the buffer the publish path already carries - so the value is encoded once,
791/// into the buffer that leaves. Nothing here depends on the format's crate. `#[wire(prost)]` is
792/// the shorthand for `prost::Message::encode`.
793///
794/// ```ignore
795/// #[derive(Serialized)]
796/// struct Export(Vec<u8>);
797///
798/// #[subscriber("orders", reply("orders-wire"))]
799/// async fn encode(order: &Order) -> Export { /* your wire format */ }
800///
801/// // What a generator's config puts on every message it emits.
802/// #[derive(Clone, PartialEq, prost::Message, Serialized)]
803/// #[wire(prost)]
804/// struct OrderDone { #[prost(uint64, tag = "1")] id: u64 }
805/// ```
806#[proc_macro_derive(Serialized, attributes(wire))]
807pub fn derive_serialized(item: TokenStream) -> TokenStream {
808 let input = parse_macro_input!(item as DeriveInput);
809 lanes::derive_serialized(&input)
810 .unwrap_or_else(syn::Error::into_compile_error)
811 .into()
812}
813
814/// The set enum's models, one per variant: each variant wraps exactly one type, and a
815/// duplicate model is rejected (its membership index inference would be ambiguous).
816fn out_messages_models(data: &syn::DataEnum) -> syn::Result<Vec<&Type>> {
817 let mut models: Vec<&Type> = Vec::new();
818 for variant in &data.variants {
819 let syn::Fields::Unnamed(fields) = &variant.fields else {
820 return Err(syn::Error::new_spanned(
821 variant,
822 "an OutMessages variant wraps exactly one message model: `Variant(Model)`",
823 ));
824 };
825 if fields.unnamed.len() != 1 {
826 return Err(syn::Error::new_spanned(
827 fields,
828 "an OutMessages variant wraps exactly one message model: `Variant(Model)`",
829 ));
830 }
831 let model = &fields.unnamed[0].ty;
832 let rendered = model.to_token_stream().to_string();
833 if models
834 .iter()
835 .any(|existing| existing.to_token_stream().to_string() == rendered)
836 {
837 return Err(syn::Error::new_spanned(
838 model,
839 "this model is already in the set: a duplicate would make the membership \
840 index inference ambiguous",
841 ));
842 }
843 models.push(model);
844 }
845 Ok(models)
846}
847
848/// Parses every `#[publishes(Type, ..)]` attribute into one list, rejecting a type listed twice
849/// (a second entry would say nothing the first does not).
850fn publishes_dictionary(attrs: &[Attribute]) -> syn::Result<Vec<Type>> {
851 let mut dictionary: Vec<Type> = Vec::new();
852 for attr in attrs
853 .iter()
854 .filter(|attr| attr.path().is_ident("publishes"))
855 {
856 let entries = attr.parse_args_with(Punctuated::<Type, Token![,]>::parse_terminated)?;
857 for ty in entries {
858 let name = ty.to_token_stream().to_string();
859 if dictionary
860 .iter()
861 .any(|existing| existing.to_token_stream().to_string() == name)
862 {
863 return Err(syn::Error::new_spanned(
864 &ty,
865 "this type is already listed on the slot: a type appears once",
866 ));
867 }
868 dictionary.push(ty);
869 }
870 }
871 Ok(dictionary)
872}