make req rep and pub sub protocol

src/bin/demoservices.rs

@@ -4,11 +4,17 @@ use std::sync::Arc;
 
 use sapvi::Result;
 
-use sapvi::service::{gateway, reqrep};
+use sapvi::service::gateway;
 
 async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
-    executor.clone().spawn(reqrep::ReqRepAPI::start()).detach();
-    gateway::GatewayService::start(executor.clone()).await?;
+
+    let gateway =
+        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(())
 }
 

src/bin/demowallet.rs

@@ -1,30 +1,58 @@
-use sapvi::service::reqrep::{Reply, Request};
-use sapvi::{serial, Result};
-
-use bytes::Bytes;
-use zeromq::*;
-
-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);
-        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);
-    }
+use async_executor::Executor;
+use easy_parallel::Parallel;
+use async_std::sync::{Arc, Mutex};
+
+use sapvi::Result;
+use sapvi::service::gateway::GatewayClient;
+
+async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
+
+    let mut client =
+        GatewayClient::new(
+            String::from("tcp://127.0.0.1:3333"),
+        );
+
+    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?;
+
+    println!("subscription ready");
+
+
+    let fetch_loop_task = executor.spawn(GatewayClient::fetch_slabs_loop(subscriber.clone(), slabs.clone()));
+
+    client.put_slab(vec![0,0,0,0]).await?;
+    client.put_slab(vec![0,0,0,0]).await?;
+    client.put_slab(vec![0,0,0,0]).await?;
+
+    fetch_loop_task.cancel().await;
+
     Ok(())
 }
 
-fn main() {
-    futures::executor::block_on(connect()).unwrap();
+
+
+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/service/gateway.rs

@@ -1,141 +1,144 @@
 use std::convert::TryInto;
+use async_std::sync::{Arc, Mutex};
 
-use super::reqrep::{Reply, Request};
-use crate::serial::{deserialize, serialize};
+use super::reqrep::{Reply, Request, RepProtocol, ReqProtocol, Publisher, Subscriber};
 use crate::{Error, Result};
 
 use async_executor::Executor;
-use async_std::sync::Arc;
-use bytes::Bytes;
-use futures::FutureExt;
-use zeromq::*;
 
 pub type Slabs = Vec<Vec<u8>>;
 
 pub struct GatewayService {
-    slabs: Slabs,
-}
-
-enum NetEvent {
-    Receive(zeromq::ZmqMessage),
-    Send(zeromq::ZmqMessage),
+    slabs: Mutex<Slabs>,
+    addr: String,
+    publisher: Mutex<Publisher>
 }
 
 impl GatewayService {
+    pub fn new(addr: String, pub_addr: String) -> Arc<GatewayService>{
+        let slabs = Mutex::new(vec![]);
+        let publisher = Mutex::new(Publisher::new(pub_addr));
+        Arc::new(GatewayService {
+            slabs,
+            addr,
+            publisher
+        })
+    }
+
     pub async fn start(self: Arc<Self>, 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::<zeromq::ZmqMessage>();
+        let (send_queue_s, send_queue_r) = async_channel::unbounded::<Reply>();
+        let (recv_queue_s, recv_queue_r) = async_channel::unbounded::<Request>();
 
-        let ex2 = executor.clone();
-        loop {
-            let event = futures::select! {
-                request = worker.recv().fuse() => NetEvent::Receive(request?),
-                reply = send_queue_r.recv().fuse() => NetEvent::Send(reply?)
-            };
-
-            let self2 = self.clone();
-            match event {
-                NetEvent::Receive(request) => {
-                    let _ = ex2.spawn(self2.clone().handle_request(send_queue_s.clone(), request));
-                }
-                NetEvent::Send(reply) => {
-                    worker.send(reply).await?;
-                }
-            }
-        }
+
+        let mut reqrep = RepProtocol::new(self.addr.clone(), send_queue_r.clone(), recv_queue_s.clone(),);
+
+
+        reqrep.start().await?;
+        println!("server started");
+
+        self.publisher.lock().await.start().await?;
+
+        println!("publisher started");
+
+        let handle_request_task = executor.spawn(self.handle_request(send_queue_s.clone(), recv_queue_r.clone()));
+
+        reqrep.run().await?;
+
+        handle_request_task.cancel().await;
+        Ok(())
     }
 
     async fn handle_request(
         self: Arc<Self>,
-        send_queue: async_channel::Sender<zeromq::ZmqMessage>,
-        request: zeromq::ZmqMessage,
+        send_queue: async_channel::Sender<Reply>,
+        recv_queue: async_channel::Receiver<Request>,
     ) -> Result<()> {
-        let request: &Bytes = request.get(0).unwrap();
-        let request: Vec<u8> = request.to_vec();
-        let req: Request = deserialize(&request)?;
 
         let data = vec![];
-        match req.get_command() {
-            0 => {
-                // PUTSLAB
-                println!("receive PUTSLAB command");
-            }
-            1 => {
-                // GETSLAB
-                println!("receive GETSLAB command");
-            }
-            2 => {
-                // GETLASTINDEX
-                println!("receive GETLASTINDEX command");
-            }
-            _ => {
-                return Err(Error::ServicesError("wrong command"));
+
+        loop {
+            match recv_queue.recv().await{
+                Ok(request) => {
+                    match request.get_command() {
+                        0 => {
+                            // PUTSLAB
+                            let slab = request.get_payload();
+                            self.slabs.lock().await.push(slab.clone());
+
+                            // publish to all subscribes
+                            self.publisher.lock().await.publish(slab).await?;
+
+                            println!("received putslab msg");
+                        }
+                        1 => {
+                            // GETSLAB
+                            println!("received getslab msg");
+                        }
+                        2 => {
+                            // GETLASTINDEX
+                            println!("received getlastindex msg");
+                        }
+                        _ => {
+                            return Err(Error::ServicesError("wrong command"));
+                        }
+                    }
+                    let rep = Reply::from(&request, 0, data.clone());
+                    send_queue.send(rep.into()).await?;
+
+                }
+                Err(_) => {}
             }
         }
-
-        let rep = Reply::from(&req, 0, data);
-        let rep: Vec<u8> = serialize(&rep);
-        let rep = Bytes::from(rep);
-        send_queue.send(rep.into()).await?;
-        Ok(())
     }
 }
 
-struct GatewayClient {
-    slabs: Slabs,
-    sender: zeromq::ReqSocket,
+pub struct GatewayClient {
+    protocol: ReqProtocol,
 }
 
 impl GatewayClient {
-    pub fn new() -> GatewayClient {
-        let sender = zeromq::ReqSocket::new();
+    pub fn new(addr: String) -> GatewayClient {
+        let protocol = ReqProtocol::new(addr);
         GatewayClient {
-            slabs: vec![],
-            sender,
+            protocol,
         }
     }
     pub async fn start(&mut self) -> Result<()> {
-        self.sender.connect("tcp://127.0.0.1:3333").await?;
+        self.protocol.start().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(crate::Error::ServicesError("response has an error"));
-        }
-
-        assert!(reply.get_id() == request.get_id());
-
-        Ok(reply.get_payload())
+    pub async fn subscribe(&self, sub_addr: String) -> Result<Arc<Mutex<Subscriber>>> {
+        let mut subscriber = Subscriber::new(sub_addr);
+        subscriber.start().await?;
+        Ok(Arc::new(Mutex::new(subscriber)))
     }
 
     pub async fn get_slab(&mut self, index: u32) -> Result<Vec<u8>> {
-        self.request(GatewayCommand::GetSlab, index.to_be_bytes().to_vec())
+        self.protocol.request(GatewayCommand::GetSlab as u8, index.to_be_bytes().to_vec())
             .await
     }
 
     pub async fn put_slab(&mut self, data: Vec<u8>) -> Result<()> {
-        self.request(GatewayCommand::PutSlab, data).await?;
+        self.protocol.request(GatewayCommand::PutSlab as u8, data.clone()).await?;
         Ok(())
     }
     pub async fn get_last_index(&mut self) -> Result<u32> {
-        let rep = self.request(GatewayCommand::GetLastIndex, vec![]).await?;
+        let rep = self.protocol.request(GatewayCommand::GetLastIndex as u8, vec![]).await?;
         let rep: [u8; 4] = rep.try_into().unwrap();
         Ok(u32::from_be_bytes(rep))
     }
+
+    pub async fn fetch_slabs_loop(subscriber: Arc<Mutex<Subscriber>>, slabs: Arc<Mutex<Slabs>>) -> Result<()>{
+        loop {
+            let mut subscriber = subscriber.lock().await;
+            let slab = subscriber.fetch().await?;
+
+            println!("received new slab from subscriber"); 
+            slabs.lock().await.push(slab);
+        }
+    }
 }
 
 #[repr(u8)]

src/service/reqrep.rs

@@ -1,50 +1,175 @@
 use std::io;
 
 use crate::{Decodable, Encodable, Result};
+use crate::serial::{deserialize, serialize};
 
+use bytes::Bytes;
 use futures::FutureExt;
 use rand::Rng;
 use zeromq::*;
 
-pub struct ReqRepAPI;
+enum NetEvent {
+    Receive(zeromq::ZmqMessage),
+    Send(Reply),
+}
+
+
+pub struct RepProtocol{
+    addr: String,
+    socket: zeromq::RepSocket,
+    recv_queue: async_channel::Receiver<Reply>,
+    send_queue: async_channel::Sender<Request>,
+}
 
-impl ReqRepAPI {
-    pub async fn start() -> Result<()> {
-        println!("start reqrep");
+impl RepProtocol {
+    pub fn new(addr: String,
+        recv_queue: async_channel::Receiver<Reply>,
+        send_queue: async_channel::Sender<Request>
+    ) -> RepProtocol {
+        let socket = zeromq::RepSocket::new();
+        RepProtocol{
+            addr,
+            socket,
+            recv_queue,
+            send_queue,
+        }
 
-        let mut frontend = zeromq::RouterSocket::new();
-        frontend.bind("tcp://127.0.0.1:3333").await?;
+    }
+    pub async fn start(&mut self) -> Result<()> {
+        self.socket.bind(self.addr.as_str()).await?;
+        Ok(())
+    }
 
-        let mut backend = zeromq::DealerSocket::new();
-        backend.bind("tcp://127.0.0.1:4444").await?;
+    pub async fn run(&mut self) -> Result<()> {
         loop {
-            println!("start reqrep loop");
-            futures::select! {
-                frontend_mess = frontend.recv().fuse() => {
-                    match frontend_mess {
-                        Ok(message) => {
-                            backend.send(message).await?;
-                        }
-                        Err(_) => {
-                            // TODO
-                        }
-                    }
-                },
-                backend_mess = backend.recv().fuse() => {
-                    match backend_mess {
-                        Ok(message) => {
-                            frontend.send(message).await?;
-                        }
-                        Err(_) => {
-                            // TODO
-                        }
-                    }
-                }
+            let event = futures::select! {
+                request = self.socket.recv().fuse() => NetEvent::Receive(request?),
+                reply = self.recv_queue.recv().fuse() => NetEvent::Send(reply?)
             };
+
+            match event {
+                NetEvent::Receive(request) => {
+                    let request: &Bytes = request.get(0).unwrap();
+                    let request: Vec<u8> = request.to_vec();
+                    let req: Request = deserialize(&request)?;
+                    self.send_queue.send(req).await?;
+                }
+                NetEvent::Send(reply) => {
+                    let reply: Vec<u8> = serialize(&reply);
+                    let reply = Bytes::from(reply);
+                    self.socket.send(reply.into()).await?;
+                }
+            }
+        }
+
+    }
+}
+
+
+pub struct ReqProtocol {
+    addr: String,
+    socket: zeromq::ReqSocket,
+}
+
+impl ReqProtocol {
+    pub fn new(addr: String) -> ReqProtocol{
+        let socket = zeromq::ReqSocket::new();
+        ReqProtocol {
+            addr,
+            socket
+        }
+    }
+
+    pub async fn start(&mut self) -> Result<()> {
+        self.socket.connect(self.addr.as_str()).await?;
+        Ok(())
+    }
+
+    pub async fn request(&mut self, command: u8, data: Vec<u8>) -> Result<Vec<u8>> {
+        let request = Request::new(command, data);
+        let req = serialize(&request);
+        let req = bytes::Bytes::from(req);
+
+        self.socket.send(req.into()).await?;
+
+        let rep: zeromq::ZmqMessage = self.socket.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(crate::Error::ServicesError("response has an error"));
         }
+
+        assert!(reply.get_id() == request.get_id());
+
+        Ok(reply.get_payload())
     }
+
+
 }
 
+pub struct Publisher {
+    addr: String,
+    socket: zeromq::PubSocket,
+}
+
+
+
+impl Publisher {
+    pub fn new(addr: String) -> Publisher {
+        let socket = zeromq::PubSocket::new();
+        Publisher{
+            addr,
+            socket
+        }
+    }
+    pub async fn start(&mut self) -> Result<()> {
+        self.socket.bind(self.addr.as_str()).await?;
+        Ok(())
+    }
+
+    pub async fn publish(&mut self, data: Vec<u8>) -> Result<()>{
+        let data = Bytes::from(data);
+        self.socket.send(data.into()).await?;
+        Ok(())
+    }
+}
+
+pub struct Subscriber{
+    addr: String,
+    socket: zeromq::SubSocket
+}
+
+impl Subscriber {
+    pub fn new(addr: String) -> Subscriber {
+        let socket = zeromq::SubSocket::new();
+        Subscriber{
+            addr,
+            socket
+        }
+    }
+
+    pub async fn start(&mut self) -> Result<()> {
+        self.socket
+            .connect(self.addr.as_str())
+            .await?;
+
+        self.socket.subscribe("").await?;
+
+        Ok(())
+    }
+
+    pub async fn fetch(&mut self) -> Result<Vec<u8>>{
+        let data = self.socket.recv().await?;
+        let data: &Bytes = data.get(0).unwrap();
+        let data = data.to_vec();
+        Ok(data)
+    }
+}
+
+
 #[derive(Debug, PartialEq)]
 pub struct Request {
     command: u8,
@@ -73,6 +198,10 @@ impl Request {
     pub fn get_command(&self) -> u8 {
         self.command
     }
+
+    pub fn get_payload(&self) -> Vec<u8> {
+        self.payload.clone()
+    }
 }
 
 #[derive(Debug, PartialEq)]
@@ -172,3 +301,6 @@ mod tests {
         assert_eq!(deserialized_reply, Some(reply));
     }
 }
+
+
+