implement Worker.terminate() and self.close() (#4684)

1 files changed, 39 insertions, 5 deletions

diff --git a/cli/ops/web_worker.rs b/cli/ops/web_worker.rs
index dd7b6e34a..8cade7d40 100644
--- a/cli/ops/web_worker.rs
+++ b/cli/ops/web_worker.rs
@@ -3,10 +3,10 @@ use super::dispatch_json::{JsonOp, Value};
 use crate::op_error::OpError;
 use crate::ops::json_op;
 use crate::state::State;
+use crate::web_worker::WebWorkerHandle;
 use crate::worker::WorkerEvent;
 use deno_core::*;
 use futures::channel::mpsc;
-use futures::sink::SinkExt;
 use std::convert::From;
 
 pub fn web_worker_op<D>(
@@ -25,7 +25,32 @@ where
         -> Result<JsonOp, OpError> { dispatcher(&sender, args, zero_copy) }
 }
 
-pub fn init(i: &mut Isolate, s: &State, sender: &mpsc::Sender<WorkerEvent>) {
+pub fn web_worker_op2<D>(
+  handle: WebWorkerHandle,
+  sender: mpsc::Sender<WorkerEvent>,
+  dispatcher: D,
+) -> impl Fn(Value, Option<ZeroCopyBuf>) -> Result<JsonOp, OpError>
+where
+  D: Fn(
+    WebWorkerHandle,
+    &mpsc::Sender<WorkerEvent>,
+    Value,
+    Option<ZeroCopyBuf>,
+  ) -> Result<JsonOp, OpError>,
+{
+  move |args: Value,
+        zero_copy: Option<ZeroCopyBuf>|
+        -> Result<JsonOp, OpError> {
+    dispatcher(handle.clone(), &sender, args, zero_copy)
+  }
+}
+
+pub fn init(
+  i: &mut Isolate,
+  s: &State,
+  sender: &mpsc::Sender<WorkerEvent>,
+  handle: WebWorkerHandle,
+) {
   i.register_op(
     "op_worker_post_message",
     s.core_op(json_op(web_worker_op(
@@ -35,7 +60,11 @@ pub fn init(i: &mut Isolate, s: &State, sender: &mpsc::Sender<WorkerEvent>) {
   );
   i.register_op(
     "op_worker_close",
-    s.core_op(json_op(web_worker_op(sender.clone(), op_worker_close))),
+    s.core_op(json_op(web_worker_op2(
+      handle,
+      sender.clone(),
+      op_worker_close,
+    ))),
   );
 }
 
@@ -47,18 +76,23 @@ fn op_worker_post_message(
 ) -> Result<JsonOp, OpError> {
   let d = Vec::from(data.unwrap().as_ref()).into_boxed_slice();
   let mut sender = sender.clone();
-  let fut = sender.send(WorkerEvent::Message(d));
-  futures::executor::block_on(fut).expect("Failed to post message to host");
+  sender
+    .try_send(WorkerEvent::Message(d))
+    .expect("Failed to post message to host");
   Ok(JsonOp::Sync(json!({})))
 }
 
 /// Notify host that guest worker closes
 fn op_worker_close(
+  handle: WebWorkerHandle,
   sender: &mpsc::Sender<WorkerEvent>,
   _args: Value,
   _data: Option<ZeroCopyBuf>,
 ) -> Result<JsonOp, OpError> {
   let mut sender = sender.clone();
+  // Notify parent that we're finished
   sender.close_channel();
+  // Terminate execution of current worker
+  handle.terminate();
   Ok(JsonOp::Sync(json!({})))
 }