cargo fmt

benchmarks/clients/common.rs

@@ -1,6 +1,6 @@
-use pprof::{protos::Message, ProfilerGuard};
 use librumqttd::rumqttlog::router::ConnectionAck;
 use librumqttd::rumqttlog::{Connection, Event, Notification, Receiver, Sender};
+use pprof::{protos::Message, ProfilerGuard};
 use serde::{Deserialize, Serialize};
 use std::fs::File;
 use std::io::Write;

benchmarks/parsers/nats.rs

@@ -158,7 +158,7 @@ use std::{
 };
 
 /// A protocol operation sent by the server.
-/// 
+///
 /// TODO: remove dead_code once contents are used, added to silence clippy
 #[derive(Debug)]
 #[allow(dead_code)]

rumqttc/examples/async_manual_acks.rs

@@ -14,7 +14,6 @@ fn create_conn() -> (AsyncClient, EventLoop) {
     AsyncClient::new(mqttoptions, 10)
 }
 
-
 #[tokio::main(worker_threads = 1)]
 async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();

rumqttc/examples/syncpubsub.rs

@@ -7,7 +7,9 @@ fn main() {
 
     let mut mqttoptions = MqttOptions::new("test-1", "localhost", 1883);
     let will = LastWill::new("hello/world", "good bye", QoS::AtMostOnce, false);
-    mqttoptions.set_keep_alive(Duration::from_secs(5)).set_last_will(will);
+    mqttoptions
+        .set_keep_alive(Duration::from_secs(5))
+        .set_last_will(will);
 
     let (client, mut connection) = Client::new(mqttoptions, 10);
     thread::spawn(move || publish(client));

rumqttc/examples/tls2.rs

@@ -14,7 +14,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     // Dummies to prevent compilation error in CI
     let ca = vec![1, 2, 3];
     let client_cert = vec![1, 2, 3];
-    let client_key= vec![1, 2, 3];
+    let client_key = vec![1, 2, 3];
     //     let ca = include_bytes!("/home/tekjar/tlsfiles/ca.cert.pem");
     //     let client_cert = include_bytes!("/home/tekjar/tlsfiles/device-1.cert.pem");
     //     let client_key = include_bytes!("/home/tekjar/tlsfiles/device-1.key.pem");

rumqttc/src/eventloop.rs

@@ -2,11 +2,11 @@ use crate::{framed::Network, Transport};
 use crate::{tls, Incoming, MqttState, Packet, Request, StateError};
 use crate::{MqttOptions, Outgoing};
 
+use crate::mqttbytes;
+use crate::mqttbytes::v4::*;
 use async_channel::{bounded, Receiver, Sender};
 #[cfg(feature = "websocket")]
 use async_tungstenite::tokio::{connect_async, connect_async_with_tls_connector};
-use crate::mqttbytes::v4::*;
-use crate::mqttbytes;
 use tokio::net::TcpStream;
 #[cfg(unix)]
 use tokio::net::UnixStream;

rumqttc/src/lib.rs

@@ -106,18 +106,18 @@ use std::time::Duration;
 mod client;
 mod eventloop;
 mod framed;
+pub mod mqttbytes;
 mod state;
 mod tls;
-pub mod mqttbytes;
 
 pub use async_channel::{SendError, Sender, TrySendError};
 pub use client::{AsyncClient, Client, ClientError, Connection};
 pub use eventloop::{ConnectionError, Event, EventLoop};
 pub use mqttbytes::v4::*;
 pub use mqttbytes::*;
 pub use state::{MqttState, StateError};
-pub use tokio_rustls::rustls::ClientConfig;
 pub use tls::Error;
+pub use tokio_rustls::rustls::ClientConfig;
 
 pub type Incoming = Packet;
 
@@ -438,7 +438,11 @@ impl MqttOptions {
     }
 
     /// Username and password
-    pub fn set_credentials<U: Into<String>, P: Into<String>>(&mut self, username: U, password: P) -> &mut Self {
+    pub fn set_credentials<U: Into<String>, P: Into<String>>(
+        &mut self,
+        username: U,
+        password: P,
+    ) -> &mut Self {
         self.credentials = Some((username.into(), password.into()));
         self
     }
@@ -672,7 +676,7 @@ impl std::convert::TryFrom<url::Url> for MqttOptions {
             inflight,
             last_will: None,
             conn_timeout,
-            manual_acks: false
+            manual_acks: false,
         })
     }
 }

rumqttc/src/mqttbytes/v4/connack.rs

@@ -30,7 +30,7 @@ impl ConnAck {
 
     fn len(&self) -> usize {
         // sesssion present + code
-        
+
         1 + 1
     }
 

rumqttc/src/mqttbytes/v4/puback.rs

@@ -13,7 +13,7 @@ impl PubAck {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttc/src/mqttbytes/v4/pubcomp.rs

@@ -13,7 +13,7 @@ impl PubComp {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttc/src/mqttbytes/v4/pubrec.rs

@@ -13,7 +13,7 @@ impl PubRec {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttc/src/mqttbytes/v4/pubrel.rs

@@ -13,7 +13,7 @@ impl PubRel {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttc/src/mqttbytes/v4/unsuback.rs

@@ -1,6 +1,6 @@
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
-use crate::mqttbytes::{FixedHeader, Error, read_u16};
+use crate::mqttbytes::{read_u16, Error, FixedHeader};
 
 /// Acknowledgement to unsubscribe
 #[derive(Debug, Clone, PartialEq)]

rumqttc/src/state.rs

@@ -1,8 +1,8 @@
 use crate::{Event, Incoming, Outgoing, Request};
 
-use bytes::BytesMut;
 use crate::mqttbytes::v4::*;
 use crate::mqttbytes::{self, *};
+use bytes::BytesMut;
 use std::collections::VecDeque;
 use std::{io, mem, time::Instant};
 
@@ -100,7 +100,7 @@ impl MqttState {
             // TODO: Optimize these sizes later
             events: VecDeque::with_capacity(100),
             write: BytesMut::with_capacity(10 * 1024),
-            manual_acks
+            manual_acks,
         }
     }
 
@@ -487,9 +487,9 @@ impl MqttState {
 #[cfg(test)]
 mod test {
     use super::{MqttState, StateError};
-    use crate::{Event, Incoming, MqttOptions, Outgoing, Request};
     use crate::mqttbytes::v4::*;
     use crate::mqttbytes::*;
+    use crate::{Event, Incoming, MqttOptions, Outgoing, Request};
 
     fn build_outgoing_publish(qos: QoS) -> Publish {
         let topic = "hello/world".to_owned();

rumqttc/tests/broker.rs

@@ -100,7 +100,6 @@ impl Broker {
         .await
         .unwrap();
 
-
         self.incoming.pop_front().unwrap()
     }
 

rumqttd/src/async_locallink.rs

@@ -1,17 +1,17 @@
 use std::sync::Arc;
 
-use crate::{
-    consolelink::{self, ConsoleLink},
-    Config, Id, Server,
-};
-use futures_util::stream::{FuturesUnordered, StreamExt};
-use log::{trace, warn, error};
 use crate::mqttbytes::v4::*;
 use crate::mqttbytes::*;
 use crate::rumqttlog::{
     Connection, ConnectionAck, Data, Event, Notification, Receiver, RecvError, Router, SendError,
     Sender,
 };
+use crate::{
+    consolelink::{self, ConsoleLink},
+    Config, Id, Server,
+};
+use futures_util::stream::{FuturesUnordered, StreamExt};
+use log::{error, trace, warn};
 
 #[derive(Debug, thiserror::Error)]
 pub enum LinkError {

rumqttd/src/consolelink.rs

@@ -1,8 +1,8 @@
-use crate::Config;
 use crate::rumqttlog::ConnectionId;
 use crate::rumqttlog::{
     Connection, ConnectionAck, Event, MetricsReply, MetricsRequest, Notification, Receiver, Sender,
 };
+use crate::Config;
 use std::sync::Arc;
 use warp::Filter;
 

rumqttd/src/lib.rs

@@ -1,15 +1,15 @@
 use jackiechan::Sender;
+use log::{error, info};
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
 use std::{io, thread};
 use std::{net::SocketAddr, sync::Arc};
-use log::{error, info};
 
 mod mqttbytes;
 pub mod rumqttlog;
 
 use mqttbytes::v4::Packet;
-use rumqttlog::{RecvError, SendError, Event, Router, Disconnection};
+use rumqttlog::{Disconnection, Event, RecvError, Router, SendError};
 use tokio::time::error::Elapsed;
 
 use crate::remotelink::RemoteLink;

rumqttd/src/mqttbytes/v4/connack.rs

@@ -30,7 +30,7 @@ impl ConnAck {
 
     fn len(&self) -> usize {
         // sesssion present + code
-        
+
         1 + 1
     }
 

rumqttd/src/mqttbytes/v4/puback.rs

@@ -13,7 +13,7 @@ impl PubAck {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttd/src/mqttbytes/v4/pubcomp.rs

@@ -13,7 +13,7 @@ impl PubComp {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttd/src/mqttbytes/v4/pubrec.rs

@@ -13,7 +13,7 @@ impl PubRec {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttd/src/mqttbytes/v4/pubrel.rs

@@ -13,7 +13,7 @@ impl PubRel {
     }
 
     fn len(&self) -> usize {
-         // pkid
+        // pkid
         2
     }
 

rumqttd/src/mqttbytes/v4/suback.rs

@@ -15,7 +15,6 @@ impl SubAck {
     }
 
     pub fn len(&self) -> usize {
-
         2 + self.return_codes.len()
     }
 

rumqttd/src/mqttbytes/v4/unsuback.rs

@@ -1,6 +1,6 @@
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
-use crate::mqttbytes::{FixedHeader, Error, read_u16};
+use crate::mqttbytes::{read_u16, Error, FixedHeader};
 
 /// Acknowledgement to unsubscribe
 #[derive(Debug, Clone, PartialEq)]

rumqttd/src/mqttbytes/v4/unsubscribe.rs

@@ -35,10 +35,7 @@ impl Unsubscribe {
     }
 
     pub fn write(&self, payload: &mut BytesMut) -> Result<usize, Error> {
-        let remaining_len = 2 + self
-            .topics
-            .iter()
-            .fold(0, |s, topic| s + topic.len() + 2);
+        let remaining_len = 2 + self.topics.iter().fold(0, |s, topic| s + topic.len() + 2);
 
         payload.put_u8(0xA2);
         let remaining_len_bytes = write_remaining_length(payload, remaining_len)?;

rumqttd/src/network.rs

@@ -1,5 +1,5 @@
-use bytes::BytesMut;
 use crate::mqttbytes::{self, v4::*};
+use bytes::BytesMut;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 use crate::state;

rumqttd/src/remotelink.rs

@@ -1,15 +1,15 @@
-use crate::network::Network;
-use crate::state::{self, State};
-use crate::{network, ConnectionSettings, Id};
 use crate::mqttbytes::v4::*;
 use crate::mqttbytes::*;
+use crate::network::Network;
 use crate::rumqttlog::{
     Connection, ConnectionAck, Event, Notification, Receiver, RecvError, SendError, Sender,
 };
+use crate::state::{self, State};
+use crate::{network, ConnectionSettings, Id};
 
+use log::{debug, trace, warn};
 use std::sync::Arc;
 use std::{io, mem};
-use log::{debug, trace, warn};
 use tokio::time::{error::Elapsed, Duration};
 use tokio::{select, time};
 

rumqttd/src/rumqttlog/router/connection.rs

@@ -1,5 +1,5 @@
-use jackiechan::{bounded, Receiver, Sender, TrySendError};
 use crate::mqttbytes::v4::LastWill;
+use jackiechan::{bounded, Receiver, Sender, TrySendError};
 use std::fmt;
 
 use super::Notification;

rumqttd/src/rumqttlog/router/router.rs

@@ -6,7 +6,7 @@ use crate::rumqttlog::logs::{ConnectionsLog, DataLog, TopicsLog};
 use crate::rumqttlog::waiters::{DataWaiters, TopicsWaiters};
 use crate::rumqttlog::{Config, ConnectionId, RouterId};
 use jackiechan::{bounded, Receiver, RecvError, Sender, TryRecvError};
-use log::{info, error, trace, warn};
+use log::{error, info, trace, warn};
 use thiserror::Error;
 
 use super::connection::ConnectionType;

rumqttd/src/rumqttlog/router/tracker.rs

@@ -3,7 +3,7 @@ use crate::mqttbytes::*;
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, HashSet, VecDeque};
 
-use super::{Request, AcksRequest, DataRequest, TopicsRequest};
+use super::{AcksRequest, DataRequest, Request, TopicsRequest};
 
 /// Used to register a new connection with the router
 /// Connection messages encompasses a handle for router to

rumqttd/src/rumqttlog/waiters/mod.rs

@@ -1,6 +1,6 @@
 use std::collections::{HashMap, VecDeque};
 
-use super::{DataRequest, router::TopicsRequest};
+use super::{router::TopicsRequest, DataRequest};
 
 type Topic = String;
 type ConnectionId = usize;