Add fn to add data from a stream (#1808)

Fixes #1759, see
#1759 (comment)

## Description

Implements a public API to add from a stream. Also changes add_reader to
use the add_stream internally

## Notes & open questions

Do the same for docs?

## Change checklist

- [ ] Self-review.
- [ ] Documentation updates if relevant.
- [ ] Tests if relevant.

Co-authored-by: Friedel Ziegelmayer <[email protected]>

Author: rklaehn

iroh/src/client.rs

@@ -287,17 +287,24 @@ where
         tag: SetTagOption,
     ) -> anyhow::Result<BlobAddProgress> {
         const CAP: usize = 1024 * 64; // send 64KB per request by default
-        let (mut sink, progress) = self.rpc.bidi(BlobAddStreamRequest { tag }).await?;
-
         let input = ReaderStream::with_capacity(reader, CAP);
+        self.add_stream(input, tag).await
+    }
+
+    /// Write a blob by passing a stream of bytes.
+    pub async fn add_stream(
+        &self,
+        input: impl Stream<Item = io::Result<Bytes>> + Send + Unpin + 'static,
+        tag: SetTagOption,
+    ) -> anyhow::Result<BlobAddProgress> {
+        let (mut sink, progress) = self.rpc.bidi(BlobAddStreamRequest { tag }).await?;
         let mut input = input.map(|chunk| match chunk {
             Ok(chunk) => Ok(BlobAddStreamUpdate::Chunk(chunk)),
             Err(err) => {
                 warn!("Abort send, reason: failed to read from source stream: {err:?}");
                 Ok(BlobAddStreamUpdate::Abort)
             }
         });
-
         tokio::spawn(async move {
             // TODO: Is it important to catch this error? It should also result in an error on the
             // response stream. If we deem it important, we could one-shot send it into the