nandi.uk
Distributed Erlang over iroh P2P + TLS, with a bot powers SDK
1//! Classic TCP transport (+ optional EPMD lookup).
2
3use crate::error::{Error, Result};
4use crate::stream::DistStream;
5use erl_dist::epmd::DEFAULT_EPMD_PORT;
6use erl_dist::node::NodeName;
7use std::net::SocketAddr;
8use tokio::io::{AsyncReadExt, AsyncWriteExt};
9use tokio::net::TcpStream;
10use tracing::debug;
11
12/// TCP carrier helpers.
13pub struct TcpTransport;
14
15impl TcpTransport {
16 /// Connect to `host:port` and wrap as a [`DistStream`].
17 pub async fn connect(host: &str, port: u16) -> Result<DistStream> {
18 debug!(%host, port, "tcp connect");
19 let stream = TcpStream::connect((host, port)).await?;
20 stream.set_nodelay(true)?;
21 Ok(DistStream::from_rw(stream))
22 }
23
24 /// Connect to a socket address.
25 pub async fn connect_addr(addr: SocketAddr) -> Result<DistStream> {
26 debug!(%addr, "tcp connect");
27 let stream = TcpStream::connect(addr).await?;
28 stream.set_nodelay(true)?;
29 Ok(DistStream::from_rw(stream))
30 }
31
32 /// Resolve a BEAM node name via EPMD, then TCP-connect to its port.
33 ///
34 /// `node` is a full name like `foo@localhost`.
35 pub async fn connect_node(node: &str) -> Result<DistStream> {
36 let name: NodeName = node
37 .parse()
38 .map_err(|e: erl_dist::node::NodeNameError| Error::InvalidName(e.to_string()))?;
39 let port = epmd_lookup(name.host(), name.name()).await?;
40 Self::connect(name.host(), port).await
41 }
42
43 /// Look up a short node name on a host's EPMD.
44 pub async fn epmd_port(host: &str, short_name: &str) -> Result<u16> {
45 epmd_lookup(host, short_name).await
46 }
47}
48
49async fn epmd_lookup(host: &str, short_name: &str) -> Result<u16> {
50 // Minimal EPMD PORT_PLEASE2 over tokio TCP.
51 let mut stream = TcpStream::connect((host, DEFAULT_EPMD_PORT)).await?;
52 // PORT_PLEASE2_REQ: 122, then name bytes. Length-prefixed packet (2-byte BE).
53 let mut body = Vec::with_capacity(1 + short_name.len());
54 body.push(122u8); // PORT_PLEASE2_REQ
55 body.extend_from_slice(short_name.as_bytes());
56 let mut packet = Vec::with_capacity(2 + body.len());
57 packet.extend_from_slice(&(body.len() as u16).to_be_bytes());
58 packet.extend_from_slice(&body);
59 stream.write_all(&packet).await?;
60
61 // Response is NOT length-prefixed: result(1) [+ port(2) + …] or result≠0 (not found)
62 let mut result = [0u8; 1];
63 stream.read_exact(&mut result).await?;
64 if result[0] != 0 {
65 return Err(Error::epmd(format!(
66 "node `{short_name}` not registered on EPMD at {host}"
67 )));
68 }
69 let mut port_buf = [0u8; 2];
70 stream.read_exact(&mut port_buf).await?;
71 let port = u16::from_be_bytes(port_buf);
72 debug!(%host, %short_name, port, "epmd lookup ok");
73 Ok(port)
74}