feat: allow specifying custom relay in dumbpipe (#60)

README.md

@@ -173,3 +173,38 @@ echo request1.bin | dumbpipe connect <ticket> --custom-alpn utf8:/iroh-bytes/2 >
 
 if request1.bin contained a valid request for the `/iroh-bytes/2` protocol, response1.bin will
 now contain the response.
+
+## Custom Relay Configuration
+
+By default, dumbpipe uses iroh's automatic relay selection, which picks the fastest responding relay server from the n0 network. You can customize this behavior using the `--relay` option:
+
+### Disable relays entirely
+
+If you want to force direct connections only:
+
+```bash
+dumbpipe listen --relay disabled
+dumbpipe connect <ticket> --relay disabled
+```
+
+This will only attempt direct peer-to-peer connections and won't fall back to relay servers.
+
+### Use default relays (default behavior)
+
+```bash
+dumbpipe listen --relay default
+dumbpipe connect <ticket> --relay default
+```
+
+This is the default behavior when no `--relay` option is specified.
+
+### Use a custom relay server
+
+If you're running your own relay server or want to use a specific one:
+
+```bash
+dumbpipe listen --relay https://your-relay-server.com
+dumbpipe connect <ticket> --relay https://your-relay-server.com
+```
+
+**Note**: Both the listener and connector should use the same relay configuration for optimal connectivity.

src/main.rs

@@ -1,12 +1,13 @@
 //! Command line arguments.
 use clap::{Parser, Subcommand};
 use dumbpipe::NodeTicket;
-use iroh::{endpoint::Connecting, Endpoint, NodeAddr, SecretKey};
+use iroh::{endpoint::Connecting, Endpoint, NodeAddr, RelayMode, RelayUrl, SecretKey};
 use n0_snafu::{Result, ResultExt};
 use std::{
+    fmt::{Display, Formatter},
     io,
     net::{SocketAddr, SocketAddrV4, SocketAddrV6, ToSocketAddrs},
-    str::FromStr,
+    str::FromStr, time::Duration,
 };
 use tokio::{
     io::{AsyncRead, AsyncWrite, AsyncWriteExt},
@@ -122,6 +123,14 @@ pub struct CommonArgs {
     #[clap(long)]
     pub custom_alpn: Option<String>,
 
+    /// The relay URL to use as a home relay,
+    ///
+    /// Can be set to "disabled" to disable relay servers, or to a URL
+    /// to configure custom servers. By default, the quickest responding
+    /// n0 relay is used if this option is not set.
+    #[clap(long, default_value_t = RelayModeOption::Default)]
+    pub relay: RelayModeOption,
+
     /// The verbosity level. Repeat to increase verbosity.
     #[clap(short = 'v', long, action = clap::ArgAction::Count)]
     pub verbose: u8,
@@ -148,6 +157,49 @@ fn parse_alpn(alpn: &str) -> Result<Vec<u8>> {
     })
 }
 
+/// Available command line options for configuring relays.
+#[derive(Clone, Debug, PartialEq)]
+pub enum RelayModeOption {
+    /// Disables relays altogether.
+    Disabled,
+    /// Uses the default relay servers.
+    Default,
+    /// Uses a single, custom relay server by URL.
+    Custom(RelayUrl),
+}
+
+impl FromStr for RelayModeOption {
+    type Err = iroh::RelayUrlParseError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        match s {
+            "disabled" => Ok(Self::Disabled),
+            "default" => Ok(Self::Default),
+            _ => Ok(Self::Custom(RelayUrl::from_str(s)?)),
+        }
+    }
+}
+
+impl Display for RelayModeOption {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Disabled => f.write_str("disabled"),
+            Self::Default => f.write_str("default"),
+            Self::Custom(url) => url.fmt(f),
+        }
+    }
+}
+
+impl From<RelayModeOption> for RelayMode {
+    fn from(value: RelayModeOption) -> Self {
+        match value {
+            RelayModeOption::Disabled => RelayMode::Disabled,
+            RelayModeOption::Default => RelayMode::Default,
+            RelayModeOption::Custom(url) => RelayMode::Custom(url.into()),
+        }
+    }
+}
+
 #[derive(Parser, Debug)]
 pub struct ListenArgs {
     /// Immediately close our sending side, indicating that we will not transmit any data
@@ -291,14 +343,24 @@ async fn create_endpoint(
     common: &CommonArgs,
     alpns: Vec<Vec<u8>>,
 ) -> Result<Endpoint> {
-    let mut builder = Endpoint::builder().secret_key(secret_key).alpns(alpns);
+    let mut builder = Endpoint::builder()
+        .secret_key(secret_key)
+        .alpns(alpns)
+        .relay_mode(common.relay.clone().into());
     if let Some(addr) = common.ipv4_addr {
         builder = builder.bind_addr_v4(addr);
     }
     if let Some(addr) = common.ipv6_addr {
         builder = builder.bind_addr_v6(addr);
     }
     let endpoint = builder.bind().await?;
+
+    let _ = tokio::time::timeout(Duration::from_secs(30), async {
+        if !(common.relay == RelayModeOption::Disabled) {
+            endpoint.online().await;
+        }
+    }).await;
+
     Ok(endpoint)
 }
 
@@ -344,8 +406,7 @@ async fn forward_bidi(
 async fn listen_stdio(args: ListenArgs) -> Result<()> {
     let secret_key = get_or_create_secret()?;
     let endpoint = create_endpoint(secret_key, &args.common, vec![args.common.alpn()?]).await?;
-    // wait for the endpoint to figure out its home relay and addresses before making a ticket
-    endpoint.online().await;
+    // wait for the endpoint to figure out its address before making a ticket
     let node = endpoint.node_addr();
     let mut short = node.clone();
     let ticket = NodeTicket::new(node);
@@ -443,9 +504,6 @@ async fn connect_tcp(args: ConnectTcpArgs) -> Result<()> {
         .context("unable to bind endpoint")?;
     tracing::info!("tcp listening on {:?}", addrs);
 
-    // Wait for our own endpoint to be ready before trying to connect.
-    endpoint.online().await;
-
     let tcp_listener = match tokio::net::TcpListener::bind(addrs.as_slice()).await {
         Ok(tcp_listener) => tcp_listener,
         Err(cause) => {
@@ -516,8 +574,6 @@ async fn listen_tcp(args: ListenTcpArgs) -> Result<()> {
     };
     let secret_key = get_or_create_secret()?;
     let endpoint = create_endpoint(secret_key, &args.common, vec![args.common.alpn()?]).await?;
-    // wait for the endpoint to figure out its address before making a ticket
-    endpoint.online().await;
     let node_addr = endpoint.node_addr();
     let mut short = node_addr.clone();
     let ticket = NodeTicket::new(node_addr);
@@ -598,8 +654,6 @@ async fn listen_unix(args: ListenUnixArgs) -> Result<()> {
     let socket_path = args.socket_path.clone();
     let secret_key = get_or_create_secret()?;
     let endpoint = create_endpoint(secret_key, &args.common, vec![args.common.alpn()?]).await?;
-    // wait for the endpoint to figure out its address before making a ticket
-    endpoint.online().await;
     let node_addr = endpoint.node_addr();
     let mut short = node_addr.clone();
     let ticket = NodeTicket::new(node_addr);
@@ -713,9 +767,6 @@ async fn connect_unix(args: ConnectUnixArgs) -> Result<()> {
         .context("unable to bind endpoint")?;
     tracing::info!("unix listening on {:?}", socket_path);
 
-    // Wait for our own endpoint to be ready before trying to connect.
-    endpoint.online().await;
-
     // Remove existing socket file if it exists
     if let Err(e) = tokio::fs::remove_file(&socket_path).await {
         if e.kind() != io::ErrorKind::NotFound {

tests/cli.rs

@@ -488,3 +488,84 @@ mod unix_socket_tests {
         connect_stderr_thread.join().ok();
     }
 }
+
+#[test]
+#[ignore = "flaky"]
+fn connect_listen_relay_disabled() {
+    let listen_to_connect = b"hello from listen";
+    let connect_to_listen = b"hello from connect";
+    let mut listen = duct::cmd(dumbpipe_bin(), ["listen", "--relay", "disabled"])
+        .env_remove("RUST_LOG")
+        .stdin_bytes(listen_to_connect)
+        .stderr_to_stdout()
+        .reader()
+        .unwrap();
+    let header = read_ascii_lines(3, &mut listen).unwrap();
+    let header = String::from_utf8(header).unwrap();
+    let ticket = header.split_ascii_whitespace().last().unwrap();
+    let ticket = NodeTicket::from_str(ticket).unwrap();
+
+    let connect = duct::cmd(
+        dumbpipe_bin(),
+        ["connect", &ticket.to_string(), "--relay", "disabled"],
+    )
+    .env_remove("RUST_LOG")
+    .stdin_bytes(connect_to_listen)
+    .stderr_null()
+    .stdout_capture()
+    .run()
+    .unwrap();
+
+    assert!(connect.status.success());
+    assert_eq!(&connect.stdout, listen_to_connect);
+
+    let mut listen_stdout = Vec::new();
+    listen.read_to_end(&mut listen_stdout).unwrap();
+    assert_eq!(&listen_stdout, connect_to_listen);
+}
+
+#[test]
+#[ignore = "flaky"]
+fn connect_listen_relay_default() {
+    let listen_to_connect = b"hello from listen";
+    let connect_to_listen = b"hello from connect";
+    let mut listen = duct::cmd(dumbpipe_bin(), ["listen", "--relay", "default"])
+        .env_remove("RUST_LOG")
+        .stdin_bytes(listen_to_connect)
+        .stderr_to_stdout()
+        .reader()
+        .unwrap();
+    let header = read_ascii_lines(3, &mut listen).unwrap();
+    let header = String::from_utf8(header).unwrap();
+    let ticket = header.split_ascii_whitespace().last().unwrap();
+    let ticket = NodeTicket::from_str(ticket).unwrap();
+
+    let connect = duct::cmd(
+        dumbpipe_bin(),
+        ["connect", &ticket.to_string(), "--relay", "default"],
+    )
+    .env_remove("RUST_LOG")
+    .stdin_bytes(connect_to_listen)
+    .stderr_null()
+    .stdout_capture()
+    .run()
+    .unwrap();
+
+    assert!(connect.status.success());
+    assert_eq!(&connect.stdout, listen_to_connect);
+
+    let mut listen_stdout = Vec::new();
+    listen.read_to_end(&mut listen_stdout).unwrap();
+    assert_eq!(&listen_stdout, connect_to_listen);
+}
+
+#[test]
+fn relay_option_invalid() {
+    let output = duct::cmd(dumbpipe_bin(), ["listen", "--relay", "invalid-relay-url"])
+        .env_remove("RUST_LOG")
+        .stderr_capture()
+        .stdout_capture()
+        .run();
+
+    assert!(output.is_err() || !output.unwrap().status.success());
+}