feat: new message format (#88)

Signed-off-by: Michele Papalini <[email protected]>

Author: micpapal

data-plane/examples/src/sdk-mock/main.rs

@@ -4,7 +4,7 @@
 use clap::Parser;
 use tracing::info;
 
-use agp_datapath::messages::encoder::{encode_agent_class, encode_agent_from_string};
+use agp_datapath::messages::encoder::{encode_agent, encode_agent_type};
 use agp_gw::config;
 
 mod args;
@@ -38,20 +38,20 @@ async fn main() {
 
     // create local agent
     let agent_id = 0;
-    let agent_name = encode_agent_from_string("cisco", "default", local_agent, agent_id);
+    let agent_name = encode_agent("cisco", "default", local_agent, agent_id);
     let mut rx = svc.create_agent(agent_name.clone());
 
     // connect to the remote gateway
     let conn_id = svc.connect(None).await.unwrap();
     info!("remote connection id = {}", conn_id);
 
-    let local_agent_class = encode_agent_class("cisco", "default", local_agent);
-    svc.subscribe(&local_agent_class, Some(agent_id), conn_id)
+    let local_agent_type = encode_agent_type("cisco", "default", local_agent);
+    svc.subscribe(&local_agent_type, Some(agent_id), conn_id)
         .await
         .unwrap();
 
     // Set a route for the remote agent
-    let route = encode_agent_class("cisco", "default", remote_agent);
+    let route = encode_agent_type("cisco", "default", remote_agent);
     info!("allowing messages to remote agent: {:?}", route);
     svc.set_route(&route, None, conn_id).await.unwrap();
 

data-plane/gateway/datapath/pubsub/proto/v1/pubsub.proto

@@ -9,39 +9,78 @@ service PubSubService {
 }
 
 message Subscribe {
-  AgentId source = 1;
-  AgentId name = 2;
+  AGPHeader header = 1;
 }
 
 message Unsubscribe {
-  AgentId source = 1;
-  AgentId name = 2;
+  AGPHeader header = 1;
 }
 
 message Publish {
-  AgentId source = 1;
-  AgentId name = 2;
+  AGPHeader header = 1;
+  ServiceHeader control = 2;
   uint32 fanout = 3;
   Content msg = 4;
 }
 
-message AgentGroup {
+// recvFrom = connection form where the sub/unsub is supposed to be received
+// forwardTo = connection where to forward the message
+// incomingConn = connection from where the packet was received
+// error = if true the publication contains an error notification
+message AGPHeader {
+  Agent source = 1;
+  Agent destination = 2;
+  optional uint64 recvFrom = 3;
+  optional uint64 forwardTo = 4;
+  optional uint64 incomingConn = 5;
+  optional bool error = 6;
+}
+
+message Agent {
   uint64 organization = 1;
   uint64 namespace = 2;
+  uint64 agent_type = 3;
+  optional uint64 agent_id = 4;
 }
 
-message AgentClass {
-  AgentGroup group = 1;
-  uint64 class = 2;
+enum ServiceHeaderType {
+  CTRL_UNSPECIFIED = 0;
+  CTRL_FNF = 1;
+  CTRL_REQUEST = 2;
+  CTRL_REPLY = 3;
+  CTRL_STREAM = 4;
+  CTRL_RTX_REQUEST = 5;
+  CTRL_RTX_REPLY = 6;
 }
 
-message AgentId {
-  AgentClass class = 1;
-  optional uint64 id = 2;
+// Service.id meaning according to ControlType
+// CTRL_FNF = nonce
+// CTRL_REQUEST = nonce
+// CTRL_REPLY = nonce of the associated CTRL_REQUEST
+// CTRL_STREAM = sequential incremental ID
+// CTRL_RTX_REQUEST = nonce
+// CTRL_RTX_REPLY = nonce of the associated CTRL_RTX_REQUEST
+
+// Service.stream meaning according to ControlType
+// CTRL_STREAM = stream for which the id field is valid
+// CTRL_RTX_REQUEST = stream for this CTRL_RTX_REQUEST
+// CTRL_RTX_REPLY = stream for this CTRL_RTX_REPLY
+// None in all the other cases
+
+// Service.rtx meaning according to ControlType
+// CTRL_RTX_REQUEST = id of the messing packet to retransmit
+// CTRL_RTX_REPLY = id of the retransmitted packet
+// None in all the other cases
+
+message ServiceHeader {
+  ServiceHeaderType header_type = 1;
+  uint32 id = 2;
+  optional uint32 stream = 3;
+  optional uint32 rtx = 4;
 }
 
 message Content {
-  string contentType = 1;
+  string content_type = 1;
   bytes blob = 2;
 }
 

data-plane/gateway/datapath/src/errors.rs

@@ -11,6 +11,8 @@ pub enum DataPathError {
     DisconnectionError(String),
     #[error("unkwon message type {0}")]
     UnknownMsgType(String),
+    #[error("unable to set incoming connection")]
+    ErrorSettingInConnection(String),
     #[error("error handling subscription: {0}")]
     SubscriptionError(String),
     #[error("error handling unsubscription: {0}")]

data-plane/gateway/datapath/src/forwarder.rs

@@ -9,7 +9,7 @@ use super::tables::remote_subscription_table::RemoteSubscriptions;
 use super::tables::subscription_table::SubscriptionTableImpl;
 use super::tables::{errors::SubscriptionTableError, SubscriptionTable};
 use crate::messages::encoder::DEFAULT_AGENT_ID;
-use crate::messages::{Agent, AgentClass};
+use crate::messages::{Agent, AgentType};
 use crate::tables::remote_subscription_table::SubscriptionInfo;
 
 #[derive(Debug)]
@@ -81,84 +81,72 @@ where
 
     pub fn on_subscription_msg(
         &self,
-        class: AgentClass,
+        agent_type: AgentType,
         agent_id: Option<u64>,
         conn_index: u64,
         is_local: bool,
     ) -> Result<(), SubscriptionTableError> {
         self.subscription_table
-            .add_subscription(class, agent_id, conn_index, is_local)
+            .add_subscription(agent_type, agent_id, conn_index, is_local)
     }
 
     pub fn on_forwarded_subscription(
         &self,
-        source_class: AgentClass,
+        source_type: AgentType,
         source_agent_id: Option<u64>,
-        name_class: AgentClass,
+        name_type: AgentType,
         name_agent_id: Option<u64>,
         conn_index: u64,
     ) {
-        let source = Agent {
-            agent_class: source_class,
-            agent_id: source_agent_id.unwrap_or(DEFAULT_AGENT_ID),
-        };
-        let name = Agent {
-            agent_class: name_class,
-            agent_id: name_agent_id.unwrap_or(DEFAULT_AGENT_ID),
-        };
+        let source = Agent::new(source_type, source_agent_id.unwrap_or(DEFAULT_AGENT_ID));
+        let name = Agent::new(name_type, name_agent_id.unwrap_or(DEFAULT_AGENT_ID));
         self.remote_subscription_table
             .add_subscription(source, name, conn_index);
     }
 
     pub fn on_unsubscription_msg(
         &self,
-        class: AgentClass,
+        agent_type: AgentType,
         agent_id: Option<u64>,
         conn_index: u64,
         is_local: bool,
     ) -> Result<(), SubscriptionTableError> {
         self.subscription_table
-            .remove_subscription(class, agent_id, conn_index, is_local)
+            .remove_subscription(agent_type, agent_id, conn_index, is_local)
     }
 
     pub fn on_forwarded_unsubscription(
         &self,
-        source_class: AgentClass,
+        source_type: AgentType,
         source_agent_id: Option<u64>,
-        name_class: AgentClass,
+        name_type: AgentType,
         name_agent_id: Option<u64>,
         conn_index: u64,
     ) {
-        let source = Agent {
-            agent_class: source_class,
-            agent_id: source_agent_id.unwrap_or(DEFAULT_AGENT_ID),
-        };
-        let name = Agent {
-            agent_class: name_class,
-            agent_id: name_agent_id.unwrap_or(DEFAULT_AGENT_ID),
-        };
+        let source = Agent::new(source_type, source_agent_id.unwrap_or(DEFAULT_AGENT_ID));
+        let name = Agent::new(name_type, name_agent_id.unwrap_or(DEFAULT_AGENT_ID));
         self.remote_subscription_table
             .remove_subscription(source, name, conn_index);
     }
 
     pub fn on_publish_msg_match_one(
         &self,
-        class: AgentClass,
+        agent_type: AgentType,
         agent_id: Option<u64>,
         incoming_conn: u64,
     ) -> Result<u64, SubscriptionTableError> {
         self.subscription_table
-            .match_one(class, agent_id, incoming_conn)
+            .match_one(agent_type, agent_id, incoming_conn)
     }
 
     pub fn on_publish_msg_match_all(
         &self,
-        class: AgentClass,
+        agent_type: AgentType,
         agent_id: Option<u64>,
         incoming_conn: u64,
     ) -> Result<Vec<u64>, SubscriptionTableError> {
         self.subscription_table
-            .match_all(class, agent_id, incoming_conn)
+            .match_all(agent_type, agent_id, incoming_conn)
     }
 
     #[allow(dead_code)]
@@ -170,13 +158,13 @@ where
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::messages::encoder::encode_agent_class;
+    use crate::messages::encoder::encode_agent_type;
     use tracing_test::traced_test;
 
     #[test]
     #[traced_test]
     fn test_forwarder() {
-        let agent_class = encode_agent_class("Cisco", "Default", "class_ONE");
+        let agent_class = encode_agent_type("Cisco", "Default", "class_ONE");
 
         let fwd = Forwarder::<u32>::new();