rumqttd: adding optional native-tls support.

Readjusting acceptor into the new task as well. That way this loop doesn't get blocked by
setting up the connection.

Author: jaredwolff

rumqttd/src/lib.rs

@@ -327,38 +327,52 @@ impl Server {
 
         let listener = TcpListener::bind(&addr).await?;
         let delay = Duration::from_millis(self.config.next_connection_delay_ms);
-        let mut count = 0;
+        let mut count: u32 = 0;
 
         let config = Arc::new(self.config.connections.clone());
         let acceptor = self.tls()?;
         let max_incoming_size = config.max_payload_size;
 
         info!("Waiting for connections on {}. Server = {}", addr, self.id);
         loop {
+            // Accept incoming connection
             let (stream, addr) = listener.accept().await?;
-            let network = match &acceptor {
-                Some(acceptor) => {
-                    info!("{}. Accepting TLS connection from: {}", count, addr);
-                    let sock = acceptor.accept(stream).await?;
-                    Network::new(sock, max_incoming_size)
-                }
-                None => {
-                    info!("{}. Accepting TCP connection from: {}", count, addr);
-                    Network::new(stream, max_incoming_size)
-                }
-            };
 
-            count += 1;
+            // Router tx needs to be outside
+            let router_tx = self.router_tx.clone();
 
+            // Acceptor cloned
+            let acceptor = acceptor.clone();
+
+            // Cloneconfig
             let config = config.clone();
-            let router_tx = self.router_tx.clone();
-            task::spawn(async {
+
+            // Then spawn a new thread to handle the connection
+            task::spawn(async move {
+                let network = match acceptor {
+                    Some(acceptor) => {
+                        info!("{}. Accepting TLS connection from: {}", count, addr);
+                        let sock = acceptor.accept(stream).await.unwrap();
+                        Network::new(sock, max_incoming_size)
+                    }
+                    None => {
+                        info!("{}. Accepting TCP connection from: {}", count, addr);
+                        Network::new(stream, max_incoming_size)
+                    }
+                };
+
+                let config = config.clone();
+
                 let connector = Connector::new(config, router_tx);
                 if let Err(e) = connector.new_connection(network).await {
                     error!("Dropping link task!! Result = {:?}", e);
                 }
             });
 
+            // Increment count
+            count += 1;
+
+            // Wait a certain amount between connection attempts.
             time::sleep(delay).await;
         }
     }