Merge branch 'master' of github.com:darkrenaissance/darkfi

Cargo.toml

@@ -127,8 +127,8 @@ name = "compile-shaders"
 path = "src/bin/compile-shaders.rs"
 
 [[bin]]
-name = "demoservices"
-path = "src/bin/demoservices.rs"
+name = "gatewayd"
+path = "src/bin/gatewayd.rs"
 
 [[bin]]
 name = "demowallet"

src/bin/demoservices.rs

@@ -1,37 +0,0 @@
-use async_executor::Executor;
-use easy_parallel::Parallel;
-use std::sync::Arc;
-
-use drk::Result;
-
-use drk::service::GatewayService;
-
-async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
-    let gateway = GatewayService::new(
-        String::from("tcp://127.0.0.1:3333"),
-        String::from("tcp://127.0.0.1:4444"),
-    );
-
-    gateway.start(executor.clone()).await?;
-    Ok(())
-}
-
-fn main() -> Result<()> {
-    let ex = Arc::new(Executor::new());
-    let (signal, shutdown) = async_channel::unbounded::<()>();
-    let ex2 = ex.clone();
-
-    let (_, result) = Parallel::new()
-        // Run four executor threads.
-        .each(0..3, |_| smol::future::block_on(ex.run(shutdown.recv())))
-        // Run the main future on the current thread.
-        .finish(|| {
-            smol::future::block_on(async move {
-                start(ex2).await?;
-                drop(signal);
-                Ok::<(), drk::Error>(())
-            })
-        });
-
-    result
-}

src/bin/demowallet.rs

@@ -6,16 +6,14 @@ use drk::service::{fetch_slabs_loop, GatewayClient};
 use drk::Result;
 
 async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
-    let mut client = GatewayClient::new(String::from("tcp://127.0.0.1:3333"));
+    let mut client = GatewayClient::new("127.0.0.1:3333".parse()?);
 
     client.start().await?;
     println!("connected to a server");
 
     let slabs = Arc::new(Mutex::new(vec![]));
 
-    let subscriber = client
-        .subscribe(String::from("tcp://127.0.0.1:4444"))
-        .await?;
+    let subscriber = client.subscribe("127.0.0.1:4444".parse()?).await?;
 
     println!("subscription ready");
 

src/bin/gatewayd.rs

@@ -0,0 +1,70 @@
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+extern crate clap;
+use async_executor::Executor;
+use easy_parallel::Parallel;
+
+use drk::Result;
+
+use drk::service::{GatewayService, ProgramOptions};
+
+fn setup_addr(address: Option<SocketAddr>, default: SocketAddr) -> SocketAddr {
+    match address {
+        Some(addr) => addr,
+        None => default,
+    }
+}
+
+async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<()> {
+    let accept_addr: SocketAddr = setup_addr(options.accept_addr, "127.0.0.1:3333".parse()?);
+    let pub_addr: SocketAddr = setup_addr(options.pub_addr, "127.0.0.1:4444".parse()?);
+
+    let gateway = GatewayService::new(accept_addr, pub_addr);
+
+    gateway.start(executor.clone()).await?;
+    Ok(())
+}
+
+fn main() -> Result<()> {
+    use simplelog::*;
+
+    let ex = Arc::new(Executor::new());
+    let (signal, shutdown) = async_channel::unbounded::<()>();
+
+    let options = ProgramOptions::load()?;
+
+    let logger_config = ConfigBuilder::new().set_time_format_str("%T%.6f").build();
+
+    let debug_level = if options.verbose {
+        LevelFilter::Debug
+    } else {
+        LevelFilter::Off
+    };
+
+    CombinedLogger::init(vec![
+        TermLogger::new(debug_level, logger_config, TerminalMode::Mixed).unwrap(),
+        WriteLogger::new(
+            LevelFilter::Debug,
+            Config::default(),
+            std::fs::File::create(options.log_path.as_path()).unwrap(),
+        ),
+    ])
+    .unwrap();
+
+    let ex2 = ex.clone();
+
+    let (_, result) = Parallel::new()
+        // Run four executor threads.
+        .each(0..3, |_| smol::future::block_on(ex.run(shutdown.recv())))
+        // Run the main future on the current thread.
+        .finish(|| {
+            smol::future::block_on(async move {
+                start(ex2, options).await?;
+                drop(signal);
+                Ok::<(), drk::Error>(())
+            })
+        });
+
+    result
+}

src/service/gateway.rs

@@ -1,11 +1,14 @@
 use async_std::sync::{Arc, Mutex};
 use std::convert::TryInto;
+use std::net::SocketAddr;
 
 use super::reqrep::{Publisher, RepProtocol, Reply, ReqProtocol, Request, Subscriber};
 use crate::{Error, Result};
 
 use async_executor::Executor;
 
+use log::*;
+
 pub type Slabs = Vec<Vec<u8>>;
 
 #[repr(u8)]
@@ -17,12 +20,12 @@ enum GatewayCommand {
 
 pub struct GatewayService {
     slabs: Mutex<Slabs>,
-    addr: String,
+    addr: SocketAddr,
     publisher: Mutex<Publisher>,
 }
 
 impl GatewayService {
-    pub fn new(addr: String, pub_addr: String) -> Arc<GatewayService> {
+    pub fn new(addr: SocketAddr, pub_addr: SocketAddr) -> Arc<GatewayService> {
         let slabs = Mutex::new(vec![]);
         let publisher = Mutex::new(Publisher::new(pub_addr));
         Arc::new(GatewayService {
@@ -36,11 +39,11 @@ impl GatewayService {
         let mut socket = RepProtocol::new(self.addr.clone());
 
         let (send, recv) = socket.start().await?;
-        println!("server started");
+        info!("server started: bind to {}", self.addr.to_string());
 
         self.publisher.lock().await.start().await?;
 
-        println!("publisher started");
+        info!("publisher started");
 
         let handle_request_task = executor.spawn(self.handle_request(send.clone(), recv.clone()));
 
@@ -69,15 +72,15 @@ impl GatewayService {
                             // publish to all subscribes
                             self.publisher.lock().await.publish(slab).await?;
 
-                            println!("received putslab msg");
+                            info!("received putslab msg");
                         }
                         1 => {
                             // GETSLAB
-                            println!("received getslab msg");
+                            info!("received getslab msg");
                         }
                         2 => {
                             // GETLASTINDEX
-                            println!("received getlastindex msg");
+                            info!("received getlastindex msg");
                         }
                         _ => {
                             return Err(Error::ServicesError("wrong command"));
@@ -97,7 +100,7 @@ pub struct GatewayClient {
 }
 
 impl GatewayClient {
-    pub fn new(addr: String) -> GatewayClient {
+    pub fn new(addr: SocketAddr) -> GatewayClient {
         let protocol = ReqProtocol::new(addr);
         GatewayClient { protocol }
     }
@@ -106,7 +109,7 @@ impl GatewayClient {
         Ok(())
     }
 
-    pub async fn subscribe(&self, sub_addr: String) -> Result<Arc<Mutex<Subscriber>>> {
+    pub async fn subscribe(&self, sub_addr: SocketAddr) -> Result<Arc<Mutex<Subscriber>>> {
         let mut subscriber = Subscriber::new(sub_addr);
         subscriber.start().await?;
         Ok(Arc::new(Mutex::new(subscriber)))
@@ -144,7 +147,7 @@ pub async fn fetch_slabs_loop(
             let mut subscriber = subscriber.lock().await;
             slab = subscriber.fetch().await?;
         }
-        println!("received new slab from subscriber");
+        info!("received new slab from subscriber");
         slabs.lock().await.push(slab);
     }
 }

src/service/mod.rs

@@ -1,4 +1,6 @@
 pub mod gateway;
+pub mod options;
 pub mod reqrep;
 
 pub use gateway::{fetch_slabs_loop, GatewayClient, GatewayService};
+pub use options::ProgramOptions;

src/service/options.rs

@@ -0,0 +1,54 @@
+use crate::Result;
+use std::net::SocketAddr;
+
+pub struct ProgramOptions {
+    pub accept_addr: Option<SocketAddr>,
+    pub pub_addr: Option<SocketAddr>,
+    pub verbose: bool,
+    pub log_path: Box<std::path::PathBuf>,
+}
+
+impl ProgramOptions {
+    pub fn load() -> Result<ProgramOptions> {
+        let app = clap_app!(dfi =>
+            (version: "0.1.0")
+            (author: "Amir Taaki <amir@dyne.org>")
+            (about: "run service daemon")
+            (@arg ACCEPT: -a --accept +takes_value "Accept add//ress")
+            (@arg PUB_ADDR: -p --pubaddr +takes_value "Publisher addr")
+            (@arg VERBOSE: -v --verbose "Increase verbosity")
+            (@arg LOG_PATH: --log +takes_value "Logfile path")
+        )
+        .get_matches();
+
+        let accept_addr = if let Some(accept_addr) = app.value_of("ACCEPT") {
+            Some(accept_addr.parse()?)
+        } else {
+            None
+        };
+
+        let pub_addr = if let Some(pub_addr) = app.value_of("PUB_ADDR") {
+            Some(pub_addr.parse()?)
+        } else {
+            None
+        };
+
+        let verbose = app.is_present("VERBOSE");
+
+        let log_path = Box::new(
+            if let Some(log_path) = app.value_of("LOG_PATH") {
+                std::path::Path::new(log_path)
+            } else {
+                std::path::Path::new("/tmp/darkfid_service_daemon.log")
+            }
+            .to_path_buf(),
+        );
+
+        Ok(ProgramOptions {
+            accept_addr,
+            pub_addr,
+            verbose,
+            log_path,
+        })
+    }
+}

src/service/reqrep.rs

@@ -1,4 +1,5 @@
 use std::io;
+use std::net::SocketAddr;
 
 use crate::serial::{deserialize, serialize};
 use crate::{Decodable, Encodable, Result};
@@ -13,8 +14,12 @@ enum NetEvent {
     Send(Reply),
 }
 
+pub fn addr_to_string(addr: SocketAddr) -> String {
+    format!("tcp://{}", addr.to_string())
+}
+
 pub struct RepProtocol {
-    addr: String,
+    addr: SocketAddr,
     socket: zeromq::RepSocket,
     recv_queue: async_channel::Receiver<Reply>,
     send_queue: async_channel::Sender<Request>,
@@ -25,7 +30,7 @@ pub struct RepProtocol {
 }
 
 impl RepProtocol {
-    pub fn new(addr: String) -> RepProtocol {
+    pub fn new(addr: SocketAddr) -> RepProtocol {
         let socket = zeromq::RepSocket::new();
         let (send_queue, recv_channel) = async_channel::unbounded::<Request>();
         let (send_channel, recv_queue) = async_channel::unbounded::<Reply>();
@@ -47,7 +52,8 @@ impl RepProtocol {
         async_channel::Sender<Reply>,
         async_channel::Receiver<Request>,
     )> {
-        self.socket.bind(self.addr.as_str()).await?;
+        let addr = addr_to_string(self.addr);
+        self.socket.bind(addr.as_str()).await?;
         Ok(self.channels.clone())
     }
 
@@ -76,18 +82,19 @@ impl RepProtocol {
 }
 
 pub struct ReqProtocol {
-    addr: String,
+    addr: SocketAddr,
     socket: zeromq::ReqSocket,
 }
 
 impl ReqProtocol {
-    pub fn new(addr: String) -> ReqProtocol {
+    pub fn new(addr: SocketAddr) -> ReqProtocol {
         let socket = zeromq::ReqSocket::new();
         ReqProtocol { addr, socket }
     }
 
     pub async fn start(&mut self) -> Result<()> {
-        self.socket.connect(self.addr.as_str()).await?;
+        let addr = addr_to_string(self.addr);
+        self.socket.connect(addr.as_str()).await?;
         Ok(())
     }
 
@@ -115,17 +122,18 @@ impl ReqProtocol {
 }
 
 pub struct Publisher {
-    addr: String,
+    addr: SocketAddr,
     socket: zeromq::PubSocket,
 }
 
 impl Publisher {
-    pub fn new(addr: String) -> Publisher {
+    pub fn new(addr: SocketAddr) -> Publisher {
         let socket = zeromq::PubSocket::new();
         Publisher { addr, socket }
     }
     pub async fn start(&mut self) -> Result<()> {
-        self.socket.bind(self.addr.as_str()).await?;
+        let addr = addr_to_string(self.addr);
+        self.socket.bind(addr.as_str()).await?;
         Ok(())
     }
 
@@ -137,18 +145,19 @@ impl Publisher {
 }
 
 pub struct Subscriber {
-    addr: String,
+    addr: SocketAddr,
     socket: zeromq::SubSocket,
 }
 
 impl Subscriber {
-    pub fn new(addr: String) -> Subscriber {
+    pub fn new(addr: SocketAddr) -> Subscriber {
         let socket = zeromq::SubSocket::new();
         Subscriber { addr, socket }
     }
 
     pub async fn start(&mut self) -> Result<()> {
-        self.socket.connect(self.addr.as_str()).await?;
+        let addr = addr_to_string(self.addr);
+        self.socket.connect(addr.as_str()).await?;
 
         self.socket.subscribe("").await?;