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))
 }