Merge branch 'entity-manager' into lightweight-watcher

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' }}
 

Cargo.lock

@@ -1,11 +1,11 @@
 [package]
 name = "iroh-blobs"
-version = "0.90.0"
+version = "0.91.0"
 edition = "2021"
 description = "content-addressed blobs for iroh"
 license = "MIT OR Apache-2.0"
 authors = ["dignifiedquire <[email protected]>", "n0 team"]
-repository = "https://github.com/n0-computer/blobs2"
+repository = "https://github.com/n0-computer/iroh-blobs"
 keywords = ["hashing", "quic", "blake3", "streaming"]
 
 # Sadly this also needs to be updated in .github/workflows/ci.yml

Cargo.toml

@@ -22,6 +22,7 @@ This crate is used together with [iroh](https://crates.io/crates/iroh). Connecti
 
 - **Requester:** The side that asks for data. It is initiating requests to one or many providers.
 
+A node can be a provider and a requester at the same time.
 
 ## Getting started
 
@@ -31,33 +32,33 @@ Iroh provides a [`Router`](https://docs.rs/iroh/latest/iroh/protocol/struct.Rout
 
 Here is a basic example of how to set up `iroh-blobs` with `iroh`:
 
-```rust
+```rust,no_run
 use iroh::{protocol::Router, Endpoint};
-use iroh_blobs::{store::Store, net_protocol::Blobs};
+use iroh_blobs::{store::mem::MemStore, BlobsProtocol};
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     // create an iroh endpoint that includes the standard discovery mechanisms
     // we've built at number0
     let endpoint = Endpoint::builder().discovery_n0().bind().await?;
 
-    // create an in-memory blob store
-    // use `iroh_blobs::net_protocol::Blobs::persistent` to load or create a
-    // persistent blob store from a path
-    let blobs = Blobs::memory().build(&endpoint);
-
-    // turn on the "rpc" feature if you need to create blobs and tags clients
-    let blobs_client = blobs.client();
-    let tags_client = blobs_client.tags();
+    // create a protocol handler using an in-memory blob store.
+    let store = MemStore::new();
+    let blobs = BlobsProtocol::new(&store, endpoint.clone(), None);
 
     // build the router
     let router = Router::builder(endpoint)
         .accept(iroh_blobs::ALPN, blobs.clone())
         .spawn();
 
-    // do fun stuff with the blobs protocol!
+    let tag = blobs.add_slice(b"Hello world").await?;
+    println!("We are now serving {}", blobs.ticket(tag).await?);
+
+    // wait for control-c
+    tokio::signal::ctrl_c().await;
+
+    // clean shutdown of router and store
     router.shutdown().await?;
-    drop(tags_client);
     Ok(())
 }
 ```
@@ -81,4 +82,4 @@ at your option.
 
 Unless you explicitly state otherwise, any contribution intentionally submitted
 for inclusion in this project by you, as defined in the Apache-2.0 license,
-shall be dual licensed as above, without any additional terms or conditions.
+shall be dual licensed as above, without any additional terms or conditions.

README.md

@@ -237,7 +237,7 @@ async fn provide(args: ProvideArgs) -> anyhow::Result<()> {
         .bind()
         .await?;
     let (dump_task, events_tx) = dump_provider_events(args.allow_push);
-    let blobs = iroh_blobs::net_protocol::Blobs::new(&store, endpoint.clone(), Some(events_tx));
+    let blobs = iroh_blobs::BlobsProtocol::new(&store, endpoint.clone(), Some(events_tx));
     let router = iroh::protocol::Router::builder(endpoint.clone())
         .accept(iroh_blobs::ALPN, blobs)
         .spawn();

examples/random_store.rs

@@ -0,0 +1,93 @@
+use std::path::PathBuf;
+
+use iroh::{protocol::Router, Endpoint};
+use iroh_blobs::{store::mem::MemStore, ticket::BlobTicket, BlobsProtocol};
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    // Create an endpoint, it allows creating and accepting
+    // connections in the iroh p2p world
+    let endpoint = Endpoint::builder().discovery_n0().bind().await?;
+
+    // We initialize an in-memory backing store for iroh-blobs
+    let store = MemStore::new();
+    // Then we initialize a struct that can accept blobs requests over iroh connections
+    let blobs = BlobsProtocol::new(&store, endpoint.clone(), None);
+
+    // Grab all passed in arguments, the first one is the binary itself, so we skip it.
+    let args: Vec<String> = std::env::args().skip(1).collect();
+    // Convert to &str, so we can pattern-match easily:
+    let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect();
+
+    match arg_refs.as_slice() {
+        ["send", filename] => {
+            let filename: PathBuf = filename.parse()?;
+            let abs_path = std::path::absolute(&filename)?;
+
+            println!("Hashing file.");
+
+            // When we import a blob, we get back a "tag" that refers to said blob in the store
+            // and allows us to control when/if it gets garbage-collected
+            let tag = store.blobs().add_path(abs_path).await?;
+
+            let node_id = endpoint.node_id();
+            let ticket = BlobTicket::new(node_id.into(), tag.hash, tag.format);
+
+            println!("File hashed. Fetch this file by running:");
+            println!(
+                "cargo run --example transfer -- receive {ticket} {}",
+                filename.display()
+            );
+
+            // For sending files we build a router that accepts blobs connections & routes them
+            // to the blobs protocol.
+            let router = Router::builder(endpoint)
+                .accept(iroh_blobs::ALPN, blobs)
+                .spawn();
+
+            tokio::signal::ctrl_c().await?;
+
+            // Gracefully shut down the node
+            println!("Shutting down.");
+            router.shutdown().await?;
+        }
+        ["receive", ticket, filename] => {
+            let filename: PathBuf = filename.parse()?;
+            let abs_path = std::path::absolute(filename)?;
+            let ticket: BlobTicket = ticket.parse()?;
+
+            // For receiving files, we create a "downloader" that allows us to fetch files
+            // from other nodes via iroh connections
+            let downloader = store.downloader(&endpoint);
+
+            println!("Starting download.");
+
+            downloader
+                .download(ticket.hash(), Some(ticket.node_addr().node_id))
+                .await?;
+
+            println!("Finished download.");
+            println!("Copying to destination.");
+
+            store.blobs().export(ticket.hash(), abs_path).await?;
+
+            println!("Finished copying.");
+
+            // Gracefully shut down the node
+            println!("Shutting down.");
+            endpoint.close().await;
+        }
+        _ => {
+            println!("Couldn't parse command line arguments: {args:?}");
+            println!("Usage:");
+            println!("    # to send:");
+            println!("    cargo run --example transfer -- send [FILE]");
+            println!("    # this will print a ticket.");
+            println!();
+            println!("    # to receive:");
+            println!("    cargo run --example transfer -- receive [TICKET] [FILE]");
+        }
+    }
+
+    Ok(())
+}

examples/transfer.rs

@@ -2,6 +2,16 @@
 //!
 //! This API is both for interacting with an in-process store and for interacting
 //! 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`]
+//! from the different store implementations
+//! (e.g. [`MemStore`](crate::store::mem::MemStore)
+//! and [`FsStore`](crate::store::fs::FsStore)) as well as on the
+//! [`BlobsProtocol`](crate::BlobsProtocol) iroh protocol handler.
+//!
+//! You can also [`connect`](Store::connect) to a remote store that is listening
+//! to rpc requests.
 use std::{io, net::SocketAddr, ops::Deref, sync::Arc};
 
 use bao_tree::io::EncodeError;

src/api.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,