Reformat long lines in tutorial examples.

content/tokio/topics/bridging.md

@@ -79,7 +79,10 @@ pub struct BlockingClient {
 }
 
 impl BlockingClient {
-    pub fn connect<T: ToSocketAddrs>(addr: T) -> crate::Result<BlockingClient> {
+    pub fn connect<T>(addr: T) -> crate::Result<BlockingClient>
+    where
+        T: ToSocketAddrs,
+    {
         let rt = tokio::runtime::Builder::new_current_thread()
             .enable_all()
             .build()?;
@@ -135,7 +138,11 @@ impl BlockingClient {
         self.rt.block_on(self.inner.set_expires(key, value, expiration))
     }
 
-    pub fn publish(&mut self, channel: &str, message: Bytes) -> crate::Result<u64> {
+    pub fn publish(
+        &mut self,
+        channel: &str,
+        message: Bytes,
+    ) -> crate::Result<u64> {
         self.rt.block_on(self.inner.publish(channel, message))
     }
 }
@@ -160,7 +167,10 @@ pub struct BlockingSubscriber {
 }
 
 impl BlockingClient {
-    pub fn subscribe(self, channels: Vec<String>) -> crate::Result<BlockingSubscriber> {
+    pub fn subscribe(
+        self,
+        channels: Vec<String>,
+    ) -> crate::Result<BlockingSubscriber> {
         let subscriber = self.rt.block_on(self.inner.subscribe(channels))?;
         Ok(BlockingSubscriber {
             inner: subscriber,
@@ -178,11 +188,17 @@ impl BlockingSubscriber {
         self.rt.block_on(self.inner.next_message())
     }
 
-    pub fn subscribe(&mut self, channels: &[String]) -> crate::Result<()> {
+    pub fn subscribe(
+        &mut self,
+        channels: &[String],
+    ) -> crate::Result<()> {
         self.rt.block_on(self.inner.subscribe(channels))
     }
 
-    pub fn unsubscribe(&mut self, channels: &[String]) -> crate::Result<()> {
+    pub fn unsubscribe(
+        &mut self,
+        channels: &[String],
+    ) -> crate::Result<()> {
         self.rt.block_on(self.inner.unsubscribe(channels))
     }
 }

content/tokio/topics/shutdown.md

@@ -55,8 +55,8 @@ async fn main() {
 
     // ... spawn application as separate task ...
     //
-    // application uses shutdown_send in case a shutdown was issued from inside
-    // the application
+    // application uses shutdown_send in case a shutdown was issued
+    // from inside the application
 
     tokio::select! {
         _ = signal::ctrl_c() => {},
@@ -113,7 +113,8 @@ let task1_handle = tokio::spawn(async move {
 tokio::spawn(async move {
     tokio::time::sleep(std::time::Duration::from_millis(10)).await;
 
-    // Step 4: Cancel the original or cloned token to notify other tasks about shutting down gracefully
+    // Step 4: Cancel the original or cloned token to notify other
+    // tasks about shutting down gracefully
     token.cancel();
 });
 

content/tokio/topics/testing.md

@@ -105,7 +105,9 @@ Consider now the actual client handler task, especially the `where`-clause of th
 function signature:
 
 ```rust
-use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
+use tokio::io::{
+    AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader,
+};
 
 async fn handle_connection<Reader, Writer>(
     reader: Reader,
@@ -124,7 +126,9 @@ where
                 break Ok(());
             }
             writer
-                .write_all(format!("Thanks for your message.\r\n").as_bytes())
+                .write_all(
+                    format!("Thanks for your message.\r\n").as_bytes()
+                )
                 .await
                 .unwrap();
         }

content/tokio/tutorial/async.md

@@ -563,8 +563,9 @@ impl MiniTokio {
 
     /// Spawn a future onto the mini-tokio instance.
     ///
-    /// The given future is wrapped with the `Task` harness and pushed into the
-    /// `scheduled` queue. The future will be executed when `run` is called.
+    /// The given future is wrapped with the `Task` harness and
+    /// pushed into the `scheduled` queue. The future will be
+    /// executed when `run` is called.
     fn spawn<F>(&self, future: F)
     where
         F: Future<Output = ()> + Send + 'static,
@@ -574,18 +575,21 @@ impl MiniTokio {
 }
 
 impl TaskFuture {
-    fn new(future: impl Future<Output = ()> + Send + 'static) -> TaskFuture {
+    fn new<F>(future: F) -> TaskFuture
+    where
+        F: Future<Output = ()> + Send + 'static,
+    {
         TaskFuture {
             future: Box::pin(future),
             poll: Poll::Pending,
         }
     }
 
     fn poll(&mut self, cx: &mut Context<'_>) {
-        // Spurious wake-ups are allowed, even after a future has                                  
-        // returned `Ready`. However, polling a future which has                                   
-        // already returned `Ready` is *not* allowed. For this                                     
-        // reason we need to check that the future is still pending                                
+        // Spurious wake-ups are allowed, even after a future has
+        // returned `Ready`. However, polling a future which has
+        // already returned `Ready` is *not* allowed. For this
+        // reason we need to check that the future is still pending
         // before we call it. Failure to do so can lead to a panic.
         if self.poll.is_pending() {
             self.poll = self.future.as_mut().poll(cx);
@@ -609,9 +613,9 @@ impl Task {
 
     // Spawns a new task with the given future.
     //
-    // Initializes a new Task harness containing the given future and pushes it
-    // onto `sender`. The receiver half of the channel will get the task and
-    // execute it.
+    // Initializes a new Task harness containing the given future and
+    // pushes it onto `sender`. The receiver half of the channel will
+    // get the task and execute it.
     fn spawn<F>(future: F, sender: &mpsc::Sender<Arc<Task>>)
     where
         F: Future<Output = ()> + Send + 'static,
@@ -731,24 +735,29 @@ struct Delay {
 impl Future for Delay {
     type Output = ();
 
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
-        // Check the current instant. If the duration has elapsed, then
-        // this future has completed so we return `Poll::Ready`.
+    fn poll(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+    ) -> Poll<()> {
+        // Check the current instant. If the duration has elapsed,
+        // then this future has completed so we return `Poll::Ready`.
         if Instant::now() >= self.when {
             return Poll::Ready(());
         }
 
-        // The duration has not elapsed. If this is the first time the future
-        // is called, spawn the timer thread. If the timer thread is already
-        // running, ensure the stored `Waker` matches the current task's waker.
+        // The duration has not elapsed. If this is the first time
+        // the future is called, spawn the timer thread. If the timer
+        // thread is already running, ensure the stored `Waker`
+        // matches the current task's waker.
         if let Some(waker) = &self.waker {
             let mut waker = waker.lock().unwrap();
 
-            // Check if the stored waker matches the current task's waker.
-            // This is necessary as the `Delay` future instance may move to
-            // a different task between calls to `poll`. If this happens, the
-            // waker contained by the given `Context` will differ and we
-            // must update our stored waker to reflect this change.
+            // Check if the stored waker matches the current task's
+            // waker. This is necessary as the `Delay` future
+            // instance may move to a different task between calls to
+            // `poll`. If this happens, the waker contained by the
+            // given `Context` will differ and we must update our
+            // stored waker to reflect this change.
             if !waker.will_wake(cx.waker()) {
                 *waker = cx.waker().clone();
             }
@@ -757,33 +766,34 @@ impl Future for Delay {
             let waker = Arc::new(Mutex::new(cx.waker().clone()));
             self.waker = Some(waker.clone());
 
-            // This is the first time `poll` is called, spawn the timer thread.
+            // This is the first time `poll` is called, spawn the
+            // timer thread.
             thread::spawn(move || {
                 let now = Instant::now();
 
                 if now < when {
                     thread::sleep(when - now);
                 }
 
-                // The duration has elapsed. Notify the caller by invoking
-                // the waker.
+                // The duration has elapsed. Notify the caller by
+                // invoking the waker.
                 let waker = waker.lock().unwrap();
                 waker.wake_by_ref();
             });
         }
 
-        // By now, the waker is stored and the timer thread is started.
-        // The duration has not elapsed (recall that we checked for this
-        // first thing), ergo the future has not completed so we must
-        // return `Poll::Pending`.
+        // By now, the waker is stored and the timer thread is
+        // started. The duration has not elapsed (recall that we
+        // checked for this first thing), ergo the future has not
+        // completed so we must return `Poll::Pending`.
         //
-        // The `Future` trait contract requires that when `Pending` is
-        // returned, the future ensures that the given waker is signalled
-        // once the future should be polled again. In our case, by
-        // returning `Pending` here, we are promising that we will
-        // invoke the given waker included in the `Context` argument
-        // once the requested duration has elapsed. We ensure this by
-        // spawning the timer thread above.
+        // The `Future` trait contract requires that when `Pending`
+        // is returned, the future ensures that the given waker is
+        // signalled once the future should be polled again. In our
+        // case, by returning `Pending` here, we are promising that
+        // we will invoke the given waker included in the `Context`
+        // argument once the requested duration has elapsed. We
+        // ensure this by spawning the timer thread above.
         //
         // If we forget to invoke the waker, the task will hang
         // indefinitely.

content/tokio/tutorial/io.md

@@ -88,7 +88,8 @@ use tokio::fs::File;
 async fn main() -> io::Result<()> {
     let mut file = File::create("foo.txt").await?;
 
-    // Writes some prefix of the byte string, but not necessarily all of it.
+    // Writes some prefix of the byte string, but not necessarily all
+    // of it.
     let n = file.write(b"some bytes").await?;
 
     println!("Wrote the first {} bytes of 'some bytes'.", n);
@@ -281,7 +282,7 @@ can use [`TcpStream::split`]. The task that processes the echo logic in the serv
 # fn dox(mut socket: TcpStream) {
 tokio::spawn(async move {
     let (mut rd, mut wr) = socket.split();
-    
+
     if io::copy(&mut rd, &mut wr).await.is_err() {
         eprintln!("failed to copy");
     }
@@ -316,20 +317,21 @@ async fn main() -> io::Result<()> {
 
             loop {
                 match socket.read(&mut buf).await {
-                    // Return value of `Ok(0)` signifies that the remote has
-                    // closed
+                    // Return value of `Ok(0)` signifies that the
+                    // remote has closed
                     Ok(0) => return,
                     Ok(n) => {
                         // Copy the data back to socket
                         if socket.write_all(&buf[..n]).await.is_err() {
-                            // Unexpected socket error. There isn't much we can
-                            // do here so just stop processing.
+                            // Unexpected socket error. There isn't
+                            // much we can do here, so just stop
+                            // processing.
                             return;
                         }
                     }
                     Err(_) => {
-                        // Unexpected socket error. There isn't much we can do
-                        // here so just stop processing.
+                        // Unexpected socket error. There isn't much
+                        // we can do here so just stop processing.
                         return;
                     }
                 }

content/tokio/tutorial/spawning.md

@@ -390,9 +390,10 @@ async fn process(socket: TcpStream) {
             }
             Get(cmd) => {
                 if let Some(value) = db.get(cmd.key()) {
-                    // `Frame::Bulk` expects data to be of type `Bytes`. This
-                    // type will be covered later in the tutorial. For now,
-                    // `&Vec<u8>` is converted to `Bytes` using `into()`.
+                    // `Frame::Bulk` expects data to be of type `Bytes`.
+                    // This type will be covered later in the
+                    // tutorial. For now, `&Vec<u8>` is converted to
+                    // `Bytes` using `into()`.
                     Frame::Bulk(value.clone().into())
                 } else {
                     Frame::Null