Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466//! RPC helper layer for remote Erlang function calls.//!//! This module provides an abstraction over erl_dist's message passing//! for making remote procedure calls to Erlang nodes.
use crate::connection::{CallRef, ConnectionManager};use crate::error::{RpcError, RpcResult};use crate::server::FormatterMode;use eetf::{Atom, List, Map, Pid, Term, Tuple};use erl_dist::message::Message;use std::collections::HashMap;use std::sync::Arc;use std::time::Duration;use tokio::sync::Mutex;use tokio::time::timeout;
use crate::connection::NodeConnection;
/// Default timeout for RPC calls in milliseconds.const DEFAULT_RPC_TIMEOUT_MS: u64 = 5000;
/// Calls `module:function(args)` on the node, through its `rex` server.////// `timeout_ms` defaults to [`DEFAULT_RPC_TIMEOUT_MS`].pub async fn rpc_call( connection_manager: &ConnectionManager, module: &str, function: &str, args: Vec<Term>, timeout_ms: Option<u64>,) -> RpcResult<Term> { let timeout_duration = Duration::from_millis(timeout_ms.unwrap_or(DEFAULT_RPC_TIMEOUT_MS)); let node = connection_manager.node();
let conn = connection_manager .connection() .await .ok_or_else(|| RpcError::NodeNotConnected { node: node.to_string(), module: module.to_string(), function: function.to_string(), })?;
let result = timeout( timeout_duration, execute_rpc( &conn, connection_manager.formatter_mode(), connection_manager.local_node_name(), node, module, function, args, ), ) .await;
match result { Ok(inner_result) => inner_result, Err(_) => Err(RpcError::Timeout { node: node.to_string(), module: module.to_string(), function: function.to_string(), timeout_ms: timeout_duration.as_millis() as u64, }), }}
/// The pid this server presents to the nodes it talks to.////// It addresses no real process: the connection task answers whatever the node/// sends here. The node part must match the name we handshook as, or replies/// are routed to a node that does not exist and every call times out.fn local_pid(local_node_name: &str) -> Pid { Pid::new(local_node_name, 0, 0, 0)}
/// Sends one `$gen_call` to `rex` and awaits its reply.async fn execute_rpc( conn: &Arc<Mutex<NodeConnection>>, mode: FormatterMode, local_node_name: &str, node: &str, module: &str, function: &str, args: Vec<Term>,) -> RpcResult<Term> { let request_sender = conn .lock() .await .request_sender() .map_err(RpcError::Connection)?;
// rex, the node's RPC server, expects // {'$gen_call', {From, Tag}, {call, Module, Function, Args, GroupLeader}}. let module_atom = Atom::from(module); let function_atom = Atom::from(function); let args_list = Term::from(List::from(args));
let from_pid = local_pid(local_node_name);
let call_tuple = Term::from(Tuple::from(vec![ Term::from(Atom::from("call")), Term::from(module_atom), Term::from(function_atom), args_list, Term::from(Atom::from("user")), // Group leader ]));
let call_ref = CallRef::next(); let from_tuple = Term::from(Tuple::from(vec![ Term::from(from_pid.clone()), call_ref.to_term(), ]));
let rex_message = Term::from(Tuple::from(vec![ Term::from(Atom::from("$gen_call")), from_tuple, call_tuple, ]));
let message = Message::reg_send(from_pid, Atom::from("rex"), rex_message);
let result = request_sender .call(message, call_ref) .await .map_err(RpcError::Connection)?;
check_badrpc_response(mode, result, node, module, function)}
/// Turns a `{badrpc, Reason}` answer into the matching error.fn check_badrpc_response( mode: FormatterMode, term: Term, node: &str, module: &str, function: &str,) -> RpcResult<Term> { if let Term::Tuple(ref tuple) = term { let elements = tuple.elements.as_slice(); if elements.len() == 2 && let Term::Atom(ref atom) = elements[0] && atom.name == "badrpc" { if is_undef(&elements[1]) { return Err(RpcError::Undef { node: node.to_string(), module: module.to_string(), function: function.to_string(), }); }
let reason = format_term_for_error(mode, &elements[1]); return Err(RpcError::BadRpc { node: node.to_string(), module: module.to_string(), function: function.to_string(), reason, }); } } Ok(term)}
/// Recognises `{'EXIT', {undef, _}}` and `{undef, _}`, which is how a node/// reports being asked for a function it has not loaded.fn is_undef(reason: &Term) -> bool { let Term::Tuple(tuple) = reason else { return false; }; match tuple.elements.as_slice() { [Term::Atom(tag), _] if tag.name == "undef" => true, [Term::Atom(tag), inner] if tag.name == "EXIT" => is_undef(inner), _ => false, }}
/// Render a term for an error message, in the server's configured syntax.pub(crate) fn format_term_for_error(mode: FormatterMode, term: &Term) -> String { crate::formatter::get_formatter(mode).format_term(term)}
pub(crate) fn atom(name: &str) -> Term { Term::from(Atom::from(name))}
/// Create a tuple term. Only the tests build these; the node sends them.#[cfg(test)]pub(crate) fn tuple(elements: Vec<Term>) -> Term { Term::from(Tuple::from(elements))}
/// Create a list term.pub(crate) fn list(elements: Vec<Term>) -> Term { Term::from(List::from(elements))}
/// Create a binary term from bytes. Test-only; [`binary_from_str`] covers the/// one case the server itself sends.#[cfg(test)]pub(crate) fn binary(bytes: Vec<u8>) -> Term { Term::from(eetf::Binary::from(bytes))}
/// Create a binary term from a string.pub(crate) fn binary_from_str(s: &str) -> Term { Term::from(eetf::Binary::from(s.as_bytes().to_vec()))}
/// Create a map term from a vector of key-value pairs.pub(crate) fn map(entries: Vec<(Term, Term)>) -> Term { let map: HashMap<Term, Term> = entries.into_iter().collect(); Term::from(Map::from(map))}
pub(crate) fn extract_atom(term: &Term) -> Option<&str> { match term { Term::Atom(a) => Some(&a.name), _ => None, }}
/// Extract tuple elements from a term.pub(crate) fn extract_tuple(term: &Term) -> Option<&[Term]> { match term { Term::Tuple(t) => Some(&t.elements), _ => None, }}
/// Extract list elements from a term.pub(crate) fn extract_list(term: &Term) -> Option<&[Term]> { match term { Term::List(l) => Some(&l.elements), _ => None, }}
/// Extract binary bytes from a term.pub(crate) fn extract_binary(term: &Term) -> Option<&[u8]> { match term { Term::Binary(b) => Some(&b.bytes), _ => None, }}
/// Extract map as a reference from a term.pub(crate) fn extract_map(term: &Term) -> Option<&HashMap<Term, Term>> { match term { Term::Map(m) => Some(&m.map), _ => None, }}
/// Pull `Value` out of an `{ok, Value}` tuple.pub(crate) fn extract_ok_value(term: &Term) -> Option<&Term> { match extract_tuple(term)? { [head, value] if is_atom(head, "ok") => Some(value), _ => None, }}
/// Check if a term is a specific atom.pub(crate) fn is_atom(term: &Term, name: &str) -> bool { matches!(term, Term::Atom(a) if a.name == name)}
#[cfg(test)]#[allow(clippy::unwrap_used, clippy::expect_used)]mod tests { use super::*; use eetf::FixInteger;
#[test] fn atom_helper() { let term = atom("ok"); assert!(matches!(term, Term::Atom(a) if a.name == "ok")); }
#[test] fn binary_helper() { let term = binary(vec![1, 2, 3]); if let Term::Binary(b) = term { assert_eq!(b.bytes, vec![1, 2, 3]); } else { panic!("Expected Binary"); } }
// ======================================================================== // Extraction helper tests // ========================================================================
#[test] fn extract_atom_success() { let term = Term::from(Atom::from("test")); assert_eq!(extract_atom(&term), Some("test")); }
#[test] fn extract_atom_failure() { let term = Term::from(FixInteger::from(42)); assert_eq!(extract_atom(&term), None); }
#[test] fn extract_tuple_success() { let term = Term::from(Tuple::from(vec![ Term::from(Atom::from("ok")), Term::from(FixInteger::from(1)), ])); let elements = extract_tuple(&term); assert!(elements.is_some()); assert_eq!(elements.unwrap().len(), 2); }
#[test] fn extract_list_success() { let term = Term::from(List::from(vec![Term::from(FixInteger::from(1))])); let elements = extract_list(&term); assert!(elements.is_some()); assert_eq!(elements.unwrap().len(), 1); }
#[test] fn extract_binary_success() { let term = Term::from(eetf::Binary::from(b"test".to_vec())); let bytes = extract_binary(&term); assert_eq!(bytes, Some(b"test".as_slice())); }
#[test] fn extract_map_success() { let mut map_entries = HashMap::new(); map_entries.insert(Term::from(Atom::from("a")), Term::from(FixInteger::from(1))); let term = Term::from(Map::from(map_entries)); let map = extract_map(&term); assert!(map.is_some()); assert_eq!(map.unwrap().len(), 1); }
#[test] fn is_atom_check() { let ok = Term::from(Atom::from("ok")); let error = Term::from(Atom::from("error"));
assert!(is_atom(&ok, "ok")); assert!(!is_atom(&ok, "error")); assert!(is_atom(&error, "error")); }
#[test] fn extract_ok_value_success() { let ok_tuple = Term::from(Tuple::from(vec![ Term::from(Atom::from("ok")), Term::from(FixInteger::from(42)), ])); let value = extract_ok_value(&ok_tuple); assert!(value.is_some()); assert!(matches!(value.unwrap(), Term::FixInteger(i) if i.value == 42)); }
#[test] fn extract_ok_value_failure() { let error_tuple = Term::from(Tuple::from(vec![ Term::from(Atom::from("error")), Term::from(Atom::from("reason")), ])); assert!(extract_ok_value(&error_tuple).is_none()); }
// ======================================================================== // badrpc detection tests // ========================================================================
#[test] fn check_badrpc_detects_badrpc() { let badrpc = Term::from(Tuple::from(vec![ Term::from(Atom::from("badrpc")), Term::from(Atom::from("nodedown")), ]));
let result = check_badrpc_response(FormatterMode::Erlang, badrpc, "node@host", "mod", "fun"); assert!(result.is_err()); if let Err(RpcError::BadRpc { reason, .. }) = result { assert_eq!(reason, "nodedown"); } else { panic!("Expected BadRpc error"); } }
#[test] fn check_badrpc_passes_ok() { let ok = Term::from(Tuple::from(vec![ Term::from(Atom::from("ok")), Term::from(FixInteger::from(42)), ]));
let result = check_badrpc_response(FormatterMode::Erlang, ok, "node@host", "mod", "fun"); assert!(result.is_ok()); }
#[test] fn check_badrpc_passes_plain_value() { let value = Term::from(FixInteger::from(123));
let result = check_badrpc_response(FormatterMode::Erlang, value, "node@host", "mod", "fun"); assert!(result.is_ok()); }
// ======================================================================== // Error formatting tests // ========================================================================
#[test] fn format_term_for_error_atom() { let term = Term::from(Atom::from("test")); assert_eq!(format_term_for_error(FormatterMode::Erlang, &term), "test"); }
#[test] fn format_term_for_error_integer() { let term = Term::from(FixInteger::from(42)); assert_eq!(format_term_for_error(FormatterMode::Erlang, &term), "42"); }
#[test] fn format_term_for_error_tuple() { let term = Term::from(Tuple::from(vec![ Term::from(Atom::from("error")), Term::from(Atom::from("reason")), ])); assert_eq!( format_term_for_error(FormatterMode::Erlang, &term), "{error, reason}" ); }
#[test] fn format_term_for_error_list() { let term = Term::from(List::from(vec![ Term::from(FixInteger::from(1)), Term::from(FixInteger::from(2)), ])); assert_eq!( format_term_for_error(FormatterMode::Erlang, &term), "[1, 2]" ); }
#[test] fn format_term_for_error_honours_the_configured_mode() { let term = Term::from(eetf::Binary::from(b"boom".to_vec())); assert_eq!( format_term_for_error(FormatterMode::Erlang, &term), "<<\"boom\">>" ); assert_eq!( format_term_for_error(FormatterMode::Elixir, &term), "\"boom\"" ); }}