Merge pull request #18 from nymtech/feature/provider_polling

Feature/provider polling

Cargo.lock

@@ -941,6 +941,7 @@ dependencies = [
  "reqwest 0.9.22 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde 1.0.103 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)",
+ "sfw-provider-requests 0.1.0",
  "sphinx 0.1.0",
  "tokio 0.2.4 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -1402,6 +1403,13 @@ dependencies = [
  "url 1.7.2 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "sfw-provider-requests"
+version = "0.1.0"
+dependencies = [
+ "sphinx 0.1.0",
+]
+
 [[package]]
 name = "sha2"
 version = "0.8.0"

Cargo.toml

@@ -15,6 +15,8 @@ pem = "0.7.0"
 reqwest = "0.9.22"
 serde = { version = "1.0", features = ["derive"] }
 sphinx = { path = "../sphinx" }
+sfw-provider-requests = { path = "../nym-sfw-provider/sfw-provider-requests" }
+
 tokio = { version = "0.2", features = ["full"] }
 
 [dev-dependencies]

src/clients/provider.rs

@@ -0,0 +1,47 @@
+use sphinx::route::Node as MixNode;
+use sphinx::SphinxPacket;
+use tokio::prelude::*;
+use sfw_provider_requests::*;
+use std::net::Shutdown;
+use std::time::Duration;
+use sfw_provider_requests::requests::{PullRequest, ProviderRequest};
+use sfw_provider_requests::responses::{PullResponse, ProviderResponse};
+
+pub struct ProviderClient {}
+
+
+impl ProviderClient {
+    pub fn new() -> Self {
+        ProviderClient {}
+    }
+
+    pub async fn retrieve_messages(
+        &self,
+//        provider: &MixNode,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let address = [42; 32];
+        let pull_request = PullRequest::new(address);
+        let bytes = pull_request.to_bytes();
+
+        let mut socket = tokio::net::TcpStream::connect("127.0.0.1:9000").await?;
+        println!("keep alive: {:?}", socket.keepalive());
+        socket.set_keepalive(Some(tokio::time::Duration::from_secs(2)));
+        socket.write_all(&bytes[..]).await?;
+        if let Err(e) = socket.shutdown(Shutdown::Write) {
+            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) {
+            eprintln!("failed to close read part of the socket; err = {:?}", e)
+        }
+
+        println!("Received the following response: {:?}", response);
+        let parsed_response = PullResponse::from_bytes(&response).unwrap();
+        for message in parsed_response.messages {
+            println!("Received: {:?}", String::from_utf8(message).unwrap())
+        }
+        Ok(())
+    }
+}

src/commands/run.rs

@@ -3,6 +3,7 @@ use crate::clients::directory::presence::{MixNodePresence, Topology};
 use crate::clients::directory::requests::presence_topology_get::PresenceTopologyGetRequester;
 use crate::clients::directory::DirectoryClient;
 use crate::clients::mix::MixClient;
+use crate::clients::provider::ProviderClient;
 use crate::utils::bytes;
 use base64;
 use clap::ArgMatches;
@@ -35,7 +36,7 @@ pub fn execute(matches: &ArgMatches) {
             interval.tick().await;
             let message = format!("Hello, Sphinx {}", i).as_bytes().to_vec();
 
-            let route_len = 3;
+            let route_len = 2;
 
             // data needed to generate a new Sphinx packet
             let route = route_from(&topology, route_len);
@@ -51,6 +52,14 @@ pub fn execute(matches: &ArgMatches) {
             let result = mix_client.send(packet, route.first().unwrap()).await;
             println!("packet sent:  {:?}", i);
             i += 1;
+
+            // retrieve messages every now and then
+            if i % 3 == 0 {
+                interval.tick().await;
+                println!("going to retrieve messages!");
+                let provider_client = ProviderClient::new();
+                provider_client.retrieve_messages().await.unwrap();
+            }
         }
     })
 }
@@ -76,10 +85,13 @@ fn route_from(topology: &Topology, route_len: usize) -> Vec<SphinxNode> {
         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);
-        let sphinx_node = SphinxNode {
+        let mut sphinx_node = SphinxNode {
             address: address_bytes,
             pub_key: key,
         };
+
+        // temporary to make it work locally:
+        sphinx_node.pub_key = Default::default();
         route.push(sphinx_node);
     }
     route
@@ -88,7 +100,7 @@ fn route_from(topology: &Topology, route_len: usize) -> Vec<SphinxNode> {
 // TODO: where do we retrieve this guy from?
 fn get_destination() -> Destination {
     Destination {
-        address: [0u8; 32],
-        identifier: [0u8; 16],
+        address: [42u8; 32],
+        identifier: [1u8; 16],
     }
 }