test: test handling for ERROR_MORE_DATA

tests/win_named_pipe.rs

@@ -1,15 +1,24 @@
 #![cfg(all(windows, feature = "os-poll", feature = "os-ext"))]
 
+use std::ffi::OsStr;
 use std::fs::OpenOptions;
 use std::io::{self, Read, Write};
+use std::os::windows::ffi::OsStrExt;
 use std::os::windows::fs::OpenOptionsExt;
-use std::os::windows::io::{FromRawHandle, IntoRawHandle};
+use std::os::windows::io::{FromRawHandle, IntoRawHandle, RawHandle};
 use std::time::Duration;
 
 use mio::windows::NamedPipe;
 use mio::{Events, Interest, Poll, Token};
 use rand::Rng;
-use windows_sys::Win32::{Foundation::ERROR_NO_DATA, Storage::FileSystem::FILE_FLAG_OVERLAPPED};
+use windows_sys::Win32::Foundation::ERROR_NO_DATA;
+use windows_sys::Win32::Storage::FileSystem::{
+    CreateFileW, FILE_FLAG_FIRST_PIPE_INSTANCE, FILE_FLAG_OVERLAPPED, OPEN_EXISTING,
+    PIPE_ACCESS_DUPLEX,
+};
+use windows_sys::Win32::System::Pipes::{
+    CreateNamedPipeW, PIPE_READMODE_MESSAGE, PIPE_TYPE_MESSAGE, PIPE_UNLIMITED_INSTANCES,
+};
 
 fn _assert_kinds() {
     fn _assert_send<T: Send>() {}
@@ -43,6 +52,38 @@ fn client(name: &str) -> NamedPipe {
     unsafe { NamedPipe::from_raw_handle(file.into_raw_handle()) }
 }
 
+fn pipe_msg_mode() -> (NamedPipe, NamedPipe) {
+    let num: u64 = rand::thread_rng().gen();
+    let name = format!(r"\\.\pipe\my-pipe-{}", num);
+    let name: Vec<_> = OsStr::new(&name).encode_wide().chain(Some(0)).collect();
+    unsafe {
+        let h = CreateNamedPipeW(
+            name.as_ptr(),
+            PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE | FILE_FLAG_OVERLAPPED,
+            PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE,
+            PIPE_UNLIMITED_INSTANCES,
+            65536,
+            65536,
+            0,
+            std::ptr::null_mut(),
+        );
+
+        let server = NamedPipe::from_raw_handle(h as RawHandle);
+
+        let h = CreateFileW(
+            name.as_ptr(),
+            PIPE_ACCESS_DUPLEX,
+            0,
+            std::ptr::null_mut(),
+            OPEN_EXISTING,
+            FILE_FLAG_OVERLAPPED,
+            0,
+        );
+        let client = NamedPipe::from_raw_handle(h as RawHandle);
+        (server, client)
+    }
+}
+
 fn pipe() -> (NamedPipe, NamedPipe) {
     let (pipe, name) = server();
     (pipe, client(&name))
@@ -108,6 +149,42 @@ fn write_then_read() {
     assert_eq!(&buf[..4], b"1234");
 }
 
+#[test]
+fn read_sz_greater_than_default_buf_size() {
+    let (mut server, mut client) = pipe_msg_mode();
+    let mut poll = t!(Poll::new());
+    t!(poll.registry().register(
+        &mut server,
+        Token(0),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+    t!(poll.registry().register(
+        &mut client,
+        Token(1),
+        Interest::READABLE | Interest::WRITABLE,
+    ));
+
+    let mut events = Events::with_capacity(128);
+    let msg = (0..4106).map(|e| e.to_string()).collect::<Vec<_>>().join("");
+
+    t!(poll.poll(&mut events, None));
+    assert_eq!(t!(client.write(msg.as_bytes())), 15314);
+
+    loop {
+        t!(poll.poll(&mut events, None));
+        let events = events.iter().collect::<Vec<_>>();
+        if let Some(event) = events.iter().find(|e| e.token() == Token(0)) {
+            if event.is_readable() {
+                break;
+            }
+        }
+    }
+
+    let mut buf = [0; 15314];
+    assert_eq!(t!(server.read(&mut buf)), 15314);
+    assert_eq!(&buf[..15314], msg.as_bytes());
+}
+
 #[test]
 fn connect_before_client() {
     let (mut server, name) = server();