feat: streaming test app (#144)

Signed-off-by: Mauro Sardara <[email protected]>
Signed-off-by: Michele Papalini <[email protected]>
Co-authored-by: Mauro Sardara <[email protected]>

Author: micpapal

data-plane/gateway/service/src/streaming.rs

@@ -242,7 +242,7 @@ impl Streaming {
                                                             &source,
                                                             pkt_src.agent_type(),
                                                             Some(pkt_src.agent_id()),
-                                                            Some(AgpHeaderFlags::default().with_incoming_conn(incoming_conn)),
+                                                            Some(AgpHeaderFlags::default().with_forward_to(incoming_conn)),
                                                         ));
 
                                                         let session_header = Some(SessionHeader::new(
@@ -340,14 +340,14 @@ impl Streaming {
                                                     }
                                                 }
                                                 for r in rtx {
-                                                    debug!("send a rtx for message {} from receiver session {}", r, session_id);
+                                                    debug!("packet loss detected on session {}, send RTX for id {}", session_id, r);
                                                     let dest = producer_name.as_ref().unwrap(); // this cannot panic a this point
 
                                                     let agp_header = Some(AgpHeader::new(
                                                         &source,
                                                         dest.agent_type(),
                                                         Some(dest.agent_id()),
-                                                        Some(AgpHeaderFlags::default().with_incoming_conn(producer_conn.unwrap())),
+                                                        Some(AgpHeaderFlags::default().with_forward_to(producer_conn.unwrap())),
                                                     ));
 
                                                     let session_header = Some(SessionHeader::new(
@@ -478,7 +478,6 @@ impl Session for Streaming {
         message: SessionMessage,
         direction: MessageDirection,
     ) -> Result<(), SessionError> {
-        debug!("receive message");
         self.tx
             .send(Ok((message.message, direction)))
             .await
@@ -801,14 +800,15 @@ mod tests {
             tx_app_receiver,
         );
 
-        let message = Message::new_publish(
+        let mut message = Message::new_publish(
             &Agent::from_strings("cisco", "default", "sender", 0),
             &AgentType::from_strings("cisco", "default", "receiver"),
             Some(0),
             Some(AgpHeaderFlags::default().with_incoming_conn(0)),
             "msg",
             vec![0x1, 0x2, 0x3, 0x4],
         );
+        message.set_incoming_conn(Some(0));
 
         let session_msg: SessionMessage = SessionMessage::new(message.clone(), Info::new(0));
         // send 3 messages from the producer app
@@ -823,14 +823,16 @@ mod tests {
         // read the 3 messages from the sender gw channel
         // forward message 1 and 3 to the receiver
         for i in 0..3 {
-            let msg = rx_gw_sender.recv().await.unwrap().unwrap();
+            let mut msg = rx_gw_sender.recv().await.unwrap().unwrap();
             let msg_header = msg.get_session_header();
             assert_eq!(msg_header.session_id, 0);
             assert_eq!(msg_header.message_id, i);
             assert_eq!(msg_header.header_type, SessionHeaderType::Stream.into());
 
             // the receiver should detect a loss for packet 1
             if i != 1 {
+                // make sure to set the incoming connection to avoid paninc
+                msg.set_incoming_conn(Some(0));
                 let session_msg: SessionMessage = SessionMessage::new(msg.clone(), Info::new(0));
                 let res = receiver
                     .on_message(session_msg.clone(), MessageDirection::North)
@@ -893,7 +895,9 @@ mod tests {
         );
 
         // send the second reply to the producer
-        let session_msg: SessionMessage = SessionMessage::new(msg.clone(), Info::new(0));
+        let mut session_msg: SessionMessage = SessionMessage::new(msg.clone(), Info::new(0));
+        // make sure to set the incoming connection to avoid paninc
+        session_msg.message.set_incoming_conn(Some(0));
         let res = sender
             .on_message(session_msg.clone(), MessageDirection::North)
             .await;
@@ -917,7 +921,9 @@ mod tests {
             )
         );
 
-        let session_msg: SessionMessage = SessionMessage::new(msg.clone(), Info::new(0));
+        let mut session_msg: SessionMessage = SessionMessage::new(msg.clone(), Info::new(0));
+        // make sure to set the incoming connection to avoid paninc
+        session_msg.message.set_incoming_conn(Some(0));
         let res = receiver
             .on_message(session_msg.clone(), MessageDirection::North)
             .await;

data-plane/testing/src/bin/publisher.rs

@@ -5,11 +5,14 @@ use std::fs::File;
 use std::io::prelude::*;
 use std::{collections::HashMap, sync::Arc};
 
-use agp_datapath::messages::Agent;
+use agp_datapath::messages::{Agent, AgentType};
+use agp_service::AgpHeaderFlags;
 use parking_lot::RwLock;
 use testing::parse_line;
 use tokio_util::sync::CancellationToken;
 
+use agp_service::streaming::StreamingConfiguration;
+
 use agp_gw::config;
 use clap::Parser;
 use indicatif::ProgressBar;
@@ -18,9 +21,19 @@ use tracing::{debug, error, info};
 #[derive(Parser, Debug)]
 #[command(version, about, long_about = None)]
 pub struct Args {
-    /// Workload input file
-    #[arg(short, long, value_name = "WORKLOAD", required = true)]
-    workload: String,
+    /// Workload input file, required if used in workload mode. If this is set the streaming mode is set to false.
+    #[arg(short, long, value_name = "WORKLOAD", required = false)]
+    workload: Option<String>,
+
+    /// Runs in streaming mode.
+    #[arg(
+        short,
+        long,
+        value_name = "STREAMING",
+        required = false,
+        default_value_t = false
+    )]
+    streaming: bool,
 
     /// Agp config file
     #[arg(short, long, value_name = "CONFIGURATION", required = true)]
@@ -54,22 +67,30 @@ pub struct Args {
     #[arg(
         short,
         long,
-        value_name = "SLEEP",
+        value_name = "FREQUENCY",
         required = false,
         default_value_t = 0
     )]
-    sleep: u32,
+    frequency: u32,
+
+    /// used only in streaming mode, defines the maximum number of packets to send
+    #[arg(short, long, value_name = "PACKETS", required = false)]
+    max_packets: Option<u64>,
 }
 
 impl Args {
     pub fn msg_size(&self) -> &u32 {
         &self.msg_size
     }
 
-    pub fn workload(&self) -> &String {
+    pub fn workload(&self) -> &Option<String> {
         &self.workload
     }
 
+    pub fn streaming(&self) -> &bool {
+        &self.streaming
+    }
+
     pub fn id(&self) -> &u64 {
         &self.id
     }
@@ -82,8 +103,12 @@ impl Args {
         &self.quite
     }
 
-    pub fn sleep(&self) -> &u32 {
-        &self.sleep
+    pub fn frequency(&self) -> &u32 {
+        &self.frequency
+    }
+
+    pub fn max_packets(&self) -> &Option<u64> {
+        &self.max_packets
     }
 }
 
@@ -95,22 +120,137 @@ async fn main() {
     let config_file = args.config();
     let msg_size = *args.msg_size();
     let id = *args.id();
-    let sleep = *args.sleep();
+    let frequency = *args.frequency();
+    let mut streaming = *args.streaming();
+    let max_packets = args.max_packets;
+    if input.is_some() {
+        streaming = false;
+    }
 
-    // setup agent config
+    info!(
+        "configuration -- workload file: {}, agent config {}, publisher id: {}, streaming mode: {}, msg size: {}",
+        input.as_ref().unwrap_or(&"None".to_string()),
+        config_file,
+        id,
+        streaming,
+        msg_size,
+    );
+
+    // start local agent
+    // get service
     let mut config = config::load_config(config_file).expect("failed to load configuration");
     let _guard = config.tracing.setup_tracing_subscriber();
+    let svc_id = agp_config::component::id::ID::new_with_str("gateway/0").unwrap();
+    let svc = config.services.get_mut(&svc_id).unwrap();
 
-    info!(
-        "configuration -- input file: {}, agent config: {}, msg size: {}",
-        input, config_file, msg_size
-    );
+    // create local agent
+    let agent_name = Agent::from_strings("cisco", "default", "publisher", id);
+    // required in streaming mode
+    let dest_name = AgentType::from_strings("cisco", "default", "subscriber");
+    let mut rx = svc
+        .create_agent(&agent_name)
+        .expect("failed to create agent");
+
+    // connect to the remote gateway
+    let conn_id = svc.connect(None).await.unwrap();
+    info!("remote connection id = {}", conn_id);
+
+    // subscribe for local name
+    match svc
+        .subscribe(
+            &agent_name,
+            agent_name.agent_type(),
+            agent_name.agent_id_option(),
+            Some(conn_id),
+        )
+        .await
+    {
+        Ok(_) => {}
+        Err(e) => {
+            panic!("an error accoured while adding a subscription {}", e);
+        }
+    }
+
+    svc.set_route(&agent_name, &dest_name, None, conn_id)
+        .await
+        .unwrap();
+
+    // wait for the connection to be established
+    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
 
+    // STREAMING MODE
+    if streaming {
+        // create streaming session
+        let res = svc
+            .create_session(
+                &agent_name,
+                agp_service::session::SessionConfig::Streaming(StreamingConfiguration {
+                    source: agent_name.clone(),
+                    max_retries: None,
+                    timeout: None,
+                }),
+            )
+            .await;
+        if res.is_err() {
+            panic!("error creating fire and forget session");
+        }
+
+        // loop to listen to errors coming from the local gateway
+        tokio::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    None => {
+                        info!(%conn_id, "end of stream");
+                        break;
+                    }
+                    Some(msg_info) => {
+                        if msg_info.is_err() {
+                            error!("received an error message {:?}", msg_info);
+                            continue;
+                        } else {
+                            panic!("received a message from the gateway, this should never happen");
+                        }
+                    }
+                }
+            }
+        });
+
+        // get the session
+        let session_info = res.unwrap();
+
+        for i in 0..max_packets.unwrap_or(u64::MAX) {
+            let payload: Vec<u8> = vec![120; msg_size as usize]; // ASCII for 'x' = 120
+            info!("publishing message {}", i);
+            // set fanout > 1 to send the message in broadcast
+            let flags = AgpHeaderFlags::new(10, None, None, None, None);
+            if svc
+                .publish_with_flags(
+                    &agent_name,
+                    session_info.clone(),
+                    &dest_name,
+                    None,
+                    flags,
+                    payload,
+                )
+                .await
+                .is_err()
+            {
+                error!("an error occurred sending publication, the test will fail",);
+            }
+            if frequency != 0 {
+                tokio::time::sleep(tokio::time::Duration::from_millis(frequency as u64)).await;
+            }
+        }
+        return;
+    }
+
+    // WORKLOAD MODE
+    // setup agent config
     let mut publication_list = HashMap::new();
     let mut oracle = HashMap::new();
     let mut routes = Vec::new();
 
-    let res = File::open(input);
+    let res = File::open(input.as_ref().unwrap());
     if res.is_err() {
         panic!("error opening the input file");
     }
@@ -142,37 +282,6 @@ async fn main() {
         }
     }
 
-    // start local agent
-    // get service
-    let svc_id = agp_config::component::id::ID::new_with_str("gateway/0").unwrap();
-    let svc = config.services.get_mut(&svc_id).unwrap();
-
-    // create local agent
-    let agent_name = Agent::from_strings("cisco", "default", "publisher", id);
-    let mut rx = svc
-        .create_agent(&agent_name)
-        .expect("failed to create agent");
-
-    // connect to the remote gateway
-    let conn_id = svc.connect(None).await.unwrap();
-    info!("remote connection id = {}", conn_id);
-
-    // subscribe for local name
-    match svc
-        .subscribe(
-            &agent_name,
-            agent_name.agent_type(),
-            agent_name.agent_id_option(),
-            Some(conn_id),
-        )
-        .await
-    {
-        Ok(_) => {}
-        Err(e) => {
-            panic!("an error accoured while adding a subscription {}", e);
-        }
-    }
-
     // set routes for all subscriptions
     for r in routes {
         match svc
@@ -317,8 +426,8 @@ async fn main() {
             bar.inc(1);
         }
 
-        if sleep != 0 {
-            tokio::time::sleep(tokio::time::Duration::from_millis(sleep as u64)).await;
+        if frequency != 0 {
+            tokio::time::sleep(tokio::time::Duration::from_millis(frequency as u64)).await;
         }
     }
     let duration = start.elapsed();