From faf558053aa3365eeb760f62014b304e72d34ed2 Mon Sep 17 00:00:00 2001 From: mfahampshire Date: Wed, 23 Aug 2023 20:47:24 +0200 Subject: [PATCH] working on rendzv example --- sdk/rust/nym-sdk/Cargo.toml | 2 +- .../examples/libp2p_rendezvous/README.md | 0 .../libp2p_rendezvous/bin/register.rs | 127 ++++++++++++++++ .../examples/libp2p_rendezvous/main.rs | 135 ++++++++++++++++++ 4 files changed, 263 insertions(+), 1 deletion(-) create mode 100644 sdk/rust/nym-sdk/examples/libp2p_rendezvous/README.md create mode 100644 sdk/rust/nym-sdk/examples/libp2p_rendezvous/bin/register.rs create mode 100644 sdk/rust/nym-sdk/examples/libp2p_rendezvous/main.rs diff --git a/sdk/rust/nym-sdk/Cargo.toml b/sdk/rust/nym-sdk/Cargo.toml index f3d1d47aae..46f521ab5b 100644 --- a/sdk/rust/nym-sdk/Cargo.toml +++ b/sdk/rust/nym-sdk/Cargo.toml @@ -38,7 +38,7 @@ tokio = { version = "1", features = ["full"] } nym-bin-common = { path = "../../../common/bin-common" } # extra dependencies for libp2p examples -libp2p = { git = "https://github.com/ChainSafe/rust-libp2p.git", rev = "e3440d25681df380c9f0f8cfdcfd5ecc0a4f2fb6", features = [ "identify", "macros", "ping", "tokio", "tcp", "dns", "websocket", "noise", "mplex", "yamux", "gossipsub" ]} +libp2p = { git = "https://github.com/ChainSafe/rust-libp2p.git", rev = "e3440d25681df380c9f0f8cfdcfd5ecc0a4f2fb6", features = [ "identify", "macros", "ping", "tokio", "tcp", "dns", "websocket", "noise", "mplex", "yamux", "gossipsub", "rendezvous" ]} tokio-stream = "0.1.12" tokio-util = { version = "0.7", features = ["codec"] } parking_lot = "0.12" diff --git a/sdk/rust/nym-sdk/examples/libp2p_rendezvous/README.md b/sdk/rust/nym-sdk/examples/libp2p_rendezvous/README.md new file mode 100644 index 0000000000..e69de29bb2 diff --git a/sdk/rust/nym-sdk/examples/libp2p_rendezvous/bin/register.rs b/sdk/rust/nym-sdk/examples/libp2p_rendezvous/bin/register.rs new file mode 100644 index 0000000000..4d903027c1 --- /dev/null +++ b/sdk/rust/nym-sdk/examples/libp2p_rendezvous/bin/register.rs @@ -0,0 +1,127 @@ + +// use futures::StreamExt; +// use libp2p::{ +// core::transport::upgrade::Version, +// identity, noise, ping, rendezvous, +// swarm::{keep_alive, NetworkBehaviour, SwarmBuilder, SwarmEvent}, +// tcp, yamux, Multiaddr, PeerId, Transport, +// }; +// use std::time::Duration; + + +use libp2p::futures::StreamExt; +use libp2p::ping::Success; +use libp2p::swarm::{keep_alive, NetworkBehaviour, SwarmEvent}; +use libp2p::{identity, ping, Multiaddr, PeerId}; +use log::{debug, info, LevelFilter}; +use nym_sdk::mixnet::MixnetClient; +use std::error::Error; +use std::time::Duration; + +#[tokio::main] +async fn main() { + env_logger::init(); + + let key_pair = identity::Keypair::generate_ed25519(); + let rendezvous_point_address = "/ip4/127.0.0.1/tcp/62649".parse::().unwrap(); + let rendezvous_point = "12D3KooWDpJ7As7BWAwRMfu1VU2WCqNjvq387JEYKDBj4kx6nXTN" + .parse() + .unwrap(); + + let mut swarm = SwarmBuilder::with_tokio_executor( + tcp::tokio::Transport::default() + .upgrade(Version::V1Lazy) + .authenticate(noise::Config::new(&key_pair).unwrap()) + .multiplex(yamux::Config::default()) + .boxed(), + MyBehaviour { + rendezvous: rendezvous::client::Behaviour::new(key_pair.clone()), + ping: ping::Behaviour::new(ping::Config::new().with_interval(Duration::from_secs(1))), + keep_alive: keep_alive::Behaviour, + }, + PeerId::from(key_pair.public()), + ) + .build(); + + // In production the external address should be the publicly facing IP address of the rendezvous point. + // This address is recorded in the registration entry by the rendezvous point. + let external_address = "/ip4/127.0.0.1/tcp/0".parse::().unwrap(); + swarm.add_external_address(external_address); + + log::info!("Local peer id: {}", swarm.local_peer_id()); + + swarm.dial(rendezvous_point_address.clone()).unwrap(); + + while let Some(event) = swarm.next().await { + match event { + SwarmEvent::NewListenAddr { address, .. } => { + log::info!("Listening on {}", address); + } + SwarmEvent::ConnectionClosed { + peer_id, + cause: Some(error), + .. + } if peer_id == rendezvous_point => { + log::error!("Lost connection to rendezvous point {}", error); + } + SwarmEvent::ConnectionEstablished { peer_id, .. } if peer_id == rendezvous_point => { + if let Err(error) = swarm.behaviour_mut().rendezvous.register( + rendezvous::Namespace::from_static("rendezvous"), + rendezvous_point, + None, + ) { + log::error!("Failed to register: {error}"); + return; + } + log::info!("Connection established with rendezvous point {}", peer_id); + } + // once `/identify` did its job, we know our external address and can register + SwarmEvent::Behaviour(MyBehaviourEvent::Rendezvous( + rendezvous::client::Event::Registered { + namespace, + ttl, + rendezvous_node, + }, + )) => { + log::info!( + "Registered for namespace '{}' at rendezvous point {} for the next {} seconds", + namespace, + rendezvous_node, + ttl + ); + } + SwarmEvent::Behaviour(MyBehaviourEvent::Rendezvous( + rendezvous::client::Event::RegisterFailed { + rendezvous_node, + namespace, + error, + }, + )) => { + log::error!( + "Failed to register: rendezvous_node={}, namespace={}, error_code={:?}", + rendezvous_node, + namespace, + error + ); + return; + } + SwarmEvent::Behaviour(MyBehaviourEvent::Ping(ping::Event { + peer, + result: Ok(rtt), + .. + })) if peer != rendezvous_point => { + log::info!("Ping to {} is {}ms", peer, rtt.as_millis()) + } + other => { + log::debug!("Unhandled {:?}", other); + } + } + } +} + +#[derive(NetworkBehaviour)] +struct MyBehaviour { + rendezvous: rendezvous::client::Behaviour, + ping: ping::Behaviour, + keep_alive: keep_alive::Behaviour, +} diff --git a/sdk/rust/nym-sdk/examples/libp2p_rendezvous/main.rs b/sdk/rust/nym-sdk/examples/libp2p_rendezvous/main.rs new file mode 100644 index 0000000000..555b5572e5 --- /dev/null +++ b/sdk/rust/nym-sdk/examples/libp2p_rendezvous/main.rs @@ -0,0 +1,135 @@ + +use futures::StreamExt; +use libp2p::{ + core::transport::upgrade::Version, + identify, identity, noise, ping, rendezvous, + tcp, yamux, PeerId, Transport, +}; +use libp2p::swarm::{keep_alive, NetworkBehaviour, SwarmBuilder, SwarmEvent}; +use std::time::Duration; +use log::{debug, info, LevelFilter}; +use nym_sdk::mixnet::MixnetClient; +use std::error::Error; + +#[path = "../libp2p_shared/lib.rs"] +mod rust_libp2p_nym; + +#[tokio::main] +async fn main() -> Result<(), Box> { + pretty_env_logger::formatted_timed_builder() + .filter_level(LevelFilter::Warn) + .filter(Some("libp2p_rendezvous"), LevelFilter::Debug) + .init(); + + let local_key = identity::Keypair::generate_ed25519(); + let local_peer_id = PeerId::from(local_key.public()); + info!("Local peer id: {local_peer_id:?}"); + + // let mut swarm = SwarmBuilder::with_tokio_executor( + // tcp::tokio::Transport::default() + // .upgrade(Version::V1Lazy) + // .authenticate(noise::Config::new(&key_pair).unwrap()) + // .multiplex(yamux::Config::default()) + // .boxed(), + // MyBehaviour { + // identify: identify::Behaviour::new(identify::Config::new( + // "rendezvous-example/1.0.0".to_string(), + // key_pair.public(), + // )), + // rendezvous: rendezvous::server::Behaviour::new(rendezvous::server::Config::default()), + // ping: ping::Behaviour::new(ping::Config::new().with_interval(Duration::from_secs(1))), + // keep_alive: keep_alive::Behaviour, + // }, + // PeerId::from(key_pair.public()), + // ) + // .build(); + + let mut swarm = { + debug!("Running `rendezvous server` example using NymTransport"); + use libp2p::core::{muxing::StreamMuxerBox, transport::Transport}; + use libp2p::swarm::SwarmBuilder; + use rust_libp2p_nym::transport::NymTransport; + + let client = MixnetClient::connect_new().await.unwrap(); + + let transport = NymTransport::new(client, local_key.clone()).await?; + SwarmBuilder::with_tokio_executor( + transport + .map(|a, _| (a.0, StreamMuxerBox::new(a.1))) + .boxed(), + MyBehaviour { + identify: identify::Behaviour::new(identify::Config::new( + "rendezvous-example/1.0.0".to_string(), + local_key.public(), + )), + rendezvous: rendezvous::server::Behaviour::new(rendezvous::server::Config::default()), + ping: ping::Behaviour::new(ping::Config::new().with_interval(Duration::from_secs(1))), + keep_alive: keep_alive::Behaviour, + }, + local_peer_id, + ) + .build() + }; + + log::info!("Local peer id: {}", swarm.local_peer_id()); + + // let _ = swarm.listen_on("/ip4/0.0.0.0/tcp/62649".parse().unwrap()); + + loop { + match swarm.select_next_some().await { + SwarmEvent::NewListenAddr { address, .. } => info!("Listening on {address:?}"), + SwarmEvent::Behaviour(event) => { + debug!("{event:?}"); + } + other => { + log::debug!("Unhandled {:?}", other); + } + } + } + + // while let Some(event) = swarm.next().await { + // match event { + // SwarmEvent::ConnectionEstablished { peer_id, .. } => { + // log::info!("Connected to {}", peer_id); + // } + // SwarmEvent::ConnectionClosed { peer_id, .. } => { + // log::info!("Disconnected from {}", peer_id); + // } + // SwarmEvent::Behaviour(MyBehaviourEvent::Rendezvous( + // rendezvous::server::Event::PeerRegistered { peer, registration }, + // )) => { + // log::info!( + // "Peer {} registered for namespace '{}'", + // peer, + // registration.namespace + // ); + // } + // SwarmEvent::Behaviour(MyBehaviourEvent::Rendezvous( + // rendezvous::server::Event::DiscoverServed { + // enquirer, + // registrations, + // }, + // )) => { + // log::info!( + // "Served peer {} with {} registrations", + // enquirer, + // registrations.len() + // ); + // } + // // add a nym-specific newlisten behaviour? + // other => { + // log::debug!("Unhandled {:?}", other); + // } + // } + // } + + Ok(()) +} + +#[derive(NetworkBehaviour)] +struct MyBehaviour { + identify: identify::Behaviour, + rendezvous: rendezvous::server::Behaviour, + ping: ping::Behaviour, + keep_alive: keep_alive::Behaviour, +}