fixup

Author: Frando

examples/limit.rs

@@ -271,7 +271,7 @@ async fn main() -> Result<()> {
             let store = MemStore::new();
             let hashes = add_paths(&store, paths).await?;
             let events = limit_by_node_id(allowed_nodes.clone());
-            let (router, addr) = setup(MemStore::new(), events).await?;
+            let (router, addr) = setup(store, events).await?;
 
             for (path, hash) in hashes {
                 let ticket = BlobTicket::new(addr.clone(), hash, BlobFormat::Raw);
@@ -299,12 +299,16 @@ async fn main() -> Result<()> {
                 }
             }
 
-            let events = limit_by_hash(allowed_hashes);
-            let (router, addr) = setup(MemStore::new(), events).await?;
+            let events = limit_by_hash(allowed_hashes.clone());
+            let (router, addr) = setup(store, events).await?;
 
-            for (i, (path, hash)) in hashes.iter().enumerate() {
+            for (path, hash) in hashes.iter() {
                 let ticket = BlobTicket::new(addr.clone(), *hash, BlobFormat::Raw);
-                let permitted = if i == 0 { "" } else { "limited" };
+                let permitted = if allowed_hashes.contains(hash) {
+                    "allowed"
+                } else {
+                    "forbidden"
+                };
                 println!("{}: {ticket} ({permitted})", path.display());
             }
             tokio::signal::ctrl_c().await?;
@@ -314,7 +318,7 @@ async fn main() -> Result<()> {
             let store = MemStore::new();
             let hashes = add_paths(&store, paths).await?;
             let events = throttle(delay_ms);
-            let (router, addr) = setup(MemStore::new(), events).await?;
+            let (router, addr) = setup(store, events).await?;
             for (path, hash) in hashes {
                 let ticket = BlobTicket::new(addr.clone(), hash, BlobFormat::Raw);
                 println!("{}: {ticket}", path.display());
@@ -329,7 +333,7 @@ async fn main() -> Result<()> {
             let store = MemStore::new();
             let hashes = add_paths(&store, paths).await?;
             let events = limit_max_connections(max_connections);
-            let (router, addr) = setup(MemStore::new(), events).await?;
+            let (router, addr) = setup(store, events).await?;
             for (path, hash) in hashes {
                 let ticket = BlobTicket::new(addr.clone(), hash, BlobFormat::Raw);
                 println!("{}: {ticket}", path.display());

examples/random_store.rs

@@ -14,7 +14,7 @@ use iroh_blobs::{
 use irpc::RpcMessage;
 use n0_future::StreamExt;
 use rand::{rngs::StdRng, Rng, SeedableRng};
-use tokio::{signal::ctrl_c, sync::mpsc};
+use tokio::signal::ctrl_c;
 use tracing::info;
 
 #[derive(Parser, Debug)]
@@ -102,7 +102,7 @@ pub fn get_or_generate_secret_key() -> Result<SecretKey> {
 }
 
 pub fn dump_provider_events(allow_push: bool) -> (tokio::task::JoinHandle<()>, EventSender) {
-    let (tx, mut rx) = mpsc::channel(100);
+    let (tx, mut rx) = EventSender::channel(100, EventMask::ALL_READONLY);
     fn dump_updates<T: RpcMessage>(mut rx: irpc::channel::mpsc::Receiver<T>) {
         tokio::spawn(async move {
             while let Ok(Some(update)) = rx.recv().await {
@@ -176,7 +176,7 @@ pub fn dump_provider_events(allow_push: bool) -> (tokio::task::JoinHandle<()>, E
             }
         }
     });
-    (dump_task, EventSender::new(tx, EventMask::ALL_READONLY))
+    (dump_task, tx)
 }
 
 #[tokio::main]

src/tests.rs

@@ -342,7 +342,7 @@ fn event_handler(
     allowed_nodes: impl IntoIterator<Item = NodeId>,
 ) -> (EventSender, watch::Receiver<usize>, AbortOnDropHandle<()>) {
     let (count_tx, count_rx) = tokio::sync::watch::channel(0usize);
-    let (events_tx, mut events_rx) = mpsc::channel::<ProviderMessage>(16);
+    let (events_tx, mut events_rx) = EventSender::channel(16, EventMask::ALL_READONLY);
     let allowed_nodes = allowed_nodes.into_iter().collect::<HashSet<_>>();
     let task = AbortOnDropHandle::new(tokio::task::spawn(async move {
         while let Some(event) = events_rx.recv().await {
@@ -370,16 +370,7 @@ fn event_handler(
             }
         }
     }));
-    (
-        EventSender::new(
-            events_tx,
-            EventMask {
-                ..EventMask::ALL_READONLY
-            },
-        ),
-        count_rx,
-        task,
-    )
+    (events_tx, count_rx, task)
 }
 
 async fn two_nodes_push_blobs(