Support split for **any** T: AsyncRead + AsyncWrite + Unpin + 'split

Author: kanarus

Cargo.toml

@@ -20,20 +20,18 @@ tokio     = { optional = true, version = "1", features = ["net", "io-util", "syn
 smol      = { optional = true, version = "2"   }
 glommio   = { optional = true, version = "0.9" }
 
-futures-util = { optional = true, version = "0.3", default-features = false, features = ["io"] }
+futures-util = { version = "0.3", default-features = false, features = ["io"] }
 
 sha1   = { version = "0.10", default-features = false }
 base64 = { version = "0.22" }
 
 [features]
-rt_tokio     = ["__splitref__", "dep:tokio","tokio/net","tokio/io-util","tokio/sync","tokio/time"]
-rt_smol      = ["__clone__",    "dep:smol"]
-rt_glommio   = ["__splitref__", "dep:glommio", "dep:futures-util"]
+rt_tokio     = ["__runtime__", "dep:tokio", "futures-util/unstable","futures-util/bilock"]
+rt_smol      = ["__runtime__", "dep:smol"]
+rt_glommio   = ["__runtime__", "dep:glommio"]
 
 ### internal ###
 __runtime__  = []
-__splitref__ = ["__runtime__"]
-__clone__    = ["__runtime__"]
 
 #####################################
 DEBUG = ["tokio?/rt"] ###############

src/connection.rs

@@ -333,36 +333,76 @@ pub mod split {
         }
     }
 
-    #[cfg(feature="__splitref__")]
-    const _: () = {
-        #[cfg(feature="rt_tokio")]
-        impl<'split> Splitable<'split> for tokio::net::TcpStream {
-            type ReadHalf  = tokio::net::tcp::ReadHalf <'split>;
-            type WriteHalf = tokio::net::tcp::WriteHalf<'split>;
-            fn split(&'split mut self) -> (Self::ReadHalf, Self::WriteHalf) {
-                <tokio::net::TcpStream>::split(self)
-            }
-        }
-        #[cfg(feature="rt_glommio")]
+    #[cfg(any(feature="rt_smol", feature="rt_glommio"))]
+    const _: (/* futures-io users */) = {
         impl<'split, T: AsyncRead + AsyncWrite + Unpin + 'split> Splitable<'split> for T {
-            type ReadHalf  = futures_util::io::ReadHalf <&'split mut T>;
+            type ReadHalf  = futures_util::io::ReadHalf<&'split mut T>;
             type WriteHalf = futures_util::io::WriteHalf<&'split mut T>;
             fn split(&'split mut self) -> (Self::ReadHalf, Self::WriteHalf) {
                 AsyncRead::split(self)
             }
         }
     };
-    #[cfg(feature="__clone__")]
-    const _: () = {
-        impl<'split, C: AsyncRead + AsyncWrite + Unpin + Sized + Clone + 'split> Splitable<'split> for C {
-            type ReadHalf = Self;
-            type WriteHalf = &'split mut Self;
+
+    #[cfg(any(feature="rt_tokio"))]
+    const _: (/* tokio::io users */) = {
+        impl<'split, T: AsyncRead + AsyncWrite + Unpin + 'split> Splitable<'split> for T {
+            type ReadHalf  = TokioIoReadHalf<'split, T>;
+            type WriteHalf = TokioIoWriteHalf<'split, T>;
             fn split(&'split mut self) -> (Self::ReadHalf, Self::WriteHalf) {
-                (self.clone(), self)
+                let (r, w) = futures_util::lock::BiLock::new(self);
+                (TokioIoReadHalf(r), TokioIoWriteHalf(w))
+            }
+        }
+        
+        /*
+         * based on https://github.com/rust-lang/futures-rs/blob/de9274e655b2fff8c9630a259a473b71a6b79dda/futures-util/src/io/split.rs
+         */
+        pub struct TokioIoReadHalf<'split, T>(futures_util::lock::BiLock<&'split mut T>);
+        pub struct TokioIoWriteHalf<'split, T>(futures_util::lock::BiLock<&'split mut T>);
+        fn lock_and_then<T, U, E>(
+            lock: &futures_util::lock::BiLock<T>,
+            cx: &mut std::task::Context<'_>,
+            f: impl FnOnce(std::pin::Pin<&mut T>, &mut std::task::Context<'_>) -> std::task::Poll<Result<U, E>>
+        ) -> std::task::Poll<Result<U, E>> {
+            let mut l = futures_util::ready!(lock.poll_lock(cx));
+            f(l.as_pin_mut(), cx)
+        }
+        impl<'split, T: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for TokioIoReadHalf<'split, T> {
+            #[inline]
+            fn poll_read(
+                self: std::pin::Pin<&mut Self>, 
+                cx: &mut std::task::Context<'_>, 
+                buf: &mut tokio::io::ReadBuf<'_>
+            ) -> std::task::Poll<std::io::Result<()>> {
+                lock_and_then(&self.0, cx, |l, cx| l.poll_read(cx, buf))
+            }
+        }
+        impl<'split, T: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for TokioIoWriteHalf<'split, T> {
+            #[inline]
+            fn poll_write(
+                self: std::pin::Pin<&mut Self>, 
+                cx: &mut std::task::Context<'_>, 
+                buf: &[u8]
+            ) -> std::task::Poll<std::io::Result<usize>> {
+                lock_and_then(&self.0, cx, |l, cx| l.poll_write(cx, buf))
+            }
+            #[inline]
+            fn poll_flush(
+                self: std::pin::Pin<&mut Self>, 
+                cx: &mut std::task::Context<'_>
+            ) -> std::task::Poll<std::io::Result<()>> {
+                lock_and_then(&self.0, cx, |l, cx| l.poll_flush(cx))
+            }    
+            fn poll_shutdown(
+                self: std::pin::Pin<&mut Self>, 
+                cx: &mut std::task::Context<'_>
+            ) -> std::task::Poll<std::io::Result<()>> {
+                lock_and_then(&self.0, cx, |l, cx| l.poll_shutdown(cx))
             }
         }
     };
-
+    
     pub struct ReadHalf<C: AsyncRead + Unpin> {
         __closed__: Arc<RwLock<bool>>,
         conn:   C,
@@ -433,3 +473,15 @@ pub mod split {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    
+    #[cfg(feature="__runtime__")]
+    #[test]
+    fn test_impl_splitable() {
+        fn assert_impl_splitable<T: split::Splitable<'static>>() {}
+        assert_impl_splitable::<crate::runtime::net::TcpStream>();
+    }
+}

src/lib.rs

@@ -32,34 +32,40 @@
 ))]
 compile_error! {"More than one runtime feature flags can't be activated"}
 
-#[cfg(feature="__runtime__")]
+#[cfg(feature="rt_tokio")]
 mod runtime {
-    #[cfg(feature="rt_tokio")]
     pub use {
         tokio::io::AsyncReadExt as AsyncRead,
         tokio::io::AsyncWriteExt as AsyncWrite,
         tokio::sync::RwLock,
         tokio::time::sleep
     };
-
-    #[cfg(feature="rt_smol")]
+    #[cfg(test)]
+    pub use tokio::net;
+}
+#[cfg(feature="rt_smol")]
+mod runtime {
     pub use {
-        smol::io::AsyncReadExt as AsyncRead,
-        smol::io::AsyncWriteExt as AsyncWrite,
+        futures_util::AsyncReadExt as AsyncRead,
+        futures_util::AsyncWriteExt as AsyncWrite,
         smol::lock::RwLock,
     };
-    #[cfg(feature="rt_smol")]
     pub async fn sleep(duration: std::time::Duration) {
         smol::Timer::after(duration).await;
     }
-
-    #[cfg(feature="rt_glommio")]
+    #[cfg(test)]
+    pub use smol::net;
+}
+#[cfg(feature="rt_glommio")]
+mod runtime {
     pub use {
         futures_util::AsyncReadExt as AsyncRead,
         futures_util::AsyncWriteExt as AsyncWrite,
         glommio::sync::RwLock,
         glommio::timer::sleep
     };
+    #[cfg(test)]
+    pub use glommio::net;
 }
 
 pub mod message;