jsonrpsee_proc_macros/

render_server.rs

// Copyright 2019-2021 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any
// person obtaining a copy of this software and associated
// documentation files (the "Software"), to deal in the
// Software without restriction, including without
// limitation the rights to use, copy, modify, merge,
// publish, distribute, sublicense, and/or sell copies of
// the Software, and to permit persons to whom the Software
// is furnished to do so, subject to the following
// conditions:
//
// The above copyright notice and this permission notice
// shall be included in all copies or substantial portions
// of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF
// ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED
// TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
// PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT
// SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
// CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
// OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR
// IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.

use std::collections::HashSet;

use super::RpcDescription;
use crate::{
    helpers::{generate_where_clause, is_option, may_borrow},
    rpc_macro::RpcFnArg,
};
use proc_macro2::{Span, TokenStream as TokenStream2};
use quote::{quote, quote_spanned};

impl RpcDescription {
    pub(super) fn render_server(&self) -> Result<TokenStream2, syn::Error> {
        let trait_name = quote::format_ident!("{}Server", &self.trait_def.ident);
        let generics = self.trait_def.generics.clone();
        let (impl_generics, _, where_clause) = generics.split_for_impl();

        let method_impls = self.render_methods()?;
        let into_rpc_impl = self.render_into_rpc()?;

        // Doc-comment to be associated with the server.
        let doc_comment =
            format!("Server trait implementation for the `{}` RPC API.", self.trait_def.ident);

        let trait_impl = quote! {
            #[doc = #doc_comment]
            pub trait #trait_name #impl_generics: Sized + Send + Sync + 'static #where_clause {
                #method_impls
                #into_rpc_impl
            }
        };

        Ok(trait_impl)
    }

    fn render_methods(&self) -> Result<TokenStream2, syn::Error> {
        let methods = self.methods.iter().map(|method| {
            let docs = &method.docs;
            let mut method_sig = method.signature.clone();
            desugar_async(&mut method_sig);

            quote! {
                #docs
                #method_sig
            }
        });

        let subscriptions = self.subscriptions.iter().map(|sub| {
            let docs = &sub.docs;
            let subscription_sink_ty = self.jrps_server_item(quote! { PendingSubscriptionSink });

            // Add `SubscriptionSink` as the second input parameter to the signature.
            let subscription_sink: syn::FnArg =
                syn::parse_quote!(subscription_sink: #subscription_sink_ty);
            let mut sub_sig = sub.signature.clone();
            sub_sig.sig.inputs.insert(1, subscription_sink);
            desugar_async(&mut sub_sig);

            quote! {
                #docs
                #sub_sig
            }
        });

        Ok(quote! {
            #(#methods)*
            #(#subscriptions)*
        })
    }

    /// Helper that will ignore results of `register_*` method calls, and panic if there have been
    /// any errors in debug builds.
    ///
    /// The debug assert is a safeguard should the contract that guarantees the method names to
    /// never conflict in the macro be broken in the future.
    fn handle_register_result(&self, tokens: TokenStream2) -> TokenStream2 {
        let reexports = self.jrps_server_item(quote! { core::__reexports });
        quote! {{
            let _res = #tokens;
            #[cfg(debug_assertions)]
            if _res.is_err() {
                #reexports::panic_fail_register();
            }
        }}
    }

    fn render_into_rpc(&self) -> Result<TokenStream2, syn::Error> {
        let rpc_module = self.jrps_server_item(quote! { RpcModule });

        let mut registered = HashSet::new();
        let mut errors = Vec::new();
        let mut check_name = |name: &str, span: Span| {
            if registered.contains(name) {
                let message = format!("{name:?} is already defined");
                errors.push(quote_spanned!(span => compile_error!(#message);));
            } else {
                registered.insert(name.to_string());
            }
        };

        let methods = self
            .methods
            .iter()
            .map(|method| {
                // Rust method to invoke (e.g. `self.<foo>(...)`).
                let rust_method_name = &method.signature.sig.ident;
                // Name of the RPC method (e.g. `foo_makeSpam`).
                let rpc_method_name = self.rpc_identifier(&method.name);
                // `parsing` is the code associated with parsing structure from the
                // provided `Params` object.
                // `params_seq` is the comma-delimited sequence of parameters we're passing to the
                // rust function called..
                let (parsing, params_seq) = self.render_params_decoding(&method.params, None);

                let into_response = self.jrps_server_item(quote! { IntoResponse });

                check_name(&rpc_method_name, rust_method_name.span());

                if method.signature.sig.asyncness.is_some() {
                    self.handle_register_result(quote! {
                        rpc.register_async_method(#rpc_method_name, |params, context| async move {
                            #parsing
                            #into_response::into_response(context.as_ref().#rust_method_name(#params_seq).await)
                        })
                    })
                } else if method.blocking {
                    self.handle_register_result(quote! {
                        rpc.register_blocking_method(#rpc_method_name, |params, context| {
                            #parsing
                            #into_response::into_response(context.#rust_method_name(#params_seq))
                        })
                    })
                } else {
                    self.handle_register_result(quote! {
                        rpc.register_method(#rpc_method_name, |params, context| {
                            #parsing
                            #into_response::into_response(context.#rust_method_name(#params_seq))
                        })
                    })
                }
            })
            .collect::<Vec<_>>();

        let subscriptions = self
            .subscriptions
            .iter()
            .map(|sub| {
                // Rust method to invoke (e.g. `self.<foo>(...)`).
                let rust_method_name = &sub.signature.sig.ident;
                // Name of the RPC method to subscribe to (e.g. `foo_sub`).
                let rpc_sub_name = self.rpc_identifier(&sub.name);
                // Name of `method` in the subscription response.
                let rpc_notif_name_override = sub.notif_name_override.as_ref().map(|m| self.rpc_identifier(m));
                // Name of the RPC method to unsubscribe (e.g. `foo_sub`).
                let rpc_unsub_name = self.rpc_identifier(&sub.unsubscribe);
                // `parsing` is the code associated with parsing structure from the
                // provided `Params` object.
                // `params_seq` is the comma-delimited sequence of parameters.
                let pending = proc_macro2::Ident::new("pending", rust_method_name.span());
                let (parsing, params_seq) = self.render_params_decoding(&sub.params, Some(pending));
                let sub_err = self.jrps_server_item(quote! { SubscriptionCloseResponse });
                let into_sub_response = self.jrps_server_item(quote! { IntoSubscriptionCloseResponse });

                check_name(&rpc_sub_name, rust_method_name.span());
                check_name(&rpc_unsub_name, rust_method_name.span());

                let rpc_notif_name = match rpc_notif_name_override {
                    Some(notif) => {
                        check_name(&notif, rust_method_name.span());
                        notif
                    }
                    None => rpc_sub_name.clone(),
                };

                if sub.signature.sig.asyncness.is_some() {
                    self.handle_register_result(quote! {
                        rpc.register_subscription(#rpc_sub_name, #rpc_notif_name, #rpc_unsub_name, |params, mut pending, context| async move {
                            #parsing
                            #into_sub_response::into_response(context.as_ref().#rust_method_name(pending, #params_seq).await)
                        })
                    })
                } else {
                    self.handle_register_result(quote! {
                        rpc.register_subscription_raw(#rpc_sub_name, #rpc_notif_name, #rpc_unsub_name, |params, mut pending, context| {
                            #parsing
                            let _ = context.as_ref().#rust_method_name(pending, #params_seq);
                            #sub_err::None
                        })
                    })
                }
            })
            .collect::<Vec<_>>();

        let method_aliases = self
            .methods
            .iter()
            .map(|method| {
                let rpc_name = self.rpc_identifier(&method.name);
                let rust_method_name = &method.signature.sig.ident;

                // Rust method to invoke (e.g. `self.<foo>(...)`).
                let aliases: Vec<TokenStream2> = method
                    .aliases
                    .iter()
                    .map(|alias| {
                        check_name(alias, rust_method_name.span());
                        self.handle_register_result(quote! {
                            rpc.register_alias(#alias, #rpc_name)
                        })
                    })
                    .collect();

                quote!( #(#aliases)* )
            })
            .collect::<Vec<_>>();

        let subscription_aliases = self
            .subscriptions
            .iter()
            .map(|method| {
                let sub_name = self.rpc_identifier(&method.name);
                let unsub_name = self.rpc_identifier(&method.unsubscribe);
                let rust_method_name = &method.signature.sig.ident;

                let sub: Vec<TokenStream2> = method
                    .aliases
                    .iter()
                    .map(|alias| {
                        check_name(alias, rust_method_name.span());
                        self.handle_register_result(quote! {
                            rpc.register_alias(#alias, #sub_name)
                        })
                    })
                    .collect();
                let unsub: Vec<TokenStream2> = method
                    .unsubscribe_aliases
                    .iter()
                    .map(|alias| {
                        check_name(alias, rust_method_name.span());
                        self.handle_register_result(quote! {
                            rpc.register_alias(#alias, #unsub_name)
                        })
                    })
                    .collect();

                quote! (
                    #(#sub)*
                    #(#unsub)*
                )
            })
            .collect::<Vec<_>>();

        let doc_comment = "Collects all the methods and subscriptions defined in the trait \
                                and adds them into a single `RpcModule`.";

        let sub_tys: Vec<syn::Type> =
            self.subscriptions.clone().into_iter().map(|s| s.item).collect();
        let where_clause =
            generate_where_clause(&self.trait_def, &sub_tys, self.server_bounds.as_ref());

        // NOTE(niklasad1): empty where clause is valid rust syntax.
        Ok(quote! {
            #[doc = #doc_comment]
            fn into_rpc(self) -> #rpc_module<Self> where #(#where_clause,)* {
                let mut rpc = #rpc_module::new(self);

                #(#errors)*
                #(#methods)*
                #(#subscriptions)*
                #(#method_aliases)*
                #(#subscription_aliases)*

                rpc
            }
        })
    }

    fn render_params_decoding(
        &self,
        params: &[RpcFnArg],
        sub: Option<proc_macro2::Ident>,
    ) -> (TokenStream2, TokenStream2) {
        if params.is_empty() {
            return (TokenStream2::default(), TokenStream2::default());
        }

        let params_fields_seq = params.iter().map(RpcFnArg::arg_pat);
        let params_fields = quote! { #(#params_fields_seq),* };

        let reexports = self.jrps_server_item(quote! { core::__reexports });

        let error_ret = if let Some(pending) = &sub {
            let tokio = quote! { #reexports::tokio };
            let sub_err = self.jrps_server_item(quote! { SubscriptionCloseResponse });
            quote! {
                #tokio::spawn(#pending.reject(e));
                return #sub_err::None;
            }
        } else {
            let response_payload = self.jrps_server_item(quote! { ResponsePayload });
            quote! {
                return #response_payload::error(e);
            }
        };

        // Code to decode sequence of parameters from a JSON array.
        let decode_array = {
            let decode_fields = params.iter().map(|RpcFnArg { arg_pat, ty, .. }| {
                let is_option = is_option(ty);
                let next_method = if is_option { quote!(optional_next) } else { quote!(next) };
                quote! {
                    let #arg_pat: #ty = match seq.#next_method() {
                        Ok(v) => v,
                        Err(e) => {
                            #reexports::log_fail_parse(stringify!(#arg_pat), stringify!(#ty), &e, #is_option);
                            #error_ret
                        }
                    };
                }
            });

            // Move the raw params into the decoding block so they are freed once decoded, instead
            // of living as long as an async method or subscription future. Arguments that may
            // borrow from the params need them to outlive the call.
            let free_params = (!params.iter().any(|arg| may_borrow(&arg.ty)))
                .then(|| quote! { let params = params; });

            quote! {
                #free_params
                let mut seq = params.sequence();
                #(#decode_fields);*
                (#params_fields)
            }
        };

        // Parameters are positional, like geth: object (named) params fail to parse as an array.
        let parsing = quote! {
            let (#params_fields) = {
                #decode_array
            };
        };

        (parsing, params_fields)
    }
}

/// Turns `async fn f(..) -> R` into `fn f(..) -> impl Future<Output = R> + Send`.
///
/// The server trait stays usable from the `Send` futures that `RpcModule` registers, without boxing
/// every call the way `async_trait` does. Implementations can still use `async fn`.
fn desugar_async(method: &mut syn::TraitItemFn) {
    let sig = &mut method.sig;
    if sig.asyncness.take().is_none() {
        return;
    }
    let output = match &sig.output {
        syn::ReturnType::Default => quote! { () },
        syn::ReturnType::Type(_, ty) => quote! { #ty },
    };
    sig.output = syn::parse_quote! {
        -> impl ::core::future::Future<Output = #output> + ::core::marker::Send
    };
    // A default body becomes the body of the returned future.
    if let Some(body) = &mut method.default {
        *body = syn::parse_quote!({ async move #body });
    }
}