feat: Use Isolate.register_op in CLI, take 2

cli/ops/dispatch_json.rs

@@ -50,37 +50,40 @@ struct AsyncArgs {
   promise_id: Option<u64>,
 }
 
-pub fn dispatch(
+pub fn json_op(
   d: Dispatcher,
-  state: &ThreadSafeState,
-  control: &[u8],
-  zero_copy: Option<PinnedBuf>,
-) -> CoreOp {
-  let async_args: AsyncArgs = serde_json::from_slice(control).unwrap();
-  let promise_id = async_args.promise_id;
-  let is_sync = promise_id.is_none();
+) -> impl Fn(&ThreadSafeState, &[u8], Option<PinnedBuf>) -> CoreOp {
+  move |state: &ThreadSafeState,
+        control: &[u8],
+        zero_copy: Option<PinnedBuf>| {
+    let async_args: AsyncArgs = serde_json::from_slice(control).unwrap();
+    let promise_id = async_args.promise_id;
+    let is_sync = promise_id.is_none();
 
-  let result = serde_json::from_slice(control)
-    .map_err(ErrBox::from)
-    .and_then(move |args| d(state, args, zero_copy));
-  match result {
-    Ok(JsonOp::Sync(sync_value)) => {
-      assert!(promise_id.is_none());
-      CoreOp::Sync(serialize_result(promise_id, Ok(sync_value)))
-    }
-    Ok(JsonOp::Async(fut)) => {
-      assert!(promise_id.is_some());
-      let fut2 = Box::new(fut.then(move |result| -> Result<Buf, ()> {
-        Ok(serialize_result(promise_id, result))
-      }));
-      CoreOp::Async(fut2)
-    }
-    Err(sync_err) => {
-      let buf = serialize_result(promise_id, Err(sync_err));
-      if is_sync {
-        CoreOp::Sync(buf)
-      } else {
-        CoreOp::Async(Box::new(futures::future::ok(buf)))
+    let result = serde_json::from_slice(control)
+      .map_err(ErrBox::from)
+      .and_then(move |args| d(state, args, zero_copy));
+
+    // Convert to CoreOp
+    match result {
+      Ok(JsonOp::Sync(sync_value)) => {
+        assert!(promise_id.is_none());
+        CoreOp::Sync(serialize_result(promise_id, Ok(sync_value)))
+      }
+      Ok(JsonOp::Async(fut)) => {
+        assert!(promise_id.is_some());
+        let fut2 = Box::new(fut.then(move |result| -> Result<Buf, ()> {
+          Ok(serialize_result(promise_id, result))
+        }));
+        CoreOp::Async(fut2)
+      }
+      Err(sync_err) => {
+        let buf = serialize_result(promise_id, Err(sync_err));
+        if is_sync {
+          CoreOp::Sync(buf)
+        } else {
+          CoreOp::Async(Box::new(futures::future::ok(buf)))
+        }
       }
     }
   }

cli/ops/dispatch_minimal.rs

@@ -72,40 +72,42 @@ fn test_parse_min_record() {
   assert_eq!(parse_min_record(&buf), None);
 }
 
-pub fn dispatch(
+pub fn minimal_op(
   d: Dispatcher,
-  _state: &ThreadSafeState,
-  control: &[u8],
-  zero_copy: Option<PinnedBuf>,
-) -> CoreOp {
-  let mut record = parse_min_record(control).unwrap();
-  let is_sync = record.promise_id == 0;
-  let rid = record.arg;
-  let min_op = d(rid, zero_copy);
+) -> impl Fn(&ThreadSafeState, &[u8], Option<PinnedBuf>) -> CoreOp {
+  move |_state: &ThreadSafeState,
+        control: &[u8],
+        zero_copy: Option<PinnedBuf>| {
+    let mut record = parse_min_record(control).unwrap();
+    let is_sync = record.promise_id == 0;
+    let rid = record.arg;
+    let min_op = d(rid, zero_copy);
 
-  let fut = Box::new(min_op.then(move |result| -> Result<Buf, ()> {
-    match result {
-      Ok(r) => {
-        record.result = r;
+    // Convert to CoreOp
+    let fut = Box::new(min_op.then(move |result| -> Result<Buf, ()> {
+      match result {
+        Ok(r) => {
+          record.result = r;
+        }
+        Err(err) => {
+          // TODO(ry) The dispatch_minimal doesn't properly pipe errors back to
+          // the caller.
+          debug!("swallowed err {}", err);
+          record.result = -1;
+        }
       }
-      Err(err) => {
-        // TODO(ry) The dispatch_minimal doesn't properly pipe errors back to
-        // the caller.
-        debug!("swallowed err {}", err);
-        record.result = -1;
-      }
-    }
-    Ok(record.into())
-  }));
+      Ok(record.into())
+    }));
 
-  if is_sync {
-    // Warning! Possible deadlocks can occur if we try to wait for a future
-    // while in a future. The safe but expensive alternative is to use
-    // tokio_util::block_on.
-    // This block is only exercised for readSync and writeSync, which I think
-    // works since they're simple polling futures.
-    Op::Sync(fut.wait().unwrap())
-  } else {
-    Op::Async(fut)
+    if is_sync {
+      // Warning! Possible deadlocks can occur if we try to wait for a future
+      // while in a future. The safe but expensive alternative is to use
+      // tokio_util::block_on.
+      // This block is only exercised for readSync and writeSync, which I think
+      // works since they're simple polling futures.
+      Op::Sync(fut.wait().unwrap())
+    } else {
+      Op::Async(fut)
+    }
   }
 }