kacrab_macros/
lib.rs

1//! Procedural macros for kacrab.
2//!
3//! This crate produces proc-macros consumed by the main `kacrab` crate via
4//! the `macros` feature. It is not intended to be used directly.
5
6use proc_macro::TokenStream;
7use proc_macro2::Span;
8use quote::{format_ident, quote};
9use syn::{
10    Attribute, Error, Expr, ExprLit, Ident, Lit, LitStr, Meta, Result, Token, Type, Visibility,
11    braced,
12    parse::{Parse, ParseStream},
13    parse_macro_input,
14    punctuated::Punctuated,
15};
16
17/// Declares a Kafka client config schema and generates a typed struct,
18/// builder, key constants, and static config metadata.
19#[proc_macro]
20pub fn kafka_config(input: TokenStream) -> TokenStream {
21    let input = parse_macro_input!(input as ConfigInput);
22    expand_config(input)
23        .unwrap_or_else(Error::into_compile_error)
24        .into()
25}
26
27struct ConfigInput {
28    attrs: Vec<Attribute>,
29    visibility: Visibility,
30    name: Ident,
31    fields: Punctuated<ConfigField, Token![,]>,
32}
33
34impl Parse for ConfigInput {
35    fn parse(input: ParseStream<'_>) -> Result<Self> {
36        let attrs = input.call(Attribute::parse_outer)?;
37        let visibility = input.parse()?;
38        let name = input.parse()?;
39        let content;
40        braced!(content in input);
41        let fields = content.parse_terminated(ConfigField::parse, Token![,])?;
42        Ok(Self {
43            attrs,
44            visibility,
45            name,
46            fields,
47        })
48    }
49}
50
51struct ConfigField {
52    attrs: Vec<Attribute>,
53    name: Ident,
54    ty: Type,
55}
56
57impl Parse for ConfigField {
58    fn parse(input: ParseStream<'_>) -> Result<Self> {
59        let attrs = input.call(Attribute::parse_outer)?;
60        let name = input.parse()?;
61        let _: Token![:] = input.parse()?;
62        let ty = input.parse()?;
63        Ok(Self { attrs, name, ty })
64    }
65}
66
67#[derive(Clone, Copy)]
68enum ClientKindAttr {
69    Producer,
70    Consumer,
71    Admin,
72}
73
74struct FieldAttrs {
75    key: LitStr,
76    required: bool,
77    default: Option<Expr>,
78    kafka_type: LitStr,
79    kafka_default: LitStr,
80    status: StatusAttr,
81    origin: OriginAttr,
82    source: LitStr,
83    platforms: Vec<LitStr>,
84    feature: Option<LitStr>,
85    comment: LitStr,
86}
87
88#[derive(Clone)]
89enum StatusAttr {
90    Native,
91    NativeReview,
92    SkipJavaOnly,
93    FeatureGated(LitStr),
94    Future(LitStr),
95}
96
97#[derive(Clone, Copy)]
98enum OriginAttr {
99    Kafka,
100    KacrabRuntime,
101}
102
103type Tokens = proc_macro2::TokenStream;
104
105struct FieldExpansion {
106    struct_field: Tokens,
107    builder_field: Tokens,
108    builder_initializer: Tokens,
109    setter: Tokens,
110    build_field: Tokens,
111    constant: Tokens,
112    metadata: Tokens,
113    key_match: Tokens,
114    property_parse: Tokens,
115}
116
117fn expand_config(input: ConfigInput) -> Result<proc_macro2::TokenStream> {
118    let client = parse_client(&input.attrs)?;
119    let client_tokens = client_tokens(client);
120    let client_label = client_label(client);
121    let visibility = input.visibility;
122    let name = input.name;
123    let builder_name = format_ident!("{name}Builder");
124    let struct_doc = LitStr::new(
125        &format!("Typed Kafka {client_label} configuration."),
126        Span::call_site(),
127    );
128    let builder_doc = LitStr::new(&format!("Builder for [`{name}`]."), Span::call_site());
129
130    let expansions = input
131        .fields
132        .into_iter()
133        .map(|field| expand_field(field, &client_tokens))
134        .collect::<Result<Vec<_>>>()?;
135    let struct_fields: Vec<_> = expansions.iter().map(|field| &field.struct_field).collect();
136    let builder_fields: Vec<_> = expansions
137        .iter()
138        .map(|field| &field.builder_field)
139        .collect();
140    let builder_initializers: Vec<_> = expansions
141        .iter()
142        .map(|field| &field.builder_initializer)
143        .collect();
144    let setters: Vec<_> = expansions.iter().map(|field| &field.setter).collect();
145    let build_fields: Vec<_> = expansions.iter().map(|field| &field.build_field).collect();
146    let constants: Vec<_> = expansions.iter().map(|field| &field.constant).collect();
147    let metadata: Vec<_> = expansions.iter().map(|field| &field.metadata).collect();
148    let key_matches: Vec<_> = expansions.iter().map(|field| &field.key_match).collect();
149    let property_parsers: Vec<_> = expansions
150        .iter()
151        .map(|field| &field.property_parse)
152        .collect();
153
154    Ok(quote! {
155        #[doc = #struct_doc]
156        #[derive(Clone, Debug)]
157        #visibility struct #name {
158            #(#struct_fields,)*
159        }
160
161        #[doc = #builder_doc]
162        #[derive(Clone, Debug, Default)]
163        #visibility struct #builder_name {
164            #(#builder_fields,)*
165        }
166
167        impl #name {
168            #(#constants)*
169
170            /// Official Kafka configuration metadata for this typed config.
171            pub const CONFIG_KEYS: &'static [::kacrab::config::ConfigEntry] = &[
172                #(#metadata,)*
173            ];
174
175            /// Creates a builder for this config.
176            #[must_use]
177            pub fn builder() -> #builder_name {
178                #builder_name {
179                    #(#builder_initializers,)*
180                }
181            }
182
183            /// Parses Java-style properties into this typed config.
184            pub fn from_properties(
185                properties: &::kacrab::config::Properties,
186                unknown_key_policy: ::kacrab::config::UnknownKeyPolicy,
187            ) -> ::core::result::Result<(#name, ::kacrab::config::WarningReport), ::kacrab::config::ConfigError> {
188                let report = ::kacrab::config::validate_properties(
189                    #client_tokens,
190                    properties,
191                    unknown_key_policy,
192                )?;
193                for (key, _value) in properties.iter() {
194                    let key = key.as_str();
195                    if !::core::matches!(key, #(#key_matches)|*) {
196                        return ::core::result::Result::Err(::kacrab::config::ConfigError::UnsupportedKey {
197                            client: #client_tokens,
198                            key: key.into(),
199                        });
200                    }
201                }
202                let mut builder = Self::builder();
203                #(#property_parsers)*
204                let config = builder.build()?;
205                ::core::result::Result::Ok((config, report))
206            }
207        }
208
209        impl #builder_name {
210            #(#setters)*
211
212            /// Builds the config, returning an error when required keys are missing.
213            pub fn build(self) -> ::core::result::Result<#name, ::kacrab::config::ConfigError> {
214                ::core::result::Result::Ok(#name {
215                    #(#build_fields,)*
216                })
217            }
218        }
219    })
220}
221
222fn expand_field(field: ConfigField, client_tokens: &Tokens) -> Result<FieldExpansion> {
223    let attrs = parse_field_attrs(&field.attrs)?;
224    if attrs.required && attrs.default.is_some() {
225        return Err(Error::new_spanned(
226            field.name,
227            "config field cannot be both required and defaulted",
228        ));
229    }
230
231    let field_name = field.name;
232    let field_ty = field.ty;
233    let option_field_ty = &field_ty;
234    let key = attrs.key;
235    let kafka_type = attrs.kafka_type;
236    let kafka_default = attrs.kafka_default;
237    let source = attrs.source;
238    let comment = attrs.comment;
239    let status = status_tokens(attrs.status);
240    let origin = origin_tokens(attrs.origin);
241    let platforms = attrs.platforms;
242    let feature = option_lit_str_tokens(attrs.feature.as_ref());
243    let const_name = format_ident!("{}_CONFIG", screaming_snake(&field_name.to_string()));
244    let const_doc = LitStr::new(
245        &format!("Kafka property key `{}`.", key.value()),
246        key.span(),
247    );
248    let setter_doc = LitStr::new(
249        &format!("Sets the `{}` Kafka property.", key.value()),
250        key.span(),
251    );
252
253    let struct_field = quote! {
254        #[doc = #comment]
255        pub #field_name: #field_ty
256    };
257    let builder_field = quote! {
258        #field_name: ::core::option::Option<#option_field_ty>
259    };
260    let builder_initializer = quote! {
261        #field_name: ::core::option::Option::None
262    };
263    let setter = quote! {
264        #[doc = #setter_doc]
265        #[must_use]
266        pub fn #field_name(mut self, value: impl ::core::convert::Into<#field_ty>) -> Self {
267            self.#field_name = ::core::option::Option::Some(value.into());
268            self
269        }
270    };
271    let constant = quote! {
272        #[doc = #const_doc]
273        pub const #const_name: &'static str = #key;
274    };
275    let metadata = quote! {
276        ::kacrab::config::ConfigEntry {
277            client: #client_tokens,
278            origin: #origin,
279            key: #key,
280            rust_field: ::core::stringify!(#field_name),
281            kafka_type: #kafka_type,
282            default: #kafka_default,
283            status: #status,
284            comment: #comment,
285            documentation: "",
286            source: #source,
287            platforms: &[#(#platforms),*],
288            feature: #feature,
289        }
290    };
291    let key_match = quote!(#key);
292    let build_field = build_field_tokens(
293        attrs.required,
294        attrs.default,
295        &field_name,
296        &key,
297        client_tokens,
298    )?;
299    let property_parse = property_parse_tokens(&field_name, &field_ty, &key, client_tokens);
300
301    Ok(FieldExpansion {
302        struct_field,
303        builder_field,
304        builder_initializer,
305        setter,
306        build_field,
307        constant,
308        metadata,
309        key_match,
310        property_parse,
311    })
312}
313
314fn property_parse_tokens(
315    field_name: &Ident,
316    field_ty: &Type,
317    key: &LitStr,
318    client_tokens: &Tokens,
319) -> Tokens {
320    quote! {
321        if let ::core::option::Option::Some(value) = properties.get(#key) {
322            let parsed = <#field_ty as ::kacrab::config::ParseConfigValue>::parse_config_value(value)
323                .map_err(|error| ::kacrab::config::ConfigError::InvalidValue {
324                    client: #client_tokens,
325                    key: #key,
326                    target: error.target,
327                    value: error.value,
328                })?;
329            builder = builder.#field_name(parsed);
330        }
331    }
332}
333
334fn build_field_tokens(
335    required: bool,
336    default: Option<Expr>,
337    field_name: &Ident,
338    key: &LitStr,
339    client_tokens: &Tokens,
340) -> Result<Tokens> {
341    if required {
342        return Ok(quote! {
343            #field_name: match self.#field_name {
344                ::core::option::Option::Some(value) => value,
345                ::core::option::Option::None => {
346                    return ::core::result::Result::Err(::kacrab::config::ConfigError::MissingRequired {
347                        client: #client_tokens,
348                        key: #key,
349                    });
350                }
351            }
352        });
353    }
354
355    let Some(default) = default else {
356        return Err(Error::new_spanned(
357            field_name,
358            "optional config field must define #[default(...)]",
359        ));
360    };
361    Ok(quote! {
362        #field_name: self.#field_name.unwrap_or_else(|| #default)
363    })
364}
365
366fn parse_client(attrs: &[Attribute]) -> Result<ClientKindAttr> {
367    for attr in attrs {
368        if !attr.path().is_ident("client") {
369            continue;
370        }
371        let expr = parse_expr_attr(attr, "client")?;
372        return parse_client_expr(&expr);
373    }
374    Err(Error::new(
375        Span::call_site(),
376        "kafka_config! requires #[client(Producer|Consumer|Admin)]",
377    ))
378}
379
380fn parse_client_expr(expr: &Expr) -> Result<ClientKindAttr> {
381    let Expr::Path(path) = expr else {
382        return Err(Error::new_spanned(
383            expr,
384            "expected Producer, Consumer, or Admin",
385        ));
386    };
387    let Some(ident) = path.path.get_ident() else {
388        return Err(Error::new_spanned(
389            expr,
390            "expected Producer, Consumer, or Admin",
391        ));
392    };
393
394    match ident.to_string().as_str() {
395        "Producer" => Ok(ClientKindAttr::Producer),
396        "Consumer" => Ok(ClientKindAttr::Consumer),
397        "Admin" => Ok(ClientKindAttr::Admin),
398        _ => Err(Error::new_spanned(
399            ident,
400            "expected Producer, Consumer, or Admin",
401        )),
402    }
403}
404
405fn parse_field_attrs(attrs: &[Attribute]) -> Result<FieldAttrs> {
406    let mut key = None;
407    let mut required = false;
408    let mut default = None;
409    let mut kafka_type = None;
410    let mut kafka_default = None;
411    let mut status = None;
412    let mut origin = None;
413    let mut source = None;
414    let mut platforms = Vec::new();
415    let mut feature = None;
416    let mut comment = None;
417
418    for attr in attrs {
419        if attr.path().is_ident("key") {
420            key = Some(parse_lit_str_attr(attr, "key")?);
421        } else if attr.path().is_ident("required") {
422            parse_marker_attr(attr, "required")?;
423            required = true;
424        } else if attr.path().is_ident("default") {
425            default = Some(parse_expr_attr(attr, "default")?);
426        } else if attr.path().is_ident("kafka_type") {
427            kafka_type = Some(parse_lit_str_attr(attr, "kafka_type")?);
428        } else if attr.path().is_ident("kafka_default") {
429            kafka_default = Some(parse_lit_str_attr(attr, "kafka_default")?);
430        } else if attr.path().is_ident("status") {
431            status = Some(parse_status(attr)?);
432        } else if attr.path().is_ident("origin") {
433            origin = Some(parse_origin(attr)?);
434        } else if attr.path().is_ident("source") {
435            source = Some(parse_lit_str_attr(attr, "source")?);
436        } else if attr.path().is_ident("platforms") {
437            platforms = parse_lit_str_list_attr(attr, "platforms")?;
438        } else if attr.path().is_ident("feature") {
439            feature = Some(parse_lit_str_attr(attr, "feature")?);
440        } else if attr.path().is_ident("comment") {
441            comment = Some(parse_lit_str_attr(attr, "comment")?);
442        }
443    }
444
445    Ok(FieldAttrs {
446        key: required_attr(key, "key")?,
447        required,
448        default,
449        kafka_type: required_attr(kafka_type, "kafka_type")?,
450        kafka_default: required_attr(kafka_default, "kafka_default")?,
451        status: required_attr(status, "status")?,
452        origin: origin.unwrap_or(OriginAttr::Kafka),
453        source: required_attr(source, "source")?,
454        platforms,
455        feature,
456        comment: required_attr(comment, "comment")?,
457    })
458}
459
460fn parse_status(attr: &Attribute) -> Result<StatusAttr> {
461    let expr = parse_expr_attr(attr, "status")?;
462    parse_status_expr(&expr)
463}
464
465fn parse_origin(attr: &Attribute) -> Result<OriginAttr> {
466    let expr = parse_expr_attr(attr, "origin")?;
467    let Expr::Path(path) = &expr else {
468        return Err(Error::new_spanned(expr, "expected kafka or kacrab_runtime"));
469    };
470    let Some(ident) = path.path.get_ident() else {
471        return Err(Error::new_spanned(expr, "expected kafka or kacrab_runtime"));
472    };
473    match ident.to_string().as_str() {
474        "kafka" => Ok(OriginAttr::Kafka),
475        "kacrab_runtime" => Ok(OriginAttr::KacrabRuntime),
476        _ => Err(Error::new_spanned(
477            ident,
478            "expected kafka or kacrab_runtime",
479        )),
480    }
481}
482
483fn parse_status_expr(expr: &Expr) -> Result<StatusAttr> {
484    if let Expr::Path(path) = expr {
485        let Some(ident) = path.path.get_ident() else {
486            return Err(Error::new_spanned(expr, "unknown config status"));
487        };
488        return match ident.to_string().as_str() {
489            "native" => Ok(StatusAttr::Native),
490            "native_review" => Ok(StatusAttr::NativeReview),
491            "skip_java_only" => Ok(StatusAttr::SkipJavaOnly),
492            _ => Err(Error::new_spanned(ident, "unknown config status")),
493        };
494    }
495
496    if let Expr::Call(call) = expr
497        && let Expr::Path(path) = call.func.as_ref()
498        && call.args.len() == 1
499        && let Some(Expr::Lit(expr_lit)) = call.args.first()
500        && let Lit::Str(feature) = &expr_lit.lit
501    {
502        if path.path.is_ident("feature_gated") {
503            return Ok(StatusAttr::FeatureGated(feature.clone()));
504        }
505        if path.path.is_ident("future") {
506            return Ok(StatusAttr::Future(feature.clone()));
507        }
508    }
509
510    Err(Error::new_spanned(
511        expr,
512        "expected status native, native_review, skip_java_only, feature_gated(\"feature\"), or \
513         future(\"feature\")",
514    ))
515}
516
517fn parse_lit_str_attr(attr: &Attribute, name: &str) -> Result<LitStr> {
518    let Expr::Lit(ExprLit {
519        lit: Lit::Str(value),
520        ..
521    }) = parse_expr_attr(attr, name)?
522    else {
523        return Err(Error::new_spanned(
524            attr,
525            format!("expected #[{name}(\"...\")]"),
526        ));
527    };
528    Ok(value)
529}
530
531fn parse_lit_str_list_attr(attr: &Attribute, name: &str) -> Result<Vec<LitStr>> {
532    let parser = Punctuated::<LitStr, Token![,]>::parse_terminated;
533    match &attr.meta {
534        Meta::List(_) => Ok(attr.parse_args_with(parser)?.into_iter().collect()),
535        _ => Err(Error::new_spanned(
536            attr,
537            format!("expected #[{name}(\"...\", ...)]"),
538        )),
539    }
540}
541
542fn parse_expr_attr(attr: &Attribute, name: &str) -> Result<Expr> {
543    match &attr.meta {
544        Meta::NameValue(meta) => Ok(meta.value.clone()),
545        Meta::List(_) => attr.parse_args(),
546        Meta::Path(_) => Err(Error::new_spanned(attr, format!("expected #[{name}(...)]"))),
547    }
548}
549
550fn parse_marker_attr(attr: &Attribute, name: &str) -> Result<()> {
551    if matches!(attr.meta, Meta::Path(_)) {
552        Ok(())
553    } else {
554        Err(Error::new_spanned(attr, format!("expected #[{name}]")))
555    }
556}
557
558fn required_attr<T>(value: Option<T>, name: &str) -> Result<T> {
559    value.ok_or_else(|| {
560        Error::new(
561            Span::call_site(),
562            format!("config field requires #[{name}(...)]"),
563        )
564    })
565}
566
567fn client_tokens(client: ClientKindAttr) -> proc_macro2::TokenStream {
568    match client {
569        ClientKindAttr::Producer => quote!(::kacrab::config::ClientKind::Producer),
570        ClientKindAttr::Consumer => quote!(::kacrab::config::ClientKind::Consumer),
571        ClientKindAttr::Admin => quote!(::kacrab::config::ClientKind::Admin),
572    }
573}
574
575const fn client_label(client: ClientKindAttr) -> &'static str {
576    match client {
577        ClientKindAttr::Producer => "producer",
578        ClientKindAttr::Consumer => "consumer",
579        ClientKindAttr::Admin => "admin",
580    }
581}
582
583fn status_tokens(status: StatusAttr) -> proc_macro2::TokenStream {
584    match status {
585        StatusAttr::Native => quote!(::kacrab::config::ConfigStatus::Native),
586        StatusAttr::NativeReview => quote!(::kacrab::config::ConfigStatus::NativeReview),
587        StatusAttr::SkipJavaOnly => quote!(::kacrab::config::ConfigStatus::SkipJavaOnly),
588        StatusAttr::FeatureGated(feature) => {
589            quote!(::kacrab::config::ConfigStatus::FeatureGated { feature: #feature })
590        },
591        StatusAttr::Future(feature) => {
592            quote!(::kacrab::config::ConfigStatus::Future { feature: #feature })
593        },
594    }
595}
596
597fn origin_tokens(origin: OriginAttr) -> proc_macro2::TokenStream {
598    match origin {
599        OriginAttr::Kafka => quote!(::kacrab::config::ConfigOrigin::Kafka),
600        OriginAttr::KacrabRuntime => quote!(::kacrab::config::ConfigOrigin::KacrabRuntime),
601    }
602}
603
604fn option_lit_str_tokens(value: Option<&LitStr>) -> proc_macro2::TokenStream {
605    value.map_or_else(
606        || quote!(::core::option::Option::None),
607        |value| quote!(::core::option::Option::Some(#value)),
608    )
609}
610
611fn screaming_snake(value: &str) -> String {
612    let mut result = String::with_capacity(value.len());
613    for (index, ch) in value.chars().enumerate() {
614        if ch.is_ascii_uppercase() && index > 0 {
615            result.push('_');
616        }
617        result.push(ch.to_ascii_uppercase());
618    }
619    result
620}
kacrab_macros/lib.rs

kacrab_macros/
lib.rs
