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

Cargo.toml

@@ -80,7 +80,9 @@ tobj = "2.0.4"
 fs_extra = "1.2"
 glob = "0.3"
 
-async_zmq = "0.3.2"
+# zmq
+zeromq = { git="https://github.com/zeromq/zmq.rs", default-features = false, features = ["async-std-runtime", "all-transport"] }
+bytes = "1.0.1"
 
 # wallet deps
 rocksdb = "0.16.0"
@@ -126,8 +128,8 @@ name = "compile-shaders"
 path = "src/bin/compile-shaders.rs"
 
 [[bin]]
-name = "services"
-path = "src/bin/services.rs"
+name = "demoservices"
+path = "src/bin/demoservices.rs"
 
 [[bin]]
 name = "demowallet"

src/bin/services.rs → src/bin/demoservices.rs

@@ -8,8 +8,7 @@ use sapvi::service::{gateway, reqrep};
 
 async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
     executor.clone().spawn(reqrep::ReqRepAPI::start()).detach();
-
-    gateway::GatewayService::start(executor.clone()).await;
+    gateway::GatewayService::start(executor.clone()).await?;
     Ok(())
 }
 

src/bin/demowallet.rs

@@ -1,33 +1,30 @@
-//! cargo run --example request --features="rt-tokio" --no-default-features
-
-use async_zmq::zmq;
-use sapvi::serial;
 use sapvi::service::reqrep::{Reply, Request};
+use sapvi::{serial, Result};
+
+use bytes::Bytes;
+use zeromq::*;
 
-fn connect() {
-    let context = zmq::Context::new();
-    let requester = context.socket(zmq::REQ).unwrap();
-    requester
-        .connect("tcp://127.0.0.1:3333")
-        .expect("failed to connect requester");
+async fn connect() -> Result<()> {
+    let mut requester = zeromq::ReqSocket::new();
+    requester.connect("tcp://127.0.0.1:3333").await?;
+
+    println!("connected");
 
     for request_nbr in 0..10 {
+        println!("start sending");
         let req = Request::new(0, "test".as_bytes().to_vec());
         let req = serial::serialize(&req);
-        requester.send(req, 0).unwrap();
-        let message = requester.recv_msg(0).unwrap();
+        let req = bytes::Bytes::from(req);
+        requester.send(req.into()).await?;
+        let message: zeromq::ZmqMessage = requester.recv().await?;
+        let message: &Bytes = message.get(0).unwrap();
+        let message: Vec<u8> = message.to_vec();
         let rep: Reply = serial::deserialize(&message).unwrap();
         println!("Received reply {:?} {:?}", request_nbr, rep);
     }
+    Ok(())
 }
-fn main() {
-    let mut thread_pools = vec![];
-    for _ in 0..20 {
-        let t = std::thread::spawn(connect);
-        thread_pools.push(t);
-    }
 
-    for t in thread_pools {
-        t.join().unwrap();
-    }
+fn main() {
+    futures::executor::block_on(connect()).unwrap();
 }

src/bin/dfi.rs

@@ -6,13 +6,13 @@ use async_std::sync::Mutex;
 use easy_parallel::Parallel;
 use http_types::{Request, Response, StatusCode};
 use log::*;
+use sapvi::rpc::jsonserver::RpcInterface;
+use sapvi::{net, Result};
 use serde_json::json;
 use smol::Async;
 use std::net::SocketAddr;
 use std::net::TcpListener;
 use std::sync::Arc;
-use sapvi::rpc::jsonserver::RpcInterface;
-use sapvi::{net, Result};
 
 /// Listens for incoming connections and serves them.
 async fn listen(

src/bin/tx.rs

@@ -48,7 +48,10 @@ fn main() {
     // Create the deposit for 110 BTC
     // Clear inputs are visible to everyone on the network
     let builder = tx::TransactionBuilder {
-        clear_inputs: vec![tx::TransactionBuilderClearInputInfo { value: 110, signature_secret: cashier_secret }],
+        clear_inputs: vec![tx::TransactionBuilderClearInputInfo {
+            value: 110,
+            signature_secret: cashier_secret,
+        }],
         inputs: vec![],
         outputs: vec![tx::TransactionBuilderOutputInfo { value: 110, public }],
         clear_outputs: vec![]
@@ -176,4 +179,3 @@ fn main() {
 
     // Step 4 withdraw the funds
 }
-

src/error.rs

@@ -1,11 +1,10 @@
 use std::fmt;
 
 use crate::net::error::NetError;
+use crate::service::ServicesError;
 use crate::vm::ZKVMError;
 use rusqlite;
 
-use async_zmq::zmq;
-
 pub type Result<T> = std::result::Result<T, Error>;
 
 #[derive(Debug)]
@@ -36,7 +35,6 @@ pub enum Error {
     VMError(ZKVMError),
     BadContract,
     Groth16Error(bellman::SynthesisError),
-    ZMQError(zmq::Error),
     RusqliteError(rusqlite::Error),
     OperationFailed,
     ConnectFailed,
@@ -46,6 +44,8 @@ pub enum Error {
     ServiceStopped,
     Utf8Error,
     NoteDecryptionFailed,
+    ServicesError(ServicesError),
+    ZMQError(zeromq::ZmqError),
 }
 
 impl std::error::Error for Error {}
@@ -81,7 +81,6 @@ impl fmt::Display for Error {
             Error::VMError(_) => f.write_str("VM error"),
             Error::BadContract => f.write_str("Contract is poorly defined"),
             Error::Groth16Error(ref err) => write!(f, "groth16 error: {}", err),
-            Error::ZMQError(ref err) => write!(f, "ZMQ error: {}", err),
             Error::RusqliteError(ref err) => write!(f, "Rusqlite error: {}", err),
             Error::OperationFailed => f.write_str("Operation failed"),
             Error::ConnectFailed => f.write_str("Connection failed"),
@@ -91,22 +90,30 @@ impl fmt::Display for Error {
             Error::ServiceStopped => f.write_str("Service stopped"),
             Error::Utf8Error => f.write_str("Malformed UTF8"),
             Error::NoteDecryptionFailed => f.write_str("Unable to decrypt mint note"),
+            Error::ServicesError(ref err) => write!(f, "Services error: {}", err),
+            Error::ZMQError(ref err) => write!(f, "zmq error: {}", err),
         }
     }
 }
 
-impl From<std::io::Error> for Error {
-    fn from(err: std::io::Error) -> Error {
-        Error::Io(err)
+impl From<ServicesError> for Error {
+    fn from(err: ServicesError) -> Error {
+        Error::ServicesError(err)
     }
 }
 
-impl From<zmq::Error> for Error {
-    fn from(err: zmq::Error) -> Error {
+impl From<zeromq::ZmqError> for Error {
+    fn from(err: zeromq::ZmqError) -> Error {
         Error::ZMQError(err)
     }
 }
 
+impl From<std::io::Error> for Error {
+    fn from(err: std::io::Error) -> Error {
+        Error::Io(err)
+    }
+}
+
 impl From<rusqlite::Error> for Error {
     fn from(err: rusqlite::Error) -> Error {
         Error::RusqliteError(err)

src/lib.rs

@@ -1,4 +1,5 @@
-#[macro_use] extern crate clap;
+#[macro_use]
+extern crate clap;
 use bellman::groth16;
 use bls12_381::{Bls12, Scalar};
 use std::collections::{HashMap, HashSet};

src/rpc/jsonserver.rs

@@ -1,3 +1,4 @@
+use crate::{net, serial, Error, Result};
 use async_executor::Executor;
 use async_native_tls::TlsAcceptor;
 use async_std::sync::Mutex;
@@ -6,6 +7,7 @@ use ff::Field;
 use http_types::{Request, Response, StatusCode};
 use log::*;
 use rand::rngs::OsRng;
+use rusqlite::Connection;
 use serde_json::json;
 use smol::Async;
 use std::fs::File;
@@ -14,8 +16,6 @@ use std::io::BufReader;
 use std::net::SocketAddr;
 use std::net::TcpListener;
 use std::sync::Arc;
-use rusqlite::Connection;
-use crate::{net, serial, Result, Error};
 
 // json RPC server goes here
 pub struct RpcInterface {
@@ -113,4 +113,3 @@ impl RpcInterface {
         Ok(self.stop_recv.recv().await?)
     }
 }
-

src/rpc/mod.rs

@@ -1,3 +1,2 @@
-pub mod test;
 pub mod jsonserver;
-
+pub mod test;

src/rpc/test.rs

@@ -1,3 +1 @@
-fn foo() {
-}
-
+fn foo() {}

src/service/error.rs

@@ -0,0 +1,18 @@
+use std::fmt;
+
+pub type Result<T> = std::result::Result<T, ServicesError>;
+
+#[derive(Debug, Copy, Clone)]
+pub enum ServicesError {
+    ResonseError(&'static str),
+}
+
+impl std::error::Error for ServicesError {}
+
+impl fmt::Display for ServicesError {
+    fn fmt(&self, f: &mut fmt::Formatter) -> std::fmt::Result {
+        match *self {
+            ServicesError::ResonseError(ref err) => write!(f, "Response: {}", err),
+        }
+    }
+}

src/service/gateway.rs

@@ -1,35 +1,39 @@
-use image::EncodableLayout;
+use std::convert::TryInto;
 
 use super::reqrep::{Reply, Request};
+use super::ServicesError;
 use crate::serial::{deserialize, serialize};
 use crate::Result;
 
 use async_executor::Executor;
 use async_std::sync::Arc;
-use async_zmq;
+use bytes::Bytes;
 use futures::FutureExt;
+use zeromq::*;
 
-pub struct GatewayService;
+pub type Slabs = Vec<Vec<u8>>;
+
+pub struct GatewayService {
+    slabs: Slabs,
+}
 
 enum NetEvent {
-    RECEIVE(async_zmq::Multipart),
-    SEND(async_zmq::Multipart),
+    RECEIVE(zeromq::ZmqMessage),
+    SEND(zeromq::ZmqMessage),
 }
 
 impl GatewayService {
-    pub async fn start(executor: Arc<Executor<'_>>) {
-        let mut worker = async_zmq::reply("tcp://127.0.0.1:4444")
-            .unwrap()
-            .connect()
-            .unwrap();
+    pub async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
+        let mut worker = zeromq::RepSocket::new();
+        worker.connect("tcp://127.0.0.1:4444").await?;
 
-        let (send_queue_s, send_queue_r) = async_channel::unbounded::<async_zmq::Multipart>();
+        let (send_queue_s, send_queue_r) = async_channel::unbounded::<zeromq::ZmqMessage>();
 
         let ex2 = executor.clone();
         loop {
             let event = futures::select! {
-                request = worker.recv().fuse() => NetEvent::RECEIVE(request.unwrap()),
-                reply = send_queue_r.recv().fuse() => NetEvent::SEND(reply.unwrap())
+                request = worker.recv().fuse() => NetEvent::RECEIVE(request?),
+                reply = send_queue_r.recv().fuse() => NetEvent::SEND(reply?)
             };
 
             match event {
@@ -38,37 +42,87 @@ impl GatewayService {
                         .detach();
                 }
                 NetEvent::SEND(reply) => {
-                    worker.send(reply).await.unwrap();
+                    worker.send(reply).await?;
                 }
             }
         }
     }
 
     async fn handle_request(
-        send_queue: async_channel::Sender<async_zmq::Multipart>,
-        request: async_zmq::Multipart,
+        send_queue: async_channel::Sender<zeromq::ZmqMessage>,
+        request: zeromq::ZmqMessage,
     ) -> Result<()> {
-        let mut messages = vec![];
-        for req in request.iter() {
-            let req = req.as_bytes();
-            let req: Request = deserialize(req).unwrap();
+        let request: &Bytes = request.get(0).unwrap();
+        let request: Vec<u8> = request.to_vec();
+        let req: Request = deserialize(&request)?;
 
-            // TODO
-            // do things
+        // TODO
+        // do things
 
-            println!("Gateway service received a msg {:?}", req);
+        println!("Gateway service received a msg {:?}", req);
 
-            let rep = Reply::from(&req, 0, "text".as_bytes().to_vec());
-            let rep = serialize(&rep);
-            let msg = async_zmq::Message::from(rep);
-            messages.push(msg);
-        }
-        send_queue.send(messages).await?;
+        let rep = Reply::from(&req, 0, "text".as_bytes().to_vec());
+        let rep: Vec<u8> = serialize(&rep);
+        let rep = Bytes::from(rep);
+        send_queue.send(rep.into()).await?;
         Ok(())
     }
 }
 
-struct GatewayClient;
+struct GatewayClient {
+    slabs: Slabs,
+    sender: zeromq::ReqSocket,
+}
+
+impl GatewayClient {
+    pub fn new() -> GatewayClient {
+        let sender = zeromq::ReqSocket::new();
+        GatewayClient {
+            slabs: vec![],
+            sender,
+        }
+    }
+    pub async fn start(&mut self) -> Result<()> {
+        self.sender.connect("tcp://127.0.0.1:3333").await?;
+        Ok(())
+    }
+    async fn request(&mut self, command: GatewayCommand, data: Vec<u8>) -> Result<Vec<u8>> {
+        let request = Request::new(command as u8, data);
+        let req = serialize(&request);
+        let req = bytes::Bytes::from(req);
+
+        self.sender.send(req.into()).await?;
+
+        let rep: zeromq::ZmqMessage = self.sender.recv().await?;
+        let rep: &Bytes = rep.get(0).unwrap();
+        let rep: Vec<u8> = rep.to_vec();
+
+        let reply: Reply = deserialize(&rep)?;
+
+        if reply.has_error() {
+            return Err(ServicesError::ResonseError("response has an error").into());
+        }
+
+        assert!(reply.get_id() == request.get_id());
+
+        Ok(reply.get_payload())
+    }
+
+    pub async fn get_slab(&mut self, index: u32) -> Result<Vec<u8>> {
+        self.request(GatewayCommand::GETSLAB, index.to_be_bytes().to_vec())
+            .await
+    }
+
+    pub async fn put_slab(&mut self, data: Vec<u8>) -> Result<()> {
+        self.request(GatewayCommand::GETSLAB, data).await?;
+        Ok(())
+    }
+    pub async fn get_last_index(&mut self) -> Result<u32> {
+        let rep = self.request(GatewayCommand::GETLASTINDEX, vec![]).await?;
+        let rep: [u8; 4] = rep.try_into().unwrap();
+        Ok(u32::from_be_bytes(rep))
+    }
+}
 
 #[repr(u8)]
 enum GatewayCommand {

src/service/mod.rs

@@ -1,2 +1,5 @@
+mod error;
 pub mod gateway;
 pub mod reqrep;
+
+pub use error::ServicesError;

src/service/reqrep.rs

@@ -2,56 +2,45 @@ use std::io;
 
 use crate::{Decodable, Encodable, Result};
 
-use async_zmq::zmq;
+use futures::FutureExt;
 use rand::Rng;
+use zeromq::*;
 
 pub struct ReqRepAPI;
 
 impl ReqRepAPI {
-    pub async fn start() {
-        let context = zmq::Context::new();
-        let frontend = context.socket(zmq::ROUTER).unwrap();
-        let backend = context.socket(zmq::DEALER).unwrap();
-
-        frontend
-            .bind("tcp://127.0.0.1:3333")
-            .expect("failed binding frontend");
-        backend
-            .bind("tcp://127.0.0.1:4444")
-            .expect("failed binding backend");
+    pub async fn start() -> Result<()> {
+        println!("start reqrep");
 
+        let mut frontend = zeromq::RouterSocket::new();
+        frontend.bind("tcp://127.0.0.1:3333").await?;
+
+        let mut backend = zeromq::DealerSocket::new();
+        backend.bind("tcp://127.0.0.1:4444").await?;
         loop {
-            let mut items = [
-                frontend.as_poll_item(zmq::POLLIN),
-                backend.as_poll_item(zmq::POLLIN),
-            ];
-
-            zmq::poll(&mut items, -1).unwrap();
-
-            if items[0].is_readable() {
-                loop {
-                    let message = frontend.recv_msg(0).unwrap();
-                    let more = message.get_more();
-                    backend
-                        .send(message, if more { zmq::SNDMORE } else { 0 })
-                        .unwrap();
-                    if !more {
-                        break;
+            println!("start reqrep loop");
+            futures::select! {
+                frontend_mess = frontend.recv().fuse() => {
+                    match frontend_mess {
+                        Ok(message) => {
+                            backend.send(message).await?;
+                        }
+                        Err(_) => {
+                            // TODO
+                        }
                     }
-                }
-            }
-            if items[1].is_readable() {
-                loop {
-                    let message = backend.recv_msg(0).unwrap();
-                    let more = message.get_more();
-                    frontend
-                        .send(message, if more { zmq::SNDMORE } else { 0 })
-                        .unwrap();
-                    if !more {
-                        break;
+                },
+                backend_mess = backend.recv().fuse() => {
+                    match backend_mess {
+                        Ok(message) => {
+                            frontend.send(message).await?;
+                        }
+                        Err(_) => {
+                            // TODO
+                        }
                     }
                 }
-            }
+            };
         }
     }
 }
@@ -97,6 +86,22 @@ impl Reply {
             payload,
         }
     }
+
+    pub fn has_error(&self) -> bool {
+        if self.error == 0 {
+            false
+        } else {
+            true
+        }
+    }
+
+    pub fn get_payload(&self) -> Vec<u8> {
+        self.payload.clone()
+    }
+
+    pub fn get_id(&self) -> u32 {
+        self.id
+    }
 }
 
 impl Encodable for Request {

src/tx.rs

@@ -67,7 +67,7 @@ impl TransactionBuilder {
             let clear_input = PartialTransactionClearInput {
                 value: input.value,
                 valcom_blind,
-                signature_public
+                signature_public,
             };
             clear_inputs.push(clear_input);
         }
@@ -331,8 +331,7 @@ impl Transaction {
 
         // Verify signatures
         let mut unsigned_tx_data = vec![];
-        self
-            .encode_without_signature(&mut unsigned_tx_data)
+        self.encode_without_signature(&mut unsigned_tx_data)
             .expect("TODO handle this");
         for input in &self.clear_inputs {
             let public = schnorr::PublicKey(input.signature_public.clone());
@@ -364,7 +363,7 @@ impl TransactionClearInput {
             value: partial.value,
             valcom_blind: partial.valcom_blind,
             signature_public: partial.signature_public,
-            signature
+            signature,
         }
     }