feat(node): implement PipeWrap for Windows

Author: CyanChanges

ext/net/04_pipe.ts

@@ -59,7 +59,7 @@ async function write(rid, data) {
 }
 
 class Pipe {
-  #rid = 0;
+  readonly #rid = 0;
   #unref = false;
   #pendingReadPromises = new SafeSet();
 
@@ -76,11 +76,11 @@ class Pipe {
     this.#rid = rid;
   }
 
-  write(buffer) {
+  write(buffer): Promise<number> {
     return write(this.#rid, buffer);
   }
 
-  async read(buffer) {
+  async read(buffer): Promise<number> {
     if (buffer.length === 0) {
       return 0;
     }
@@ -101,7 +101,7 @@ class Pipe {
   close() {
     core.close(this.#rid);
   }
-  get readable() {
+  get readable(): ReadableStream<Uint8Array> {
     if (this.#readable === undefined) {
       this.#readable = readableStreamForRidUnrefable(this.#rid);
       if (this.#unref) {
@@ -111,7 +111,7 @@ class Pipe {
     return this.#readable;
   }
 
-  get writable() {
+  get writable(): WritableStream<Uint8Array> {
     if (this.#writable === undefined) {
       this.#writable = writableStreamForRid(this.#rid);
     }
@@ -146,14 +146,14 @@ class Pipe {
   }
 }
 
-function connect(opts: Options | WindowsConnectOptions) {
+async function connect(opts: Options | WindowsConnectOptions) {
   let rid: number;
   switch (opts.kind) {
     case "unix":
       rid = op_pipe_connect(opts.path, "Deno.pipe.connect");
       return new Pipe(rid);
     case "windows":
-      rid = op_pipe_connect(opts, "Deno.pipe.connect");
+      rid = await op_pipe_connect(opts, "Deno.pipe.connect");
       return new Pipe(rid);
     default:
       throw new Error(`Unsupported kind: ${opts.kind}`);

ext/net/ops_win_pipe.rs

@@ -102,7 +102,7 @@ where
 
 #[op2(stack_trace)]
 #[smi]
-pub fn op_pipe_connect<NP>(
+pub async fn op_pipe_connect<NP>(
   state: &mut OpState,
   #[serde] args: ConnectArgs,
   #[string] api_name: &str,
@@ -132,10 +132,9 @@ where
   };
 
   let mut opts = named_pipe::ClientOptions::new();
-  opts
-      .read(args.read)
-      .write(args.write);
+  opts.read(args.read).write(args.write);
   let pipe = NamedPipe::new_client(path.as_ref(), &opts)?;
+  pipe.connect().await?;
   let rid = state.resource_table.add(pipe);
   Ok(rid)
 }

ext/net/win_pipe.rs

@@ -3,10 +3,8 @@
 use std::borrow::Cow;
 use std::ffi::OsStr;
 use std::io;
-use std::path::Path;
 use std::rc::Rc;
 
-use deno_core::AsyncRef;
 use deno_core::AsyncRefCell;
 use deno_core::AsyncResult;
 use deno_core::CancelHandle;
@@ -23,19 +21,18 @@ pub struct NamedPipe {
 }
 
 enum Inner {
-  Server((bool, named_pipe::NamedPipeServer)),
+  Server(named_pipe::NamedPipeServer),
   Client(named_pipe::NamedPipeClient),
 }
 
 impl NamedPipe {
-  pub fn new_server(
+  pub async fn new_server(
     addr: impl AsRef<OsStr>,
     options: &named_pipe::ServerOptions,
   ) -> io::Result<NamedPipe> {
     let server = options.create(addr)?;
-    // todo: should we connect here
     Ok(NamedPipe {
-      inner: AsyncRefCell::new(Inner::Server((false, server))),
+      inner: AsyncRefCell::new(Inner::Server(server)),
       cancel: Default::default(),
     })
   }
@@ -51,17 +48,20 @@ impl NamedPipe {
     })
   }
 
+  pub async fn connect(&self) -> io::Result<()> {
+    match &self.inner {
+      Inner::Server(ref inner) => {
+        inner.connect().try_or_cancel(&self.cancel).await
+      }
+      Inner::Client(ref inner) => Ok(()),
+    }
+  }
+
   pub async fn write(self: Rc<Self>, buf: &[u8]) -> io::Result<usize> {
     let mut inner = RcRef::map(&self, |s| &s.inner).borrow_mut().await;
     let cancel = RcRef::map(&self, |s| &s.cancel);
     match &mut *inner {
-      Inner::Server((mut connected, server)) => {
-        if !connected {
-          server.connect().await?;
-          connected = true;
-        }
-        server.write(buf).try_or_cancel(cancel).await
-      },
+      Inner::Server(server) => server.write(buf).try_or_cancel(cancel).await,
       Inner::Client(client) => client.write(buf).try_or_cancel(cancel).await,
     }
   }
@@ -70,13 +70,7 @@ impl NamedPipe {
     let mut inner = RcRef::map(&self, |s| &s.inner).borrow_mut().await;
     let cancel = RcRef::map(&self, |s| &s.cancel);
     match &mut *inner {
-      Inner::Server((mut connected, server)) => {
-        if !connected {
-          server.connect().await?;
-          connected = true;
-        }
-        server.read(buf).try_or_cancel(cancel).await
-      },
+      Inner::Server(server) => server.read(buf).try_or_cancel(cancel).await,
       Inner::Client(client) => client.read(buf).try_or_cancel(cancel).await,
     }
   }

ext/node/polyfills/internal_binding/pipe_wrap.ts

@@ -59,7 +59,7 @@ export class Pipe extends ConnectionWrap {
   override reading = false;
   ipc: boolean;
 
-  // REF: https://github.com/nodejs/node/blob/master/deps/uv/src/win/pipe.c#L48
+  // REF: https://github.com/nodejs/node/blob/main/deps/uv/src/win/pipe.c#L48
   #pendingInstances = 4;
 
   #address?: string;
@@ -71,7 +71,10 @@ export class Pipe extends ConnectionWrap {
   #closed = false;
   #acceptBackoffDelay?: number;
 
-  constructor(type: number, conn?: Deno.UnixConn | StreamBase) {
+  constructor(
+    type: number,
+    conn?: Deno.UnixConn | Deno.PipeNs.Pipe | StreamBase,
+  ) {
     let provider: providerType;
     let ipc: boolean;
 
@@ -137,8 +140,30 @@ export class Pipe extends ConnectionWrap {
    */
   connect(req: PipeConnectWrap, address: string) {
     if (isWindows) {
-      // REF: https://github.com/denoland/deno/issues/10244
-      notImplemented("Pipe.prototype.connect - Windows");
+      try {
+        this[kStreamBaseField] = Deno.pipe.connect({
+          path: address,
+          kind: "windows",
+        });
+        this.#address = req.address;
+
+        try {
+          this.afterConnect(req, 0);
+        } catch {
+          // swallow callback errors.
+        }
+      } catch (e) {
+        const code = codeMap.get(e.code ?? "UNKNOWN") ??
+          codeMap.get("UNKNOWN")!;
+
+        try {
+          this.afterConnect(req, code);
+        } catch {
+          // swallow callback errors.
+        }
+      }
+
+      return 0;
     }
 
     const connectOptions: Deno.UnixConnectOptions = {
@@ -180,15 +205,30 @@ export class Pipe extends ConnectionWrap {
    * @return An error status code.
    */
   listen(backlog: number): number {
-    if (isWindows) {
-      // REF: https://github.com/denoland/deno/issues/10244
-      notImplemented("Pipe.prototype.listen - Windows");
-    }
-
     this.#backlog = isWindows
       ? this.#pendingInstances
       : ceilPowOf2(backlog + 1);
 
+    if (isWindows) {
+      let listener;
+      try {
+        listener = Deno.pipe.listen({
+          path: this.#address!,
+          kind: "windows",
+          pipeMode: "byte",
+        });
+      } catch (e) {
+        if (e instanceof Deno.errors.NotCapable) {
+          throw e;
+        }
+        return codeMap.get(e.code ?? "UNKNOWN") ?? codeMap.get("UNKNOWN")!;
+      }
+
+      this.#listener = listener;
+
+      return 0;
+    }
+
     const listenOptions = {
       path: this.#address!,
       transport: "unix" as const,
@@ -299,6 +339,10 @@ export class Pipe extends ConnectionWrap {
       return;
     }
 
+    if (isWindows) {
+      return;
+    }
+
     if (this.#connections > this.#backlog!) {
       this.#acceptBackoff();