5 files changed, 275 insertions, 0 deletions

diff --git a/src/handlers.rs b/src/handlers.rs
index 80eb2871e..ad701c4e1 100644
--- a/src/handlers.rs
+++ b/src/handlers.rs
@@ -9,6 +9,7 @@ use isolate::Isolate;
 use isolate::IsolateState;
 use isolate::Op;
 use msg;
+use tokio_util;
 
 use flatbuffers::FlatBufferBuilder;
 use futures;
@@ -21,14 +22,18 @@ use remove_dir_all::remove_dir_all;
 use resources;
 use std;
 use std::fs;
+use std::net::SocketAddr;
 #[cfg(any(unix))]
 use std::os::unix::fs::PermissionsExt;
 use std::path::Path;
 use std::path::PathBuf;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::time::UNIX_EPOCH;
 use std::time::{Duration, Instant};
 use tokio;
+use tokio::net::TcpListener;
+use tokio::net::TcpStream;
 use tokio_io;
 use tokio_threadpool;
 
@@ -74,6 +79,7 @@ pub fn msg_from_js(
       msg::Any::Open => handle_open,
       msg::Any::Read => handle_read,
       msg::Any::Write => handle_write,
+      msg::Any::Close => handle_close,
       msg::Any::Remove => handle_remove,
       msg::Any::ReadFile => handle_read_file,
       msg::Any::ReadDir => handle_read_dir,
@@ -86,6 +92,9 @@ pub fn msg_from_js(
       msg::Any::WriteFile => handle_write_file,
       msg::Any::Exit => handle_exit,
       msg::Any::CopyFile => handle_copy_file,
+      msg::Any::Listen => handle_listen,
+      msg::Any::Accept => handle_accept,
+      msg::Any::Dial => handle_dial,
       _ => panic!(format!(
         "Unhandled message {}",
         msg::enum_name_any(msg_type)
@@ -581,6 +590,26 @@ fn handle_open(
   Box::new(op)
 }
 
+fn handle_close(
+  _state: Arc<IsolateState>,
+  base: &msg::Base,
+  data: &'static mut [u8],
+) -> Box<Op> {
+  assert_eq!(data.len(), 0);
+  let msg = base.msg_as_close().unwrap();
+  let rid = msg.rid();
+  match resources::lookup(rid) {
+    None => odd_future(errors::new(
+      errors::ErrorKind::BadFileDescriptor,
+      String::from("Bad File Descriptor"),
+    )),
+    Some(mut resource) => {
+      resource.close();
+      ok_future(empty_buf())
+    }
+  }
+}
+
 fn handle_read(
   _state: Arc<IsolateState>,
   base: &msg::Base,
@@ -994,3 +1023,124 @@ fn handle_truncate(
     Ok(empty_buf())
   })
 }
+
+fn handle_listen(
+  state: Arc<IsolateState>,
+  base: &msg::Base,
+  data: &'static mut [u8],
+) -> Box<Op> {
+  assert_eq!(data.len(), 0);
+  if !state.flags.allow_net {
+    return odd_future(permission_denied());
+  }
+
+  let cmd_id = base.cmd_id();
+  let msg = base.msg_as_listen().unwrap();
+  let network = msg.network().unwrap();
+  assert_eq!(network, "tcp");
+  let address = msg.address().unwrap();
+
+  Box::new(futures::future::result((move || {
+    // TODO properly parse addr
+    let addr = SocketAddr::from_str(address).unwrap();
+
+    let listener = TcpListener::bind(&addr)?;
+    let resource = resources::add_tcp_listener(listener);
+
+    let builder = &mut FlatBufferBuilder::new();
+    let msg = msg::ListenRes::create(
+      builder,
+      &msg::ListenResArgs {
+        rid: resource.rid,
+        ..Default::default()
+      },
+    );
+    Ok(serialize_response(
+      cmd_id,
+      builder,
+      msg::BaseArgs {
+        msg: Some(msg.as_union_value()),
+        msg_type: msg::Any::ListenRes,
+        ..Default::default()
+      },
+    ))
+  })()))
+}
+
+fn new_conn(cmd_id: u32, tcp_stream: TcpStream) -> OpResult {
+  let tcp_stream_resource = resources::add_tcp_stream(tcp_stream);
+  // TODO forward socket_addr to client.
+
+  let builder = &mut FlatBufferBuilder::new();
+  let msg = msg::NewConn::create(
+    builder,
+    &msg::NewConnArgs {
+      rid: tcp_stream_resource.rid,
+      ..Default::default()
+    },
+  );
+  Ok(serialize_response(
+    cmd_id,
+    builder,
+    msg::BaseArgs {
+      msg: Some(msg.as_union_value()),
+      msg_type: msg::Any::NewConn,
+      ..Default::default()
+    },
+  ))
+}
+
+fn handle_accept(
+  state: Arc<IsolateState>,
+  base: &msg::Base,
+  data: &'static mut [u8],
+) -> Box<Op> {
+  assert_eq!(data.len(), 0);
+  if !state.flags.allow_net {
+    return odd_future(permission_denied());
+  }
+
+  let cmd_id = base.cmd_id();
+  let msg = base.msg_as_accept().unwrap();
+  let server_rid = msg.rid();
+
+  match resources::lookup(server_rid) {
+    None => odd_future(errors::new(
+      errors::ErrorKind::BadFileDescriptor,
+      String::from("Bad File Descriptor"),
+    )),
+    Some(server_resource) => {
+      let op = tokio_util::accept(server_resource)
+        .map_err(|err| DenoError::from(err))
+        .and_then(move |(tcp_stream, _socket_addr)| {
+          new_conn(cmd_id, tcp_stream)
+        });
+      Box::new(op)
+    }
+  }
+}
+
+fn handle_dial(
+  state: Arc<IsolateState>,
+  base: &msg::Base,
+  data: &'static mut [u8],
+) -> Box<Op> {
+  assert_eq!(data.len(), 0);
+  if !state.flags.allow_net {
+    return odd_future(permission_denied());
+  }
+
+  let cmd_id = base.cmd_id();
+  let msg = base.msg_as_dial().unwrap();
+  let network = msg.network().unwrap();
+  assert_eq!(network, "tcp");
+  let address = msg.address().unwrap();
+
+  // TODO properly parse addr
+  let addr = SocketAddr::from_str(address).unwrap();
+
+  let op = TcpStream::connect(&addr)
+    .map_err(|err| err.into())
+    .and_then(move |tcp_stream| new_conn(cmd_id, tcp_stream));
+  Box::new(op)
+}
diff --git a/src/main.rs b/src/main.rs
index d07eb5b07..82681cc34 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -1,5 +1,6 @@
 // Copyright 2018 the Deno authors. All rights reserved. MIT license.
 extern crate flatbuffers;
+#[macro_use]
 extern crate futures;
 extern crate hyper;
 extern crate libc;
diff --git a/src/msg.fbs b/src/msg.fbs
index 83dca11e6..b47230389 100644
--- a/src/msg.fbs
+++ b/src/msg.fbs
@@ -35,6 +35,11 @@ union Any {
   Write,
   WriteRes,
   Close,
+  Listen,
+  ListenRes,
+  Accept,
+  Dial,
+  NewConn,
 }
 
 enum ErrorKind: byte {
@@ -285,4 +290,29 @@ table Close {
   rid: int;
 }
 
+table Listen {
+  network: string;
+  address: string;
+}
+
+table ListenRes {
+  rid: int;
+}
+
+table Accept {
+  rid: int;
+}
+
+table Dial {
+  network: string;
+  address: string;
+}
+
+// Response to Accept and Dial.
+table NewConn {
+  rid: int;
+  remote_addr: string;
+  local_addr: string;
+}
+
 root_type Base;
diff --git a/src/resources.rs b/src/resources.rs
index 940b14d21..75bad04b7 100644
--- a/src/resources.rs
+++ b/src/resources.rs
@@ -14,11 +14,13 @@ use std;
 use std::collections::HashMap;
 use std::io::Error;
 use std::io::{Read, Write};
+use std::net::SocketAddr;
 use std::sync::atomic::AtomicIsize;
 use std::sync::atomic::Ordering;
 use std::sync::Mutex;
 use tokio;
 use tokio::io::{AsyncRead, AsyncWrite};
+use tokio::net::TcpStream;
 
 pub type ResourceId = i32; // Sometimes referred to RID.
 
@@ -45,14 +47,40 @@ enum Repr {
   Stdout(tokio::io::Stdout),
   Stderr(tokio::io::Stderr),
   FsFile(tokio::fs::File),
+  TcpListener(tokio::net::TcpListener),
+  TcpStream(tokio::net::TcpStream),
 }
 
 // Abstract async file interface.
 // Ideally in unix, if Resource represents an OS rid, it will be the same.
+#[derive(Debug)]
 pub struct Resource {
   pub rid: ResourceId,
 }
 
+impl Resource {
+  // TODO Should it return a Resource instead of net::TcpStream?
+  pub fn poll_accept(&mut self) -> Poll<(TcpStream, SocketAddr), Error> {
+    let mut table = RESOURCE_TABLE.lock().unwrap();
+    let maybe_repr = table.get_mut(&self.rid);
+    match maybe_repr {
+      None => panic!("bad rid"),
+      Some(repr) => match repr {
+        Repr::TcpListener(ref mut s) => s.poll_accept(),
+        _ => panic!("Cannot accept"),
+      },
+    }
+  }
+
+  // close(2) is done by dropping the value. Therefore we just need to remove
+  // the resource from the RESOURCE_TABLE.
+  pub fn close(&mut self) {
+    let mut table = RESOURCE_TABLE.lock().unwrap();
+    let r = table.remove(&self.rid);
+    assert!(r.is_some());
+  }
+}
+
 impl Read for Resource {
   fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
     unimplemented!();
@@ -68,9 +96,11 @@ impl AsyncRead for Resource {
       Some(repr) => match repr {
         Repr::FsFile(ref mut f) => f.poll_read(buf),
         Repr::Stdin(ref mut f) => f.poll_read(buf),
+        Repr::TcpStream(ref mut f) => f.poll_read(buf),
         Repr::Stdout(_) | Repr::Stderr(_) => {
           panic!("Cannot read from stdout/stderr")
         }
+        Repr::TcpListener(_) => panic!("Cannot read"),
       },
     }
   }
@@ -96,7 +126,9 @@ impl AsyncWrite for Resource {
         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::TcpStream(ref mut f) => f.poll_write(buf),
         Repr::Stdin(_) => panic!("Cannot write to stdin"),
+        Repr::TcpListener(_) => panic!("Cannot write"),
       },
     }
   }
@@ -120,6 +152,22 @@ pub fn add_fs_file(fs_file: tokio::fs::File) -> Resource {
   }
 }
 
+pub fn add_tcp_listener(listener: tokio::net::TcpListener) -> Resource {
+  let rid = new_rid();
+  let mut tg = RESOURCE_TABLE.lock().unwrap();
+  let r = tg.insert(rid, Repr::TcpListener(listener));
+  assert!(r.is_none());
+  Resource { rid }
+}
+
+pub fn add_tcp_stream(stream: tokio::net::TcpStream) -> Resource {
+  let rid = new_rid();
+  let mut tg = RESOURCE_TABLE.lock().unwrap();
+  let r = tg.insert(rid, Repr::TcpStream(stream));
+  assert!(r.is_none());
+  Resource { rid }
+}
+
 pub fn lookup(rid: ResourceId) -> Option<Resource> {
   let table = RESOURCE_TABLE.lock().unwrap();
   table.get(&rid).map(|_| Resource { rid })
diff --git a/src/tokio_util.rs b/src/tokio_util.rs
index de81620ef..0a2a34917 100644
--- a/src/tokio_util.rs
+++ b/src/tokio_util.rs
@@ -1,8 +1,15 @@
 // Copyright 2018 the Deno authors. All rights reserved. MIT license.
 
+use resources::Resource;
+
 use futures;
 use futures::Future;
+use futures::Poll;
+use std::io;
+use std::mem;
+use std::net::SocketAddr;
 use tokio;
+use tokio::net::TcpStream;
 use tokio_executor;
 
 pub fn block_on<F, R, E>(future: F) -> Result<R, E>
@@ -28,3 +35,42 @@ where
   let mut enter = tokio_executor::enter().expect("Multiple executors at once");
   tokio_executor::with_default(&mut executor, &mut enter, move |_enter| f());
 }
+
+#[derive(Debug)]
+enum AcceptState {
+  Pending(Resource),
+  Empty,
+}
+
+/// Simply accepts a connection.
+pub fn accept(r: Resource) -> Accept {
+  Accept {
+    state: AcceptState::Pending(r),
+  }
+}
+
+/// A future which can be used to easily read available number of bytes to fill
+/// a buffer.
+///
+/// Created by the [`read`] function.
+#[derive(Debug)]
+pub struct Accept {
+  state: AcceptState,
+}
+
+impl Future for Accept {
+  type Item = (TcpStream, SocketAddr);
+  type Error = io::Error;
+
+  fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+    let (stream, addr) = match self.state {
+      AcceptState::Pending(ref mut r) => try_ready!(r.poll_accept()),
+      AcceptState::Empty => panic!("poll Accept after it's done"),
+    };
+
+    match mem::replace(&mut self.state, AcceptState::Empty) {
+      AcceptState::Pending(_) => Ok((stream, addr).into()),
+      AcceptState::Empty => panic!("invalid internal state"),
+    }
+  }
+}