▮ RECOMMENDED USE
Generating idiomatic tokio building blocks (task functions, mock_fetch, error enums) but not a wired-up, compilable module — needs heavy human completion.
▮ REFACTORED PATCH
// 1. Add missing imports at top:
use std::convert::Infallible;
// and add `tracing = "0.1"` to Cargo.toml with `use tracing;` OR replace
// all tracing::info!/warn!/error! with eprintln!/log macros.
// 2. Fix V1 shutdown — JoinHandle is not Clone; move-join instead:
pub async fn shutdown(&self) -> Result<(), ServiceError> {
let _ = self.shutdown_tx.send(());
let mut set = self.inner.write().await;
let entries: Vec<(_, tokio::task::JoinHandle<()>)> =
set.watchers.drain().map(|(k,v)| (k, v.handle)).collect();
drop(set); // release lock before awaiting joins
for (id, handle) in entries {
if handle.await.is_err() { tracing::warn!(watcher_id = id, "watcher panicked"); }
}
Ok(())
}
// 3. Finish V2 add_watcher (the design is sound; just complete it):
pub async fn add_watcher(&self, id: u32) -> Result<(), ServiceError> {
let mut set = self.inner.write().await;
if set.watchers.contains_key(&id) { return Err(ServiceError::WatcherNotFound(id)); }
let health_flag = Arc::new(AtomicBool::new(true));
set.health_flags.insert(id, health_flag.clone());
let tx = self.shared_tx.clone();
let mut shutdown_rx = self.shutdown_tx.subscribe();
let handle = tokio::spawn(async move {
let mut tick = tokio::time::interval(Duration::from_millis(80));
let mut fails = 0u32;
loop {
tokio::select! {
_ = shutdown_rx.recv() => break,
_ = tick.tick() => match mock_fetch(id).await {
Ok(vs) => { fails = 0; for v in vs { let _ = tx.send(WatchedItem{watcher_id:id, value:v}).await; } }
Err(_) => { fails += 1; if fails > 5 { health_flag.store(false, Ordering::Relaxed); break; } }
}
}
}
});
set.watchers.insert(id, WatcherEntryV2 { id, handle });
Ok(())
}
// 4. Add a real main + a backpressure/shutdown test:
#[tokio::main]
async fn main() {
let (mgr, mut out_rx) = WatcherManagerV2::new();
mgr.add_watcher(1).await.unwrap();
tokio::time::sleep(Duration::from_millis(500)).await;
mgr.shutdown().await.unwrap();
if let Some(o) = out_rx.recv().await { println!("items={}.", o.total_items); }
}