diff --git a/src/os/windows/named_pipe/stream/impl/ctor.rs b/src/os/windows/named_pipe/stream/impl/ctor.rs
index 8db1dab..b95e364 100644
--- a/src/os/windows/named_pipe/stream/impl/ctor.rs
+++ b/src/os/windows/named_pipe/stream/impl/ctor.rs
@@ -4,19 +4,19 @@ use widestring::U16CStr;
 use windows_sys::Win32::System::Pipes::PIPE_READMODE_MESSAGE;
 
 impl RawPipeStream {
-	pub(super) fn new(handle: FileHandle, is_server: bool) -> Self {
+	pub(super) fn new(handle: FileHandle, is_server: bool, nfv: NeedsFlushVal) -> Self {
 		Self {
 			handle: Some(handle),
 			is_server,
-			needs_flush: NeedsFlush::from(NeedsFlushVal::No),
+			needs_flush: NeedsFlush::from(nfv),
 			concurrency_detector: ConcurrencyDetector::new(),
 		}
 	}
 	pub(crate) fn new_server(handle: FileHandle) -> Self {
-		Self::new(handle, true)
+		Self::new(handle, true, NeedsFlushVal::No)
 	}
 	fn new_client(handle: FileHandle) -> Self {
-		Self::new(handle, false)
+		Self::new(handle, false, NeedsFlushVal::No)
 	}
 	fn connect(
 		path: &U16CStr,
diff --git a/src/os/windows/named_pipe/stream/impl/handle.rs b/src/os/windows/named_pipe/stream/impl/handle.rs
index 6282ae8..450d510 100644
--- a/src/os/windows/named_pipe/stream/impl/handle.rs
+++ b/src/os/windows/named_pipe/stream/impl/handle.rs
@@ -17,7 +17,11 @@ derive_asraw!(RawPipeStream);
 
 impl RawPipeStream {
 	fn from_handle_given_flags(handle: OwnedHandle, flags: u32) -> Self {
-		Self::new(FileHandle::from(handle), flags & PIPE_SERVER_END != 0)
+		Self::new(
+			FileHandle::from(handle),
+			flags & PIPE_SERVER_END != 0,
+			NeedsFlushVal::Once,
+		)
 	}
 }
 
@@ -95,8 +99,7 @@ impl<Rm: PipeModeTag, Sm: PipeModeTag> TryClone for PipeStream<Rm, Sm> {
 	fn try_clone(&self) -> io::Result<Self> {
 		let handle = duplicate_handle(self.as_handle())?;
 		self.raw.needs_flush.on_clone();
-		let mut new = RawPipeStream::new(handle.into(), self.is_server());
-		new.needs_flush = NeedsFlushVal::Always.into();
+		let new = RawPipeStream::new(handle.into(), self.is_server(), NeedsFlushVal::Always);
 		Ok(Self::new(new))
 	}
 }
diff --git a/src/os/windows/named_pipe/tokio/stream/impl/ctor.rs b/src/os/windows/named_pipe/tokio/stream/impl/ctor.rs
index 46483f0..3c0f189 100644
--- a/src/os/windows/named_pipe/tokio/stream/impl/ctor.rs
+++ b/src/os/windows/named_pipe/tokio/stream/impl/ctor.rs
@@ -5,18 +5,18 @@ use crate::os::windows::{named_pipe::WaitTimeout, path_conversion::*, NeedsFlush
 use std::{borrow::Cow, mem::take};
 
 impl RawPipeStream {
-	pub(super) fn new(inner: InnerTokio) -> Self {
+	pub(super) fn new(inner: InnerTokio, nfv: NeedsFlushVal) -> Self {
 		Self {
 			inner: Some(inner),
-			needs_flush: NeedsFlush::from(NeedsFlushVal::No),
+			needs_flush: NeedsFlush::from(nfv),
 			//recv_msg_state: Mutex::new(RecvMsgState::NotRecving),
 		}
 	}
 	pub(crate) fn new_server(server: TokioNPServer) -> Self {
-		Self::new(InnerTokio::Server(server))
+		Self::new(InnerTokio::Server(server), NeedsFlushVal::No)
 	}
 	fn new_client(client: TokioNPClient) -> Self {
-		Self::new(InnerTokio::Client(client))
+		Self::new(InnerTokio::Client(client), NeedsFlushVal::No)
 	}
 
 	async fn wait_for_server(path: U16CString) -> io::Result<U16CString> {
diff --git a/src/os/windows/named_pipe/tokio/stream/impl/handle.rs b/src/os/windows/named_pipe/tokio/stream/impl/handle.rs
index fb96e01..c658dab 100644
--- a/src/os/windows/named_pipe/tokio/stream/impl/handle.rs
+++ b/src/os/windows/named_pipe/tokio/stream/impl/handle.rs
@@ -1,6 +1,7 @@
 use windows_sys::Win32::System::Pipes::{PIPE_SERVER_END, PIPE_TYPE_MESSAGE};
 
 use super::*;
+use crate::os::windows::NeedsFlushVal;
 use std::mem::ManuallyDrop;
 
 impl AsHandle for InnerTokio {
@@ -34,7 +35,7 @@ impl RawPipeStream {
 			}
 		};
 		match tkresult {
-			Ok(s) => Ok(Self::new(s)),
+			Ok(s) => Ok(Self::new(s, NeedsFlushVal::Once)),
 			Err(e) => Err(FromHandleError {
 				details: FromHandleErrorKind::TokioError,
 				cause: Some(e),