1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
use crate::permissions::Permissions;
use deno_core::error::bad_resource_id;
use deno_core::error::AnyError;
use deno_core::futures::future::poll_fn;
use deno_core::serde_json;
use deno_core::serde_json::json;
use deno_core::serde_json::Value;
use deno_core::BufVec;
use deno_core::OpState;
use deno_core::ZeroCopyBuf;
use notify::event::Event as NotifyEvent;
use notify::Error as NotifyError;
use notify::EventKind;
use notify::RecommendedWatcher;
use notify::RecursiveMode;
use notify::Watcher;
use serde::Deserialize;
use serde::Serialize;
use std::cell::RefCell;
use std::convert::From;
use std::path::PathBuf;
use std::rc::Rc;
use tokio::sync::mpsc;
pub fn init(rt: &mut deno_core::JsRuntime) {
super::reg_json_sync(rt, "op_fs_events_open", op_fs_events_open);
super::reg_json_async(rt, "op_fs_events_poll", op_fs_events_poll);
}
struct FsEventsResource {
#[allow(unused)]
watcher: RecommendedWatcher,
receiver: mpsc::Receiver<Result<FsEvent, AnyError>>,
}
/// Represents a file system event.
///
/// We do not use the event directly from the notify crate. We flatten
/// the structure into this simpler structure. We want to only make it more
/// complex as needed.
///
/// Feel free to expand this struct as long as you can add tests to demonstrate
/// the complexity.
#[derive(Serialize, Debug)]
struct FsEvent {
kind: String,
paths: Vec<PathBuf>,
}
impl From<NotifyEvent> for FsEvent {
fn from(e: NotifyEvent) -> Self {
let kind = match e.kind {
EventKind::Any => "any",
EventKind::Access(_) => "access",
EventKind::Create(_) => "create",
EventKind::Modify(_) => "modify",
EventKind::Remove(_) => "remove",
EventKind::Other => todo!(), // What's this for? Leaving it out for now.
}
.to_string();
FsEvent {
kind,
paths: e.paths,
}
}
}
fn op_fs_events_open(
state: &mut OpState,
args: Value,
_zero_copy: &mut [ZeroCopyBuf],
) -> Result<Value, AnyError> {
#[derive(Deserialize)]
struct OpenArgs {
recursive: bool,
paths: Vec<String>,
}
let args: OpenArgs = serde_json::from_value(args)?;
let (sender, receiver) = mpsc::channel::<Result<FsEvent, AnyError>>(16);
let sender = std::sync::Mutex::new(sender);
let mut watcher: RecommendedWatcher =
Watcher::new_immediate(move |res: Result<NotifyEvent, NotifyError>| {
let res2 = res.map(FsEvent::from).map_err(AnyError::from);
let mut sender = sender.lock().unwrap();
// Ignore result, if send failed it means that watcher was already closed,
// but not all messages have been flushed.
let _ = sender.try_send(res2);
})?;
let recursive_mode = if args.recursive {
RecursiveMode::Recursive
} else {
RecursiveMode::NonRecursive
};
for path in &args.paths {
state
.borrow::<Permissions>()
.check_read(&PathBuf::from(path))?;
watcher.watch(path, recursive_mode)?;
}
let resource = FsEventsResource { watcher, receiver };
let rid = state.resource_table.add("fsEvents", Box::new(resource));
Ok(json!(rid))
}
async fn op_fs_events_poll(
state: Rc<RefCell<OpState>>,
args: Value,
_zero_copy: BufVec,
) -> Result<Value, AnyError> {
#[derive(Deserialize)]
struct PollArgs {
rid: u32,
}
let PollArgs { rid } = serde_json::from_value(args)?;
poll_fn(move |cx| {
let mut state = state.borrow_mut();
let watcher = state
.resource_table
.get_mut::<FsEventsResource>(rid)
.ok_or_else(bad_resource_id)?;
watcher
.receiver
.poll_recv(cx)
.map(|maybe_result| match maybe_result {
Some(Ok(value)) => Ok(json!({ "value": value, "done": false })),
Some(Err(err)) => Err(err),
None => Ok(json!({ "done": true })),
})
})
.await
}
|
