Add windows NamedPipe

Cargo.toml

@@ -49,6 +49,7 @@ ntapi  = "0.3"
 [dev-dependencies]
 env_logger = { version = "0.6.2", default-features = false }
 net2       = "0.2.33"
+rand = "0.4"
 
 [package.metadata.docs.rs]
 all-features = true

src/lib.rs

@@ -89,6 +89,16 @@ pub mod unix {
     pub use crate::sys::SourceFd;
 }
 
+#[cfg(all(windows, feature = "os-util"))]
+#[cfg_attr(docsrs, doc(cfg(all(windows, feature = "os-util"))))]
+pub mod windows {
+    //! Windows only extensions.
+
+    cfg_os_poll! {
+        pub use crate::sys::named_pipe::NamedPipe;
+    }
+}
+
 // Enable with `cargo doc --features extra-docs`.
 #[cfg(feature = "extra-docs")]
 pub mod features {

src/sys/mod.rs

@@ -80,7 +80,7 @@ cfg_os_poll! {
 #[cfg(windows)]
 cfg_os_poll! {
     mod windows;
-    pub(crate) use self::windows::*;
+    pub use self::windows::*;
 }
 
 cfg_not_os_poll! {

src/sys/windows/afd.rs

@@ -7,9 +7,7 @@ use std::io;
 use std::mem::size_of;
 use std::os::windows::io::AsRawHandle;
 use std::ptr::null_mut;
-use winapi::shared::ntdef::{
-    HANDLE, LARGE_INTEGER, NTSTATUS, PVOID, ULONG,
-};
+use winapi::shared::ntdef::{HANDLE, LARGE_INTEGER, NTSTATUS, PVOID, ULONG};
 use winapi::shared::ntstatus::{STATUS_NOT_FOUND, STATUS_PENDING, STATUS_SUCCESS};
 
 const IOCTL_AFD_POLL: ULONG = 0x00012024;
@@ -196,7 +194,11 @@ cfg_net! {
                     ));
                 }
                 let fd = File::from_raw_handle(afd_helper_handle as RawHandle);
-                let token = NEXT_TOKEN.fetch_add(1, Ordering::Relaxed) + 1;
+                // Increment by 2 to reserve space for other types of handles.
+                // Non-AFD types (currently only NamedPipe), use odd numbered
+                // tokens. This allows the selector to differentate between them
+                // and dispatch events accordingly.
+                let token = NEXT_TOKEN.fetch_add(2, Ordering::Relaxed) + 2;
                 let afd = Afd { fd };
                 cp.add_handle(token, &afd.fd)?;
                 match SetFileCompletionNotificationModes(

src/sys/windows/event.rs

@@ -14,6 +14,35 @@ pub fn token(event: &Event) -> Token {
     Token(event.data as usize)
 }
 
+impl Event {
+    pub(super) fn new(token: Token) -> Event {
+        Event {
+            flags: 0,
+            data: usize::from(token) as u64,
+        }
+    }
+
+    pub(super) fn set_readable(&mut self) {
+        self.flags |= afd::POLL_RECEIVE
+    }
+
+    #[cfg(feature = "os-util")]
+    pub(super) fn set_writable(&mut self) {
+        self.flags |= afd::POLL_SEND;
+    }
+
+    pub(super) fn from_completion_status(status: &CompletionStatus) -> Event {
+        Event {
+            flags: status.bytes_transferred(),
+            data: status.token() as u64,
+        }
+    }
+
+    pub(super) fn to_completion_status(&self) -> CompletionStatus {
+        CompletionStatus::new(self.flags, self.data as usize, std::ptr::null_mut())
+    }
+}
+
 pub(crate) const READABLE_FLAGS: u32 = afd::POLL_RECEIVE
     | afd::POLL_DISCONNECT
     | afd::POLL_ACCEPT

src/sys/windows/mod.rs

@@ -7,6 +7,9 @@ pub use event::{Event, Events};
 mod selector;
 pub use selector::{Selector, SelectorInner, SockState};
 
+mod overlapped;
+use overlapped::Overlapped;
+
 // Macros must be defined before the modules that use them
 cfg_net! {
     /// Helper macro to execute a system call that returns an `io::Result`.
@@ -32,6 +35,9 @@ cfg_udp! {
     pub(crate) mod udp;
 }
 
+#[cfg(feature = "os-util")]
+pub(crate) mod named_pipe;
+
 mod waker;
 pub(crate) use waker::Waker;