1use battler_wamp_uri::Uri;
2use proc_macro2::Span;
3use quote::quote;
4use syn::{
5 Error,
6 Ident,
7 ItemEnum,
8 LitStr,
9 Path,
10 Result,
11 parse::{
12 Parse,
13 ParseStream,
14 },
15 parse_macro_input,
16 spanned::Spanned,
17};
18
19#[allow(dead_code)]
20enum UriAttribute {
21 Uri(LitStr),
22 Pattern(Path),
23}
24
25#[allow(dead_code)]
26struct RpcAttribute {
27 input: Path,
28 output: Path,
29 error: Option<Path>,
30 uri: UriAttribute,
31 progressive: bool,
32}
33
34#[allow(dead_code)]
35struct PubSubAttribute {
36 event: Path,
37 uri: UriAttribute,
38 subscription: Option<Path>,
39}
40
41#[allow(dead_code)]
42enum Attribute {
43 Rpc(RpcAttribute),
44 PubSub(PubSubAttribute),
45}
46
47#[allow(dead_code)]
48struct Variant {
49 span: Span,
50 ident: Ident,
51 attribute: Attribute,
52}
53
54#[allow(dead_code)]
55struct Input {
56 ident: Ident,
57 realm: LitStr,
58 variants: Vec<Variant>,
59}
60
61impl Parse for Input {
62 fn parse(input: ParseStream) -> Result<Self> {
63 let call_site = Span::call_site();
64 let input =
65 ItemEnum::parse(input).map_err(|_| Error::new(call_site, "input must be an enum"))?;
66 let ident = input.ident;
67 let realm = input
68 .attrs
69 .iter()
70 .find(|attr| attr.path().is_ident("realm"))
71 .map(|attr| attr.parse_args_with(|input: ParseStream| input.parse::<LitStr>()))
72 .ok_or_else(|| Error::new(call_site, "missing realm attribute"))??;
73 Uri::try_from(realm.value()).map_err(|_| Error::new(call_site, "invalid realm uri"))?;
74 let variants = input
75 .variants
76 .into_iter()
77 .map(|variant| {
78 let span = variant.span();
79 let ident = variant.ident;
80 let rpc: Option<Result<RpcAttribute>> = variant
81 .attrs
82 .iter()
83 .find(|attr| attr.path().is_ident("rpc"))
84 .map(|attr| {
85 let mut input = None;
86 let mut output = None;
87 let mut error = None;
88 let mut uri = None;
89 let mut progressive = false;
90 attr.parse_nested_meta(|meta| {
91 if meta.path.is_ident("input") {
92 input = Some(meta.value()?.parse::<Path>()?);
93 Ok(())
94 } else if meta.path.is_ident("output") {
95 output = Some(meta.value()?.parse::<Path>()?);
96 Ok(())
97 } else if meta.path.is_ident("error") {
98 error = Some(meta.value()?.parse::<Path>()?);
99 Ok(())
100 } else if meta.path.is_ident("uri") {
101 uri = Some(UriAttribute::Uri(meta.value()?.parse::<LitStr>()?));
102 Ok(())
103 } else if meta.path.is_ident("pattern") {
104 uri = Some(UriAttribute::Pattern(meta.value()?.parse::<Path>()?));
105 Ok(())
106 } else if meta.path.is_ident("progressive") {
107 progressive = true;
108 Ok(())
109 } else {
110 Ok(())
111 }
112 })?;
113 Ok(RpcAttribute {
114 input: input
115 .ok_or_else(|| Error::new(span, "missing input attribute"))?,
116 output: output
117 .ok_or_else(|| Error::new(span, "missing output attribute"))?,
118 error,
119 uri: uri
120 .ok_or_else(|| Error::new(span, "missing uri/pattern attribute"))?,
121 progressive,
122 })
123 });
124 let pub_sub: Option<Result<PubSubAttribute>> = variant
125 .attrs
126 .iter()
127 .find(|attr| attr.path().is_ident("pubsub"))
128 .map(|attr| {
129 let mut event = None;
130 let mut uri = None;
131 let mut subscription = None;
132 attr.parse_nested_meta(|meta| {
133 if meta.path.is_ident("event") {
134 event = Some(meta.value()?.parse::<Path>()?);
135 Ok(())
136 } else if meta.path.is_ident("uri") {
137 uri = Some(UriAttribute::Uri(meta.value()?.parse::<LitStr>()?));
138 Ok(())
139 } else if meta.path.is_ident("pattern") {
140 uri = Some(UriAttribute::Pattern(meta.value()?.parse::<Path>()?));
141 Ok(())
142 } else if meta.path.is_ident("subscription") {
143 subscription = Some(meta.value()?.parse::<Path>()?);
144 Ok(())
145 } else {
146 Ok(())
147 }
148 })?;
149 Ok(PubSubAttribute {
150 event: event
151 .ok_or_else(|| Error::new(span, "missing event attribute"))?,
152 uri: uri
153 .ok_or_else(|| Error::new(span, "missing uri/pattern attribute"))?,
154 subscription,
155 })
156 });
157 let attribute = match (rpc, pub_sub) {
158 (Some(_), Some(_)) | (None, None) => {
159 return Err(Error::new(span, "variant must be oneof rpc, pubsub"));
160 }
161 (Some(rpc), None) => Attribute::Rpc(rpc?),
162 (None, Some(pub_sub)) => Attribute::PubSub(pub_sub?),
163 };
164 Ok(Variant {
165 span,
166 ident,
167 attribute,
168 })
169 })
170 .collect::<Result<Vec<_>>>()?;
171 Ok(Input {
172 ident,
173 realm,
174 variants,
175 })
176 }
177}
178
179fn variant_to_function_name(s: &str) -> String {
180 let name = s
181 .chars()
182 .map(|c| {
183 if c.is_uppercase() {
184 format!("_{}", c.to_lowercase())
185 } else {
186 format!("{c}")
187 }
188 })
189 .collect::<Vec<_>>()
190 .join("");
191 if let Some('_') = name.chars().nth(0) {
192 name[1..].to_owned()
193 } else {
194 name
195 }
196}
197
198#[proc_macro_derive(WampSchema, attributes(realm, rpc, pubsub))]
201pub fn derive_wamp_schema(input: proc_macro::TokenStream) -> proc_macro::TokenStream {
202 #[allow(unused)]
203 let input = parse_macro_input!(input as Input);
204
205 let ident = &input.ident;
206 let realm = &input.realm;
207
208 let peer = quote!(self.__peer_handle);
209 let peer_builder = quote!(self.__peer_builder);
210
211 let subscriptions = input.variants.iter().map(|variant| match &variant.attribute {
212 Attribute::Rpc(_) => quote!(),
213 Attribute::PubSub(pubsub) => {
214 let variant_ident = &variant.ident;
215 let name = Ident::new(&format!("{variant_ident}Subscription"), variant.span);
216 let event = &pubsub.event;
217 let underlying = match &pubsub.uri {
218 UriAttribute::Uri(_) => quote!(::battler_wamprat::subscription::TypedSubscription<Event = #event>),
219 UriAttribute::Pattern(pattern) => quote!(::battler_wamprat::subscription::TypedPatternMatchedSubscription<Pattern = #pattern, Event = #event>),
220 };
221 quote! {
222 #[doc = "Subscription for handling events of the"]
223 #[doc = concat!("[`", stringify!(#ident), "::", stringify!(#variant_ident), "`]")]
224 #[doc = "topic."]
225 pub trait #name: #underlying {}
226 }
227 }
228 }).collect::<Vec<_>>();
229
230 let consumer_methods = input
231 .variants
232 .iter()
233 .map(|variant| match &variant.attribute {
234 Attribute::Rpc(rpc) => {
235 let variant_ident = &variant.ident;
236 let name = variant_to_function_name(&variant_ident.to_string());
237 const RUST_KEYWORDS: &[&str] = &[
238 "as", "break", "const", "continue", "crate", "else", "enum", "extern", "false", "fn",
239 "for", "if", "impl", "in", "let", "loop", "match", "mod", "move", "mut", "pub",
240 "ref", "return", "self", "Self", "static", "struct", "super", "trait", "true",
241 "type", "unsafe", "use", "where", "while", "async", "await", "dyn",
242 ];
243 let name = if RUST_KEYWORDS.contains(&name.as_str()) {
244 Ident::new_raw(&name, variant.span)
245 } else {
246 Ident::new(&name, variant.span)
247 };
248 let input = &rpc.input;
249 let output = &rpc.output;
250 let error = &rpc.error;
251 let error = match error {
252 Some(error) => quote!(#error),
253 None => quote!(::anyhow::Error),
254 };
255 let (uri_input, uri_arg) = match &rpc.uri {
256 UriAttribute::Uri(uri) => (quote!(), quote!(::battler_wamp_uri::Uri::try_from(#uri)?)),
257 UriAttribute::Pattern(pattern) => (quote!(uri: #pattern,), quote!(uri.wamp_generate_uri()?)),
258 };
259 let (output_rpc, method) = if rpc.progressive {
260 (quote!(::battler_wamprat_schema::ProgressivePendingRpc), quote!(call_with_progress))
261 } else {
262 (quote!(::battler_wamprat_schema::SimplePendingRpc), quote!(call))
263 };
264 quote! {
265 #[doc = "Calls the"]
266 #[doc = concat!("[`", stringify!(#ident), "::", stringify!(#variant_ident), "`]")]
267 #[doc = "procedure."]
268 pub async fn #name(&self, #uri_input input: #input, call_options: ::battler_wamprat::peer::CallOptions) -> ::anyhow::Result<#output_rpc<#output, #error>> {
269 Ok(#peer.#method::<#input, #output>(#uri_arg, input, call_options).await?.into())
270 }
271 }
272 }
273 Attribute::PubSub(pubsub) => {
274 let variant_ident = &variant.ident;
275 let name = variant_to_function_name(&variant_ident.to_string());
276 let subscribe_name = Ident::new(&format!("subscribe_{name}"), variant.span);
277 let unsubscribe_name = Ident::new(&format!("unsubscribe_{name}"), variant.span);
278 let (parameters, subscribe_method_call, unsubscribe_method_call) = match &pubsub.uri {
279 UriAttribute::Uri(uri) => (quote!(), quote!(subscribe(::battler_wamp_uri::Uri::try_from(#uri)?, subscription)), quote!(unsubscribe(&::battler_wamp_uri::WildcardUri::try_from(#uri)?))),
280 UriAttribute::Pattern(pattern) => match &pubsub.subscription {
281 Some(subscription) => (quote!(, generator: &#subscription), quote!(subscribe_pattern_matched_with_generator(generator, subscription)), quote!(unsubscribe_with_generator(generator))),
282 None => (quote!(), quote!(subscribe_pattern_matched(subscription)), quote!(unsubscribe(&#pattern::uri_for_router()))),
283 },
284 };
285 let subscription_type = Ident::new(&format!("{variant_ident}Subscription"), variant.span);
286
287 quote! {
288 #[doc = "Subscribes to the"]
289 #[doc = concat!("[`", stringify!(#ident), "::", stringify!(#variant_ident), "`]")]
290 #[doc = "topic."]
291 pub async fn #subscribe_name<T>(&self #parameters, subscription: T) -> ::anyhow::Result<()> where T: #subscription_type + 'static {
292 #peer.#subscribe_method_call.await
293 }
294
295 #[doc = "Unsubscribes from the"]
296 #[doc = concat!("[`", stringify!(#ident), "::", stringify!(#variant_ident), "`]")]
297 #[doc = "topic."]
298 pub async fn #unsubscribe_name(&self #parameters) -> ::anyhow::Result<()> {
299 #peer.#unsubscribe_method_call.await
300 }
301 }
302 },
303 })
304 .collect::<Vec<_>>();
305
306 let procedures = input
307 .variants
308 .iter()
309 .map(|variant| match &variant.attribute {
310 Attribute::Rpc(rpc) => {
311 let variant_ident = &variant.ident;
312 let name = Ident::new(&format!("{variant_ident}Procedure"), variant.span);
313 let input = &rpc.input;
314 let output = &rpc.output;
315 let error = &rpc.error;
316 let error = match error {
317 Some(error) => quote!(#error),
318 None => quote!(::anyhow::Error),
319 };
320 let underlying = match &rpc.uri {
321 UriAttribute::Uri(_) => if rpc.progressive {
322 quote!(::battler_wamprat::procedure::TypedProgressiveProcedure<Input = #input, Output = #output, Error = #error>)
323
324 } else {
325 quote!(::battler_wamprat::procedure::TypedProcedure<Input = #input, Output = #output, Error = #error>)
326 }
327 UriAttribute::Pattern(pattern) => if rpc.progressive {
328 quote!(::battler_wamprat::procedure::TypedPatternMatchedProgressiveProcedure<Pattern = #pattern, Input = #input, Output = #output, Error = #error>)
329
330 } else {
331 quote!(::battler_wamprat::procedure::TypedPatternMatchedProcedure<Pattern = #pattern, Input = #input, Output = #output, Error = #error>)
332 }
333 };
334 quote! {
335 #[doc = "Procedure for handling invocations of the"]
336 #[doc = concat!("[`", stringify!(#ident), "::", stringify!(#variant_ident), "`]")]
337 #[doc = "procedure."]
338 pub trait #name: #underlying {}
339 }
340 }
341 Attribute::PubSub(_) => quote!(),
342 })
343 .collect::<Vec<_>>();
344
345 let producer_builder_methods = input
346 .variants
347 .iter()
348 .map(|variant| match &variant.attribute {
349 Attribute::Rpc(rpc) => {
350 let variant_ident = &variant.ident;
351 let name = variant_to_function_name(&variant_ident.to_string());
352 let name = Ident::new(&format!("register_{name}"), variant.span);
353 let method_call = match &rpc.uri {
354 UriAttribute::Uri(uri) => if rpc.progressive {
355 quote!(add_procedure_progressive(::battler_wamp_uri::Uri::try_from(#uri)?, procedure))
356
357 } else {
358 quote!(add_procedure(::battler_wamp_uri::Uri::try_from(#uri)?, procedure))
359 }
360 UriAttribute::Pattern(_) => if rpc.progressive {
361 quote!(add_procedure_pattern_matched_progressive(procedure))
362 } else {
363 quote!(add_procedure_pattern_matched(procedure))
364 }
365 };
366 let procedure_type = Ident::new(&format!("{variant_ident}Procedure"), variant.span);
367 quote! {
368 #[doc = "Registers a procedure for invocations to the"]
369 #[doc = concat!("[`", stringify!(#ident), "::", stringify!(#variant_ident), "`]")]
370 #[doc = "procedure."]
371 pub fn #name<T>(&mut self, procedure: T) -> ::anyhow::Result<()> where T: #procedure_type + 'static {
372 #peer_builder.#method_call;
373 Ok(())
374 }
375 }
376 }
377 Attribute::PubSub(_) => quote!(),
378 })
379 .collect::<Vec<_>>();
380
381 let producer_methods = input
382 .variants
383 .iter()
384 .map(|variant| match &variant.attribute {
385 Attribute::Rpc(_) => quote!(),
386 Attribute::PubSub(pubsub) => {
387 let variant_ident = &variant.ident;
388 let name = variant_to_function_name(&variant_ident.to_string());
389 let name = Ident::new(&format!("publish_{name}"), variant.span);
390 let event = &pubsub.event;
391 let (uri_input, uri_arg) = match &pubsub.uri {
392 UriAttribute::Uri(uri) => (
393 quote!(),
394 quote!(::battler_wamp_uri::Uri::try_from(#uri)?),
395 ),
396 UriAttribute::Pattern(pattern) => {
397 (quote!(uri: #pattern,), quote!(uri.wamp_generate_uri()?))
398 }
399 };
400 quote! {
401 #[doc = "Publishes an event to the"]
402 #[doc = concat!("[`", stringify!(#ident), "::", stringify!(#variant_ident), "`]")]
403 #[doc = "topic."]
404 pub async fn #name(&self, #uri_input event: #event, options: ::battler_wamprat::peer::PublishOptions) -> ::anyhow::Result<()> {
405 #peer.publish(#uri_arg, event, options).await
406 }
407 }
408 }
409 })
410 .collect::<Vec<_>>();
411
412 let consumer = Ident::new(&format!("{ident}Consumer"), Span::call_site());
413 let producer = Ident::new(&format!("{ident}Producer"), Span::call_site());
414 let producer_builder = Ident::new(&format!("{producer}Builder"), producer.span());
415
416 quote! {
417 #(#subscriptions)*
418
419 #(#procedures)*
420
421 #[doc = "A consumer (client) of the"]
422 #[doc = concat!("[`", stringify!(#ident), "`]")]
423 #[doc = "service."]
424 pub struct #consumer<S> {
425 __peer_handle: ::battler_wamprat::peer::PeerHandle<S>,
426 __join_handle: ::tokio::task::JoinHandle<()>,
427 }
428
429 impl<S> #consumer<S> where S: Send + 'static {
430 fn new(config: ::battler_wamprat_schema::PeerConfig, peer: ::battler_wamp::peer::Peer<S>) -> ::anyhow::Result<Self> {
431 let mut peer_builder = ::battler_wamprat::peer::PeerBuilder::new(config.connection.connection_type.clone());
432 *peer_builder.connection_config_mut() = config.connection;
433 peer_builder.set_auth_methods(config.auth_methods);
434 let (peer_handle, join_handle) = peer_builder.start(
435 peer,
436 ::battler_wamp_uri::Uri::try_from(#realm)?,
437 );
438 Ok(Self { __peer_handle: peer_handle, __join_handle: join_handle })
439 }
440
441 #[doc = "Cancels and waits for the peer to be fully cleaned up by joining the asynchronous task."]
442 pub async fn stop(self) -> ::core::result::Result<(), ::anyhow::Error> {
443 #peer.cancel()?;
444 self.__join_handle.await.map_err(|err| err.into())
445 }
446
447 #[doc = "Waits until the consumer is known to be in a ready state."]
448 pub async fn wait_until_ready(&self) -> ::core::result::Result<(), ::anyhow::Error> {
449 #peer.wait_until_ready().await
450 }
451
452 #(#consumer_methods)*
453 }
454
455 #[doc = "A producer (server) of the"]
456 #[doc = concat!("[`", stringify!(#ident), "`]")]
457 #[doc = "service."]
458 pub struct #producer<S> {
459 __peer_handle: ::battler_wamprat::peer::PeerHandle<S>,
460 __join_handle: ::tokio::task::JoinHandle<()>,
461 }
462
463 impl<S> #producer<S> where S: Send + 'static {
464 #[doc = "Cancels and waits for the peer to be fully cleaned up by joining the asynchronous task."]
465 pub async fn stop(self) -> ::core::result::Result<(), ::anyhow::Error> {
466 #peer.cancel()?;
467 self.__join_handle.await.map_err(|err| err.into())
468 }
469
470 #[doc = "Waits until the producer is known to be in a ready state."]
471 pub async fn wait_until_ready(&self) -> ::core::result::Result<(), ::anyhow::Error> {
472 #peer.wait_until_ready().await
473 }
474
475 #(#producer_methods)*
476 }
477
478 #[doc = "A builder for a"]
479 #[doc = concat!("[`", stringify!(#producer), "`]")]
480 #[doc = "service producer."]
481 pub struct #producer_builder {
482 __peer_builder: ::battler_wamprat::peer::PeerBuilder,
483 }
484
485 impl #producer_builder {
486 fn new(config: ::battler_wamprat_schema::PeerConfig) -> Self {
487 let mut peer_builder = ::battler_wamprat::peer::PeerBuilder::new(config.connection.connection_type.clone());
488 *peer_builder.connection_config_mut() = config.connection;
489 peer_builder.set_auth_methods(config.auth_methods);
490 Self { __peer_builder: peer_builder }
491 }
492
493 #[doc = "Starts the producer on the given peer."]
494 pub fn start<S>(self, peer: ::battler_wamp::peer::Peer<S>) -> ::anyhow::Result<#producer<S>> where S: Send + 'static {
495 let (peer_handle, join_handle) = #peer_builder.start(peer, ::battler_wamp_uri::Uri::try_from(#realm)?);
496 Ok(#producer { __peer_handle: peer_handle, __join_handle: join_handle })
497 }
498
499 #(#producer_builder_methods)*
500 }
501
502 impl #ident {
503 #[doc = "Creates a peer that consumes the"]
504 #[doc = concat!("[`", stringify!(#ident), "`]")]
505 #[doc = "service."]
506 pub fn consumer<S>(config: ::battler_wamprat_schema::PeerConfig, peer: ::battler_wamp::peer::Peer<S>) -> ::anyhow::Result<#consumer<S>> where S: Send + 'static {
507 #consumer::<S>::new(config, peer)
508 }
509
510 #[doc = "Creates a peer builder for a producer of the"]
511 #[doc = concat!("[`", stringify!(#ident), "`]")]
512 #[doc = "service."]
513 pub fn producer_builder(config: ::battler_wamprat_schema::PeerConfig) -> #producer_builder {
514 #producer_builder::new(config)
515 }
516 }
517 }
518 .into()
519}