repository·main·Indexed 41 days ago
https://github.com/sile/erl_distA Rust implementation of the Erlang Distribution Protocol (version 0.8.0) that enables Rust applications to communicate with distributed Erlang nodes. It provides tools for node discovery via the Erlang Port Mapper Daemon (EPMD), handshake execution using ClientSideHandshake, and bidirectional message passing through asynchronous channels.
erl_dist is a Rust implementation of the Erlang Distribution Protocol. It allows Rust applications to communicate with distributed Erlang nodes, enabling features like node discovery via EPMD and message passing between Rust and Erlang environments.Many distribution messages have a corresponding version that includes a trace_token. These are used for distributed tracing. When you need to propagate trace information, use the Tt variants of the message constructors (e.g., send_tt, exit_tt, spawn_request_tt, alias_send_tt).
// Using a trace token with a Send message
let msg = Message::send_tt(to_pid, message, trace_token);
// Using a trace token with an AliasSend message
let msg = Message::alias_send_tt(from_pid, alias, message, trace_token);The HandshakeStatus enum represents the response from a peer during the first phase of the handshake. It dictates how the handshake should proceed.
| Variant | Meaning |
|---|---|
Ok | The handshake will continue normally. |
OkSimultaneous | The handshake will continue, but indicates a simultaneous connection attempt where the client's name is greater than the server's. |
Nok | The handshake will not continue (the client already has an ongoing handshake). |
NotAllowed | The connection is disallowed for security reasons. |
Alive | A connection to the node is already active; the client must decide whether to continue. |
Named { name, creation } | The handshake continues, but the server has assigned a dynamic node name and creation value to the client. |
To initiate a connection to an Erlang node from the client side, use the ClientSideHandshake struct. The process involves two distinct phases: sending the local node's name and then executing the remainder of the protocol based on the peer's response.
ClientSideHandshake::new(connection, local_node, cookie).execute_send_name(protocol_version) to send your node's identity and receive a HandshakeStatus from the peer.execute_rest(do_continue) to complete the challenge/response and authentication. If the status from Phase 1 was HandshakeStatus::Alive, the do_continue boolean determines if you should proceed with the connection.Upon success, execute_rest returns the underlying connection and the PeerNode information.
// Assuming connection, local_node, and cookie are already defined
let mut handshake = ClientSideHandshake::new(connection, local_node, cookie);
// Phase 1: Send name and get status
let status = handshake.execute_send_name(LOWEST_DISTRIBUTION_PROTOCOL_VERSION).await?;
// Phase 2: Complete the handshake
// Pass `true` to `do_continue` if you want to proceed with the connection
let (connection, peer_node) = handshake.execute_rest(true).await?;To accept an incoming connection from an Erlang node, use the ServerSideHandshake struct. The server must first receive the peer's name and then respond with a status to complete the handshake.
ServerSideHandshake::new(connection, local_node, cookie).execute_recv_name() to receive the peer's node name. This returns Ok(None) if the peer requested a dynamic node name, or Ok(Some(NodeName)) if the name is fixed.execute_rest(status) where status is the HandshakeStatus you wish to grant the peer (e.g., HandshakeStatus::Ok).Upon success, execute_rest returns the underlying connection and the PeerNode information.
// Assuming connection, local_node, and cookie are already defined
let mut handshake = ServerSideHandshake::new(connection, local_node, cookie);
// Phase 1: Receive peer name
let peer_name = handshake.execute_recv_name().await?;
// Phase 2: Respond with status and complete handshake
let (connection, peer_node) = handshake.execute_rest(HandshakeStatus::Ok).await?;Use the EpmdClient to query the Erlang Port Mapper Daemon (EPMD) to retrieve information about registered Erlang nodes. This typically involves connecting to the local EPMD using DEFAULT_EPMD_PORT via a TcpStream.
use erl_dist::epmd::{DEFAULT_EPMD_PORT, EpmdClient};
use tokio::net::TcpStream;
// Connect to the local EPMD.
let connection = TcpStream::connect(("localhost", DEFAULT_EPMD_PORT)).await?;
let client = EpmdClient::new(connection);
// Get the information of a node.
let node_name = "foo";
if let Some(node) = client.get_node(node_name).await? {
println!("Found: {:?}", node);
} else {
println!("Not found");
}To communicate with a remote Erlang node, you must perform a handshake using ClientSideHandshake, establish a channel, and then send a message.
Steps:
TcpStream (port usually retrieved from EPMD).LocalNode and Creation.ClientSideHandshake with the node's name and the distribution cookie.LOWEST_DISTRIBUTION_PROTOCOL_VERSION for the initial handshake step.channel() with capability flags derived from the intersection of local and peer node flags.Message::reg_send) and send it via the channel's transmitter (tx).use erl_dist::LOWEST_DISTRIBUTION_PROTOCOL_VERSION;
use erl_dist::node::{Creation, LocalNode};
use erl_dist::handshake::ClientSideHandshake;
use erl_dist::term::{Atom, Pid};
use erl_dist::message::{channel, Message};
use tokio::net::TcpStream;
// Connect to a peer node.
let peer_host = "localhost";
let peer_port = 7483; // NOTE: Usually, port number is retrieved from EPMD.
let connection = TcpStream::connect((peer_host, peer_port)).await?;
// Local node information.
let creation = Creation::random();
let local_node = LocalNode::new("foo@localhost".parse()?, creation);
// Do handshake.
let mut handshake = ClientSideHandshake::new(connection, local_node.clone(), "cookie");
let _status = handshake.execute_send_name(LOWEST_DISTRIBUTION_PROTOCOL_VERSION).await?;
let (connection, peer_node) = handshake.execute_rest(true).await?;
// Create a channel.
let capability_flags = local_node.flags & peer_node.flags;
let (mut tx, _) = channel(connection, capability_flags);
// Send a message.
let from_pid = Pid::new(local_node.name.to_string(), 0, 0, local_node.creation.get());
let to_name = Atom::from("bar");
let msg = Message::reg_send(from_pid, to_name, Atom::from("hello").into());
tx.send(msg).await?;Use EpmdClient::get_names to retrieve a list of all currently registered nodes in the form of (String, u16) pairs, representing the node name and its port.
let names = client.get_names().await?;
for (name, port) in names {
println!("Node: {}, Port: {}", name, port);
}The try_from_term<T>(term: Term, expected: &str) function is a helper for converting a Term into a type T that implements TryInto<T>. It is particularly useful when you want to provide a custom error message indicating what type was expected if the conversion fails.
If the conversion fails, it returns DecodeError::UnexpectedType with the expected string provided as the error context.
// Example: Using try_from_term with a custom expected type label
let val: MyType = try_from_term(term, "my custom type")?;The HandshakeMessageReader is used to read a single handshake message from a Connection. You first initialize it using connection.handshake_message_reader(), which reads a u16 length prefix from the stream. Subsequent calls to read_* methods will only consume bytes up to that specified length. Calling finish() ensures any remaining bytes in the declared message size are consumed.
let mut reader = connection.handshake_message_reader().await?;
// Reads are constrained by the initial size prefix
let tag = reader.read_u8().await?;
let payload = reader.read_bytes().await?;
// Consumes any remaining bytes in the message and returns
reader.finish().await?;To interact with the Erlang Port Mapper Daemon (EPMD), use the EpmdClient struct. You must provide a connection that implements AsyncRead, AsyncWrite, and Unpin (such as a TCP stream).
use erl_dist::epmd::EpmdClient;
// Assuming `stream` is an object implementing AsyncRead + AsyncWrite + Unpin
let client = EpmdClient::new(stream);The library provides two primary structures to represent node identities:
LocalNode: Represents the identity of the current node. It includes a name, flags (of type DistributionFlags), and a creation identifier.PeerNode: Represents the identity of a remote node discovered in the distribution. It includes a name, flags, and an optional creation identifier (Option<Creation>). The creation field is optional because older nodes might not provide this information during the handshake.Use LocalNode::new(name, creation) to initialize a local node with default distribution flags.
use erl_dist::{LocalNode, NodeName, Creation};
let name = NodeName::new("node1", "host1").unwrap();
let creation = Creation::new(1);
let local = LocalNode::new(name, creation);