fix: fixed statistics counters to properly record incoming/outgoing/f…

…ailed/broken bundles

core/dtn7/src/cla/httppull.rs

@@ -89,6 +89,7 @@ async fn http_pull_from_node(
             let bundle = match bp7::Bundle::try_from(bundle_buf.as_ref()) {
                 Ok(bundle) => bundle,
                 Err(e) => {
+                    crate::STATS.lock().broken += 1;
                     warn!("could not parse bundle from remote: {}", e);
                     continue;
                 }

core/dtn7/src/cla/mtcp.rs

@@ -329,6 +329,7 @@ impl MtcpConvergenceLayer {
                             });
                         }
                     } else {
+                        crate::STATS.lock().broken += 1;
                         info!("Error decoding bundle from {}", peer_addr);
                         break;
                     }

core/dtn7/src/cla/tcp/mod.rs

@@ -235,6 +235,7 @@ impl TcpSession {
             Err(err) => {
                 error!("Failed to parse bundle: {}", err);
                 //error!("Failed bytes: {}", bp7::helpers::hexify(&vec));
+                crate::STATS.lock().broken += 1;
                 TcpClPacket::XferRefuse(XferRefuseData {
                     reason: XferRefuseReasonCode::NotAcceptable,
                     tid,

core/dtn7/src/core/mod.rs

@@ -34,6 +34,7 @@ pub struct DtnStatistics {
     pub dups: u64,
     pub outgoing: u64,
     pub delivered: u64,
+    pub failed: u64,
     pub broken: u64,
 }
 
@@ -44,6 +45,7 @@ impl DtnStatistics {
             dups: 0,
             outgoing: 0,
             delivered: 0,
+            failed: 0,
             broken: 0,
         }
     }

core/dtn7/src/core/processing.rs

@@ -97,11 +97,13 @@ pub async fn transmit(mut bp: BundlePack) -> Result<()> {
 pub async fn receive(mut bndl: Bundle) -> Result<()> {
     if store_add_bundle_if_unknown(&bndl)? {
         info!("Received new bundle: {}", bndl.id());
+        STATS.lock().incoming += 1;
     } else {
         debug!(
             "Received an already known bundle, skip processing: {}",
             bndl.id()
         );
+        STATS.lock().dups += 1;
 
         // bundleDeletion is _not_ called because this would delete the already
         // stored BundlePack.
@@ -360,6 +362,7 @@ pub async fn forward(mut bp: BundlePack) -> Result<()> {
                         n.next_hop,
                         start_time.elapsed()
                     );
+                    STATS.lock().failed += 1;
                     debug!("Error while transferring bundle {}: {}", &bpid, err);
                     let mut failed_peer = None;
 
@@ -402,6 +405,7 @@ pub async fn forward(mut bp: BundlePack) -> Result<()> {
                         n.cla_name,
                         start_time.elapsed()
                     );
+                    STATS.lock().outgoing += 1;
                     bundle_sent.store(true, Ordering::Relaxed);
                     if let Err(err) = routing_notify(RoutingNotifcation::SendingSucceeded(
                         bpid,

core/dtn7/src/dtnd/httpd.rs

@@ -539,6 +539,7 @@ async fn push_post(body: bytes::Bytes) -> Result<String, (StatusCode, String)> {
         //});
         Ok(format!("Received {} bytes", b_len))
     } else {
+        crate::STATS.lock().broken += 1;
         Err((
             StatusCode::BAD_REQUEST,
             "Error decoding bundle!".to_string(),