refactor(rumqttc): clean-up examples (#794)

Co-authored-by: swanandx <[email protected]>

Author: Devdutt Shenoi

Cargo.lock

@@ -54,7 +54,24 @@ color-backtrace = "0.6"
 matches = "0.1"
 pretty_assertions = "1"
 pretty_env_logger = "0.5"
-rustls = "0.21"
-rustls-native-certs = "0.6"
 serde = { version = "1", features = ["derive"] }
-tokio = { version = "1", features = ["full", "macros"] }
+
+[[example]]
+name = "tls"
+path = "examples/tls.rs"
+required-features = ["use-rustls"]
+
+[[example]]
+name = "tls2"
+path = "examples/tls2.rs"
+required-features = ["use-rustls"]
+
+[[example]]
+name = "websocket"
+path = "examples/websocket.rs"
+required-features = ["websocket"]
+
+[[example]]
+name = "websocket_proxy"
+path = "examples/websocket_proxy.rs"
+required-features = ["websocket", "proxy"]

rumqttc/Cargo.toml

@@ -1,6 +1,6 @@
 use tokio::{task, time};
 
-use rumqttc::{self, AsyncClient, Event, EventLoop, Incoming, MqttOptions, QoS};
+use rumqttc::{AsyncClient, Event, EventLoop, Incoming, MqttOptions, QoS};
 use std::error::Error;
 use std::time::Duration;
 
@@ -14,7 +14,7 @@ fn create_conn() -> (AsyncClient, EventLoop) {
     AsyncClient::new(mqttoptions, 10)
 }
 
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
 

rumqttc/examples/async_manual_acks.rs

@@ -1,4 +1,3 @@
-#![allow(dead_code, unused_imports)]
 use rumqttc::v5::mqttbytes::v5::Packet;
 use rumqttc::v5::mqttbytes::QoS;
 use tokio::{task, time};
@@ -17,7 +16,7 @@ fn create_conn() -> (AsyncClient, EventLoop) {
     AsyncClient::new(mqttoptions, 10)
 }
 
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     // todo!("fix this example with new way of spawning clients")
     pretty_env_logger::init();

rumqttc/examples/async_manual_acks_v5.rs

@@ -1,10 +1,10 @@
 use tokio::{task, time};
 
-use rumqttc::{self, AsyncClient, MqttOptions, QoS};
+use rumqttc::{AsyncClient, MqttOptions, QoS};
 use std::error::Error;
 use std::time::Duration;
 
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
     // color_backtrace::install();

rumqttc/examples/asyncpubsub.rs

@@ -5,7 +5,7 @@ use rumqttc::v5::{AsyncClient, MqttOptions};
 use std::error::Error;
 use std::time::Duration;
 
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
     // color_backtrace::install();

rumqttc/examples/asyncpubsub_v5.rs

@@ -51,8 +51,8 @@ fn main() {
     });
 
     // Iterate to poll the eventloop for connection progress
-    for notification in connection.iter() {
-        if let Ok(Event::Incoming(Incoming::Publish(packet))) = notification {
+    for notification in connection.iter().flatten() {
+        if let Event::Incoming(Incoming::Publish(packet)) = notification {
             match Message::try_from(packet.payload.as_ref()) {
                 Ok(message) => println!("Payload = {message:?}"),
                 Err(error) => println!("Error = {error}"),

rumqttc/examples/serde.rs

@@ -6,7 +6,7 @@ use rumqttc::v5::{AsyncClient, MqttOptions};
 use std::error::Error;
 use std::time::Duration;
 
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
 

rumqttc/examples/subscription_ids.rs

@@ -1,4 +1,4 @@
-use rumqttc::{self, Client, LastWill, MqttOptions, QoS};
+use rumqttc::{Client, LastWill, MqttOptions, QoS};
 use std::thread;
 use std::time::Duration;
 

rumqttc/examples/syncpubsub.rs

@@ -1,4 +1,4 @@
-use rumqttc::{self, Client, LastWill, MqttOptions, QoS};
+use rumqttc::{Client, LastWill, MqttOptions, QoS};
 use std::thread;
 use std::time::Duration;
 

rumqttc/examples/syncrecv.rs

@@ -1,12 +1,11 @@
 //! Example of how to configure rumqttd to connect to a server using TLS and authentication.
 use std::error::Error;
 
-#[cfg(feature = "use-rustls")]
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    use rumqttc::{self, AsyncClient, Event, Incoming, MqttOptions, Transport};
-    use tokio_rustls::rustls::ClientConfig;
+use rumqttc::{AsyncClient, Event, Incoming, MqttOptions, Transport};
+use tokio_rustls::rustls::ClientConfig;
 
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
     color_backtrace::install();
 
@@ -45,8 +44,3 @@ async fn main() -> Result<(), Box<dyn Error>> {
         }
     }
 }
-
-#[cfg(not(feature = "use-rustls"))]
-fn main() -> Result<(), Box<dyn Error>> {
-    panic!("Enable feature 'use-rustls'");
-}

rumqttc/examples/tls.rs

@@ -1,11 +1,10 @@
 //! Example of how to configure rumqttd to connect to a server using TLS and authentication.
 use std::error::Error;
 
-#[cfg(feature = "use-rustls")]
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error>> {
-    use rumqttc::{self, AsyncClient, MqttOptions, TlsConfiguration, Transport};
+use rumqttc::{AsyncClient, MqttOptions, TlsConfiguration, Transport};
 
+#[tokio::main(flavor = "current_thread")]
+async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
     color_backtrace::install();
 
@@ -44,8 +43,3 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
     Ok(())
 }
-
-#[cfg(not(feature = "use-rustls"))]
-fn main() -> Result<(), Box<dyn Error>> {
-    panic!("Enable feature 'use-rustls'");
-}

rumqttc/examples/tls2.rs

@@ -5,7 +5,7 @@ use rumqttc::v5::{AsyncClient, MqttOptions};
 use std::error::Error;
 use std::time::Duration;
 
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
 

rumqttc/examples/topic_alias.rs

@@ -1,12 +1,8 @@
-#[cfg(feature = "websocket")]
-use rumqttc::{self, AsyncClient, MqttOptions, QoS, Transport};
-#[cfg(feature = "websocket")]
+use rumqttc::{AsyncClient, MqttOptions, QoS, Transport};
 use std::{error::Error, time::Duration};
-#[cfg(feature = "websocket")]
 use tokio::{task, time};
 
-#[cfg(feature = "websocket")]
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     pretty_env_logger::init();
 
@@ -39,7 +35,6 @@ async fn main() -> Result<(), Box<dyn Error>> {
     }
 }
 
-#[cfg(feature = "websocket")]
 async fn requests(client: AsyncClient) {
     client
         .subscribe("hello/world", QoS::AtMostOnce)
@@ -57,8 +52,3 @@ async fn requests(client: AsyncClient) {
 
     time::sleep(Duration::from_secs(120)).await;
 }
-
-#[cfg(not(feature = "websocket"))]
-fn main() {
-    panic!("Enable websocket feature with `--features=websocket`");
-}

rumqttc/examples/websocket.rs

@@ -1,12 +1,8 @@
-#[cfg(all(feature = "websocket", feature = "proxy"))]
-use rumqttc::{self, AsyncClient, Proxy, ProxyAuth, ProxyType, QoS, Transport};
-#[cfg(all(feature = "websocket", feature = "proxy"))]
+use rumqttc::{AsyncClient, Proxy, ProxyAuth, ProxyType, QoS, Transport};
 use std::{error::Error, time::Duration};
-#[cfg(all(feature = "websocket", feature = "proxy"))]
 use tokio::{task, time};
 
-#[cfg(all(feature = "websocket", feature = "proxy"))]
-#[tokio::main(worker_threads = 1)]
+#[tokio::main(flavor = "current_thread")]
 async fn main() -> Result<(), Box<dyn Error>> {
     use rumqttc::MqttOptions;
 
@@ -48,7 +44,6 @@ async fn main() -> Result<(), Box<dyn Error>> {
     }
 }
 
-#[cfg(all(feature = "websocket", feature = "proxy"))]
 async fn requests(client: AsyncClient) {
     client
         .subscribe("hello/world", QoS::AtMostOnce)
@@ -66,8 +61,3 @@ async fn requests(client: AsyncClient) {
 
     time::sleep(Duration::from_secs(120)).await;
 }
-
-#[cfg(not(all(feature = "websocket", feature = "proxy")))]
-fn main() {
-    panic!("Enable websocket and proxy feature with `--features=websocket, proxy`");
-}

rumqttc/examples/websocket_proxy.rs

@@ -38,7 +38,7 @@
 //! use std::time::Duration;
 //! use std::error::Error;
 //!
-//! # #[tokio::main(worker_threads = 1)]
+//! # #[tokio::main(flavor = "current_thread")]
 //! # async fn main() {
 //! let mut mqttoptions = MqttOptions::new("rumqtt-async", "test.mosquitto.org", 1883);
 //! mqttoptions.set_keep_alive(Duration::from_secs(5));