refactor(h2): refactor legacy unsafe wrapper code

src/core/client/dispatch.rs

@@ -264,6 +264,7 @@ pin_project! {
 impl<B> Future for SendWhen<B>
 where
     B: Body + 'static,
+    B::Data: Send,
 {
     type Output = ();
 

src/core/client/proto/h2/client.rs

@@ -34,7 +34,7 @@ use crate::core::{
         body::{self, Incoming as IncomingBody},
         bounds::Http2ClientConnExec,
         dispatch::{self, Callback, SendWhen, TrySendError},
-        proto::{Dispatched, h2::UpgradedSendStream, headers},
+        proto::{Dispatched, headers},
         upgrade::{self, Upgraded},
     },
     common::{io::Compat, time::Time},
@@ -295,7 +295,8 @@ pin_project! {
 
 impl<B, T> Future for H2ClientFuture<B, T>
 where
-    B: http_body::Body + 'static,
+    B: Body + 'static,
+    B::Data: Send,
     B::Error: Into<BoxError>,
     T: Read + Write + Unpin,
 {
@@ -454,6 +455,7 @@ pin_project! {
 impl<B> Future for ResponseFutMap<B>
 where
     B: Body + 'static,
+    B::Data: Send,
 {
     type Output = Result<Response<body::Incoming>, (Error, Option<Request<B>>)>;
 
@@ -487,7 +489,7 @@ where
                     let (pending, on_upgrade) = upgrade::pending();
                     let io = H2Upgraded {
                         ping,
-                        send_stream: unsafe { UpgradedSendStream::new(send_stream) },
+                        send_stream,
                         recv_stream,
                         buf: Bytes::new(),
                     };

src/core/client/proto/h2/mod.rs

@@ -4,7 +4,6 @@ pub(crate) mod ping;
 use std::{
     future::Future,
     io::{Cursor, IoSlice},
-    mem,
     pin::Pin,
     task::{Context, Poll, ready},
 };
@@ -263,7 +262,7 @@ where
     B: Buf,
 {
     ping: Recorder,
-    send_stream: UpgradedSendStream<B>,
+    send_stream: SendStream<SendBuf<B>>,
     recv_stream: RecvStream,
     buf: Bytes,
 }
@@ -328,7 +327,7 @@ where
             None => Some(0),
             Some(Ok(cnt)) => self
                 .send_stream
-                .write(&buf[..cnt], false)
+                .send_data(SendBuf::Cursor(Cursor::new(buf[..cnt].into())), false)
                 .ok()
                 .map(|()| cnt),
             Some(Err(_)) => None,
@@ -357,7 +356,11 @@ where
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<(), std::io::Error>> {
-        if self.send_stream.write(&[], true).is_ok() {
+        if self
+            .send_stream
+            .send_data(SendBuf::Cursor(Cursor::new([].into())), true)
+            .is_ok()
+        {
             return Poll::Ready(Ok(()));
         }
 
@@ -381,65 +384,3 @@ fn h2_to_io_error(e: http2::Error) -> std::io::Error {
         std::io::Error::other(e)
     }
 }
-
-struct UpgradedSendStream<B>(SendStream<SendBuf<Neutered<B>>>);
-
-impl<B> UpgradedSendStream<B>
-where
-    B: Buf,
-{
-    unsafe fn new(inner: SendStream<SendBuf<B>>) -> Self {
-        assert_eq!(mem::size_of::<B>(), mem::size_of::<Neutered<B>>());
-        #[allow(clippy::missing_transmute_annotations)]
-        Self(unsafe { mem::transmute(inner) })
-    }
-
-    fn reserve_capacity(&mut self, cnt: usize) {
-        unsafe { self.as_inner_unchecked().reserve_capacity(cnt) }
-    }
-
-    fn poll_capacity(&mut self, cx: &mut Context<'_>) -> Poll<Option<Result<usize, http2::Error>>> {
-        unsafe { self.as_inner_unchecked().poll_capacity(cx) }
-    }
-
-    fn poll_reset(&mut self, cx: &mut Context<'_>) -> Poll<Result<http2::Reason, http2::Error>> {
-        unsafe { self.as_inner_unchecked().poll_reset(cx) }
-    }
-
-    fn write(&mut self, buf: &[u8], end_of_stream: bool) -> Result<(), std::io::Error> {
-        let send_buf = SendBuf::Cursor(Cursor::new(buf.into()));
-        unsafe {
-            self.as_inner_unchecked()
-                .send_data(send_buf, end_of_stream)
-                .map_err(h2_to_io_error)
-        }
-    }
-
-    unsafe fn as_inner_unchecked(&mut self) -> &mut SendStream<SendBuf<B>> {
-        unsafe { &mut *(&mut self.0 as *mut _ as *mut _) }
-    }
-}
-
-#[repr(transparent)]
-struct Neutered<B> {
-    _inner: B,
-    impossible: Impossible,
-}
-
-enum Impossible {}
-
-unsafe impl<B> Send for Neutered<B> {}
-
-impl<B> Buf for Neutered<B> {
-    fn remaining(&self) -> usize {
-        match self.impossible {}
-    }
-
-    fn chunk(&self) -> &[u8] {
-        match self.impossible {}
-    }
-
-    fn advance(&mut self, _cnt: usize) {
-        match self.impossible {}
-    }
-}