1 files changed, 84 insertions, 199 deletions

diff --git a/cli/compiler.rs b/cli/compiler.rs
index d4913a4e2..e1bb56130 100644
--- a/cli/compiler.rs
+++ b/cli/compiler.rs
@@ -1,10 +1,6 @@
 // Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
-use crate::js_errors;
-use crate::js_errors::JSErrorColor;
 use crate::msg;
-use crate::ops::op_selector_compiler;
 use crate::resources;
-use crate::resources::ResourceId;
 use crate::startup_data;
 use crate::state::*;
 use crate::tokio_util;
@@ -12,31 +8,10 @@ use crate::worker::Worker;
 use deno::js_check;
 use deno::Buf;
 use deno::JSError;
-use futures::future::*;
-use futures::sync::oneshot;
 use futures::Future;
 use futures::Stream;
-use serde_json;
-use std::collections::HashMap;
 use std::str;
-use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
-use std::sync::Mutex;
-use tokio::runtime::Runtime;
-
-type CmdId = u32;
-type ResponseSenderTable = HashMap<CmdId, oneshot::Sender<Buf>>;
-
-lazy_static! {
-  static ref C_NEXT_CMD_ID: AtomicUsize = AtomicUsize::new(1);
-  // Map of response senders
-  static ref C_RES_SENDER_TABLE: Mutex<ResponseSenderTable> = Mutex::new(ResponseSenderTable::new());
-  // Shared worker resources so we can spawn
-  static ref C_RID: Mutex<Option<ResourceId>> = Mutex::new(None);
-  // tokio runtime specifically for spawning logic that is dependent on
-  // completetion of the compiler worker future
-  static ref C_RUNTIME: Mutex<Runtime> = Mutex::new(tokio_util::create_threadpool_runtime());
-}
 
 // This corresponds to JS ModuleMetaData.
 // TODO Rename one or the other so they correspond.
@@ -72,91 +47,22 @@ impl ModuleMetaData {
   }
 }
 
-fn new_cmd_id() -> CmdId {
-  let next_rid = C_NEXT_CMD_ID.fetch_add(1, Ordering::SeqCst);
-  next_rid as CmdId
-}
-
-fn parse_cmd_id(res_json: &str) -> CmdId {
-  match serde_json::from_str::<serde_json::Value>(res_json) {
-    Ok(serde_json::Value::Object(map)) => match map["cmdId"].as_u64() {
-      Some(cmd_id) => cmd_id as CmdId,
-      _ => panic!("Error decoding compiler response: expected cmdId"),
-    },
-    _ => panic!("Error decoding compiler response"),
-  }
-}
-
-fn lazy_start(parent_state: ThreadSafeState) -> ResourceId {
-  let mut cell = C_RID.lock().unwrap();
-  cell
-    .get_or_insert_with(|| {
-      let child_state = ThreadSafeState::new(
-        parent_state.flags.clone(),
-        parent_state.argv.clone(),
-        op_selector_compiler,
-        parent_state.progress.clone(),
-      );
-      let rid = child_state.resource.rid;
-      let resource = child_state.resource.clone();
-
-      let mut worker = Worker::new(
-        "TS".to_string(),
-        startup_data::compiler_isolate_init(),
-        child_state,
-      );
-
-      js_check(worker.execute("denoMain()"));
-      js_check(worker.execute("workerMain()"));
-      js_check(worker.execute("compilerMain()"));
+type CompilerConfig = Option<(String, Vec<u8>)>;
 
-      let mut runtime = C_RUNTIME.lock().unwrap();
-      runtime.spawn(lazy(move || {
-        worker.then(move |result| -> Result<(), ()> {
-          // Close resource so the future created by
-          // handle_worker_message_stream exits
-          resource.close();
-          debug!("Compiler worker exited!");
-          if let Err(e) = result {
-            eprintln!("{}", JSErrorColor(&e).to_string());
-          }
-          std::process::exit(1);
-        })
-      }));
-      runtime.spawn(lazy(move || {
-        debug!("Start worker stream handler!");
-        let worker_stream = resources::get_message_stream_from_worker(rid);
-        worker_stream
-          .for_each(|msg: Buf| {
-            // All worker responses are handled here first before being sent via
-            // their respective sender. This system can be compared to the
-            // promise system used on the js side. This provides a way to
-            // resolve many futures via the same channel.
-            let res_json = std::str::from_utf8(&msg).unwrap();
-            debug!("Got message from worker: {}", res_json);
-            // Get the intended receiver's cmd_id from the message.
-            let cmd_id = parse_cmd_id(res_json);
-            let mut table = C_RES_SENDER_TABLE.lock().unwrap();
-            debug!("Cmd id for get message handler: {}", cmd_id);
-            // Get the corresponding response sender from the table and
-            // send a response.
-            let response_sender = table.remove(&(cmd_id as CmdId)).unwrap();
-            response_sender.send(msg).unwrap();
-            Ok(())
-          }).map_err(|_| ())
-      }));
-      rid
-    }).to_owned()
-}
-
-fn req(specifier: &str, referrer: &str, cmd_id: u32) -> Buf {
-  json!({
-    "specifier": specifier,
-    "referrer": referrer,
-    "cmdId": cmd_id,
-  }).to_string()
-  .into_boxed_str()
-  .into_boxed_bytes()
+/// Creates the JSON message send to compiler.ts's onmessage.
+fn req(root_names: Vec<String>, compiler_config: CompilerConfig) -> Buf {
+  let j = if let Some((config_path, config_data)) = compiler_config {
+    json!({
+      "rootNames": root_names,
+      "configPath": config_path,
+      "config": str::from_utf8(&config_data).unwrap(),
+    })
+  } else {
+    json!({
+      "rootNames": root_names,
+    })
+  };
+  j.to_string().into_boxed_str().into_boxed_bytes()
 }
 
 /// Returns an optional tuple which represents the state of the compiler
@@ -165,7 +71,7 @@ fn req(specifier: &str, referrer: &str, cmd_id: u32) -> Buf {
 pub fn get_compiler_config(
   parent_state: &ThreadSafeState,
   _compiler_type: &str,
-) -> Option<(String, Vec<u8>)> {
+) -> CompilerConfig {
   // The compiler type is being passed to make it easier to implement custom
   // compilers in the future.
   match (&parent_state.config_path, &parent_state.config) {
@@ -177,7 +83,7 @@ pub fn get_compiler_config(
 }
 
 pub fn compile_async(
-  parent_state: ThreadSafeState,
+  state: ThreadSafeState,
   specifier: &str,
   referrer: &str,
   module_meta_data: &ModuleMetaData,
@@ -186,100 +92,86 @@ pub fn compile_async(
     "Running rust part of compile_sync. specifier: {}, referrer: {}",
     &specifier, &referrer
   );
-  let cmd_id = new_cmd_id();
 
-  let req_msg = req(&specifier, &referrer, cmd_id);
+  let root_names = vec![module_meta_data.module_name.clone()];
+  let compiler_config = get_compiler_config(&state, "typescript");
+  let req_msg = req(root_names, compiler_config);
+
   let module_meta_data_ = module_meta_data.clone();
 
-  let compiler_rid = lazy_start(parent_state.clone());
+  // Count how many times we start the compiler worker.
+  state.metrics.compiler_starts.fetch_add(1, Ordering::SeqCst);
+
+  let mut worker = Worker::new(
+    "TS".to_string(),
+    startup_data::compiler_isolate_init(),
+    // TODO(ry) Maybe we should use a separate state for the compiler.
+    // as was done previously.
+    state.clone(),
+  );
+  js_check(worker.execute("denoMain()"));
+  js_check(worker.execute("workerMain()"));
+  js_check(worker.execute("compilerMain()"));
 
-  let compiling_job = parent_state
+  let compiling_job = state
     .progress
     .add(format!("Compiling {}", module_meta_data_.module_name));
 
-  let (local_sender, local_receiver) =
-    oneshot::channel::<Result<ModuleMetaData, Option<JSError>>>();
-
-  let (response_sender, response_receiver) = oneshot::channel::<Buf>();
-
-  // Scoping to auto dispose of locks when done using them
-  {
-    let mut table = C_RES_SENDER_TABLE.lock().unwrap();
-    debug!("Cmd id for response sender insert: {}", cmd_id);
-    // Place our response sender in the table so we can find it later.
-    table.insert(cmd_id, response_sender);
-
-    let mut runtime = C_RUNTIME.lock().unwrap();
-    runtime.spawn(lazy(move || {
-      resources::post_message_to_worker(compiler_rid, req_msg)
-        .then(move |_| {
-          debug!("Sent message to worker");
-          response_receiver.map_err(|_| None)
-        }).and_then(move |res_msg| {
-          debug!("Received message from worker");
-          let res_json = std::str::from_utf8(res_msg.as_ref()).unwrap();
-          let res = serde_json::from_str::<serde_json::Value>(res_json)
-            .expect("Error decoding compiler response");
-          let res_data = res["data"].as_object().expect(
-            "Error decoding compiler response: expected object field 'data'",
-          );
+  let resource = worker.state.resource.clone();
+  let compiler_rid = resource.rid;
+  let first_msg_fut = resources::post_message_to_worker(compiler_rid, req_msg)
+    .then(move |_| worker)
+    .then(move |result| {
+      if let Err(err) = result {
+        // TODO(ry) Need to forward the error instead of exiting.
+        eprintln!("{}", err.to_string());
+        std::process::exit(1);
+      }
+      debug!("Sent message to worker");
+      let stream_future =
+        resources::get_message_stream_from_worker(compiler_rid).into_future();
+      stream_future.map(|(f, _rest)| f).map_err(|(f, _rest)| f)
+    });
+
+  first_msg_fut
+    .map_err(|_| panic!("not handled"))
+    .and_then(move |maybe_msg: Option<Buf>| {
+      let _res_msg = maybe_msg.unwrap();
+
+      debug!("Received message from worker");
+
+      // TODO res is EmitResult, use serde_derive to parse it. Errors from the
+      // worker or Diagnostics should be somehow forwarded to the caller!
+      // Currently they are handled inside compiler.ts with os.exit(1) and above
+      // with std::process::exit(1). This bad.
+
+      let r = state.dir.fetch_module_meta_data(
+        &module_meta_data_.module_name,
+        ".",
+        true,
+        true,
+      );
+      let module_meta_data_after_compile = r.unwrap();
 
-          // Explicit drop to keep reference alive until future completes.
-          drop(compiling_job);
+      // Explicit drop to keep reference alive until future completes.
+      drop(compiling_job);
 
-          match res["success"].as_bool() {
-            Some(true) => Ok(ModuleMetaData {
-              maybe_output_code: res_data["outputCode"]
-                .as_str()
-                .map(|s| s.as_bytes().to_owned()),
-              maybe_source_map: res_data["sourceMap"]
-                .as_str()
-                .map(|s| s.as_bytes().to_owned()),
-              ..module_meta_data_
-            }),
-            Some(false) => {
-              let js_error = JSError::from_json_value(
-                serde_json::Value::Object(res_data.clone()),
-              ).expect(
-                "Error decoding compiler response: failed to parse error",
-              );
-              Err(Some(js_errors::apply_source_map(
-                &js_error,
-                &parent_state.dir,
-              )))
-            }
-            _ => panic!(
-              "Error decoding compiler response: expected bool field 'success'"
-            ),
-          }
-        }).then(move |result| {
-          local_sender.send(result).expect("Oneshot send() failed");
-          Ok(())
-        })
-    }));
-  }
-
-  local_receiver
-    .map_err(|e| {
-      panic!(
-        "Local channel canceled before compile request could be completed: {}",
-        e
-      )
-    }).and_then(move |result| match result {
-      Ok(v) => futures::future::result(Ok(v)),
-      Err(Some(err)) => futures::future::result(Err(err)),
-      Err(None) => panic!("Failed to communicate with the compiler worker."),
+      Ok(module_meta_data_after_compile)
+    }).then(move |r| {
+      // TODO(ry) do this in worker's destructor.
+      // resource.close();
+      r
     })
 }
 
 pub fn compile_sync(
-  parent_state: ThreadSafeState,
+  state: ThreadSafeState,
   specifier: &str,
   referrer: &str,
   module_meta_data: &ModuleMetaData,
 ) -> Result<ModuleMetaData, JSError> {
   tokio_util::block_on(compile_async(
-    parent_state,
+    state,
     specifier,
     referrer,
     module_meta_data,
@@ -298,9 +190,13 @@ mod tests {
 
       let specifier = "./tests/002_hello.ts";
       let referrer = cwd_string + "/";
+      use crate::worker;
+      let module_name = worker::root_specifier_to_url(specifier)
+        .unwrap()
+        .to_string();
 
       let mut out = ModuleMetaData {
-        module_name: "xxx".to_owned(),
+        module_name,
         module_redirect_source_name: None,
         filename: "/tests/002_hello.ts".to_owned(),
         media_type: msg::MediaType::TypeScript,
@@ -323,17 +219,6 @@ mod tests {
   }
 
   #[test]
-  fn test_parse_cmd_id() {
-    let cmd_id = new_cmd_id();
-
-    let msg = req("Hello", "World", cmd_id);
-
-    let res_json = std::str::from_utf8(&msg).unwrap();
-
-    assert_eq!(parse_cmd_id(res_json), cmd_id);
-  }
-
-  #[test]
   fn test_get_compiler_config_no_flag() {
     let compiler_type = "typescript";
     let state = ThreadSafeState::mock();