time blob messages

1 files changed, 19 insertions, 0 deletions

diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs
index 53f1aa3..f43a45a 100644
--- a/src/swarm/peer/inbound.rs
+++ b/src/swarm/peer/inbound.rs
@@ -15,6 +15,7 @@ use super::keepalive::{
     heal_if_diverged, heard_from, local_status, note_alive, note_status, status_received,
 };
 use super::lanes::SyncLanes;
+use crate::locks::MutexExt;
 use crate::state::AppState;
 use crate::swarm::sync::SyncEvent;
 use crate::transport::{self, ParsedMessage};
@@ -151,6 +152,24 @@ pub(crate) async fn handle_message(
             lanes.enqueue(&peer, envelope);
         }
         transport::ParsedMessage::Blob(blob) => {
+            // How long it took to come, and of a chunk, the throughput (it was asked
+            // for: the request's round trip is no exchange of sync bytes to time).
+            if let Some(delay_ms) = crate::swarm::hop::note(state, &peer, blob.sent).await
+                && delay_ms > 0
+                && let transport::BlobBody::Chunk { data, .. } = &blob.body
+            {
+                let rtt =
+                    state
+                        .lock_rtt()
+                        .effective(&peer, Instant::now(), state.timing.rtt_fresh_for());
+                state.throughput.lock_or_recover().message(
+                    &peer,
+                    data.len() as u64,
+                    Duration::from_millis(delay_ms as u64),
+                    Instant::now(),
+                    rtt,
+                );
+            }
             crate::swarm::blob::handle(state, &peer, blob).await;
         }
         transport::ParsedMessage::Status(remote) => {