✨ True Async notifykit API #46
Conversation
There was a problem hiding this comment.
Pull Request Overview
This PR introduces true asynchronous support to the notifykit API by replacing the existing blocking/polling mechanism with Tokio-based async operations. The changes enable proper async file watching without blocking threads.
- Replaced synchronous
get()polling method with async event streaming using Tokio broadcast channels - Converted
watch()andunwatch()methods to async operations usingpyo3_asyncio - Introduced background task-based event processing with proper async iteration support
Reviewed Changes
Copilot reviewed 10 out of 12 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| src/watcher.rs | Core async refactor: added Tokio broadcast channels, background event processing task, and proper async event handling |
| src/lib.rs | Updated Python bindings to use async methods and added EventBatchIter for proper async iteration |
| src/events/mod.rs | Added Clone trait to EventType enum for broadcast channel compatibility |
| notifykit/_notifier.py | Simplified Python interface to use native async patterns instead of polling |
| examples/awatch_dir_recursively.py | Updated example to use new async watch() method |
| Cargo.toml | Added required async dependencies (pyo3-asyncio, tokio) |
Comments suppressed due to low confidence (1)
src/lib.rs:1
- Silently dropping events on channel overflow could lead to missed file system events. Consider logging when events are dropped or implementing backpressure handling.
mod events;
Tip: Customize your code reviews with copilot-instructions.md. Create the file or learn how to get started.
| thread::spawn(move || { | ||
| let rt = tokio::runtime::Runtime::new().expect("tokio rt"); | ||
| rt.block_on(async move { | ||
| let mut ticker = time::interval(debounce_delay); | ||
|
|
||
| loop { | ||
| tokio::select! { | ||
| _ = &mut stop_rx => break, | ||
| _ = ticker.tick() => { | ||
| let (raw, errs) = { | ||
| let mut p = proc.lock().unwrap(); | ||
| (p.get_events(), p.get_errors()) | ||
| }; | ||
| if debug && !raw.is_empty() { println!("processed: {:?}", raw); } | ||
| if !errs.is_empty() { eprintln!("errors: {:?}", errs); } | ||
| if raw.is_empty() { continue; } | ||
|
|
||
| let mut batch = Vec::with_capacity(raw.len()); | ||
| for r in raw { | ||
| if let Some(ev) = create_event(&r) { batch.push(ev); } | ||
| } | ||
|
|
||
| if !batch.is_empty() { let _ = tx.send(batch); } // drop on overflow | ||
| } | ||
| } | ||
| } | ||
| }); |
There was a problem hiding this comment.
Creating a new Tokio runtime inside a spawned thread is inefficient and can cause issues. Consider using Handle::current() to use the existing runtime or restructure to avoid nested runtimes.
| .await | ||
| .ok(); |
There was a problem hiding this comment.
Silently ignoring the Result with .ok() hides potential errors from spawn_blocking. Consider proper error handling or at least logging the error.
| .await | ||
| .ok(); |
There was a problem hiding this comment.
Silently ignoring the Result with .ok() hides potential errors from spawn_blocking. Consider proper error handling or at least logging the error.
1 participant
Changes