use proc_macro::TokenStream;
use quote::quote;
use syn::{
braced, parse, parse_macro_input, Data, DeriveInput, Expr, ExprLit, Ident, Lit,
Meta, MetaNameValue, Path, Token, Type,
};
fn camel_to_dotted(name: &str) -> String {
let mut result = String::with_capacity(name.len() + 8);
let mut chars = name.chars().peekable();
let mut prev_lower = false;
while let Some(ch) = chars.next() {
if ch.is_uppercase() {
let next_lower = chars.peek().is_some_and(|c| c.is_lowercase());
if prev_lower || next_lower {
if !result.is_empty() {
result.push('.');
}
} else if !result.is_empty() {
} else {
}
result.push(ch.to_ascii_lowercase());
prev_lower = false;
} else {
if result.is_empty() {
}
result.push(ch);
prev_lower = ch.is_lowercase() || ch.is_ascii_digit();
}
}
result
}
struct EventAttrs {
name: Option<String>,
topic: Option<String>,
}
fn parse_event_attrs(attrs: &[syn::Attribute]) -> EventAttrs {
let mut name = None;
let mut topic = None;
for attr in attrs {
if !attr.path().is_ident("event") {
continue;
}
let nested = attr.parse_args_with(
syn::punctuated::Punctuated::<Meta, syn::Token![,]>::parse_terminated,
);
if let Ok(metas) = nested {
for meta in metas {
if let Meta::NameValue(MetaNameValue {
path,
value:
Expr::Lit(ExprLit {
lit: Lit::Str(lit_str),
..
}),
..
}) = meta
{
if path.is_ident("name") {
name = Some(lit_str.value());
} else if path.is_ident("topic") {
topic = Some(lit_str.value());
}
}
}
}
}
EventAttrs { name, topic }
}
fn first_segment(name: &str) -> &str {
name.split('.').next().unwrap_or(name)
}
#[proc_macro_derive(Event, attributes(event))]
pub fn derive_event(input: TokenStream) -> TokenStream {
let input = parse_macro_input!(input as DeriveInput);
match &input.data {
Data::Struct(_) => {}
Data::Enum(_) => {
return syn::Error::new_spanned(
&input.ident,
"Event can only be derived for structs, not enums",
)
.to_compile_error()
.into();
}
Data::Union(_) => {
return syn::Error::new_spanned(
&input.ident,
"Event can only be derived for structs, not unions",
)
.to_compile_error()
.into();
}
}
let ident = &input.ident;
let attrs = parse_event_attrs(&input.attrs);
let event_name = attrs.name.unwrap_or_else(|| camel_to_dotted(&ident.to_string()));
let topic = attrs
.topic
.unwrap_or_else(|| first_segment(&event_name).to_owned());
let expanded = quote! {
impl ::anycms_event::Event for #ident {
fn event_name() -> &'static str {
#event_name
}
fn topic() -> &'static str {
#topic
}
fn to_json(&self) -> Option<::serde_json::Value> {
::serde_json::to_value(self).ok()
}
fn from_json(json: &str) -> Option<Self>
where
Self: Sized,
{
::serde_json::from_str(json).ok()
}
}
};
expanded.into()
}
struct EventDecl {
name: Ident,
fields: Vec<(Ident, Type)>,
}
struct TopicDecl {
method_name: Ident,
event_types: Vec<Ident>,
}
struct EventBusDef {
bus_name: Ident,
events: Vec<EventDecl>,
topics: Vec<TopicDecl>,
enable_redis: bool,
}
fn peek_keyword(input: &parse::ParseBuffer, keyword: &str) -> bool {
input.peek(Ident)
&& input
.cursor()
.ident()
.is_some_and(|(ident, _)| ident == keyword)
}
fn parse_keyword(input: parse::ParseStream, keyword: &str) -> syn::Result<()> {
let ident: Ident = input.parse()?;
if ident != keyword {
return Err(syn::Error::new(ident.span(), format!("expected `{}`", keyword)));
}
Ok(())
}
impl parse::Parse for EventBusDef {
fn parse(input: parse::ParseStream) -> syn::Result<Self> {
parse_keyword(input, "bus")?;
let bus_name: Ident = input.parse()?;
let mut enable_redis = false;
if input.peek(syn::token::Paren) {
let paren_content;
syn::parenthesized!(paren_content in input);
let attr: Ident = paren_content.parse()?;
if attr != "redis" {
return Err(syn::Error::new(attr.span(), "expected `redis`"));
}
enable_redis = true;
}
let content;
braced!(content in input);
let mut events = Vec::new();
let mut topics = Vec::new();
while !content.is_empty() {
if peek_keyword(&content, "event") {
parse_keyword(&content, "event")?;
let name: Ident = content.parse()?;
let fields_content;
braced!(fields_content in content);
let mut fields = Vec::new();
while !fields_content.is_empty() {
let field_name: Ident = fields_content.parse()?;
fields_content.parse::<Token![:]>()?;
let field_type: Type = fields_content.parse()?;
fields.push((field_name, field_type));
if fields_content.peek(Token![,]) {
fields_content.parse::<Token![,]>()?;
}
}
events.push(EventDecl { name, fields });
} else if peek_keyword(&content, "topic") {
parse_keyword(&content, "topic")?;
let method_name: Ident = content.parse()?;
content.parse::<Token![=>]>()?;
let types_content;
let _bracket = syn::bracketed!(types_content in content);
let mut event_types = Vec::new();
while !types_content.is_empty() {
let event_type: Path = types_content.parse()?;
if let Some(segment) = event_type.segments.last() {
event_types.push(segment.ident.clone());
}
if types_content.peek(Token![,]) {
types_content.parse::<Token![,]>()?;
}
}
topics.push(TopicDecl {
method_name,
event_types,
});
} else {
return Err(content.error("expected `event` or `topic`"));
}
}
Ok(EventBusDef {
bus_name,
events,
topics,
enable_redis,
})
}
}
#[proc_macro]
pub fn event_bus(input: TokenStream) -> TokenStream {
let def = match syn::parse::<EventBusDef>(input) {
Ok(d) => d,
Err(e) => return e.to_compile_error().into(),
};
let bus_name = &def.bus_name;
let enum_name = quote::format_ident!("{}TopicEvent", def.bus_name);
let event_structs: Vec<proc_macro2::TokenStream> = def
.events
.iter()
.map(|event| {
let name = &event.name;
let event_name_str = camel_to_dotted(&name.to_string());
let topic_str = first_segment(&event_name_str).to_owned();
let fields: Vec<proc_macro2::TokenStream> = event
.fields
.iter()
.map(|(fname, ftype)| {
quote! { pub #fname: #ftype }
})
.collect();
quote! {
#[derive(::std::fmt::Debug, ::std::clone::Clone, ::serde::Serialize, ::serde::Deserialize)]
pub struct #name {
#(#fields),*
}
impl ::anycms_event::Event for #name {
fn event_name() -> &'static str {
#event_name_str
}
fn topic() -> &'static str {
#topic_str
}
fn to_json(&self) -> Option<::serde_json::Value> {
::serde_json::to_value(self).ok()
}
fn from_json(json: &str) -> Option<Self>
where
Self: Sized,
{
::serde_json::from_str(json).ok()
}
}
}
})
.collect();
let topic_enum = if def.topics.is_empty() {
quote! {}
} else {
let variants: Vec<proc_macro2::TokenStream> = def
.topics
.iter()
.flat_map(|topic| &topic.event_types)
.map(|event_type| {
quote! { #event_type(#event_type) }
})
.collect();
if variants.is_empty() {
quote! {}
} else {
quote! {
#[derive(::std::fmt::Debug, ::std::clone::Clone, ::serde::Serialize, ::serde::Deserialize)]
#[serde(tag = "event_type")]
pub enum #enum_name {
#(#variants),*
}
}
}
};
let topic_subscribe_methods: Vec<proc_macro2::TokenStream> = if def.topics.is_empty() || def.topics.iter().all(|t| t.event_types.is_empty()) {
Vec::new()
} else {
def.topics
.iter()
.map(|topic| {
let method_name = quote::format_ident!("subscribe_topic_{}", topic.method_name);
let subscribe_arms: Vec<proc_macro2::TokenStream> = topic
.event_types
.iter()
.map(|event_type| {
let variant = event_type;
quote! {
{
let h = handler.clone();
self.inner.subscribe::<#variant, _, _>(move |e| {
let h = h.clone();
async move { h(#enum_name::#variant(e)).await }
}).await
}
}
})
.collect();
quote! {
pub async fn #method_name<F, Fut>(&self, handler: F) -> ::std::vec::Vec<::anycms_event::Result<::anycms_event::bus::Subscription>>
where
F: Fn(#enum_name) -> Fut + ::std::clone::Clone + ::std::marker::Send + ::std::marker::Sync + 'static,
Fut: ::std::future::Future<Output = ::anycms_event::Result<()>> + ::std::marker::Send + 'static,
{
let mut subs = ::std::vec::Vec::new();
#(
subs.push(#subscribe_arms);
)*
subs
}
}
})
.collect()
};
let forward_calls: Vec<proc_macro2::TokenStream> = def
.events
.iter()
.map(|event| {
let name = &event.name;
quote! {
handles.push(bridged.forward_from_redis::<#name>().await?);
}
})
.collect();
let redis_impl = if def.enable_redis && !def.events.is_empty() {
let bridged_name = quote::format_ident!("Bridged{}", def.bus_name);
quote! {
pub struct #bridged_name {
inner: ::anycms_event_redis::BridgedEventBus,
_forwarder_handles: ::std::vec::Vec<::anycms_event_redis::ForwarderHandle>,
}
impl #bridged_name {
pub async fn publish<E>(&self, event: E) -> ::anycms_event::Result<()>
where
E: ::anycms_event::Event + ::std::clone::Clone
+ ::serde::Serialize + ::serde::de::DeserializeOwned,
{
self.inner.publish(event).await
}
pub async fn subscribe<E, F, Fut>(&self, handler: F) -> ::anycms_event::Result<::anycms_event::bus::Subscription>
where
E: ::anycms_event::Event,
F: Fn(E) -> Fut + ::std::marker::Send + ::std::marker::Sync + 'static,
Fut: ::std::future::Future<Output = ::anycms_event::Result<()>> + ::std::marker::Send + 'static,
{
self.inner.subscribe::<E, F, Fut>(handler).await
}
pub fn inner(&self) -> &::anycms_event_redis::BridgedEventBus {
&self.inner
}
}
impl ::std::clone::Clone for #bridged_name {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
_forwarder_handles: ::std::vec::Vec::new(),
}
}
}
}
} else {
quote! {}
};
let redis_bridge_method = if def.enable_redis && !def.events.is_empty() {
let bridged_name = quote::format_ident!("Bridged{}", def.bus_name);
quote! {
pub async fn bridge(
&self,
transport: &::anycms_event_redis::RedisTransport,
) -> ::std::result::Result<#bridged_name, ::anycms_event_redis::RedisTransportError> {
let bridged = transport.bridge(self.inner.clone()).await?;
let mut handles = ::std::vec::Vec::new();
#(#forward_calls)*
Ok(#bridged_name {
inner: bridged,
_forwarder_handles: handles,
})
}
}
} else {
quote! {}
};
let bus_impl = quote! {
pub struct #bus_name {
inner: ::anycms_event::EventBus,
}
impl #bus_name {
pub fn new() -> Self {
Self {
inner: ::anycms_event::EventBus::new(),
}
}
pub fn inner(&self) -> &::anycms_event::EventBus {
&self.inner
}
pub fn into_inner(self) -> ::anycms_event::EventBus {
self.inner
}
pub async fn publish<E: ::anycms_event::Event>(&self, event: E) -> ::anycms_event::Result<()> {
self.inner.publish(event).await
}
pub async fn subscribe<E, F, Fut>(&self, handler: F) -> ::anycms_event::Result<::anycms_event::bus::Subscription>
where
E: ::anycms_event::Event,
F: Fn(E) -> Fut + ::std::marker::Send + ::std::marker::Sync + 'static,
Fut: ::std::future::Future<Output = ::anycms_event::Result<()>> + ::std::marker::Send + 'static,
{
self.inner.subscribe::<E, F, Fut>(handler).await
}
#(#topic_subscribe_methods)*
#redis_bridge_method
}
impl ::std::clone::Clone for #bus_name {
fn clone(&self) -> Self {
Self {
inner: self.inner.clone(),
}
}
}
impl ::std::default::Default for #bus_name {
fn default() -> Self {
Self::new()
}
}
};
let expanded = quote! {
#(#event_structs)*
#topic_enum
#bus_impl
#redis_impl
};
expanded.into()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_camel_to_dotted_simple() {
assert_eq!(camel_to_dotted("UserCreated"), "user.created");
}
#[test]
fn test_camel_to_dotted_three_words() {
assert_eq!(camel_to_dotted("UserProfileUpdated"), "user.profile.updated");
}
#[test]
fn test_camel_to_dotted_single_word() {
assert_eq!(camel_to_dotted("Order"), "order");
}
#[test]
fn test_camel_to_dotted_acronym() {
assert_eq!(camel_to_dotted("HTTPServer"), "http.server");
}
#[test]
fn test_camel_to_dotted_order_placed() {
assert_eq!(camel_to_dotted("OrderPlaced"), "order.placed");
}
#[test]
fn test_first_segment_simple() {
assert_eq!(first_segment("user.created"), "user");
}
#[test]
fn test_first_segment_single() {
assert_eq!(first_segment("order"), "order");
}
}