feat: Read and seek (#119)

## Description

Add an api for reading and seeking a blob using tokio::io::AsyncRead and
tokio::io::AsyncSeek

## Breaking Changes

Changes api::blobs::ExportRangesProgress::stream to be non-async - it
does not have to be!

## Notes & open questions

Note: the entire machinery of seek and read is horribly undocumented
regarding the non happy case. So I went with the following approach: if
you ever drop a future of either seek or read before completing, the
reader becomes poisioned and will never work again, you have to grab a
new one, which is cheap.

This bypasses the whole question about what to do if somebody calls x
without having completed y, and in normal usage you should never notice
this.

## Change checklist

- [ ] Self-review.
- [ ] Documentation updates following the [style
guide](https://rust-lang.github.io/rfcs/1574-more-api-documentation-conventions.html#appendix-a-full-conventions-text),
if relevant.
- [ ] Tests if relevant.
- [ ] All breaking changes documented.

Author: rklaehn

.github/workflows/ci.yaml

@@ -143,7 +143,7 @@ jobs:
     - uses: taiki-e/install-action@cross
 
     - name: test
-      run: cross test --all --target ${{ matrix.target }} -- --test-threads=12
+      run: cross test --all --target ${{ matrix.target }} -- --test-threads=4
       env:
         RUST_LOG: ${{ runner.debug && 'TRACE' || 'DEBUG' }}
 

src/api.rs

@@ -4,7 +4,7 @@
 //! with a remote store via rpc calls.
 //!
 //! The entry point for the api is the [`Store`] struct. There are several ways
-//! to obtain a `Store` instance: it is available via [`Deref`](std::ops::Deref)
+//! to obtain a `Store` instance: it is available via [`Deref`]
 //! from the different store implementations
 //! (e.g. [`MemStore`](crate::store::mem::MemStore)
 //! and [`FsStore`](crate::store::fs::FsStore)) as well as on the

src/api/blobs.rs

@@ -29,8 +29,11 @@ use n0_future::{future, stream, Stream, StreamExt};
 use quinn::SendStream;
 use range_collections::{range_set::RangeSetRange, RangeSet2};
 use ref_cast::RefCast;
+use serde::{Deserialize, Serialize};
 use tokio::io::AsyncWriteExt;
 use tracing::trace;
+mod reader;
+pub use reader::BlobReader;
 
 // Public reexports from the proto module.
 //
@@ -102,6 +105,38 @@ impl Blobs {
         })
     }
 
+    /// Create a reader for the given hash. The reader implements [`tokio::io::AsyncRead`] and [`tokio::io::AsyncSeek`]
+    /// and therefore can be used to read the blob's content.
+    ///
+    /// Any access to parts of the blob that are not present will result in an error.
+    ///
+    /// Example:
+    /// ```rust
+    /// use iroh_blobs::{store::mem::MemStore, api::blobs::Blobs};
+    /// use tokio::io::AsyncReadExt;
+    ///
+    /// # async fn example() -> anyhow::Result<()> {
+    /// let store = MemStore::new();
+    /// let tag = store.add_slice(b"Hello, world!").await?;
+    /// let mut reader = store.reader(tag.hash);
+    /// let mut buf = String::new();
+    /// reader.read_to_string(&mut buf).await?;
+    /// assert_eq!(buf, "Hello, world!");
+    /// # Ok(())
+    /// }
+    /// ```
+    pub fn reader(&self, hash: impl Into<Hash>) -> BlobReader {
+        self.reader_with_opts(ReaderOptions { hash: hash.into() })
+    }
+
+    /// Create a reader for the given options. The reader implements [`tokio::io::AsyncRead`] and [`tokio::io::AsyncSeek`]
+    /// and therefore can be used to read the blob's content.
+    ///
+    /// Any access to parts of the blob that are not present will result in an error.
+    pub fn reader_with_opts(&self, options: ReaderOptions) -> BlobReader {
+        BlobReader::new(self.clone(), options)
+    }
+
     /// Delete a blob.
     ///
     /// This function is not public, because it does not work as expected when called manually,
@@ -647,6 +682,12 @@ impl<'a> AddProgress<'a> {
     }
 }
 
+/// Options for an async reader for blobs that supports AsyncRead and AsyncSeek.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ReaderOptions {
+    pub hash: Hash,
+}
+
 /// An observe result. Awaiting this will return the current state.
 ///
 /// Calling [`ObserveProgress::stream`] will return a stream of updates, where
@@ -856,7 +897,7 @@ impl ExportRangesProgress {
     /// range of 0..100, you will get the entire first chunk, 0..1024.
     ///
     /// It is up to the caller to clip the ranges to the requested ranges.
-    pub async fn stream(self) -> impl Stream<Item = ExportRangesItem> {
+    pub fn stream(self) -> impl Stream<Item = ExportRangesItem> {
         Gen::new(|co| async move {
             let mut rx = match self.inner.await {
                 Ok(rx) => rx,