Get rid of one Gen::new, also clippy and stuff

Author: rklaehn

src/api/blobs.rs

@@ -866,15 +866,15 @@ impl IrpcStreamItem for ListBlobsItem {
     fn into_result_opt(self) -> Option<Result<Hash, super::Error>> {
         match self {
             Self::Item(hash) => Some(Ok(hash)),
-            Self::Error(e) => Some(Err(e.into())),
+            Self::Error(e) => Some(Err(e)),
             Self::Done => None,
         }
     }
 
     fn from_result(item: std::result::Result<Hash, super::Error>) -> Self {
         match item {
             Ok(hash) => Self::Item(hash),
-            Err(e) => Self::Error(e.into()),
+            Err(e) => Self::Error(e),
         }
     }
 

src/api/proto.rs

@@ -29,7 +29,6 @@ use bao_tree::{
     ChunkRanges,
 };
 use bytes::Bytes;
-use genawaiter::sync::Gen;
 use irpc::{
     channel::{mpsc, oneshot},
     rpc_requests,
@@ -398,7 +397,7 @@ impl IrpcStreamItem for ListTagsItem {
     fn from_result(item: std::result::Result<TagInfo, super::Error>) -> Self {
         match item {
             Ok(i) => Self::Item(i),
-            Err(e) => Self::Error(e.into()),
+            Err(e) => Self::Error(e),
         }
     }
 
@@ -431,41 +430,7 @@ impl ListTagsProgress {
     }
 
     pub fn stream(self) -> impl Stream<Item = super::Result<TagInfo>> {
-        Gen::new(|co| async move {
-            let mut rx = match self.inner.await {
-                Ok(rx) => rx,
-                Err(err) => {
-                    co.yield_(Err(super::Error::from(err))).await;
-                    return;
-                }
-            };
-            loop {
-                match rx.recv().await {
-                    Ok(Some(ListTagsItem::Item(item))) => {
-                        co.yield_(Ok(item)).await;
-                    }
-                    Ok(Some(ListTagsItem::Done)) => {
-                        break;
-                    }
-                    Ok(Some(ListTagsItem::Error(err))) => {
-                        co.yield_(Err(err.into())).await;
-                        break;
-                    }
-                    Ok(None) => {
-                        co.yield_(Err(super::Error::Io(io::Error::new(
-                            io::ErrorKind::UnexpectedEof,
-                            "stream ended",
-                        ))))
-                        .await;
-                        break;
-                    }
-                    Err(cause) => {
-                        co.yield_(Err(super::Error::from(cause))).await;
-                        break;
-                    }
-                }
-            }
-        })
+        self.inner.into_stream()
     }
 }
 

src/store/readonly_mem.rs

@@ -42,7 +42,7 @@ use crate::{
         ApiClient, TempTag,
     },
     store::{mem::CompleteStorage, IROH_BLOCK_SIZE},
-    util::ChunkRangesExt,
+    util::{irpc::MpscSenderExt, ChunkRangesExt},
     Hash,
 };
 
@@ -196,7 +196,7 @@ impl Actor {
                 cmd.tx.send(status).await.ok();
             }
             Command::ListTags(cmd) => {
-                cmd.tx.send(proto::ListTagsItem::Done).await.ok();
+                cmd.tx.forward_iter(std::iter::empty()).await.ok();
             }
             Command::SetTag(cmd) => {
                 cmd.tx

src/util/irpc.rs

@@ -29,6 +29,7 @@ pub(crate) trait MpscSenderExt<T: IrpcStreamItem>: Sized {
     ///
     /// This will convert items and errors into the item enum type, and add
     /// a done marker if the stream ends without an error.
+    #[allow(dead_code)]
     async fn forward_stream(
         self,
         stream: impl Stream<Item = std::result::Result<T::Item, T::Error>>,
@@ -122,7 +123,7 @@ where
         while let Some(item) = stream.next().await {
             match item {
                 Ok(i) => items.extend(Some(i)),
-                Err(e) => return Err(E::from(e)),
+                Err(e) => return Err(e),
             }
         }
         Ok(items)