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, 41 insertions, 27 deletions

diff --git a/op_crates/fetch/lib.rs b/op_crates/fetch/lib.rs
index 8a4c1ee16..c2c08d2cf 100644
--- a/op_crates/fetch/lib.rs
+++ b/op_crates/fetch/lib.rs
@@ -5,15 +5,19 @@
 use deno_core::error::bad_resource_id;
 use deno_core::error::type_error;
 use deno_core::error::AnyError;
-use deno_core::futures;
 use deno_core::serde_json;
 use deno_core::serde_json::json;
 use deno_core::serde_json::Value;
 use deno_core::url;
 use deno_core::url::Url;
+use deno_core::AsyncRefCell;
 use deno_core::BufVec;
+use deno_core::CancelFuture;
+use deno_core::CancelHandle;
 use deno_core::JsRuntime;
 use deno_core::OpState;
+use deno_core::RcRef;
+use deno_core::Resource;
 use deno_core::ZeroCopyBuf;
 
 use reqwest::header::HeaderName;
@@ -23,6 +27,7 @@ use reqwest::Client;
 use reqwest::Method;
 use reqwest::Response;
 use serde::Deserialize;
+use std::borrow::Cow;
 use std::cell::RefCell;
 use std::convert::From;
 use std::fs::File;
@@ -172,10 +177,10 @@ where
     }
   }
 
-  let rid = state
-    .borrow_mut()
-    .resource_table
-    .add("httpBody", Box::new(res));
+  let rid = state.borrow_mut().resource_table.add(HttpBodyResource {
+    response: AsyncRefCell::new(res),
+    cancel: Default::default(),
+  });
 
   Ok(json!({
     "bodyRid": rid,
@@ -199,32 +204,43 @@ pub async fn op_fetch_read(
   let args: Args = serde_json::from_value(args)?;
   let rid = args.rid;
 
-  use futures::future::poll_fn;
-  use futures::ready;
-  use futures::FutureExt;
-  let f = poll_fn(move |cx| {
-    let mut state = state.borrow_mut();
-    let response = state
-      .resource_table
-      .get_mut::<Response>(rid as u32)
-      .ok_or_else(bad_resource_id)?;
+  let resource = state
+    .borrow()
+    .resource_table
+    .get::<HttpBodyResource>(rid as u32)
+    .ok_or_else(bad_resource_id)?;
+  let mut response = RcRef::map(&resource, |r| &r.response).borrow_mut().await;
+  let cancel = RcRef::map(resource, |r| &r.cancel);
+  let maybe_chunk = response.chunk().or_cancel(cancel).await??;
+  if let Some(chunk) = maybe_chunk {
+    // TODO(ry) This is terribly inefficient. Make this zero-copy.
+    Ok(json!({ "chunk": &*chunk }))
+  } else {
+    Ok(json!({ "chunk": null }))
+  }
+}
 
-    let mut chunk_fut = response.chunk().boxed_local();
-    let r = ready!(chunk_fut.poll_unpin(cx))?;
-    if let Some(chunk) = r {
-      // TODO(ry) This is terribly inefficient. Make this zero-copy.
-      Ok(json!({ "chunk": &*chunk })).into()
-    } else {
-      Ok(json!({ "chunk": null })).into()
-    }
-  });
-  f.await
+struct HttpBodyResource {
+  response: AsyncRefCell<Response>,
+  cancel: CancelHandle,
+}
+
+impl Resource for HttpBodyResource {
+  fn name(&self) -> Cow<str> {
+    "httpBody".into()
+  }
 }
 
 struct HttpClientResource {
   client: Client,
 }
 
+impl Resource for HttpClientResource {
+  fn name(&self) -> Cow<str> {
+    "httpClient".into()
+  }
+}
+
 impl HttpClientResource {
   fn new(client: Client) -> Self {
     Self { client }
@@ -255,9 +271,7 @@ where
 
   let client = create_http_client(args.ca_file.as_deref()).unwrap();
 
-  let rid = state
-    .resource_table
-    .add("httpClient", Box::new(HttpClientResource::new(client)));
+  let rid = state.resource_table.add(HttpClientResource::new(client));
   Ok(json!(rid))
 }