|
|
@@ -1,132 +1,156 @@
|
|
|
+use async_std::sync::{Arc, Mutex};
|
|
|
use std::convert::TryInto;
|
|
|
|
|
|
-use super::reqrep::{Reply, Request};
|
|
|
-use super::ServicesError;
|
|
|
-use crate::serial::{deserialize, serialize};
|
|
|
-use crate::Result;
|
|
|
+use super::reqrep::{Publisher, RepProtocol, Reply, ReqProtocol, Request, 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 async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
|
|
|
- let mut worker = zeromq::RepSocket::new();
|
|
|
- worker.connect("tcp://127.0.0.1:4444").await?;
|
|
|
+ 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,
|
|
|
+ })
|
|
|
+ }
|
|
|
|
|
|
- let (send_queue_s, send_queue_r) = async_channel::unbounded::<zeromq::ZmqMessage>();
|
|
|
+ pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
|
|
|
+ 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?)
|
|
|
- };
|
|
|
-
|
|
|
- match event {
|
|
|
- NetEvent::RECEIVE(request) => {
|
|
|
- ex2.spawn(Self::handle_request(send_queue_s.clone(), request))
|
|
|
- .detach();
|
|
|
- }
|
|
|
- NetEvent::SEND(reply) => {
|
|
|
- worker.send(reply).await?;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
+ let mut reqrep = RepProtocol::new(
|
|
|
+ self.addr.clone(),
|
|
|
+ send_queue_r.clone(),
|
|
|
+ recv_queue_s.clone(),
|
|
|
+ );
|
|
|
|
|
|
- async fn handle_request(
|
|
|
- send_queue: async_channel::Sender<zeromq::ZmqMessage>,
|
|
|
- request: zeromq::ZmqMessage,
|
|
|
- ) -> Result<()> {
|
|
|
- let request: &Bytes = request.get(0).unwrap();
|
|
|
- let request: Vec<u8> = request.to_vec();
|
|
|
- let req: Request = deserialize(&request)?;
|
|
|
+ reqrep.start().await?;
|
|
|
+ println!("server started");
|
|
|
+
|
|
|
+ self.publisher.lock().await.start().await?;
|
|
|
|
|
|
- // TODO
|
|
|
- // do things
|
|
|
+ println!("publisher started");
|
|
|
|
|
|
- println!("Gateway service received a msg {:?}", req);
|
|
|
+ let handle_request_task =
|
|
|
+ executor.spawn(self.handle_request(send_queue_s.clone(), recv_queue_r.clone()));
|
|
|
|
|
|
- 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?;
|
|
|
+ reqrep.run().await?;
|
|
|
+
|
|
|
+ handle_request_task.cancel().await;
|
|
|
Ok(())
|
|
|
}
|
|
|
+
|
|
|
+ async fn handle_request(
|
|
|
+ self: Arc<Self>,
|
|
|
+ send_queue: async_channel::Sender<Reply>,
|
|
|
+ recv_queue: async_channel::Receiver<Request>,
|
|
|
+ ) -> Result<()> {
|
|
|
+ let data = vec![];
|
|
|
+
|
|
|
+ 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(_) => {}
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
-struct GatewayClient {
|
|
|
- slabs: Slabs,
|
|
|
- sender: zeromq::ReqSocket,
|
|
|
+pub struct GatewayClient {
|
|
|
+ protocol: ReqProtocol,
|
|
|
}
|
|
|
|
|
|
impl GatewayClient {
|
|
|
- pub fn new() -> GatewayClient {
|
|
|
- let sender = zeromq::ReqSocket::new();
|
|
|
- GatewayClient {
|
|
|
- slabs: vec![],
|
|
|
- sender,
|
|
|
- }
|
|
|
+ pub fn new(addr: String) -> GatewayClient {
|
|
|
+ let protocol = ReqProtocol::new(addr);
|
|
|
+ GatewayClient { 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(ServicesError::ResonseError("response has an error").into());
|
|
|
- }
|
|
|
-
|
|
|
- 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::GETSLAB, 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)]
|
|
|
enum GatewayCommand {
|
|
|
- PUTSLAB,
|
|
|
- GETSLAB,
|
|
|
- GETLASTINDEX,
|
|
|
+ PutSlab,
|
|
|
+ GetSlab,
|
|
|
+ GetLastIndex,
|
|
|
}
|