GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove old libp2p code - new example usage coming in smolmix-companions-docs branch

Browse Source
This commit is contained in:
mfahampshire
2026-04-29 12:36:50 +01:00
parent 0a8f18bd06
commit 8ab290ea03
13 changed files with 0 additions and 2900 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sdk/rust/nym-sdk/Cargo.toml
-2
View File
@@ -97,8 +97,6 @@ tokio = { workspace = true, features = ["full", "test-util"] }
time = { workspace = true }
nym-bin-common = {workspace = true, features = ["basic_tracing"] }
# 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" ]}
tokio-stream = { workspace = true }
tokio-util = { workspace = true, features = ["codec"] }
parking_lot = { workspace = true }
sdk/rust/nym-sdk/examples/libp2p_chat/README.md
-64
View File
@@ -1,64 +0,0 @@
# rust-libp2p-nym
This repo contains an example implementation of a libp2p transport using the Nym mixnet. It relies on the ChainSafe's fork of libp2p: https://github.com/ChainSafe/rust-libp2p
## Requirements
- Rust 1.68.2
- `Protoc` protobuf compiler. On Debian/Ubuntu distributed via `apt` as `protobuf-compiler` & on Arch/Manjaro via AUR as `[python-protobuf-compiler](https://aur.archlinux.org/packages/python-protobuf-compiler)`.
## Usage
To instantiate a libp2p swarm using the transport:
```rust
use libp2p::core::{muxing::StreamMuxerBox, transport::Transport};
use libp2p::swarm::{keep_alive::Behaviour, SwarmBuilder};
use libp2p::{identity, PeerId};
use nym_sdk::mixnet::MixnetClient;
use rust_libp2p_nym::transport::NymTransport;
use rust_libp2p_nym::test_utils::create_nym_client;
use std::error::Error;
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let local_key = identity::Keypair::generate_ed25519();
let local_peer_id = PeerId::from(local_key.public());
info!("Local peer id: {local_peer_id:?}");
let nym_id = rand::random::<u64>().to_string();
let nym_client = MixnetClient::connect_new().await.unwrap();
let transport = NymTransport::new(nym_client, local_key.clone()).await?;
let _swarm = SwarmBuilder::with_tokio_executor(
transport
.map(|a, _| (a.0, StreamMuxerBox::new(a.1)))
.boxed(),
Behaviour::default(),
local_peer_id,
)
.build();
Ok(())
}
```
## Chat example
To run the libp2p chat example, run the following in one terminal:
```bash
cargo run --example libp2p_chat
# Local peer id: PeerId("12D3KooWLukBu6q2FerWPFhFFhiYaJkhn2sBmceh9UCaXe6hJf5D")
# Listening on "/nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve"
```
In another terminal, run ping again, passing the Nym multiaddress printed previously:
```bash
cargo run --example libp2p_chat -- /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve
# Local peer id: PeerId("12D3KooWNsuRwG6DHnFJCDR8B3zdvja6xLcfnbtKCsQWJ8eppyWC")
# Dialed /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve
# Listening on "/nym/2oiRW5C9ivyF3Bo3Gpm4H9EqSKH7A6GpcrRRwVSDVUQ9.EajgCnhzimsP6KskUwKcEj8VFCmHR78s2J6FHWcZ4etR@Fo4f4SQLdoyoGkFae5TpVhRVoXCF8UiypLVGtGjujVPf"
```
You should see that the nodes connected and sent messages to each other:
```bash
# 2023-08-10T14:06:28.116Z INFO libp2p_chat > Got message: 'hello world' with id: 37393732353836333838333537303637303237 from peer: 12D3KooWB6k8ZGDF44N4FMRhgVBNihwk1wMYSumosxiZq9pUTbAz
```
sdk/rust/nym-sdk/examples/libp2p_chat/main.rs
-173
View File
@@ -1,173 +0,0 @@
// Copyright 2018 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
//! A basic chat application with logs demonstrating libp2p and the gossipsub protocol
//! combined with mDNS for the discovery of peers to gossip with.
//!
//! Using two terminal windows, start two instances, typing the following in each:
//!
//! ```sh
//! cargo run
//! ```
//!
//! Mutual mDNS discovery may take a few seconds. When each peer does discover the other
//! it will print a message like:
//!
//! ```sh
//! mDNS discovered a new peer: {peerId}
//! ```
//!
//! Type a message and hit return: the message is sent and printed in the other terminal.
//! Close with Ctrl-c.
//!
//! You can open more terminal windows and add more peers using the same line above.
//!
//! Once an additional peer is mDNS discovered it can participate in the conversation
//! and all peers will receive messages sent from it.
//!
//! If a participant exits (Control-C or otherwise) the other peers will receive an mDNS expired
//! event and remove the expired peer from the list of known peers.
// use crate::rust_libp2p_nym::transport::NymTransport;
// use futures::{prelude::*, select};
// use libp2p::Multiaddr;
// use libp2p::{
// core::muxing::StreamMuxerBox,
// gossipsub, identity,
// swarm::NetworkBehaviour,
// swarm::{SwarmBuilder, SwarmEvent},
// PeerId, Transport,
// };
// use log::{error, info, LevelFilter};
// use nym_sdk::mixnet::MixnetClient;
// use std::collections::hash_map::DefaultHasher;
use std::error::Error;
// use std::hash::{Hash, Hasher};
// use std::time::Duration;
// use tokio::io;
// use tokio_util::codec;
// #[path = "../libp2p_shared/lib.rs"]
// mod rust_libp2p_nym;
//
// // We create a custom network behaviour that uses Gossipsub
// #[derive(NetworkBehaviour)]
// struct Behaviour {
// gossipsub: gossipsub::Behaviour,
// }
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
unimplemented!("temporarily disabled")
// pretty_env_logger::formatted_timed_builder()
// .filter_level(LevelFilter::Warn)
// .filter(Some("libp2p_chat"), LevelFilter::Info)
// .init();
//
// // Create a random PeerId
// let id_keys = identity::Keypair::generate_ed25519();
// let local_peer_id = PeerId::from(id_keys.public());
// info!("Local peer id: {local_peer_id}");
//
// // To content-address message, we can take the hash of message and use it as an ID.
// let message_id_fn = |message: &gossipsub::Message| {
// let mut s = DefaultHasher::new();
// message.data.hash(&mut s);
// gossipsub::MessageId::from(s.finish().to_string())
// };
//
// // Set a custom gossipsub configuration
// let gossipsub_config = gossipsub::ConfigBuilder::default()
// .heartbeat_interval(Duration::from_secs(10)) // This is set to aid debugging by not cluttering the log space
// .validation_mode(gossipsub::ValidationMode::Strict) // This sets the kind of message validation. The default is Strict (enforce message signing)
// .message_id_fn(message_id_fn) // content-address messages. No two messages of the same content will be propagated.
// .build()
// .expect("Valid config");
//
// // build a gossipsub network behaviour
// let mut gossipsub = gossipsub::Behaviour::new(
// gossipsub::MessageAuthenticity::Signed(id_keys),
// gossipsub_config,
// )
// .expect("Correct configuration");
// // Create a Gossipsub topic
// let topic = gossipsub::IdentTopic::new("test-net");
// // subscribes to our topic
// gossipsub.subscribe(&topic)?;
//
// let client = MixnetClient::connect_new().await.unwrap();
// info!("client address: {}", client.nym_address());
//
// let local_key = identity::Keypair::generate_ed25519();
// let local_peer_id = PeerId::from(local_key.public());
// info!("Local peer id: {local_peer_id:?}");
//
// let transport = NymTransport::new(client, local_key).await?;
//
// let mut swarm = SwarmBuilder::with_tokio_executor(
// transport
// .map(|a, _| (a.0, StreamMuxerBox::new(a.1)))
// .boxed(),
// Behaviour { gossipsub },
// local_peer_id,
// )
// .build();
//
// if let Some(addr) = std::env::args().nth(1) {
// let remote: Multiaddr = addr.parse()?;
// swarm.dial(remote)?;
// info!("Dialed {addr}")
// }
//
// // Read full lines from stdin
// let mut stdin = codec::FramedRead::new(io::stdin(), codec::LinesCodec::new()).fuse();
//
// info!("Enter messages via STDIN and they will be sent to connected peers using Gossipsub");
//
// // Kick it off
// loop {
// select! {
// line = stdin.select_next_some() => {
// if let Err(e) = swarm
// .behaviour_mut().gossipsub
// .publish(topic.clone(), line.expect("Stdin not to close").as_bytes()) {
// error!("Publish error: {e:?}");
// }
// },
// event = swarm.select_next_some() => {
// match event {
// SwarmEvent::Behaviour(BehaviourEvent::Gossipsub(gossipsub::Event::Message {
// propagation_source: peer_id,
// message_id: id,
// message,
// })) => info!(
// "Got message: '{}' with id: {id} from peer: {peer_id}",
// String::from_utf8_lossy(&message.data),
// ),
// SwarmEvent::NewListenAddr { address, .. } => {
// info!("Local node is listening on {address}");
// }
// other => {info!("other event: {:?}", other)}
// }
// }
// }
// }
}
sdk/rust/nym-sdk/examples/libp2p_ping/README.md
-79
View File
@@ -1,79 +0,0 @@
# rust-libp2p-nym
This repo contains an example implementation of a libp2p transport using the Nym mixnet. It relies on the ChainSafe's fork of libp2p: https://github.com/ChainSafe/rust-libp2p
## Requirements
- Rust 1.68.2
- `Protoc` protobuf compiler. On Debian/Ubuntu distributed via `apt` as `protobuf-compiler` & on Arch/Manjaro via AUR as `[python-protobuf-compiler](https://aur.archlinux.org/packages/python-protobuf-compiler)`.
## Usage
To instantiate a libp2p swarm using the transport:
```rust
use libp2p::core::{muxing::StreamMuxerBox, transport::Transport};
use libp2p::swarm::{keep_alive::Behaviour, SwarmBuilder};
use libp2p::{identity, PeerId};
use nym_sdk::mixnet::MixnetClient;
use rust_libp2p_nym::transport::NymTransport;
use rust_libp2p_nym::test_utils::create_nym_client;
use std::error::Error;
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
let local_key = identity::Keypair::generate_ed25519();
let local_peer_id = PeerId::from(local_key.public());
info!("Local peer id: {local_peer_id:?}");
let nym_id = rand::random::<u64>().to_string();
let nym_client = MixnetClient::connect_new().await.unwrap();
let transport = NymTransport::new(nym_client, local_key.clone()).await?;
let _swarm = SwarmBuilder::with_tokio_executor(
transport
.map(|a, _| (a.0, StreamMuxerBox::new(a.1)))
.boxed(),
Behaviour::default(),
local_peer_id,
)
.build();
Ok(())
}
```
## Ping example
To run the libp2p ping example, run the following in one terminal:
```bash
cargo run --example libp2p_ping
# Local peer id: PeerId("12D3KooWLukBu6q2FerWPFhFFhiYaJkhn2sBmceh9UCaXe6hJf5D")
# Listening on "/nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve"
```
In another terminal, run ping again, passing the Nym multiaddress printed previously:
```bash
cargo run --example libp2p_ping -- /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve
# Local peer id: PeerId("12D3KooWNsuRwG6DHnFJCDR8B3zdvja6xLcfnbtKCsQWJ8eppyWC")
# Dialed /nym/FhtkzizQg2JbZ19kGkRKXdjV2QnFbT5ww88ZAKaD4nkF.7Remi4UVYzn1yL3qYtEcQBGh6tzTYxMdYB4uqyHVc5Z4@62F81C9GrHDRja9WCqozemRFSzFPMecY85MbGwn6efve
# Listening on "/nym/2oiRW5C9ivyF3Bo3Gpm4H9EqSKH7A6GpcrRRwVSDVUQ9.EajgCnhzimsP6KskUwKcEj8VFCmHR78s2J6FHWcZ4etR@Fo4f4SQLdoyoGkFae5TpVhRVoXCF8UiypLVGtGjujVPf"
```
You should see that the nodes connected and pinged each other:
```bash
# Mar 30 22:56:36.400 INFO ping: BehaviourEvent: Event { peer: PeerId("12D3KooWGf2oYd6U2nrLzfDrN9zxsjSQjPsMh2oDJPUQ9hiHMNtf"), result: Ok(Ping { rtt: 1.06836675s }) }
```
```bash
# Mar 30 22:56:35.595 INFO ping: BehaviourEvent: Event { peer: PeerId("12D3KooWMd5ak31DXuZq7x1JuFSR6toA5RDQrPaHrfXEhy7vqqpC"), result: Ok(Pong) }
```
In order to run the ping example with vanilla libp2p, which uses tcp, pass the
`--features libp2p-vanilla` flag to the example and follow the instructions on the
rust-libp2p project as usual.
```bash
RUST_LOG=ping=debug cargo run --example ping --features libp2p-vanilla
```
```bash
RUST_LOG=ping=debug cargo run --example ping --features libp2p-vanilla -- "/ip4/127.0.0.1/tcp/$PORT"
```
sdk/rust/nym-sdk/examples/libp2p_ping/main.rs
-143
View File
@@ -1,143 +0,0 @@
// Copyright 2018 Parity Technologies (UK) Ltd.
//
// Permission is hereby granted, free of charge, to any person obtaining a
// copy of this software and associated documentation files (the "Software"),
// to deal in the Software without restriction, including without limitation
// the rights to use, copy, modify, merge, publish, distribute, sublicense,
// and/or sell copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
// OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
//! Ping example
//!
//! See ../src/tutorial.rs for a step-by-step guide building the example below.
//!
//! In the first terminal window, run:
//!
//! ```sh
//! cargo run --example ping --features=full
//! ```
//!
//! It will print the PeerId and the listening addresses, e.g. `Listening on
//! "/ip4/0.0.0.0/tcp/24915"`
//!
//! In the second terminal window, start a new instance of the example with:
//!
//! ```sh
//! cargo run --example ping --features=full -- /ip4/127.0.0.1/tcp/24915
//! ```
//!
//! The two nodes establish a connection, negotiate the ping protocol
//! and begin pinging each other.
// 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 std::error::Error;
// use std::time::Duration;
//
// #[path = "../libp2p_shared/lib.rs"]
// mod rust_libp2p_nym;
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
unimplemented!("temporarily disabled")
//
// pretty_env_logger::formatted_timed_builder()
// .filter_level(LevelFilter::Warn)
// .filter(Some("libp2p_ping"), 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:?}");
//
// #[cfg(not(feature = "libp2p-vanilla"))]
// let mut swarm = {
// debug!("Running `ping` example using NymTransport");
// use libp2p::core::{muxing::StreamMuxerBox, transport::Transport};
// use libp2p::swarm::SwarmBuilder;
// use rust_libp2p_nym::transport::NymTransport;
//
// let client = nym_sdk::mixnet::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(),
// Behaviour::default(),
// local_peer_id,
// )
// .build()
// };
//
// #[cfg(feature = "libp2p-vanilla")]
// let mut swarm = {
// debug!("Running `ping` example using the vanilla libp2p tokio_development_transport");
// let transport = libp2p::tokio_development_transport(local_key)?;
// let mut swarm =
// libp2p::Swarm::with_tokio_executor(transport, Behaviour::default(), local_peer_id);
// swarm.listen_on("/ip4/0.0.0.0/tcp/0".parse()?)?;
// swarm
// };
//
// // Dial the peer identified by the multi-address given as the second
// // command-line argument, if any.
// if let Some(addr) = std::env::args().nth(1) {
// let remote: Multiaddr = addr.parse()?;
// swarm.dial(remote)?;
// info!("Dialed {addr}")
// }
//
// let mut total_ping_rtt: Duration = Duration::from_micros(0);
// let mut counter: u128 = 0;
// loop {
// match swarm.select_next_some().await {
// SwarmEvent::NewListenAddr { address, .. } => info!("Listening on {address:?}"),
// SwarmEvent::Behaviour(event) => {
// // Get the round-trip duration for the pings.
// // This value is already captured in the BehaviourEvent::Ping's `Success::Ping`
// // field.
// debug!("{event:?}");
// if let BehaviourEvent::Ping(ping_event) = event {
// let result: Success = ping_event.result?;
// match result {
// Success::Ping { rtt } => {
// counter += 1;
// total_ping_rtt += rtt;
// let average_ping_rtt = Duration::from_micros(
// (total_ping_rtt.as_micros() / counter).try_into().unwrap(),
// );
// info!("Ping RTT: {rtt:?} AVERAGE RTT: ({counter} pings): {average_ping_rtt:?}");
// }
// Success::Pong => info!("Pong Event"),
// }
// }
// }
// _ => {}
// }
// }
}
//
// /// Our network behaviour.
// ///
// /// For illustrative purposes, this includes the [`KeepAlive`](behaviour::KeepAlive) behaviour so a continuous sequence of
// /// pings can be observed.
// #[derive(NetworkBehaviour, Default)]
// struct Behaviour {
// keep_alive: keep_alive::Behaviour,
// ping: ping::Behaviour,
// }
sdk/rust/nym-sdk/examples/libp2p_shared/connection.rs
-418
View File
@@ -1,418 +0,0 @@
use libp2p::core::{muxing::StreamMuxerEvent, PeerId, StreamMuxer};
use log::debug;
use nym_sphinx::addressing::clients::Recipient;
use std::{
collections::{HashMap, HashSet},
pin::Pin,
sync::{
atomic::{AtomicU64, Ordering},
Arc,
},
task::{Context, Poll, Waker},
};
use tokio::sync::{
mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
oneshot,
};
use super::error::Error;
use super::message::{
ConnectionId, Message, OutboundMessage, SubstreamId, SubstreamMessage, SubstreamMessageType,
TransportMessage,
};
use super::substream::Substream;
/// Connection represents the result of a connection setup process.
/// It implements `StreamMuxer` and thus has stream multiplexing built in.
#[derive(Debug)]
pub struct Connection {
pub(crate) peer_id: PeerId,
pub(crate) remote_recipient: Recipient,
pub(crate) id: ConnectionId,
/// receive inbound messages from the `InnerConnection`
pub(crate) inbound_rx: UnboundedReceiver<SubstreamMessage>,
/// substream ID -> outbound pending substream exists
/// the key is deleted when the response is received, or the request times out
pending_substreams: HashSet<SubstreamId>,
/// substream ID -> substream's inbound_tx channel
substream_inbound_txs: HashMap<SubstreamId, UnboundedSender<Vec<u8>>>,
/// substream ID -> substream's close_tx channel
substream_close_txs: HashMap<SubstreamId, oneshot::Sender<()>>,
/// send messages to the mixnet
/// used for sending `SubstreamMessageType::OpenRequest` messages
/// also passed to each substream so they can write to the mixnet
pub(crate) mixnet_outbound_tx: UnboundedSender<OutboundMessage>,
/// inbound substream open requests; used in poll_inbound
inbound_open_tx: UnboundedSender<Substream>,
inbound_open_rx: UnboundedReceiver<Substream>,
/// closed substream IDs; used in poll_close
close_tx: UnboundedSender<SubstreamId>,
close_rx: UnboundedReceiver<SubstreamId>,
/// message nonce contains the next nonce that should be used when
/// sending a message over the connection
pub(crate) message_nonce: Arc<AtomicU64>,
waker: Option<Waker>,
}
impl Connection {
pub(crate) fn new(
peer_id: PeerId,
remote_recipient: Recipient,
id: ConnectionId,
inbound_rx: UnboundedReceiver<SubstreamMessage>,
mixnet_outbound_tx: UnboundedSender<OutboundMessage>,
) -> Self {
let (inbound_open_tx, inbound_open_rx) = unbounded_channel();
let (close_tx, close_rx) = unbounded_channel();
Connection {
peer_id,
remote_recipient,
id,
inbound_rx,
pending_substreams: HashSet::new(),
substream_inbound_txs: HashMap::new(),
substream_close_txs: HashMap::new(),
mixnet_outbound_tx,
inbound_open_tx,
inbound_open_rx,
close_tx,
close_rx,
message_nonce: Arc::new(AtomicU64::new(1)),
waker: None,
}
}
fn new_outbound_substream(&mut self) -> Result<Substream, Error> {
let substream_id = SubstreamId::generate();
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
// send the substream open request that requests to open a substream with the given ID
self.mixnet_outbound_tx
.send(OutboundMessage {
recipient: self.remote_recipient,
message: Message::TransportMessage(TransportMessage {
nonce,
id: self.id.clone(),
message: SubstreamMessage {
substream_id: substream_id.clone(),
message_type: SubstreamMessageType::OpenRequest,
},
}),
})
.map_err(|e| Error::OutboundSendFailure(e.to_string()))?;
// track pending outbound substreams
// TODO we should probably lock this? storing map values should be atomic
let res = self.new_substream(substream_id.clone());
if res.is_ok() {
self.pending_substreams.insert(substream_id);
}
res
}
// creates a new substream instance with the given ID.
fn new_substream(&mut self, id: SubstreamId) -> Result<Substream, Error> {
// check we don't already have a substream with this ID
if self.substream_inbound_txs.contains_key(&id) {
return Err(Error::SubstreamIdExists(id));
}
let (inbound_tx, inbound_rx) = unbounded_channel::<Vec<u8>>();
let (close_tx, close_rx) = oneshot::channel::<()>();
self.substream_inbound_txs.insert(id.clone(), inbound_tx);
self.substream_close_txs.insert(id.clone(), close_tx);
if let Some(waker) = self.waker.take() {
waker.wake();
}
Ok(Substream::new(
self.remote_recipient,
self.id.clone(),
id,
inbound_rx,
self.mixnet_outbound_tx.clone(),
close_rx,
self.message_nonce.clone(),
))
}
fn handle_close(&mut self, substream_id: SubstreamId) -> Result<(), Error> {
if self.substream_inbound_txs.remove(&substream_id).is_none() {
return Err(Error::SubstreamIdDoesNotExist(substream_id));
}
// notify substream that it's closed
let close_tx = self.substream_close_txs.remove(&substream_id);
close_tx.unwrap().send(()).unwrap();
// notify poll_close that the substream is closed
self.close_tx
.send(substream_id)
.map_err(|e| Error::InboundSendFailure(e.to_string()))
}
}
impl StreamMuxer for Connection {
type Substream = Substream;
type Error = Error;
fn poll_inbound(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<Self::Substream, Self::Error>> {
if let Poll::Ready(Some(substream)) = self.inbound_open_rx.poll_recv(cx) {
return Poll::Ready(Ok(substream));
}
Poll::Pending
}
fn poll_outbound(
mut self: Pin<&mut Self>,
_cx: &mut Context<'_>,
) -> Poll<Result<Self::Substream, Self::Error>> {
Poll::Ready(self.new_outbound_substream())
}
fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
if let Poll::Ready(Some(_)) = self.close_rx.poll_recv(cx) {
return Poll::Ready(Ok(()));
}
Poll::Pending
}
fn poll(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
while let Poll::Ready(Some(msg)) = self.inbound_rx.poll_recv(cx) {
match msg.message_type {
SubstreamMessageType::OpenRequest => {
// create a new substream with the given ID
let substream = self.new_substream(msg.substream_id.clone())?;
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
// send the response to the remote peer
self.mixnet_outbound_tx
.send(OutboundMessage {
recipient: self.remote_recipient,
message: Message::TransportMessage(TransportMessage {
nonce,
id: self.id.clone(),
message: SubstreamMessage {
substream_id: msg.substream_id.clone(),
message_type: SubstreamMessageType::OpenResponse,
},
}),
})
.map_err(|e| Error::OutboundSendFailure(e.to_string()))?;
debug!("wrote OpenResponse for substream: {:?}", &msg.substream_id);
// send the substream to our own channel to be returned in poll_inbound
self.inbound_open_tx
.send(substream)
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
debug!("new inbound substream: {:?}", &msg.substream_id);
}
SubstreamMessageType::OpenResponse => {
if !self.pending_substreams.remove(&msg.substream_id) {
debug!(
"SubstreamMessageType::OpenResponse no substream pending for ID: {:?}",
&msg.substream_id
);
}
}
SubstreamMessageType::Close => {
self.handle_close(msg.substream_id)?;
}
SubstreamMessageType::Data(data) => {
debug!("SubstreamMessageType::Data: {:?}", &data);
let inbound_tx = self
.substream_inbound_txs
.get_mut(&msg.substream_id)
.expect("must have a substream channel for substream");
// NOTE: this ignores channel closed errors, which is fine because the substream
// might have been closed/dropped
inbound_tx.send(data).ok();
}
}
}
self.waker = Some(cx.waker().clone());
Poll::Pending
}
}
/// PendingConnection represents a connection that's been initiated, but not completed.
pub(crate) struct PendingConnection {
pub(crate) remote_recipient: Recipient,
pub(crate) connection_tx: oneshot::Sender<Connection>,
}
impl PendingConnection {
pub(crate) fn new(
remote_recipient: Recipient,
connection_tx: oneshot::Sender<Connection>,
) -> Self {
PendingConnection {
remote_recipient,
connection_tx,
}
}
}
#[cfg(test)]
mod test {
use super::super::message::InboundMessage;
use super::super::mixnet::initialize_mixnet;
use super::*;
use futures::future::poll_fn;
use futures::{AsyncReadExt, AsyncWriteExt, FutureExt};
use nym_sdk::mixnet::MixnetClient;
async fn inbound_receive_and_send(
connection_id: ConnectionId,
mixnet_inbound_rx: &mut UnboundedReceiver<InboundMessage>,
inbound_tx: &UnboundedSender<SubstreamMessage>,
expected_nonce: u64,
) {
let recv_msg = mixnet_inbound_rx.recv().await.unwrap();
match recv_msg.0 {
Message::TransportMessage(TransportMessage {
nonce,
id,
message: msg,
}) => {
assert_eq!(nonce, expected_nonce);
assert_eq!(id, connection_id);
inbound_tx.send(msg).unwrap();
}
_ => panic!("unexpected message"),
}
}
#[tokio::test]
async fn test_connection_stream_muxer() {
let client = MixnetClient::connect_new().await.unwrap();
let (sender_address, mut sender_mixnet_inbound_rx, sender_outbound_tx) =
initialize_mixnet(client, None).await.unwrap();
let client2 = MixnetClient::connect_new().await.unwrap();
let (recipient_address, mut recipient_mixnet_inbound_rx, recipient_outbound_tx) =
initialize_mixnet(client2, None).await.unwrap();
let connection_id = ConnectionId::generate();
let recipient_peer_id = PeerId::random();
let sender_peer_id = PeerId::random();
// create the connections
let (sender_inbound_tx, sender_inbound_rx) = unbounded_channel::<SubstreamMessage>();
let mut sender_connection = Connection::new(
recipient_peer_id,
recipient_address,
connection_id.clone(),
sender_inbound_rx,
sender_outbound_tx,
);
let (recipient_inbound_tx, recipient_inbound_rx) = unbounded_channel::<SubstreamMessage>();
let mut recipient_connection = Connection::new(
sender_peer_id,
sender_address,
connection_id.clone(),
recipient_inbound_rx,
recipient_outbound_tx,
);
// send the substream OpenRequest to the mixnet
let mut sender_substream = sender_connection.new_outbound_substream().unwrap();
assert!(sender_connection
.pending_substreams
.contains(&sender_substream.substream_id));
assert_eq!(sender_connection.message_nonce.load(Ordering::SeqCst), 2);
// poll the recipient inbound stream; should receive the OpenRequest and create the substream
inbound_receive_and_send(
connection_id.clone(),
&mut recipient_mixnet_inbound_rx,
&recipient_inbound_tx,
1,
)
.await;
poll_fn(|cx| Pin::new(&mut recipient_connection).as_mut().poll(cx)).now_or_never();
assert_eq!(recipient_connection.message_nonce.load(Ordering::SeqCst), 2);
// poll recipient's poll_inbound to receive the substream
let maybe_recipient_substream = poll_fn(|cx| {
Pin::new(&mut recipient_connection)
.as_mut()
.poll_inbound(cx)
})
.now_or_never();
let mut recipient_substream = maybe_recipient_substream.unwrap().unwrap();
// poll sender's connection to receive the OpenResponse and send it to the Connection inbound channel
inbound_receive_and_send(
connection_id.clone(),
&mut sender_mixnet_inbound_rx,
&sender_inbound_tx,
1,
)
.await;
// poll sender's poll_outbound to get the substream
poll_fn(|cx| Pin::new(&mut sender_connection).as_mut().poll(cx)).now_or_never();
assert!(sender_connection.pending_substreams.is_empty());
// finally, write message to the substream
let data = b"hello world";
sender_substream.write_all(data).await.unwrap();
assert_eq!(sender_connection.message_nonce.load(Ordering::SeqCst), 3);
// receive message from the mixnet, push to the recipient Connection inbound channel
inbound_receive_and_send(
connection_id.clone(),
&mut recipient_mixnet_inbound_rx,
&recipient_inbound_tx,
2,
)
.await;
// poll the sender's connection to send the msg from the connection inbound channel to the substream's
poll_fn(|cx| Pin::new(&mut sender_connection).as_mut().poll(cx)).now_or_never();
// poll the recipient's connection to read the msg from the mixnet and mux it into the substream
poll_fn(|cx| Pin::new(&mut recipient_connection).as_mut().poll(cx)).now_or_never();
let mut buf = [0u8; 11];
let n = recipient_substream.read(&mut buf).await.unwrap();
assert_eq!(n, 11);
assert_eq!(buf, data[..]);
// test closing the stream; assert the stream is closed on both sides
sender_substream.close().await.unwrap();
assert_eq!(sender_connection.message_nonce.load(Ordering::SeqCst), 4);
inbound_receive_and_send(
connection_id.clone(),
&mut recipient_mixnet_inbound_rx,
&recipient_inbound_tx,
3,
)
.await;
}
}
sdk/rust/nym-sdk/examples/libp2p_shared/error.rs
-64
View File
@@ -1,64 +0,0 @@
use libp2p::core::multiaddr;
use nym_sphinx::addressing::clients::RecipientFormattingError;
use super::message::SubstreamId;
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("unimplemented")]
Unimplemented,
#[error("failed to format multiaddress from nym address")]
FailedToFormatMultiaddr(#[from] multiaddr::Error),
#[error("unexpected protocol in multiaddress")]
InvalidProtocolForMultiaddr,
#[error("failed to decode message")]
InvalidMessageBytes,
#[error("no connection found for ConnectionResponse")]
NoConnectionForResponse,
#[error("received ConnectionResponse but connection was already established")]
ConnectionAlreadyEstablished,
#[error("received None recipient in ConnectionRequest")]
NoneRecipientInConnectionRequest,
#[error("cannot handle connection request; already have connection with given ID")]
ConnectionIDExists,
#[error("no connection found for TransportMessage")]
NoConnectionForTransportMessage,
#[error("failed to decode ConnectionMessage; too short")]
ConnectionMessageBytesTooShort,
#[error("failed to decode ConnectionMessage; no recipient")]
ConnectionMessageBytesNoRecipient,
#[error("failed to decode ConnectionMessage; no peer ID")]
ConnectionMessageBytesNoPeerId,
#[error("invalid peer ID bytes")]
InvalidPeerIdBytes,
#[error("invalid recipient bytes")]
InvalidRecipientBytes(#[from] RecipientFormattingError),
#[error("invalid recipient prefix byte")]
InvalidRecipientPrefixByte,
#[error("failed to decode TransportMessage; too short")]
TransportMessageBytesTooShort,
#[error("failed to decode TransportMessage; invalid nonce")]
InvalidNonce,
#[error("invalid substream ID")]
InvalidSubstreamMessageBytes,
#[error("invalid substream message type byte")]
InvalidSubstreamMessageType,
#[error("substrean with given ID already exists")]
SubstreamIdExists(SubstreamId),
#[error("no substream found for given ID")]
SubstreamIdDoesNotExist(SubstreamId),
#[error("recv error: channel closed")]
OneshotRecvFailure(#[from] tokio::sync::oneshot::error::RecvError),
#[error("recv error: channel closed")]
RecvFailure,
#[error("outbound send error")]
OutboundSendFailure(String),
#[error("inbound send error")]
InboundSendFailure(String),
#[error("failed to send new connection; receiver dropped")]
ConnectionSendFailure,
#[error("failed to send initial TransportEvent::NewAddress")]
SendErrorTransportEvent,
#[error("dial timed out")]
DialTimeout(#[from] tokio::time::error::Elapsed),
}
sdk/rust/nym-sdk/examples/libp2p_shared/lib.rs
-10
View File
@@ -1,10 +0,0 @@
pub(crate) mod connection;
pub mod error;
pub(crate) mod message;
pub(crate) mod mixnet;
pub(crate) mod queue;
pub mod substream;
pub mod transport;
/// The deafult timeout secs for [`transport::Upgrade`] future.
const DEFAULT_HANDSHAKE_TIMEOUT_SECS: u64 = 5;
sdk/rust/nym-sdk/examples/libp2p_shared/message.rs
-335
View File
@@ -1,335 +0,0 @@
use libp2p::core::PeerId;
use nym_sphinx::addressing::clients::Recipient;
use rand::rngs::OsRng;
use rand::RngCore;
use std::fmt::{Debug, Formatter};
use super::error::Error;
const RECIPIENT_LENGTH: usize = Recipient::LEN;
const CONNECTION_ID_LENGTH: usize = 32;
const SUBSTREAM_ID_LENGTH: usize = 32;
const NONCE_BYTES_LEN: usize = 8; // length of u64
const MIN_CONNECTION_MESSAGE_LEN: usize = CONNECTION_ID_LENGTH + NONCE_BYTES_LEN;
/// ConnectionId is a unique, randomly-generated per-connection ID that's used to
/// identify which connection a message belongs to.
#[derive(Clone, Default, Eq, Hash, PartialEq)]
pub(crate) struct ConnectionId([u8; 32]);
impl ConnectionId {
pub(crate) fn generate() -> Self {
let mut bytes = [0u8; 32];
OsRng.fill_bytes(&mut bytes);
ConnectionId(bytes)
}
fn from_bytes(bytes: &[u8]) -> Self {
let mut id = [0u8; 32];
id[..].copy_from_slice(&bytes[0..CONNECTION_ID_LENGTH]);
ConnectionId(id)
}
}
impl Debug for ConnectionId {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", hex::encode(self.0))
}
}
/// SubstreamId is a unique, randomly-generated per-substream ID that's used to
/// identify which substream a message belongs to.
#[derive(Clone, Default, Eq, Hash, PartialEq)]
pub struct SubstreamId(pub(crate) [u8; 32]);
impl SubstreamId {
pub(crate) fn generate() -> Self {
let mut bytes = [0u8; 32];
OsRng.fill_bytes(&mut bytes);
SubstreamId(bytes)
}
fn from_bytes(bytes: &[u8]) -> Self {
let mut id = [0u8; 32];
id[..].copy_from_slice(&bytes[0..SUBSTREAM_ID_LENGTH]);
SubstreamId(id)
}
}
impl Debug for SubstreamId {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", hex::encode(self.0))
}
}
#[derive(Debug)]
#[allow(clippy::enum_variant_names)]
pub(crate) enum Message {
ConnectionRequest(ConnectionMessage),
ConnectionResponse(ConnectionMessage),
TransportMessage(TransportMessage),
}
/// ConnectionMessage is exchanged to open a new connection.
#[derive(Debug)]
pub(crate) struct ConnectionMessage {
pub(crate) peer_id: PeerId,
pub(crate) id: ConnectionId,
/// recipient is the sender's Nym address.
/// only required if this is a ConnectionRequest.
pub(crate) recipient: Option<Recipient>,
}
/// TransportMessage is sent over a connection after establishment.
#[derive(Debug, Clone)]
pub(crate) struct TransportMessage {
/// increments by 1 for every TransportMessage sent over a connection.
/// required for ordering, since Nym does not guarantee ordering.
/// ConnectionMessages do not need nonces, as we know that they will
/// be the first messages sent over a connection.
/// the first TransportMessage sent over a connection will have nonce 1.
pub(crate) nonce: u64,
pub(crate) message: SubstreamMessage,
pub(crate) id: ConnectionId,
}
impl Message {
fn try_from_bytes(bytes: Vec<u8>) -> Result<Self, Error> {
if bytes.len() < 2 {
return Err(Error::InvalidMessageBytes);
}
Ok(match bytes[0] {
0 => Message::ConnectionRequest(ConnectionMessage::try_from_bytes(&bytes[1..])?),
1 => Message::ConnectionResponse(ConnectionMessage::try_from_bytes(&bytes[1..])?),
2 => Message::TransportMessage(TransportMessage::try_from_bytes(&bytes[1..])?),
_ => return Err(Error::InvalidMessageBytes),
})
}
}
impl ConnectionMessage {
fn to_bytes(&self) -> Vec<u8> {
let mut bytes = self.id.0.to_vec();
match self.recipient {
Some(recipient) => {
bytes.push(1u8);
bytes.append(&mut recipient.to_bytes().to_vec());
}
None => bytes.push(0u8),
}
bytes.append(&mut self.peer_id.to_bytes());
bytes
}
fn try_from_bytes(bytes: &[u8]) -> Result<Self, Error> {
if bytes.len() < CONNECTION_ID_LENGTH + 1 {
return Err(Error::ConnectionMessageBytesTooShort);
}
let id = ConnectionId::from_bytes(&bytes[0..CONNECTION_ID_LENGTH]);
let recipient = match bytes[CONNECTION_ID_LENGTH] {
0u8 => None,
1u8 => {
if bytes.len() < CONNECTION_ID_LENGTH + 1 + RECIPIENT_LENGTH {
return Err(Error::ConnectionMessageBytesNoRecipient);
}
let mut recipient_bytes = [0u8; RECIPIENT_LENGTH];
recipient_bytes[..].copy_from_slice(
&bytes[CONNECTION_ID_LENGTH + 1..CONNECTION_ID_LENGTH + 1 + RECIPIENT_LENGTH],
);
Some(
Recipient::try_from_bytes(recipient_bytes)
.map_err(Error::InvalidRecipientBytes)?,
)
}
_ => {
return Err(Error::InvalidRecipientPrefixByte);
}
};
let peer_id = match recipient {
Some(_) => {
if bytes.len() < CONNECTION_ID_LENGTH + RECIPIENT_LENGTH + 2 {
return Err(Error::ConnectionMessageBytesNoPeerId);
}
PeerId::from_bytes(&bytes[CONNECTION_ID_LENGTH + 1 + RECIPIENT_LENGTH..])
.map_err(|_| Error::InvalidPeerIdBytes)?
}
None => {
if bytes.len() < CONNECTION_ID_LENGTH + 2 {
return Err(Error::ConnectionMessageBytesNoPeerId);
}
PeerId::from_bytes(&bytes[CONNECTION_ID_LENGTH + 1..])
.map_err(|_| Error::InvalidPeerIdBytes)?
}
};
Ok(ConnectionMessage {
peer_id,
recipient,
id,
})
}
}
impl TransportMessage {
fn to_bytes(&self) -> Vec<u8> {
let mut bytes = self.nonce.to_be_bytes().to_vec();
bytes.extend_from_slice(self.id.0.as_ref());
bytes.extend_from_slice(&self.message.to_bytes());
bytes
}
fn try_from_bytes(bytes: &[u8]) -> Result<Self, Error> {
if bytes.len() < MIN_CONNECTION_MESSAGE_LEN + 1 {
return Err(Error::TransportMessageBytesTooShort);
}
let nonce = u64::from_be_bytes(
bytes[0..NONCE_BYTES_LEN]
.to_vec()
.try_into()
.map_err(|_| Error::InvalidNonce)?,
);
let id = ConnectionId::from_bytes(&bytes[NONCE_BYTES_LEN..MIN_CONNECTION_MESSAGE_LEN]);
let message = SubstreamMessage::try_from_bytes(&bytes[MIN_CONNECTION_MESSAGE_LEN..])?;
Ok(TransportMessage { nonce, message, id })
}
}
impl Ord for TransportMessage {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
self.nonce.cmp(&other.nonce)
}
}
impl std::cmp::PartialOrd for TransportMessage {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl std::cmp::Eq for TransportMessage {}
impl std::cmp::PartialEq for TransportMessage {
fn eq(&self, other: &Self) -> bool {
self.nonce == other.nonce
}
}
#[derive(Debug, Clone)]
pub(crate) enum SubstreamMessageType {
OpenRequest,
OpenResponse,
Close,
Data(Vec<u8>),
}
impl SubstreamMessageType {
fn to_u8(&self) -> u8 {
match self {
SubstreamMessageType::OpenRequest => 0,
SubstreamMessageType::OpenResponse => 1,
SubstreamMessageType::Close => 2,
SubstreamMessageType::Data(_) => 3,
}
}
}
/// SubstreamMessage is a message sent over a substream.
#[derive(Debug, Clone)]
pub(crate) struct SubstreamMessage {
pub(crate) substream_id: SubstreamId,
pub(crate) message_type: SubstreamMessageType,
}
impl SubstreamMessage {
pub(crate) fn new_with_data(substream_id: SubstreamId, message: Vec<u8>) -> Self {
SubstreamMessage {
substream_id,
message_type: SubstreamMessageType::Data(message),
}
}
pub(crate) fn new_close(substream_id: SubstreamId) -> Self {
SubstreamMessage {
substream_id,
message_type: SubstreamMessageType::Close,
}
}
pub(crate) fn to_bytes(&self) -> Vec<u8> {
let mut bytes = self.substream_id.0.clone().to_vec();
bytes.push(self.message_type.to_u8());
if let SubstreamMessageType::Data(message) = &self.message_type {
bytes.extend_from_slice(message);
}
bytes
}
pub(crate) fn try_from_bytes(bytes: &[u8]) -> Result<Self, Error> {
if bytes.len() < SUBSTREAM_ID_LENGTH + 1 {
return Err(Error::InvalidSubstreamMessageBytes);
}
let substream_id = SubstreamId::from_bytes(&bytes[0..SUBSTREAM_ID_LENGTH]);
let message_type = match bytes[SUBSTREAM_ID_LENGTH] {
0 => SubstreamMessageType::OpenRequest,
1 => SubstreamMessageType::OpenResponse,
2 => SubstreamMessageType::Close,
3 => {
if bytes.len() < SUBSTREAM_ID_LENGTH + 2 {
return Err(Error::InvalidSubstreamMessageBytes);
}
SubstreamMessageType::Data(bytes[SUBSTREAM_ID_LENGTH + 1..].to_vec())
}
_ => return Err(Error::InvalidSubstreamMessageType),
};
Ok(SubstreamMessage {
substream_id,
message_type,
})
}
}
impl Message {
pub(crate) fn to_bytes(&self) -> Vec<u8> {
match self {
Message::ConnectionRequest(msg) => {
let mut bytes = 0_u8.to_be_bytes().to_vec();
bytes.append(&mut msg.to_bytes());
bytes
}
Message::ConnectionResponse(msg) => {
let mut bytes = 1_u8.to_be_bytes().to_vec();
bytes.append(&mut msg.to_bytes());
bytes
}
Message::TransportMessage(msg) => {
let mut bytes = 2_u8.to_be_bytes().to_vec();
bytes.append(&mut msg.to_bytes());
bytes
}
}
}
}
/// InboundMessage represents an inbound mixnet message.
pub(crate) struct InboundMessage(pub(crate) Message);
/// OutboundMessage represents an outbound mixnet message.
#[derive(Debug)]
pub(crate) struct OutboundMessage {
pub(crate) message: Message,
pub(crate) recipient: Recipient,
}
pub(crate) fn parse_message_data(data: &[u8]) -> Result<InboundMessage, Error> {
if data.len() < 2 {
return Err(Error::InvalidMessageBytes);
}
let msg = Message::try_from_bytes(data.to_vec())?;
Ok(InboundMessage(msg))
}
sdk/rust/nym-sdk/examples/libp2p_shared/mixnet.rs
-164
View File
@@ -1,164 +0,0 @@
use futures::{pin_mut, select};
use futures::{FutureExt, StreamExt};
use log::debug;
use nym_sdk::mixnet::{IncludedSurbs, MixnetClient, MixnetClientSender, MixnetMessageSender};
use nym_sphinx::addressing::clients::Recipient;
use nym_sphinx::receiver::ReconstructedMessage;
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
use super::error::Error;
use super::message::*;
/// initialize_mixnet initializes a read/write connection to a Nym websockets endpoint.
/// It starts a task that listens for inbound messages from the endpoint and writes outbound messages to the endpoint.
pub(crate) async fn initialize_mixnet(
client: MixnetClient,
notify_inbound_tx: Option<UnboundedSender<()>>,
) -> Result<
(
Recipient,
UnboundedReceiver<InboundMessage>,
UnboundedSender<OutboundMessage>,
),
Error,
> {
let recipient = *client.nym_address();
// a channel of inbound messages from the mixnet..
// the transport reads from (listens) to the inbound_rx.
// TODO: this is probably a DOS vector; we should limit the size of the channel.
let (inbound_tx, inbound_rx) = unbounded_channel::<InboundMessage>();
// a channel of outbound messages to be written to the mixnet.
// the transport writes to outbound_tx.
let (outbound_tx, mut outbound_rx) = unbounded_channel::<OutboundMessage>();
let sink = client.split_sender();
let mut stream = client;
tokio::task::spawn(async move {
loop {
let t1 = check_inbound(&mut stream, &inbound_tx, &notify_inbound_tx).fuse();
let t2 = check_outbound(&sink, &mut outbound_rx).fuse();
pin_mut!(t1, t2);
select! {
_ = t1 => {},
_ = t2 => {},
};
}
});
Ok((recipient, inbound_rx, outbound_tx))
}
async fn check_inbound(
client: &mut MixnetClient,
inbound_tx: &UnboundedSender<InboundMessage>,
notify_inbound_tx: &Option<UnboundedSender<()>>,
) -> Result<(), Error> {
if let Some(msg) = client.next().await {
if let Some(notify_tx) = notify_inbound_tx {
notify_tx
.send(())
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
}
handle_inbound(msg, inbound_tx).await?;
}
Err(Error::Unimplemented)
}
async fn handle_inbound(
msg: ReconstructedMessage,
inbound_tx: &UnboundedSender<InboundMessage>,
) -> Result<(), Error> {
let data = parse_message_data(&msg.message)?;
inbound_tx
.send(data)
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
Ok(())
}
async fn check_outbound(
mixnet_sender: &MixnetClientSender,
outbound_rx: &mut UnboundedReceiver<OutboundMessage>,
) -> Result<(), Error> {
match outbound_rx.recv().await {
Some(message) => {
write_bytes(
mixnet_sender,
message.recipient,
&message.message.to_bytes(),
)
.await
}
None => Err(Error::RecvFailure),
}
}
async fn write_bytes(
mixnet_sender: &MixnetClientSender,
recipient: Recipient,
message: &[u8],
) -> Result<(), Error> {
if let Err(_err) = mixnet_sender
.send_message(recipient, message, IncludedSurbs::ExposeSelfAddress)
.await
{
return Err(Error::Unimplemented);
}
debug!(
"wrote message to mixnet: recipient: {:?}",
recipient.to_string()
);
Ok(())
}
#[cfg(test)]
mod test {
use super::super::message::{
self, ConnectionId, Message, SubstreamId, SubstreamMessage, SubstreamMessageType,
TransportMessage,
};
use super::super::mixnet::initialize_mixnet;
use nym_sdk::mixnet::MixnetClient;
#[tokio::test]
async fn test_mixnet_poll_inbound_and_outbound() {
let client = MixnetClient::connect_new().await.unwrap();
let (self_address, mut inbound_rx, outbound_tx) =
initialize_mixnet(client, None).await.unwrap();
let msg_inner = "hello".as_bytes();
let substream_id = SubstreamId::generate();
let msg = Message::TransportMessage(TransportMessage {
nonce: 1, // arbitrary
id: ConnectionId::generate(),
message: SubstreamMessage::new_with_data(substream_id.clone(), msg_inner.to_vec()),
});
// send a message to ourselves through the mixnet
let out_msg = message::OutboundMessage {
message: msg,
recipient: self_address,
};
outbound_tx.send(out_msg).unwrap();
// receive the message from ourselves over the mixnet
let received_msg = inbound_rx.recv().await.unwrap();
if let Message::TransportMessage(recv_msg) = received_msg.0 {
assert_eq!(substream_id, recv_msg.message.substream_id);
if let SubstreamMessageType::Data(data) = recv_msg.message.message_type {
assert_eq!(msg_inner, data.as_slice());
} else {
panic!("expected SubstreamMessage::Data")
}
} else {
panic!("expected Message::TransportMessage")
}
}
}
sdk/rust/nym-sdk/examples/libp2p_shared/queue.rs
-138
View File
@@ -1,138 +0,0 @@
use log::{debug, warn};
use std::collections::BTreeSet;
use super::message::TransportMessage;
/// MessageQueue is a queue of messages, ordered by nonce, that we've
/// received but are not yet able to process because we're waiting for
/// a message with the next expected nonce first.
/// This is required because Nym does not guarantee any sort of message
/// ordering, only delivery.
/// TODO: is there a DOS vector here where a malicious peer sends us
/// messages only with nonce higher than the next expected nonce?
pub(crate) struct MessageQueue {
/// nonce of the next message we expect to receive on the
/// connection.
/// any messages with a nonce greater than this are pushed into
/// the queue.
/// if we get a message with a nonce equal to this, then we
/// immediately handle it in the transport and increment the nonce.
next_expected_nonce: u64,
/// the actual queue of messages, ordered by nonce.
/// the head of the queue's nonce is always greater
/// than the next expected nonce.
queue: BTreeSet<TransportMessage>,
}
impl MessageQueue {
pub(crate) fn new() -> Self {
MessageQueue {
next_expected_nonce: 0,
queue: BTreeSet::new(),
}
}
pub(crate) fn print_nonces(&self) {
let nonces = self.queue.iter().map(|msg| msg.nonce).collect::<Vec<_>>();
debug!("MessageQueue: {:?}", nonces);
}
/// sets the next expected nonce to 1, indicating that we've received
/// a ConnectionRequest or ConnectionResponse.
pub(crate) fn set_connection_message_received(&mut self) {
if self.next_expected_nonce != 0 {
panic!("connection message received twice");
}
self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1);
}
/// tries to push a message into the queue.
/// if the message has the next expected nonce, then the message is returned,
/// and should be processed by the caller.
/// in that case, the internal queue's next expected nonce is incremented.
pub(crate) fn try_push(&mut self, msg: TransportMessage) -> Option<TransportMessage> {
if msg.nonce == self.next_expected_nonce {
self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1);
Some(msg)
} else {
if msg.nonce < self.next_expected_nonce {
// this shouldn't happen normally, only if the other node
// is not following the protocol
warn!("received a message with a nonce that is too low");
return None;
}
if !self.queue.insert(msg) {
// this shouldn't happen normally, only if the other node
// is not following the protocol
warn!("received a message with a duplicate nonce");
return None;
}
None
}
}
pub(crate) fn pop(&mut self) -> Option<TransportMessage> {
let head = self.queue.first()?;
if head.nonce == self.next_expected_nonce {
self.next_expected_nonce = self.next_expected_nonce.wrapping_add(1);
Some(self.queue.pop_first().unwrap())
} else {
None
}
}
}
#[cfg(test)]
mod test {
use super::super::message::{ConnectionId, SubstreamId, SubstreamMessage};
use super::*;
impl TransportMessage {
fn new(nonce: u64, message: SubstreamMessage, id: ConnectionId) -> Self {
TransportMessage { nonce, message, id }
}
}
#[test]
fn test_message_queue() {
let mut queue = MessageQueue::new();
let test_substream_message =
SubstreamMessage::new_with_data(SubstreamId::generate(), vec![1, 2, 3]);
let connection_id = ConnectionId::generate();
let msg1 = TransportMessage::new(1, test_substream_message.clone(), connection_id.clone());
let msg2 = TransportMessage::new(2, test_substream_message.clone(), connection_id.clone());
let msg3 = TransportMessage::new(3, test_substream_message.clone(), connection_id.clone());
assert_eq!(queue.try_push(msg1.clone()), None);
assert_eq!(queue.try_push(msg3.clone()), None);
assert_eq!(queue.try_push(msg2.clone()), None);
assert_eq!(queue.pop(), None);
// set expected nonce to 1
queue.set_connection_message_received();
assert_eq!(queue.pop(), Some(msg1));
let msg4 = TransportMessage::new(4, test_substream_message.clone(), connection_id.clone());
assert_eq!(queue.try_push(msg4.clone()), None);
assert_eq!(queue.pop(), Some(msg2));
assert_eq!(queue.pop(), Some(msg3));
assert_eq!(queue.pop(), Some(msg4));
assert_eq!(queue.pop(), None);
assert_eq!(queue.next_expected_nonce, 5);
// should just return the message and increment nonce when message nonce = next expected nonce
let msg5 = TransportMessage::new(5, test_substream_message, connection_id);
assert_eq!(queue.try_push(msg5.clone()), Some(msg5));
assert_eq!(queue.next_expected_nonce, 6);
}
}
sdk/rust/nym-sdk/examples/libp2p_shared/substream.rs
-412
View File
@@ -1,412 +0,0 @@
use super::message::{
ConnectionId, Message, OutboundMessage, SubstreamId, SubstreamMessage, TransportMessage,
};
use futures::{
io::{Error as IoError, ErrorKind},
AsyncRead, AsyncWrite,
};
use log::debug;
use nym_sphinx::addressing::clients::Recipient;
use parking_lot::Mutex;
use std::{
pin::Pin,
sync::{
atomic::{AtomicU64, Ordering},
Arc,
},
task::{Context, Poll},
};
use tokio::sync::{
mpsc::{UnboundedReceiver, UnboundedSender},
oneshot::Receiver,
};
#[derive(Debug)]
pub struct Substream {
remote_recipient: Recipient,
connection_id: ConnectionId,
pub(crate) substream_id: SubstreamId,
/// inbound messages; inbound_tx is in the corresponding Connection
pub(crate) inbound_rx: UnboundedReceiver<Vec<u8>>,
/// outbound messages; go directly to the mixnet
outbound_tx: UnboundedSender<OutboundMessage>,
/// used to signal when the substream is closed
close_rx: Receiver<()>,
closed: Mutex<bool>,
// buffer of data that's been written to the stream,
// but not yet read by the application.
unread_data: Mutex<Vec<u8>>,
message_nonce: Arc<AtomicU64>,
}
impl Substream {
pub(crate) fn new(
remote_recipient: Recipient,
connection_id: ConnectionId,
substream_id: SubstreamId,
inbound_rx: UnboundedReceiver<Vec<u8>>,
outbound_tx: UnboundedSender<OutboundMessage>,
close_rx: Receiver<()>,
message_nonce: Arc<AtomicU64>,
) -> Self {
Substream {
remote_recipient,
connection_id,
substream_id,
inbound_rx,
outbound_tx,
close_rx,
closed: Mutex::new(false),
unread_data: Mutex::new(vec![]),
message_nonce,
}
}
fn check_closed(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Result<(), IoError> {
let closed_err = IoError::other("stream closed");
// close_rx will return an error if the channel is closed (ie. sender was dropped),
// or if it's empty
let received_closed = self.close_rx.try_recv();
let mut closed = self.closed.lock();
if *closed {
return Err(closed_err);
}
if received_closed.is_ok() {
*closed = true;
return Err(closed_err);
}
Ok(())
}
}
impl AsyncRead for Substream {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut [u8],
) -> Poll<Result<usize, IoError>> {
let closed_result = self.as_mut().check_closed(cx);
if let Err(e) = closed_result {
return Poll::Ready(Err(e));
}
let inbound_rx_data = self.inbound_rx.poll_recv(cx);
// first, write any previously unread data to the buf
let mut unread_data = self.unread_data.lock();
let filled_len = if unread_data.len() > 0 {
let unread_len = unread_data.len();
let buf_len = buf.len();
let copy_len = std::cmp::min(unread_len, buf_len);
buf[..copy_len].copy_from_slice(&unread_data[..copy_len]);
*unread_data = unread_data[copy_len..].to_vec();
copy_len
} else {
0
};
if let Poll::Ready(Some(data)) = inbound_rx_data {
if filled_len == buf.len() {
// we've filled the buffer, so we'll have to save the rest for later
let mut new = vec![];
new.extend(unread_data.drain(..));
new.extend(data.iter());
*unread_data = new;
return Poll::Ready(Ok(filled_len));
}
// otherwise, there's still room in the buffer, so we'll copy the rest of the data
let remaining_len = buf.len() - filled_len;
let data_len = data.len();
// we have more data than buffer room remaining, save the extra for later
if remaining_len < data_len {
unread_data.extend_from_slice(&data[remaining_len..]);
}
let copied = std::cmp::min(remaining_len, data_len);
buf[filled_len..filled_len + copied].copy_from_slice(&data[..copied]);
debug!("poll_read copied {} bytes: data {:?}", copied, buf);
return Poll::Ready(Ok(copied));
}
if filled_len > 0 {
debug!("poll_read copied {} bytes: data {:?}", filled_len, buf);
return Poll::Ready(Ok(filled_len));
}
Poll::Pending
}
}
impl AsyncWrite for Substream {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<Result<usize, IoError>> {
if let Err(e) = self.as_mut().check_closed(cx) {
return Poll::Ready(Err(e));
}
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
self.outbound_tx
.send(OutboundMessage {
recipient: self.remote_recipient,
message: Message::TransportMessage(TransportMessage {
nonce,
id: self.connection_id.clone(),
message: SubstreamMessage::new_with_data(
self.substream_id.clone(),
buf.to_vec(),
),
}),
})
.map_err(|e| IoError::other(format!("poll_write outbound_tx error: {}", e)))?;
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), IoError>> {
if let Err(e) = self.check_closed(cx) {
return Poll::Ready(Err(e));
}
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), IoError>> {
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
let mut closed = self.closed.lock();
if *closed {
return Poll::Ready(Err(IoError::other("stream closed")));
}
*closed = true;
// send a close message to the mixnet
self.outbound_tx
.send(OutboundMessage {
recipient: self.remote_recipient,
message: Message::TransportMessage(TransportMessage {
nonce,
id: self.connection_id.clone(),
message: SubstreamMessage::new_close(self.substream_id.clone()),
}),
})
.map_err(|e| IoError::other(format!("poll_close outbound_rx error: {}", e)))?;
Poll::Ready(Ok(()))
}
}
#[cfg(test)]
mod test {
use super::super::message::{
ConnectionId, Message, SubstreamId, SubstreamMessage, TransportMessage,
};
use super::super::mixnet::initialize_mixnet;
use super::Substream;
use futures::{AsyncReadExt, AsyncWriteExt};
use nym_sdk::mixnet::MixnetClient;
use nym_sphinx::addressing::clients::Recipient;
use std::sync::atomic::AtomicU64;
use std::sync::Arc;
#[tokio::test]
async fn test_substream_poll_read_unread_data() {
let (outbound_tx, _) = tokio::sync::mpsc::unbounded_channel();
let connection_id = ConnectionId::generate();
let substream_id = SubstreamId::generate();
let (inbound_tx, inbound_rx) = tokio::sync::mpsc::unbounded_channel();
let (_, close_rx) = tokio::sync::oneshot::channel();
let mut substream = Substream::new(
Recipient::try_from_base58_string("D1rrpsysCGCYXy9saP8y3kmNpGtJZUXN9SvFoUcqAsM9.9Ssso1ea5NfkbMASdiseDSjTN1fSWda5SgEVjdSN4CvV@GJqd3ZxpXWSNxTfx7B1pPtswpetH4LnJdFeLeuY5KUuN").unwrap(),
connection_id,
substream_id,
inbound_rx,
outbound_tx,
close_rx,
Arc::new(AtomicU64::new(1)),
);
// test writing and reading w/ same length data
let data = b"hello".to_vec();
inbound_tx.send(data.clone()).unwrap();
let mut buf = [0u8; 5];
let read_len = substream.read(&mut buf).await.unwrap();
assert_eq!(read_len, data.len());
assert_eq!(buf.to_vec(), data);
// test writing data longer than read buffer
let data = b"nootwashere".to_vec();
inbound_tx.send(data.clone()).unwrap();
let mut buf = [0u8; 4];
let read_len = substream.read(&mut buf).await.unwrap();
assert_eq!(read_len, buf.len());
assert_eq!(buf.to_vec(), b"noot".to_vec());
let mut buf = [0u8; 7];
let read_len = substream.read(&mut buf).await.unwrap();
assert_eq!(read_len, buf.len());
assert_eq!(buf.to_vec(), b"washere".to_vec());
// test read buffer larger than written data
let data = b"nootwashere".to_vec();
inbound_tx.send(data.clone()).unwrap();
let mut buf = [0u8; 16];
let read_len = substream.read(&mut buf).await.unwrap();
assert_eq!(read_len, data.len());
assert_eq!(buf[..data.len()], data);
assert_eq!(buf[data.len()..].to_vec(), vec![0u8; 16 - data.len()]);
// test writing data longer than read buffer multiple times
let data = b"nootwashere".to_vec();
inbound_tx.send(data.clone()).unwrap();
let mut buf = [0u8; 4];
let read_len = substream.read(&mut buf).await.unwrap();
assert_eq!(read_len, buf.len());
assert_eq!(buf.to_vec(), b"noot".to_vec());
let data = b"asdf".to_vec();
inbound_tx.send(data.clone()).unwrap();
let mut buf = [0u8; 4];
let read_len = substream.read(&mut buf).await.unwrap();
assert_eq!(read_len, buf.len());
assert_eq!(buf.to_vec(), b"wash".to_vec());
let mut buf = [0u8; 8];
let read_len = substream.read(&mut buf).await.unwrap();
assert_eq!(read_len, 7);
assert_eq!(buf[..7], b"ereasdf".to_vec());
}
#[tokio::test]
async fn test_substream_read_write() {
let client = MixnetClient::connect_new().await.unwrap();
let (self_address, mut mixnet_inbound_rx, outbound_tx) =
initialize_mixnet(client, None).await.unwrap();
const MSG_INNER: &[u8] = "hello".as_bytes();
let connection_id = ConnectionId::generate();
let substream_id = SubstreamId::generate();
let (inbound_tx, inbound_rx) = tokio::sync::mpsc::unbounded_channel();
let (_, close_rx) = tokio::sync::oneshot::channel();
let mut substream = Substream::new(
self_address,
connection_id,
substream_id,
inbound_rx,
outbound_tx,
close_rx,
Arc::new(AtomicU64::new(1)),
);
// send message to ourselves over the mixnet
substream.write_all(MSG_INNER).await.unwrap();
// receive full message over the mixnet
let recv_msg = mixnet_inbound_rx.recv().await.unwrap();
match recv_msg.0 {
Message::TransportMessage(TransportMessage {
nonce,
id: _,
message:
SubstreamMessage {
substream_id: _,
message_type: msg,
},
}) => {
assert_eq!(nonce, 1);
match msg {
super::super::message::SubstreamMessageType::Data(data) => {
assert_eq!(data, MSG_INNER);
// send message to substream inbound channel
inbound_tx.send(data).unwrap();
}
_ => panic!("unexpected message type"),
}
}
_ => panic!("unexpected message"),
}
// read message from substream
let mut buf = [0u8; MSG_INNER.len()];
substream.read_exact(&mut buf).await.unwrap();
assert_eq!(buf, MSG_INNER);
// close substream
substream.close().await.unwrap();
// try to read/write to closed substream; should error
substream.write_all(MSG_INNER).await.unwrap_err();
substream.read_exact(&mut buf).await.unwrap_err();
// assert a close message was sent over the mixnet
let recv_msg = mixnet_inbound_rx.recv().await.unwrap();
match recv_msg.0 {
Message::TransportMessage(TransportMessage {
nonce: _,
id: _,
message:
SubstreamMessage {
substream_id: _,
message_type: msg,
},
}) => match msg {
super::super::message::SubstreamMessageType::Close => {}
_ => panic!("unexpected message type"),
},
_ => panic!("unexpected message: {:?}", recv_msg.0),
}
}
#[tokio::test]
async fn test_substream_recv_close() {
let client = MixnetClient::connect_new().await.unwrap();
let (self_address, _, outbound_tx) = initialize_mixnet(client, None).await.unwrap();
const MSG_INNER: &[u8] = "hello".as_bytes();
let connection_id = ConnectionId::generate();
let substream_id = SubstreamId::generate();
let (_, inbound_rx) = tokio::sync::mpsc::unbounded_channel();
let (close_tx, close_rx) = tokio::sync::oneshot::channel();
let mut substream = Substream::new(
self_address,
connection_id,
substream_id,
inbound_rx,
outbound_tx,
close_rx,
Arc::new(AtomicU64::new(1)),
);
// close substream
close_tx.send(()).unwrap();
// try to read/write to closed substream; should error
substream.write_all(MSG_INNER).await.unwrap_err();
let mut buf = [0u8; MSG_INNER.len()];
substream.read_exact(&mut buf).await.unwrap_err();
}
}
sdk/rust/nym-sdk/examples/libp2p_shared/transport.rs
-898
View File
@@ -1,898 +0,0 @@
use futures::prelude::*;
use libp2p::core::{
identity::Keypair,
multiaddr::{Multiaddr, Protocol},
transport::{ListenerId, TransportError, TransportEvent},
PeerId, Transport,
};
use log::debug;
use nym_sdk::mixnet::MixnetClient;
use nym_sphinx::addressing::clients::Recipient;
use std::{
collections::HashMap,
pin::Pin,
str::FromStr,
task::{Context, Poll, Waker},
};
use tokio::{
sync::{
mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender},
oneshot,
},
time::{timeout, Duration},
};
use tokio_stream::wrappers::UnboundedReceiverStream;
use super::connection::{Connection, PendingConnection};
use super::error::Error;
use super::message::{
ConnectionId, ConnectionMessage, InboundMessage, Message, OutboundMessage, SubstreamMessage,
TransportMessage,
};
use super::mixnet::initialize_mixnet;
use super::queue::MessageQueue;
use super::DEFAULT_HANDSHAKE_TIMEOUT_SECS;
/// InboundTransportEvent represents an inbound event from the mixnet.
pub enum InboundTransportEvent {
ConnectionRequest(Upgrade),
ConnectionResponse,
TransportMessage,
}
/// NymTransport implements the Transport trait using the Nym mixnet.
pub struct NymTransport {
/// our Nym address
self_address: Recipient,
pub(crate) listen_addr: Multiaddr,
pub(crate) listener_id: ListenerId,
/// our libp2p keypair; currently not really used
keypair: Keypair,
/// established connections -> channel which sends messages received from
/// the mixnet to the corresponding Connection
connections: HashMap<ConnectionId, UnboundedSender<SubstreamMessage>>,
/// outbound pending dials
pending_dials: HashMap<ConnectionId, PendingConnection>,
/// connection message queues
message_queues: HashMap<ConnectionId, MessageQueue>,
/// inbound mixnet messages
inbound_stream: UnboundedReceiverStream<InboundMessage>,
/// outbound mixnet messages
outbound_tx: UnboundedSender<OutboundMessage>,
/// inbound messages for Transport.poll()
poll_rx: UnboundedReceiver<TransportEvent<Upgrade, Error>>,
/// outbound messages to Transport.poll()
poll_tx: UnboundedSender<TransportEvent<Upgrade, Error>>,
waker: Option<Waker>,
/// Timeout for the [`Upgrade`] future.
handshake_timeout: Duration,
}
impl NymTransport {
/// New transport.
#[allow(unused)]
pub async fn new(client: MixnetClient, keypair: Keypair) -> Result<Self, Error> {
Self::new_maybe_with_notify_inbound(client, keypair, None, None).await
}
/// New transport with a timeout.
#[allow(dead_code)]
pub async fn new_with_timeout(
client: MixnetClient,
keypair: Keypair,
timeout: Duration,
) -> Result<Self, Error> {
Self::new_maybe_with_notify_inbound(client, keypair, None, Some(timeout)).await
}
/// Add timeout to transport and return self.
#[allow(dead_code)]
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.handshake_timeout = timeout;
self
}
async fn new_maybe_with_notify_inbound(
client: MixnetClient,
keypair: Keypair,
notify_inbound_tx: Option<UnboundedSender<()>>,
timeout: Option<Duration>,
) -> Result<Self, Error> {
let (self_address, inbound_rx, outbound_tx) =
initialize_mixnet(client, notify_inbound_tx).await?;
let listen_addr = nym_address_to_multiaddress(self_address)?;
let listener_id = ListenerId::new();
let (poll_tx, poll_rx) = unbounded_channel::<TransportEvent<Upgrade, Error>>();
poll_tx
.send(TransportEvent::NewAddress {
listener_id,
listen_addr: listen_addr.clone(),
})
.map_err(|_| Error::SendErrorTransportEvent)?;
let inbound_stream = UnboundedReceiverStream::new(inbound_rx);
let handshake_timeout =
timeout.unwrap_or_else(|| Duration::from_secs(DEFAULT_HANDSHAKE_TIMEOUT_SECS));
Ok(Self {
self_address,
listen_addr,
listener_id,
keypair,
connections: HashMap::new(),
pending_dials: HashMap::new(),
message_queues: HashMap::new(),
inbound_stream,
outbound_tx,
poll_rx,
poll_tx,
waker: None,
handshake_timeout,
})
}
pub(crate) fn peer_id(&self) -> PeerId {
PeerId::from_public_key(&self.keypair.public())
}
fn handle_message_queue_on_connection_initiation(
&mut self,
id: &ConnectionId,
) -> Result<(), Error> {
debug!("handle_message_queue_on_connection_initiation");
let Some(inbound_tx) = self.connections.get(id) else {
// this should not happen
return Err(Error::NoConnectionForTransportMessage);
};
match self.message_queues.get_mut(id) {
Some(queue) => {
// update expected nonce
queue.set_connection_message_received();
// push pending inbound some messages in this case
while let Some(msg) = queue.pop() {
debug!(
"popped queued message with nonce {} for connection",
msg.nonce
);
inbound_tx
.send(msg.message.clone())
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
}
}
None => {
// no queue exists for this connection, create one
let queue = MessageQueue::new();
self.message_queues.insert(id.clone(), queue);
let queue = self.message_queues.get_mut(id).unwrap();
queue.set_connection_message_received();
}
};
debug!("returning from handle_message_queue_on_connection_initiation");
Ok(())
}
// handle_connection_response resolves the pending connection corresponding to the response
// (if there is one) into a Connection.
fn handle_connection_response(&mut self, msg: &ConnectionMessage) -> Result<(), Error> {
if self.connections.contains_key(&msg.id) {
return Err(Error::ConnectionAlreadyEstablished);
}
if let Some(pending_conn) = self.pending_dials.remove(&msg.id) {
// resolve connection and put into pending_conn channel
let (conn, conn_tx) = self.create_connection_types(
msg.peer_id,
pending_conn.remote_recipient,
msg.id.clone(),
);
self.connections.insert(msg.id.clone(), conn_tx);
self.handle_message_queue_on_connection_initiation(&msg.id)?;
pending_conn
.connection_tx
.send(conn)
.map_err(|_| Error::ConnectionSendFailure)?;
if let Some(waker) = self.waker.take() {
waker.wake();
}
Ok(())
} else {
Err(Error::NoConnectionForResponse)
}
}
/// handle_connection_request handles an incoming connection request, sends back a
/// connection response, and finally completes the upgrade into a Connection.
fn handle_connection_request(&mut self, msg: &ConnectionMessage) -> Result<Connection, Error> {
if msg.recipient.is_none() {
return Err(Error::NoneRecipientInConnectionRequest);
}
// ensure we don't already have a conn with the same id
if self.connections.contains_key(&msg.id) {
return Err(Error::ConnectionIDExists);
}
let (conn, conn_tx) =
self.create_connection_types(msg.peer_id, msg.recipient.unwrap(), msg.id.clone());
self.connections.insert(msg.id.clone(), conn_tx);
self.handle_message_queue_on_connection_initiation(&msg.id)?;
let resp = ConnectionMessage {
peer_id: self.peer_id(),
recipient: None,
id: msg.id.clone(),
};
self.outbound_tx
.send(OutboundMessage {
message: Message::ConnectionResponse(resp),
recipient: msg.recipient.unwrap(),
})
.map_err(|e| Error::OutboundSendFailure(e.to_string()))?;
if let Some(waker) = self.waker.take() {
waker.wake();
}
Ok(conn)
}
fn handle_transport_message(&mut self, msg: TransportMessage) -> Result<(), Error> {
let queue = match self.message_queues.get_mut(&msg.id) {
Some(queue) => queue,
None => {
// no queue exists for this connection, create one
let queue = MessageQueue::new();
self.message_queues.insert(msg.id.clone(), queue);
self.message_queues.get_mut(&msg.id).unwrap()
}
};
queue.print_nonces();
let nonce = msg.nonce;
let Some(msg) = queue.try_push(msg) else {
// don't push the message yet, it's been queued
debug!("message with nonce {} queued for connection", nonce);
return Ok(());
};
let Some(inbound_tx) = self.connections.get(&msg.id) else {
return Err(Error::NoConnectionForTransportMessage);
};
// send original message
debug!(
"sending original message with nonce {} for connection",
nonce
);
inbound_tx
.send(msg.message.clone())
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
// try to pop queued messages and send them on inbound channel
while let Some(msg) = queue.pop() {
debug!(
"popped queued message with nonce {} for connection",
msg.nonce
);
inbound_tx
.send(msg.message.clone())
.map_err(|e| Error::InboundSendFailure(e.to_string()))?;
}
if let Some(waker) = self.waker.clone().take() {
waker.wake();
}
Ok(())
}
fn create_connection_types(
&self,
remote_peer_id: PeerId,
recipient: Recipient,
id: ConnectionId,
) -> (Connection, UnboundedSender<SubstreamMessage>) {
let (inbound_tx, inbound_rx) = unbounded_channel::<SubstreamMessage>();
// representation of a connection; this contains channels for applications to read/write to.
let conn = Connection::new(
remote_peer_id,
recipient,
id,
inbound_rx,
self.outbound_tx.clone(),
);
// inbound_tx is what we write to when receiving messages on the mixnet,
(conn, inbound_tx)
}
/// handle_inbound handles an inbound message from the mixnet, received via self.inbound_stream.
fn handle_inbound(&mut self, msg: Message) -> Result<InboundTransportEvent, Error> {
match msg {
Message::ConnectionRequest(inner) => {
debug!("got inbound connection request {:?}", inner);
match self.handle_connection_request(&inner) {
Ok(conn) => {
let (connection_tx, connection_rx) =
oneshot::channel::<(PeerId, Connection)>();
let upgrade = Upgrade::new(connection_rx);
connection_tx
.send((inner.peer_id, conn))
.map_err(|_| Error::ConnectionSendFailure)?;
Ok(InboundTransportEvent::ConnectionRequest(upgrade))
}
Err(e) => Err(e),
}
}
Message::ConnectionResponse(msg) => {
debug!("got inbound connection response {:?}", msg);
self.handle_connection_response(&msg)
.map(|_| InboundTransportEvent::ConnectionResponse)
}
Message::TransportMessage(msg) => {
debug!("got inbound TransportMessage: {:?}", msg);
self.handle_transport_message(msg)
.map(|_| InboundTransportEvent::TransportMessage)
}
}
}
}
/// Upgrade represents a transport listener upgrade.
/// Note: we immediately upgrade a connection request to a connection,
/// so this only contains a channel for receiving that connection.
pub struct Upgrade {
connection_tx: oneshot::Receiver<(PeerId, Connection)>,
}
impl Upgrade {
fn new(connection_tx: oneshot::Receiver<(PeerId, Connection)>) -> Upgrade {
Upgrade { connection_tx }
}
}
impl Future for Upgrade {
type Output = Result<(PeerId, Connection), Error>;
// poll checks if the upgrade has turned into a connection yet
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.connection_tx
.poll_unpin(cx)
.map_err(|_| Error::RecvFailure)
}
}
impl Transport for NymTransport {
type Output = (PeerId, Connection);
type Error = Error;
type ListenerUpgrade = Upgrade;
type Dial = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
fn listen_on(&mut self, _: Multiaddr) -> Result<ListenerId, TransportError<Self::Error>> {
// we should only allow listening on the multiaddress containing our Nym address
Ok(self.listener_id)
}
fn remove_listener(&mut self, id: ListenerId) -> bool {
if self.listener_id != id {
return false;
}
// TODO: close channels?
self.poll_tx
.send(TransportEvent::ListenerClosed {
listener_id: id,
reason: Ok(()),
})
.expect("failed to send listener closed event");
true
}
fn dial(&mut self, addr: Multiaddr) -> Result<Self::Dial, TransportError<Self::Error>> {
debug!("dialing {}", addr);
let id = ConnectionId::generate();
// create remote recipient address
let recipient = multiaddress_to_nym_address(addr).map_err(TransportError::Other)?;
// create pending conn structs and store
let (connection_tx, connection_rx) = oneshot::channel::<Connection>();
let inner_pending_conn = PendingConnection::new(recipient, connection_tx);
self.pending_dials.insert(id.clone(), inner_pending_conn);
// put ConnectionRequest message into outbound message channel
let msg = ConnectionMessage {
peer_id: self.peer_id(),
recipient: Some(self.self_address),
id,
};
let outbound_tx = self.outbound_tx.clone();
let mut waker = self.waker.clone();
let handshake_timeout = self.handshake_timeout;
Ok(async move {
outbound_tx
.send(OutboundMessage {
message: Message::ConnectionRequest(msg),
recipient,
})
.map_err(|e| Error::OutboundSendFailure(e.to_string()))?;
debug!("sent outbound ConnectionRequest");
if let Some(waker) = waker.take() {
waker.wake();
};
let conn = timeout(handshake_timeout, connection_rx).await??;
Ok((conn.peer_id, conn))
}
.boxed())
}
// dial_as_listener currently just calls self.dial().
fn dial_as_listener(
&mut self,
addr: Multiaddr,
) -> Result<Self::Dial, TransportError<Self::Error>> {
self.dial(addr)
}
fn poll(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<TransportEvent<Self::ListenerUpgrade, Self::Error>> {
// new addresses + listener close events
if let Poll::Ready(Some(res)) = self.poll_rx.recv().boxed().poll_unpin(cx) {
return Poll::Ready(res);
}
// check for and handle inbound messages
while let Poll::Ready(Some(msg)) = self.inbound_stream.poll_next_unpin(cx) {
match self.handle_inbound(msg.0) {
Ok(event) => match event {
InboundTransportEvent::ConnectionRequest(upgrade) => {
debug!("InboundTransportEvent::ConnectionRequest");
return Poll::Ready(TransportEvent::Incoming {
listener_id: self.listener_id,
upgrade,
local_addr: self.listen_addr.clone(),
send_back_addr: self.listen_addr.clone(),
});
}
InboundTransportEvent::ConnectionResponse => {
debug!("InboundTransportEvent::ConnectionResponse");
}
InboundTransportEvent::TransportMessage => {
debug!("InboundTransportEvent::TransportMessage");
}
},
Err(e) => {
return Poll::Ready(TransportEvent::ListenerError {
listener_id: self.listener_id,
error: e,
});
}
};
}
self.waker = Some(cx.waker().clone());
Poll::Pending
}
fn address_translation(&self, _listen: &Multiaddr, _observed: &Multiaddr) -> Option<Multiaddr> {
None
}
}
fn nym_address_to_multiaddress(addr: Recipient) -> Result<Multiaddr, Error> {
Multiaddr::from_str(&format!("/nym/{}", addr)).map_err(Error::FailedToFormatMultiaddr)
}
fn multiaddress_to_nym_address(multiaddr: Multiaddr) -> Result<Recipient, Error> {
let mut multiaddr = multiaddr;
match multiaddr.pop().unwrap() {
Protocol::Nym(addr) => Recipient::from_str(&addr).map_err(Error::InvalidRecipientBytes),
_ => Err(Error::InvalidProtocolForMultiaddr),
}
}
#[cfg(test)]
mod test {
use super::super::connection::Connection;
use super::super::error::Error;
use super::super::message::{
Message, OutboundMessage, SubstreamId, SubstreamMessage, SubstreamMessageType,
TransportMessage,
};
use super::super::substream::Substream;
use super::{nym_address_to_multiaddress, NymTransport};
use futures::{future::poll_fn, AsyncReadExt, AsyncWriteExt, FutureExt};
use libp2p::core::{
identity::Keypair,
transport::{Transport, TransportEvent},
Multiaddr, StreamMuxer,
};
use log::info;
use nym_bin_common::logging::setup_tracing_logger;
use nym_sdk::mixnet::MixnetClient;
use std::{pin::Pin, str::FromStr, sync::atomic::Ordering};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
impl Connection {
fn write(&self, msg: SubstreamMessage) -> Result<(), Error> {
let nonce = self.message_nonce.fetch_add(1, Ordering::SeqCst);
self.mixnet_outbound_tx
.send(OutboundMessage {
recipient: self.remote_recipient,
message: Message::TransportMessage(TransportMessage {
nonce,
id: self.id.clone(),
message: msg,
}),
})
.map_err(|e| Error::OutboundSendFailure(e.to_string()))?;
Ok(())
}
}
impl NymTransport {
async fn new_with_notify_inbound(
client: MixnetClient,
notify_inbound_tx: UnboundedSender<()>,
) -> Result<Self, Error> {
let local_key = Keypair::generate_ed25519();
Self::new_maybe_with_notify_inbound(client, local_key, Some(notify_inbound_tx), None)
.await
}
}
#[tokio::test]
async fn test_transport_connection() {
setup_tracing_logger();
let client = MixnetClient::connect_new().await.unwrap();
let (dialer_notify_inbound_tx, mut dialer_notify_inbound_rx) = unbounded_channel();
let mut dialer_transport =
NymTransport::new_with_notify_inbound(client, dialer_notify_inbound_tx)
.await
.unwrap();
let client2 = MixnetClient::connect_new().await.unwrap();
let (listener_notify_inbound_tx, mut listener_notify_inbound_rx) = unbounded_channel();
let mut listener_transport =
NymTransport::new_with_notify_inbound(client2, listener_notify_inbound_tx)
.await
.unwrap();
let listener_multiaddr =
nym_address_to_multiaddress(listener_transport.self_address).unwrap();
assert_new_address_event(Pin::new(&mut dialer_transport)).await;
assert_new_address_event(Pin::new(&mut listener_transport)).await;
// dial the remote peer
let mut dial = dialer_transport.dial(listener_multiaddr).unwrap();
// poll the dial to send the connection request message
assert!(poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx))
.now_or_never()
.is_none());
listener_notify_inbound_rx.recv().await.unwrap();
// should receive the connection request from the mixnet and send the connection response
let res = poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx)).await;
let mut upgrade = match res {
TransportEvent::Incoming {
listener_id,
upgrade,
local_addr,
send_back_addr,
} => {
assert_eq!(listener_id, listener_transport.listener_id);
assert_eq!(local_addr, listener_transport.listen_addr);
assert_eq!(send_back_addr, listener_transport.listen_addr);
upgrade
}
_ => panic!("expected TransportEvent::Incoming, got {:?}", res),
};
dialer_notify_inbound_rx.recv().await.unwrap();
// should receive the connection response from the mixnet
assert!(
poll_fn(|cx| Pin::new(&mut dialer_transport).as_mut().poll(cx))
.now_or_never()
.is_none()
);
info!("waiting for connections...");
// should be able to resolve the connections now
let (_, mut listener_conn) = poll_fn(|cx| Pin::new(&mut upgrade).as_mut().poll_unpin(cx))
.now_or_never()
.expect("the upgrade should be ready")
.expect("the upgrade should not error");
let (_, mut dialer_conn) = poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx))
.now_or_never()
.expect("the upgrade should be ready")
.expect("the upgrade should not error");
info!("connections established");
// write messages from the dialer to the listener and vice versa
send_and_receive_over_conns(
b"hello".to_vec(),
&mut dialer_conn,
&mut listener_conn,
Pin::new(&mut listener_transport),
&mut listener_notify_inbound_rx,
)
.await;
send_and_receive_over_conns(
b"hi".to_vec(),
&mut dialer_conn,
&mut listener_conn,
Pin::new(&mut listener_transport),
&mut listener_notify_inbound_rx,
)
.await;
send_and_receive_over_conns(
b"world".to_vec(),
&mut listener_conn,
&mut dialer_conn,
Pin::new(&mut dialer_transport),
&mut dialer_notify_inbound_rx,
)
.await;
}
async fn assert_new_address_event(mut transport: Pin<&mut NymTransport>) {
match poll_fn(|cx| transport.as_mut().poll(cx)).await {
TransportEvent::NewAddress {
listener_id,
listen_addr,
} => {
assert_eq!(listener_id, transport.listener_id);
assert_eq!(listen_addr, transport.listen_addr);
}
_ => panic!("expected TransportEvent::NewAddress"),
}
}
async fn send_and_receive_over_conns(
msg: Vec<u8>,
conn1: &mut Connection,
conn2: &mut Connection,
mut transport2: Pin<&mut NymTransport>,
notify_inbound_rx: &mut UnboundedReceiver<()>,
) {
// send message over conn1 to conn2
let substream_id = SubstreamId::generate();
conn1
.write(SubstreamMessage::new_with_data(
substream_id.clone(),
msg.clone(),
))
.unwrap();
notify_inbound_rx.recv().await.unwrap();
// poll transport2 to push message from transport to connection
assert!(poll_fn(|cx| transport2.as_mut().poll(cx))
.now_or_never()
.is_none());
let substream_msg = conn2.inbound_rx.recv().await.unwrap();
if let SubstreamMessageType::Data(data) = substream_msg.message_type {
assert_eq!(data, msg);
} else {
panic!("expected data message");
}
}
#[tokio::test]
async fn test_transport_substream() {
let client = MixnetClient::connect_new().await.unwrap();
let (dialer_notify_inbound_tx, mut dialer_notify_inbound_rx) = unbounded_channel();
let mut dialer_transport =
NymTransport::new_with_notify_inbound(client, dialer_notify_inbound_tx)
.await
.unwrap();
let client2 = MixnetClient::connect_new().await.unwrap();
let (listener_notify_inbound_tx, mut listener_notify_inbound_rx) = unbounded_channel();
let mut listener_transport =
NymTransport::new_with_notify_inbound(client2, listener_notify_inbound_tx)
.await
.unwrap();
let listener_multiaddr =
nym_address_to_multiaddress(listener_transport.self_address).unwrap();
assert_new_address_event(Pin::new(&mut dialer_transport)).await;
assert_new_address_event(Pin::new(&mut listener_transport)).await;
// dial the remote peer
let mut dial = dialer_transport.dial(listener_multiaddr).unwrap();
// poll the dial to send the connection request message
assert!(poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx))
.now_or_never()
.is_none());
listener_notify_inbound_rx.recv().await.unwrap();
// should receive the connection request from the mixnet and send the connection response
let res = poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx)).await;
let mut upgrade = match res {
TransportEvent::Incoming {
listener_id,
upgrade,
local_addr,
send_back_addr,
} => {
assert_eq!(listener_id, listener_transport.listener_id);
assert_eq!(local_addr, listener_transport.listen_addr);
assert_eq!(send_back_addr, listener_transport.listen_addr);
upgrade
}
_ => panic!("expected TransportEvent::Incoming, got {:?}", res),
};
dialer_notify_inbound_rx.recv().await.unwrap();
// should receive the connection response from the mixnet
assert!(
poll_fn(|cx| Pin::new(&mut dialer_transport).as_mut().poll(cx))
.now_or_never()
.is_none()
);
info!("waiting for connections...");
// should be able to resolve the connections now
let (_, mut listener_conn) = poll_fn(|cx| Pin::new(&mut upgrade).as_mut().poll_unpin(cx))
.now_or_never()
.unwrap()
.unwrap();
let (_, mut dialer_conn) = poll_fn(|cx| Pin::new(&mut dial).as_mut().poll_unpin(cx))
.now_or_never()
.unwrap()
.unwrap();
info!("connections established");
// initiate a new substream from the dialer
let mut dialer_substream =
poll_fn(|cx| Pin::new(&mut dialer_conn).as_mut().poll_outbound(cx))
.await
.unwrap();
listener_notify_inbound_rx.recv().await.unwrap();
// accept the substream on the listener
assert!(
poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx))
.now_or_never()
.is_none()
);
poll_fn(|cx| Pin::new(&mut listener_conn).as_mut().poll(cx)).now_or_never();
// poll recipient's poll_inbound to receive the substream; sends a response to the sender
let mut listener_substream =
poll_fn(|cx| Pin::new(&mut listener_conn).as_mut().poll_inbound(cx))
.now_or_never()
.unwrap()
.unwrap();
info!("got listener substream");
dialer_notify_inbound_rx.recv().await.unwrap();
// poll sender to finalize the substream
assert!(
poll_fn(|cx| Pin::new(&mut dialer_transport).as_mut().poll(cx))
.now_or_never()
.is_none()
);
poll_fn(|cx| Pin::new(&mut dialer_conn).as_mut().poll(cx)).now_or_never();
info!("got dialer substream");
// write message from dialer to listener
send_and_receive_substream_message(
b"hello world".to_vec(),
Pin::new(&mut dialer_substream),
Pin::new(&mut listener_substream),
Pin::new(&mut listener_transport),
Pin::new(&mut listener_conn),
&mut listener_notify_inbound_rx,
)
.await;
// write message from listener to dialer
send_and_receive_substream_message(
b"hello back".to_vec(),
Pin::new(&mut listener_substream),
Pin::new(&mut dialer_substream),
Pin::new(&mut dialer_transport),
Pin::new(&mut dialer_conn),
&mut dialer_notify_inbound_rx,
)
.await;
// close the substream from the dialer side
info!("closing dialer substream");
dialer_substream.close().await.unwrap();
listener_notify_inbound_rx.recv().await.unwrap();
info!("dialer substream closed");
// assert we can't read or write to either substream
dialer_substream.write_all(b"hello").await.unwrap_err();
// poll listener transport and conn to receive the substream close message
poll_fn(|cx| Pin::new(&mut listener_transport).as_mut().poll(cx)).now_or_never();
poll_fn(|cx| Pin::new(&mut listener_conn).as_mut().poll(cx)).now_or_never();
listener_substream.write_all(b"hello").await.unwrap_err();
let mut buf = vec![0u8; 5];
dialer_substream.read(&mut buf).await.unwrap_err();
listener_substream.read(&mut buf).await.unwrap_err();
dialer_substream.close().await.unwrap_err();
listener_substream.close().await.unwrap_err();
}
async fn send_and_receive_substream_message(
data: Vec<u8>,
mut sender_substream: Pin<&mut Substream>,
mut recipient_substream: Pin<&mut Substream>,
mut recipient_transport: Pin<&mut NymTransport>,
mut recipient_conn: Pin<&mut Connection>,
recipient_notify_inbound_rx: &mut UnboundedReceiver<()>,
) {
// write message
sender_substream.write_all(&data).await.unwrap();
recipient_notify_inbound_rx.recv().await.unwrap();
// poll recipient for message
poll_fn(|cx| recipient_transport.as_mut().poll(cx)).now_or_never();
poll_fn(|cx| recipient_conn.as_mut().poll(cx)).now_or_never();
let mut buf = vec![0u8; data.len()];
let n = recipient_substream.read(&mut buf).await.unwrap();
assert_eq!(n, data.len());
assert_eq!(buf, data[..]);
}
#[tokio::test]
async fn test_transport_timeout() {
let client = MixnetClient::connect_new().await.unwrap();
let (dialer_notify_inbound_tx, _) = unbounded_channel();
let mut dialer_transport =
NymTransport::new_with_notify_inbound(client, dialer_notify_inbound_tx)
.await
.unwrap()
.with_timeout(std::time::Duration::from_millis(100));
// mock a transport that will never resolve the connection.
let empty_addr = Multiaddr::from_str(
"/nym/Hmer6Ndt3PV13YW53HM8ri4NvqqtfDQUQBhzvKqb1dag.2g478dyxtrQXGWc1Mk2VEqdPcWXpz7EhAcjhdAJtVZdA@AnnYnEtBjB2a5sHmeRCnBq43qxyHDf95Bqd7cwQyKNLR"
)
.expect("unable to parse multiaddress");
let dial = dialer_transport.dial(empty_addr).unwrap();
assert!(dial
.await
.expect_err("should have timed out")
.to_string()
.contains("dial timed out"));
}
}