use crate::deno_error;
use crate::deno_error::bad_resource;
use crate::http_body::HttpBody;
use crate::repl::Repl;
use crate::state::WorkerChannels;
use deno::Buf;
use deno::ErrBox;
use futures;
use futures::Future;
use futures::Poll;
use futures::Sink;
use futures::Stream;
use reqwest::r#async::Decoder as ReqwestDecoder;
use std;
use std::collections::BTreeMap;
use std::io::{Error, Read, Seek, SeekFrom, Write};
use std::net::{Shutdown, SocketAddr};
use std::process::ExitStatus;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use tokio;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use tokio_process;
pub type ResourceId = u32;
type ResourceTable = BTreeMap<ResourceId, Repr>;
#[cfg(not(windows))]
use std::os::unix::io::FromRawFd;
#[cfg(windows)]
use std::os::windows::io::FromRawHandle;
#[cfg(windows)]
extern crate winapi;
lazy_static! {
static ref NEXT_RID: AtomicUsize = AtomicUsize::new(3);
static ref RESOURCE_TABLE: Mutex<ResourceTable> = Mutex::new({
let mut m = BTreeMap::new();
m.insert(0, Repr::Stdin(tokio::io::stdin()));
m.insert(1, Repr::Stdout({
#[cfg(not(windows))]
let stdout = unsafe { std::fs::File::from_raw_fd(1) };
#[cfg(windows)]
let stdout = unsafe {
std::fs::File::from_raw_handle(winapi::um::processenv::GetStdHandle(
winapi::um::winbase::STD_OUTPUT_HANDLE))
};
tokio::fs::File::from_std(stdout)
}));
m.insert(2, Repr::Stderr(tokio::io::stderr()));
m
});
}
enum Repr {
Stdin(tokio::io::Stdin),
Stdout(tokio::fs::File),
Stderr(tokio::io::Stderr),
FsFile(tokio::fs::File),
TcpListener(tokio::net::TcpListener, Option<futures::task::Task>),
TcpStream(tokio::net::TcpStream),
HttpBody(HttpBody),
Repl(Arc<Mutex<Repl>>),
Child(Box<tokio_process::Child>),
ChildStdin(tokio_process::ChildStdin),
ChildStdout(tokio_process::ChildStdout),
ChildStderr(tokio_process::ChildStderr),
Worker(WorkerChannels),
}
pub fn get_type(rid: ResourceId) -> Option<String> {
let table = RESOURCE_TABLE.lock().unwrap();
table.get(&rid).map(inspect_repr)
}
pub fn table_entries() -> Vec<(u32, String)> {
let table = RESOURCE_TABLE.lock().unwrap();
table
.iter()
.map(|(key, value)| (*key, inspect_repr(&value)))
.collect()
}
#[test]
fn test_table_entries() {
let mut entries = table_entries();
entries.sort();
assert_eq!(entries[0], (0, String::from("stdin")));
assert_eq!(entries[1], (1, String::from("stdout")));
assert_eq!(entries[2], (2, String::from("stderr")));
}
fn inspect_repr(repr: &Repr) -> String {
let h_repr = match repr {
Repr::Stdin(_) => "stdin",
Repr::Stdout(_) => "stdout",
Repr::Stderr(_) => "stderr",
Repr::FsFile(_) => "fsFile",
Repr::TcpListener(_, _) => "tcpListener",
Repr::TcpStream(_) => "tcpStream",
Repr::HttpBody(_) => "httpBody",
Repr::Repl(_) => "repl",
Repr::Child(_) => "child",
Repr::ChildStdin(_) => "childStdin",
Repr::ChildStdout(_) => "childStdout",
Repr::ChildStderr(_) => "childStderr",
Repr::Worker(_) => "worker",
};
String::from(h_repr)
}
#[derive(Clone, Debug)]
pub struct Resource {
pub rid: ResourceId,
}
impl Resource {
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 => Err(std::io::Error::new(
std::io::ErrorKind::Other,
"Listener has been closed",
)),
Some(repr) => match repr {
Repr::TcpListener(ref mut s, _) => s.poll_accept(),
_ => panic!("Cannot accept"),
},
}
}
pub fn track_task(&mut self) -> Result<(), std::io::Error> {
let mut table = RESOURCE_TABLE.lock().unwrap();
if let Some(Repr::TcpListener(_, t)) = table.get_mut(&self.rid) {
if t.is_some() {
return Err(std::io::Error::new(
std::io::ErrorKind::Other,
"Another accept task is ongoing",
));
}
t.replace(futures::task::current());
}
Ok(())
}
pub fn untrack_task(&mut self) {
let mut table = RESOURCE_TABLE.lock().unwrap();
if let Some(Repr::TcpListener(_, t)) = table.get_mut(&self.rid) {
if t.is_some() {
t.take();
}
}
}
pub fn close(&self) {
let mut table = RESOURCE_TABLE.lock().unwrap();
let r = table.remove(&self.rid).unwrap();
if let Repr::TcpListener(_, Some(t)) = r {
t.notify();
}
}
pub fn shutdown(&mut self, how: Shutdown) -> Result<(), ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let repr = table.get_mut(&self.rid).ok_or_else(bad_resource)?;
match repr {
Repr::TcpStream(ref mut f) => {
TcpStream::shutdown(f, how).map_err(ErrBox::from)
}
_ => Err(bad_resource()),
}
}
}
impl Read for Resource {
fn read(&mut self, _buf: &mut [u8]) -> std::io::Result<usize> {
unimplemented!();
}
}
pub trait DenoAsyncRead {
fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, ErrBox>;
}
impl DenoAsyncRead for Resource {
fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let repr = table.get_mut(&self.rid).ok_or_else(bad_resource)?;
let r = 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::HttpBody(ref mut f) => f.poll_read(buf),
Repr::ChildStdout(ref mut f) => f.poll_read(buf),
Repr::ChildStderr(ref mut f) => f.poll_read(buf),
_ => {
return Err(bad_resource());
}
};
r.map_err(ErrBox::from)
}
}
impl Write for Resource {
fn write(&mut self, _buf: &[u8]) -> std::io::Result<usize> {
unimplemented!()
}
fn flush(&mut self) -> std::io::Result<()> {
unimplemented!()
}
}
pub trait DenoAsyncWrite {
fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, ErrBox>;
fn shutdown(&mut self) -> Poll<(), ErrBox>;
}
impl DenoAsyncWrite for Resource {
fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let repr = table.get_mut(&self.rid).ok_or_else(bad_resource)?;
let r = 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::TcpStream(ref mut f) => f.poll_write(buf),
Repr::ChildStdin(ref mut f) => f.poll_write(buf),
_ => {
return Err(bad_resource());
}
};
r.map_err(ErrBox::from)
}
fn shutdown(&mut self) -> futures::Poll<(), ErrBox> {
unimplemented!()
}
}
fn new_rid() -> ResourceId {
let next_rid = NEXT_RID.fetch_add(1, Ordering::SeqCst);
next_rid as ResourceId
}
pub fn add_fs_file(fs_file: tokio::fs::File) -> Resource {
let rid = new_rid();
let mut tg = RESOURCE_TABLE.lock().unwrap();
let r = tg.insert(rid, Repr::FsFile(fs_file));
assert!(r.is_none());
Resource { rid }
}
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, None));
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 add_reqwest_body(body: ReqwestDecoder) -> Resource {
let rid = new_rid();
let mut tg = RESOURCE_TABLE.lock().unwrap();
let body = HttpBody::from(body);
let r = tg.insert(rid, Repr::HttpBody(body));
assert!(r.is_none());
Resource { rid }
}
pub fn add_repl(repl: Repl) -> Resource {
let rid = new_rid();
let mut tg = RESOURCE_TABLE.lock().unwrap();
let r = tg.insert(rid, Repr::Repl(Arc::new(Mutex::new(repl))));
assert!(r.is_none());
Resource { rid }
}
pub fn add_worker(wc: WorkerChannels) -> Resource {
let rid = new_rid();
let mut tg = RESOURCE_TABLE.lock().unwrap();
let r = tg.insert(rid, Repr::Worker(wc));
assert!(r.is_none());
Resource { rid }
}
pub fn post_message_to_worker(
rid: ResourceId,
buf: Buf,
) -> futures::sink::Send<mpsc::Sender<Buf>> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let maybe_repr = table.get_mut(&rid);
match maybe_repr {
Some(Repr::Worker(ref mut wc)) => {
wc.0.clone().send(buf)
}
_ => panic!("bad resource"), }
}
pub struct WorkerReceiver {
rid: ResourceId,
}
impl Future for WorkerReceiver {
type Item = Option<Buf>;
type Error = ErrBox;
fn poll(&mut self) -> Poll<Option<Buf>, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let maybe_repr = table.get_mut(&self.rid);
match maybe_repr {
Some(Repr::Worker(ref mut wc)) => wc.1.poll().map_err(ErrBox::from),
_ => Err(bad_resource()),
}
}
}
pub fn get_message_from_worker(rid: ResourceId) -> WorkerReceiver {
WorkerReceiver { rid }
}
pub struct WorkerReceiverStream {
rid: ResourceId,
}
impl Stream for WorkerReceiverStream {
type Item = Buf;
type Error = ErrBox;
fn poll(&mut self) -> Poll<Option<Buf>, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let maybe_repr = table.get_mut(&self.rid);
match maybe_repr {
Some(Repr::Worker(ref mut wc)) => wc.1.poll().map_err(ErrBox::from),
_ => Err(bad_resource()),
}
}
}
pub fn get_message_stream_from_worker(rid: ResourceId) -> WorkerReceiverStream {
WorkerReceiverStream { rid }
}
pub struct ChildResources {
pub child_rid: ResourceId,
pub stdin_rid: Option<ResourceId>,
pub stdout_rid: Option<ResourceId>,
pub stderr_rid: Option<ResourceId>,
}
pub fn add_child(mut c: tokio_process::Child) -> ChildResources {
let child_rid = new_rid();
let mut tg = RESOURCE_TABLE.lock().unwrap();
let mut resources = ChildResources {
child_rid,
stdin_rid: None,
stdout_rid: None,
stderr_rid: None,
};
if c.stdin().is_some() {
let stdin = c.stdin().take().unwrap();
let rid = new_rid();
let r = tg.insert(rid, Repr::ChildStdin(stdin));
assert!(r.is_none());
resources.stdin_rid = Some(rid);
}
if c.stdout().is_some() {
let stdout = c.stdout().take().unwrap();
let rid = new_rid();
let r = tg.insert(rid, Repr::ChildStdout(stdout));
assert!(r.is_none());
resources.stdout_rid = Some(rid);
}
if c.stderr().is_some() {
let stderr = c.stderr().take().unwrap();
let rid = new_rid();
let r = tg.insert(rid, Repr::ChildStderr(stderr));
assert!(r.is_none());
resources.stderr_rid = Some(rid);
}
let r = tg.insert(child_rid, Repr::Child(Box::new(c)));
assert!(r.is_none());
resources
}
pub struct ChildStatus {
rid: ResourceId,
}
impl Future for ChildStatus {
type Item = ExitStatus;
type Error = ErrBox;
fn poll(&mut self) -> Poll<ExitStatus, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let maybe_repr = table.get_mut(&self.rid);
match maybe_repr {
Some(Repr::Child(ref mut child)) => child.poll().map_err(ErrBox::from),
_ => Err(bad_resource()),
}
}
}
pub fn child_status(rid: ResourceId) -> Result<ChildStatus, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let maybe_repr = table.get_mut(&rid);
match maybe_repr {
Some(Repr::Child(ref mut _child)) => Ok(ChildStatus { rid }),
_ => Err(bad_resource()),
}
}
pub fn get_repl(rid: ResourceId) -> Result<Arc<Mutex<Repl>>, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let maybe_repr = table.get_mut(&rid);
match maybe_repr {
Some(Repr::Repl(ref mut r)) => Ok(r.clone()),
_ => Err(bad_resource()),
}
}
pub fn get_file(rid: ResourceId) -> Result<std::fs::File, ErrBox> {
let mut table = RESOURCE_TABLE.lock().unwrap();
let maybe_repr = table.remove(&rid);
match maybe_repr {
Some(Repr::FsFile(r)) => {
let std_file = r.into_std();
let maybe_std_file_copy = std_file.try_clone();
table.insert(rid, Repr::FsFile(tokio_fs::File::from_std(std_file)));
if maybe_std_file_copy.is_err() {
return Err(ErrBox::from(maybe_std_file_copy.unwrap_err()));
}
let std_file_copy = maybe_std_file_copy.unwrap();
Ok(std_file_copy)
}
_ => Err(bad_resource()),
}
}
pub fn lookup(rid: ResourceId) -> Result<Resource, ErrBox> {
debug!("resource lookup {}", rid);
let table = RESOURCE_TABLE.lock().unwrap();
table
.get(&rid)
.ok_or_else(bad_resource)
.map(|_| Resource { rid })
}
pub fn seek(
resource: Resource,
offset: i32,
whence: u32,
) -> Box<dyn Future<Item = (), Error = ErrBox> + Send> {
let seek_from = match whence {
0 => SeekFrom::Start(offset as u64),
1 => SeekFrom::Current(i64::from(offset)),
2 => SeekFrom::End(i64::from(offset)),
_ => {
return Box::new(futures::future::err(
deno_error::DenoError::new(
deno_error::ErrorKind::InvalidSeekMode,
format!("Invalid seek mode: {}", whence),
)
.into(),
));
}
};
match get_file(resource.rid) {
Ok(mut file) => Box::new(futures::future::lazy(move || {
let result = file.seek(seek_from).map(|_| {}).map_err(ErrBox::from);
futures::future::result(result)
})),
Err(err) => Box::new(futures::future::err(err)),
}
}