chore: format code and enable rustfmt CI task (#1212)

22 files changed, 94 insertions, 60 deletions

diff --git a/azure-pipelines.yml b/azure-pipelines.yml
index be2b8443..2c397e60 100644
--- a/azure-pipelines.yml
+++ b/azure-pipelines.yml
@@ -5,10 +5,11 @@ variables:
   nightly: nightly-2019-06-10
 
 jobs:
-# # Check formatting
-# - template: ci/azure-rustfmt.yml
-#   parameters:
-#     name: rustfmt
+# Check formatting
+- template: ci/azure-rustfmt.yml
+  parameters:
+    rust: $(nightly)
+    name: rustfmt
 
 # Test top level crate
 - template: ci/azure-test-stable.yml
diff --git a/ci/azure-rustfmt.yml b/ci/azure-rustfmt.yml
index 0f50b6c2..7e01bfbb 100644
--- a/ci/azure-rustfmt.yml
+++ b/ci/azure-rustfmt.yml
@@ -7,7 +7,7 @@ jobs:
   steps:
     - template: azure-install-rust.yml
       parameters:
-        rust_version: stable
+        rust_version: ${{ parameters.rust }}
     - script: |
         rustup component add rustfmt
         cargo fmt --version
diff --git a/tokio-io/src/async_write.rs b/tokio-io/src/async_write.rs
index 98f9c849..1914b5a4 100644
--- a/tokio-io/src/async_write.rs
+++ b/tokio-io/src/async_write.rs
@@ -119,8 +119,7 @@ pub trait AsyncWrite {
     ///
     /// This function will panic if not called within the context of a future's
     /// task.
-    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>)
-        -> Poll<Result<(), io::Error>>;
+    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>>;
 
     /// Write a `Buf` into this value, returning how many bytes were written.
     ///
@@ -175,9 +174,11 @@ where
     P: DerefMut + Unpin,
     P::Target: AsyncWrite,
 {
-    fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8])
-        -> Poll<io::Result<usize>>
-    {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
         self.get_mut().as_mut().poll_write(cx, buf)
     }
 
diff --git a/tokio-io/tests/async_read.rs b/tokio-io/tests/async_read.rs
index 1e159588..bc5b2949 100644
--- a/tokio-io/tests/async_read.rs
+++ b/tokio-io/tests/async_read.rs
@@ -1,6 +1,6 @@
 use tokio_io::AsyncRead;
-use tokio_test::{assert_ready_ok, assert_ready_err};
 use tokio_test::task::MockTask;
+use tokio_test::{assert_ready_err, assert_ready_ok};
 
 use bytes::{BufMut, BytesMut};
 use pin_utils::pin_mut;
@@ -22,8 +22,8 @@ fn read_buf_success() {
         fn poll_read(
             self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
-            buf: &mut [u8]) -> Poll<io::Result<usize>>
-        {
+            buf: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
             buf[0..11].copy_from_slice(b"hello world");
             Poll::Ready(Ok(11))
         }
@@ -51,8 +51,8 @@ fn read_buf_error() {
         fn poll_read(
             self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
-            _buf: &mut [u8]) -> Poll<io::Result<usize>>
-        {
+            _buf: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
             let err = io::ErrorKind::Other.into();
             Poll::Ready(Err(err))
         }
@@ -78,8 +78,8 @@ fn read_buf_no_capacity() {
         fn poll_read(
             self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
-            _buf: &mut [u8]) -> Poll<io::Result<usize>>
-        {
+            _buf: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
             unimplemented!();
         }
     }
@@ -107,8 +107,8 @@ fn read_buf_no_uninitialized() {
         fn poll_read(
             self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
-            buf: &mut [u8]) -> Poll<io::Result<usize>>
-        {
+            buf: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
             for b in buf {
                 assert_eq!(0, *b);
             }
@@ -141,8 +141,8 @@ fn read_buf_uninitialized_ok() {
         fn poll_read(
             self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
-            buf: &mut [u8]) -> Poll<io::Result<usize>>
-        {
+            buf: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
             assert_eq!(buf[0..11], b"hello world"[..]);
             Poll::Ready(Ok(0))
         }
diff --git a/tokio-sync/src/mpsc/bounded.rs b/tokio-sync/src/mpsc/bounded.rs
index 7f8cb905..67134156 100644
--- a/tokio-sync/src/mpsc/bounded.rs
+++ b/tokio-sync/src/mpsc/bounded.rs
@@ -200,12 +200,10 @@ impl<T> async_sink::Sink<T> for Sender<T> {
     }
 
     fn start_send(mut self: Pin<&mut Self>, msg: T) -> Result<(), Self::Error> {
-        self.as_mut()
-            .try_send(msg)
-            .map_err(|err| {
-                assert!(err.is_full(), "call `poll_ready` before sending");
-                SendError(())
-            })
+        self.as_mut().try_send(msg).map_err(|err| {
+            assert!(err.is_full(), "call `poll_ready` before sending");
+            SendError(())
+        })
     }
 
     fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
diff --git a/tokio-tcp/src/listener.rs b/tokio-tcp/src/listener.rs
index 2f639a66..c469a38f 100644
--- a/tokio-tcp/src/listener.rs
+++ b/tokio-tcp/src/listener.rs
@@ -98,7 +98,10 @@ impl TcpListener {
     /// }
     /// # Ok::<_, Box<dyn std::error::Error>>(())
     /// ```
-    pub fn poll_accept(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<(TcpStream, SocketAddr)>> {
+    pub fn poll_accept(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<(TcpStream, SocketAddr)>> {
         let (io, addr) = ready!(self.poll_accept_std(cx))?;
 
         let io = mio::net::TcpStream::from_stream(io)?;
@@ -143,7 +146,10 @@ impl TcpListener {
     /// }
     /// # Ok::<_, Box<dyn std::error::Error>>(())
     /// ```
-    pub fn poll_accept_std(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<(net::TcpStream, SocketAddr)>> {
+    pub fn poll_accept_std(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<io::Result<(net::TcpStream, SocketAddr)>> {
         ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
 
         match self.io.get_ref().accept_std() {
diff --git a/tokio-tcp/src/stream.rs b/tokio-tcp/src/stream.rs
index 68ab150e..ada486ac 100644
--- a/tokio-tcp/src/stream.rs
+++ b/tokio-tcp/src/stream.rs
@@ -193,7 +193,11 @@ impl TcpStream {
     /// });
     /// # Ok::<_, Box<dyn std::error::Error>>(())
     /// ```
-    pub fn poll_read_ready(&self, cx: &mut Context<'_>, mask: mio::Ready) -> Poll<io::Result<mio::Ready>> {
+    pub fn poll_read_ready(
+        &self,
+        cx: &mut Context<'_>,
+        mask: mio::Ready,
+    ) -> Poll<io::Result<mio::Ready>> {
         self.io.poll_read_ready(cx, mask)
     }
 
@@ -729,11 +733,19 @@ impl AsyncRead for TcpStream {
         false
     }
 
-    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<io::Result<usize>> {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut [u8],
+    ) -> Poll<io::Result<usize>> {
         Pin::new(&mut self.io).poll_read(cx, buf)
     }
 
-    fn poll_read_buf<B: BufMut>(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut B) -> Poll<io::Result<usize>> {
+    fn poll_read_buf<B: BufMut>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
         ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
 
         let r = unsafe {
@@ -795,7 +807,11 @@ impl AsyncRead for TcpStream {
 }
 
 impl AsyncWrite for TcpStream {
-    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<io::Result<usize>> {
         Pin::new(&mut self.io).poll_write(cx, buf)
     }
 
@@ -809,7 +825,11 @@ impl AsyncWrite for TcpStream {
         Poll::Ready(Ok(()))
     }
 
-    fn poll_write_buf<B: Buf>(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut B) -> Poll<io::Result<usize>> {
+    fn poll_write_buf<B: Buf>(
+        self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut B,
+    ) -> Poll<io::Result<usize>> {
         ready!(self.io.poll_write_ready(cx))?;
 
         let r = {
diff --git a/tokio-test/tests/macros.rs b/tokio-test/tests/macros.rs
index 2dd519eb..ea3e5b5c 100644
--- a/tokio-test/tests/macros.rs
+++ b/tokio-test/tests/macros.rs
@@ -1,8 +1,8 @@
 #![cfg(feature = "broken")]
 #![deny(warnings, rust_2018_idioms)]
 
-use tokio_macros::{assert_ready, assert_not_ready, assert_ready_eq};
 use futures::{future, Async, Future, Poll};
+use tokio_macros::{assert_not_ready, assert_ready, assert_ready_eq};
 
 #[test]
 fn assert_ready() {
diff --git a/tokio-timer/src/delay.rs b/tokio-timer/src/delay.rs
index 47c4fab3..77534682 100644
--- a/tokio-timer/src/delay.rs
+++ b/tokio-timer/src/delay.rs
@@ -1,8 +1,8 @@
 use crate::timer::{HandlePriv, Registration};
 use std::future::Future;
 use std::pin::Pin;
-use std::time::{Duration, Instant};
 use std::task::{self, Poll};
+use std::time::{Duration, Instant};
 
 /// A future that completes at a specified instant in time.
 ///
diff --git a/tokio-timer/src/lib.rs b/tokio-timer/src/lib.rs
index 083d8d87..2d724d2b 100644
--- a/tokio-timer/src/lib.rs
+++ b/tokio-timer/src/lib.rs
@@ -32,12 +32,12 @@
 //! [`Timer`]: timer/struct.Timer.html
 
 macro_rules! ready {
-    ($e:expr) => (
+    ($e:expr) => {
         match $e {
             ::std::task::Poll::Ready(v) => v,
             ::std::task::Poll::Pending => return ::std::task::Poll::Pending,
         }
-    )
+    };
 }
 
 pub mod clock;
diff --git a/tokio-timer/src/throttle.rs b/tokio-timer/src/throttle.rs
index ab8733fd..a2654dd5 100644
--- a/tokio-timer/src/throttle.rs
+++ b/tokio-timer/src/throttle.rs
@@ -67,7 +67,10 @@ impl<T: Stream> Stream for Throttle<T> {
                 self.as_mut().get_unchecked_mut().has_delayed = true;
             }
 
-            let value = ready!(self.as_mut().map_unchecked_mut(|me| &mut me.stream).poll_next(cx));
+            let value = ready!(self
+                .as_mut()
+                .map_unchecked_mut(|me| &mut me.stream)
+                .poll_next(cx));
 
             if value.is_some() {
                 self.as_mut().get_unchecked_mut().delay.reset_timeout();
diff --git a/tokio-timer/src/timeout.rs b/tokio-timer/src/timeout.rs
index 42c0d9b1..d95f4a56 100644
--- a/tokio-timer/src/timeout.rs
+++ b/tokio-timer/src/timeout.rs
@@ -72,7 +72,6 @@ pub struct Timeout<T> {
     delay: Delay,
 }
 
-
 /// Error returned by `Timeout`.
 #[derive(Debug)]
 pub struct Elapsed(());
@@ -170,7 +169,7 @@ where
         unsafe {
             match self.map_unchecked_mut(|me| &mut me.delay).poll(cx) {
                 Poll::Ready(()) => Poll::Ready(Err(Elapsed(()))),
-                Poll::Pending => Poll::Pending
+                Poll::Pending => Poll::Pending,
             }
         }
     }
diff --git a/tokio-timer/src/timer/entry.rs b/tokio-timer/src/timer/entry.rs
index c1676e97..3cd0f815 100644
--- a/tokio-timer/src/timer/entry.rs
+++ b/tokio-timer/src/timer/entry.rs
@@ -290,7 +290,6 @@ impl Entry {
     }
 
     pub fn poll_elapsed(&self, cx: &mut task::Context<'_>) -> Poll<Result<(), Error>> {
-
         let mut curr = self.state.load(SeqCst);
 
         if is_elapsed(curr) {
diff --git a/tokio-timer/src/timer/handle.rs b/tokio-timer/src/timer/handle.rs
index 128723f3..0ac06491 100644
--- a/tokio-timer/src/timer/handle.rs
+++ b/tokio-timer/src/timer/handle.rs
@@ -3,7 +3,7 @@ use crate::{Delay, Error, /*Interval,*/ Timeout};
 use std::cell::RefCell;
 use std::fmt;
 use std::sync::{Arc, Weak};
-use std::time::{/*Duration,*/ Instant};
+use std::time::Instant;
 use tokio_executor::Enter;
 
 /// Handle to timer instance.
diff --git a/tokio/src/io/async_read_ext.rs b/tokio/src/io/async_read_ext.rs
index cca0906a..62b239a8 100644
--- a/tokio/src/io/async_read_ext.rs
+++ b/tokio/src/io/async_read_ext.rs
@@ -6,7 +6,6 @@ use tokio_io::{AsyncRead, AsyncWrite};
 
 /// An extension trait which adds utility methods to `AsyncRead` types.
 pub trait AsyncReadExt: AsyncRead {
-
     /// Copy all data from `self` into the provided `AsyncWrite`.
     ///
     /// The returned future will copy all the bytes read from `reader` into the
@@ -24,9 +23,9 @@ pub trait AsyncReadExt: AsyncRead {
     /// unimplemented!();
     /// ```
     fn copy<'a, W>(&'a mut self, dst: &'a mut W) -> Copy<'a, Self, W>
-        where
-            Self: Unpin,
-            W: AsyncWrite + Unpin + ?Sized,
+    where
+        Self: Unpin,
+        W: AsyncWrite + Unpin + ?Sized,
     {
         copy(self, dst)
     }
@@ -42,7 +41,8 @@ pub trait AsyncReadExt: AsyncRead {
     /// unimplemented!();
     /// ```
     fn read<'a>(&'a mut self, dst: &'a mut [u8]) -> Read<'a, Self>
-        where Self: Unpin,
+    where
+        Self: Unpin,
     {
         read(self, dst)
     }
@@ -55,7 +55,8 @@ pub trait AsyncReadExt: AsyncRead {
     /// unimplemented!();
     /// ```
     fn read_exact<'a>(&'a mut self, dst: &'a mut [u8]) -> ReadExact<'a, Self>
-        where Self: Unpin,
+    where
+        Self: Unpin,
     {
         read_exact(self, dst)
     }
diff --git a/tokio/src/io/async_write_ext.rs b/tokio/src/io/async_write_ext.rs
index 759dac99..ffb5794f 100644
--- a/tokio/src/io/async_write_ext.rs
+++ b/tokio/src/io/async_write_ext.rs
@@ -15,7 +15,8 @@ pub trait AsyncWriteExt: AsyncWrite {
     /// unimplemented!();
     /// ````
     fn write<'a>(&'a mut self, src: &'a [u8]) -> Write<'a, Self>
-        where Self: Unpin,
+    where
+        Self: Unpin,
     {
         write(self, src)
     }
diff --git a/tokio/src/io/mod.rs b/tokio/src/io/mod.rs
index 20730811..5fb4951c 100644
--- a/tokio/src/io/mod.rs
+++ b/tokio/src/io/mod.rs
@@ -40,8 +40,8 @@ mod async_read_ext;
 mod async_write_ext;
 mod copy;
 mod read;
-mod write;
 mod read_exact;
+mod write;
 
 pub use self::async_read_ext::AsyncReadExt;
 pub use self::async_write_ext::AsyncWriteExt;
diff --git a/tokio/src/io/read_exact.rs b/tokio/src/io/read_exact.rs
index c8b2f884..32320814 100644
--- a/tokio/src/io/read_exact.rs
+++ b/tokio/src/io/read_exact.rs
@@ -11,11 +11,15 @@ use tokio_io::AsyncRead;
 /// Created by the [`read_exact`] function.
 ///
 /// [`read_exact`]: fn.read_exact.html
-pub(crate) fn read_exact<'a, A>(reader: &'a mut A, buf: &'a mut[u8]) -> ReadExact<'a, A>
+pub(crate) fn read_exact<'a, A>(reader: &'a mut A, buf: &'a mut [u8]) -> ReadExact<'a, A>
 where
-    A: AsyncRead + Unpin + ?Sized
+    A: AsyncRead + Unpin + ?Sized,
 {
-    ReadExact { reader, buf, pos: 0 }
+    ReadExact {
+        reader,
+        buf,
+        pos: 0,
+    }
 }
 
 /// Creates a future which will read exactly enough bytes to fill `buf`,
@@ -29,7 +33,6 @@ pub struct ReadExact<'a, A: ?Sized> {
     pos: usize,
 }
 
-
 fn eof() -> io::Error {
     io::Error::new(io::ErrorKind::UnexpectedEof, "early eof")
 }
diff --git a/tokio/src/net.rs b/tokio/src/net.rs
index 872ec22a..896bca3d 100644
--- a/tokio/src/net.rs
+++ b/tokio/src/net.rs
@@ -58,7 +58,7 @@ pub mod udp {
     //! [`Send`]: struct.Send.html
     //! [`RecvFrom`]: struct.RecvFrom.html
     //! [`SendTo`]: struct.SendTo.html
-    pub use tokio_udp::{UdpSocket, Recv, Send, RecvFrom, SendTo};
+    pub use tokio_udp::{Recv, RecvFrom, Send, SendTo, UdpSocket};
 }
 #[cfg(feature = "udp")]
 pub use self::udp::UdpSocket;
diff --git a/tokio/tests/io_copy.rs b/tokio/tests/io_copy.rs
index 0179cfb7..2fac7e85 100644
--- a/tokio/tests/io_copy.rs
+++ b/tokio/tests/io_copy.rs
@@ -1,7 +1,7 @@
 #![deny(warnings, rust_2018_idioms)]
 #![feature(async_await)]
 
-use tokio::io::{AsyncRead, AsyncWrite, AsyncReadExt};
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
 use tokio_test::assert_ok;
 
 use bytes::BytesMut;
diff --git a/tokio/tests/io_read.rs b/tokio/tests/io_read.rs
index b1fc42c4..0ec2bef3 100644
--- a/tokio/tests/io_read.rs
+++ b/tokio/tests/io_read.rs
@@ -22,7 +22,7 @@ async fn read() {
             buf: &mut [u8],
         ) -> Poll<io::Result<usize>> {
             assert_eq!(0, self.poll_cnt);
-            self.poll_cnt +=1 ;
+            self.poll_cnt += 1;
 
             buf[0..11].copy_from_slice(b"hello world");
             Poll::Ready(Ok(11))
diff --git a/tokio/tests/io_read_exact.rs b/tokio/tests/io_read_exact.rs
index 1e3ab4ed..eae7435f 100644
--- a/tokio/tests/io_read_exact.rs
+++ b/tokio/tests/io_read_exact.rs
@@ -18,7 +18,7 @@ async fn read_exact() {
         fn poll_read(
             mut self: Pin<&mut Self>,
             _cx: &mut Context<'_>,
-            buf: &mut [u8]
+            buf: &mut [u8],
         ) -> Poll<io::Result<usize>> {
             let me = &mut *self;
             let len = buf.len();
@@ -29,7 +29,9 @@ async fn read_exact() {
     }
 
     let mut buf = Box::new([0; 8]);
-    let mut rd = Rd { val: b"hello world" };
+    let mut rd = Rd {
+        val: b"hello world",
+    };
 
     let n = assert_ok!(rd.read_exact(&mut buf[..]).await);
     assert_eq!(n, 8);