denoland-deno/cli/ops/net.rs

// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license.
use super::dispatch_json::{Deserialize, JsonOp, Value};
use crate::resolve_addr::resolve_addr;
use crate::resources;
use crate::resources::Resource;
use crate::state::ThreadSafeState;
use crate::tokio_util;
use deno::*;
use futures::Future;
use std;
use std::convert::From;
use std::net::Shutdown;
use tokio;
use tokio::net::TcpListener;
use tokio::net::TcpStream;

#[derive(Deserialize)]
struct AcceptArgs {
  rid: i32,
}

pub fn op_accept(
  _state: &ThreadSafeState,
  args: Value,
  _zero_copy: Option<PinnedBuf>,
) -> Result<JsonOp, ErrBox> {
  let args: AcceptArgs = serde_json::from_value(args)?;
  let server_rid = args.rid as u32;

  let server_resource = resources::lookup(server_rid)?;
  let op = tokio_util::accept(server_resource)
    .and_then(move |(tcp_stream, _socket_addr)| {
      let local_addr = tcp_stream.local_addr()?;
      let remote_addr = tcp_stream.peer_addr()?;
      let tcp_stream_resource = resources::add_tcp_stream(tcp_stream);
      Ok((tcp_stream_resource, local_addr, remote_addr))
    })
    .map_err(ErrBox::from)
    .and_then(move |(tcp_stream_resource, local_addr, remote_addr)| {
      futures::future::ok(json!({
        "rid": tcp_stream_resource.rid,
        "localAddr": local_addr.to_string(),
        "remoteAddr": remote_addr.to_string(),
      }))
    });

  Ok(JsonOp::Async(Box::new(op)))
}

#[derive(Deserialize)]
struct DialArgs {
  network: String,
  address: String,
}

pub fn op_dial(
  state: &ThreadSafeState,
  args: Value,
  _zero_copy: Option<PinnedBuf>,
) -> Result<JsonOp, ErrBox> {
  let args: DialArgs = serde_json::from_value(args)?;
  let network = args.network;
  assert_eq!(network, "tcp"); // TODO Support others.
  let address = args.address;

  state.check_net(&address)?;

  let op = resolve_addr(&address).and_then(move |addr| {
    TcpStream::connect(&addr)
      .map_err(ErrBox::from)
      .and_then(move |tcp_stream| {
        let local_addr = tcp_stream.local_addr()?;
        let remote_addr = tcp_stream.peer_addr()?;
        let tcp_stream_resource = resources::add_tcp_stream(tcp_stream);
        Ok((tcp_stream_resource, local_addr, remote_addr))
      })
      .map_err(ErrBox::from)
      .and_then(move |(tcp_stream_resource, local_addr, remote_addr)| {
        futures::future::ok(json!({
          "rid": tcp_stream_resource.rid,
          "localAddr": local_addr.to_string(),
          "remoteAddr": remote_addr.to_string(),
        }))
      })
  });

  Ok(JsonOp::Async(Box::new(op)))
}

#[derive(Deserialize)]
struct ShutdownArgs {
  rid: i32,
  how: i32,
}

pub fn op_shutdown(
  _state: &ThreadSafeState,
  args: Value,
  _zero_copy: Option<PinnedBuf>,
) -> Result<JsonOp, ErrBox> {
  let args: ShutdownArgs = serde_json::from_value(args)?;

  let rid = args.rid as u32;
  let how = args.how;
  let mut resource = resources::lookup(rid)?;

  let shutdown_mode = match how {
    0 => Shutdown::Read,
    1 => Shutdown::Write,
    _ => unimplemented!(),
  };

  // Use UFCS for disambiguation
  Resource::shutdown(&mut resource, shutdown_mode)?;
  Ok(JsonOp::Sync(json!({})))
}

#[derive(Deserialize)]
struct ListenArgs {
  network: String,
  address: String,
}

pub fn op_listen(
  state: &ThreadSafeState,
  args: Value,
  _zero_copy: Option<PinnedBuf>,
) -> Result<JsonOp, ErrBox> {
  let args: ListenArgs = serde_json::from_value(args)?;

  let network = args.network;
  assert_eq!(network, "tcp");
  let address = args.address;

  state.check_net(&address)?;

  let addr = resolve_addr(&address).wait()?;
  let listener = TcpListener::bind(&addr)?;
  let local_addr = listener.local_addr()?;
  let resource = resources::add_tcp_listener(listener);

  Ok(JsonOp::Sync(json!({
    "rid": resource.rid,
    "localAddr": local_addr.to_string()
  })))
}