feat(io): add SyncStream::into_parts (#847)

Add methods to safely extract the underlying stream and buffered data
from the compat adapter layers, needed for protocol upgrades (e.g.
WebSocket) to bypass double-buffering.

- `SyncStream::into_parts()`: returns `(S, Vec<u8>)` with unread data

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: Xerxes-2

compio-io/src/compat/async_stream.rs

@@ -68,7 +68,7 @@ impl<S> AsyncStream<S> {
         self.inner.get_mut()
     }
 
-    /// Consumes the `SyncStream`, returning the underlying stream.
+    /// Consumes the `AsyncStream`, returning the underlying stream.
     pub fn into_inner(self) -> S {
         self.inner.into_inner()
     }

compio-io/src/compat/sync_stream.rs

@@ -102,10 +102,33 @@ impl<S> SyncStream<S> {
     }
 
     /// Consumes the `SyncStream`, returning the underlying stream.
+    ///
+    /// Any buffered data is discarded. Use [`into_parts`](Self::into_parts)
+    /// if you need to preserve unread data.
     pub fn into_inner(self) -> S {
         self.inner
     }
 
+    /// Consumes the `SyncStream`, returning the underlying stream and any
+    /// unread buffered data.
+    ///
+    /// If the read buffer is currently lent to an IO operation, the returned
+    /// `Vec` will be empty.
+    pub fn into_parts(mut self) -> (S, Vec<u8>) {
+        let remaining = if self.read_buf.has_inner() {
+            let slice = self.read_buf.take_inner();
+            let begin = slice.begin();
+            let mut vec = slice.into_inner();
+            if begin > 0 {
+                vec.drain(..begin);
+            }
+            vec
+        } else {
+            Vec::new()
+        };
+        (self.inner, remaining)
+    }
+
     /// Returns `true` if the stream has reached EOF.
     pub fn is_eof(&self) -> bool {
         self.eof

compio-io/tests/compat.rs

@@ -1,6 +1,6 @@
-use std::io::Cursor;
+use std::io::{Cursor, Read};
 
-use compio_io::compat::{AsyncReadStream, AsyncWriteStream};
+use compio_io::compat::{AsyncReadStream, AsyncWriteStream, SyncStream};
 use futures_executor::block_on;
 use futures_util::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt};
 
@@ -87,3 +87,17 @@ fn async_compat_flush_fail() {
         assert_eq!(err.kind(), std::io::ErrorKind::WriteZero);
     })
 }
+
+#[test]
+fn sync_stream_into_parts_keeps_unread_buffer() {
+    let mut stream = SyncStream::new(Cursor::new(b"hello".to_vec()));
+    let mut buf = [0; 2];
+
+    Read::read(&mut stream, &mut buf).unwrap_err();
+    futures_executor::block_on(stream.fill_read_buf()).unwrap();
+    assert_eq!(Read::read(&mut stream, &mut buf).unwrap(), 2);
+    assert_eq!(&buf, b"he");
+
+    let (_, remaining) = stream.into_parts();
+    assert_eq!(remaining, b"llo");
+}