refactor: split message from grpc message

Cargo.toml

@@ -1,12 +1,12 @@
-[workspace]
-
-members = [
-    "queues",
-    "exchanges",
-    "protos",
-    "ruuster_grpc",
-    "utils",
-    "apps"
-]
-
-resolver = "2"
+[workspace]
+
+members = [
+    "queues",
+    "exchanges",
+    "protos",
+    "ruuster_grpc",
+    "utils",
+    "apps"
+, "internals"]
+
+resolver = "2"

README.md

@@ -4,15 +4,18 @@
 
 # Setup
 
+To run you can use GitPod (super fast), devcontainers (local docker + vscode) or just linux box
+
 1. Clone the repo
 1. Open in vscode
-1. In vscode run command "reopen in devcontainer"
+1. Run `docker compose up -d` to start logs/traces collector UI: http://localhost:16686
 1. Build & run in the terminal
 
-```rs
-// to run the server
-cargo run --bin server --release'
-```
+```bash
+# to run the server
+cargo watch -x 'run --bin server'  --ignore protos/ #gitpod keeps regenerating proto files
+# to have watch on tests
+cargo watch -x 'test'  --ignore protos/ #gitpod keeps regenerating proto files
 
 ```bash
 # to run docker for local tracing and metrics

docker-compose.yml

@@ -0,0 +1,11 @@
+version: "3.9"
+
+services:
+  jaeger:
+    image: jaegertracing/all-in-one:latest
+    ports:
+      - "6831:6831/udp"
+      - "6832:6832/udp"
+      - "16686:16686"
+      - "14268:14268"
+      - "4317:4317"

exchanges/Cargo.toml

@@ -12,9 +12,11 @@ path = "src/exchanges.rs"
 tracing = "0.1"
 prost = "0.12.4"
 tonic = "0.11.0"
-uuid = { version = "1.8.0", features = ["v4"]}
+uuid = { version = "1.8.0", features = ["v4"] }
 protos = { path = "../protos" }
-serde = "1.0.201"
+serde = { version = "1.0.201", features = ["derive"] }
 serde_json = "1.0.117"
 lazy_static = "1.4.0"
 env_logger = "0.11.3"
+
+internals = { path = '../internals' }

exchanges/src/exchanges.rs

@@ -1,13 +1,12 @@
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::fmt::{self, Display};
 use std::sync::{Arc, Mutex, RwLock};
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::Instant;
 
-use serde_json::json;
+use internals::{DeadLetterMetadata, Message, Metadata};
+use serde::{Deserialize, Serialize};
 
-use protos::ruuster::Message;
-use protos::ruuster::Metadata;
-use tracing::{debug, error, warn};
+use tracing::{debug, error, info, warn};
 use types::{DirectExchange, FanoutExchange};
 
 pub mod types;
@@ -30,7 +29,6 @@ pub enum ExchangeKind {
 pub enum ExchangeError {
     BindFail,
     EmptyPayloadFail,
-    GetSystemTimeFail {},
     DeadLetterQueueLockFail {},
     NoRouteKey,
     NoMatchingQueue { route_key: String },
@@ -45,12 +43,6 @@ impl fmt::Display for ExchangeError {
             ExchangeError::EmptyPayloadFail => {
                 write!(f, "handling message failed")
             }
-            ExchangeError::GetSystemTimeFail {} => {
-                write!(
-                    f,
-                    "handling message failed: SystemTime::now().duration_since"
-                )
-            }
             ExchangeError::DeadLetterQueueLockFail {} => write!(
                 f,
                 "handling message failed: queue().lock() failed for dead letter queue"
@@ -67,7 +59,11 @@ impl fmt::Display for ExchangeError {
 }
 
 pub trait Exchange {
-    fn bind(&mut self, queue_name: &QueueName, metadata: Option<&Metadata>) -> Result<(), ExchangeError>;
+    fn bind(
+        &mut self,
+        queue_name: &QueueName,
+        metadata: Option<&protos::BindMetadata>,
+    ) -> Result<(), ExchangeError>;
     fn get_bound_queue_names(&self) -> HashSet<QueueName>;
     fn handle_message(
         &self,
@@ -139,39 +135,39 @@ pub(crate) trait PushToQueueStrategy {
             HashMap<String, Arc<Mutex<VecDeque<Message>>>>,
         >,
     ) -> Result<PushResult, ExchangeError> {
-        let timestamp = SystemTime::now()
-            .duration_since(UNIX_EPOCH)
-            .map(|d| d.as_millis() as i64)
-            .map_err(|_| ExchangeError::GetSystemTimeFail {})?;
-
         let queue_lock = &mut queue.lock().unwrap();
         if queue_lock.len() >= QUEUE_MAX_LENGTH {
             warn!("queue size reached for queue {}", name);
             if let Some(dead_letter_queue) = queues_read.get(DEADLETTER_QUEUE_NAME) {
-                // FIXME: use the deadletter queue defined per exchange
-                debug!("moving the message {} to the dead letter queue", message.uuid);
-
-                // FIXME: convert to ruuster headers
-                let val = json!({
-                    "count": 1,
-                    "exchange": exchange_name,
-                    "original_message": message.payload,
-                    "reason": "max_len",
-                    "time": timestamp,
-                    "queue": name.to_string(),
-                })
-                .to_string();
+                // TODO: use the deadletter queue defined per exchange
+                debug!(
+                    "moving the message {} to the dead letter queue",
+                    message.uuid
+                );
+                let mut meta = message.metadata.unwrap_or(Metadata {
+                    created_at: Some(Instant::now()),
+                    routing_key: None,
+                    dead_letter: None,
+                });
+                let deadletter_metadata = DeadLetterMetadata {
+                    count: Some(1),
+                    exchange: Some(exchange_name.to_string()),
+                    queue: Some(name.to_string()),
+                };
+                meta.dead_letter = Some(deadletter_metadata);
+
+                let message = Message {
+                    uuid: message.uuid,
+                    payload: message.payload,
+                    metadata: Some(meta),
+                };
 
                 dead_letter_queue
                     .lock()
                     .map_err(|_| ExchangeError::DeadLetterQueueLockFail {})?
-                    .push_back(Message {
-                        uuid: message.uuid,
-                        payload: val,
-                        metadata: None
-                    });
+                    .push_back(message);
             } else {
-                debug!("message {} dropped", message.uuid);
+                info!("message {} dropped", message.uuid);
             }
             return Ok(PushResult::QueueOverflow);
         } else {
@@ -180,3 +176,12 @@ pub(crate) trait PushToQueueStrategy {
         }
     }
 }
+
+#[derive(Serialize, Deserialize)]
+struct DeadLetterMessage {
+    count: i32,
+    exchange: String,
+    original_message: String,
+    time: i64,
+    queue: String,
+}