transports/tcp/: Call take_error on tokio TcpStream (#2725)

See e04e95a for the rational.

With tokio `v1.19.0` released, `TcStream` exposes `take_error`.

This commit applies the same fix from e04e95a to the tokio TCP provider.

transports/tcp/CHANGELOG.md

@@ -2,6 +2,10 @@
 
 - Update to `libp2p-core` `v0.34.0`.
 
+- Call `TcpStream::take_error` in tokio `Provider` to report connection
+  establishment errors early. See also [PR 2458] for the related async-io
+  change.
+
 # 0.33.0
 
 - Update to `libp2p-core` `v0.33.0`.
@@ -16,7 +20,9 @@
 
 # 0.31.1 [2022-02-02]
 
-- Call `TcpSocket::take_error` to report connection establishment errors early.
+- Call `TcpSocket::take_error` to report connection establishment errors early. See [PR 2458].
+
+[PR 2458]: https://github.com/libp2p/rust-libp2p/pull/2458
 
 # 0.31.0 [2022-01-27]
 

transports/tcp/Cargo.toml

@@ -21,7 +21,7 @@ libc = "0.2.80"
 libp2p-core = { version = "0.34.0", path = "../../core", default-features = false  }
 log = "0.4.11"
 socket2 = { version = "0.4.0", features = ["all"] }
-tokio-crate = { package = "tokio", version = "1.0.1", default-features = false, features = ["net"], optional = true }
+tokio-crate = { package = "tokio", version = "1.19.0", default-features = false, features = ["net"], optional = true }
 
 [features]
 default = ["async-io"]

transports/tcp/src/provider/tokio.rs

@@ -64,8 +64,22 @@ impl Provider for Tcp {
 
     fn new_stream(s: net::TcpStream) -> BoxFuture<'static, io::Result<Self::Stream>> {
         async move {
+            // Taken from [`tokio_crate::net::TcpStream::connect_mio`].
+
             let stream = tokio_crate::net::TcpStream::try_from(s)?;
+
+            // Once we've connected, wait for the stream to be writable as
+            // that's when the actual connection has been initiated. Once we're
+            // writable we check for `take_socket_error` to see if the connect
+            // actually hit an error or not.
+            //
+            // If all that succeeded then we ship everything on up.
             stream.writable().await?;
+
+            if let Some(e) = stream.take_error()? {
+                return Err(e);
+            }
+
             Ok(TcpStream(stream))
         }
         .boxed()