started on rendezvous example

working on rendzv example
network-requester: disable poisson process by default (#3783 )
2023-08-23 21:52:52 +02:00 · 2023-08-23 20:47:24 +02:00 · 2023-08-22 14:17:03 +02:00 · 2023-08-22 13:12:13 +02:00 · 2023-08-22 12:21:44 +02:00 · 2023-08-22 11:56:23 +02:00
50 changed files with 4025 additions and 237 deletions
@@ -57,7 +57,7 @@ jobs:
          echo $OUTPUT_DIR

      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
+        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools protobuf-compiler
        continue-on-error: true

      - name: Install Rust stable
@@ -42,7 +42,7 @@ jobs:
      RUSTC_WRAPPER: /home/ubuntu/.cargo/bin/sccache
    steps:
      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
+        run: sudo apt-get update && sudo apt-get -y install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools protobuf-compiler
        continue-on-error: true

      - name: Check out repository code
@@ -24,7 +24,7 @@ jobs:
    continue-on-error: ${{ matrix.rust == 'nightly' || matrix.rust == 'beta' || matrix.rust == 'stable' }}
    steps:
      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
+        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools protobuf-compiler
        continue-on-error: true
        if: matrix.os == 'ubuntu-20.04'

@@ -37,7 +37,7 @@ jobs:
    continue-on-error: ${{ matrix.rust == 'nightly' || matrix.rust == 'beta' || matrix.rust == 'stable' }}
    steps:
      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools
+        run: sudo apt-get update && sudo apt-get install libwebkit2gtk-4.0-dev build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools protobuf-compiler
        continue-on-error: true
        if: matrix.os == 'ubuntu-20.04'

@@ -100,6 +100,7 @@ jobs:
            nym-connect/desktop/target/release/bundle/msi/*.msi.zip*
      - id: release-info
        name: Prepare release info
+        shell: bash
        run: |
          ref=${{ github.ref_name }}
          semver="${ref##nym-connect-}" && semver="${semver##v}"
@@ -32,5 +32,5 @@ jobs:
        run: yarn

      - name: Run tests
-        run: yarn test:qa
+        run: yarn test:sandbox
        working-directory: nym-api/tests
@@ -1,79 +0,0 @@
-name: Upload nyxd to CI
-
-on:
-  workflow_dispatch:
-
-jobs:
-  publish-nyxd:
-    strategy:
-      fail-fast: false
-      matrix:
-        platform: [ubuntu-20.04]
-
-    runs-on: ${{ matrix.platform }}
-    steps:
-      - uses: actions/checkout@v3
-
-      - name: Prepare build output directory
-        shell: bash
-        env:
-          OUTPUT_DIR: ci-builds/nyxd
-        run: |
-          rm -rf ci-builds || true
-          mkdir -p $OUTPUT_DIR
-          echo $OUTPUT_DIR
-
-      - name: Install Dependencies (Linux)
-        run: sudo apt-get update && sudo apt-get -y install build-essential curl wget libssl-dev libgtk-3-dev libudev-dev squashfs-tools git
-        continue-on-error: true
-      
-      - name: Update env variables to include go
-        run: | 
-          sudo rm -rf /usr/local/go
-          curl https://dl.google.com/go/go1.19.2.linux-amd64.tar.gz | sudo tar -C/usr/local -zxvf -
-          cat <<'EOF' >>$HOME/.profile
-          export GOROOT=/usr/local/go
-          export GOPATH=$HOME/go
-          export GO111MODULE=on
-          export PATH=$PATH:/usr/local/go/bin:$HOME/go/bin
-          EOF
-          source $HOME/.profile
-
-      - name: Verify Go is installed
-        run: go version
-
-      - name: Clone nyxd repo
-        run: | 
-          git clone https://github.com/tommyv1987/nyxd
-          cd nyxd
-          git checkout release/v0.30.2
-
-      - name: Run nyxd
-        run: |
-          pwd
-          cd nyxd && make build
-          sleep 10
-          ls /home/runner/work/nym/nym/nyxd/build
-
-      - name: Prepare build output
-        shell: bash
-        env:
-          OUTPUT_DIR: ci-builds/nyxd
-        run: |
-          cp /home/runner/work/nym/nym/nyxd/build/nyxd $OUTPUT_DIR
-          WASMVM_SO=$(ldd /home/runner/work/nym/nym/nyxd/build/nyxd | grep "libwasm*" | awk '{ print $3 }')
-          ls $WASMVM_SO
-          sleep 3
-          cp $(echo $WASMVM_SO) $OUTPUT_DIR
-
-      - name: Deploy nyxd to CI www
-        continue-on-error: true
-        uses: easingthemes/ssh-deploy@main
-        env:
-          SSH_PRIVATE_KEY: ${{ secrets.CI_WWW_SSH_PRIVATE_KEY }}
-          ARGS: "-avzr"
-          SOURCE: "ci-builds/"
-          REMOTE_HOST: ${{ secrets.CI_WWW_REMOTE_HOST }}
-          REMOTE_USER: ${{ secrets.CI_WWW_REMOTE_USER }}
-          TARGET: ${{ secrets.CI_WWW_REMOTE_TARGET }}/builds/
-          EXCLUDE: "/dist/, /node_modules/"
@@ -176,7 +176,7 @@ mod tests {
    use cosmwasm_std::Order;
    use rand_chacha::rand_core::RngCore;

-    fn read_entire_set(storage: &mut dyn Storage) -> HashMap<MixId, RewardedSetNodeStatus> {
+    fn read_entire_set(storage: &dyn Storage) -> HashMap<MixId, RewardedSetNodeStatus> {
        REWARDED_SET
            .range(storage, None, None, Order::Ascending)
            .map(|r| r.unwrap())
@@ -286,7 +286,6 @@ mod tests {
        let owner = "steve";
        let (name, owner_signature) =
            new_name_details_with_sign(deps.as_mut(), &mut rng, "my-name", owner, deposit.clone());
-        dbg!(&name);

        // Register
        let msg = ExecuteMsg::Register {
@@ -265,7 +265,6 @@ fn can_register_multiple_names_for_the_same_nym_address(mut setup: TestSetup) {
    let reg_name2 = reg_name2.sign(payload);
    setup.register(&reg_name2, &owner);

-    dbg!(&setup.query_all().names);
    assert_eq!(
        setup.query_all().names,
        vec![
@@ -1,6 +1,6 @@
 use nym_api_requests::models::NodePerformance;
 use nym_contracts_common::Percent;
-use nym_mixnet_contract_common::{Addr, Coin, Layer, MixId, MixNode};
+use nym_mixnet_contract_common::{Addr, Coin, Gateway, Layer, MixId, MixNode};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};

@@ -42,3 +42,13 @@ pub struct Location {
    pub latitude: Option<f64>,
    pub longitude: Option<f64>,
 }
+
+#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
+pub struct PrettyDetailedGatewayBond {
+    pub pledge_amount: Coin,
+    pub owner: Addr,
+    pub block_height: u64,
+    pub gateway: Gateway,
+    pub proxy: Option<Addr>,
+    pub location: Option<Location>,
+}
@@ -17,13 +17,14 @@ impl GeoLocateTask {
    }

    pub(crate) fn start(mut self) {
-        info!("Spawning mix node locator task runner...");
+        info!("Spawning locator task runner...");
        tokio::spawn(async move {
            let mut interval_timer = tokio::time::interval(std::time::Duration::from_millis(50));
            while !self.shutdown.is_shutdown() {
                tokio::select! {
                    _ = interval_timer.tick() => {
                        self.locate_mix_nodes().await;
+                        self.locate_gateways().await;
                    }
                    _ = self.shutdown.recv() => {
                        trace!("Listener: Received shutdown");
@@ -107,4 +108,68 @@ impl GeoLocateTask {

        trace!("All mix nodes located");
    }
+
+    async fn locate_gateways(&mut self) {
+        let gateways = self.state.inner.gateways.get_gateways().await;
+
+        let geo_ip = self.state.inner.geo_ip.0.clone();
+
+        for (i, cache_item) in gateways.iter().enumerate() {
+            if self
+                .state
+                .inner
+                .gateways
+                .is_location_valid(cache_item.identity().to_owned())
+                .await
+            {
+                // when the cached location is valid, don't locate and continue to next gateway
+                continue;
+            }
+
+            match geo_ip.query(&cache_item.gateway.host, Some(cache_item.gateway.mix_port)) {
+                Ok(opt) => match opt {
+                    Some(location) => {
+                        let location: Location = location.into();
+
+                        trace!(
+                            "{} gateways already located. Ip {} is located in {:#?}",
+                            i,
+                            cache_item.gateway.host,
+                            location.three_letter_iso_country_code,
+                        );
+
+                        if i > 0 && (i % 100) == 0 {
+                            info!("Located {} gateways...", i + 1,);
+                        }
+
+                        self.state
+                            .inner
+                            .gateways
+                            .set_location(cache_item.identity().to_owned(), Some(location))
+                            .await;
+
+                        // one node has been located, so return out of the loop
+                        return;
+                    }
+                    None => {
+                        warn!("❌ Location for {} not found.", cache_item.gateway.host);
+                        self.state
+                            .inner
+                            .gateways
+                            .set_location(cache_item.identity().to_owned(), None)
+                            .await;
+                    }
+                },
+                Err(_e) => {
+                    // warn!(
+                    //     "❌ Oh no! Location for {} failed. Error: {:#?}",
+                    //     cache_item.gateway.host,
+                    //     e
+                    // );
+                }
+            };
+        }
+
+        trace!("All gateways located");
+    }
 }
@@ -1,6 +1,7 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0

+use nym_explorer_api_requests::PrettyDetailedGatewayBond;
 use rocket::response::status::NotFound;
 use rocket::serde::json::Json;
 use rocket::{Route, State};
@@ -9,7 +10,6 @@ use rocket_okapi::openapi_get_routes_spec;
 use rocket_okapi::settings::OpenApiSettings;

 use crate::state::ExplorerApiStateContext;
-use nym_mixnet_contract_common::GatewayBond;

 pub fn gateways_make_default_routes(settings: &OpenApiSettings) -> (Vec<Route>, OpenApi) {
    openapi_get_routes_spec![settings: list]
@@ -19,6 +19,6 @@ pub fn gateways_make_default_routes(settings: &OpenApiSettings) -> (Vec<Route>,
 #[get("/")]
 pub(crate) async fn list(
    state: &State<ExplorerApiStateContext>,
-) -> Result<Json<Vec<GatewayBond>>, NotFound<String>> {
-    Ok(Json(state.inner.gateways.get_gateways().await))
+) -> Result<Json<Vec<PrettyDetailedGatewayBond>>, NotFound<String>> {
+    Ok(Json(state.inner.gateways.get_detailed_gateways().await))
 }
@@ -0,0 +1,8 @@
+// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use nym_contracts_common::IdentityKey;
+
+use crate::location::LocationCache;
+
+pub(crate) type GatewayLocationCache = LocationCache<IdentityKey>;
@@ -1,2 +1,3 @@
 pub(crate) mod http;
+pub(crate) mod location;
 pub(crate) mod models;
@@ -1,12 +1,15 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0

-use crate::cache::Cache;
+use crate::{cache::Cache, location::LocationCacheItem};
+use nym_explorer_api_requests::{Location, PrettyDetailedGatewayBond};
 use nym_mixnet_contract_common::{GatewayBond, IdentityKey};
 use serde::Serialize;
-use std::sync::Arc;
+use std::{sync::Arc, time::SystemTime};
 use tokio::sync::RwLock;

+use super::location::GatewayLocationCache;
+
 pub(crate) struct GatewayCache {
    pub(crate) gateways: Cache<IdentityKey, GatewayBond>,
 }
@@ -18,30 +21,93 @@ pub(crate) struct GatewaySummary {

 #[derive(Clone)]
 pub(crate) struct ThreadsafeGatewayCache {
-    inner: Arc<RwLock<GatewayCache>>,
+    gateways: Arc<RwLock<GatewayCache>>,
+    locations: Arc<RwLock<GatewayLocationCache>>,
 }

 impl ThreadsafeGatewayCache {
    pub(crate) fn new() -> Self {
        ThreadsafeGatewayCache {
-            inner: Arc::new(RwLock::new(GatewayCache {
+            gateways: Arc::new(RwLock::new(GatewayCache {
                gateways: Cache::new(),
            })),
+            locations: Arc::new(RwLock::new(GatewayLocationCache::new())),
+        }
+    }
+
+    fn create_detailed_gateway(
+        &self,
+        bond: GatewayBond,
+        location: Option<&LocationCacheItem>,
+    ) -> PrettyDetailedGatewayBond {
+        PrettyDetailedGatewayBond {
+            pledge_amount: bond.pledge_amount,
+            owner: bond.owner,
+            block_height: bond.block_height,
+            gateway: bond.gateway,
+            proxy: bond.proxy,
+            location: location.and_then(|l| l.location.clone()),
        }
    }

    pub(crate) async fn get_gateways(&self) -> Vec<GatewayBond> {
-        self.inner.read().await.gateways.get_all()
+        self.gateways.read().await.gateways.get_all()
+    }
+
+    pub(crate) async fn get_detailed_gateways(&self) -> Vec<PrettyDetailedGatewayBond> {
+        let gateways_guard = self.gateways.read().await;
+        let location_guard = self.locations.read().await;
+
+        gateways_guard
+            .gateways
+            .get_all()
+            .iter()
+            .map(|bond| {
+                let location = location_guard.get(bond.identity());
+                self.create_detailed_gateway(bond.to_owned(), location)
+            })
+            .collect()
    }

    pub(crate) async fn get_gateway_summary(&self) -> GatewaySummary {
        GatewaySummary {
-            count: self.inner.read().await.gateways.len(),
+            count: self.gateways.read().await.gateways.len(),
        }
    }

+    pub(crate) fn new_with_location_cache(locations: GatewayLocationCache) -> Self {
+        ThreadsafeGatewayCache {
+            gateways: Arc::new(RwLock::new(GatewayCache {
+                gateways: Cache::new(),
+            })),
+            locations: Arc::new(RwLock::new(locations)),
+        }
+    }
+
+    pub(crate) async fn is_location_valid(&self, identity_key: IdentityKey) -> bool {
+        self.locations
+            .read()
+            .await
+            .get(&identity_key)
+            .map_or(false, |cache_item| {
+                cache_item.valid_until > SystemTime::now()
+            })
+    }
+
+    pub(crate) async fn get_locations(&self) -> GatewayLocationCache {
+        self.locations.read().await.clone()
+    }
+
+    pub(crate) async fn set_location(&self, identy_key: IdentityKey, location: Option<Location>) {
+        // cache the location for this mix node so that it can be used when the mix node list is refreshed
+        self.locations
+            .write()
+            .await
+            .insert(identy_key, LocationCacheItem::new_from_location(location));
+    }
+
    pub(crate) async fn update_cache(&self, gateways: Vec<GatewayBond>) {
-        let mut guard = self.inner.write().await;
+        let mut guard = self.gateways.write().await;

        for gateway in gateways {
            guard
@@ -0,0 +1,40 @@
+// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use nym_explorer_api_requests::Location;
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+use std::time::{Duration, SystemTime};
+
+pub(crate) type LocationCache<T> = HashMap<T, LocationCacheItem>;
+
+#[allow(dead_code)]
+#[derive(Debug, Deserialize)]
+pub(crate) struct GeoLocation {
+    pub(crate) ip: String,
+    pub(crate) country_code: String,
+    pub(crate) country_name: String,
+    pub(crate) region_code: String,
+    pub(crate) region_name: String,
+    pub(crate) city: String,
+    pub(crate) zip_code: String,
+    pub(crate) time_zone: String,
+    pub(crate) latitude: f32,
+    pub(crate) longitude: f32,
+    pub(crate) metro_code: u32,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub(crate) struct LocationCacheItem {
+    pub(crate) location: Option<Location>,
+    pub(crate) valid_until: SystemTime,
+}
+
+impl LocationCacheItem {
+    pub(crate) fn new_from_location(location: Option<Location>) -> Self {
+        LocationCacheItem {
+            location,
+            valid_until: SystemTime::now() + Duration::from_secs(60 * 60 * 24), // valid for 1 day
+        }
+    }
+}
@@ -19,6 +19,7 @@ mod geo_ip;
 mod guards;
 mod helpers;
 mod http;
+mod location;
 mod mix_node;
 pub(crate) mod mix_nodes;
 mod overview;
@@ -1,41 +1,8 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0

-use nym_explorer_api_requests::Location;
 use nym_mixnet_contract_common::MixId;
-use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use std::time::{Duration, SystemTime};

-pub(crate) type LocationCache = HashMap<MixId, LocationCacheItem>;
+use crate::location::LocationCache;

-#[allow(dead_code)]
-#[derive(Debug, Deserialize)]
-pub(crate) struct GeoLocation {
-    pub(crate) ip: String,
-    pub(crate) country_code: String,
-    pub(crate) country_name: String,
-    pub(crate) region_code: String,
-    pub(crate) region_name: String,
-    pub(crate) city: String,
-    pub(crate) zip_code: String,
-    pub(crate) time_zone: String,
-    pub(crate) latitude: f32,
-    pub(crate) longitude: f32,
-    pub(crate) metro_code: u32,
-}
-
-#[derive(Clone, Debug, Serialize, Deserialize)]
-pub(crate) struct LocationCacheItem {
-    pub(crate) location: Option<Location>,
-    pub(crate) valid_until: SystemTime,
-}
-
-impl LocationCacheItem {
-    pub(crate) fn new_from_location(location: Option<Location>) -> Self {
-        LocationCacheItem {
-            location,
-            valid_until: SystemTime::now() + Duration::from_secs(60 * 60 * 24), // valid for 1 day
-        }
-    }
-}
+pub(crate) type MixnodeLocationCache = LocationCache<MixId>;
@@ -12,10 +12,11 @@ use serde::Serialize;
 use tokio::sync::{RwLock, RwLockReadGuard};

 use crate::helpers::best_effort_small_dec_to_f64;
+use crate::location::LocationCacheItem;
 use nym_validator_client::models::MixNodeBondAnnotated;

+use super::location::MixnodeLocationCache;
 use super::utils::family_numerical_id;
-use crate::mix_nodes::location::{LocationCache, LocationCacheItem};
 use crate::mix_nodes::CACHE_ENTRY_TTL;

 #[derive(Clone, Debug, Serialize, JsonSchema)]
@@ -83,18 +84,18 @@ impl MixNodesResult {
 #[derive(Clone)]
 pub(crate) struct ThreadsafeMixNodesCache {
    mixnodes: Arc<RwLock<MixNodesResult>>,
-    locations: Arc<RwLock<LocationCache>>,
+    locations: Arc<RwLock<MixnodeLocationCache>>,
 }

 impl ThreadsafeMixNodesCache {
    pub(crate) fn new() -> Self {
        ThreadsafeMixNodesCache {
            mixnodes: Arc::new(RwLock::new(MixNodesResult::new())),
-            locations: Arc::new(RwLock::new(LocationCache::new())),
+            locations: Arc::new(RwLock::new(MixnodeLocationCache::new())),
        }
    }

-    pub(crate) fn new_with_location_cache(locations: LocationCache) -> Self {
+    pub(crate) fn new_with_location_cache(locations: MixnodeLocationCache) -> Self {
        ThreadsafeMixNodesCache {
            mixnodes: Arc::new(RwLock::new(MixNodesResult::new())),
            locations: Arc::new(RwLock::new(locations)),
@@ -111,7 +112,7 @@ impl ThreadsafeMixNodesCache {
            })
    }

-    pub(crate) async fn get_locations(&self) -> LocationCache {
+    pub(crate) async fn get_locations(&self) -> MixnodeLocationCache {
        self.locations.read().await.clone()
    }

@@ -1,13 +1,13 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0

+use isocountry::CountryCode;
 use nym_mixnet_contract_common::families::FamilyHead;
 use rand::Rng;
 use rand_pcg::Pcg64;
 use rand_seeder::Seeder;

-use crate::mix_nodes::location::GeoLocation;
-use isocountry::CountryCode;
+use crate::location::GeoLocation;

 #[allow(dead_code)]
 pub(crate) fn map_2_letter_to_3_letter_country_code(geo: &GeoLocation) -> String {
@@ -13,9 +13,10 @@ use nym_validator_client::models::MixNodeBondAnnotated;
 use crate::country_statistics::country_nodes_distribution::{
    CountryNodesDistribution, ThreadsafeCountryNodesDistribution,
 };
+use crate::gateways::location::GatewayLocationCache;
 use crate::gateways::models::ThreadsafeGatewayCache;
 use crate::mix_node::models::ThreadsafeMixNodeCache;
-use crate::mix_nodes::location::LocationCache;
+use crate::mix_nodes::location::MixnodeLocationCache;
 use crate::mix_nodes::models::ThreadsafeMixNodesCache;
 use crate::ping::models::ThreadsafePingCache;
 use crate::validators::models::ThreadsafeValidatorCache;
@@ -46,7 +47,8 @@ impl ExplorerApiState {
 #[derive(Debug, Serialize, Deserialize)]
 pub struct ExplorerApiStateOnDisk {
    pub(crate) country_node_distribution: CountryNodesDistribution,
-    pub(crate) location_cache: LocationCache,
+    pub(crate) mixnode_location_cache: MixnodeLocationCache,
+    pub(crate) gateway_location_cache: GatewayLocationCache,
    pub(crate) as_at: DateTime<Utc>,
 }

@@ -76,9 +78,13 @@ impl ExplorerApiStateContext {
                    ThreadsafeCountryNodesDistribution::new_from_distribution(
                        state.country_node_distribution,
                    ),
-                gateways: ThreadsafeGatewayCache::new(),
+                gateways: ThreadsafeGatewayCache::new_with_location_cache(
+                    state.gateway_location_cache,
+                ),
                mixnode: ThreadsafeMixNodeCache::new(),
-                mixnodes: ThreadsafeMixNodesCache::new_with_location_cache(state.location_cache),
+                mixnodes: ThreadsafeMixNodesCache::new_with_location_cache(
+                    state.mixnode_location_cache,
+                ),
                ping: ThreadsafePingCache::new(),
                validators: ThreadsafeValidatorCache::new(),
                validator_client: ThreadsafeValidatorClient::new(),
@@ -109,7 +115,8 @@ impl ExplorerApiStateContext {
        let file = File::create(json_file_path).expect("unable to create state json file");
        let state = ExplorerApiStateOnDisk {
            country_node_distribution: self.inner.country_node_distribution.get_all().await,
-            location_cache: self.inner.mixnodes.get_locations().await,
+            mixnode_location_cache: self.inner.mixnodes.get_locations().await,
+            gateway_location_cache: self.inner.gateways.get_locations().await,
            as_at: Utc::now(),
        };
        serde_json::to_writer(file, &state).expect("error writing state to disk");
@@ -11,6 +11,8 @@ mod routes;

 pub(crate) fn network_routes(settings: &OpenApiSettings) -> (Vec<Route>, OpenApi) {
    openapi_get_routes_spec![
-        settings: routes::network_details, routes::nym_contracts, routes::nym_contracts_detailed
+        settings: routes::network_details,
+        routes::nym_contracts,
+        routes::nym_contracts_detailed
    ]
 }
@@ -7,7 +7,7 @@
    "test": "test"
  },
  "scripts": {
-    "test:qa": "TEST_ENV=qa jest --forceExit --detectOpenHandles --passWithNoTests",
+    "test:sandbox": "TEST_ENV=sandbox jest --forceExit --detectOpenHandles --passWithNoTests",
    "test:prod": "TEST_ENV=prod jest --forceExit --detectOpenHandles --passWithNoTests",
    "build": "tsc",
    "lint": "eslint --ext .js,.ts,.tsx .",
@@ -2,12 +2,13 @@ common:
  request_headers:
    Accept: application/json
    Content-Type: application/json
-qa:
-  api_base_url: https://qwerty-validator-api.qa.nymte.ch/api/v1
-  mix_id: 63
-  identity_key: 4Yr4qmEHd9sgsuQ83191FR2hD88RfsbMmB4tzhhZWriz
-  gateway_identity: 336yuXAeGEgedRfqTJZsG2YV7P13QH1bHv1SjCZYarc9
+sandbox:
+  api_base_url: https://sandbox-nym-api1.nymtech.net/api/v1
+  mix_id: 7
+  identity_key: 2wEzyW6Pihk19xbCK9kB81vYuDFUvTzTAr77arvxdfSg
+  gateway_identity: HjNEDJuotWV8VD4ufeA1jeheTnfNJ7Jorevp57hgaZua
  log_level: error
+  time_zone: utc
 prod:
  api_base_url: https://validator.nymtech.net/api/v1
  mix_id: 730
@@ -6,7 +6,7 @@ import YAML from "yaml";
 class ConfigHandler {
  private static instance: ConfigHandler;

-  private validEnvironments = ["qa", "prod"];
+  private validEnvironments = ["sandbox", "prod"];

  public commonConfig: { request_headers: object };

@@ -23,7 +23,7 @@ class ConfigHandler {

  private constructor() {
    this.setCommonConfig();
-    this.setEnvironmentConfig(process.env.TEST_ENV || "qa" || "prod");
+    this.setEnvironmentConfig(process.env.TEST_ENV || "sandbox" || "prod");
  }

  public static getInstance(): ConfigHandler {
@@ -36,3 +36,13 @@ reqwest = { workspace = true, features = ["json", "socks"] }
 thiserror = { workspace = true }
 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", "rendezvous" ]}
+tokio-stream = "0.1.12"
+tokio-util = { version = "0.7", features = ["codec"] }
+parking_lot = "0.12"
+hex = "0.4"
+
+[features]
+libp2p-vanilla = []
@@ -0,0 +1,64 @@
+# 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
+```
@@ -0,0 +1,171 @@
+// 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>> {
+    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)}
+                }
+            }
+        }
+    }
+}
@@ -0,0 +1,79 @@
+# 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"
+```
@@ -0,0 +1,142 @@
+// 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 nym_sdk::mixnet::MixnetClient;
+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>> {
+    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 = 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,
+}
@@ -0,0 +1,135 @@
+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() {
+    pretty_env_logger::formatted_timed_builder()
+        .filter_level(LevelFilter::Warn)
+        .filter(Some("libp2p_rendezvous"), LevelFilter::Debug)
+        .init();
+
+    let key_pair = identity::Keypair::generate_ed25519();
+    let local_peer_id = PeerId::from(local_key.public());
+    info!("Local peer id: {local_peer_id:?}"); 
+
+    // copy and paste in from the cli for the moment 
+    // TODO take from cli as args 
+    let rendezvous_point_address = "/nym/6a9Lxos7r5oh1jHNUDHdXW6NjoAiMk3fvSUYeCdP7p6f.6scJ3WbQKHw6m1vowTHedonGzunDXABnDjLV5Jgg95UT@FyHgeVWeXTysBd7515ndZ2tpzWhv9myLfuv4S9oeoFpR".parse::<Multiaddr>().unwrap();
+    let rendezvous_point = "12D3KooWJFjuHa68RpcwgRFLYwEiT6TcwWRneEgzSExP5dQ81Yrk"
+        .parse()
+        .unwrap();
+
+    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 {
+                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());
+
+    // 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.
+    /* 
+    TODO we need the nym multiaddr of this instance
+    */
+    let external_address = "/ip4/127.0.0.1/tcp/0".parse::<Multiaddr>().unwrap();
+    swarm.add_external_address(external_address);
+
+    log::info!("Local peer id: {}", swarm.local_peer_id());
+
+    swarm.dial(rendezvous_point_address.clone()).unwrap();
+
+    match swarm.select_next_some().await {
+            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,
+}
@@ -0,0 +1,109 @@
+use futures::StreamExt;
+use libp2p::{identify, identity, ping, rendezvous, PeerId};
+use libp2p::swarm::{keep_alive, NetworkBehaviour, 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<dyn Error>> {
+    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 = {
+        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());
+
+    loop {
+        match swarm.select_next_some().await {
+            SwarmEvent::NewListenAddr { address, .. } => info!("Listening on {address:?}"),
+            SwarmEvent::Behaviour(event) => {
+                debug!("{event:?}");
+            }
+            other => {
+                log::debug!("Unhandled {:?}", other);
+            }
+        }
+    }
+
+    // TODO incorporate 
+    // 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);
+    //         }
+    //     }
+    // }
+
+}
+
+#[derive(NetworkBehaviour)]
+struct MyBehaviour {
+    identify: identify::Behaviour,
+    rendezvous: rendezvous::server::Behaviour,
+    ping: ping::Behaviour,
+    keep_alive: keep_alive::Behaviour,
+}
@@ -0,0 +1,418 @@
+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::OutboundSendError(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.get(&id).is_some() {
+            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::InboundSendError(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::OutboundSendError(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::InboundSendError(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;
+    }
+}
@@ -0,0 +1,64 @@
+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")]
+    OneshotRecvError(#[from] tokio::sync::oneshot::error::RecvError),
+    #[error("recv error: channel closed")]
+    RecvError,
+    #[error("outbound send error")]
+    OutboundSendError(String),
+    #[error("inbound send error")]
+    InboundSendError(String),
+    #[error("failed to send new connection; receiver dropped")]
+    ConnectionSendError,
+    #[error("failed to send initial TransportEvent::NewAddress")]
+    SendErrorTransportEvent,
+    #[error("dial timed out")]
+    DialTimeout(#[from] tokio::time::error::Elapsed),
+}
@@ -0,0 +1,10 @@
+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;
@@ -0,0 +1,335 @@
+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))
+}
@@ -0,0 +1,164 @@
+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::InboundSendError(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::InboundSendError(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::RecvError),
+    }
+}
+
+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")
+        }
+    }
+}
@@ -0,0 +1,140 @@
+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 Some(head) = self.queue.first() else {
+            return None;
+        };
+
+        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);
+    }
+}
@@ -0,0 +1,422 @@
+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::new(ErrorKind::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::new(
+                    ErrorKind::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::new(ErrorKind::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::new(
+                    ErrorKind::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();
+    }
+}
@@ -0,0 +1,897 @@
+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.
+    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::InboundSendError(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::ConnectionSendError)?;
+
+            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.get(&msg.id).is_some() {
+            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::OutboundSendError(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::InboundSendError(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::InboundSendError(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::ConnectionSendError)?;
+                        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::RecvError)
+    }
+}
+
+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::OutboundSendError(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_logging;
+    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::OutboundSendError(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_logging();
+
+        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"));
+    }
+}
@@ -100,7 +100,7 @@ impl HostsStore {
            .unwrap();

        if let Err(e) = writeln!(file, "{text}") {
-            eprintln!("Couldn't write to file: {e}");
+            log::error!("Couldn't write to file: {e}");
        }
    }

@@ -106,7 +106,7 @@ fn init_paths(id: &str) -> io::Result<()> {
 }

 pub(crate) async fn execute(args: &Init) -> Result<(), NetworkRequesterError> {
-    eprintln!("Initialising client...");
+    log::info!("Initialising client...");

    let id = &args.id;

@@ -114,7 +114,7 @@ pub(crate) async fn execute(args: &Init) -> Result<(), NetworkRequesterError> {
        // in case we're using old config, try to upgrade it
        // (if we're using the current version, it's a no-op)
        try_upgrade_config(id)?;
-        eprintln!("Client \"{id}\" was already initialised before");
+        log::info!("Client \"{id}\" was already initialised before");
        true
    } else {
        init_paths(&args.id)?;
@@ -125,7 +125,7 @@ pub(crate) async fn execute(args: &Init) -> Result<(), NetworkRequesterError> {
    // re-registering if wanted.
    let user_wants_force_register = args.force_register_gateway;
    if user_wants_force_register {
-        eprintln!("Instructed to force registering gateway. This might overwrite keys!");
+        log::warn!("Instructed to force registering gateway. This might overwrite keys!");
    }

    // If the client was already initialized, don't generate new keys and don't re-register with
@@ -142,6 +142,7 @@ pub(crate) async fn execute(args: &Init) -> Result<(), NetworkRequesterError> {

    // Load and potentially override config
    let config = override_config(Config::new(id), OverrideConfig::from(args.clone()));
+    log::debug!("Using config: {:#?}", config);

    // Setup gateway by either registering a new one, or creating a new config from the selected
    // one but with keys kept, or reusing the gateway configuration.
@@ -156,21 +157,21 @@ pub(crate) async fn execute(args: &Init) -> Result<(), NetworkRequesterError> {
        Some(&config.base.client.nym_api_urls),
    )
    .await
-    .tap_err(|err| eprintln!("Failed to setup gateway\nError: {err}"))?
+    .tap_err(|err| log::error!("Failed to setup gateway\nError: {err}"))?
    .details;

    let config_save_location = config.default_location();
    config.save_to_default_location().tap_err(|_| {
        log::error!("Failed to save the config file");
    })?;
-    eprintln!(
+    log::info!(
        "Saved configuration file to {}",
        config_save_location.display()
    );

    let address = init_details.client_address()?;

-    eprintln!("Client configuration completed.\n");
+    log::info!("Client configuration completed.\n");

    let init_results = InitResults::new(&config, &address, &init_details.gateway_details);
    println!("{}", args.output.format(&init_results));
@@ -84,7 +84,12 @@ pub(crate) struct OverrideConfig {
 }

 pub(crate) fn override_config(config: Config, args: OverrideConfig) -> Config {
-    let disable_cover_traffic_with_keepalive = args.medium_toggle;
+    // These flags have overlapping effects, meaning the order matters here. Making it a bit messy.
+    // Since a big chunk of these are hidden experimental flags there is hope we can remove them
+    // soonish and clean this up.
+
+    let disable_cover_traffic_with_keepalive =
+        config.network_requester.disable_poisson_rate || args.medium_toggle;
    let secondary_packet_size = args.medium_toggle.then_some(PacketSize::ExtendedPacket16);
    let no_per_hop_delays = args.medium_toggle;

@@ -84,6 +84,7 @@ pub(crate) async fn execute(args: &Run) -> Result<(), NetworkRequesterError> {

    let mut config = try_load_current_config(&args.id)?;
    config = override_config(config, OverrideConfig::from(args.clone()));
+    log::debug!("Using config: {:#?}", config);

    if !version_check(&config) {
        error!("failed the local version check");
@@ -159,9 +159,19 @@ impl Config {
    }
 }

-#[derive(Debug, Default, Clone, Deserialize, PartialEq, Serialize)]
+#[derive(Debug, Clone, Deserialize, PartialEq, Serialize)]
 #[serde(default, deny_unknown_fields)]
-pub struct NetworkRequester {}
+pub struct NetworkRequester {
+    pub disable_poisson_rate: bool,
+}
+
+impl Default for NetworkRequester {
+    fn default() -> Self {
+        NetworkRequester {
+            disable_poisson_rate: true,
+        }
+    }
+}

 #[derive(Debug, Clone, Copy, Deserialize, PartialEq, Serialize)]
 #[serde(default, deny_unknown_fields)]
@@ -79,7 +79,7 @@ pub struct NetworkRequesterV1_1_20_2 {}

 impl From<NetworkRequesterV1_1_20_2> for NetworkRequester {
    fn from(_value: NetworkRequesterV1_1_20_2) -> Self {
-        NetworkRequester {}
+        NetworkRequester::default()
    }
 }

@@ -37,7 +37,7 @@ nym_api_urls = [
    {{/each}}
 ]

-[storage_paths] 
+[storage_paths]

 # Path to file containing private identity key.
 keys.private_identity_key_file = '{{ storage_paths.keys.private_identity_key_file }}'
@@ -76,6 +76,14 @@ allowed_list_location = '{{ storage_paths.allowed_list_location }}'
 unknown_list_location = '{{ storage_paths.unknown_list_location }}'


+[network_requester]
+
+# Disable Poisson sending rate, and only send cover traffic occasionally as keepalive messages.
+# This is equivalent to setting debug.traffic.disable_main_poisson_packet_distribution = true,
+# and debug.cover_traffic.loop_cover_traffic_average_delay = 5s.
+disable_poisson_rate = {{ network_requester.disable_poisson_rate }}
+
+
 ##### logging configuration options #####

 [logging]
Author	SHA1	Message	Date
mfahampshire	9a22f3dbba	started on rendezvous example	2023-08-23 21:52:52 +02:00
mfahampshire	faf558053a	working on rendzv example	2023-08-23 20:47:24 +02:00
Jon Häggblad	92350daca8	network-requester: disable poisson process by default (#3783 ) * network-requester: disable poisson process by default * network-requester: instead add new top-level config field * Remove quoation marks in template	2023-08-22 14:17:03 +02:00
Jędrzej Stuczyński	b429c64168	[demo] libp2p example with nym-sdk (#3763 ) * imported libp2p to monorepo * fixed vanilla ping example * added libp2p client to workspace * naively replaced dockerised client with the sdk * moved libp2p code to sdk examples * reduced number of dependencies required for libp2p example * updated readmes * added protobuf compiler to build dependencies * added protoc dependency to readme for chat and ping examples --------- Co-authored-by: mfahampshire <maxhampshire@pm.me>	2023-08-22 13:12:13 +02:00
Tommy Verrall	f88622ac08	Delete upload-nyxd-ci.yml not needed anymore removing	2023-08-22 12:21:44 +02:00
Jędrzej Stuczyński	0a5a2c6747	removed needless_pass_by_ref_mut in mixnet contract tests (#3798 )	2023-08-22 11:56:23 +02:00
Tommy Verrall	824bd636f9	Merge pull request #3795 from nymtech/release/v1.1.28 Release/v1.1.28	2023-08-22 11:17:29 +02:00
Tommy Verrall	cf6411ac08	Merge pull request #3794 from nymtech/qa/fix-windows-builder Update nym-connect-publish-windows10.yml	2023-08-22 10:59:23 +02:00
Tommy Verrall	6428133122	Merge pull request #3793 from nymtech/qa/fix-api-tests change ci to run against sandbox for tests	2023-08-22 10:55:24 +02:00
Tommy Verrall	3c69f9c2f9	Update nym-connect-publish-windows10.yml	2023-08-22 10:35:46 +02:00
Tommy Verrall	c1e4b87744	change ci to run against sandbox for tests	2023-08-22 10:31:37 +02:00
Pierre Dommerc	dffe171b7f	feat(explorer-api): add gateway geoip location (#3785 ) * feat(explorer-api): add gateway geoip location * fix lint	2023-08-21 15:50:18 +02:00
mfahampshire	6d3b198f00	change minibolt description	2023-08-21 14:04:21 +02:00
mfahampshire	d4920b82f0	include 2faktor's PR material	2023-08-21 14:04:08 +02:00
mfahampshire	8dfe8f4678	version bump in toml files fr docs projects	2023-08-21 09:54:09 +02:00