chore: remove BoxFuture's (non-breaking) (#3629)

* chore: reduce BoxFuture's when using recursion.

* remove BoxFuture's in WithSocket

* chore: better document previous changes

sqlx-core/src/net/socket/mod.rs

@@ -143,15 +143,18 @@ where
 pub trait WithSocket {
     type Output;
 
-    fn with_socket<S: Socket>(self, socket: S) -> Self::Output;
+    fn with_socket<S: Socket>(
+        self,
+        socket: S,
+    ) -> impl std::future::Future<Output = Self::Output> + Send;
 }
 
 pub struct SocketIntoBox;
 
 impl WithSocket for SocketIntoBox {
     type Output = Box<dyn Socket>;
 
-    fn with_socket<S: Socket>(self, socket: S) -> Self::Output {
+    async fn with_socket<S: Socket>(self, socket: S) -> Self::Output {
         Box::new(socket)
     }
 }
@@ -197,7 +200,7 @@ pub async fn connect_tcp<Ws: WithSocket>(
         let stream = TcpStream::connect((host, port)).await?;
         stream.set_nodelay(true)?;
 
-        return Ok(with_socket.with_socket(stream));
+        return Ok(with_socket.with_socket(stream).await);
     }
 
     #[cfg(feature = "_rt-async-std")]
@@ -217,7 +220,7 @@ pub async fn connect_tcp<Ws: WithSocket>(
                     Ok(s)
                 });
             match stream {
-                Ok(stream) => return Ok(with_socket.with_socket(stream)),
+                Ok(stream) => return Ok(with_socket.with_socket(stream).await),
                 Err(e) => last_err = Some(e),
             }
         }
@@ -255,7 +258,7 @@ pub async fn connect_uds<P: AsRef<Path>, Ws: WithSocket>(
 
             let stream = UnixStream::connect(path).await?;
 
-            return Ok(with_socket.with_socket(stream));
+            return Ok(with_socket.with_socket(stream).await);
         }
 
         #[cfg(feature = "_rt-async-std")]
@@ -265,7 +268,7 @@ pub async fn connect_uds<P: AsRef<Path>, Ws: WithSocket>(
 
             let stream = Async::<UnixStream>::connect(path).await?;
 
-            Ok(with_socket.with_socket(stream))
+            Ok(with_socket.with_socket(stream).await)
         }
 
         #[cfg(not(feature = "_rt-async-std"))]

sqlx-core/src/net/tls/mod.rs

@@ -75,10 +75,14 @@ where
     Ws: WithSocket,
 {
     #[cfg(feature = "_tls-native-tls")]
-    return Ok(with_socket.with_socket(tls_native_tls::handshake(socket, config).await?));
+    return Ok(with_socket
+        .with_socket(tls_native_tls::handshake(socket, config).await?)
+        .await);
 
     #[cfg(all(feature = "_tls-rustls", not(feature = "_tls-native-tls")))]
-    return Ok(with_socket.with_socket(tls_rustls::handshake(socket, config).await?));
+    return Ok(with_socket
+        .with_socket(tls_rustls::handshake(socket, config).await?)
+        .await);
 
     #[cfg(not(any(feature = "_tls-native-tls", feature = "_tls-rustls")))]
     {

sqlx-mysql/src/connection/establish.rs

@@ -1,6 +1,5 @@
 use bytes::buf::Buf;
 use bytes::Bytes;
-use futures_core::future::BoxFuture;
 
 use crate::collation::{CharSet, Collation};
 use crate::common::StatementCache;
@@ -22,7 +21,7 @@ impl MySqlConnection {
             None => crate::net::connect_tcp(&options.host, options.port, do_handshake).await?,
         };
 
-        let stream = handshake.await?;
+        let stream = handshake?;
 
         Ok(Self {
             inner: Box::new(MySqlConnectionInner {
@@ -187,9 +186,9 @@ impl<'a> DoHandshake<'a> {
 }
 
 impl<'a> WithSocket for DoHandshake<'a> {
-    type Output = BoxFuture<'a, Result<MySqlStream, Error>>;
+    type Output = Result<MySqlStream, Error>;
 
-    fn with_socket<S: Socket>(self, socket: S) -> Self::Output {
-        Box::pin(self.do_handshake(socket))
+    async fn with_socket<S: Socket>(self, socket: S) -> Self::Output {
+        self.do_handshake(socket).await
     }
 }

sqlx-mysql/src/connection/tls.rs

@@ -94,7 +94,7 @@ pub(super) async fn maybe_upgrade<S: Socket>(
 impl WithSocket for MapStream {
     type Output = MySqlStream;
 
-    fn with_socket<S: Socket>(self, socket: S) -> Self::Output {
+    async fn with_socket<S: Socket>(self, socket: S) -> Self::Output {
         MySqlStream {
             socket: BufferedSocket::new(Box::new(socket)),
             server_version: self.server_version,