refactor(core): Turn the `wasm_streaming_feed` binding into ops (#11985)

Async WebAssembly compilation was implemented by adding two
bindings: `set_wasm_streaming_callback`, which registered a callback to
be called whenever a streaming wasm compilation was started, and
`wasm_streaming_feed`, which let the JS callback modify the state of the
v8 wasm compiler.

`set_wasm_streaming_callback` cannot currently be implemented as
anything other than a binding, but `wasm_streaming_feed` does not really
need to use anything specific to bindings, and could indeed be
implemented as one or more ops. This PR does that, resulting in a
simplification of the relevant code.

There are three operations on the state of the v8 wasm compiler that
`wasm_streaming_feed` allowed: feeding new bytes into the compiler,
letting it know that there are no more bytes coming from the network,
and aborting the compilation. This PR provides `op_wasm_streaming_feed`
to feed new bytes into the compiler, and `op_wasm_streaming_abort` to
abort the compilation. It doesn't provide an op to let v8 know that the
response is finished, but closing the resource with `Deno.core.close()`
will achieve that.

1 files changed, 56 insertions, 0 deletions

diff --git a/core/ops_builtin.rs b/core/ops_builtin.rs
index 06de59054..e1313fa32 100644
--- a/core/ops_builtin.rs
+++ b/core/ops_builtin.rs
@@ -5,7 +5,11 @@ use crate::op_sync;
 use crate::resources::ResourceId;
 use crate::Extension;
 use crate::OpState;
+use crate::Resource;
+use crate::ZeroCopyBuf;
+use std::cell::RefCell;
 use std::io::{stderr, stdout, Write};
+use std::rc::Rc;
 
 pub(crate) fn init_builtins() -> Extension {
   Extension::builder()
@@ -20,6 +24,8 @@ pub(crate) fn init_builtins() -> Extension {
       ("op_try_close", op_sync(op_try_close)),
       ("op_print", op_sync(op_print)),
       ("op_resources", op_sync(op_resources)),
+      ("op_wasm_streaming_feed", op_sync(op_wasm_streaming_feed)),
+      ("op_wasm_streaming_abort", op_sync(op_wasm_streaming_abort)),
     ])
     .build()
 }
@@ -81,3 +87,53 @@ pub fn op_print(
   }
   Ok(())
 }
+
+pub struct WasmStreamingResource(pub(crate) RefCell<rusty_v8::WasmStreaming>);
+
+impl Resource for WasmStreamingResource {
+  fn close(self: Rc<Self>) {
+    // At this point there are no clones of Rc<WasmStreamingResource> on the
+    // resource table, and no one should own a reference outside of the stack.
+    // Therefore, we can be sure `self` is the only reference.
+    if let Ok(wsr) = Rc::try_unwrap(self) {
+      wsr.0.into_inner().finish();
+    } else {
+      panic!("Couldn't consume WasmStreamingResource.");
+    }
+  }
+}
+
+/// Feed bytes to WasmStreamingResource.
+pub fn op_wasm_streaming_feed(
+  state: &mut OpState,
+  rid: ResourceId,
+  bytes: ZeroCopyBuf,
+) -> Result<(), AnyError> {
+  let wasm_streaming =
+    state.resource_table.get::<WasmStreamingResource>(rid)?;
+
+  wasm_streaming.0.borrow_mut().on_bytes_received(&bytes);
+
+  Ok(())
+}
+
+/// Abort a WasmStreamingResource.
+pub fn op_wasm_streaming_abort(
+  state: &mut OpState,
+  rid: ResourceId,
+  exception: serde_v8::Value,
+) -> Result<(), AnyError> {
+  let wasm_streaming =
+    state.resource_table.take::<WasmStreamingResource>(rid)?;
+
+  // At this point there are no clones of Rc<WasmStreamingResource> on the
+  // resource table, and no one should own a reference because we're never
+  // cloning them. So we can be sure `wasm_streaming` is the only reference.
+  if let Ok(wsr) = Rc::try_unwrap(wasm_streaming) {
+    wsr.0.into_inner().abort(Some(exception.v8_value));
+  } else {
+    panic!("Couldn't consume WasmStreamingResource.");
+  }
+
+  Ok(())
+}