fix: msg_id for internal message and allow custom trace_id (#46)

* fix: msg_id for internal message and allow custom trace_id
* feat: Add ServiceError code to help for metrics

Signed-off-by: Jeremy HERGAULT <jeremy.hergault@worldline.com>

Author: reneca

prosa/examples/proc.rs

@@ -38,7 +38,6 @@ where
             .add_service_proc(vec![String::from("PROC_TEST")])
             .await?;
         let mut interval = time::interval(time::Duration::from_secs(4));
-        let mut msg_id: u64 = 0;
         let mut pending_msgs: PendingMsgs<RequestMsg<M>, M> = Default::default();
         loop {
             tokio::select! {
@@ -80,17 +79,15 @@ where
                     tvf.put_string(2, String::from("request"));
 
                     let stub_service_name = String::from("STUB_TEST");
-                    if let Some(service) = self.service.get_proc_service(&stub_service_name, msg_id) {
+                    if let Some(service) = self.service.get_proc_service(&stub_service_name) {
                         debug!("The service is find: {:?}", service);
-                        service.proc_queue.send(InternalMsg::Request(RequestMsg::new(msg_id, stub_service_name, tvf.clone(), self.proc.get_service_queue()))).await.unwrap();
-                        msg_id += 1;
+                        service.proc_queue.send(InternalMsg::Request(RequestMsg::new(stub_service_name, tvf.clone(), self.proc.get_service_queue()))).await.unwrap();
                     }
 
                     let proc_service_name = String::from("PROC_TEST");
-                    if let Some(service) = self.service.get_proc_service(&proc_service_name, msg_id) {
+                    if let Some(service) = self.service.get_proc_service(&proc_service_name) {
                         debug!("The service is find: {:?}", service);
-                        service.proc_queue.send(InternalMsg::Request(RequestMsg::new(msg_id, proc_service_name, tvf, self.proc.get_service_queue()))).await.unwrap();
-                        msg_id += 1;
+                        service.proc_queue.send(InternalMsg::Request(RequestMsg::new(proc_service_name, tvf, self.proc.get_service_queue()))).await.unwrap();
                     }
                 },
                 Some(msg) = pending_msgs.pull(), if !pending_msgs.is_empty() => {

prosa/src/core/msg.rs

@@ -1,12 +1,12 @@
 use std::{
-    sync::Arc,
+    sync::{Arc, atomic},
     time::{Duration, SystemTime},
 };
 
 use prosa_utils::msg::tvf::Tvf;
 use tokio::sync::mpsc;
-use tracing::span;
 use tracing::{Level, Span, event};
+use tracing::{info_span, span};
 
 use crate::core::error::BusError;
 
@@ -139,6 +139,8 @@ where
         P: FnOnce(&mut M) -> bool;
 }
 
+pub(crate) static ATOMIC_INTERNAL_MSG_ID: atomic::AtomicU64 = atomic::AtomicU64::new(0);
+
 /// ProSA request message that define a data message that need to be process by a processor
 #[derive(Debug)]
 pub struct RequestMsg<M>
@@ -206,16 +208,30 @@ where
     M: Sized + Clone + Tvf,
 {
     /// Method to create a new RequestMessage
-    pub fn new(
-        id: u64,
+    pub fn new(service: String, data: M, response_queue: mpsc::Sender<InternalMsg<M>>) -> Self {
+        let begin_time = SystemTime::now();
+        let span = info_span!("prosa::Msg", service = service);
+        RequestMsg {
+            id: ATOMIC_INTERNAL_MSG_ID.fetch_add(1, atomic::Ordering::Relaxed),
+            service,
+            data: Some(data),
+            begin_time,
+            span,
+            response_queue,
+        }
+    }
+
+    /// Method to create a new RequestMessage with a specific trace Id
+    pub fn new_with_trace_id(
         service: String,
         data: M,
         response_queue: mpsc::Sender<InternalMsg<M>>,
+        trace_id: tracing::span::Id,
     ) -> Self {
         let begin_time = SystemTime::now();
-        let span = span!(Level::INFO, "prosa::Msg", service = service);
+        let span = info_span!(parent: trace_id, "prosa::Msg", service = service);
         RequestMsg {
-            id,
+            id: ATOMIC_INTERNAL_MSG_ID.fetch_add(1, atomic::Ordering::Relaxed),
             service,
             data: Some(data),
             begin_time,
@@ -389,18 +405,6 @@ impl<M> ErrorMsg<M>
 where
     M: Sized + Clone + Tvf,
 {
-    /// Method to create a new ErrorMsg
-    pub fn new(id: u64, service: String, span: Span, data: Option<M>, err: ServiceError) -> Self {
-        ErrorMsg {
-            id,
-            service,
-            span,
-            error_time: SystemTime::now(),
-            data,
-            err,
-        }
-    }
-
     /// Getter of the service error
     pub fn get_err(&self) -> &ServiceError {
         &self.err

prosa/src/core/service.rs

@@ -1,3 +1,5 @@
+use crate::core::msg;
+
 use super::{
     msg::InternalMsg,
     proc::{ProcBusParam, ProcParam},
@@ -6,6 +8,7 @@ use prosa_utils::msg::tvf::{Tvf, TvfError};
 use std::{
     collections::HashMap,
     fmt::{self, Debug},
+    sync::atomic,
 };
 use thiserror::Error;
 use tokio::sync::mpsc;
@@ -47,10 +50,13 @@ where
     /// Method to get a processor that respond to the service
     ///
     /// Call by the processor to send a transaction to a processor that give the corresponding service
-    pub fn get_proc_service(&self, name: &str, msg_id: u64) -> Option<&ProcService<M>> {
+    pub fn get_proc_service(&self, name: &str) -> Option<&ProcService<M>> {
         if let Some(services) = self.table.get(name) {
             match services.len() {
-                2.. => services.get(msg_id as usize % services.len()),
+                2.. => services.get(
+                    msg::ATOMIC_INTERNAL_MSG_ID.load(atomic::Ordering::Relaxed) as usize
+                        % services.len(),
+                ),
                 1 => services.first(),
                 _ => None,
             }
@@ -231,6 +237,22 @@ pub enum ServiceError {
     ProtocolError(String),
 }
 
+impl ServiceError {
+    /// Method to get the error code of the service error
+    /// - 0: No error
+    /// - 1: Unable to reach service
+    /// - 2: Timeout
+    /// - 3: Protocol error
+    pub fn get_code(&self) -> u8 {
+        match self {
+            ServiceError::NoError(_) => 0,
+            ServiceError::UnableToReachService(_) => 1,
+            ServiceError::Timeout(_, _) => 2,
+            ServiceError::ProtocolError(_) => 3,
+        }
+    }
+}
+
 impl From<TvfError> for ServiceError {
     fn from(err: TvfError) -> Self {
         match err {

prosa/src/event/pending.rs

@@ -260,7 +260,7 @@ where
     ///     let mut pending_msg: PendingMsgs<RequestMsg<SimpleStringTvf>, SimpleStringTvf> = Default::default();
     ///     let mut msg: Option<RequestMsg<SimpleStringTvf>> = pending_msg.pull().await;
     ///     assert!(msg.is_none());
-    ///     pending_msg.push(RequestMsg::new(1, String::from("service"), tvf, queue), Duration::from_millis(200));
+    ///     pending_msg.push(RequestMsg::new(String::from("service"), tvf, queue), Duration::from_millis(200));
     ///     msg = pending_msg.pull().await;
     ///     assert!(msg.is_some());
     ///     println!("Timeout message {:?}", msg);
@@ -344,8 +344,8 @@ mod tests {
                                 assert_eq!(1, pending_timer.len());
                             },
                             InternalMsg::Service(table) => {
-                                if let Some(service) = table.get_proc_service("TEST", 1) {
-                                    service.proc_queue.send(InternalMsg::Request(RequestMsg::new(1, String::from("TEST"), Default::default(), self.proc.get_service_queue().clone()))).await.unwrap();
+                                if let Some(service) = table.get_proc_service("TEST") {
+                                    service.proc_queue.send(InternalMsg::Request(RequestMsg::new(String::from("TEST"), Default::default(), self.proc.get_service_queue().clone()))).await.unwrap();
                                 }
                             },
                             _ => return Err(BusError::ProcComm(self.get_proc_id(), 0, String::from("Wrong message"))),
@@ -380,10 +380,10 @@ mod tests {
                                 assert_eq!(1, pending_msg.len());
                             },
                             InternalMsg::Service(table) => {
-                                if let Some(service) = table.get_proc_service("TEST", 1) {
+                                if let Some(service) = table.get_proc_service("TEST") {
                                     let mut msg: SimpleStringTvf = Default::default();
                                     msg.put_string(1, "good");
-                                    service.proc_queue.send(InternalMsg::Request(RequestMsg::new(1, String::from("TEST"), msg, self.proc.get_service_queue().clone()))).await.unwrap();
+                                    service.proc_queue.send(InternalMsg::Request(RequestMsg::new(String::from("TEST"), msg, self.proc.get_service_queue().clone()))).await.unwrap();
                                 }
                             },
                             _ => return Err(BusError::ProcComm(self.get_proc_id(), 0, String::from("Wrong message"))),

prosa/src/inj/proc.rs

@@ -212,7 +212,6 @@ where
         // Create a message regulator
         let mut regulator = self.settings.get_regulator();
         let mut next_transaction = Some(adaptor.build_transaction());
-        let mut msg_id: u64 = 0;
 
         // Wait for service table
         while !self.service.exist_proc_service(&self.settings.service_name) {
@@ -231,17 +230,15 @@ where
 
         // Send first transaction
         self.service
-            .get_proc_service(&self.settings.service_name, msg_id)
+            .get_proc_service(&self.settings.service_name)
             .unwrap()
             .proc_queue
             .send(InternalMsg::Request(RequestMsg::new(
-                msg_id,
                 self.settings.service_name.clone(),
                 next_transaction.take().unwrap(),
                 self.proc.get_service_queue(),
             )))
             .await?;
-        msg_id += 1;
         regulator.notify_send_transaction();
 
         loop {
@@ -250,17 +247,16 @@ where
                     self.process_internal(name.as_str(), msg, &mut adaptor, &mut regulator, &mut next_transaction, &meter_trans_duration).await?;
                 }
                 _ = regulator.tick() => {
-                    if let Some(service) = self.service.get_proc_service(&self.settings.service_name, msg_id) {
+                    if let Some(service) = self.service.get_proc_service(&self.settings.service_name) {
                         let trans = if let Some(transaction) = next_transaction.take() {
-                            RequestMsg::new(msg_id, self.settings.service_name.clone(), transaction, self.proc.get_service_queue())
+                            RequestMsg::new(self.settings.service_name.clone(), transaction, self.proc.get_service_queue())
                         } else {
-                            RequestMsg::new(msg_id, self.settings.service_name.clone(), adaptor.build_transaction(), self.proc.get_service_queue())
+                            RequestMsg::new(self.settings.service_name.clone(), adaptor.build_transaction(), self.proc.get_service_queue())
                         };
 
                         debug!(name: "inj_proc", target: "prosa::inj::proc", parent: trans.get_span(), proc_name = name, service = self.settings.service_name, request = format!("{:?}", trans.get_data()));
                         service.proc_queue.send(InternalMsg::Request(trans)).await?;
 
-                        msg_id += 1;
                         regulator.notify_send_transaction();
                     }
                 },

prosa/src/lib.rs

@@ -93,7 +93,7 @@ mod tests {
             request: M,
         ) -> MaybeAsync<Result<M, ServiceError>> {
             assert!(!request.is_empty());
-            COUNTER.fetch_add(1, Ordering::Relaxed);
+            COUNTER.fetch_add(1, Ordering::SeqCst);
             Ok(request.clone()).into()
         }
     }
@@ -126,7 +126,7 @@ mod tests {
         main_task.await.unwrap();
 
         // Check exchanges messages
-        let nb_trans = COUNTER.load(Ordering::Relaxed) as u64;
+        let nb_trans = COUNTER.load(Ordering::SeqCst) as u64;
         let estimated_trans = WAIT_TIME.as_secs() * 5;
         assert!(nb_trans > (estimated_trans - 2) && nb_trans < (estimated_trans + 2));
         // Should have a coherent number of transaction with the regulator

prosa_book/src/ch03-05-service.md

@@ -165,9 +165,6 @@ After that, you are free to call any services.
         // Register the processor
         self.proc.add_proc().await?;
 
-        // Need to have a unique message id when you send a message
-        let mut msg_id = 0;
-
         // Wait for the service table before sending messages to a service
         loop {
             if let Some(msg) = self.internal_rx_queue.recv().await {
@@ -192,16 +189,13 @@ After that, you are free to call any services.
             }
 
             // Attempt to send a message if the service is available
-            if let Some(service) = self.service.get_proc_service("SERVICE_NAME", msg_id) {
+            if let Some(service) = self.service.get_proc_service("SERVICE_NAME") {
                 let trans = RequestMsg::new(
-                    msg_id,
                     String::from("SERVICE_NAME"),
                     M::default(),
                     self.proc.get_service_queue()
                 );
                 service.proc_queue.send(InternalMsg::Request(trans)).await?;
-
-                msg_id += 1;
             }
         }
 
@@ -228,9 +222,6 @@ The logic is similar to single senders, but you specify the queue when sending m
         // Register the processor
         self.proc.add_proc().await?;
 
-        // Need to have a unique message id when you send a message
-        let mut msg_id = 0;
-
         // Create a queue for subtask communication
         let (tx_queue, mut rx_queue) = tokio::sync::mpsc::channel(2048);
         let tx_msg_queue = tx_queue.clone();
@@ -265,16 +256,13 @@ The logic is similar to single senders, but you specify the queue when sending m
             }
 
             // Attempt to send a message if the service is available
-            if let Some(service) = self.service.get_proc_service("SERVICE_NAME", msg_id) {
+            if let Some(service) = self.service.get_proc_service("SERVICE_NAME") {
                 let trans = RequestMsg::new(
-                    msg_id,
                     String::from("SERVICE_NAME"),
                     M::default(),
                     tx_msg_queue.clone()
                 );
                 service.proc_queue.send(InternalMsg::Request(trans)).await?;
-
-                msg_id += 1;
             }
         })
 

prosa_book/src/ch03-06-events.md

@@ -33,7 +33,6 @@ There are three important methods you need to use for this object:
             .add_service_proc(vec![String::from("PROC_TEST")])
             .await?;
         let mut interval = time::interval(time::Duration::from_secs(4));
-        let mut msg_id: u64 = 0;
         let mut pending_msgs: PendingMsgs<RequestMsg<M>, M> = Default::default();
         loop {
             tokio::select! {