refactor(server): make command smaller and read more data from storage

Author: manhunto

server/src/bin/dispatcher.rs

@@ -1,11 +1,14 @@
+use actix_web::web::Data;
+
 use server::amqp::establish_connection_with_rabbit;
 use server::dispatch_consumer::consume;
 use server::logs::init_log;
+use server::storage::Storage;
 
 #[tokio::main]
 async fn main() {
     init_log();
     let channel = establish_connection_with_rabbit().await;
 
-    consume(channel, "dispatcher").await
+    consume(channel, "dispatcher", Data::new(Storage::new())).await
 }

server/src/bin/server.rs

@@ -19,6 +19,7 @@ async fn main() -> std::io::Result<()> {
     let port: u16 = Env::env_or("SERVER_PORT", 8090);
 
     let storage = Data::new(Storage::new());
+    let storage_for_consumer = storage.clone();
     let channel = establish_connection_with_rabbit().await;
     let publisher = Data::new(Publisher::new(channel.clone()));
     let app = move || {
@@ -33,7 +34,7 @@ async fn main() -> std::io::Result<()> {
         ip, port
     );
 
-    rt::spawn(async move { consume(channel, "dispatcher-in-server").await });
+    rt::spawn(async move { consume(channel, "dispatcher-in-server", storage_for_consumer).await });
 
     HttpServer::new(app).bind((ip, port))?.run().await
 }

server/src/circuit_breaker.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::future::Future;
 
 use log::debug;
 
@@ -33,17 +34,18 @@ impl CircuitBreaker {
         self.states.get(&key).unwrap_or(&State::Open) == &State::Closed
     }
 
-    pub fn call<T, E, F>(&mut self, key: String, function: F) -> Result<T, Error<E>>
+    pub async fn call<T, E, F, Fut>(&mut self, key: String, function: F) -> Result<T, Error<E>>
     where
-        F: FnOnce() -> Result<T, E>,
+        F: FnOnce() -> Fut,
+        Fut: Future<Output = Result<T, E>>,
     {
         if self.is_call_permitted(key.clone()) {
             debug!("Service {} closed", key);
 
             return Err(Error::Rejected);
         }
 
-        match function() {
+        match function().await {
             Ok(ok) => {
                 self.storage.entry(key.clone()).or_insert(0);
 

server/src/cmd.rs

@@ -1,8 +1,7 @@
 use serde::{Deserialize, Serialize};
-use url::Url;
 
 use crate::configuration::domain::EndpointId;
-use crate::events::domain::{MessageId, Payload};
+use crate::events::domain::MessageId;
 
 #[derive(Serialize, Deserialize)]
 #[serde(tag = "t", content = "c")]
@@ -12,18 +11,14 @@ pub enum AsyncMessage {
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct SentMessage {
-    pub payload: String,
-    pub url: String,
-    pub msg_id: String,
+    msg_id: String,
     pub attempt: usize,
-    pub endpoint_id: String,
+    endpoint_id: String,
 }
 
 impl SentMessage {
-    pub fn new(payload: Payload, url: Url, msg_id: MessageId, endpoint_id: EndpointId) -> Self {
+    pub fn new(msg_id: MessageId, endpoint_id: EndpointId) -> Self {
         Self {
-            payload: payload.to_string(),
-            url: url.to_string(),
             msg_id: msg_id.to_string(),
             attempt: 1,
             endpoint_id: endpoint_id.to_string(),
@@ -32,11 +27,17 @@ impl SentMessage {
 
     pub fn with_increased_attempt(&self) -> SentMessage {
         Self {
-            payload: self.payload.clone(),
-            url: self.url.clone(),
             msg_id: self.msg_id.clone(),
             attempt: self.attempt + 1,
             endpoint_id: self.endpoint_id.clone(),
         }
     }
+
+    pub fn msg_id(&self) -> MessageId {
+        MessageId::try_from(self.msg_id.clone()).unwrap()
+    }
+
+    pub fn endpoint_id(&self) -> EndpointId {
+        EndpointId::try_from(self.endpoint_id.clone()).unwrap()
+    }
 }

server/src/dispatch_consumer.rs

@@ -1,17 +1,19 @@
 use std::time::Duration;
 
+use actix_web::web::Data;
 use futures_lite::StreamExt;
 use lapin::options::{BasicAckOptions, BasicConsumeOptions};
 use lapin::types::FieldTable;
 use lapin::Channel;
-use log::{debug, info};
+use log::{debug, error, info};
 
 use crate::amqp::{Publisher, Serializer, SENT_MESSAGE_QUEUE};
 use crate::circuit_breaker::{CircuitBreaker, Error};
 use crate::cmd::AsyncMessage;
 use crate::retry::RetryPolicyBuilder;
+use crate::storage::Storage;
 
-pub async fn consume(channel: Channel, consumer_tag: &str) {
+pub async fn consume(channel: Channel, consumer_tag: &str, storage: Data<Storage>) {
     let retry_policy = RetryPolicyBuilder::new()
         .max_retries(5)
         .exponential(2, Duration::from_secs(2))
@@ -44,22 +46,52 @@ pub async fn consume(channel: Channel, consumer_tag: &str) {
         info!("message consumed: {:?}", cmd);
 
         // todo allow to revive endpoint, check endpoint status and reset circuit breaker
+        let endpoint_id = cmd.endpoint_id();
+        let msg = storage.messages.get(cmd.msg_id());
+
+        if msg.is_err() {
+            error!(
+                "Message {} doesn't not exists and cannot be dispatched",
+                cmd.msg_id()
+            );
+
+            delivery.ack(BasicAckOptions::default()).await.expect("ack");
+
+            continue;
+        }
+
+        let endpoint = storage.endpoints.get(&endpoint_id);
+        if endpoint.is_err() {
+            error!(
+                "Endpoint {} doesn't not exists and message {} cannot be dispatched",
+                endpoint_id,
+                cmd.msg_id()
+            );
+
+            delivery.ack(BasicAckOptions::default()).await.expect("ack");
+
+            continue;
+        }
+
+        let msg = msg.unwrap();
+        let endpoint = endpoint.unwrap();
+
+        debug!(
+            "Message {} for endpoint {} is being prepared to send",
+            msg.id.to_string(),
+            endpoint.id.to_string()
+        );
 
-        let key = cmd.endpoint_id.to_string();
         let func = || {
-            reqwest::blocking::Client::new()
-                .post(&cmd.url)
-                .json(&cmd.payload.as_str())
+            reqwest::Client::new()
+                .post(endpoint.url)
+                .json(msg.payload.to_string().as_str())
                 .send()
         };
 
-        let log_error_response = |res: reqwest::Error| {
-            let status: String = res.status().map_or(String::from("-"), |s| s.to_string());
-
-            debug!("Error response! Status: {}, Error: {}", status, res);
-        };
+        let key = endpoint_id.to_string();
 
-        match circuit_breaker.call(key.clone(), func) {
+        match circuit_breaker.call(key.clone(), func).await {
             Ok(res) => {
                 debug!("Success! {}", res.status())
             }
@@ -92,7 +124,8 @@ pub async fn consume(channel: Channel, consumer_tag: &str) {
                 Error::Rejected => {
                     debug!(
                         "Endpoint {} is closed. Message {} rejected.",
-                        key, cmd.msg_id
+                        key,
+                        cmd.msg_id()
                     );
 
                     // todo do something with message? add to some "not delivered" bucket?
@@ -103,3 +136,9 @@ pub async fn consume(channel: Channel, consumer_tag: &str) {
         delivery.ack(BasicAckOptions::default()).await.expect("ack");
     }
 }
+
+fn log_error_response(res: reqwest::Error) {
+    let status: String = res.status().map_or(String::from("-"), |s| s.to_string());
+
+    debug!("Error response! Status: {}, Error: {}", status, res);
+}

server/src/events/domain.rs

@@ -1,6 +1,7 @@
-use crate::configuration::domain::{ApplicationId, Topic};
 use std::fmt::{Display, Formatter};
 
+use crate::configuration::domain::{ApplicationId, Topic};
+
 #[derive(Debug, Clone)]
 pub struct Payload {
     body: String,
@@ -18,7 +19,7 @@ impl Display for Payload {
     }
 }
 
-#[derive(Debug, Clone, derive::Ksuid)]
+#[derive(Debug, Clone, derive::Ksuid, Eq, PartialEq)]
 #[prefix = "msg"]
 pub struct MessageId {
     id: String,

server/src/events/handlers.rs

@@ -50,12 +50,7 @@ pub async fn create_message_handler(
     for endpoint in active_endpoints {
         debug!("{} sending to {}", msg.id, endpoint.url);
 
-        let cmd = SentMessage::new(
-            msg.payload.clone(),
-            endpoint.url,
-            msg.id.clone(),
-            endpoint.id,
-        );
+        let cmd = SentMessage::new(msg.id.clone(), endpoint.id);
         let message = AsyncMessage::SentMessage(cmd);
 
         dispatcher.publish(message).await;

server/src/events/storage.rs

@@ -1,10 +1,15 @@
-use crate::events::domain::Message;
 use std::sync::Mutex;
 
+use crate::error::Error;
+use crate::error::Error::EntityNotFound;
+use crate::events::domain::{Message, MessageId};
+
 pub trait MessageStorage {
     fn save(&self, app: Message);
 
     fn count(&self) -> usize;
+
+    fn get(&self, message_id: MessageId) -> Result<Message, Error>;
 }
 
 pub struct InMemoryMessageStorage {
@@ -31,4 +36,14 @@ impl MessageStorage for InMemoryMessageStorage {
 
         messages.len()
     }
+
+    fn get(&self, message_id: MessageId) -> Result<Message, Error> {
+        let messages = self.messages.lock().unwrap();
+
+        messages
+            .clone()
+            .into_iter()
+            .find(|msg| msg.id.eq(&message_id))
+            .ok_or_else(|| EntityNotFound("Message not found".to_string()))
+    }
 }