use proc_macro2::{Span, TokenStream};
use quote::quote;
use syn::{Token, Type, punctuated::Punctuated};
fn parse_message_list(ast: &syn::DeriveInput) -> syn::Result<Vec<Type>> {
let attr = ast
.attrs
.iter()
.find(|a| a.path().is_ident("message"))
.ok_or_else(|| {
syn::Error::new(
Span::call_site(),
"`#[derive(RemoteAddressable)]` requires a `#[message(..)]` attribute listing at \
least one message type",
)
})?;
let list = attr.parse_args_with(Punctuated::<Type, Token![,]>::parse_terminated)?;
if list.is_empty() {
return Err(syn::Error::new_spanned(
attr,
"`#[message(..)]` requires at least one message type",
));
}
Ok(list.into_iter().collect())
}
pub fn expand(ast: &syn::DeriveInput) -> TokenStream {
let messages = match parse_message_list(ast) {
Ok(list) => list,
Err(err) => return err.to_compile_error(),
};
let name = &ast.ident;
let (impl_generics, ty_generics, where_clause) = ast.generics.split_for_impl();
let self_impl = quote! {
impl #impl_generics ::acktor::RemoteAddressable for #name #ty_generics #where_clause {}
};
let codec_impl = quote! {
impl #impl_generics ::acktor::codec::Codec for #name #ty_generics #where_clause {
fn codec_table() -> &'static ::acktor::codec::CodecTable {
static TABLE: ::std::sync::OnceLock<::acktor::codec::CodecTable> =
::std::sync::OnceLock::new();
TABLE.get_or_init(|| {
let mut map: ::acktor::utils::TypeMap<::acktor::codec::MessageCodec> =
::core::default::Default::default();
#(
map.insert(
::core::any::TypeId::of::<#messages>(),
::acktor::codec::MessageCodec {
message_id: <#messages as ::acktor::MessageId>::ID,
encode_msg: |any, ctx| {
let m = any
.downcast_ref::<#messages>()
.expect("TypeId invariant");
::acktor::codec::Encode::encode_to_bytes(m, ctx)
},
decode_res: |bytes, ctx| {
let r = <<#messages as ::acktor::Message>::Result
as ::acktor::codec::Decode>::decode(bytes, ctx)?;
::core::result::Result::Ok(
::std::boxed::Box::new(r)
as ::std::boxed::Box<dyn ::core::any::Any>
)
},
},
);
)*
::acktor::codec::CodecTable::new(map)
})
}
}
};
let arms = messages.iter().map(|m| {
quote! {
<#m as ::acktor::MessageId>::ID => {
match <#m as ::acktor::codec::Decode>::decode(bytes, decode_msg_ctx) {
::core::result::Result::Ok(msg) => {
let result =
<Self as ::acktor::Handler<#m>>::handle(self, msg, ctx).await;
if let ::core::option::Option::Some(result_tx) = result_tx {
let (tx, rx) = ::acktor::channel::oneshot::channel();
::acktor::MessageResponse::<Self, #m>::handle(result, ctx, Some(tx))
.await;
let result = rx.await;
let m_name = ::acktor::utils::ShortName::of::<#m>();
send_result(result, result_tx, encode_res_ctx, m_name);
}
else {
::acktor::MessageResponse::<Self, #m>::handle(result, ctx, None)
.await;
}
}
::core::result::Result::Err(e) => {
::acktor::tracing::warn!(
"Could not handle the binary message: {}",
::acktor::ErrorReport::report(&e)
);
if let ::core::option::Option::Some(tx) = result_tx {
send_err(tx, e);
}
}
}
}
}
});
let handler_impl = quote! {
impl #impl_generics ::acktor::Handler<::acktor::message::BinaryMessage>
for #name #ty_generics #where_clause
{
type Result = ();
async fn handle(
&mut self,
msg: ::acktor::message::BinaryMessage,
ctx: &mut <Self as ::acktor::Actor>::Context,
) -> Self::Result {
let ::acktor::message::BinaryMessage {
message_id,
bytes,
result_tx,
decode_msg_ctx,
encode_res_ctx,
..
} = msg;
#[inline]
fn send_err(
tx: ::acktor::channel::oneshot::Sender<::acktor::bytes::Bytes>,
err: impl ::core::convert::Into<::acktor::error::BoxError>,
) {
if let Err(e) = tx.send_err(err) {
::acktor::tracing::warn!(
"Could not report the error to the original sender: {}",
::acktor::ErrorReport::report(&e)
);
}
}
#[inline]
fn send_result(
result: ::core::result::Result<
impl ::acktor::codec::Encode,
::acktor::error::RecvError,
>,
tx: ::acktor::channel::oneshot::Sender<::acktor::bytes::Bytes>,
encode_ctx: ::core::option::Option<
::std::sync::Arc<
dyn ::acktor::codec::EncodeContext
+ ::core::marker::Send
+ ::core::marker::Sync,
>,
>,
m_name: ::acktor::utils::ShortName,
) {
match result {
::core::result::Result::Ok(result) => {
let encode_ctx = encode_ctx
.as_deref()
.map(|ctx| ctx as &dyn ::acktor::codec::EncodeContext);
match ::acktor::codec::Encode::encode_to_bytes(&result, encode_ctx) {
::core::result::Result::Ok(bytes) => {
if let Err(e) = tx.send(bytes) {
::acktor::tracing::warn!(
"Could not send the message response to the original \
sender: {}",
::acktor::ErrorReport::report(&e)
);
}
}
::core::result::Result::Err(e) => {
::acktor::tracing::warn!(
"Could not encode the message response: {}",
::acktor::ErrorReport::report(&e)
);
send_err(tx, e);
}
}
}
::core::result::Result::Err(e) => {
::acktor::tracing::warn!(
"Could not handle the binary message: could not receive the \
message response from Handler<{}>: {}",
m_name,
::acktor::ErrorReport::report(&e)
);
send_err(tx, e);
}
};
}
let decode_msg_ctx = decode_msg_ctx
.as_deref()
.map(|ctx| ctx as &dyn ::acktor::codec::DecodeContext);
match message_id {
#(#arms)*
_ => {
::acktor::tracing::debug!(
"Received a message with unknown message id {}",
message_id
);
if let ::core::option::Option::Some(tx) = result_tx {
send_err(
tx,
::acktor::codec::DecodeError::UnknownMessageId(message_id),
);
}
}
}
}
}
};
let ids = messages.iter().map(|m| {
quote! { <#m as ::acktor::MessageId>::ID }
});
let n = messages.len();
let uniqueness_check = quote! {
const _: () = {
const IDS: [u64; #n] = [#(#ids),*];
let mut i = 0;
while i < IDS.len() {
let mut j = i + 1;
while j < IDS.len() {
assert!(
IDS[i] != IDS[j],
"duplicate message ids in #[message(...)]",
);
j += 1;
}
i += 1;
}
};
};
quote! {
#self_impl
#codec_impl
#handler_impl
#uniqueness_check
}
}
#[cfg(test)]
mod tests {
use anyhow::Result;
use pretty_assertions::assert_eq;
use super::*;
fn input(src: &str) -> syn::DeriveInput {
syn::parse_str(src).unwrap()
}
#[test]
fn test_parse_message_list() -> Result<()> {
assert!(parse_message_list(&input("struct Foo;")).is_err());
assert!(parse_message_list(&input("#[message()] struct Foo;")).is_err());
let list = parse_message_list(&input("#[message(Ping, Echo)] struct Foo;"))?;
assert_eq!(list.len(), 2);
let list = parse_message_list(&input("#[message(Observer<Pong>)] struct Foo;"))?;
assert_eq!(list.len(), 1);
Ok(())
}
}