6 files changed, 211 insertions, 216 deletions

diff --git a/cli/ops/compiler.rs b/cli/ops/compiler.rs
index efdcd2c9b..39d3a6d7f 100644
--- a/cli/ops/compiler.rs
+++ b/cli/ops/compiler.rs
@@ -1,86 +1,68 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
-use super::dispatch_flatbuffers::serialize_response;
-use super::utils::*;
-use crate::deno_error;
-use crate::msg;
+use super::dispatch_json::{Deserialize, JsonOp, Value};
 use crate::state::ThreadSafeState;
 use crate::tokio_util;
 use deno::*;
-use flatbuffers::FlatBufferBuilder;
-use futures::Future;
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct CacheArgs {
+  module_id: String,
+  contents: String,
+  extension: String,
+}
 
 pub fn op_cache(
   state: &ThreadSafeState,
-  base: &msg::Base<'_>,
-  data: Option<PinnedBuf>,
-) -> CliOpResult {
-  assert!(data.is_none());
-  let inner = base.inner_as_cache().unwrap();
-  let extension = inner.extension().unwrap();
-  // TODO: rename to something with 'url'
-  let module_id = inner.module_id().unwrap();
-  let contents = inner.contents().unwrap();
+  args: Value,
+  _zero_copy: Option<PinnedBuf>,
+) -> Result<JsonOp, ErrBox> {
+  let args: CacheArgs = serde_json::from_value(args)?;
 
-  let module_specifier = ModuleSpecifier::resolve_url(module_id)
+  let module_specifier = ModuleSpecifier::resolve_url(&args.module_id)
     .expect("Should be valid module specifier");
 
   state.ts_compiler.cache_compiler_output(
     &module_specifier,
-    extension,
-    contents,
+    &args.extension,
+    &args.contents,
   )?;
 
-  ok_buf(empty_buf())
+  Ok(JsonOp::Sync(json!({})))
+}
+
+#[derive(Deserialize)]
+struct FetchSourceFileArgs {
+  specifier: String,
+  referrer: String,
 }
 
 pub fn op_fetch_source_file(
   state: &ThreadSafeState,
-  base: &msg::Base<'_>,
-  data: Option<PinnedBuf>,
-) -> CliOpResult {
-  if !base.sync() {
-    return Err(deno_error::no_async_support());
-  }
-  assert!(data.is_none());
-  let inner = base.inner_as_fetch_source_file().unwrap();
-  let cmd_id = base.cmd_id();
-  let specifier = inner.specifier().unwrap();
-  let referrer = inner.referrer().unwrap();
+  args: Value,
+  _zero_copy: Option<PinnedBuf>,
+) -> Result<JsonOp, ErrBox> {
+  let args: FetchSourceFileArgs = serde_json::from_value(args)?;
 
   // TODO(ry) Maybe a security hole. Only the compiler worker should have access
   // to this. Need a test to demonstrate the hole.
   let is_dyn_import = false;
 
   let resolved_specifier =
-    state.resolve(specifier, referrer, false, is_dyn_import)?;
+    state.resolve(&args.specifier, &args.referrer, false, is_dyn_import)?;
 
   let fut = state
     .file_fetcher
-    .fetch_source_file_async(&resolved_specifier)
-    .and_then(move |out| {
-      let builder = &mut FlatBufferBuilder::new();
-      let data_off = builder.create_vector(out.source_code.as_slice());
-      let msg_args = msg::FetchSourceFileResArgs {
-        module_name: Some(builder.create_string(&out.url.to_string())),
-        filename: Some(builder.create_string(&out.filename.to_str().unwrap())),
-        media_type: out.media_type,
-        data: Some(data_off),
-      };
-      let inner = msg::FetchSourceFileRes::create(builder, &msg_args);
-      Ok(serialize_response(
-        cmd_id,
-        builder,
-        msg::BaseArgs {
-          inner: Some(inner.as_union_value()),
-          inner_type: msg::Any::FetchSourceFileRes,
-          ..Default::default()
-        },
-      ))
-    });
+    .fetch_source_file_async(&resolved_specifier);
 
   // WARNING: Here we use tokio_util::block_on() which starts a new Tokio
   // runtime for executing the future. This is so we don't inadvernently run
   // out of threads in the main runtime.
-  let result_buf = tokio_util::block_on(fut)?;
-  Ok(Op::Sync(result_buf))
+  let out = tokio_util::block_on(fut)?;
+  Ok(JsonOp::Sync(json!({
+    "moduleName": out.url.to_string(),
+    "filename": out.filename.to_str().unwrap(),
+    "mediaType": out.media_type as i32,
+    "sourceCode": String::from_utf8(out.source_code).unwrap(),
+  })))
 }
diff --git a/cli/ops/dispatch_flatbuffers.rs b/cli/ops/dispatch_flatbuffers.rs
index d9aa2fb83..c785d6c06 100644
--- a/cli/ops/dispatch_flatbuffers.rs
+++ b/cli/ops/dispatch_flatbuffers.rs
@@ -6,10 +6,7 @@ use deno::*;
 use flatbuffers::FlatBufferBuilder;
 use hyper::rt::Future;
 
-use super::compiler::{op_cache, op_fetch_source_file};
-use super::errors::{op_apply_source_map, op_format_error};
-use super::fetch::op_fetch;
-use super::files::{op_close, op_open, op_read, op_seek, op_write};
+use super::files::{op_read, op_write};
 use super::fs::{
   op_chdir, op_chmod, op_chown, op_copy_file, op_cwd, op_link,
   op_make_temp_dir, op_mkdir, op_read_dir, op_read_link, op_remove, op_rename,
@@ -142,19 +139,13 @@ pub fn serialize_response(
 pub fn op_selector_std(inner_type: msg::Any) -> Option<CliDispatchFn> {
   match inner_type {
     msg::Any::Accept => Some(op_accept),
-    msg::Any::ApplySourceMap => Some(op_apply_source_map),
-    msg::Any::Cache => Some(op_cache),
     msg::Any::Chdir => Some(op_chdir),
     msg::Any::Chmod => Some(op_chmod),
     msg::Any::Chown => Some(op_chown),
-    msg::Any::Close => Some(op_close),
     msg::Any::CopyFile => Some(op_copy_file),
     msg::Any::CreateWorker => Some(op_create_worker),
     msg::Any::Cwd => Some(op_cwd),
     msg::Any::Dial => Some(op_dial),
-    msg::Any::Fetch => Some(op_fetch),
-    msg::Any::FetchSourceFile => Some(op_fetch_source_file),
-    msg::Any::FormatError => Some(op_format_error),
     msg::Any::GetRandomValues => Some(op_get_random_values),
     msg::Any::GlobalTimer => Some(op_global_timer),
     msg::Any::GlobalTimerStop => Some(op_global_timer_stop),
@@ -168,7 +159,6 @@ pub fn op_selector_std(inner_type: msg::Any) -> Option<CliDispatchFn> {
     msg::Any::Metrics => Some(op_metrics),
     msg::Any::Mkdir => Some(op_mkdir),
     msg::Any::Now => Some(op_now),
-    msg::Any::Open => Some(op_open),
     msg::Any::PermissionRevoke => Some(op_revoke_permission),
     msg::Any::Permissions => Some(op_permissions),
     msg::Any::Read => Some(op_read),
@@ -181,7 +171,6 @@ pub fn op_selector_std(inner_type: msg::Any) -> Option<CliDispatchFn> {
     msg::Any::Resources => Some(op_resources),
     msg::Any::Run => Some(op_run),
     msg::Any::RunStatus => Some(op_run_status),
-    msg::Any::Seek => Some(op_seek),
     msg::Any::Shutdown => Some(op_shutdown),
     msg::Any::Stat => Some(op_stat),
     msg::Any::Symlink => Some(op_symlink),
diff --git a/cli/ops/errors.rs b/cli/ops/errors.rs
index a27f3656e..cd21a3880 100644
--- a/cli/ops/errors.rs
+++ b/cli/ops/errors.rs
@@ -1,88 +1,56 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
-use super::dispatch_flatbuffers::serialize_response;
-use super::utils::*;
-use crate::deno_error;
+use super::dispatch_json::{Deserialize, JsonOp, Value};
 use crate::fmt_errors::JSError;
-use crate::msg;
 use crate::source_maps::get_orig_position;
 use crate::source_maps::CachedMaps;
 use crate::state::ThreadSafeState;
 use deno::*;
-use flatbuffers::FlatBufferBuilder;
 use std::collections::HashMap;
 
+#[derive(Deserialize)]
+struct FormatErrorArgs {
+  error: String,
+}
+
 pub fn op_format_error(
   state: &ThreadSafeState,
-  base: &msg::Base<'_>,
-  data: Option<PinnedBuf>,
-) -> CliOpResult {
-  assert!(data.is_none());
-  let inner = base.inner_as_format_error().unwrap();
-  let json_str = inner.error().unwrap();
-  let error = JSError::from_json(json_str, &state.ts_compiler);
-  let error_string = error.to_string();
-
-  let mut builder = FlatBufferBuilder::new();
-  let new_error = builder.create_string(&error_string);
-
-  let inner = msg::FormatErrorRes::create(
-    &mut builder,
-    &msg::FormatErrorResArgs {
-      error: Some(new_error),
-    },
-  );
-
-  let response_buf = serialize_response(
-    base.cmd_id(),
-    &mut builder,
-    msg::BaseArgs {
-      inner_type: msg::Any::FormatErrorRes,
-      inner: Some(inner.as_union_value()),
-      ..Default::default()
-    },
-  );
+  args: Value,
+  _zero_copy: Option<PinnedBuf>,
+) -> Result<JsonOp, ErrBox> {
+  let args: FormatErrorArgs = serde_json::from_value(args)?;
+  let error = JSError::from_json(&args.error, &state.ts_compiler);
+
+  Ok(JsonOp::Sync(json!({
+    "error": error.to_string(),
+  })))
+}
 
-  ok_buf(response_buf)
+#[derive(Deserialize)]
+struct ApplySourceMap {
+  filename: String,
+  line: i32,
+  column: i32,
 }
 
 pub fn op_apply_source_map(
   state: &ThreadSafeState,
-  base: &msg::Base<'_>,
-  data: Option<PinnedBuf>,
-) -> CliOpResult {
-  if !base.sync() {
-    return Err(deno_error::no_async_support());
-  }
-  assert!(data.is_none());
-  let inner = base.inner_as_apply_source_map().unwrap();
-  let cmd_id = base.cmd_id();
-  let filename = inner.filename().unwrap();
-  let line = inner.line();
-  let column = inner.column();
+  args: Value,
+  _zero_copy: Option<PinnedBuf>,
+) -> Result<JsonOp, ErrBox> {
+  let args: ApplySourceMap = serde_json::from_value(args)?;
 
   let mut mappings_map: CachedMaps = HashMap::new();
   let (orig_filename, orig_line, orig_column) = get_orig_position(
-    filename.to_owned(),
-    line.into(),
-    column.into(),
+    args.filename,
+    args.line.into(),
+    args.column.into(),
     &mut mappings_map,
     &state.ts_compiler,
   );
 
-  let builder = &mut FlatBufferBuilder::new();
-  let msg_args = msg::ApplySourceMapArgs {
-    filename: Some(builder.create_string(&orig_filename)),
-    line: orig_line as i32,
-    column: orig_column as i32,
-  };
-  let res_inner = msg::ApplySourceMap::create(builder, &msg_args);
-  ok_buf(serialize_response(
-    cmd_id,
-    builder,
-    msg::BaseArgs {
-      inner: Some(res_inner.as_union_value()),
-      inner_type: msg::Any::ApplySourceMap,
-      ..Default::default()
-    },
-  ))
+  Ok(JsonOp::Sync(json!({
+    "filename": orig_filename.to_string(),
+    "line": orig_line as u32,
+    "column": orig_column as u32,
+  })))
 }
diff --git a/cli/ops/fetch.rs b/cli/ops/fetch.rs
index 7661eb6e9..e2ab81a81 100644
--- a/cli/ops/fetch.rs
+++ b/cli/ops/fetch.rs
@@ -1,38 +1,57 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
-use super::dispatch_flatbuffers::serialize_response;
-use super::utils::CliOpResult;
+use super::dispatch_json::{Deserialize, JsonOp, Value};
 use crate::http_util;
-use crate::msg;
-use crate::msg_util;
 use crate::resources;
 use crate::state::ThreadSafeState;
 use deno::*;
-use flatbuffers::FlatBufferBuilder;
+use http::header::HeaderName;
+use http::uri::Uri;
+use http::Method;
 use hyper;
+use hyper::header::HeaderValue;
 use hyper::rt::Future;
+use hyper::Request;
 use std;
 use std::convert::From;
+use std::str::FromStr;
+
+#[derive(Deserialize)]
+struct FetchArgs {
+  method: Option<String>,
+  url: String,
+  headers: Vec<(String, String)>,
+}
 
 pub fn op_fetch(
   state: &ThreadSafeState,
-  base: &msg::Base<'_>,
+  args: Value,
   data: Option<PinnedBuf>,
-) -> CliOpResult {
-  let inner = base.inner_as_fetch().unwrap();
-  let cmd_id = base.cmd_id();
-
-  let header = inner.header().unwrap();
-  assert!(header.is_request());
-  let url = header.url().unwrap();
+) -> Result<JsonOp, ErrBox> {
+  let args: FetchArgs = serde_json::from_value(args)?;
+  let url = args.url;
 
   let body = match data {
     None => hyper::Body::empty(),
     Some(buf) => hyper::Body::from(Vec::from(&*buf)),
   };
 
-  let req = msg_util::deserialize_request(header, body)?;
+  let mut req = Request::new(body);
+  let uri = Uri::from_str(&url).map_err(ErrBox::from)?;
+  *req.uri_mut() = uri;
+
+  if let Some(method) = args.method {
+    let method = Method::from_str(&method).unwrap();
+    *req.method_mut() = method;
+  }
+
+  let headers = req.headers_mut();
+  for header_pair in args.headers {
+    let name = HeaderName::from_bytes(header_pair.0.as_bytes()).unwrap();
+    let v = HeaderValue::from_str(&header_pair.1).unwrap();
+    headers.insert(name, v);
+  }
 
-  let url_ = url::Url::parse(url).map_err(ErrBox::from)?;
+  let url_ = url::Url::parse(&url).map_err(ErrBox::from)?;
   state.check_net_url(&url_)?;
 
   let client = http_util::get_client();
@@ -42,32 +61,22 @@ pub fn op_fetch(
     .request(req)
     .map_err(ErrBox::from)
     .and_then(move |res| {
-      let builder = &mut FlatBufferBuilder::new();
-      let header_off = msg_util::serialize_http_response(builder, &res);
+      let status = res.status().as_u16();
+      let mut res_headers = Vec::new();
+      for (key, val) in res.headers().iter() {
+        res_headers.push((key.to_string(), val.to_str().unwrap().to_owned()));
+      }
       let body = res.into_body();
       let body_resource = resources::add_hyper_body(body);
-      let inner = msg::FetchRes::create(
-        builder,
-        &msg::FetchResArgs {
-          header: Some(header_off),
-          body_rid: body_resource.rid,
-        },
-      );
 
-      Ok(serialize_response(
-        cmd_id,
-        builder,
-        msg::BaseArgs {
-          inner: Some(inner.as_union_value()),
-          inner_type: msg::Any::FetchRes,
-          ..Default::default()
-        },
-      ))
+      let json_res = json!({
+        "bodyRid": body_resource.rid,
+        "status": status,
+        "headers": res_headers
+      });
+
+      futures::future::ok(json_res)
     });
-  if base.sync() {
-    let result_buf = future.wait()?;
-    Ok(Op::Sync(result_buf))
-  } else {
-    Ok(Op::Async(Box::new(future)))
-  }
+
+  Ok(JsonOp::Async(Box::new(future)))
 }
diff --git a/cli/ops/files.rs b/cli/ops/files.rs
index 023bd65f9..c02a69b9c 100644
--- a/cli/ops/files.rs
+++ b/cli/ops/files.rs
@@ -1,5 +1,6 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
 use super::dispatch_flatbuffers::serialize_response;
+use super::dispatch_json::{Deserialize, JsonOp, Value};
 use super::utils::*;
 use crate::deno_error;
 use crate::fs as deno_fs;
@@ -14,17 +15,22 @@ use std;
 use std::convert::From;
 use tokio;
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct OpenArgs {
+  promise_id: Option<u64>,
+  filename: String,
+  mode: String,
+}
+
 pub fn op_open(
   state: &ThreadSafeState,
-  base: &msg::Base<'_>,
-  data: Option<PinnedBuf>,
-) -> CliOpResult {
-  assert!(data.is_none());
-  let cmd_id = base.cmd_id();
-  let inner = base.inner_as_open().unwrap();
-  let (filename, filename_) =
-    deno_fs::resolve_from_cwd(inner.filename().unwrap())?;
-  let mode = inner.mode().unwrap();
+  args: Value,
+  _zero_copy: Option<PinnedBuf>,
+) -> Result<JsonOp, ErrBox> {
+  let args: OpenArgs = serde_json::from_value(args)?;
+  let (filename, filename_) = deno_fs::resolve_from_cwd(&args.filename)?;
+  let mode = args.mode.as_ref();
 
   let mut open_options = tokio::fs::OpenOptions::new();
 
@@ -75,44 +81,39 @@ pub fn op_open(
     }
   }
 
+  let is_sync = args.promise_id.is_none();
   let op = open_options.open(filename).map_err(ErrBox::from).and_then(
     move |fs_file| {
       let resource = resources::add_fs_file(fs_file);
-      let builder = &mut FlatBufferBuilder::new();
-      let inner =
-        msg::OpenRes::create(builder, &msg::OpenResArgs { rid: resource.rid });
-      Ok(serialize_response(
-        cmd_id,
-        builder,
-        msg::BaseArgs {
-          inner: Some(inner.as_union_value()),
-          inner_type: msg::Any::OpenRes,
-          ..Default::default()
-        },
-      ))
+      futures::future::ok(json!(resource.rid))
     },
   );
-  if base.sync() {
+
+  if is_sync {
     let buf = op.wait()?;
-    Ok(Op::Sync(buf))
+    Ok(JsonOp::Sync(buf))
   } else {
-    Ok(Op::Async(Box::new(op)))
+    Ok(JsonOp::Async(Box::new(op)))
   }
 }
 
+#[derive(Deserialize)]
+struct CloseArgs {
+  rid: i32,
+}
+
 pub fn op_close(
   _state: &ThreadSafeState,
-  base: &msg::Base<'_>,
-  data: Option<PinnedBuf>,
-) -> CliOpResult {
-  assert!(data.is_none());
-  let inner = base.inner_as_close().unwrap();
-  let rid = inner.rid();
-  match resources::lookup(rid) {
+  args: Value,
+  _zero_copy: Option<PinnedBuf>,
+) -> Result<JsonOp, ErrBox> {
+  let args: CloseArgs = serde_json::from_value(args)?;
+
+  match resources::lookup(args.rid as u32) {
     None => Err(deno_error::bad_resource()),
     Some(resource) => {
       resource.close();
-      ok_buf(empty_buf())
+      Ok(JsonOp::Sync(json!({})))
     }
   }
 }
@@ -202,27 +203,32 @@ pub fn op_write(
   }
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct SeekArgs {
+  promise_id: Option<u64>,
+  rid: i32,
+  offset: i32,
+  whence: i32,
+}
+
 pub fn op_seek(
   _state: &ThreadSafeState,
-  base: &msg::Base<'_>,
-  data: Option<PinnedBuf>,
-) -> CliOpResult {
-  assert!(data.is_none());
-  let inner = base.inner_as_seek().unwrap();
-  let rid = inner.rid();
-  let offset = inner.offset();
-  let whence = inner.whence();
+  args: Value,
+  _zero_copy: Option<PinnedBuf>,
+) -> Result<JsonOp, ErrBox> {
+  let args: SeekArgs = serde_json::from_value(args)?;
 
-  match resources::lookup(rid) {
+  match resources::lookup(args.rid as u32) {
     None => Err(deno_error::bad_resource()),
     Some(resource) => {
-      let op = resources::seek(resource, offset, whence)
-        .and_then(move |_| Ok(empty_buf()));
-      if base.sync() {
+      let op = resources::seek(resource, args.offset, args.whence as u32)
+        .and_then(move |_| futures::future::ok(json!({})));
+      if args.promise_id.is_none() {
         let buf = op.wait()?;
-        Ok(Op::Sync(buf))
+        Ok(JsonOp::Sync(buf))
       } else {
-        Ok(Op::Async(Box::new(op)))
+        Ok(JsonOp::Async(Box::new(op)))
       }
     }
   }
diff --git a/cli/ops/mod.rs b/cli/ops/mod.rs
index d945c5e5f..6a80e610f 100644
--- a/cli/ops/mod.rs
+++ b/cli/ops/mod.rs
@@ -37,6 +37,14 @@ pub const OP_UTIME: OpId = 7;
 pub const OP_SET_ENV: OpId = 8;
 pub const OP_HOME_DIR: OpId = 9;
 pub const OP_START: OpId = 10;
+pub const OP_APPLY_SOURCE_MAP: OpId = 11;
+pub const OP_FORMAT_ERROR: OpId = 12;
+pub const OP_CACHE: OpId = 13;
+pub const OP_FETCH_SOURCE_FILE: OpId = 14;
+pub const OP_OPEN: OpId = 15;
+pub const OP_CLOSE: OpId = 16;
+pub const OP_SEEK: OpId = 17;
+pub const OP_FETCH: OpId = 18;
 
 pub fn dispatch(
   state: &ThreadSafeState,
@@ -74,6 +82,39 @@ pub fn dispatch(
     OP_START => {
       dispatch_json::dispatch(os::op_start, state, control, zero_copy)
     }
+    OP_APPLY_SOURCE_MAP => dispatch_json::dispatch(
+      errors::op_apply_source_map,
+      state,
+      control,
+      zero_copy,
+    ),
+    OP_FORMAT_ERROR => dispatch_json::dispatch(
+      errors::op_format_error,
+      state,
+      control,
+      zero_copy,
+    ),
+    OP_CACHE => {
+      dispatch_json::dispatch(compiler::op_cache, state, control, zero_copy)
+    }
+    OP_FETCH_SOURCE_FILE => dispatch_json::dispatch(
+      compiler::op_fetch_source_file,
+      state,
+      control,
+      zero_copy,
+    ),
+    OP_OPEN => {
+      dispatch_json::dispatch(files::op_open, state, control, zero_copy)
+    }
+    OP_CLOSE => {
+      dispatch_json::dispatch(files::op_close, state, control, zero_copy)
+    }
+    OP_SEEK => {
+      dispatch_json::dispatch(files::op_seek, state, control, zero_copy)
+    }
+    OP_FETCH => {
+      dispatch_json::dispatch(fetch::op_fetch, state, control, zero_copy)
+    }
     OP_FLATBUFFER => dispatch_flatbuffers::dispatch(state, control, zero_copy),
     _ => panic!("bad op_id"),
   };