clippy

Author: rklaehn

src/api/blobs.rs

@@ -57,7 +57,6 @@ use super::{
 };
 use crate::{
     api::proto::{BatchRequest, ImportByteStreamUpdate},
-    provider::WriterContext,
     store::IROH_BLOCK_SIZE,
     util::temp_tag::TempTag,
     BlobFormat, Hash, HashAndFormat,
@@ -1167,19 +1166,3 @@ pub(crate) trait WriteProgress {
     /// Notify the progress writer that a transfer has started.
     async fn send_transfer_started(&mut self, index: u64, hash: &Hash, size: u64);
 }
-
-impl WriteProgress for WriterContext {
-    async fn notify_payload_write(&mut self, _index: u64, offset: u64, len: usize) {
-        let end_offset = offset + len as u64;
-        self.payload_bytes_written += len as u64;
-        self.tracker.transfer_progress(end_offset).await.ok();
-    }
-
-    fn log_other_write(&mut self, len: usize) {
-        self.other_bytes_written += len as u64;
-    }
-
-    async fn send_transfer_started(&mut self, index: u64, hash: &Hash, size: u64) {
-        self.tracker.transfer_started(index, hash, size).await.ok();
-    }
-}

src/net_protocol.rs

@@ -74,7 +74,7 @@ impl BlobsProtocol {
             inner: Arc::new(BlobsInner {
                 store: store.clone(),
                 endpoint,
-                events: events.unwrap_or(EventSender::NONE),
+                events: events.unwrap_or(EventSender::DEFAULT),
             }),
         }
     }

src/protocol.rs

@@ -392,7 +392,7 @@ pub use range_spec::{ChunkRangesSeq, NonEmptyRequestRangeSpecIter, RangeSpec};
 use snafu::{GenerateImplicitData, Snafu};
 use tokio::io::AsyncReadExt;
 
-use crate::{api::blobs::Bitfield, provider::CountingReader, BlobFormat, Hash, HashAndFormat};
+use crate::{api::blobs::Bitfield, provider::RecvStreamExt, BlobFormat, Hash, HashAndFormat};
 
 /// Maximum message size is limited to 100MiB for now.
 pub const MAX_MESSAGE_SIZE: usize = 1024 * 1024;
@@ -441,9 +441,7 @@ pub enum RequestType {
 }
 
 impl Request {
-    pub async fn read_async(
-        reader: &mut CountingReader<&mut iroh::endpoint::RecvStream>,
-    ) -> io::Result<Self> {
+    pub async fn read_async(reader: &mut iroh::endpoint::RecvStream) -> io::Result<(Self, usize)> {
         let request_type = reader.read_u8().await?;
         let request_type: RequestType = postcard::from_bytes(std::slice::from_ref(&request_type))
             .map_err(|_| {
@@ -453,22 +451,31 @@ impl Request {
             )
         })?;
         Ok(match request_type {
-            RequestType::Get => reader
-                .read_to_end_as::<GetRequest>(MAX_MESSAGE_SIZE)
-                .await?
-                .into(),
-            RequestType::GetMany => reader
-                .read_to_end_as::<GetManyRequest>(MAX_MESSAGE_SIZE)
-                .await?
-                .into(),
-            RequestType::Observe => reader
-                .read_to_end_as::<ObserveRequest>(MAX_MESSAGE_SIZE)
-                .await?
-                .into(),
-            RequestType::Push => reader
-                .read_length_prefixed::<PushRequest>(MAX_MESSAGE_SIZE)
-                .await?
-                .into(),
+            RequestType::Get => {
+                let (r, size) = reader
+                    .read_to_end_as::<GetRequest>(MAX_MESSAGE_SIZE)
+                    .await?;
+                (r.into(), size)
+            }
+            RequestType::GetMany => {
+                let (r, size) = reader
+                    .read_to_end_as::<GetManyRequest>(MAX_MESSAGE_SIZE)
+                    .await?;
+                (r.into(), size)
+            }
+            RequestType::Observe => {
+                let (r, size) = reader
+                    .read_to_end_as::<ObserveRequest>(MAX_MESSAGE_SIZE)
+                    .await?;
+                (r.into(), size)
+            }
+            RequestType::Push => {
+                let r = reader
+                    .read_length_prefixed::<PushRequest>(MAX_MESSAGE_SIZE)
+                    .await?;
+                let size = postcard::experimental::serialized_size(&r).unwrap();
+                (r.into(), size)
+            }
             _ => {
                 return Err(io::Error::new(
                     io::ErrorKind::InvalidData,