[WIP] [Does not compile] Exploring dynamic discovery mechanism

.vscode/launch.json

@@ -0,0 +1,83 @@
+{
+    // Use IntelliSense to learn about possible attributes.
+    // Hover to view descriptions of existing attributes.
+    // For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
+    "version": "0.2.0",
+    "configurations": [
+        {
+            "type": "lldb",
+            "request": "launch",
+            "name": "Debug unit tests in library 'hyperswarm'",
+            "cargo": {
+                "args": [
+                    "test",
+                    "--no-run",
+                    "--lib",
+                    "--package=hyperswarm"
+                ],
+                "filter": {
+                    "name": "hyperswarm",
+                    "kind": "lib"
+                }
+            },
+            "args": [],
+            "cwd": "${workspaceFolder}"
+        },
+        {
+            "type": "lldb",
+            "request": "launch",
+            "name": "Debug example 'simple'",
+            "cargo": {
+                "args": [
+                    "build",
+                    "--example=simple",
+                    "--package=hyperswarm"
+                ],
+                "filter": {
+                    "name": "simple",
+                    "kind": "example"
+                }
+            },
+            "args": [],
+            "cwd": "${workspaceFolder}"
+        },
+        {
+            "type": "lldb",
+            "request": "launch",
+            "name": "Debug unit tests in example 'simple'",
+            "cargo": {
+                "args": [
+                    "test",
+                    "--no-run",
+                    "--example=simple",
+                    "--package=hyperswarm"
+                ],
+                "filter": {
+                    "name": "simple",
+                    "kind": "example"
+                }
+            },
+            "args": [],
+            "cwd": "${workspaceFolder}"
+        },
+        {
+            "type": "lldb",
+            "request": "launch",
+            "name": "Debug integration test 'test'",
+            "cargo": {
+                "args": [
+                    "test",
+                    "--no-run",
+                    "--test=test",
+                    "--package=hyperswarm"
+                ],
+                "filter": {
+                    "name": "test",
+                    "kind": "test"
+                }
+            },
+            "args": [],
+            "cwd": "${workspaceFolder}"
+        }
+    ]
+}

Cargo.toml

@@ -29,7 +29,7 @@ multicast-socket = "0.2.1"
 hex = "0.4.3"
 pretty-hash = "0.4.1"
 hyperswarm-dht = { git = "https://github.com/Frando/hyperswarm-dht.git", branch = "hyperspace" }
-colmeia-hyperswarm-mdns = { git = "https://github.com/bltavares/colmeia.git", rev = "e92ab71981356197a21592b7ce6854e209582985" }
+colmeia-hyperswarm-mdns = { git = "https://github.com/bltavares/colmeia.git", rev = "53761799f7a9ee123875534e0108d7483a117885" }
 libutp-rs = { git = "https://github.com/Frando/libutp-rs.git", branch = "feat/clone", optional = true }
 
 [dev-dependencies]

examples/lan.rs

@@ -0,0 +1,80 @@
+use async_std::prelude::*;
+use async_std::stream::StreamExt;
+use async_std::task;
+// use std::net::{SocketAddr, ToSocketAddrs};
+
+use hyperswarm::{
+    discovery::mdns::MdnsDiscovery, DhtOptions, Hyperswarm, HyperswarmStream, TopicConfig,
+};
+
+#[async_std::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::init();
+
+    let transport = Hyperswarm::listen().await?;
+    let discovery = MdnsDiscovery::bind(transport.local_addr().port()).await?;
+    let mut swarm1 = Hyperswarm::with(transport, discovery);
+
+    let transport = Hyperswarm::listen().await?;
+    let discovery = MdnsDiscovery::bind(transport.local_addr().port()).await?;
+    let mut swarm2 = Hyperswarm::with(transport, discovery);
+
+    let handle1 = swarm1.handle();
+    let handle2 = swarm2.handle();
+
+    let task1 = task::spawn(async move {
+        while let Some(stream) = swarm1.next().await {
+            let stream = stream.unwrap();
+            on_connection(stream, "rust1".into());
+        }
+    });
+
+    let task2 = task::spawn(async move {
+        while let Some(stream) = swarm2.next().await {
+            let stream = stream.unwrap();
+            on_connection(stream, "rust2".into());
+        }
+    });
+
+    let topic = [0u8; 32];
+    handle1.configure(topic, TopicConfig::both());
+    handle2.configure(topic, TopicConfig::both());
+
+    task1.await;
+    task2.await;
+
+    Ok(())
+}
+
+fn on_connection(mut stream: HyperswarmStream, local_name: String) {
+    let label = format!(
+        "[{} -> {}://{}]",
+        local_name,
+        stream.protocol(),
+        stream.peer_addr()
+    );
+    eprintln!("{} connect", label);
+    task::spawn(async move {
+        stream
+            .write_all(format!("hi from {}", local_name).as_bytes())
+            .await
+            .unwrap();
+        let mut buf = vec![0u8; 100];
+        loop {
+            match stream.read(&mut buf).await {
+                Ok(n) if n > 0 => {
+                    let text = String::from_utf8(buf[..n].to_vec()).unwrap();
+                    eprintln!("{} read: {}", label, text);
+                }
+                Ok(_) => {
+                    eprintln!("{} close", label);
+                    break;
+                }
+                Err(e) => {
+                    eprintln!("{} error: {}", label, e);
+                    break;
+                }
+            }
+        }
+    });
+}

examples/simple.rs

@@ -3,7 +3,7 @@ use async_std::stream::StreamExt;
 use async_std::task;
 // use std::net::{SocketAddr, ToSocketAddrs};
 
-use hyperswarm::{run_bootstrap_node, Config, Hyperswarm, HyperswarmStream, TopicConfig};
+use hyperswarm::{run_bootstrap_node, DhtOptions, Hyperswarm, HyperswarmStream, TopicConfig};
 
 #[async_std::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -12,7 +12,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let (bs_addr, bs_task) = run_bootstrap_node(Some(bs_addr)).await?;
     // let bs_addr: SocketAddr = bs_addr.to_socket_addrs().unwrap().next().unwrap();
 
-    let config = Config::default().set_bootstrap_nodes(Some(vec![bs_addr]));
+    let config = DhtOptions::default().set_bootstrap_nodes(Some(vec![bs_addr]));
 
     let mut swarm1 = Hyperswarm::bind(config.clone()).await?;
     let mut swarm2 = Hyperswarm::bind(config).await?;

src/config.rs

@@ -1,12 +1,12 @@
 use std::net::SocketAddr;
 
 #[derive(Debug, Default, Clone)]
-pub struct Config {
+pub struct DhtOptions {
     pub bootstrap: Option<Vec<SocketAddr>>,
     pub ephemeral: bool,
 }
 
-impl Config {
+impl DhtOptions {
     pub fn set_bootstrap_nodes(mut self, nodes: Option<Vec<SocketAddr>>) -> Self {
         self.bootstrap = nodes;
         self

src/discovery/combined.rs

@@ -7,7 +7,7 @@ use std::task::{Context, Poll};
 use super::dht::DhtDiscovery;
 use super::mdns::MdnsDiscovery;
 use super::{Discovery, PeerInfo, Topic};
-use crate::config::Config;
+use crate::config::DhtOptions;
 
 #[derive(Debug)]
 pub struct CombinedDiscovery {
@@ -16,8 +16,8 @@ pub struct CombinedDiscovery {
 }
 
 impl CombinedDiscovery {
-    pub async fn bind(local_port: u16, config: Config) -> io::Result<Self> {
-        let mdns = MdnsDiscovery::bind(local_port, config.clone()).await?;
+    pub async fn bind(local_port: u16, config: DhtOptions) -> io::Result<Self> {
+        let mdns = MdnsDiscovery::bind(local_port).await?;
         let dht = DhtDiscovery::bind(local_port, config).await?;
         Ok(Self { mdns, dht })
     }

src/discovery/dht.rs

@@ -8,7 +8,7 @@ use std::io;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use crate::config::Config;
+use crate::DhtOptions;
 
 use super::{Discovery, DiscoveryMethod, PeerInfo, Topic};
 
@@ -37,7 +37,7 @@ enum Command {
 }
 
 impl DhtDiscovery {
-    pub async fn bind(local_port: u16, config: Config) -> io::Result<Self> {
+    pub async fn bind(local_port: u16, config: DhtOptions) -> io::Result<Self> {
         let dht_config = DhtConfig::default();
         let dht_config = if let Some(bootstrap) = config.bootstrap.as_ref() {
             dht_config.set_bootstrap_nodes(bootstrap)

src/discovery/mdns.rs

@@ -2,6 +2,7 @@ use async_std::channel;
 use async_std::stream::Stream;
 use async_std::task::{Context, Poll};
 use colmeia_hyperswarm_mdns::{self_id, Announcer, Locator};
+use futures::FutureExt;
 use futures_lite::ready;
 // use log::*;
 use std::convert::TryInto;
@@ -11,8 +12,6 @@ use std::io;
 use std::pin::Pin;
 use std::time::Duration;
 
-use crate::Config;
-
 use super::{Discovery, DiscoveryMethod, PeerInfo, Topic};
 
 mod socket {
@@ -52,7 +51,7 @@ impl fmt::Debug for MdnsDiscovery {
 }
 
 impl MdnsDiscovery {
-    pub async fn bind(local_port: u16, _config: Config) -> io::Result<Self> {
+    pub async fn bind(local_port: u16) -> io::Result<Self> {
         let self_id = self_id();
         let socket = socket::create()?;
         let lookup_interval = Duration::from_secs(60);
@@ -108,29 +107,26 @@ impl Stream for MdnsDiscovery {
             return Poll::Ready(Some(Err(e)));
         }
 
-        if let Poll::Ready(Some(_command)) = Pin::new(&mut this.pending_commands_rx).poll_next(cx) {
-            // TODO: Boxing the add_topic future does not work because there's no valid
-            // lifetime. Best would be to make the add_topic functions sync, or return
-            // a future that can be boxed.
-            // let fut = match command {
-            //     Command::Lookup(topic) => {
-            //         let fut = this.locator.add_topic(&topic);
-            //         let fut = fut.map(|r| {
-            //             r.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{}", e)))
-            //         });
-            //         let fut: CommandFut = fut.boxed();
-            //         fut
-            //     }
-            //     Command::Announce(topic) => {
-            //         let fut = this.announcer.add_topic(&topic);
-            //         let fut = fut.map(|r| {
-            //             r.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{}", e)))
-            //         });
-            //         let fut: CommandFut = fut.boxed();
-            //         fut
-            //     }
-            // };
-            // this.pending_future = Some(fut);
+        if let Poll::Ready(Some(command)) = Pin::new(&mut this.pending_commands_rx).poll_next(cx) {
+            let fut = match command {
+                Command::Lookup(topic) => {
+                    let fut = this.locator.add_topic(topic.to_vec());
+                    let fut = fut.map(|r| {
+                        r.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{}", e)))
+                    });
+                    let fut: CommandFut = fut.boxed();
+                    fut
+                }
+                Command::Announce(topic) => {
+                    let fut = this.announcer.add_topic(topic.to_vec());
+                    let fut = fut.map(|r| {
+                        r.map_err(|e| io::Error::new(io::ErrorKind::Other, format!("{}", e)))
+                    });
+                    let fut: CommandFut = fut.boxed();
+                    fut
+                }
+            };
+            this.pending_future = Some(fut);
         }
 
         if let Err(e) = ready!(this.poll_pending_future(cx)) {

src/lib.rs

@@ -18,7 +18,7 @@ pub mod discovery;
 pub mod transport;
 
 pub use bootstrap::run_bootstrap_node;
-pub use config::{Config, TopicConfig};
+pub use config::{DhtOptions, TopicConfig};
 pub use swarm::Hyperswarm;
 
 use transport::combined::CombinedStream;