Merge branch 'release/0.2.0'

CHANGELOG.md

@@ -0,0 +1,12 @@
+# nym-client Changelog
+
+* removed the `--local` flag
+* introduced `--directory` argument to support arbitrary directory servers. Leaving it out will point the node at the "https://directory.nymtech.net" alpha testnet server
+* IPv6 support
+* client version number is now shown at node start
+* directory server location is now shown at node start
+* decrease default delays
+
+## 0.1.0 - Initial Release
+
+* The bare minimum set of features required by a Nym Client

Cargo.toml

@@ -1,10 +1,39 @@
 [package]
+build = "build.rs"
 name = "nym-client"
-version = "0.1.0"
-authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
+version = "0.2.0"
+authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 edition = "2018"
 
+[lib]
+name = "nym_client"
+path = "src/lib.rs"
+
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-sphinx = { path = "../sphinx" }
+base64 = "0.11.0"
+clap = "2.33.0"
+curve25519-dalek = "1.2.3"
+dirs = "2.0.2"
+futures = "0.3.1"
+hex = "0.4.0"
+pem = "0.7.0"
+rand = "0.7.2"
+rand_distr = "0.2.2"
+reqwest = "0.9.22"
+serde = { version = "1.0.104", features = ["derive"] }
+serde_json = "1.0.44"
+sphinx = { path = "../sphinx" }
+sfw-provider-requests = { path = "../nym-sfw-provider/sfw-provider-requests" }
+tokio = { version = "0.2", features = ["full"] }
+tungstenite = "0.9.2"
+
+# putting this explicitly below everything and most likely, the next time we look into it, it will already have a proper release
+tokio-tungstenite = { git = "https://github.com/dbcfd/tokio-tungstenite", rev="6dc2018cbfe8fe7ddd75ff977343086503135b38" }
+
+[build-dependencies]
+built = "0.3.2"
+
+[dev-dependencies]
+mockito = "0.22.0"

build.rs

@@ -0,0 +1,5 @@
+use built;
+
+fn main() {
+    built::write_built_file().expect("Failed to acquire build-time information");
+}

src/clients/directory/metrics.rs

@@ -0,0 +1,19 @@
+use serde::{Deserialize, Serialize};
+use std::collections::HashMap;
+
+#[derive(Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct PersistedMixMetric {
+    pub pub_key: String,
+    pub received: u64,
+    pub sent: HashMap<String, u64>,
+    pub timestamp: u64,
+}
+
+#[derive(Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MixMetric {
+    pub pub_key: String,
+    pub received: u64,
+    pub sent: HashMap<String, u64>,
+}

src/clients/directory/mod.rs

@@ -0,0 +1,68 @@
+use crate::clients::directory::requests::health_check_get::{
+    HealthCheckRequester, Request as HealthCheckRequest,
+};
+use crate::clients::directory::requests::metrics_mixes_get::{
+    MetricsMixRequester, Request as MetricsMixRequest,
+};
+use crate::clients::directory::requests::metrics_mixes_post::{
+    MetricsMixPoster, Request as MetricsMixPost,
+};
+use crate::clients::directory::requests::presence_coconodes_post::{
+    PresenceCocoNodesPoster, Request as PresenceCocoNodesPost,
+};
+use crate::clients::directory::requests::presence_mixnodes_post::{
+    PresenceMixNodesPoster, Request as PresenceMixNodesPost,
+};
+use crate::clients::directory::requests::presence_providers_post::{
+    PresenceMixProviderPoster, Request as PresenceProvidersPost,
+};
+use crate::clients::directory::requests::presence_topology_get::{
+    PresenceTopologyGetRequester, Request as PresenceTopologyRequest,
+};
+
+pub mod metrics;
+pub mod presence;
+pub mod requests;
+
+pub struct Config {
+    pub base_url: String,
+}
+
+pub trait DirectoryClient {
+    fn new(config: Config) -> Self;
+}
+
+pub struct Client {
+    pub health_check: HealthCheckRequest,
+    pub metrics_mixes: MetricsMixRequest,
+    pub metrics_post: MetricsMixPost,
+    pub presence_coconodes_post: PresenceCocoNodesPost,
+    pub presence_mix_nodes_post: PresenceMixNodesPost,
+    pub presence_providers_post: PresenceProvidersPost,
+    pub presence_topology: PresenceTopologyRequest,
+}
+
+impl DirectoryClient for Client {
+    fn new(config: Config) -> Client {
+        let health_check: HealthCheckRequest = HealthCheckRequest::new(config.base_url.clone());
+        let metrics_mixes: MetricsMixRequest = MetricsMixRequest::new(config.base_url.clone());
+        let metrics_post: MetricsMixPost = MetricsMixPost::new(config.base_url.clone());
+        let presence_topology: PresenceTopologyRequest =
+            PresenceTopologyRequest::new(config.base_url.clone());
+        let presence_coconodes_post: PresenceCocoNodesPost =
+            PresenceCocoNodesPost::new(config.base_url.clone());
+        let presence_mix_nodes_post: PresenceMixNodesPost =
+            PresenceMixNodesPost::new(config.base_url.clone());
+        let presence_providers_post: PresenceProvidersPost =
+            PresenceProvidersPost::new(config.base_url.clone());
+        Client {
+            health_check,
+            metrics_mixes,
+            metrics_post,
+            presence_coconodes_post,
+            presence_mix_nodes_post,
+            presence_providers_post,
+            presence_topology,
+        }
+    }
+}

src/clients/directory/presence.rs

@@ -0,0 +1,41 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct CocoPresence {
+    pub host: String,
+    pub pub_key: String,
+    pub last_seen: i64,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MixNodePresence {
+    pub host: String,
+    pub pub_key: String,
+    pub layer: u64,
+    pub last_seen: u64,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MixProviderPresence {
+    pub host: String,
+    pub pub_key: String,
+    pub registered_clients: Vec<MixProviderClient>,
+}
+
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct MixProviderClient {
+    pub pub_key: String,
+}
+
+// Topology shows us the current state of the overall Nym network
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Topology {
+    pub coco_nodes: Vec<CocoPresence>,
+    pub mix_nodes: Vec<MixNodePresence>,
+    pub mix_provider_nodes: Vec<MixProviderPresence>,
+}

src/clients/directory/requests/health_check_get.rs

@@ -0,0 +1,60 @@
+use reqwest::Response;
+
+pub struct Request {
+    base_url: String,
+    path: String,
+}
+
+pub trait HealthCheckRequester {
+    fn new(base_url: String) -> Self;
+    fn get(&self) -> Result<Response, reqwest::Error>;
+}
+
+impl HealthCheckRequester for Request {
+    fn new(base_url: String) -> Self {
+        Request {
+            base_url,
+            path: "/api/healthcheck".to_string(),
+        }
+    }
+
+    fn get(&self) -> Result<Response, reqwest::Error> {
+        let url = format!("{}{}", self.base_url, self.path);
+        reqwest::get(&url)
+    }
+}
+
+#[cfg(test)]
+mod healthcheck_requests {
+    use super::*;
+
+    #[cfg(test)]
+    use mockito::mock;
+
+    #[cfg(test)]
+    mod on_a_400_status {
+        use super::*;
+
+        #[test]
+        #[should_panic]
+        fn it_returns_an_error() {
+            let _m = mock("GET", "/api/healthcheck").with_status(400).create();
+            let req = Request::new(mockito::server_url());
+            assert!(req.get().is_err());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod on_a_200 {
+        use super::*;
+
+        #[test]
+        fn it_returns_a_response_with_200_status() {
+            let _m = mock("GET", "/api/healthcheck").with_status(200).create();
+            let req = Request::new(mockito::server_url());
+            assert!(req.get().is_ok());
+            _m.assert();
+        }
+    }
+}

src/clients/directory/requests/metrics_mixes_get.rs

@@ -0,0 +1,239 @@
+use crate::clients::directory::metrics::PersistedMixMetric;
+
+pub struct Request {
+    base_url: String,
+    path: String,
+}
+
+pub trait MetricsMixRequester {
+    fn new(base_url: String) -> Self;
+    fn get(&self) -> Result<Vec<PersistedMixMetric>, reqwest::Error>;
+}
+
+impl MetricsMixRequester for Request {
+    fn new(base_url: String) -> Self {
+        Request {
+            base_url,
+            path: "/api/metrics/mixes".to_string(),
+        }
+    }
+
+    fn get(&self) -> Result<Vec<PersistedMixMetric>, reqwest::Error> {
+        let url = format!("{}{}", self.base_url, self.path);
+        let mix_metric_vec = reqwest::get(&url)?.json()?;
+        Ok(mix_metric_vec)
+    }
+}
+
+#[cfg(test)]
+mod metrics_get_request {
+    use super::*;
+
+    #[cfg(test)]
+    use mockito::mock;
+
+    #[cfg(test)]
+    mod on_a_400_status {
+        use super::*;
+
+        #[test]
+        #[should_panic]
+        fn it_returns_an_error() {
+            let _m = mock("GET", "/api/metrics/mixes").with_status(400).create();
+            let req = Request::new(mockito::server_url());
+            req.get().unwrap();
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod on_a_200 {
+        use super::*;
+        #[test]
+        fn it_returns_a_response_with_200_status_and_a_correct_topology() {
+            let json = fixtures::mix_metrics_response_json();
+            let _m = mock("GET", "/api/metrics/mixes")
+                .with_status(200)
+                .with_body(json)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let result = req.get();
+            assert_eq!(true, result.is_ok());
+            assert_eq!(
+                1576061080635800000,
+                result.unwrap().first().unwrap().timestamp
+            );
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod fixtures {
+        #[cfg(test)]
+        pub fn mix_metrics_response_json() -> String {
+            r#"[
+              {
+                "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=",
+                "sent": {
+                  "35.178.213.77:1789": 1,
+                  "52.56.99.196:1789": 2
+                },
+                "received": 10,
+                "timestamp": 1576061080635800000
+              },
+              {
+                "pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=",
+                "sent": {
+                  "3.8.176.11:1789": 2,
+                  "35.178.212.193:1789": 2
+                },
+                "received": 4,
+                "timestamp": 1576061080806225700
+              },
+              {
+                "pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=",
+                "sent": {
+                  "3.9.12.238:1789": 3
+                },
+                "received": 3,
+                "timestamp": 1576061080894667300
+              },
+              {
+                "pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=",
+                "sent": {
+                  "3.9.12.238:1789": 5,
+                  "35.176.155.107:1789": 3
+                },
+                "received": 8,
+                "timestamp": 1576061081254846500
+              },
+              {
+                "pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=",
+                "sent": {
+                  "35.178.213.77:1789": 4,
+                  "52.56.99.196:1789": 6
+                },
+                "received": 19,
+                "timestamp": 1576061081371549000
+              },
+              {
+                "pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=",
+                "sent": {
+                  "3.8.176.11:1789": 4,
+                  "35.178.212.193:1789": 2
+                },
+                "received": 6,
+                "timestamp": 1576061081498404900
+              },
+              {
+                "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=",
+                "sent": {
+                  "35.178.213.77:1789": 2,
+                  "52.56.99.196:1789": 3
+                },
+                "received": 6,
+                "timestamp": 1576061081637625000
+              },
+              {
+                "pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=",
+                "sent": {
+                  "3.8.176.11:1789": 5,
+                  "35.178.212.193:1789": 4
+                },
+                "received": 9,
+                "timestamp": 1576061081805484800
+              },
+              {
+                "pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=",
+                "sent": {
+                  "3.9.12.238:1789": 4,
+                  "35.176.155.107:1789": 4
+                },
+                "received": 8,
+                "timestamp": 1576061081896562400
+              },
+              {
+                "pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=",
+                "sent": {
+                  "3.9.12.238:1789": 2,
+                  "35.176.155.107:1789": 4
+                },
+                "received": 6,
+                "timestamp": 1576061079255938600
+              },
+              {
+                "pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=",
+                "sent": {
+                  "35.178.213.77:1789": 6
+                },
+                "received": 10,
+                "timestamp": 1576061079370829300
+              },
+              {
+                "pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=",
+                "sent": {
+                  "3.8.176.11:1789": 2,
+                  "35.178.212.193:1789": 5
+                },
+                "received": 7,
+                "timestamp": 1576061079497993200
+              },
+              {
+                "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=",
+                "sent": {
+                  "35.178.213.77:1789": 5,
+                  "52.56.99.196:1789": 2
+                },
+                "received": 13,
+                "timestamp": 1576061079637208600
+              },
+              {
+                "pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=",
+                "sent": {
+                  "3.8.176.11:1789": 5,
+                  "35.178.212.193:1789": 4
+                },
+                "received": 9,
+                "timestamp": 1576061079806557200
+              },
+              {
+                "pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=",
+                "sent": {
+                  "3.9.12.238:1789": 2,
+                  "35.176.155.107:1789": 7
+                },
+                "received": 9,
+                "timestamp": 1576061079895988000
+              },
+              {
+                "pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=",
+                "sent": {
+                  "3.9.12.238:1789": 3,
+                  "35.176.155.107:1789": 2
+                },
+                "received": 5,
+                "timestamp": 1576061080255701500
+              },
+              {
+                "pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=",
+                "sent": {
+                  "35.178.213.77:1789": 3,
+                  "52.56.99.196:1789": 3
+                },
+                "received": 7,
+                "timestamp": 1576061080370956300
+              },
+              {
+                "pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=",
+                "sent": {
+                  "3.8.176.11:1789": 5,
+                  "35.178.212.193:1789": 1
+                },
+                "received": 6,
+                "timestamp": 1576061080501732900
+              }
+            ]"#
+            .to_string()
+        }
+    }
+}

src/clients/directory/requests/metrics_mixes_post.rs

@@ -0,0 +1,98 @@
+use crate::clients::directory::metrics::MixMetric;
+use reqwest::Response;
+
+pub struct Request {
+    base_url: String,
+    path: String,
+}
+
+pub trait MetricsMixPoster {
+    fn new(base_url: String) -> Self;
+    fn post(&self, metric: &MixMetric) -> Result<Response, reqwest::Error>;
+}
+
+impl MetricsMixPoster for Request {
+    fn new(base_url: String) -> Self {
+        Request {
+            base_url,
+            path: "/api/metrics/mixes".to_string(),
+        }
+    }
+
+    fn post(&self, metric: &MixMetric) -> Result<Response, reqwest::Error> {
+        let url = format!("{}{}", self.base_url, self.path);
+        let client = reqwest::Client::new();
+        let mix_metric_vec = client.post(&url).json(&metric).send()?;
+        Ok(mix_metric_vec)
+    }
+}
+
+#[cfg(test)]
+mod metrics_get_request {
+    use super::*;
+
+    #[cfg(test)]
+    use mockito::mock;
+
+    #[cfg(test)]
+    mod on_a_400_status {
+        use super::*;
+
+        #[test]
+        fn it_returns_an_error() {
+            let _m = mock("POST", "/api/metrics/mixes").with_status(400).create();
+            let req = Request::new(mockito::server_url());
+            let metric = fixtures::new_metric();
+            let result = req.post(&metric);
+            assert_eq!(400, result.unwrap().status());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod on_a_200 {
+        use super::*;
+        #[test]
+        fn it_returns_a_response_with_200() {
+            let json = fixtures::mix_metrics_response_json();
+            let _m = mock("POST", "/api/metrics/mixes")
+                .with_status(201)
+                .with_body(json)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let metric = fixtures::new_metric();
+            let result = req.post(&metric);
+            assert_eq!(true, result.is_ok());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod fixtures {
+        use crate::clients::directory::metrics::MixMetric;
+
+        pub fn new_metric() -> MixMetric {
+            MixMetric {
+                pub_key: "abc".to_string(),
+                received: 666,
+                sent: Default::default(),
+            }
+        }
+
+        #[cfg(test)]
+        pub fn mix_metrics_response_json() -> String {
+            r#"
+              {
+                "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=",
+                "sent": {
+                  "35.178.213.77:1789": 1,
+                  "52.56.99.196:1789": 2
+                },
+                "received": 10,
+                "timestamp": 1576061080635800000
+              }
+            "#
+            .to_string()
+        }
+    }
+}

src/clients/directory/requests/mod.rs

@@ -0,0 +1,7 @@
+pub mod health_check_get;
+pub mod metrics_mixes_get;
+pub mod metrics_mixes_post;
+pub mod presence_coconodes_post;
+pub mod presence_mixnodes_post;
+pub mod presence_providers_post;
+pub mod presence_topology_get;

src/clients/directory/requests/presence_coconodes_post.rs

@@ -0,0 +1,86 @@
+use crate::clients::directory::presence::CocoPresence;
+use reqwest::Response;
+
+pub struct Request {
+    base_url: String,
+    path: String,
+}
+
+pub trait PresenceCocoNodesPoster {
+    fn new(base_url: String) -> Self;
+    fn post(&self, presence: &CocoPresence) -> Result<Response, reqwest::Error>;
+}
+
+impl PresenceCocoNodesPoster for Request {
+    fn new(base_url: String) -> Self {
+        Request {
+            base_url,
+            path: "/api/presence/coconodes".to_string(),
+        }
+    }
+
+    fn post(&self, presence: &CocoPresence) -> Result<Response, reqwest::Error> {
+        let url = format!("{}{}", self.base_url, self.path);
+        let client = reqwest::Client::new();
+        let p = client.post(&url).json(&presence).send()?;
+        Ok(p)
+    }
+}
+
+#[cfg(test)]
+mod metrics_get_request {
+    use super::*;
+
+    #[cfg(test)]
+    use mockito::mock;
+
+    #[cfg(test)]
+    mod on_a_400_status {
+        use super::*;
+
+        #[test]
+        fn it_returns_an_error() {
+            let _m = mock("POST", "/api/presence/coconodes")
+                .with_status(400)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let presence = fixtures::new_presence();
+            let result = req.post(&presence);
+            assert_eq!(400, result.unwrap().status());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod on_a_200 {
+        use super::*;
+        #[test]
+        fn it_returns_a_response_with_201() {
+            let json = r#"{
+                          "ok": true
+                      }"#;
+            let _m = mock("POST", "/api/presence/coconodes")
+                .with_status(201)
+                .with_body(json)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let presence = fixtures::new_presence();
+            let result = req.post(&presence);
+            assert_eq!(true, result.is_ok());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod fixtures {
+        use crate::clients::directory::presence::CocoPresence;
+
+        pub fn new_presence() -> CocoPresence {
+            CocoPresence {
+                host: "foo.com".to_string(),
+                pub_key: "abc".to_string(),
+                last_seen: 666,
+            }
+        }
+    }
+}

src/clients/directory/requests/presence_mixnodes_post.rs

@@ -0,0 +1,87 @@
+use crate::clients::directory::presence::MixNodePresence;
+use reqwest::Response;
+
+pub struct Request {
+    base_url: String,
+    path: String,
+}
+
+pub trait PresenceMixNodesPoster {
+    fn new(base_url: String) -> Self;
+    fn post(&self, presence: &MixNodePresence) -> Result<Response, reqwest::Error>;
+}
+
+impl PresenceMixNodesPoster for Request {
+    fn new(base_url: String) -> Self {
+        Request {
+            base_url,
+            path: "/api/presence/mixnodes".to_string(),
+        }
+    }
+
+    fn post(&self, presence: &MixNodePresence) -> Result<Response, reqwest::Error> {
+        let url = format!("{}{}", self.base_url, self.path);
+        let client = reqwest::Client::new();
+        let p = client.post(&url).json(&presence).send()?;
+        Ok(p)
+    }
+}
+
+#[cfg(test)]
+mod metrics_get_request {
+    use super::*;
+
+    #[cfg(test)]
+    use mockito::mock;
+
+    #[cfg(test)]
+    mod on_a_400_status {
+        use super::*;
+
+        #[test]
+        fn it_returns_an_error() {
+            let _m = mock("POST", "/api/presence/mixnodes")
+                .with_status(400)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let presence = fixtures::new_presence();
+            let result = req.post(&presence);
+            assert_eq!(400, result.unwrap().status());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod on_a_200 {
+        use super::*;
+        #[test]
+        fn it_returns_a_response_with_201() {
+            let json = r#"{
+                          "ok": true
+                      }"#;
+            let _m = mock("POST", "/api/presence/mixnodes")
+                .with_status(201)
+                .with_body(json)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let presence = fixtures::new_presence();
+            let result = req.post(&presence);
+            assert_eq!(true, result.is_ok());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod fixtures {
+        use crate::clients::directory::presence::MixNodePresence;
+
+        pub fn new_presence() -> MixNodePresence {
+            MixNodePresence {
+                host: "foo.com".to_string(),
+                pub_key: "abc".to_string(),
+                layer: 1,
+                last_seen: 0,
+            }
+        }
+    }
+}

src/clients/directory/requests/presence_providers_post.rs

@@ -0,0 +1,85 @@
+use crate::clients::directory::presence::MixProviderPresence;
+use reqwest::Response;
+
+pub struct Request {
+    base_url: String,
+    path: String,
+}
+
+pub trait PresenceMixProviderPoster {
+    fn new(base_url: String) -> Self;
+    fn post(&self, presence: &MixProviderPresence) -> Result<Response, reqwest::Error>;
+}
+
+impl PresenceMixProviderPoster for Request {
+    fn new(base_url: String) -> Self {
+        Request {
+            base_url,
+            path: "/api/presence/mixproviders".to_string(),
+        }
+    }
+
+    fn post(&self, presence: &MixProviderPresence) -> Result<Response, reqwest::Error> {
+        let url = format!("{}{}", self.base_url, self.path);
+        let client = reqwest::Client::new();
+        let p = client.post(&url).json(&presence).send()?;
+        Ok(p)
+    }
+}
+
+#[cfg(test)]
+mod metrics_get_request {
+    use super::*;
+
+    #[cfg(test)]
+    use mockito::mock;
+
+    #[cfg(test)]
+    mod on_a_400_status {
+        use super::*;
+
+        #[test]
+        fn it_returns_an_error() {
+            let _m = mock("POST", "/api/presence/mixproviders")
+                .with_status(400)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let presence = fixtures::new_presence();
+            let result = req.post(&presence);
+            assert_eq!(400, result.unwrap().status());
+            _m.assert();
+        }
+    }
+
+    #[cfg(test)]
+    mod on_a_200 {
+        use super::*;
+        #[test]
+        fn it_returns_a_response_with_201() {
+            let json = r#"{
+                          "ok": true
+                      }"#;
+            let _m = mock("POST", "/api/presence/mixproviders")
+                .with_status(201)
+                .with_body(json)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let presence = fixtures::new_presence();
+            let result = req.post(&presence);
+            assert_eq!(true, result.is_ok());
+            _m.assert();
+        }
+    }
+    #[cfg(test)]
+    mod fixtures {
+        use crate::clients::directory::presence::MixProviderPresence;
+
+        pub fn new_presence() -> MixProviderPresence {
+            MixProviderPresence {
+                host: "foo.com".to_string(),
+                pub_key: "abc".to_string(),
+                registered_clients: vec![],
+            }
+        }
+    }
+}

src/clients/directory/requests/presence_topology_get.rs

@@ -0,0 +1,306 @@
+use crate::clients::directory::presence::Topology;
+
+pub struct Request {
+    base_url: String,
+    path: String,
+}
+
+pub trait PresenceTopologyGetRequester {
+    fn new(base_url: String) -> Self;
+    fn get(&self) -> Result<Topology, reqwest::Error>;
+}
+
+impl PresenceTopologyGetRequester for Request {
+    fn new(base_url: String) -> Self {
+        Request {
+            base_url,
+            path: "/api/presence/topology".to_string(),
+        }
+    }
+
+    fn get(&self) -> Result<Topology, reqwest::Error> {
+        let url = format!("{}{}", self.base_url, self.path);
+        let topology: Topology = reqwest::get(&url)?.json()?;
+        Ok(topology)
+    }
+}
+
+#[cfg(test)]
+mod topology_requests {
+    use super::*;
+    #[cfg(test)]
+    use mockito::mock;
+    #[cfg(test)]
+    mod on_a_400_status {
+        use super::*;
+        #[test]
+        #[should_panic]
+        fn it_panics() {
+            let _m = mock("GET", "/api/presence/topology")
+                .with_status(400)
+                .with_body("bad body")
+                .create();
+            let req = Request::new(mockito::server_url());
+            req.get().unwrap();
+            _m.assert();
+        }
+    }
+    #[cfg(test)]
+    mod on_a_200 {
+        use super::*;
+        #[test]
+        fn it_returns_a_response_with_200_status_and_a_correct_topology() {
+            let json = fixtures::topology_response_json();
+            let _m = mock("GET", "/api/presence/topology")
+                .with_status(200)
+                .with_body(json)
+                .create();
+            let req = Request::new(mockito::server_url());
+            let result = req.get();
+            assert_eq!(true, result.is_ok());
+            assert_eq!(
+                1575915097085539300,
+                result.unwrap().coco_nodes.first().unwrap().last_seen
+            );
+            _m.assert();
+        }
+    }
+    #[cfg(test)]
+    pub mod fixtures {
+        #[cfg(test)]
+        pub fn topology_response_json() -> String {
+            r#"{
+                   "cocoNodes": [
+                     {
+                       "host": "3.8.244.109:4000",
+                       "pubKey": "AAAAAAAAAAEKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAEFuhRpy4HfcuxwcRI9sIWMo_LVmbk19g1gfMRlBrmZqoEQL6rDApVLZ9eMp-5IQK8WLlZpWf4Zjy7kZolARAyp_rHUQkH4PrDjgoPrKbm6qK_iejYpL7qx28Q3VeInMpwMIMaSbbW9y36sEVtGc2I0Iu5vS0sp8ESiVlQ5NaBz72deZ8oKJJ4IEPPHP99-b0UQX80fVIrNM88mMzKy0bHri9NFlmIG-e0G1cqmw_ry3XWGQkcr1M5RuNa6oX50w5QawAEVxd5FP5bE8bS4x54Csof11sQWUTwMp6Q7_3H7ZCTSlKSqujlOhmfqSHfGPO2sDIYPHDhDzjakZpKAZWWhn_hiR6DfPpomQ01ZYUhVKKSMxz7_VPjsQplP0bZXA2gfnkADUN8UQ0N9g_usIw73r4aZsOviMsRM8oByvsjVfUWc4_HTLSdnQyImFkHz9CiCmrIYL2dYQRePRatWggvBAyeRzntxI4jDqLKiBdi54ZlAKgV6MCRaJ7Bu7BtmLXrtK4sawAED3QYxuvOSZrbZdUr4yG-U9yVvJ9Klkf-5Mo4EYp3qTL2KBB6_LrZepjAQqp486YkZ03mTIezcsZ48EboXVTWKBZ3QnTI5tX-j4gGxQb7klOJc97qJkDxsvpz4F0ChgCUIZhpIItWHia7_R3Gi-b5siLIdQdUho9isn3kiDGm6t0NED2Bgy3ZxxQwzqsBZm4kPr2_fPX4YyvIoP9895YcGjZyE5iiRC_TE41RJmB1GZYdxegTMq3lNDllKgiqaiPgawAEJASDkmZHTwlg9YOev5OWpQD-FnhPkqVNo_QcDyRu9eoGcWSGFp2sYqjG2SpmiXq0VNnAO7AcKxRzDFu7TjfhlU3Kt0uTKIcrWVU1zFNbJNMjYEq90pp50nowwx8INz20IXET2ZNX6kIXYFCsEvPLZFlG2OoL6xg3uQS1qMl3lIS_VxdO_JfVe0rT65WsJ_P4Nkc1jYiuNPHY6d_iFO0BVYqX0sOCX73GC_TT13BR0jnPwDAVw0rGtYHsXBb8TKOsawAEZIClauuT1V3qOZnb7uRZhFXO-PKTxgc1LCzJt2ChOrMZaBpjlkf3IPpJ2UF4JH4kGaDeBf2k_S-FLAs3drK21efbi5P6_a4QTxAiiRimXGoQIyvOg462s6kP_ZRFufo8YYQHS4olaOeqU4564dNskg_uBPsFMz_2GNOhmn_15cJqP1jfkyD49Z16GTS5YLHgVl9bJKqyvLuypsToLbt1BJzipEP0L2OohuRm-_MvqvwwWKyjNQsubgee1K728d9AawAEBkGggcNVCtXyhoSqi3_w0tVxtkAYeud8sBeAtZHGs06me_QL8co0MFLlO-zdkUb4ZBq08rFEbgLOma8_3whleM8NIPaHNISp1q3IsIhB5zdXcZoGsqLixODBFHtID3YEHAlr4f9T_yh11yJ95xGCl_6Y37hpwLQVGyrfSfccM24mVFqnV3TT5Wdq3ile-jesUx1Q2G1yK_xVqc6itmk-kDuBjyZgzYi1-jsIXAjnhM9G7t8J_Bv5yGGZhLK2dCzM=",
+                       "type": "validator",
+                       "lastSeen": 1575915097085539300
+                     },
+                     {
+                       "host": "3.9.129.61:4000",
+                       "pubKey": "AAAAAAAAAAMKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAEEv6RMevAQmLGkeK0uJKnMPPAtm8GgXjWSQijYdnxlPh5SJSNeJUbPZKWFFWdk8yIFXKa8jnzETtdGFKgUUt5AVUDpTBmEdwaHCzlFhXrttshy0V5OhPUlV8cGABmxbagMYm0bFPg0r-snSkrB9YG6wqJYQVeIMOCGYCPbHmDA8R_0-h8VkRKWs1d9KvQOK4kShqgZtYN71KJW8uDE4q2jsGDVvxFt1AgmU9b93xsXF17KrpZy5WxlLZ73HtnTD_oawAED4vd_rK-Kx_n8x_OdDiiEOPUlYDlDCQUqenU9XHKH3B6ijfkJ368wd3LDDVStjDwNORrAyUSw_VlSNUpd1XLC8d17gTaIq5ZI2fWuwwZaoN1JCsYU8fQ6USgtIehQX7IPP8EkFuNmuCBCmpr4schtYniGe9J8Q4dsV-TYPr2uLJkdx1r7luzF--I22k7NfQQM14QDci_0kgrgmZ54CJGkjXyOhCppBXg3fqLC6aFvT3ZocfiiXBJt0huGgPMDtYsawAECLh8KUdNsDolERwJ8v04bS5jI_KKf7uUnCHWuCELwbJSUI3OK1ufS1qSpauvSzVQSbrhEzrEfwQn4VtxQxJlX4UdDU-R-hafiZvVC6DLLAbuORBAC3FScn9W58CnezH4DvCp_w7nftDfdxeuungbZT9XaxS3iNC6PnFsWF6WM3DxMwrzOrFe6wEEoTSPe1mcUDrtwM5UksIvJr6MBRAXrdl0IdBTQr7cLwKe_KYi4siwdjfJEJtOh7oxQBxBg2UkawAEJAPZK2Gg2MQwpxdDT24lNQHF7FVfkO_LuhJwn0RbwNDSVeA4P6-tWL5TkCpqr8xYHfwQ6Z3ILfpGCZr8PspwIoRzqZHQ16f8Pq9xnr0hLEI9BOQU0FS2EtuyPgju5iwsAJAfehUzu6kNLphuLGsXoIZdXDG5mbylwh9JzAVXTwgaR0hNqyXVJxgbt7jcYaSEBFcMGV-hjXyVVNzBleE-G9o_noI_KWU4Ce7K-qOMcewMKfy_VEw-gVaD6dHz6AMoawAEE9XuOLwRttvKybAssZ9gsK-_YRUwuFOeRDIr3NX___9bx6pCc18adCIlH_8EJWFwXZ05ZpNNE88mYx7ZQ3aqaArZJRoWeZeKhqH_s05V10xbzkYX71G5cqz--8vr9ZlQRb2BeETF_Tdq_PLk7qbT8WTGIoq7ZwyDRQTgzvkCgyzj_hBLh2o7sSVNgUo38SFUTMn7YtvVFYlSrTDE3WKE-T-nh5SWdDBxgDTc3Bw8JpzNH-WkoJ4Lim7sB4Op1gEUawAEW4-kenlffwsNr_3b3aV0YuusLpxB03sxPzQ5B0CWNiVtbja1Z4tWhKGUUrdq_eUgMV0y5Of-BqNi5FspAQnhJBFSSxtOzRGV1h3qyUTksfZyed9z8zPI-ZPP9XXm7hYgJgDz_kxte-NfS9UG9q5AZetHUN4kGxXutjjzfUQZ9yTvhBKgKgTI2Dp_R_jZrWQ8F1BoWzIJzjddT1K2MvCQEkARYw08isbOeFmCwgVUcjxYZO45WyOmLQA7QJRL9WvA=",
+                       "type": "validator",
+                       "lastSeen": 1575915097388409000
+                     },
+                     {
+                       "host": "3.9.222.1:4000",
+                       "pubKey": "AAAAAAAAAAQKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAECh9xcxpjOp1r7kiNIgrI9GgAlvXwgHkTchOxUiyOzTq6FDWdGN64KiC3NDeyGTg8FmzvGzS3jREeJqOdr4G9ZGtWkauAITgLFiH62t-YntRslhr8_1shxlmzKiNKJN_QFflEq79pZIlWtp3N8LIHMvXRtl-zt2DMze4s02XDmEkviyVE4CkQUDtCc-2MfPT4JcmEFqtFIxjrXn18SbYg3c6XUQHsGIkuDrKuCTRlpC8kvmM0uVoIeWdmwDlZk4jUawAEJhRwK5ozjqIWRP1bFzBPS9VhaJnfKU9PeFYtN5beiAHrYr2ylIB3yDfmAQUdKDowDUm5nfJATejEjEnrTGxh70QtfoNV391rSns3F71tBwY62KLaNr8qnVfeSFHV3FcQTMHHF_8mDb5_11Rj6aiMvW0y6eetHo7CDPMdEyDPmok_U2ZM5BzOUnwjT21HtnvcKxKKwHJ_QGfnAHPyDIhNOMgxJCrVazOidLCHeYGpyCLw1ipeTyKOQX0_ByB8dH6AawAEGV1GuF5SSlT67B1ityPJK2ZwXjeeKB4gGdCG3qRtWxLTZfGhVm7YAYm2f5tw_wrsJAZ9FubVhateGg0ZN67NxZtsvOOejXz6743f7ijnQopPgd_8pH-iVf6BEcSO8ZdcHxNRUTayzjVLs99bwMo2zaPevW4X4G_bN4mh---aPkdGYHwaiklzUhqJ-eqycrYAFyjyEXaPBXLQm1rpczqluNvnKbd8Q9LZWukgm7_uWv_HxufIvdWgoq8bAt78UU3oawAEP9VDehhqrQG5-WHMB66XVxo1TgMM8aVV0SwAq3lCRkpiFBz_9kw8T1F9Hx2AiNrEGT1QLbdMkpms1cG_5gBBahQofdt_NmUs1jfTFXY9iyMy1Q7A6ZYaLP8Z6q-orc1cKqySY-BJZQ_CpGFfXS0OVniFDQ6v78ytPK7K-yRgT1PxFgm3rZqrG0Tjbrpsg2PUL5S5fuXfMhUosP0uoLj0D1guWAR9Y7kfFBIXaTSFMoa8fghVBUTRNhK9f72a8SxQawAEOiv71taLjKqaaWQ_QjcDhWbvjG1EnsCyI0toNjGkcF19x4Vk-5NC96_4ioUGz404IC0XN03roRnibRT_78D9vZFVCWCqve9EjdF5TcApx03zIP4JT2g2q0MKIGgGrwt4Pz6LO6yOfMm7B8Yraps8IV-nP1w7K1m9XKP_FvH8egl5GHJe-_omlC2YyL_b28jMLENbxDFD-3KPjZFBhSLrRukX2PlayYTwEiTtokA2R9_11vQvJgP8KFEjGHg6zsAMawAEBn2H_hz2knb8ltnpEA5YSKVcV3nUtojkCNi_WUz7xUKd7efw1oI_lbnKrS7HkyC0JkQUZ1pCWUlSXNmgjMEhsn823a1LFzpV7rOv4vayYvvFX61hB9R78VjpyxJiYpDwRZLiUY3AK4WY8NqFDbjXR7rT4CkFHEf-VhSQQ8ZNvlpod1nmeVQVizHH9e7Tq7wsWz-LWEk3Hx6LmcrgDsL79LZYG9JXU5IdvG8RvLNx9cSwEI8yxcchpISAaot7UoYQ=",
+                       "type": "validator",
+                       "lastSeen": 1575915094734973000
+                     },
+                     {
+                       "host": "3.9.102.214:4000",
+                       "pubKey": "AAAAAAAAAAIKwAECSqKy8I8KkSYIBSctxRBRxuR61PpAOwK0UQtkeuPRdwusAyaoBbvv1IBWyMEhvbgT4CtgUnGfYH2s06CIJ09lWWvQ0Jkgthq12mG73H9QSTNM8RITlF1X5ax9BV0EK34M5dUncn1uEYzJzcbaLjUarf2bqoy906dtQpppUWDRLJI6ycw7rKKJ4ZNUhgi4KAEGBsSgLqc0zDKs0rArwouZyz4ofoWnY68mdJKrVy6Zqz83DSdc7B2hqqkHX_Bfeb4SwAEOVCUN3EwiVroS5-TOq2o7hYSxphK9X0G23N-IBZ0Tr1Rl8XEiJ-OEy0rqnAKwmhAZJWnx3u8oXqbZtOWIZmzQSpcoxhgwfhdmTZJCqT2RVzZyeFItX4sVeilEP3z2xdsJs8-a1kg6UZnx1s1BNLBo7eZrreZygWojPCIDBn03fSAflXoVc5PpY2CGy5MA_IgWgSYBHDdoZEtigp_amjqK7Us44Db20XpLxMXfbahiqa7WKNnMgi6Ca2H67VtaaD8awAEF3zbE1nZRAa7a8vbU25c80YBYJBaW8P6FwXQI-K0Xk5MakwYeMMnIrm6w6IS_0XAO5YlD453GLqnxY8H1BEnRpfOnT7PE4el9mJ8MuYQMo6R2up0lGCmYM0YA9FORjroM3ng69SEPfJPCReG7LfJkERl_m2U403ertDRBYrlqCDagDfyI500srBcMrjSvV3oNouyyx3yZUrjLQfbHhDteQFsYdmakJs8Y-Q9-5MXCcrz6Qa4xwv522Euv0CCxkHcawAEYjfsU_zDhUZA1ey1aquWXlFOnx-iEALqxW1slDYHwQ1M2SILc-v_E6i1doa5e_bAZHVezBHFAlaNAVedNyHFFJxYAqAK3hbzbvl2glw3Q6h_rTXElymloqtaqVFIJ-oUWWOHsZBmu8EDA-HzvGCiBa_GbRaVfh2lE4ObeMXoJrEm_5dbxxeEic2l3IYeIz40N9ooQQOkQcOZdY4AXWYCavIAwWEJBjLtptJgCLu9a_zM1S5GsiyJHpdDs46WbP0EawAEWZ-95Sf0YAHujxRNLdXgpqe0ZF8loVwzZfvyMvqaxF1Ug274BqHuY_c5NdPAzuqoTwjfEn8NKEoaNqlumM75FUYbaTd7mXvk4WVYWjVnkO40dfQjRB7DYhvj0LBlbndAJ4wJIA2ilPYgjZsXVbNNh3e2j3u9eABd0VaFMbSb8Sz5_31r8HzoWmPJs3HiyuyANGFUA6CvAnMN6K3b-D8BhFZU_nPUTgu80o8_n6LQt-XWbaC_mTHzsnOjzBiPJxlYawAEW3bmOEtStH2T8q7vMkhchImp2-hg9MFYGBmEe9sSByTn3NUf8eksqXOC1dUjHkXoZm298FgUYLkNdnlxWpf993j5mEDoFxjcTB7scBD7k6nu6Nrs_wK0-seS8gsHrx9UK7GwAsi10q82Cm4PFyAtrWjmy_d9WLHuZt6VIOKunTs8cf0FwNUiMcvZsruqIFJcP7iWxdiFdUkh65P_iCz1ZEjJcj2GEZoq4v3a3by1aizGPaaiKc1jd_T-XJg_YpncawAEWnstu5b9WiZv0x8xfsiMk6YRlU0Cnj5svxLLXz_8drvwAa--GBY5yH0ke2EM6udMEi2EPeFcGTe6Sjs0YEhSbY7Uad_8suD2J4tIWJSWBbiyvh7rSqzv57m7BlsVcHfQJn_wNH-UlC9xkx8vg-LwfN8_FlxvHNPTc7XZG3lKYbwpUWlZxAziOYT1VQ-2K2bQQBBMdix-ht_SjccL1Dc2dP5kDazQ8yZV_8xnyeheazEedWe63uutfkHlZRg9YwP8=",
+                       "type": "validator",
+                       "lastSeen": 1575915094967382800
+                     }
+                   ],
+                   "mixNodes": [
+                     {
+                       "host": "35.176.155.107:1789",
+                       "pubKey": "zSob16499jT7C3S3ky4GihNOjlU6aLfSRkf1xAxOwV0=",
+                       "layer": 3,
+                       "lastSeen": 1575915096805374500
+                     },
+                     {
+                       "host": "18.130.86.190:1789",
+                       "pubKey": "vCdpFc0NvW0NSqsuTxtjFtiSY35aXesgT3JNA8sSIXk=",
+                       "layer": 1,
+                       "lastSeen": 1575915097370376000
+                     },
+                     {
+                       "host": "3.10.22.152:1789",
+                       "pubKey": "OwOqwWjh_IlnaWS2PxO6odnhNahOYpRCkju50beQCTA=",
+                       "layer": 1,
+                       "lastSeen": 1575915097639423500
+                     },
+                     {
+                       "host": "35.178.213.77:1789",
+                       "pubKey": "nkkrUjgL8UJk05QydvWvFSvtRB6nmeV8RMvH5540J3s=",
+                       "layer": 2,
+                       "lastSeen": 1575915097895166500
+                     },
+                     {
+                       "host": "52.56.99.196:1789",
+                       "pubKey": "whHuBuEc6zyOZOquKbuATaH4Crml61V_3Y-MztpWhF4=",
+                       "layer": 2,
+                       "lastSeen": 1575915096255174700
+                     },
+                     {
+                       "host": "3.9.12.238:1789",
+                       "pubKey": "vk5Sr-Xyi0cTbugACv8U42ZJ6hs6cGDox0rpmXY94Fc=",
+                       "layer": 3,
+                       "lastSeen": 1575915096497827600
+                     }
+                   ],
+                   "mixProviderNodes": [
+                     {
+                       "host": "3.8.176.11:1789",
+                       "pubKey": "54U6krAr-j9nQXFlsHk3io04_p0tctuqH71t7w_usgI=",
+                       "registeredClients": [
+                         {
+                           "pubKey": "zOqdJFH49HcgGSCRnmbXGzovnwRLEPN0YGN1SCafTyo="
+                         },
+                         {
+                           "pubKey": "fy9xo69hZ2UJ9uxhIS1YzKHZsH8saV-02AiyCNXPNUc="
+                         },
+                         {
+                           "pubKey": "6hFCz42d5AODAoMXqcBWtoOoZh-7hPMDXbLKKhS7x3I="
+                         },
+                         {
+                           "pubKey": "pPZMDzw0FyZ-LBAhwCvjlPGj1p2bO_vaWbRBc7Ojq3Q="
+                         },
+                         {
+                           "pubKey": "kjl5poXt2GdIrHLuMBjmTBSChV-zMqsXFBhKZoIREVI="
+                         },
+                         {
+                           "pubKey": "T1jMGuk7-rcIUWnVopAsdGzhAJ7ZVymwON2LzjuOfnY="
+                         },
+                         {
+                           "pubKey": "3X0eKPkflx5Vzok0Gk0jm5-YEfvPAuWP55ovyBUOtXA="
+                         },
+                         {
+                           "pubKey": "5wC3i8rCZolLKbWJ9U6eLieXNGLKM21dtL6lR30u_hE="
+                         },
+                         {
+                           "pubKey": "1P6p7fgjwjlEepD9JgbN1V-rk9n36-hCmPN5P6y62n4="
+                         },
+                         {
+                           "pubKey": "aBjlLQKKFroqhX_kvYLnMm1uq3FJdQWqVy9Q35zzERA="
+                         },
+                         {
+                           "pubKey": "I0gVOPj6lv9ha60xPYKeAgbeUU8pdyMD-Y7Nb1nS9EQ="
+                         },
+                         {
+                           "pubKey": "WgdvQ74QH1uFDWDL2YeApvv7oniNGh9BQJ_HZam20QA="
+                         },
+                         {
+                           "pubKey": "Mlw23KaSL2hyrIjEZM76bZStt2iMzxVAqXwO5clJfxg="
+                         },
+                         {
+                           "pubKey": "F9xzbjnMQVN4ZidcqN2ip9kVnI9wbS39aVayZGiMihY="
+                         },
+                         {
+                           "pubKey": "s6pfVkZrUG--RNjfzS55N2oPvFkMdvgb1LUut6gqRy4="
+                         },
+                         {
+                           "pubKey": "bSi-9k0jJNKc8PGx8M3SWFaNpORFjYw-NkWXRZVRWGU="
+                         },
+                         {
+                           "pubKey": "pz6ahQcGOQcZBFx1tGmzRngqk6BecXB_wFd4WVdOQDU="
+                         },
+                         {
+                           "pubKey": "5sfwIMcG2zRCxhDh4D9Evw1WPI5bfKZAShM_6o9Pu1I="
+                         },
+                         {
+                           "pubKey": "9fZnxXy9onGPpZ3Ygckqw0okqCw3di02sLr-NTBr4SE="
+                         },
+                         {
+                           "pubKey": "Q0TbbggOwzZjalUdi5eEHVFi9VMv-rMm5mJPUZZs12A="
+                         },
+                         {
+                           "pubKey": "aPNyox_qAIGFB2-wZ0lc9iAWDN6jzLojApSiWVFjCks="
+                         },
+                         {
+                           "pubKey": "DUKLEsIGMw-ucs3DjS7Ag9qCb5-C_A84DuIsZuLkdwI="
+                         },
+                         {
+                           "pubKey": "YV84vPoSrLf1p9Sw6FnnrcCpS3kvpJUfKyKpnwk8z0Q="
+                         },
+                         {
+                           "pubKey": "_IYEzZQoBAYeTxqzpEe_ez1-7pn7aId8AKliazy0qlE="
+                         },
+                         {
+                           "pubKey": "srkOAoVU-a02lnEsoH_wOLLw7HFx_xHIZUSbnhjFwDw="
+                         },
+                         {
+                           "pubKey": "LxSCBn_OEQq-hI2xDi6bfGoioRO_lSTIq6AQ9l1k5jg="
+                         },
+                         {
+                           "pubKey": "OC0OOqtGfAytgZjthpjoKeYNa0VrtzfgZ0iO5Fag5y8="
+                         },
+                         {
+                           "pubKey": "ImVEch2focRhm1ial1gA6YJPr6WDyW3oh-OgYcO5Ll4="
+                         },
+                         {
+                           "pubKey": "yTDnzLvEaSq0mC9xNrtyjpAKtsIU6yRuBepCuWQMBm4="
+                         },
+                         {
+                           "pubKey": "-LCUUc46HuL7iUEOMkrlVAkvvulRiQ7dR1QYh7bkKxw="
+                         },
+                         {
+                           "pubKey": "Bx1MGpISig25rqe7mhoX68EROUPPzmF7yGLYah9DPgM="
+                         },
+                         {
+                           "pubKey": "Z4Nu6iwLmgJ93yoIFTbTEBeDAHRwS-vo1T_K2Kv1FQo="
+                         },
+                         {
+                           "pubKey": "cKFAGxllwAmEXCtDxG_T1iEm3-lKWUVQxxpDBje6mQU="
+                         },
+                         {
+                           "pubKey": "pQV40whlQWUSXtrNTTePzO6sdq3zr1JUIWZWvD443nY="
+                         },
+                         {
+                           "pubKey": "6Bb5HwnVqJPy5wcNsaHY-0y__coZCE7XC80kUkesnRU="
+                         },
+                         {
+                           "pubKey": "COGdpfhmzNGR6YX820GqJIkjOihL8mr6-h-d3JlTDFA="
+                         }
+                       ],
+                       "lastSeen": 1575915097358694100
+                     },
+                     {
+                       "host": "35.178.212.193:1789",
+                       "pubKey": "sA-sxi038pEbGy4lgZWG-RdHHDkA6kZzu44G0LUxFSc=",
+                       "registeredClients": [
+                         {
+                           "pubKey": "UE-7r6-bpw0b4T3GxOBVxlg02psx23DF2p5Tuf-OBSE="
+                         },
+                         {
+                           "pubKey": "UnZuLpzq64_EPtIcr1Fd-5AESBCBLFnDMDsjUaOqrUA="
+                         },
+                         {
+                           "pubKey": "4ExXPrW5w0nZIQ5ravBRT6H9r0RH0MXuOcGIF8HzUhg="
+                         },
+                         {
+                           "pubKey": "wPsRpJPi1e2sjItyRKDkFACbxwu3Cw5GlYVPmdYxk2U="
+                         },
+                         {
+                           "pubKey": "UfK5UvT3HUkT1SGbv1QGafy3in3uQ9a6NSy5EOT6k0k="
+                         },
+                         {
+                           "pubKey": "5Giu-tJpGXU0S9Av75iAv5qDO0k6l8v_k9-UCcRUCl4="
+                         },
+                         {
+                           "pubKey": "MnQxlmmKDybku4CfxsQQxfftilsaphF9Gq1w3MB1ZCE="
+                         },
+                         {
+                           "pubKey": "GdP1fHVs2R65EkuWVWKZSz6WPDh0MgThyuBOv6_xsmQ="
+                         },
+                         {
+                           "pubKey": "4JEtSrKsonmBuDvxJ9nITSu7iC4f8reutXRAVugPgS4="
+                         },
+                         {
+                           "pubKey": "q4XyuUJbSGJaoRb3SmWzeX88V2dKB7sPTf72BAtQp3k="
+                         },
+                         {
+                           "pubKey": "gTzKd1Ph5bpUw-JxTZiCe8RBfO-FsZiVYDYioQ-6dVg="
+                         },
+                         {
+                           "pubKey": "NX4oaDLYEOmMUP_9pcEaZv5MJHJ4ZYAoxPQDxov7tRs="
+                         },
+                         {
+                           "pubKey": "7fbk4oGQNlTW-tnWjVz8rWtKrtAicTsiNWgO98sqMyk="
+                         },
+                         {
+                           "pubKey": "w1bfLpnd3rWu5JczB0nQfnE2S6nUCbx2AA7HDE48DQo="
+                         }
+                       ],
+                       "lastSeen": 1575915097869025000
+                     }
+                   ]
+                 }"#.to_string()
+        }
+    }
+}

src/clients/mix.rs

@@ -0,0 +1,52 @@
+use sphinx::route::NodeAddressBytes;
+use sphinx::SphinxPacket;
+use std::net::SocketAddr;
+use tokio::prelude::*;
+
+pub struct MixClient {}
+
+impl MixClient {
+    pub fn new() -> MixClient {
+        MixClient {}
+    }
+
+    // Sends a Sphinx packet to a mixnode.
+    pub async fn send(
+        &self,
+        packet: SphinxPacket,
+        mix_addr: SocketAddr,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let bytes = packet.to_bytes();
+
+        println!("socket addr: {:?}", mix_addr);
+
+        let mut stream = tokio::net::TcpStream::connect(mix_addr).await?;
+        stream.write_all(&bytes[..]).await?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod sending_a_sphinx_packet {
+    // use super::*;
+    // use sphinx::SphinxPacket;
+
+    #[test]
+    fn works() {
+        // arrange
+        //        let directory = Client::new();
+        //        let message = "Hello, Sphinx!".as_bytes().to_vec();
+        //        let mixes = directory.get_mixes();
+        //        let destination = directory.get_destination();
+        //        let delays = sphinx::header::delays::generate(2);
+        //        let packet = SphinxPacket::new(message, &mixes, &destination, &delays).unwrap();
+        //        let mix_client = MixClient::new();
+        //        let first_hop = mixes.first().unwrap();
+        //
+        //        // act
+        //        mix_client.send(packet, first_hop);
+
+        // assert
+        // wtf are we supposed to assert here?
+    }
+}

src/clients/mod.rs

@@ -0,0 +1,334 @@
+use crate::clients::directory::presence::Topology;
+use crate::clients::mix::MixClient;
+use crate::clients::provider::ProviderClient;
+use crate::sockets::tcp;
+use crate::sockets::ws;
+use crate::utils;
+use crate::utils::topology::get_topology;
+use futures::channel::{mpsc, oneshot};
+use futures::join;
+use futures::lock::Mutex as FMutex;
+use futures::select;
+use futures::{SinkExt, StreamExt};
+use sfw_provider_requests::AuthToken;
+use sphinx::route::{Destination, DestinationAddressBytes};
+use sphinx::SphinxPacket;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::runtime::Runtime;
+
+pub mod directory;
+pub mod mix;
+pub mod provider;
+pub mod validator;
+
+const LOOP_COVER_AVERAGE_DELAY: f64 = 0.5;
+// seconds
+const MESSAGE_SENDING_AVERAGE_DELAY: f64 = 0.5;
+//  seconds;
+const FETCH_MESSAGES_DELAY: f64 = 1.0; // seconds;
+
+// provider-poller sends polls service provider; receives messages
+// provider-poller sends (TX) to ReceivedBufferController (RX)
+// ReceivedBufferController sends (TX) to ... ??Client??
+// outQueueController sends (TX) to TrafficStreamController (RX)
+// TrafficStreamController sends messages to mixnet
+// ... ??Client?? sends (TX) to outQueueController (RX)
+// Loop cover traffic stream just sends messages to mixnet without any channel communication
+
+struct MixMessage(SocketAddr, SphinxPacket);
+
+struct MixTrafficController;
+
+impl MixTrafficController {
+    // this was way more difficult to implement than what this code may suggest...
+    async fn run(mut rx: mpsc::UnboundedReceiver<MixMessage>) {
+        let mix_client = MixClient::new();
+        while let Some(mix_message) = rx.next().await {
+            println!(
+                "[MIX TRAFFIC CONTROL] - got a mix_message for {:?}",
+                mix_message.0
+            );
+            let send_res = mix_client.send(mix_message.1, mix_message.0).await;
+            match send_res {
+                Ok(_) => println!("We successfully sent the message!"),
+                Err(e) => eprintln!("We failed to send the message :( - {:?}", e),
+            };
+        }
+    }
+}
+
+pub type BufferResponse = oneshot::Sender<Vec<Vec<u8>>>;
+
+struct ReceivedMessagesBuffer {
+    messages: Vec<Vec<u8>>,
+}
+
+impl ReceivedMessagesBuffer {
+    fn add_arc_futures_mutex(self) -> Arc<FMutex<Self>> {
+        Arc::new(FMutex::new(self))
+    }
+
+    fn new() -> Self {
+        ReceivedMessagesBuffer {
+            messages: Vec::new(),
+        }
+    }
+
+    async fn add_new_messages(buf: Arc<FMutex<Self>>, msgs: Vec<Vec<u8>>) {
+        println!("Adding new messages to the buffer! {:?}", msgs);
+        let mut unlocked = buf.lock().await;
+        unlocked.messages.extend(msgs);
+    }
+
+    async fn run_poller_input_controller(
+        buf: Arc<FMutex<Self>>,
+        mut poller_rx: mpsc::UnboundedReceiver<Vec<Vec<u8>>>,
+    ) {
+        while let Some(new_messages) = poller_rx.next().await {
+            ReceivedMessagesBuffer::add_new_messages(buf.clone(), new_messages).await;
+        }
+    }
+
+    async fn acquire_and_empty(buf: Arc<FMutex<Self>>) -> Vec<Vec<u8>> {
+        let mut unlocked = buf.lock().await;
+        std::mem::replace(&mut unlocked.messages, Vec::new())
+    }
+
+    async fn run_query_output_controller(
+        buf: Arc<FMutex<Self>>,
+        mut query_receiver: mpsc::UnboundedReceiver<BufferResponse>,
+    ) {
+        while let Some(request) = query_receiver.next().await {
+            let messages = ReceivedMessagesBuffer::acquire_and_empty(buf.clone()).await;
+            // if this fails, the whole application needs to blow
+            // because currently only this thread would fail
+            request.send(messages).unwrap();
+        }
+    }
+}
+
+pub enum SocketType {
+    TCP,
+    WebSocket,
+    None,
+}
+
+pub struct NymClient {
+    // to be replaced by something else I guess
+    address: DestinationAddressBytes,
+    pub input_tx: mpsc::UnboundedSender<InputMessage>,
+    // to be used by "send" function or socket, etc
+    input_rx: mpsc::UnboundedReceiver<InputMessage>,
+    socket_listening_address: SocketAddr,
+    directory: String,
+    auth_token: Option<AuthToken>,
+    socket_type: SocketType,
+}
+
+#[derive(Debug)]
+pub struct InputMessage(pub Destination, pub Vec<u8>);
+
+impl NymClient {
+    pub fn new(
+        address: DestinationAddressBytes,
+        socket_listening_address: SocketAddr,
+        directory: String,
+        auth_token: Option<AuthToken>,
+        socket_type: SocketType,
+    ) -> Self {
+        let (input_tx, input_rx) = mpsc::unbounded::<InputMessage>();
+
+        NymClient {
+            address,
+            input_tx,
+            input_rx,
+            socket_listening_address,
+            directory,
+            auth_token,
+            socket_type,
+        }
+    }
+
+    async fn start_loop_cover_traffic_stream(
+        mut tx: mpsc::UnboundedSender<MixMessage>,
+        our_info: Destination,
+        topology: Topology,
+    ) {
+        loop {
+            println!("[LOOP COVER TRAFFIC STREAM] - next cover message!");
+            let delay = utils::poisson::sample(LOOP_COVER_AVERAGE_DELAY);
+            let delay_duration = Duration::from_secs_f64(delay);
+            tokio::time::delay_for(delay_duration).await;
+            let cover_message =
+                utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology);
+            tx.send(MixMessage(cover_message.0, cover_message.1))
+                .await
+                .unwrap();
+        }
+    }
+
+    async fn control_out_queue(
+        mut mix_tx: mpsc::UnboundedSender<MixMessage>,
+        mut input_rx: mpsc::UnboundedReceiver<InputMessage>,
+        our_info: Destination,
+        topology: Topology,
+    ) {
+        loop {
+            println!("[OUT QUEUE] here I will be sending real traffic (or loop cover if nothing is available)");
+            select! {
+                real_message = input_rx.next() => {
+                    println!("[OUT QUEUE] - we got a real message!");
+                    let real_message = real_message.expect("The channel must have closed! - if the client hasn't crashed, it should have!");
+                    println!("real: {:?}", real_message);
+                    let encapsulated_message = utils::sphinx::encapsulate_message(real_message.0, real_message.1, &topology);
+                    mix_tx.send(MixMessage(encapsulated_message.0, encapsulated_message.1)).await.unwrap();
+                },
+
+                default => {
+                    println!("[OUT QUEUE] - no real message - going to send extra loop cover");
+                    let cover_message = utils::sphinx::loop_cover_message(our_info.address, our_info.identifier, &topology);
+                    mix_tx.send(MixMessage(cover_message.0, cover_message.1)).await.unwrap();
+                }
+            };
+
+            let delay_duration = Duration::from_secs_f64(MESSAGE_SENDING_AVERAGE_DELAY);
+            tokio::time::delay_for(delay_duration).await;
+        }
+    }
+
+    async fn start_provider_polling(
+        provider_client: ProviderClient,
+        mut poller_tx: mpsc::UnboundedSender<Vec<Vec<u8>>>,
+    ) {
+        let loop_message = &utils::sphinx::LOOP_COVER_MESSAGE_PAYLOAD.to_vec();
+        let dummy_message = &sfw_provider_requests::DUMMY_MESSAGE_CONTENT.to_vec();
+        loop {
+            let delay_duration = Duration::from_secs_f64(FETCH_MESSAGES_DELAY);
+            tokio::time::delay_for(delay_duration).await;
+            println!("[FETCH MSG] - Polling provider...");
+            let messages = provider_client.retrieve_messages().await.unwrap();
+            let good_messages = messages
+                .into_iter()
+                .filter(|message| message != loop_message && message != dummy_message)
+                .collect();
+            // if any of those fails, whole application should blow...
+            poller_tx.send(good_messages).await.unwrap();
+        }
+    }
+
+    pub fn start(self) -> Result<(), Box<dyn std::error::Error>> {
+        println!("Starting nym client");
+        let mut rt = Runtime::new()?;
+
+        let topology = get_topology(self.directory.clone());
+        // this is temporary and assumes there exists only a single provider.
+        let provider_address: SocketAddr = topology
+            .mix_provider_nodes
+            .first()
+            .unwrap()
+            .host
+            .parse()
+            .unwrap();
+
+        let mut provider_client =
+            ProviderClient::new(provider_address, self.address, self.auth_token);
+
+        // registration
+        rt.block_on(async {
+            match self.auth_token {
+                None => {
+                    let auth_token = provider_client.register().await.unwrap();
+                    provider_client.update_token(auth_token);
+                    println!("Obtained new token! - {:?}", auth_token);
+                }
+                Some(token) => println!("Already got the token! - {:?}", token),
+            }
+        });
+
+        // channels for intercomponent communication
+        let (mix_tx, mix_rx) = mpsc::unbounded();
+        let (poller_input_tx, poller_input_rx) = mpsc::unbounded();
+        let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) =
+            mpsc::unbounded();
+
+        let received_messages_buffer = ReceivedMessagesBuffer::new().add_arc_futures_mutex();
+
+        let received_messages_buffer_input_controller_future =
+            rt.spawn(ReceivedMessagesBuffer::run_poller_input_controller(
+                received_messages_buffer.clone(),
+                poller_input_rx,
+            ));
+        let received_messages_buffer_output_controller_future =
+            rt.spawn(ReceivedMessagesBuffer::run_query_output_controller(
+                received_messages_buffer,
+                received_messages_buffer_output_rx,
+            ));
+
+        let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx));
+        let loop_cover_traffic_future = rt.spawn(NymClient::start_loop_cover_traffic_stream(
+            mix_tx.clone(),
+            Destination::new(self.address, Default::default()),
+            topology.clone(),
+        ));
+
+        let out_queue_control_future = rt.spawn(NymClient::control_out_queue(
+            mix_tx,
+            self.input_rx,
+            Destination::new(self.address, Default::default()),
+            topology.clone(),
+        ));
+
+        let provider_polling_future = rt.spawn(NymClient::start_provider_polling(
+            provider_client,
+            poller_input_tx,
+        ));
+
+        match self.socket_type {
+            SocketType::WebSocket => {
+                rt.spawn(ws::start_websocket(
+                    self.socket_listening_address,
+                    self.input_tx,
+                    received_messages_buffer_output_tx,
+                    self.address,
+                    topology,
+                ));
+            }
+            SocketType::TCP => {
+                rt.spawn(tcp::start_tcpsocket(
+                    self.socket_listening_address,
+                    self.input_tx,
+                    received_messages_buffer_output_tx,
+                    self.address,
+                    topology,
+                ));
+            }
+            SocketType::None => (),
+        }
+
+        rt.block_on(async {
+            let future_results = join!(
+                received_messages_buffer_input_controller_future,
+                received_messages_buffer_output_controller_future,
+                mix_traffic_future,
+                loop_cover_traffic_future,
+                out_queue_control_future,
+                provider_polling_future,
+            );
+
+            assert!(
+                future_results.0.is_ok()
+                    && future_results.1.is_ok()
+                    && future_results.2.is_ok()
+                    && future_results.3.is_ok()
+                    && future_results.4.is_ok()
+                    && future_results.5.is_ok()
+            );
+        });
+
+        // this line in theory should never be reached as the runtime should be permanently blocked on traffic senders
+        eprintln!("The client went kaput...");
+        Ok(())
+    }
+}

src/clients/provider.rs

@@ -0,0 +1,118 @@
+use futures::io::Error;
+use sfw_provider_requests::requests::{ProviderRequest, PullRequest, RegisterRequest};
+use sfw_provider_requests::responses::{
+    ProviderResponse, ProviderResponseError, PullResponse, RegisterResponse,
+};
+use sfw_provider_requests::AuthToken;
+use sphinx::route::DestinationAddressBytes;
+use std::net::{Shutdown, SocketAddr};
+use std::time::Duration;
+use tokio::prelude::*;
+
+#[derive(Debug)]
+pub enum ProviderClientError {
+    ClientAlreadyRegisteredError,
+    EmptyAuthTokenError,
+    NetworkError,
+
+    InvalidRequestError,
+    InvalidResponseError,
+    InvalidResponseLengthError,
+}
+
+impl From<io::Error> for ProviderClientError {
+    fn from(_: Error) -> Self {
+        use ProviderClientError::*;
+
+        NetworkError
+    }
+}
+
+impl From<ProviderResponseError> for ProviderClientError {
+    fn from(err: ProviderResponseError) -> Self {
+        use ProviderClientError::*;
+        match err {
+            ProviderResponseError::MarshalError => InvalidRequestError,
+            ProviderResponseError::UnmarshalError => InvalidResponseError,
+            ProviderResponseError::UnmarshalErrorInvalidLength => InvalidResponseLengthError,
+        }
+    }
+}
+
+pub struct ProviderClient {
+    provider_network_address: SocketAddr,
+    our_address: DestinationAddressBytes,
+    auth_token: Option<AuthToken>,
+}
+
+impl ProviderClient {
+    pub fn new(
+        provider_network_address: SocketAddr,
+        our_address: DestinationAddressBytes,
+        auth_token: Option<AuthToken>,
+    ) -> Self {
+        // DH temporary: the provider's client port is not in the topology, but we can't change that
+        // right now without messing up the existing Go mixnet. So I'm going to hardcode this
+        // for the moment until the Go mixnet goes away.
+        let provider_socket = SocketAddr::new(provider_network_address.ip(), 9000);
+
+        ProviderClient {
+            provider_network_address: provider_socket,
+            our_address,
+            auth_token,
+        }
+    }
+
+    pub fn update_token(&mut self, auth_token: AuthToken) {
+        self.auth_token = Some(auth_token)
+    }
+
+    pub async fn send_request(&self, bytes: Vec<u8>) -> Result<Vec<u8>, ProviderClientError> {
+        let mut socket = tokio::net::TcpStream::connect(self.provider_network_address).await?;
+        println!("keep alive: {:?}", socket.keepalive());
+        socket.set_keepalive(Some(Duration::from_secs(2))).unwrap();
+        socket.write_all(&bytes[..]).await?;
+        if let Err(_e) = socket.shutdown(Shutdown::Write) {
+            // TODO: make it a silent log once we have a proper logging library
+            //            eprintln!("failed to close write part of the socket; err = {:?}", e)
+        }
+
+        let mut response = Vec::new();
+        socket.read_to_end(&mut response).await?;
+        if let Err(_e) = socket.shutdown(Shutdown::Read) {
+            // TODO: make it a silent log once we have a proper logging library
+            //            eprintln!("failed to close read part of the socket; err = {:?}", e)
+        }
+
+        Ok(response)
+    }
+
+    pub async fn retrieve_messages(&self) -> Result<Vec<Vec<u8>>, ProviderClientError> {
+        if self.auth_token.is_none() {
+            return Err(ProviderClientError::EmptyAuthTokenError);
+        }
+
+        let pull_request = PullRequest::new(self.our_address, self.auth_token.unwrap());
+        let bytes = pull_request.to_bytes();
+
+        let response = self.send_request(bytes).await?;
+        println!("Received the following response: {:?}", response);
+
+        let parsed_response = PullResponse::from_bytes(&response)?;
+        Ok(parsed_response.messages)
+    }
+
+    pub async fn register(&self) -> Result<AuthToken, ProviderClientError> {
+        if self.auth_token.is_some() {
+            return Err(ProviderClientError::ClientAlreadyRegisteredError);
+        }
+
+        let register_request = RegisterRequest::new(self.our_address);
+        let bytes = register_request.to_bytes();
+
+        let response = self.send_request(bytes).await?;
+        let parsed_response = RegisterResponse::from_bytes(&response)?;
+
+        Ok(parsed_response.auth_token)
+    }
+}

src/clients/validator.rs

@@ -0,0 +1 @@
+

src/commands/init.rs

@@ -0,0 +1,20 @@
+use crate::banner;
+use crate::identity::mixnet;
+use crate::persistence::pathfinder::Pathfinder;
+use crate::persistence::pemstore::PemStore;
+use clap::ArgMatches;
+
+pub fn execute(matches: &ArgMatches) {
+    println!("{}", banner());
+    println!("Initialising client...");
+
+    let id = matches.value_of("id").unwrap().to_string(); // required for now
+    let pathfinder = Pathfinder::new(id);
+
+    println!("Writing keypairs to {:?}...", pathfinder.config_dir);
+    let mix_keys = mixnet::KeyPair::new();
+    let pem_store = PemStore::new(pathfinder);
+    pem_store.write(mix_keys);
+
+    println!("Client configuration completed.\n\n\n")
+}

src/commands/mod.rs

@@ -0,0 +1,4 @@
+pub mod init;
+pub mod run;
+pub mod tcpsocket;
+pub mod websocket;

src/commands/run.rs

@@ -0,0 +1,17 @@
+use crate::banner;
+//use crate::clients::NymClient;
+//use crate::persistence::pemstore;
+use clap::ArgMatches;
+
+pub fn execute(_matches: &ArgMatches) {
+    println!("{}", banner());
+    panic!("For time being this command is deprecated! Please use 'websocket' instead");
+    //
+    //    let is_local = matches.is_present("local");
+    //    let id = matches.value_of("id").unwrap().to_string();
+    //    println!("Starting client...");
+    //
+    //    let keypair = pemstore::read_keypair_from_disk(id);
+    //    let client = NymClient::new(keypair.public_bytes(), is_local);
+    //    client.start("127.0.0.1:9000".parse().unwrap()).unwrap();
+}

src/commands/tcpsocket.rs

@@ -0,0 +1,44 @@
+use crate::banner;
+use crate::clients::{NymClient, SocketType};
+use crate::persistence::pemstore;
+use crate::sockets::tcp;
+
+use clap::ArgMatches;
+use std::net::ToSocketAddrs;
+
+pub fn execute(matches: &ArgMatches) {
+    println!("{}", banner());
+
+    let id = matches.value_of("id").unwrap().to_string();
+    let port = match matches.value_of("port").unwrap_or("9001").parse::<u16>() {
+        Ok(n) => n,
+        Err(err) => panic!("Invalid port value provided - {:?}", err),
+    };
+
+    let directory_server = matches
+        .value_of("directory")
+        .unwrap_or("https://directory.nymtech.net")
+        .to_string();
+
+    println!("Starting TCP socket on port: {:?}", port);
+    println!("Listening for messages...");
+
+    let socket_address = ("127.0.0.1", port)
+        .to_socket_addrs()
+        .expect("Failed to combine host and port")
+        .next()
+        .expect("Failed to extract the socket address from the iterator");
+
+    let keypair = pemstore::read_keypair_from_disk(id);
+    let auth_token = None;
+
+    let client = NymClient::new(
+        keypair.public_bytes(),
+        socket_address.clone(),
+        directory_server,
+        auth_token,
+        SocketType::TCP,
+    );
+
+    client.start().unwrap();
+}

src/commands/websocket.rs

@@ -0,0 +1,43 @@
+use crate::banner;
+use crate::clients::{NymClient, SocketType};
+use crate::persistence::pemstore;
+
+use clap::ArgMatches;
+use std::net::ToSocketAddrs;
+
+pub fn execute(matches: &ArgMatches) {
+    println!("{}", banner());
+
+    let id = matches.value_of("id").unwrap().to_string();
+    let port = match matches.value_of("port").unwrap_or("9001").parse::<u16>() {
+        Ok(n) => n,
+        Err(err) => panic!("Invalid port value provided - {:?}", err),
+    };
+
+    let directory_server = matches
+        .value_of("directory")
+        .unwrap_or("https://directory.nymtech.net")
+        .to_string();
+
+    println!("Starting websocket on port: {:?}", port);
+    println!("Listening for messages...");
+
+    let socket_address = ("127.0.0.1", port)
+        .to_socket_addrs()
+        .expect("Failed to combine host and port")
+        .next()
+        .expect("Failed to extract the socket address from the iterator");
+
+    let keypair = pemstore::read_keypair_from_disk(id);
+    // TODO: reading auth_token from disk (if exists);
+    let auth_token = None;
+    let client = NymClient::new(
+        keypair.public_bytes(),
+        socket_address,
+        directory_server,
+        auth_token,
+        SocketType::WebSocket,
+    );
+
+    client.start().unwrap();
+}

src/directory.rs

@@ -1,70 +0,0 @@
-use sphinx::route::{Node as SphinxNode, Destination};
-
-pub struct DirectoryClient {}
-
-impl DirectoryClient {
-    pub fn new() -> DirectoryClient {
-        DirectoryClient {}
-    }
-
-    // Hardcoded for now. Later, this should make a network request to the directory server (if one
-    // has not yet been made), parse the returned JSON, memoize the full tree of active nodes,
-    // and return the list of currently active mix nodes.
-    pub fn get_mixes(&self) -> Vec<SphinxNode> {
-        fake_directory_mixes()
-    }
-
-    pub fn get_destination(&self) -> Destination {
-        Destination {
-            address: [0u8;32],
-            identifier: [0u8; 16],
-        }
-    }
-}
-
-
-
-#[cfg(test)]
-mod retrieving_mixnode_list {
-    use super::*;
-
-    #[test]
-    fn always_works_because_it_is_hardcoded() {
-        let directory = DirectoryClient::new();
-        let expected = fake_directory_mixes().as_slice().to_owned();
-
-        let mixes = directory.get_mixes();
-
-        assert_eq!(expected, mixes.as_slice());
-    }
-}
-
-#[cfg(test)]
-mod retrieving_destinations {
-    use super::*;
-    
-    #[test]
-    fn always_works_because_it_is_hardcoded() {
-        let directory = DirectoryClient::new();
-        let expected = Destination {
-            address: [0u8;32],
-            identifier: [0u8; 16],
-        };
-
-        let destination = directory.get_destination();
-
-        assert_eq!(expected, destination);
-    }
-}
-
-fn fake_directory_mixes() -> Vec<SphinxNode> {
-    let node1 = sphinx::route::Node{
-        address: [0u8;32],
-        pub_key: Default::default()
-    };
-    let node2 = sphinx::route::Node{
-        address: [1u8;32],
-        pub_key: Default::default()
-    };
-    vec![node1, node2]
-}

src/identity/mixnet.rs

@@ -0,0 +1,33 @@
+use curve25519_dalek::montgomery::MontgomeryPoint;
+use curve25519_dalek::scalar::Scalar;
+
+// This keypair serves as the user's identity within the Mixnet
+pub struct KeyPair {
+    pub private: Scalar,
+    pub public: MontgomeryPoint,
+}
+
+impl KeyPair {
+    pub fn new() -> KeyPair {
+        let (private, public) = sphinx::crypto::keygen();
+        KeyPair { private, public }
+    }
+
+    pub fn from_bytes(private_bytes: Vec<u8>, public_bytes: Vec<u8>) -> KeyPair {
+        let mut bytes = [0; 32];
+        bytes.copy_from_slice(&private_bytes[..]);
+        let private = Scalar::from_canonical_bytes(bytes).unwrap();
+        let mut bytes = [0; 32];
+        bytes.copy_from_slice(&public_bytes[..]);
+        let public = MontgomeryPoint(bytes);
+        KeyPair { private, public }
+    }
+
+    pub fn private_bytes(&self) -> [u8; 32] {
+        self.private.to_bytes()
+    }
+
+    pub fn public_bytes(&self) -> [u8; 32] {
+        self.public.to_bytes()
+    }
+}

src/identity/mod.rs

@@ -0,0 +1 @@
+pub mod mixnet;

src/identity/validators.rs

@@ -0,0 +1 @@
+// TODO types for Validator keys will go in here once we hook this up.

src/lib.rs

@@ -0,0 +1,5 @@
+pub mod clients;
+pub mod identity;
+pub mod persistence;
+pub mod sockets;
+pub mod utils;

src/main.rs

@@ -1,8 +1,136 @@
-mod directory;
-mod mix;
+use clap::{App, Arg, ArgMatches, SubCommand};
+use std::process;
+
+pub mod clients;
+mod commands;
+mod identity;
+mod persistence;
+mod sockets;
+mod utils;
 
 fn main() {
-    let _ = "Hello, Sphinx!".as_bytes().to_vec();
+    let arg_matches = App::new("Nym Client")
+        .version("0.1.0")
+        .author("Nymtech")
+        .about("Implementation of the Nym Client")
+        .subcommand(
+            SubCommand::with_name("init")
+                .about("Initialise a Nym client. Do this first!")
+                .arg(Arg::with_name("id")
+                    .long("id")
+                    .help("Id of the nym-mixnet-client we want to create config for.")
+                    .takes_value(true)
+                    .required(true)
+                )
+                .arg(Arg::with_name("provider")
+                    .long("provider")
+                    .help("Id of the provider we have preference to connect to. If left empty, a random provider will be chosen.")
+                    .takes_value(true)
+                )
+        )
+        .subcommand(
+            SubCommand::with_name("run")
+                .about("Run a persistent Nym client process")
+                .arg(Arg::with_name("id")
+                    .long("id")
+                    .help("Id of the nym-mixnet-client we want to run.")
+                    .takes_value(true)
+                    .required(true)
+                )
+                .arg(
+                    Arg::with_name("directory")
+                        .long("directory")
+                        .help("Address of the directory server the client is getting topology from")
+                        .takes_value(true),
+                )
+        )
+        .subcommand(
+            SubCommand::with_name("tcpsocket")
+                .about("Run Nym client that listens for bytes on a TCP socket")
+                .arg(
+                    Arg::with_name("port")
+                        .short("p")
+                        .long("port")
+                        .help("Port for TCP socket to listen on")
+                        .takes_value(true)
+                        .required(true),
+                )
+                .arg(
+                    Arg::with_name("directory")
+                        .long("directory")
+                        .help("Address of the directory server the client is getting topology from")
+                        .takes_value(true),
+                )
+                .arg(Arg::with_name("id")
+                    .long("id")
+                    .help("Id of the nym-mixnet-client we want to run.")
+                    .takes_value(true)
+                    .required(true)
+                )
+        )
+        .subcommand(
+            SubCommand::with_name("websocket")
+                .about("Run Nym client that listens on a websocket")
+                .arg(
+                    Arg::with_name("port")
+                        .short("p")
+                        .long("port")
+                        .help("Port for websocket to listen on")
+                        .takes_value(true)
+                )
+                .arg(
+                    Arg::with_name("directory")
+                        .long("directory")
+                        .help("Address of the directory server the client is getting topology from")
+                        .takes_value(true),
+                )
+                .arg(Arg::with_name("id")
+                    .long("id")
+                    .help("Id of the nym-mixnet-client we want to run.")
+                    .takes_value(true)
+                    .required(true)
+                )
+        )
+        .get_matches();
+
+    if let Err(e) = execute(arg_matches) {
+        println!("{}", e);
+        process::exit(1);
+    }
 }
 
+pub mod built_info {
+    // The file has been placed there by the build script.
+    include!(concat!(env!("OUT_DIR"), "/built.rs"));
+}
 
+fn execute(matches: ArgMatches) -> Result<(), String> {
+    match matches.subcommand() {
+        ("init", Some(m)) => Ok(commands::init::execute(m)),
+        ("run", Some(m)) => Ok(commands::run::execute(m)),
+        ("tcpsocket", Some(m)) => Ok(commands::tcpsocket::execute(m)),
+        ("websocket", Some(m)) => Ok(commands::websocket::execute(m)),
+        _ => Err(usage()),
+    }
+}
+
+fn usage() -> String {
+    banner() + "usage: --help to see available options.\n\n"
+}
+
+fn banner() -> String {
+    format!(
+        r#"
+
+      _ __  _   _ _ __ ___
+     | '_ \| | | | '_ \ _ \
+     | | | | |_| | | | | | |
+     |_| |_|\__, |_| |_| |_|
+            |___/
+
+             (client - version {:})
+
+    "#,
+        built_info::PKG_VERSION
+    )
+}

src/mix.rs

@@ -1,41 +0,0 @@
-use crate::directory::DirectoryClient;
-use sphinx::SphinxPacket;
-use sphinx::route::Node;
-
-struct MixClient {}
-
-impl MixClient {
-    fn new() -> MixClient {
-        MixClient {}
-    }
-
-    fn send(&self, packet: SphinxPacket, mix: &Node) {
-        let bytes = packet.to_bytes();
-        // now we shoot it into space!
-    }
-}
-
-
-#[cfg(test)]
-mod sending_a_sphinx_packet {
-    use super::*;
-    use sphinx::SphinxPacket;
-
-    #[test]
-    fn works() {
-        // arrange
-        let directory = DirectoryClient::new();
-        let message = "Hello, Sphinx!".as_bytes().to_vec();
-        let mixes = directory.get_mixes();
-        let destination = directory.get_destination();
-        let packet = SphinxPacket::new(message, &mixes, &destination);
-        let mix_client = MixClient::new();
-        let first_hop = mixes.first().unwrap();
-
-        // act
-        mix_client.send(packet, first_hop);
-
-        // assert
-        // wtf are we supposed to assert here?
-    }
-}

src/persistence/mod.rs

@@ -0,0 +1,2 @@
+pub mod pathfinder;
+pub mod pemstore;

src/persistence/pathfinder.rs

@@ -0,0 +1,21 @@
+use std::path::PathBuf;
+
+pub struct Pathfinder {
+    pub config_dir: PathBuf,
+    pub private_mix_key: PathBuf,
+    pub public_mix_key: PathBuf,
+}
+
+impl Pathfinder {
+    pub fn new(id: String) -> Pathfinder {
+        let os_config_dir = dirs::config_dir().unwrap(); // grabs the OS default config dir
+        let config_dir = os_config_dir.join("nym").join("clients").join(id);
+        let private_mix_key = config_dir.join("private.pem");
+        let public_mix_key = config_dir.join("public.pem");
+        Pathfinder {
+            config_dir,
+            private_mix_key,
+            public_mix_key,
+        }
+    }
+}

src/persistence/pemstore.rs

@@ -0,0 +1,72 @@
+use crate::identity::mixnet::KeyPair;
+use crate::persistence::pathfinder::Pathfinder;
+use pem::{encode, parse, Pem};
+use std::fs::File;
+use std::io::prelude::*;
+use std::path::PathBuf;
+
+pub fn read_keypair_from_disk(id: String) -> KeyPair {
+    let pathfinder = Pathfinder::new(id);
+    let pem_store = PemStore::new(pathfinder);
+    let keypair = pem_store.read();
+    keypair
+}
+
+pub struct PemStore {
+    config_dir: PathBuf,
+    private_mix_key: PathBuf,
+    public_mix_key: PathBuf,
+}
+
+impl PemStore {
+    pub fn new(pathfinder: Pathfinder) -> PemStore {
+        PemStore {
+            config_dir: pathfinder.config_dir,
+            private_mix_key: pathfinder.private_mix_key,
+            public_mix_key: pathfinder.public_mix_key,
+        }
+    }
+
+    pub fn read(&self) -> KeyPair {
+        let private = self.read_file(self.private_mix_key.clone());
+        let public = self.read_file(self.public_mix_key.clone());
+
+        KeyPair::from_bytes(private, public)
+    }
+
+    fn read_file(&self, filepath: PathBuf) -> Vec<u8> {
+        let mut pem_bytes = File::open(filepath).unwrap();
+        let mut buf = Vec::new();
+        pem_bytes.read_to_end(&mut buf).unwrap();
+        let pem = parse(&buf).unwrap();
+        pem.contents
+    }
+    // This should be refactored and made more generic for when we have other kinds of
+    // KeyPairs that we want to persist (e.g. validator keypairs, or keys for
+    // signing vs encryption). However, for the moment, it does the job.
+    pub fn write(&self, key_pair: KeyPair) {
+        std::fs::create_dir_all(self.config_dir.clone()).unwrap();
+
+        self.write_pem_file(
+            self.private_mix_key.clone(),
+            key_pair.private_bytes(),
+            String::from("SPHINX CURVE25519 PRIVATE KEY"),
+        );
+        self.write_pem_file(
+            self.public_mix_key.clone(),
+            key_pair.public_bytes(),
+            String::from("SPHINX CURVE25519 PUBLIC KEY"),
+        );
+    }
+
+    fn write_pem_file(&self, filepath: PathBuf, data: [u8; 32], tag: String) {
+        let pem = Pem {
+            tag,
+            contents: data.to_vec(),
+        };
+        let key = encode(&pem);
+
+        let mut file = File::create(filepath).unwrap();
+        file.write_all(key.as_bytes()).unwrap();
+    }
+}

src/persistence/toml.rs

@@ -0,0 +1 @@
+// TODO: we can put all the TOML config templating code in here once we get to that.

src/sockets/mod.rs

@@ -0,0 +1,2 @@
+pub mod tcp;
+pub mod ws;

src/sockets/tcp.rs

@@ -0,0 +1,305 @@
+use crate::clients::directory::presence::Topology;
+use crate::clients::BufferResponse;
+use crate::clients::InputMessage;
+use futures::channel::{mpsc, oneshot};
+use futures::future::FutureExt;
+use futures::io::Error;
+use futures::SinkExt;
+use sphinx::route::{Destination, DestinationAddressBytes};
+use std::io;
+use std::net::SocketAddr;
+use tokio::prelude::*;
+use std::convert::TryFrom;
+use std::sync::Arc;
+use std::borrow::Borrow;
+
+const SEND_REQUEST_PREFIX: u8 = 1;
+const FETCH_REQUEST_PREFIX: u8 = 2;
+const GET_CLIENTS_REQUEST_PREFIX: u8 = 3;
+const OWN_DETAILS_REQUEST_PREFIX: u8 = 4;
+
+#[derive(Debug)]
+pub enum TCPSocketError {
+    FailedToStartSocketError,
+    UnknownSocketError,
+    IncompleteDataError,
+    UnknownRequestError,
+}
+
+impl From<io::Error> for TCPSocketError {
+    fn from(err: Error) -> Self {
+        use TCPSocketError::*;
+        match err.kind() {
+            io::ErrorKind::ConnectionRefused => FailedToStartSocketError,
+            io::ErrorKind::ConnectionReset => FailedToStartSocketError,
+            io::ErrorKind::ConnectionAborted => FailedToStartSocketError,
+            io::ErrorKind::NotConnected => FailedToStartSocketError,
+
+            io::ErrorKind::AddrInUse => FailedToStartSocketError,
+            io::ErrorKind::AddrNotAvailable => FailedToStartSocketError,
+            _ => UnknownSocketError,
+        }
+    }
+}
+
+
+enum ClientRequest {
+    Send {
+        message: Vec<u8>,
+        recipient_address: DestinationAddressBytes,
+    },
+    Fetch,
+    GetClients,
+    OwnDetails,
+}
+
+impl TryFrom<&[u8]> for ClientRequest {
+    type Error = TCPSocketError;
+
+    fn try_from(data: &[u8]) -> Result<Self, Self::Error> {
+        use TCPSocketError::*;
+        if data.is_empty() {
+            return Err(IncompleteDataError);
+        }
+
+        match data[0] {
+            SEND_REQUEST_PREFIX => parse_send_request(data),
+            FETCH_REQUEST_PREFIX => Ok(ClientRequest::Fetch),
+            GET_CLIENTS_REQUEST_PREFIX => Ok(ClientRequest::GetClients),
+            OWN_DETAILS_REQUEST_PREFIX => Ok(ClientRequest::OwnDetails),
+            _ => Err(UnknownRequestError)
+        }
+    }
+}
+
+fn parse_send_request(data: &[u8]) -> Result<ClientRequest, TCPSocketError> {
+    if data.len() < 1 + 32 + 1 {
+        // make sure it has the prefix, destination and at least single byte of data
+        return Err(TCPSocketError::IncompleteDataError);
+    }
+
+    let mut recipient_address = [0u8; 32];
+    recipient_address.copy_from_slice(&data[1..33]);
+
+    let message = data[33..].to_vec();
+
+    Ok(ClientRequest::Send {
+        message,
+        recipient_address,
+    })
+}
+
+impl ClientRequest {
+    async fn handle_send(
+        msg: Vec<u8>,
+        recipient_address: DestinationAddressBytes,
+        mut input_tx: mpsc::UnboundedSender<InputMessage>,
+    ) -> ServerResponse {
+        println!("send handle. sending to: {:?}, msg: {:?}", recipient_address, msg);
+        let dummy_surb = [0; 16];
+        let input_msg = InputMessage(Destination::new(recipient_address, dummy_surb), msg);
+
+        println!("ALMOST ABOUT TO SOMEDAY SEND {:?}", input_msg);
+        input_tx.send(input_msg).await.unwrap();
+
+        ServerResponse::Send
+    }
+
+    async fn handle_fetch(mut msg_query: mpsc::UnboundedSender<BufferResponse>) -> ServerResponse {
+        println!("fetch handle");
+        let (res_tx, res_rx) = oneshot::channel();
+        if msg_query.send(res_tx).await.is_err() {
+            return ServerResponse::Error {
+                message: "Server failed to receive messages".to_string(),
+            };
+        }
+
+        let messages = res_rx.map(|msg| msg).await;
+
+        if messages.is_err() {
+            return ServerResponse::Error {
+                message: "Server failed to receive messages".to_string(),
+            };
+        }
+
+        let messages = messages.unwrap();
+        println!("fetched {} messages", messages.len());
+        ServerResponse::Fetch {
+            messages,
+        }
+    }
+
+    async fn handle_get_clients(topology: &Topology) -> ServerResponse {
+        println!("get clients handle");
+        let clients = topology
+            .mix_provider_nodes
+            .iter()
+            .flat_map(|provider| provider.registered_clients.iter())
+            .map(|client| base64::decode_config(&client.pub_key, base64::URL_SAFE).unwrap()) // TODO: this can potentially throw an error
+            .collect();
+        ServerResponse::GetClients { clients }
+    }
+
+    async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse {
+        println!("own details handle");
+        ServerResponse::OwnDetails {
+            address: self_address_bytes.to_vec(),
+        }
+    }
+}
+
+
+enum ServerResponse {
+    Send,
+    Fetch { messages: Vec<Vec<u8>> },
+    GetClients { clients: Vec<Vec<u8>> },
+    OwnDetails { address: Vec<u8> },
+    Error { message: String },
+}
+
+impl Into<Vec<u8>> for ServerResponse {
+    fn into(self) -> Vec<u8> {
+        match self {
+            ServerResponse::Send => b"ok".to_vec(),
+            ServerResponse::Fetch {messages} =>encode_fetched_messages(messages),
+            ServerResponse::GetClients {clients} => encode_list_of_clients(clients),
+            ServerResponse::OwnDetails {address} => address,
+            ServerResponse::Error { message } => message.as_bytes().to_vec()
+        }
+    }
+}
+
+// num_msgs || len1 || len2 || ... || msg1 || msg2 || ...
+fn encode_fetched_messages(messages: Vec<Vec<u8>>) -> Vec<u8> {
+    // for reciprocal of this look into sfw-provider-requests::responses::PullResponse::from_bytes()
+
+    let num_msgs = messages.len() as u16;
+    let msgs_lens: Vec<u16> = messages.iter().map(|msg| msg.len() as u16).collect();
+
+    num_msgs
+        .to_be_bytes()
+        .to_vec()
+        .into_iter()
+        .chain(
+            msgs_lens
+                .into_iter()
+                .flat_map(|len| len.to_be_bytes().to_vec().into_iter()),
+        )
+        .chain(messages.iter().flat_map(|msg| msg.clone().into_iter()))
+        .collect()
+}
+
+fn encode_list_of_clients(clients: Vec<Vec<u8>>) -> Vec<u8> {
+    println!("clients: {:?}", clients);
+    // we can just concat all clients since all of them got to be 32 bytes long
+    // (if not, then we have bigger problem somewhere up the line)
+
+    // converts [[1,2,3],[4,5,6],...] into [1,2,3,4,5,6,...]
+    clients.into_iter().flatten().collect()
+}
+
+impl ServerResponse {
+    fn new_error(message: String) -> ServerResponse {
+        ServerResponse::Error { message }
+    }
+}
+
+
+async fn handle_connection(data: &[u8], request_handling_data: RequestHandlingData) -> Result<ServerResponse, TCPSocketError> {
+    let request = ClientRequest::try_from(data)?;
+    let response = match request {
+        ClientRequest::Send {
+            message,
+            recipient_address
+        } => ClientRequest::handle_send(message, recipient_address, request_handling_data.msg_input).await,
+        ClientRequest::Fetch => ClientRequest::handle_fetch(request_handling_data.msg_query).await,
+        ClientRequest::GetClients => ClientRequest::handle_get_clients(request_handling_data.topology.borrow()).await,
+        ClientRequest::OwnDetails => ClientRequest::handle_own_details(request_handling_data.self_address).await,
+    };
+
+    Ok(response)
+}
+
+struct RequestHandlingData {
+    msg_input: mpsc::UnboundedSender<InputMessage>,
+    msg_query: mpsc::UnboundedSender<BufferResponse>,
+    self_address: DestinationAddressBytes,
+    topology: Arc<Topology>,
+}
+
+async fn accept_connection(
+    mut socket: tokio::net::TcpStream,
+    msg_input: mpsc::UnboundedSender<InputMessage>,
+    msg_query: mpsc::UnboundedSender<BufferResponse>,
+    self_address: DestinationAddressBytes,
+    topology: Topology,
+) {
+    let address = socket
+        .peer_addr()
+        .expect("connected streams should have a peer address");
+    println!("Peer address: {}", address);
+
+    let topology = Arc::new(topology);
+
+    let mut buf = [0u8; 2048];
+
+    // In a loop, read data from the socket and write the data back.
+    loop {
+        // TODO: shutdowns?
+
+        let response = match socket.read(&mut buf).await {
+            // socket closed
+            Ok(n) if n == 0 => {
+                println!("Remote connection closed.");
+                return;
+            }
+            Ok(n) => {
+                let request_handling_data = RequestHandlingData {
+                    topology: topology.clone(),
+                    msg_input: msg_input.clone(),
+                    msg_query: msg_query.clone(),
+                    self_address: self_address.clone(),
+                };
+                match handle_connection(&buf[..n], request_handling_data).await {
+                    Ok(res) => res,
+                    Err(e) => ServerResponse::new_error(format!("{:?}", e))
+                }
+            }
+            Err(e) => {
+                eprintln!("failed to read from socket; err = {:?}", e);
+                return;
+            }
+        };
+
+        let response_vec: Vec<u8> = response.into();
+        if let Err(e) = socket.write_all(&response_vec).await {
+            eprintln!("failed to write reply to socket; err = {:?}", e);
+            return;
+        }
+    }
+}
+
+pub async fn start_tcpsocket(
+    address: SocketAddr,
+    message_tx: mpsc::UnboundedSender<InputMessage>,
+    received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
+    self_address: DestinationAddressBytes,
+    topology: Topology,
+) -> Result<(), TCPSocketError> {
+    let mut listener = tokio::net::TcpListener::bind(address).await?;
+
+    while let Ok((stream, _)) = listener.accept().await {
+        // it's fine to be cloning the channel on all new connection, because in principle
+        // this server should only EVER have a single client connected
+        tokio::spawn(accept_connection(
+            stream,
+            message_tx.clone(),
+            received_messages_query_tx.clone(),
+            self_address,
+            topology.clone(),
+        ));
+    }
+
+    eprintln!("The tcpsocket went kaput...");
+    Ok(())
+}

src/sockets/ws.rs

@@ -0,0 +1,260 @@
+use crate::clients::directory::presence::Topology;
+use crate::clients::BufferResponse;
+use crate::clients::InputMessage;
+use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender};
+use futures::channel::{mpsc, oneshot};
+use futures::future::FutureExt;
+use futures::io::Error;
+use futures::{SinkExt, StreamExt};
+use serde::{Deserialize, Serialize};
+use sphinx::route::{Destination, DestinationAddressBytes};
+use std::io;
+use std::net::SocketAddr;
+use tungstenite::protocol::Message;
+
+struct Connection {
+    address: SocketAddr,
+    msg_input: mpsc::UnboundedSender<InputMessage>,
+    msg_query: mpsc::UnboundedSender<BufferResponse>,
+    rx: UnboundedReceiver<Message>,
+    self_address: DestinationAddressBytes,
+    topology: Topology,
+    tx: UnboundedSender<Message>,
+}
+
+#[derive(Debug)]
+pub enum WebSocketError {
+    FailedToStartSocketError,
+    UnknownSocketError,
+}
+
+impl From<io::Error> for WebSocketError {
+    fn from(err: Error) -> Self {
+        use WebSocketError::*;
+        match err.kind() {
+            io::ErrorKind::ConnectionRefused => FailedToStartSocketError,
+            io::ErrorKind::ConnectionReset => FailedToStartSocketError,
+            io::ErrorKind::ConnectionAborted => FailedToStartSocketError,
+            io::ErrorKind::NotConnected => FailedToStartSocketError,
+
+            io::ErrorKind::AddrInUse => FailedToStartSocketError,
+            io::ErrorKind::AddrNotAvailable => FailedToStartSocketError,
+            _ => UnknownSocketError,
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(tag = "type", rename_all = "camelCase")]
+enum ClientRequest {
+    Send {
+        message: String,
+        recipient_address: String,
+    },
+    Fetch,
+    GetClients,
+    OwnDetails,
+}
+
+impl From<Message> for ClientRequest {
+    fn from(msg: Message) -> Self {
+        let text_msg = match msg {
+            Message::Text(msg) => msg,
+            Message::Binary(_) => panic!("binary messages are not supported!"),
+            Message::Close(_) => panic!("todo: handle close!"),
+            _ => panic!("Other types of messages are also unsupported!"),
+        };
+        serde_json::from_str(&text_msg).unwrap()
+    }
+}
+
+impl ClientRequest {
+    async fn handle_send(
+        msg: String,
+        recipient_address: String,
+        mut input_tx: mpsc::UnboundedSender<InputMessage>,
+    ) -> ServerResponse {
+        let address_vec = match base64::decode_config(&recipient_address, base64::URL_SAFE) {
+            Err(e) => {
+                return ServerResponse::Error {
+                    message: e.to_string(),
+                }
+            }
+            Ok(hex) => hex,
+        };
+
+        if address_vec.len() != 32 {
+            return ServerResponse::Error {
+                message: "InvalidDestinationLength".to_string(),
+            };
+        }
+
+        let mut address = [0; 32];
+        address.copy_from_slice(&address_vec);
+
+        let dummy_surb = [0; 16];
+
+        let input_msg = InputMessage(Destination::new(address, dummy_surb), msg.into_bytes());
+
+        println!("ALMOST ABOUT TO SOMEDAY SEND {:?}", input_msg);
+        input_tx.send(input_msg).await.unwrap();
+
+        ServerResponse::Send
+    }
+
+    async fn handle_fetch(mut msg_query: mpsc::UnboundedSender<BufferResponse>) -> ServerResponse {
+        let (res_tx, res_rx) = oneshot::channel();
+        if msg_query.send(res_tx).await.is_err() {
+            return ServerResponse::Error {
+                message: "Server failed to receive messages".to_string(),
+            };
+        }
+
+        let messages = res_rx.map(|msg| msg).await;
+
+        if messages.is_err() {
+            return ServerResponse::Error {
+                message: "Server failed to receive messages".to_string(),
+            };
+        }
+
+        let messages = messages.unwrap();
+        let messages_as_b64 = messages
+            .iter()
+            .map(|message| base64::encode_config(message, base64::URL_SAFE))
+            .collect();
+        ServerResponse::Fetch {
+            messages: messages_as_b64,
+        }
+    }
+
+    async fn handle_get_clients(topology: Topology) -> ServerResponse {
+        let clients = topology
+            .mix_provider_nodes
+            .into_iter()
+            .flat_map(|provider| provider.registered_clients.into_iter())
+            .map(|client| client.pub_key)
+            .collect();
+        ServerResponse::GetClients { clients }
+    }
+
+    async fn handle_own_details(self_address_bytes: DestinationAddressBytes) -> ServerResponse {
+        let self_address = base64::encode_config(&self_address_bytes, base64::URL_SAFE);
+        ServerResponse::OwnDetails {
+            address: self_address,
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(tag = "type", rename_all = "camelCase")]
+enum ServerResponse {
+    Send,
+    Fetch { messages: Vec<String> },
+    GetClients { clients: Vec<String> },
+    OwnDetails { address: String },
+    Error { message: String },
+}
+
+impl Into<Message> for ServerResponse {
+    fn into(self) -> Message {
+        let str_res = serde_json::to_string(&self).unwrap();
+        Message::Text(str_res)
+    }
+}
+
+async fn handle_connection(conn: Connection) {
+    let mut conn = conn;
+    while let Some(msg) = conn.rx.next().await {
+        println!("Received a message from {}: {}", conn.address, msg);
+        let request: ClientRequest = msg.into();
+
+        let response = match request {
+            ClientRequest::Send {
+                message,
+                recipient_address,
+            } => {
+                ClientRequest::handle_send(message, recipient_address, conn.msg_input.clone()).await
+            }
+            ClientRequest::Fetch => ClientRequest::handle_fetch(conn.msg_query.clone()).await,
+            ClientRequest::GetClients => {
+                ClientRequest::handle_get_clients(conn.topology.clone()).await
+            }
+            ClientRequest::OwnDetails => ClientRequest::handle_own_details(conn.self_address).await,
+        };
+
+        conn.tx
+            .unbounded_send(response.into())
+            .expect("Failed to forward message");
+    }
+}
+
+async fn accept_connection(
+    stream: tokio::net::TcpStream,
+    msg_input: mpsc::UnboundedSender<InputMessage>,
+    msg_query: mpsc::UnboundedSender<BufferResponse>,
+    self_address: DestinationAddressBytes,
+    topology: Topology,
+) {
+    let address = stream
+        .peer_addr()
+        .expect("connected streams should have a peer address");
+    println!("Peer address: {}", address);
+
+    let mut ws_stream = tokio_tungstenite::accept_async(stream)
+        .await
+        .expect("Error during the websocket handshake occurred");
+
+    println!("New WebSocket connection: {}", address);
+
+    // Create a channel for our stream, which other sockets will use to
+    // send us messages. Then register our address with the stream to send
+    // data to us.
+    let (msg_tx, msg_rx) = futures::channel::mpsc::unbounded();
+    let (response_tx, mut response_rx) = futures::channel::mpsc::unbounded();
+    let conn = Connection {
+        address,
+        rx: msg_rx,
+        tx: response_tx,
+        topology,
+        msg_input,
+        msg_query,
+        self_address,
+    };
+    tokio::spawn(handle_connection(conn));
+
+    while let Some(message) = ws_stream.next().await {
+        let message = message.expect("Failed to get request");
+        msg_tx
+            .unbounded_send(message)
+            .expect("Failed to forward request");
+        if let Some(resp) = response_rx.next().await {
+            ws_stream.send(resp).await.expect("Failed to send response");
+        }
+    }
+}
+
+pub async fn start_websocket(
+    address: SocketAddr,
+    message_tx: mpsc::UnboundedSender<InputMessage>,
+    received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
+    self_address: DestinationAddressBytes,
+    topology: Topology,
+) -> Result<(), WebSocketError> {
+    let mut listener = tokio::net::TcpListener::bind(address).await?;
+
+    while let Ok((stream, _)) = listener.accept().await {
+        // it's fine to be cloning the channel on all new connection, because in principle
+        // this server should only EVER have a single client connected
+        tokio::spawn(accept_connection(
+            stream,
+            message_tx.clone(),
+            received_messages_query_tx.clone(),
+            self_address,
+            topology.clone(),
+        ));
+    }
+
+    eprintln!("The websocket went kaput...");
+    Ok(())
+}

src/utils/addressing.rs

@@ -0,0 +1,80 @@
+use std::convert::{TryFrom, TryInto};
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+
+pub enum AddressType {
+    V4,
+    V6,
+}
+
+impl Into<u8> for AddressType {
+    fn into(self) -> u8 {
+        use AddressType::*;
+
+        match self {
+            V4 => 4,
+            V6 => 6,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub enum AddressTypeError {
+    InvalidPrefixError,
+}
+
+impl TryFrom<u8> for AddressType {
+    type Error = AddressTypeError;
+
+    fn try_from(value: u8) -> Result<Self, Self::Error> {
+        use AddressType::*;
+        use AddressTypeError::*;
+
+        match value {
+            4 => Ok(V4),
+            6 => Ok(V6),
+            _ => Err(InvalidPrefixError),
+        }
+    }
+}
+
+/// FLAG || port || octets || zeropad
+pub fn encoded_bytes_from_socket_address(address: SocketAddr) -> [u8; 32] {
+    let port_bytes = address.port().to_be_bytes();
+
+    let encoded_host: Vec<u8> = match address.ip() {
+        IpAddr::V4(ip) => std::iter::once(AddressType::V4.into())
+            .chain(port_bytes.iter().cloned())
+            .chain(ip.octets().iter().cloned())
+            .chain(std::iter::repeat(0))
+            .take(32)
+            .collect(),
+        IpAddr::V6(ip) => std::iter::once(AddressType::V6.into())
+            .chain(port_bytes.iter().cloned())
+            .chain(ip.octets().iter().cloned())
+            .chain(std::iter::repeat(0))
+            .take(32)
+            .collect(),
+    };
+
+    let mut address_bytes = [0u8; 32];
+    address_bytes.copy_from_slice(&encoded_host[..32]);
+
+    address_bytes
+}
+
+pub fn socket_address_from_encoded_bytes(b: [u8; 32]) -> SocketAddr {
+    let address_type: AddressType = b[0].try_into().unwrap();
+
+    let port: u16 = u16::from_be_bytes([b[1], b[2]]);
+
+    let ip = match address_type {
+        AddressType::V4 => IpAddr::V4(Ipv4Addr::new(b[3], b[4], b[5], b[6])),
+        AddressType::V6 => {
+            let mut address_octets = [0u8; 16];
+            address_octets.copy_from_slice(&b[3..19]);
+            IpAddr::V6(Ipv6Addr::from(address_octets))
+        }
+    };
+
+    SocketAddr::new(ip, port)
+}

src/utils/bytes.rs

@@ -0,0 +1,63 @@
+pub fn zero_pad_to_32(mut bytes: Vec<u8>) -> [u8; 32] {
+    assert!(bytes.len() <= 32);
+    if bytes.len() != 32 {
+        bytes.resize(32, 0);
+    }
+    let mut padded_bytes = [0; 32];
+    padded_bytes.copy_from_slice(&bytes[..]);
+    padded_bytes
+}
+
+#[cfg(test)]
+mod zero_padding_to_32_bytes {
+    use super::*;
+
+    #[cfg(test)]
+    mod with_empty_input {
+        use super::*;
+
+        #[test]
+        fn it_returns_32_zeros() {
+            let input = vec![];
+            let result = zero_pad_to_32(input);
+            assert_eq!([0u8; 32], result);
+        }
+    }
+
+    #[cfg(test)]
+    mod with_all_bytes_set_to_1 {
+        use super::*;
+        #[test]
+        fn it_returns_32_ones() {
+            let input = vec![1u8; 32];
+            let result = zero_pad_to_32(input);
+            assert_eq!([1u8; 32], result);
+        }
+    }
+
+    #[cfg(test)]
+    mod with_3_bytes_set {
+        use super::*;
+        #[test]
+        fn it_returns_input_zero_padded_to_32_bytes() {
+            let input = vec![1u8; 3];
+            let result = zero_pad_to_32(input);
+            let expected_content = vec![
+                1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+                0, 0, 0, 0,
+            ];
+            assert_eq!(expected_content, result.to_vec());
+        }
+    }
+
+    #[cfg(test)]
+    mod with_oversized_input {
+        use super::*;
+        #[test]
+        #[should_panic]
+        fn it_panics() {
+            let input = vec![1u8; 33];
+            zero_pad_to_32(input);
+        }
+    }
+}

src/utils/mod.rs

@@ -0,0 +1,5 @@
+pub mod addressing;
+pub mod bytes;
+pub mod poisson;
+pub mod sphinx;
+pub mod topology;

src/utils/poisson.rs

@@ -0,0 +1,6 @@
+use rand_distr::{Distribution, Exp};
+
+pub fn sample(average_delay: f64) -> f64 {
+    let exp = Exp::new(1.0 / average_delay).unwrap();
+    exp.sample(&mut rand::thread_rng())
+}

src/utils/sphinx.rs

@@ -0,0 +1,48 @@
+use crate::clients::directory::presence::Topology;
+use crate::utils::{addressing, bytes, topology};
+use curve25519_dalek::montgomery::MontgomeryPoint;
+use sphinx::route::{Destination, DestinationAddressBytes, Node, NodeAddressBytes, SURBIdentifier};
+use sphinx::SphinxPacket;
+use std::net::SocketAddr;
+
+pub const LOOP_COVER_MESSAGE_PAYLOAD: &[u8] = b"The cake is a lie!";
+
+pub fn loop_cover_message(
+    our_address: DestinationAddressBytes,
+    surb_id: SURBIdentifier,
+    topology: &Topology,
+) -> (SocketAddr, SphinxPacket) {
+    let destination = Destination::new(our_address, surb_id);
+
+    encapsulate_message(destination, LOOP_COVER_MESSAGE_PAYLOAD.to_vec(), topology)
+}
+
+pub fn encapsulate_message(
+    recipient: Destination,
+    message: Vec<u8>,
+    topology: &Topology,
+) -> (SocketAddr, SphinxPacket) {
+    let mixes_route = topology::route_from(&topology);
+    let first_provider = topology.mix_provider_nodes.first().unwrap();
+    let decoded_key_bytes =
+        base64::decode_config(&first_provider.pub_key, base64::URL_SAFE).unwrap();
+    let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes);
+    let key = MontgomeryPoint(key_bytes);
+
+    let provider = Node::new(
+        addressing::encoded_bytes_from_socket_address(first_provider.host.clone().parse().unwrap()),
+        key,
+    );
+
+    let route = [mixes_route, vec![provider]].concat();
+
+    let delays = sphinx::header::delays::generate(route.len());
+
+    // build the packet
+    let packet = sphinx::SphinxPacket::new(message, &route[..], &recipient, &delays).unwrap();
+
+    let first_node_address =
+        addressing::socket_address_from_encoded_bytes(route.first().unwrap().address);
+
+    (first_node_address, packet)
+}

src/utils/topology.rs

@@ -0,0 +1,58 @@
+use crate::clients::directory;
+use crate::clients::directory::presence::MixNodePresence;
+use crate::clients::directory::presence::Topology;
+use crate::clients::directory::requests::presence_topology_get::PresenceTopologyGetRequester;
+use crate::clients::directory::DirectoryClient;
+use crate::utils::{addressing, bytes};
+use curve25519_dalek::montgomery::MontgomeryPoint;
+use rand::seq::SliceRandom;
+use sphinx::route::Node as SphinxNode;
+use std::collections::HashMap;
+use std::net::SocketAddr;
+
+pub(crate) fn get_topology(directory_server: String) -> Topology {
+    println!("Using directory server: {:?}", directory_server);
+    let directory_config = directory::Config {
+        base_url: directory_server,
+    };
+    let directory = directory::Client::new(directory_config);
+
+    let topology = directory
+        .presence_topology
+        .get()
+        .expect("Failed to retrieve network topology.");
+    topology
+}
+
+pub(crate) fn route_from(topology: &Topology) -> Vec<SphinxNode> {
+    let mut layered_topology: HashMap<u64, Vec<MixNodePresence>> = HashMap::new();
+    let mixes = topology.mix_nodes.iter();
+    for mix in mixes {
+        let layer_nodes = layered_topology.entry(mix.layer).or_insert(Vec::new());
+        layer_nodes.push(mix.clone());
+    }
+
+    let num_layers = layered_topology.len() as u64;
+    let mut route = vec![];
+
+    for x in 1..=num_layers {
+        let nodes = &layered_topology[&x];
+        let the_node = nodes.choose(&mut rand::thread_rng()).unwrap();
+        route.push(the_node);
+    }
+
+    route
+        .iter()
+        .map(|mix| {
+            let address_bytes =
+                addressing::encoded_bytes_from_socket_address(mix.host.clone().parse().unwrap());
+            let decoded_key_bytes = base64::decode_config(&mix.pub_key, base64::URL_SAFE).unwrap();
+            let key_bytes = bytes::zero_pad_to_32(decoded_key_bytes);
+            let key = MontgomeryPoint(key_bytes);
+            SphinxNode {
+                address: address_bytes,
+                pub_key: key,
+            }
+        })
+        .collect()
+}