zmq api test code

Cargo.toml

@@ -78,6 +78,9 @@ tobj = "2.0.4"
 fs_extra = "1.2"
 glob = "0.3"
 
+async_zmq = "0.3.2"
+
+
 [[bin]]
 name = "lisp"
 path = "lisp/lisp.rs"
@@ -110,6 +113,13 @@ path = "src/bin/dfg.rs"
 name = "compile-shaders"
 path = "src/bin/compile-shaders.rs"
 
+[[bin]]
+name = "services"
+path = "src/bin/services.rs"
+[[bin]]
+name = "wallet"
+path = "src/bin/demowallet.rs"
+
 [profile.release]
 debug = 1
 

src/bin/demowallet.rs

@@ -0,0 +1,42 @@
+
+//! cargo run --example request --features="rt-tokio" --no-default-features
+
+use async_zmq::zmq;
+use sapvi::service::reqrep::{Request, Reply};
+use sapvi::serial;
+
+
+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");
+
+    for request_nbr in 0..10 {
+        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 rep: Reply = serial::deserialize(&message).unwrap();
+        println!(
+            "Received reply {:?} {:?}",
+            request_nbr,
+            rep
+        );
+    }
+}
+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();
+    }
+
+}

src/bin/services.rs

@@ -0,0 +1,40 @@
+use async_executor::Executor;
+use easy_parallel::Parallel;
+use std::sync::Arc;
+
+use sapvi::Result;
+
+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; 
+    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::<(), sapvi::Error>(())
+            })
+        });
+
+    result
+}
+
+
+

src/error.rs

@@ -3,6 +3,8 @@ use std::fmt;
 use crate::net::error::NetError;
 use crate::vm::ZKVMError;
 
+use async_zmq::zmq;
+
 pub type Result<T> = std::result::Result<T, Error>;
 
 #[derive(Debug)]
@@ -33,6 +35,7 @@ pub enum Error {
     VMError(ZKVMError),
     BadContract,
     Groth16Error(bellman::SynthesisError),
+    ZMQError(zmq::Error),
     OperationFailed,
     ConnectFailed,
     ConnectTimeout,
@@ -75,6 +78,7 @@ 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::OperationFailed => f.write_str("Operation failed"),
             Error::ConnectFailed => f.write_str("Connection failed"),
             Error::ConnectTimeout => f.write_str("Connection timed out"),
@@ -92,6 +96,12 @@ impl From<std::io::Error> for Error {
     }
 }
 
+impl From<zmq::Error> for Error {
+    fn from(err: zmq::Error) -> Error {
+        Error::ZMQError(err)
+    }
+}
+
 impl From<ZKVMError> for Error {
     fn from(err: ZKVMError) -> Error {
         Error::VMError(err)

src/lib.rs

@@ -15,6 +15,7 @@ pub mod serial;
 pub mod system;
 pub mod vm;
 pub mod vm_serial;
+pub mod service;
 
 pub use crate::bls_extensions::BlsStringConversion;
 pub use crate::error::{Error, Result};

src/service/gateway.rs

@@ -0,0 +1,82 @@
+use image::EncodableLayout;
+
+use crate::Result;
+use crate::serial::{serialize, deserialize};
+use super::reqrep::{Request, Reply};
+
+use async_zmq;
+use async_std::sync::Arc;
+use async_executor::Executor;
+use futures::FutureExt;
+
+
+
+pub struct GatewayService;
+
+
+enum NetEvent{
+    RECEIVE(async_zmq::Multipart),
+    SEND(async_zmq::Multipart)
+}
+
+
+impl GatewayService {
+
+    pub async fn start(
+        executor: Arc<Executor<'_>>,
+    ) {
+        let mut worker = async_zmq::reply("tcp://127.0.0.1:4444").unwrap().connect().unwrap();
+
+        let (send_queue_s, send_queue_r) = async_channel::unbounded::<async_zmq::Multipart>();
+
+        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())
+            };
+
+            match event {
+                NetEvent::RECEIVE(request) => {
+                    ex2.spawn(Self::handle_request(send_queue_s.clone(), request)).detach();
+                },
+                NetEvent::SEND(reply) => {
+                    worker.send(reply).await.unwrap();
+                },
+            }
+        }
+
+    }
+
+    async fn handle_request(send_queue: async_channel::Sender<async_zmq::Multipart>, request: async_zmq::Multipart) -> Result<()> {
+        let mut messages = vec![];
+        for req in request.iter() {
+            let req = req.as_bytes();
+            let req: Request = deserialize(req).unwrap();
+
+            // TODO
+            // do things
+
+            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?;
+        Ok(())
+    }
+}
+
+
+struct GatewayClient;
+
+
+#[repr(u8)]
+enum GatewayCommand{
+    PUTSLAB,
+    GETSLAB,
+    GETLASTINDEX,
+}
+

src/service/mod.rs

@@ -0,0 +1,4 @@
+
+pub mod reqrep;
+pub mod gateway;
+

src/service/reqrep.rs

@@ -0,0 +1,172 @@
+use std::io;
+
+use crate::{Decodable, Encodable, Result};
+
+use async_zmq::zmq;
+use rand::Rng;
+
+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");
+
+        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
+                    }
+                }
+            }
+            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
+                    }
+                }
+            }
+        }
+    }
+}
+
+
+#[derive(Debug, PartialEq)]
+pub struct Request {
+    command: u8,
+    id: u32,
+    payload: Vec<u8>,
+}
+
+impl Request {
+    pub fn new(command: u8, payload: Vec<u8>) -> Request {
+        let id = Self::gen_id();
+        Request {
+            command,
+            id,
+            payload,
+        }
+    }
+    fn gen_id() -> u32 {
+        let mut rng = rand::thread_rng();
+        rng.gen()
+    }
+
+    pub fn get_id(&self) -> u32 {
+        self.id
+    }
+}
+
+#[derive(Debug, PartialEq)]
+pub struct Reply {
+    id: u32,
+    error: u32,
+    payload: Vec<u8>,
+}
+
+impl Reply {
+    pub fn from(request: &Request, error: u32, payload: Vec<u8>) -> Reply {
+        Reply {
+            id: request.get_id(),
+            error,
+            payload
+        }
+    }
+}
+
+
+
+impl Encodable for Request {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.command.encode(&mut s)?;
+        len += self.id.encode(&mut s)?;
+        len += self.payload.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Encodable for Reply {
+    fn encode<S: io::Write>(&self, mut s: S) -> Result<usize> {
+        let mut len = 0;
+        len += self.id.encode(&mut s)?;
+        len += self.error.encode(&mut s)?;
+        len += self.payload.encode(&mut s)?;
+        Ok(len)
+    }
+}
+
+impl Decodable for Request {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            command: Decodable::decode(&mut d)?,
+            id: Decodable::decode(&mut d)?,
+            payload: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+impl Decodable for Reply {
+    fn decode<D: io::Read>(mut d: D) -> Result<Self> {
+        Ok(Self {
+            id: Decodable::decode(&mut d)?,
+            error: Decodable::decode(&mut d)?,
+            payload: Decodable::decode(&mut d)?,
+        })
+    }
+}
+
+
+
+
+#[cfg(test)]
+mod tests {
+    use crate::serial::{deserialize, serialize};
+    use super::{Request, Reply, Result};
+
+    #[test]
+    fn serialize_and_deserialize_request_test(){
+        let request = Request::new(2, vec![2,3,4,6,4]);
+        let serialized_request = serialize(&request);
+        assert!((deserialize(&serialized_request) as Result<bool>).is_err());
+        let deserialized_request = deserialize(&serialized_request).ok();
+        assert_eq!(deserialized_request, Some(request));
+    }
+
+    #[test]
+    fn serialize_and_deserialize_reply_test(){
+        let request = Request::new(2, vec![2,3,4,6,4]);
+        let reply = Reply::from(&request, 0, vec![2,3,4,6,4]);
+        let serialized_reply = serialize(&reply);
+        assert!((deserialize(&serialized_reply) as Result<bool>).is_err());
+        let deserialized_reply = deserialize(&serialized_reply).ok();
+        assert_eq!(deserialized_reply, Some(reply));
+    }
+
+}