Revert "Revert "refactor(ext/net): clean up variadic network ops (#16… (#16422)

…392)" (#16417)"

This reverts commit 8e3f825c921b38141afa7a69a0664881c5c94461.

8 files changed, 454 insertions, 716 deletions

diff --git a/ext/net/01_net.js b/ext/net/01_net.js
index d7a093ba6..acd8ee179 100644
--- a/ext/net/01_net.js
+++ b/ext/net/01_net.js
@@ -10,10 +10,10 @@
     Error,
     ObjectPrototypeIsPrototypeOf,
     PromiseResolve,
-    Symbol,
     SymbolAsyncIterator,
     SymbolFor,
     TypedArrayPrototypeSubarray,
+    TypeError,
     Uint8Array,
   } = window.__bootstrap.primordials;
 
@@ -38,30 +38,6 @@
     return core.shutdown(rid);
   }
 
-  function opAccept(rid, transport) {
-    return core.opAsync("op_net_accept", { rid, transport });
-  }
-
-  function opListen(args) {
-    return ops.op_net_listen(args);
-  }
-
-  function opConnect(args) {
-    return core.opAsync("op_net_connect", args);
-  }
-
-  function opReceive(rid, transport, zeroCopy) {
-    return core.opAsync(
-      "op_dgram_recv",
-      { rid, transport },
-      zeroCopy,
-    );
-  }
-
-  function opSend(args, zeroCopy) {
-    return core.opAsync("op_dgram_send", args, zeroCopy);
-  }
-
   function resolveDns(query, recordType, options) {
     return core.opAsync("op_dns_resolve", { query, recordType, options });
   }
@@ -135,11 +111,6 @@
 
   class UnixConn extends Conn {}
 
-  // Use symbols for method names to hide these in stable API.
-  // TODO(kt3k): Remove these symbols when ref/unref become stable.
-  const listenerRef = Symbol("listenerRef");
-  const listenerUnref = Symbol("listenerUnref");
-
   class Listener {
     #rid = 0;
     #addr = null;
@@ -159,21 +130,35 @@
       return this.#addr;
     }
 
-    accept() {
-      const promise = opAccept(this.rid, this.addr.transport);
+    async accept() {
+      let promise;
+      switch (this.addr.transport) {
+        case "tcp":
+          promise = core.opAsync("op_net_accept_tcp", this.rid);
+          break;
+        case "unix":
+          promise = core.opAsync("op_net_accept_unix", this.rid);
+          break;
+        default:
+          throw new Error(`Unsupported transport: ${this.addr.transport}`);
+      }
       this.#promiseId = promise[promiseIdSymbol];
-      if (this.#unref) {
-        this.#unrefOpAccept();
+      if (this.#unref) core.unrefOp(this.#promiseId);
+      const [rid, localAddr, remoteAddr] = await promise;
+      this.#promiseId = null;
+      if (this.addr.transport == "tcp") {
+        localAddr.transport = "tcp";
+        remoteAddr.transport = "tcp";
+        return new TcpConn(rid, remoteAddr, localAddr);
+      } else if (this.addr.transport == "unix") {
+        return new UnixConn(
+          rid,
+          { transport: "unix", path: remoteAddr },
+          { transport: "unix", path: localAddr },
+        );
+      } else {
+        throw new Error("unreachable");
       }
-      return promise.then((res) => {
-        if (this.addr.transport == "tcp") {
-          return new TcpConn(res.rid, res.remoteAddr, res.localAddr);
-        } else if (this.addr.transport == "unix") {
-          return new UnixConn(res.rid, res.remoteAddr, res.localAddr);
-        } else {
-          throw new Error("unreachable");
-        }
-      });
     }
 
     async next() {
@@ -205,22 +190,15 @@
       return this;
     }
 
-    [listenerRef]() {
+    ref() {
       this.#unref = false;
-      this.#refOpAccept();
-    }
-
-    [listenerUnref]() {
-      this.#unref = true;
-      this.#unrefOpAccept();
-    }
-
-    #refOpAccept() {
       if (typeof this.#promiseId === "number") {
         core.refOp(this.#promiseId);
       }
     }
-    #unrefOpAccept() {
+
+    unref() {
+      this.#unref = true;
       if (typeof this.#promiseId === "number") {
         core.unrefOp(this.#promiseId);
       }
@@ -247,18 +225,54 @@
 
     async receive(p) {
       const buf = p || new Uint8Array(this.bufSize);
-      const { size, remoteAddr } = await opReceive(
-        this.rid,
-        this.addr.transport,
-        buf,
-      );
-      const sub = TypedArrayPrototypeSubarray(buf, 0, size);
+      let nread;
+      let remoteAddr;
+      switch (this.addr.transport) {
+        case "udp": {
+          [nread, remoteAddr] = await core.opAsync(
+            "op_net_recv_udp",
+            this.rid,
+            buf,
+          );
+          remoteAddr.transport = "udp";
+          break;
+        }
+        case "unixpacket": {
+          let path;
+          [nread, path] = await core.opAsync(
+            "op_net_recv_unixpacket",
+            this.rid,
+            buf,
+          );
+          remoteAddr = { transport: "unixpacket", path };
+          break;
+        }
+        default:
+          throw new Error(`Unsupported transport: ${this.addr.transport}`);
+      }
+      const sub = TypedArrayPrototypeSubarray(buf, 0, nread);
       return [sub, remoteAddr];
     }
 
-    send(p, addr) {
-      const args = { hostname: "127.0.0.1", ...addr, rid: this.rid };
-      return opSend(args, p);
+    async send(p, opts) {
+      switch (this.addr.transport) {
+        case "udp":
+          return await core.opAsync(
+            "op_net_send_udp",
+            this.rid,
+            { hostname: opts.hostname ?? "127.0.0.1", port: opts.port },
+            p,
+          );
+        case "unixpacket":
+          return await core.opAsync(
+            "op_net_send_unixpacket",
+            this.rid,
+            opts.path,
+            p,
+          );
+        default:
+          throw new Error(`Unsupported transport: ${this.addr.transport}`);
+      }
     }
 
     close() {
@@ -282,40 +296,100 @@
     }
   }
 
-  function listen({ hostname, ...options }, constructor = Listener) {
-    const res = opListen({
-      transport: "tcp",
-      hostname: typeof hostname === "undefined" ? "0.0.0.0" : hostname,
-      ...options,
-    });
+  function listen(args) {
+    switch (args.transport ?? "tcp") {
+      case "tcp": {
+        const [rid, addr] = ops.op_net_listen_tcp({
+          hostname: args.hostname ?? "0.0.0.0",
+          port: args.port,
+        });
+        addr.transport = "tcp";
+        return new Listener(rid, addr);
+      }
+      case "unix": {
+        const [rid, path] = ops.op_net_listen_unix(args.path);
+        const addr = {
+          transport: "unix",
+          path,
+        };
+        return new Listener(rid, addr);
+      }
+      default:
+        throw new TypeError(`Unsupported transport: '${transport}'`);
+    }
+  }
 
-    return new constructor(res.rid, res.localAddr);
+  function listenDatagram(args) {
+    switch (args.transport) {
+      case "udp": {
+        const [rid, addr] = ops.op_net_listen_udp(
+          {
+            hostname: args.hostname ?? "127.0.0.1",
+            port: args.port,
+          },
+          args.reuseAddress ?? false,
+        );
+        addr.transport = "udp";
+        return new Datagram(rid, addr);
+      }
+      case "unixpacket": {
+        const [rid, path] = ops.op_net_listen_unixpacket(args.path);
+        const addr = {
+          transport: "unixpacket",
+          path,
+        };
+        return new Datagram(rid, addr);
+      }
+      default:
+        throw new TypeError(`Unsupported transport: '${transport}'`);
+    }
   }
 
-  async function connect(options) {
-    if (options.transport === "unix") {
-      const res = await opConnect(options);
-      return new UnixConn(res.rid, res.remoteAddr, res.localAddr);
+  async function connect(args) {
+    switch (args.transport ?? "tcp") {
+      case "tcp": {
+        const [rid, localAddr, remoteAddr] = await core.opAsync(
+          "op_net_connect_tcp",
+          {
+            hostname: args.hostname ?? "127.0.0.1",
+            port: args.port,
+          },
+        );
+        localAddr.transport = "tcp";
+        remoteAddr.transport = "tcp";
+        return new TcpConn(rid, remoteAddr, localAddr);
+      }
+      case "unix": {
+        const [rid, localAddr, remoteAddr] = await core.opAsync(
+          "op_net_connect_unix",
+          args.path,
+        );
+        return new UnixConn(
+          rid,
+          { transport: "unix", path: remoteAddr },
+          { transport: "unix", path: localAddr },
+        );
+      }
+      default:
+        throw new TypeError(`Unsupported transport: '${transport}'`);
     }
+  }
 
-    const res = await opConnect({
-      transport: "tcp",
-      hostname: "127.0.0.1",
-      ...options,
-    });
-    return new TcpConn(res.rid, res.remoteAddr, res.localAddr);
+  function setup(unstable) {
+    if (!unstable) {
+      delete Listener.prototype.ref;
+      delete Listener.prototype.unref;
+    }
   }
 
   window.__bootstrap.net = {
+    setup,
     connect,
     Conn,
     TcpConn,
     UnixConn,
-    opConnect,
     listen,
-    listenerRef,
-    listenerUnref,
-    opListen,
+    listenDatagram,
     Listener,
     shutdown,
     Datagram,
diff --git a/ext/net/02_tls.js b/ext/net/02_tls.js
index 04b25caf9..d3f906dbd 100644
--- a/ext/net/02_tls.js
+++ b/ext/net/02_tls.js
@@ -5,20 +5,7 @@
   const core = window.Deno.core;
   const ops = core.ops;
   const { Listener, Conn } = window.__bootstrap.net;
-
-  function opConnectTls(
-    args,
-  ) {
-    return core.opAsync("op_tls_connect", args);
-  }
-
-  function opAcceptTLS(rid) {
-    return core.opAsync("op_tls_accept", rid);
-  }
-
-  function opListenTls(args) {
-    return ops.op_tls_listen(args);
-  }
+  const { TypeError } = window.__bootstrap.primordials;
 
   function opStartTls(args) {
     return core.opAsync("op_tls_start", args);
@@ -44,23 +31,28 @@
     privateKey = undefined,
     alpnProtocols = undefined,
   }) {
-    const res = await opConnectTls({
-      port,
-      hostname,
-      transport,
-      certFile,
-      caCerts,
-      certChain,
-      privateKey,
-      alpnProtocols,
-    });
-    return new TlsConn(res.rid, res.remoteAddr, res.localAddr);
+    if (transport !== "tcp") {
+      throw new TypeError(`Unsupported transport: '${transport}'`);
+    }
+    const [rid, localAddr, remoteAddr] = await core.opAsync(
+      "op_net_connect_tls",
+      { hostname, port },
+      { certFile, caCerts, certChain, privateKey, alpnProtocols },
+    );
+    localAddr.transport = "tcp";
+    remoteAddr.transport = "tcp";
+    return new TlsConn(rid, remoteAddr, localAddr);
   }
 
   class TlsListener extends Listener {
     async accept() {
-      const res = await opAcceptTLS(this.rid);
-      return new TlsConn(res.rid, res.remoteAddr, res.localAddr);
+      const [rid, localAddr, remoteAddr] = await core.opAsync(
+        "op_net_accept_tls",
+        this.rid,
+      );
+      localAddr.transport = "tcp";
+      remoteAddr.transport = "tcp";
+      return new TlsConn(rid, remoteAddr, localAddr);
     }
   }
 
@@ -74,17 +66,14 @@
     transport = "tcp",
     alpnProtocols = undefined,
   }) {
-    const res = opListenTls({
-      port,
-      cert,
-      certFile,
-      key,
-      keyFile,
-      hostname,
-      transport,
-      alpnProtocols,
-    });
-    return new TlsListener(res.rid, res.localAddr);
+    if (transport !== "tcp") {
+      throw new TypeError(`Unsupported transport: '${transport}'`);
+    }
+    const [rid, localAddr] = ops.op_net_listen_tls(
+      { hostname, port },
+      { cert, certFile, key, keyFile, alpnProtocols },
+    );
+    return new TlsListener(rid, localAddr);
   }
 
   async function startTls(
@@ -96,14 +85,14 @@
       alpnProtocols = undefined,
     } = {},
   ) {
-    const res = await opStartTls({
+    const [rid, localAddr, remoteAddr] = await opStartTls({
       rid: conn.rid,
       hostname,
       certFile,
       caCerts,
       alpnProtocols,
     });
-    return new TlsConn(res.rid, res.remoteAddr, res.localAddr);
+    return new TlsConn(rid, remoteAddr, localAddr);
   }
 
   window.__bootstrap.tls = {
diff --git a/ext/net/04_net_unstable.js b/ext/net/04_net_unstable.js
deleted file mode 100644
index fcdb3c547..000000000
--- a/ext/net/04_net_unstable.js
+++ /dev/null
@@ -1,60 +0,0 @@
-// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
-"use strict";
-
-((window) => {
-  const net = window.__bootstrap.net;
-
-  function listen(options) {
-    if (options.transport === "unix") {
-      const res = net.opListen(options);
-      return new Listener(res.rid, res.localAddr);
-    } else {
-      return net.listen(options, Listener);
-    }
-  }
-
-  function listenDatagram(
-    options,
-  ) {
-    let res;
-    if (options.transport === "unixpacket") {
-      res = net.opListen(options);
-    } else {
-      res = net.opListen({
-        transport: "udp",
-        hostname: "127.0.0.1",
-        ...options,
-      });
-    }
-
-    return new net.Datagram(res.rid, res.localAddr);
-  }
-
-  async function connect(
-    options,
-  ) {
-    if (options.transport === "unix") {
-      const res = await net.opConnect(options);
-      return new net.Conn(res.rid, res.remoteAddr, res.localAddr);
-    } else {
-      return net.connect(options);
-    }
-  }
-
-  class Listener extends net.Listener {
-    ref() {
-      this[net.listenerRef]();
-    }
-
-    unref() {
-      this[net.listenerUnref]();
-    }
-  }
-
-  window.__bootstrap.netUnstable = {
-    connect,
-    listenDatagram,
-    listen,
-    Listener,
-  };
-})(this);
diff --git a/ext/net/README.md b/ext/net/README.md
index 1928fc375..1f9301a2d 100644
--- a/ext/net/README.md
+++ b/ext/net/README.md
@@ -9,14 +9,21 @@ This crate depends on following extensions:
 
 Following ops are provided:
 
-- "op_net_accept"
-- "op_net_connect"
-- "op_net_listen"
-- "op_dgram_recv"
-- "op_dgram_send"
+- "op_net_accept_tcp"
+- "op_net_accept_unix"
+- "op_net_connect_tcp"
+- "op_net_connect_unix"
+- "op_net_listen_tcp"
+- "op_net_listen_udp"
+- "op_net_listen_unix"
+- "op_net_listen_unixpacket"
+- "op_net_recv_udp"
+- "op_net_recv_unixpacket"
+- "op_net_send_udp"
+- "op_net_send_unixpacket"
 - "op_dns_resolve"
+- "op_net_connect_tls"
+- "op_net_listen_tls"
+- "op_net_accept_tls"
 - "op_tls_start"
-- "op_tls_connect"
-- "op_tls_listen"
-- "op_tls_accept"
 - "op_tls_handshake"
diff --git a/ext/net/lib.rs b/ext/net/lib.rs
index 35d612598..ddeeeb1a8 100644
--- a/ext/net/lib.rs
+++ b/ext/net/lib.rs
@@ -90,7 +90,6 @@ pub fn init<P: NetPermissions + 'static>(
       prefix "deno:ext/net",
       "01_net.js",
       "02_tls.js",
-      "04_net_unstable.js",
     ))
     .ops(ops)
     .state(move |state| {
diff --git a/ext/net/ops.rs b/ext/net/ops.rs
index 399baa4fd..9a6d95586 100644
--- a/ext/net/ops.rs
+++ b/ext/net/ops.rs
@@ -7,7 +7,6 @@ use crate::NetPermissions;
 use deno_core::error::bad_resource;
 use deno_core::error::custom_error;
 use deno_core::error::generic_error;
-use deno_core::error::type_error;
 use deno_core::error::AnyError;
 use deno_core::op;
 
@@ -21,7 +20,6 @@ use deno_core::RcRef;
 use deno_core::Resource;
 use deno_core::ResourceId;
 use deno_core::ZeroCopyBuf;
-use log::debug;
 use serde::Deserialize;
 use serde::Serialize;
 use socket2::Domain;
@@ -45,69 +43,51 @@ use trust_dns_resolver::error::ResolveErrorKind;
 use trust_dns_resolver::system_conf;
 use trust_dns_resolver::AsyncResolver;
 
-#[cfg(unix)]
-use super::ops_unix as net_unix;
-#[cfg(unix)]
-use crate::io::UnixStreamResource;
-#[cfg(unix)]
-use std::path::Path;
-
 pub fn init<P: NetPermissions + 'static>() -> Vec<OpDecl> {
   vec![
-    op_net_accept::decl(),
-    op_net_connect::decl::<P>(),
-    op_net_listen::decl::<P>(),
-    op_dgram_recv::decl(),
-    op_dgram_send::decl::<P>(),
+    op_net_accept_tcp::decl(),
+    #[cfg(unix)]
+    crate::ops_unix::op_net_accept_unix::decl(),
+    op_net_connect_tcp::decl::<P>(),
+    #[cfg(unix)]
+    crate::ops_unix::op_net_connect_unix::decl::<P>(),
+    op_net_listen_tcp::decl::<P>(),
+    op_net_listen_udp::decl::<P>(),
+    #[cfg(unix)]
+    crate::ops_unix::op_net_listen_unix::decl::<P>(),
+    #[cfg(unix)]
+    crate::ops_unix::op_net_listen_unixpacket::decl::<P>(),
+    op_net_recv_udp::decl(),
+    #[cfg(unix)]
+    crate::ops_unix::op_net_recv_unixpacket::decl(),
+    op_net_send_udp::decl::<P>(),
+    #[cfg(unix)]
+    crate::ops_unix::op_net_send_unixpacket::decl::<P>(),
     op_dns_resolve::decl::<P>(),
     op_set_nodelay::decl::<P>(),
     op_set_keepalive::decl::<P>(),
   ]
 }
 
-#[derive(Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct OpConn {
-  pub rid: ResourceId,
-  pub remote_addr: Option<OpAddr>,
-  pub local_addr: Option<OpAddr>,
-}
-
-#[derive(Serialize)]
-#[serde(tag = "transport", rename_all = "lowercase")]
-pub enum OpAddr {
-  Tcp(IpAddr),
-  Udp(IpAddr),
-  #[cfg(unix)]
-  Unix(net_unix::UnixAddr),
-  #[cfg(unix)]
-  UnixPacket(net_unix::UnixAddr),
-}
-
-#[derive(Serialize)]
-#[serde(rename_all = "camelCase")]
-/// A received datagram packet (from udp or unixpacket)
-pub struct OpPacket {
-  pub size: usize,
-  pub remote_addr: OpAddr,
-}
-
 #[derive(Serialize, Clone, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct TlsHandshakeInfo {
   pub alpn_protocol: Option<ByteString>,
 }
 
-#[derive(Serialize)]
+#[derive(Deserialize, Serialize)]
 pub struct IpAddr {
   pub hostname: String,
   pub port: u16,
 }
 
-#[derive(Deserialize)]
-pub(crate) struct AcceptArgs {
-  pub rid: ResourceId,
-  pub transport: String,
+impl From<SocketAddr> for IpAddr {
+  fn from(addr: SocketAddr) -> Self {
+    Self {
+      hostname: addr.ip().to_string(),
+      port: addr.port(),
+    }
+  }
 }
 
 pub(crate) fn accept_err(e: std::io::Error) -> AnyError {
@@ -119,13 +99,11 @@ pub(crate) fn accept_err(e: std::io::Error) -> AnyError {
   }
 }
 
-async fn accept_tcp(
+#[op]
+async fn op_net_accept_tcp(
   state: Rc<RefCell<OpState>>,
-  args: AcceptArgs,
-  _: (),
-) -> Result<OpConn, AnyError> {
-  let rid = args.rid;
-
+  rid: ResourceId,
+) -> Result<(ResourceId, IpAddr, IpAddr), AnyError> {
   let resource = state
     .borrow()
     .resource_table
@@ -147,51 +125,15 @@ async fn accept_tcp(
   let rid = state
     .resource_table
     .add(TcpStreamResource::new(tcp_stream.into_split()));
-  Ok(OpConn {
-    rid,
-    local_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: local_addr.ip().to_string(),
-      port: local_addr.port(),
-    })),
-    remote_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: remote_addr.ip().to_string(),
-      port: remote_addr.port(),
-    })),
-  })
+  Ok((rid, IpAddr::from(local_addr), IpAddr::from(remote_addr)))
 }
 
 #[op]
-async fn op_net_accept(
+async fn op_net_recv_udp(
   state: Rc<RefCell<OpState>>,
-  args: AcceptArgs,
-) -> Result<OpConn, AnyError> {
-  match args.transport.as_str() {
-    "tcp" => accept_tcp(state, args, ()).await,
-    #[cfg(unix)]
-    "unix" => net_unix::accept_unix(state, args, ()).await,
-    other => Err(bad_transport(other)),
-  }
-}
-
-fn bad_transport(transport: &str) -> AnyError {
-  generic_error(format!("Unsupported transport protocol {}", transport))
-}
-
-#[derive(Deserialize)]
-pub(crate) struct ReceiveArgs {
-  pub rid: ResourceId,
-  pub transport: String,
-}
-
-async fn receive_udp(
-  state: Rc<RefCell<OpState>>,
-  args: ReceiveArgs,
-  zero_copy: ZeroCopyBuf,
-) -> Result<OpPacket, AnyError> {
-  let mut zero_copy = zero_copy.clone();
-
-  let rid = args.rid;
-
+  rid: ResourceId,
+  mut buf: ZeroCopyBuf,
+) -> Result<(usize, IpAddr), AnyError> {
   let resource = state
     .borrow_mut()
     .resource_table
@@ -199,192 +141,75 @@ async fn receive_udp(
     .map_err(|_| bad_resource("Socket has been closed"))?;
   let socket = RcRef::map(&resource, |r| &r.socket).borrow().await;
   let cancel_handle = RcRef::map(&resource, |r| &r.cancel);
-  let (size, remote_addr) = socket
-    .recv_from(&mut zero_copy)
+  let (nread, remote_addr) = socket
+    .recv_from(&mut buf)
     .try_or_cancel(cancel_handle)
     .await?;
-  Ok(OpPacket {
-    size,
-    remote_addr: OpAddr::Udp(IpAddr {
-      hostname: remote_addr.ip().to_string(),
-      port: remote_addr.port(),
-    }),
-  })
+  Ok((nread, IpAddr::from(remote_addr)))
 }
 
 #[op]
-async fn op_dgram_recv(
+async fn op_net_send_udp<NP>(
   state: Rc<RefCell<OpState>>,
-  args: ReceiveArgs,
-  zero_copy: ZeroCopyBuf,
-) -> Result<OpPacket, AnyError> {
-  match args.transport.as_str() {
-    "udp" => receive_udp(state, args, zero_copy).await,
-    #[cfg(unix)]
-    "unixpacket" => net_unix::receive_unix_packet(state, args, zero_copy).await,
-    other => Err(bad_transport(other)),
-  }
-}
-
-#[derive(Deserialize)]
-struct SendArgs {
   rid: ResourceId,
-  transport: String,
-  #[serde(flatten)]
-  transport_args: ArgsEnum,
-}
-
-#[op]
-async fn op_dgram_send<NP>(
-  state: Rc<RefCell<OpState>>,
-  args: SendArgs,
+  addr: IpAddr,
   zero_copy: ZeroCopyBuf,
 ) -> Result<usize, AnyError>
 where
   NP: NetPermissions + 'static,
 {
-  let zero_copy = zero_copy.clone();
-
-  match args {
-    SendArgs {
-      rid,
-      transport,
-      transport_args: ArgsEnum::Ip(args),
-    } if transport == "udp" => {
-      {
-        let mut s = state.borrow_mut();
-        s.borrow_mut::<NP>().check_net(
-          &(&args.hostname, Some(args.port)),
-          "Deno.DatagramConn.send()",
-        )?;
-      }
-      let addr = resolve_addr(&args.hostname, args.port)
-        .await?
-        .next()
-        .ok_or_else(|| generic_error("No resolved address found"))?;
-
-      let resource = state
-        .borrow_mut()
-        .resource_table
-        .get::<UdpSocketResource>(rid)
-        .map_err(|_| bad_resource("Socket has been closed"))?;
-      let socket = RcRef::map(&resource, |r| &r.socket).borrow().await;
-      let byte_length = socket.send_to(&zero_copy, &addr).await?;
-      Ok(byte_length)
-    }
-    #[cfg(unix)]
-    SendArgs {
-      rid,
-      transport,
-      transport_args: ArgsEnum::Unix(args),
-    } if transport == "unixpacket" => {
-      let address_path = Path::new(&args.path);
-      {
-        let mut s = state.borrow_mut();
-        s.borrow_mut::<NP>()
-          .check_write(address_path, "Deno.DatagramConn.send()")?;
-      }
-      let resource = state
-        .borrow()
-        .resource_table
-        .get::<net_unix::UnixDatagramResource>(rid)
-        .map_err(|_| custom_error("NotConnected", "Socket has been closed"))?;
-      let socket = RcRef::map(&resource, |r| &r.socket)
-        .try_borrow_mut()
-        .ok_or_else(|| custom_error("Busy", "Socket already in use"))?;
-      let byte_length = socket.send_to(&zero_copy, address_path).await?;
-      Ok(byte_length)
-    }
-    _ => Err(type_error("Wrong argument format!")),
+  {
+    let mut s = state.borrow_mut();
+    s.borrow_mut::<NP>().check_net(
+      &(&addr.hostname, Some(addr.port)),
+      "Deno.DatagramConn.send()",
+    )?;
   }
-}
+  let addr = resolve_addr(&addr.hostname, addr.port)
+    .await?
+    .next()
+    .ok_or_else(|| generic_error("No resolved address found"))?;
 
-#[derive(Deserialize)]
-pub struct ConnectArgs {
-  transport: String,
-  #[serde(flatten)]
-  transport_args: ArgsEnum,
+  let resource = state
+    .borrow_mut()
+    .resource_table
+    .get::<UdpSocketResource>(rid)
+    .map_err(|_| bad_resource("Socket has been closed"))?;
+  let socket = RcRef::map(&resource, |r| &r.socket).borrow().await;
+  let nwritten = socket.send_to(&zero_copy, &addr).await?;
+
+  Ok(nwritten)
 }
 
 #[op]
-pub async fn op_net_connect<NP>(
+pub async fn op_net_connect_tcp<NP>(
   state: Rc<RefCell<OpState>>,
-  args: ConnectArgs,
-) -> Result<OpConn, AnyError>
+  addr: IpAddr,
+) -> Result<(ResourceId, IpAddr, IpAddr), AnyError>
 where
   NP: NetPermissions + 'static,
 {
-  match args {
-    ConnectArgs {
-      transport,
-      transport_args: ArgsEnum::Ip(args),
-    } if transport == "tcp" => {
-      {
-        let mut state_ = state.borrow_mut();
-        state_
-          .borrow_mut::<NP>()
-          .check_net(&(&args.hostname, Some(args.port)), "Deno.connect()")?;
-      }
-      let addr = resolve_addr(&args.hostname, args.port)
-        .await?
-        .next()
-        .ok_or_else(|| generic_error("No resolved address found"))?;
-      let tcp_stream = TcpStream::connect(&addr).await?;
-      let local_addr = tcp_stream.local_addr()?;
-      let remote_addr = tcp_stream.peer_addr()?;
-
-      let mut state_ = state.borrow_mut();
-      let rid = state_
-        .resource_table
-        .add(TcpStreamResource::new(tcp_stream.into_split()));
-      Ok(OpConn {
-        rid,
-        local_addr: Some(OpAddr::Tcp(IpAddr {
-          hostname: local_addr.ip().to_string(),
-          port: local_addr.port(),
-        })),
-        remote_addr: Some(OpAddr::Tcp(IpAddr {
-          hostname: remote_addr.ip().to_string(),
-          port: remote_addr.port(),
-        })),
-      })
-    }
-    #[cfg(unix)]
-    ConnectArgs {
-      transport,
-      transport_args: ArgsEnum::Unix(args),
-    } if transport == "unix" => {
-      let address_path = Path::new(&args.path);
-      super::check_unstable2(&state, "Deno.connect");
-      {
-        let mut state_ = state.borrow_mut();
-        state_
-          .borrow_mut::<NP>()
-          .check_read(address_path, "Deno.connect()")?;
-        state_
-          .borrow_mut::<NP>()
-          .check_write(address_path, "Deno.connect()")?;
-      }
-      let path = args.path;
-      let unix_stream = net_unix::UnixStream::connect(Path::new(&path)).await?;
-      let local_addr = unix_stream.local_addr()?;
-      let remote_addr = unix_stream.peer_addr()?;
-
-      let mut state_ = state.borrow_mut();
-      let resource = UnixStreamResource::new(unix_stream.into_split());
-      let rid = state_.resource_table.add(resource);
-      Ok(OpConn {
-        rid,
-        local_addr: Some(OpAddr::Unix(net_unix::UnixAddr {
-          path: local_addr.as_pathname().and_then(net_unix::pathstring),
-        })),
-        remote_addr: Some(OpAddr::Unix(net_unix::UnixAddr {
-          path: remote_addr.as_pathname().and_then(net_unix::pathstring),
-        })),
-      })
-    }
-    _ => Err(type_error("Wrong argument format!")),
+  {
+    let mut state_ = state.borrow_mut();
+    state_
+      .borrow_mut::<NP>()
+      .check_net(&(&addr.hostname, Some(addr.port)), "Deno.connect()")?;
   }
+
+  let addr = resolve_addr(&addr.hostname, addr.port)
+    .await?
+    .next()
+    .ok_or_else(|| generic_error("No resolved address found"))?;
+  let tcp_stream = TcpStream::connect(&addr).await?;
+  let local_addr = tcp_stream.local_addr()?;
+  let remote_addr = tcp_stream.peer_addr()?;
+
+  let mut state_ = state.borrow_mut();
+  let rid = state_
+    .resource_table
+    .add(TcpStreamResource::new(tcp_stream.into_split()));
+
+  Ok((rid, IpAddr::from(local_addr), IpAddr::from(remote_addr)))
 }
 
 pub struct TcpListenerResource {
@@ -417,33 +242,20 @@ impl Resource for UdpSocketResource {
   }
 }
 
-#[derive(Deserialize)]
-#[serde(rename_all = "camelCase")]
-struct IpListenArgs {
-  hostname: String,
-  port: u16,
-  reuse_address: Option<bool>,
-}
-
-#[derive(Deserialize)]
-#[serde(untagged)]
-enum ArgsEnum {
-  Ip(IpListenArgs),
-  #[cfg(unix)]
-  Unix(net_unix::UnixListenArgs),
-}
-
-#[derive(Deserialize)]
-struct ListenArgs {
-  transport: String,
-  #[serde(flatten)]
-  transport_args: ArgsEnum,
-}
-
-fn listen_tcp(
+#[op]
+fn op_net_listen_tcp<NP>(
   state: &mut OpState,
-  addr: SocketAddr,
-) -> Result<(u32, SocketAddr), AnyError> {
+  addr: IpAddr,
+) -> Result<(ResourceId, IpAddr), AnyError>
+where
+  NP: NetPermissions + 'static,
+{
+  state
+    .borrow_mut::<NP>()
+    .check_net(&(&addr.hostname, Some(addr.port)), "Deno.listen()")?;
+  let addr = resolve_addr_sync(&addr.hostname, addr.port)?
+    .next()
+    .ok_or_else(|| generic_error("No resolved address found"))?;
   let domain = if addr.is_ipv4() {
     Domain::IPV4
   } else {
@@ -465,21 +277,33 @@ fn listen_tcp(
   };
   let rid = state.resource_table.add(listener_resource);
 
-  Ok((rid, local_addr))
+  Ok((rid, IpAddr::from(local_addr)))
 }
 
-fn listen_udp(
+#[op]
+fn op_net_listen_udp<NP>(
   state: &mut OpState,
-  addr: SocketAddr,
-  reuse_address: Option<bool>,
-) -> Result<(u32, SocketAddr), AnyError> {
+  addr: IpAddr,
+  reuse_address: bool,
+) -> Result<(ResourceId, IpAddr), AnyError>
+where
+  NP: NetPermissions + 'static,
+{
+  super::check_unstable(state, "Deno.listenDatagram");
+  state
+    .borrow_mut::<NP>()
+    .check_net(&(&addr.hostname, Some(addr.port)), "Deno.listenDatagram()")?;
+  let addr = resolve_addr_sync(&addr.hostname, addr.port)?
+    .next()
+    .ok_or_else(|| generic_error("No resolved address found"))?;
+
   let domain = if addr.is_ipv4() {
     Domain::IPV4
   } else {
     Domain::IPV6
   };
   let socket_tmp = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
-  if reuse_address.unwrap_or(false) {
+  if reuse_address {
     // This logic is taken from libuv:
     //
     // On the BSDs, SO_REUSEPORT implies SO_REUSEADDR but with some additional
@@ -508,110 +332,7 @@ fn listen_udp(
   };
   let rid = state.resource_table.add(socket_resource);
 
-  Ok((rid, local_addr))
-}
-
-#[op]
-fn op_net_listen<NP>(
-  state: &mut OpState,
-  args: ListenArgs,
-) -> Result<OpConn, AnyError>
-where
-  NP: NetPermissions + 'static,
-{
-  match args {
-    ListenArgs {
-      transport,
-      transport_args: ArgsEnum::Ip(args),
-    } => {
-      {
-        if transport == "udp" {
-          super::check_unstable(state, "Deno.listenDatagram");
-        }
-        state.borrow_mut::<NP>().check_net(
-          &(&args.hostname, Some(args.port)),
-          "Deno.listenDatagram()",
-        )?;
-      }
-      let addr = resolve_addr_sync(&args.hostname, args.port)?
-        .next()
-        .ok_or_else(|| generic_error("No resolved address found"))?;
-      let (rid, local_addr) = if transport == "tcp" {
-        if args.reuse_address.is_some() {
-          return Err(generic_error(
-            "The reuseAddress option is not supported for TCP",
-          ));
-        }
-        listen_tcp(state, addr)?
-      } else {
-        listen_udp(state, addr, args.reuse_address)?
-      };
-      debug!(
-        "New listener {} {}:{}",
-        rid,
-        local_addr.ip().to_string(),
-        local_addr.port()
-      );
-      let ip_addr = IpAddr {
-        hostname: local_addr.ip().to_string(),
-        port: local_addr.port(),
-      };
-      Ok(OpConn {
-        rid,
-        local_addr: Some(match transport.as_str() {
-          "udp" => OpAddr::Udp(ip_addr),
-          "tcp" => OpAddr::Tcp(ip_addr),
-          // NOTE: This could be unreachable!()
-          other => return Err(bad_transport(other)),
-        }),
-        remote_addr: None,
-      })
-    }
-    #[cfg(unix)]
-    ListenArgs {
-      transport,
-      transport_args: ArgsEnum::Unix(args),
-    } if transport == "unix" || transport == "unixpacket" => {
-      let address_path = Path::new(&args.path);
-      {
-        if transport == "unix" {
-          super::check_unstable(state, "Deno.listen");
-        }
-        if transport == "unixpacket" {
-          super::check_unstable(state, "Deno.listenDatagram");
-        }
-        let api_name = if transport == "unix" {
-          "Deno.listen()"
-        } else {
-          "Deno.listenDatagram()"
-        };
-        let permissions = state.borrow_mut::<NP>();
-        permissions.check_read(address_path, api_name)?;
-        permissions.check_write(address_path, api_name)?;
-      }
-      let (rid, local_addr) = if transport == "unix" {
-        net_unix::listen_unix(state, address_path)?
-      } else {
-        net_unix::listen_unix_packet(state, address_path)?
-      };
-      debug!("New listener {} {:?}", rid, local_addr);
-      let unix_addr = net_unix::UnixAddr {
-        path: local_addr.as_pathname().and_then(net_unix::pathstring),
-      };
-
-      Ok(OpConn {
-        rid,
-        local_addr: Some(match transport.as_str() {
-          "unix" => OpAddr::Unix(unix_addr),
-          "unixpacket" => OpAddr::UnixPacket(unix_addr),
-          other => return Err(bad_transport(other)),
-        }),
-        remote_addr: None,
-      })
-    }
-    #[cfg(unix)]
-    _ => Err(type_error("Wrong argument format!")),
-  }
+  Ok((rid, IpAddr::from(local_addr)))
 }
 
 #[derive(Serialize, Eq, PartialEq, Debug)]
@@ -1128,21 +849,15 @@ mod tests {
     let conn_state = runtime.op_state();
 
     let server_addr: Vec<&str> = clone_addr.split(':').collect();
-    let ip_args = IpListenArgs {
+    let ip_addr = IpAddr {
       hostname: String::from(server_addr[0]),
       port: server_addr[1].parse().unwrap(),
-      reuse_address: None,
-    };
-    let connect_args = ConnectArgs {
-      transport: String::from("tcp"),
-      transport_args: ArgsEnum::Ip(ip_args),
     };
 
     let connect_fut =
-      op_net_connect::call::<TestPermission>(conn_state, connect_args);
-    let conn = connect_fut.await.unwrap();
+      op_net_connect_tcp::call::<TestPermission>(conn_state, ip_addr);
+    let (rid, _, _) = connect_fut.await.unwrap();
 
-    let rid = conn.rid;
     let state = runtime.op_state();
     set_sockopt_fn(&mut state.borrow_mut(), rid);
 
diff --git a/ext/net/ops_tls.rs b/ext/net/ops_tls.rs
index a59cd747e..b27426894 100644
--- a/ext/net/ops_tls.rs
+++ b/ext/net/ops_tls.rs
@@ -2,8 +2,6 @@
 
 use crate::io::TcpStreamResource;
 use crate::ops::IpAddr;
-use crate::ops::OpAddr;
-use crate::ops::OpConn;
 use crate::ops::TlsHandshakeInfo;
 use crate::resolve_addr::resolve_addr;
 use crate::resolve_addr::resolve_addr_sync;
@@ -658,9 +656,9 @@ impl Write for ImplementWriteTrait<'_, TcpStream> {
 pub fn init<P: NetPermissions + 'static>() -> Vec<OpDecl> {
   vec![
     op_tls_start::decl::<P>(),
-    op_tls_connect::decl::<P>(),
-    op_tls_listen::decl::<P>(),
-    op_tls_accept::decl(),
+    op_net_connect_tls::decl::<P>(),
+    op_net_listen_tls::decl::<P>(),
+    op_net_accept_tls::decl(),
     op_tls_handshake::decl(),
   ]
 }
@@ -751,9 +749,6 @@ impl Resource for TlsStreamResource {
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ConnectTlsArgs {
-  transport: String,
-  hostname: String,
-  port: u16,
   cert_file: Option<String>,
   ca_certs: Vec<String>,
   cert_chain: Option<String>,
@@ -774,7 +769,7 @@ pub struct StartTlsArgs {
 pub async fn op_tls_start<NP>(
   state: Rc<RefCell<OpState>>,
   args: StartTlsArgs,
-) -> Result<OpConn, AnyError>
+) -> Result<(ResourceId, IpAddr, IpAddr), AnyError>
 where
   NP: NetPermissions + 'static,
 {
@@ -853,33 +848,18 @@ where
       .add(TlsStreamResource::new(tls_stream.into_split()))
   };
 
-  Ok(OpConn {
-    rid,
-    local_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: local_addr.ip().to_string(),
-      port: local_addr.port(),
-    })),
-    remote_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: remote_addr.ip().to_string(),
-      port: remote_addr.port(),
-    })),
-  })
+  Ok((rid, IpAddr::from(local_addr), IpAddr::from(remote_addr)))
 }
 
 #[op]
-pub async fn op_tls_connect<NP>(
+pub async fn op_net_connect_tls<NP>(
   state: Rc<RefCell<OpState>>,
+  addr: IpAddr,
   args: ConnectTlsArgs,
-) -> Result<OpConn, AnyError>
+) -> Result<(ResourceId, IpAddr, IpAddr), AnyError>
 where
   NP: NetPermissions + 'static,
 {
-  assert_eq!(args.transport, "tcp");
-  let hostname = match &*args.hostname {
-    "" => "localhost",
-    n => n,
-  };
-  let port = args.port;
   let cert_file = args.cert_file.as_deref();
   let unsafely_ignore_certificate_errors = state
     .borrow()
@@ -896,7 +876,8 @@ where
   {
     let mut s = state.borrow_mut();
     let permissions = s.borrow_mut::<NP>();
-    permissions.check_net(&(hostname, Some(port)), "Deno.connectTls()")?;
+    permissions
+      .check_net(&(&addr.hostname, Some(addr.port)), "Deno.connectTls()")?;
     if let Some(path) = cert_file {
       permissions.check_read(Path::new(path), "Deno.connectTls()")?;
     }
@@ -919,10 +900,9 @@ where
     .borrow::<DefaultTlsOptions>()
     .root_cert_store
     .clone();
-  let hostname_dns =
-    ServerName::try_from(hostname).map_err(|_| invalid_hostname(hostname))?;
-
-  let connect_addr = resolve_addr(hostname, port)
+  let hostname_dns = ServerName::try_from(&*addr.hostname)
+    .map_err(|_| invalid_hostname(&addr.hostname))?;
+  let connect_addr = resolve_addr(&addr.hostname, addr.port)
     .await?
     .next()
     .ok_or_else(|| generic_error("No resolved address found"))?;
@@ -968,17 +948,7 @@ where
       .add(TlsStreamResource::new(tls_stream.into_split()))
   };
 
-  Ok(OpConn {
-    rid,
-    local_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: local_addr.ip().to_string(),
-      port: local_addr.port(),
-    })),
-    remote_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: remote_addr.ip().to_string(),
-      port: remote_addr.port(),
-    })),
-  })
+  Ok((rid, IpAddr::from(local_addr), IpAddr::from(remote_addr)))
 }
 
 fn load_certs_from_file(path: &str) -> Result<Vec<Certificate>, AnyError> {
@@ -1013,9 +983,6 @@ impl Resource for TlsListenerResource {
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ListenTlsArgs {
-  transport: String,
-  hostname: String,
-  port: u16,
   cert: Option<String>,
   // TODO(kt3k): Remove this option at v2.0.
   cert_file: Option<String>,
@@ -1026,16 +993,14 @@ pub struct ListenTlsArgs {
 }
 
 #[op]
-pub fn op_tls_listen<NP>(
+pub fn op_net_listen_tls<NP>(
   state: &mut OpState,
+  addr: IpAddr,
   args: ListenTlsArgs,
-) -> Result<OpConn, AnyError>
+) -> Result<(ResourceId, IpAddr), AnyError>
 where
   NP: NetPermissions + 'static,
 {
-  assert_eq!(args.transport, "tcp");
-  let hostname = &*args.hostname;
-  let port = args.port;
   let cert_file = args.cert_file.as_deref();
   let key_file = args.key_file.as_deref();
   let cert = args.cert.as_deref();
@@ -1043,7 +1008,8 @@ where
 
   {
     let permissions = state.borrow_mut::<NP>();
-    permissions.check_net(&(hostname, Some(port)), "Deno.listenTls()")?;
+    permissions
+      .check_net(&(&addr.hostname, Some(addr.port)), "Deno.listenTls()")?;
     if let Some(path) = cert_file {
       permissions.check_read(Path::new(path), "Deno.listenTls()")?;
     }
@@ -1084,7 +1050,7 @@ where
       alpn_protocols.into_iter().map(|s| s.into_bytes()).collect();
   }
 
-  let bind_addr = resolve_addr_sync(hostname, port)?
+  let bind_addr = resolve_addr_sync(&addr.hostname, addr.port)?
     .next()
     .ok_or_else(|| generic_error("No resolved address found"))?;
   let domain = if bind_addr.is_ipv4() {
@@ -1111,21 +1077,14 @@ where
 
   let rid = state.resource_table.add(tls_listener_resource);
 
-  Ok(OpConn {
-    rid,
-    local_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: local_addr.ip().to_string(),
-      port: local_addr.port(),
-    })),
-    remote_addr: None,
-  })
+  Ok((rid, IpAddr::from(local_addr)))
 }
 
 #[op]
-pub async fn op_tls_accept(
+pub async fn op_net_accept_tls(
   state: Rc<RefCell<OpState>>,
   rid: ResourceId,
-) -> Result<OpConn, AnyError> {
+) -> Result<(ResourceId, IpAddr, IpAddr), AnyError> {
   let resource = state
     .borrow()
     .resource_table
@@ -1159,17 +1118,7 @@ pub async fn op_tls_accept(
       .add(TlsStreamResource::new(tls_stream.into_split()))
   };
 
-  Ok(OpConn {
-    rid,
-    local_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: local_addr.ip().to_string(),
-      port: local_addr.port(),
-    })),
-    remote_addr: Some(OpAddr::Tcp(IpAddr {
-      hostname: remote_addr.ip().to_string(),
-      port: remote_addr.port(),
-    })),
-  })
+  Ok((rid, IpAddr::from(local_addr), IpAddr::from(remote_addr)))
 }
 
 #[op]
diff --git a/ext/net/ops_unix.rs b/ext/net/ops_unix.rs
index 181dcacec..b45b02343 100644
--- a/ext/net/ops_unix.rs
+++ b/ext/net/ops_unix.rs
@@ -1,20 +1,18 @@
 // Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
 
 use crate::io::UnixStreamResource;
-use crate::ops::AcceptArgs;
-use crate::ops::OpAddr;
-use crate::ops::OpConn;
-use crate::ops::OpPacket;
-use crate::ops::ReceiveArgs;
+use crate::NetPermissions;
 use deno_core::error::bad_resource;
 use deno_core::error::custom_error;
 use deno_core::error::AnyError;
+use deno_core::op;
 use deno_core::AsyncRefCell;
 use deno_core::CancelHandle;
 use deno_core::CancelTryFuture;
 use deno_core::OpState;
 use deno_core::RcRef;
 use deno_core::Resource;
+use deno_core::ResourceId;
 use deno_core::ZeroCopyBuf;
 use serde::Deserialize;
 use serde::Serialize;
@@ -74,13 +72,11 @@ pub struct UnixListenArgs {
   pub path: String,
 }
 
-pub(crate) async fn accept_unix(
+#[op]
+pub async fn op_net_accept_unix(
   state: Rc<RefCell<OpState>>,
-  args: AcceptArgs,
-  _: (),
-) -> Result<OpConn, AnyError> {
-  let rid = args.rid;
-
+  rid: ResourceId,
+) -> Result<(ResourceId, Option<String>, Option<String>), AnyError> {
   let resource = state
     .borrow()
     .resource_table
@@ -98,27 +94,52 @@ pub(crate) async fn accept_unix(
 
   let local_addr = unix_stream.local_addr()?;
   let remote_addr = unix_stream.peer_addr()?;
+  let local_addr_path = local_addr.as_pathname().map(pathstring).transpose()?;
+  let remote_addr_path =
+    remote_addr.as_pathname().map(pathstring).transpose()?;
   let resource = UnixStreamResource::new(unix_stream.into_split());
   let mut state = state.borrow_mut();
   let rid = state.resource_table.add(resource);
-  Ok(OpConn {
-    rid,
-    local_addr: Some(OpAddr::Unix(UnixAddr {
-      path: local_addr.as_pathname().and_then(pathstring),
-    })),
-    remote_addr: Some(OpAddr::Unix(UnixAddr {
-      path: remote_addr.as_pathname().and_then(pathstring),
-    })),
-  })
+  Ok((rid, local_addr_path, remote_addr_path))
 }
 
-pub(crate) async fn receive_unix_packet(
+#[op]
+pub async fn op_net_connect_unix<NP>(
   state: Rc<RefCell<OpState>>,
-  args: ReceiveArgs,
-  mut buf: ZeroCopyBuf,
-) -> Result<OpPacket, AnyError> {
-  let rid = args.rid;
+  path: String,
+) -> Result<(ResourceId, Option<String>, Option<String>), AnyError>
+where
+  NP: NetPermissions + 'static,
+{
+  let address_path = Path::new(&path);
+  super::check_unstable2(&state, "Deno.connect");
+  {
+    let mut state_ = state.borrow_mut();
+    state_
+      .borrow_mut::<NP>()
+      .check_read(address_path, "Deno.connect()")?;
+    state_
+      .borrow_mut::<NP>()
+      .check_write(address_path, "Deno.connect()")?;
+  }
+  let unix_stream = UnixStream::connect(Path::new(&path)).await?;
+  let local_addr = unix_stream.local_addr()?;
+  let remote_addr = unix_stream.peer_addr()?;
+  let local_addr_path = local_addr.as_pathname().map(pathstring).transpose()?;
+  let remote_addr_path =
+    remote_addr.as_pathname().map(pathstring).transpose()?;
+  let mut state_ = state.borrow_mut();
+  let resource = UnixStreamResource::new(unix_stream.into_split());
+  let rid = state_.resource_table.add(resource);
+  Ok((rid, local_addr_path, remote_addr_path))
+}
 
+#[op]
+pub async fn op_net_recv_unixpacket(
+  state: Rc<RefCell<OpState>>,
+  rid: ResourceId,
+  mut buf: ZeroCopyBuf,
+) -> Result<(usize, Option<String>), AnyError> {
   let resource = state
     .borrow()
     .resource_table
@@ -128,46 +149,90 @@ pub(crate) async fn receive_unix_packet(
     .try_borrow_mut()
     .ok_or_else(|| custom_error("Busy", "Socket already in use"))?;
   let cancel = RcRef::map(resource, |r| &r.cancel);
-  let (size, remote_addr) =
+  let (nread, remote_addr) =
     socket.recv_from(&mut buf).try_or_cancel(cancel).await?;
-  Ok(OpPacket {
-    size,
-    remote_addr: OpAddr::UnixPacket(UnixAddr {
-      path: remote_addr.as_pathname().and_then(pathstring),
-    }),
-  })
+  let path = remote_addr.as_pathname().map(pathstring).transpose()?;
+  Ok((nread, path))
+}
+
+#[op]
+async fn op_net_send_unixpacket<NP>(
+  state: Rc<RefCell<OpState>>,
+  rid: ResourceId,
+  path: String,
+  zero_copy: ZeroCopyBuf,
+) -> Result<usize, AnyError>
+where
+  NP: NetPermissions + 'static,
+{
+  let address_path = Path::new(&path);
+  {
+    let mut s = state.borrow_mut();
+    s.borrow_mut::<NP>()
+      .check_write(address_path, "Deno.DatagramConn.send()")?;
+  }
+
+  let resource = state
+    .borrow()
+    .resource_table
+    .get::<UnixDatagramResource>(rid)
+    .map_err(|_| custom_error("NotConnected", "Socket has been closed"))?;
+  let socket = RcRef::map(&resource, |r| &r.socket)
+    .try_borrow_mut()
+    .ok_or_else(|| custom_error("Busy", "Socket already in use"))?;
+  let nwritten = socket.send_to(&zero_copy, address_path).await?;
+
+  Ok(nwritten)
 }
 
-pub fn listen_unix(
+#[op]
+pub fn op_net_listen_unix<NP>(
   state: &mut OpState,
-  addr: &Path,
-) -> Result<(u32, tokio::net::unix::SocketAddr), AnyError> {
-  let listener = UnixListener::bind(&addr)?;
+  path: String,
+) -> Result<(ResourceId, Option<String>), AnyError>
+where
+  NP: NetPermissions + 'static,
+{
+  let address_path = Path::new(&path);
+  super::check_unstable(state, "Deno.listen");
+  let permissions = state.borrow_mut::<NP>();
+  permissions.check_read(address_path, "Deno.listen()")?;
+  permissions.check_write(address_path, "Deno.listen()")?;
+  let listener = UnixListener::bind(&address_path)?;
   let local_addr = listener.local_addr()?;
+  let pathname = local_addr.as_pathname().map(pathstring).transpose()?;
   let listener_resource = UnixListenerResource {
     listener: AsyncRefCell::new(listener),
     cancel: Default::default(),
   };
   let rid = state.resource_table.add(listener_resource);
-
-  Ok((rid, local_addr))
+  Ok((rid, pathname))
 }
 
-pub fn listen_unix_packet(
+#[op]
+pub fn op_net_listen_unixpacket<NP>(
   state: &mut OpState,
-  addr: &Path,
-) -> Result<(u32, tokio::net::unix::SocketAddr), AnyError> {
-  let socket = UnixDatagram::bind(&addr)?;
+  path: String,
+) -> Result<(ResourceId, Option<String>), AnyError>
+where
+  NP: NetPermissions + 'static,
+{
+  let address_path = Path::new(&path);
+  super::check_unstable(state, "Deno.listenDatagram");
+  let permissions = state.borrow_mut::<NP>();
+  permissions.check_read(address_path, "Deno.listenDatagram()")?;
+  permissions.check_write(address_path, "Deno.listenDatagram()")?;
+  let socket = UnixDatagram::bind(&address_path)?;
   let local_addr = socket.local_addr()?;
+  let pathname = local_addr.as_pathname().map(pathstring).transpose()?;
   let datagram_resource = UnixDatagramResource {
     socket: AsyncRefCell::new(socket),
     cancel: Default::default(),
   };
   let rid = state.resource_table.add(datagram_resource);
-
-  Ok((rid, local_addr))
+  Ok((rid, pathname))
 }
 
-pub fn pathstring(pathname: &Path) -> Option<String> {
-  into_string(pathname.into()).ok()
+pub fn pathstring(pathname: &Path) -> Result<String, AnyError> {
+  into_string(pathname.into())
 }