refactor: rewrite ops to use ResourceTable2 (#8512)

This commit migrates all ops to use new resource table
and "AsyncRefCell".

Old implementation of resource table was completely 
removed and all code referencing it was updated to use
new system.

1 files changed, 38 insertions, 20 deletions

diff --git a/runtime/ops/fs_events.rs b/runtime/ops/fs_events.rs
index 4832c915c..38661e1d4 100644
--- a/runtime/ops/fs_events.rs
+++ b/runtime/ops/fs_events.rs
@@ -3,12 +3,16 @@
 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::AsyncRefCell;
 use deno_core::BufVec;
+use deno_core::CancelFuture;
+use deno_core::CancelHandle;
 use deno_core::OpState;
+use deno_core::RcRef;
+use deno_core::Resource;
 use deno_core::ZeroCopyBuf;
 use notify::event::Event as NotifyEvent;
 use notify::Error as NotifyError;
@@ -18,6 +22,7 @@ use notify::RecursiveMode;
 use notify::Watcher;
 use serde::Deserialize;
 use serde::Serialize;
+use std::borrow::Cow;
 use std::cell::RefCell;
 use std::convert::From;
 use std::path::PathBuf;
@@ -32,7 +37,18 @@ pub fn init(rt: &mut deno_core::JsRuntime) {
 struct FsEventsResource {
   #[allow(unused)]
   watcher: RecommendedWatcher,
-  receiver: mpsc::Receiver<Result<FsEvent, AnyError>>,
+  receiver: AsyncRefCell<mpsc::Receiver<Result<FsEvent, AnyError>>>,
+  cancel: CancelHandle,
+}
+
+impl Resource for FsEventsResource {
+  fn name(&self) -> Cow<str> {
+    "fsEvents".into()
+  }
+
+  fn close(self: Rc<Self>) {
+    self.cancel.cancel();
+  }
 }
 
 /// Represents a file system event.
@@ -99,8 +115,12 @@ fn op_fs_events_open(
       .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));
+  let resource = FsEventsResource {
+    watcher,
+    receiver: AsyncRefCell::new(receiver),
+    cancel: Default::default(),
+  };
+  let rid = state.resource_table.add(resource);
   Ok(json!(rid))
 }
 
@@ -114,20 +134,18 @@ async fn op_fs_events_poll(
     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
+
+  let resource = state
+    .borrow()
+    .resource_table
+    .get::<FsEventsResource>(rid)
+    .ok_or_else(bad_resource_id)?;
+  let mut receiver = RcRef::map(&resource, |r| &r.receiver).borrow_mut().await;
+  let cancel = RcRef::map(resource, |r| &r.cancel);
+  let maybe_result = receiver.recv().or_cancel(cancel).await?;
+  match maybe_result {
+    Some(Ok(value)) => Ok(json!({ "value": value, "done": false })),
+    Some(Err(err)) => Err(err),
+    None => Ok(json!({ "done": true })),
+  }
 }