Adds basic File I/O and FD table.

Adds deno.stdin, deno.stdout, deno.stderr, deno.open(), deno.write(),
deno.read(), deno.Reader, deno.Writer, deno.copy().

Fixes #721. tests/cat.ts works.

5 files changed, 312 insertions, 2 deletions

diff --git a/src/errors.rs b/src/errors.rs
index 1bedf8d40..872f3492e 100644
--- a/src/errors.rs
+++ b/src/errors.rs
@@ -15,15 +15,22 @@ pub struct DenoError {
 
 #[derive(Debug)]
 enum Repr {
-  // Simple(ErrorKind),
+  Simple(ErrorKind, String),
   IoErr(io::Error),
   UrlErr(url::ParseError),
   HyperErr(hyper::Error),
 }
 
+pub fn new(kind: ErrorKind, msg: String) -> DenoError {
+  DenoError {
+    repr: Repr::Simple(kind, msg),
+  }
+}
+
 impl DenoError {
   pub fn kind(&self) -> ErrorKind {
     match self.repr {
+      Repr::Simple(kind, ref _msg) => kind,
       // Repr::Simple(kind) => kind,
       Repr::IoErr(ref err) => {
         use std::io::ErrorKind::*;
@@ -87,10 +94,10 @@ impl DenoError {
 impl fmt::Display for DenoError {
   fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
     match self.repr {
+      Repr::Simple(_kind, ref _msg) => panic!("todo"),
       Repr::IoErr(ref err) => err.fmt(f),
       Repr::UrlErr(ref err) => err.fmt(f),
       Repr::HyperErr(ref err) => err.fmt(f),
-      // Repr::Simple(..) => Ok(()),
     }
   }
 }
@@ -98,6 +105,7 @@ impl fmt::Display for DenoError {
 impl std::error::Error for DenoError {
   fn description(&self) -> &str {
     match self.repr {
+      Repr::Simple(_kind, ref msg) => msg.as_str(),
       Repr::IoErr(ref err) => err.description(),
       Repr::UrlErr(ref err) => err.description(),
       Repr::HyperErr(ref err) => err.description(),
@@ -107,6 +115,7 @@ impl std::error::Error for DenoError {
 
   fn cause(&self) -> Option<&std::error::Error> {
     match self.repr {
+      Repr::Simple(_kind, ref _msg) => None,
       Repr::IoErr(ref err) => Some(err),
       Repr::UrlErr(ref err) => Some(err),
       Repr::HyperErr(ref err) => Some(err),
diff --git a/src/files.rs b/src/files.rs
new file mode 100644
index 000000000..64160bb84
--- /dev/null
+++ b/src/files.rs
@@ -0,0 +1,119 @@
+// Copyright 2018 the Deno authors. All rights reserved. MIT license.
+
+use futures;
+use futures::Poll;
+use std;
+use std::collections::HashMap;
+use std::io::Error;
+use std::io::{Read, Write};
+use std::sync::atomic::AtomicIsize;
+use std::sync::atomic::Ordering;
+use std::sync::Mutex;
+use tokio;
+use tokio::io::{AsyncRead, AsyncWrite};
+
+// These store Deno's file descriptors. These are not necessarally the operating
+// system ones.
+type FdTable = HashMap<i32, Repr>;
+
+lazy_static! {
+  // Starts at 3 because stdio is [0-2].
+  static ref NEXT_FD: AtomicIsize = AtomicIsize::new(3);
+  static ref FD_TABLE: Mutex<FdTable> = Mutex::new({
+    let mut m = HashMap::new();
+    // TODO Load these lazily during lookup?
+    m.insert(0, Repr::Stdin(tokio::io::stdin()));
+    m.insert(1, Repr::Stdout(tokio::io::stdout()));
+    m.insert(2, Repr::Stderr(tokio::io::stderr()));
+    m
+  });
+}
+
+// Internal representation of DFile.
+enum Repr {
+  Stdin(tokio::io::Stdin),
+  Stdout(tokio::io::Stdout),
+  Stderr(tokio::io::Stderr),
+  FsFile(tokio::fs::File),
+}
+
+// Abstract async file interface.
+// fd does not necessarally correspond to an OS fd.
+// Ideally in unix, if DFile represents an OS fd, it will be the same.
+pub struct DFile {
+  pub fd: i32,
+}
+
+impl Read for DFile {
+  fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
+    unimplemented!();
+  }
+}
+
+impl AsyncRead for DFile {
+  fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, Error> {
+    let mut table = FD_TABLE.lock().unwrap();
+    let maybe_repr = table.get_mut(&self.fd);
+    match maybe_repr {
+      None => panic!("bad fd"),
+      Some(repr) => match repr {
+        Repr::FsFile(ref mut f) => f.poll_read(buf),
+        Repr::Stdin(ref mut f) => f.poll_read(buf),
+        Repr::Stdout(_) | Repr::Stderr(_) => {
+          panic!("Cannot read from stdout/stderr")
+        }
+      },
+    }
+  }
+}
+
+impl Write for DFile {
+  fn write(&mut self, _buf: &[u8]) -> std::io::Result<usize> {
+    unimplemented!()
+  }
+
+  fn flush(&mut self) -> std::io::Result<()> {
+    unimplemented!()
+  }
+}
+
+impl AsyncWrite for DFile {
+  fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, Error> {
+    let mut table = FD_TABLE.lock().unwrap();
+    let maybe_repr = table.get_mut(&self.fd);
+    match maybe_repr {
+      None => panic!("bad fd"),
+      Some(repr) => match repr {
+        Repr::FsFile(ref mut f) => f.poll_write(buf),
+        Repr::Stdout(ref mut f) => f.poll_write(buf),
+        Repr::Stderr(ref mut f) => f.poll_write(buf),
+        Repr::Stdin(_) => panic!("Cannot write to stdin"),
+      },
+    }
+  }
+
+  fn shutdown(&mut self) -> futures::Poll<(), std::io::Error> {
+    unimplemented!()
+  }
+}
+
+fn new_fd() -> i32 {
+  // TODO If on unix, just extract the real FD of fs_file.
+  // let fd = AsRawFd::as_raw_fd(fs_file.std());
+  let next_fd = NEXT_FD.fetch_add(1, Ordering::SeqCst);
+  next_fd as i32
+}
+
+pub fn add_fs_file(fs_file: tokio::fs::File) -> DFile {
+  let fd = new_fd();
+  let mut tg = FD_TABLE.lock().unwrap();
+  match tg.insert(fd, Repr::FsFile(fs_file)) {
+    Some(_) => panic!("There is already a file with that fd"),
+    None => DFile { fd },
+  }
+}
+
+pub fn lookup(fd: i32) -> Option<DFile> {
+  let table = FD_TABLE.lock().unwrap();
+  table.get(&fd).map(|_| DFile { fd })
+}
diff --git a/src/handlers.rs b/src/handlers.rs
index 62bdabb61..b1ef67d94 100644
--- a/src/handlers.rs
+++ b/src/handlers.rs
@@ -1,5 +1,6 @@
 // Copyright 2018 the Deno authors. All rights reserved. MIT license.
 
+use errors;
 use errors::DenoError;
 use errors::DenoResult;
 use fs as deno_fs;
@@ -8,6 +9,7 @@ use isolate::IsolateState;
 use isolate::Op;
 use msg;
 
+use files;
 use flatbuffers::FlatBufferBuilder;
 use futures;
 use futures::future::poll_fn;
@@ -26,7 +28,10 @@ use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::UNIX_EPOCH;
 use std::time::{Duration, Instant};
+use tokio;
 use tokio::timer::Delay;
+use tokio_io::AsyncRead;
+use tokio_io::AsyncWrite;
 use tokio_threadpool;
 
 type OpResult = DenoResult<Buf>;
@@ -62,6 +67,9 @@ pub fn msg_from_js(
     msg::Any::TimerClear => handle_timer_clear,
     msg::Any::MakeTempDir => handle_make_temp_dir,
     msg::Any::Mkdir => handle_mkdir,
+    msg::Any::Open => handle_open,
+    msg::Any::Read => handle_read,
+    msg::Any::Write => handle_write,
     msg::Any::Remove => handle_remove,
     msg::Any::ReadFile => handle_read_file,
     msg::Any::Rename => handle_rename,
@@ -551,6 +559,129 @@ fn handle_mkdir(
   })
 }
 
+fn handle_open(
+  _state: Arc<IsolateState>,
+  base: &msg::Base,
+  data: &'static mut [u8],
+) -> Box<Op> {
+  assert_eq!(data.len(), 0);
+  let cmd_id = base.cmd_id();
+  let msg = base.msg_as_open().unwrap();
+  let filename = PathBuf::from(msg.filename().unwrap());
+  // TODO let perm = msg.perm();
+
+  let op = tokio::fs::File::open(filename)
+    .map_err(|err| DenoError::from(err))
+    .and_then(move |fs_file| -> OpResult {
+      let dfile = files::add_fs_file(fs_file);
+      let builder = &mut FlatBufferBuilder::new();
+      let msg = msg::OpenRes::create(
+        builder,
+        &msg::OpenResArgs {
+          fd: dfile.fd,
+          ..Default::default()
+        },
+      );
+      Ok(serialize_response(
+        cmd_id,
+        builder,
+        msg::BaseArgs {
+          msg: Some(msg.as_union_value()),
+          msg_type: msg::Any::OpenRes,
+          ..Default::default()
+        },
+      ))
+    });
+  Box::new(op)
+}
+
+fn handle_read(
+  _state: Arc<IsolateState>,
+  base: &msg::Base,
+  data: &'static mut [u8],
+) -> Box<Op> {
+  let cmd_id = base.cmd_id();
+  let msg = base.msg_as_read().unwrap();
+  let fd = msg.fd();
+
+  match files::lookup(fd) {
+    None => odd_future(errors::new(
+      errors::ErrorKind::BadFileDescriptor,
+      String::from("Bad File Descriptor"),
+    )),
+    Some(mut dfile) => {
+      let op = futures::future::poll_fn(move || {
+        let poll = dfile.poll_read(data);
+        poll
+      }).map_err(|err| DenoError::from(err))
+      .and_then(move |nread: usize| {
+        let builder = &mut FlatBufferBuilder::new();
+        let msg = msg::ReadRes::create(
+          builder,
+          &msg::ReadResArgs {
+            nread: nread as u32,
+            eof: nread == 0,
+            ..Default::default()
+          },
+        );
+        Ok(serialize_response(
+          cmd_id,
+          builder,
+          msg::BaseArgs {
+            msg: Some(msg.as_union_value()),
+            msg_type: msg::Any::ReadRes,
+            ..Default::default()
+          },
+        ))
+      });
+      Box::new(op)
+    }
+  }
+}
+
+fn handle_write(
+  _state: Arc<IsolateState>,
+  base: &msg::Base,
+  data: &'static mut [u8],
+) -> Box<Op> {
+  let cmd_id = base.cmd_id();
+  let msg = base.msg_as_write().unwrap();
+  let fd = msg.fd();
+
+  match files::lookup(fd) {
+    None => odd_future(errors::new(
+      errors::ErrorKind::BadFileDescriptor,
+      String::from("Bad File Descriptor"),
+    )),
+    Some(mut dfile) => {
+      let op = futures::future::poll_fn(move || {
+        let poll = dfile.poll_write(data);
+        poll
+      }).map_err(|err| DenoError::from(err))
+      .and_then(move |bytes_written: usize| {
+        let builder = &mut FlatBufferBuilder::new();
+        let msg = msg::WriteRes::create(
+          builder,
+          &msg::WriteResArgs {
+            nbyte: bytes_written as u32,
+            ..Default::default()
+          },
+        );
+        Ok(serialize_response(
+          cmd_id,
+          builder,
+          msg::BaseArgs {
+            msg: Some(msg.as_union_value()),
+            msg_type: msg::Any::WriteRes,
+            ..Default::default()
+          },
+        ))
+      });
+      Box::new(op)
+    }
+  }
+}
+
 fn handle_remove(
   state: Arc<IsolateState>,
   base: &msg::Base,
diff --git a/src/main.rs b/src/main.rs
index cc762f1ae..01be538e1 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -8,6 +8,8 @@ extern crate rand;
 extern crate tempfile;
 extern crate tokio;
 extern crate tokio_executor;
+extern crate tokio_fs;
+extern crate tokio_io;
 extern crate tokio_threadpool;
 extern crate url;
 #[macro_use]
@@ -21,6 +23,7 @@ extern crate ring;
 
 mod deno_dir;
 mod errors;
+mod files;
 mod flags;
 mod fs;
 pub mod handlers;
diff --git a/src/msg.fbs b/src/msg.fbs
index 0d78395ea..7f42cd637 100644
--- a/src/msg.fbs
+++ b/src/msg.fbs
@@ -27,6 +27,13 @@ union Any {
   Stat,
   StatRes,
   SetEnv,
+  Open,
+  OpenRes,
+  Read,
+  ReadRes,
+  Write,
+  WriteRes,
+  Close,
 }
 
 enum ErrorKind: byte {
@@ -53,6 +60,8 @@ enum ErrorKind: byte {
   Other,
   UnexpectedEof,
 
+  BadFileDescriptor,
+
   // url errors
 
   EmptyHost,
@@ -199,6 +208,7 @@ table ReadFileRes {
 
 table WriteFile {
   filename: string;
+  data: [ubyte];
   perm: uint;
   // perm specified by https://godoc.org/os#FileMode
 }
@@ -237,4 +247,42 @@ table StatRes {
   has_mode: bool; // false on windows
 }
 
+table WriteFileSync {
+  filename: string;
+  data: [ubyte];
+  perm: uint;
+  // perm specified by https://godoc.org/os#FileMode
+}
+
+table Open {
+  filename: string;
+  perm: uint;
+}
+
+table OpenRes {
+  fd: int;
+}
+
+table Read {
+  fd: int;
+  // (ptr, len) is passed as second parameter to libdeno.send().
+}
+
+table ReadRes {
+  nread: uint;
+  eof: bool;
+}
+
+table Write {
+  fd: int;
+}
+
+table WriteRes {
+  nbyte: uint;
+}
+
+table Close {
+  fd: int;
+}
+
 root_type Base;