Added MQTT requests

Cargo.toml

@@ -34,6 +34,10 @@ reqwest-middleware = { version = "=0.5.1", features = ["rustls", "json", "multip
 reqwest-websocket = { version = "=0.6.0", features = ["json", "middleware"] }
 reqwest_cookie_store = "=0.10.0"
 cookie_store = "=0.22.0"
+## MQTT client, 3.1.1 and 5
+rumqttc = "=0.25.1"
+## Platform root certificates for MQTT over TLS
+rustls-native-certs = "=0.8.3"
 
 # TUI
 ## Terminal UI framework
@@ -88,7 +92,7 @@ digest_auth = "0.3.1"
 
 # Async
 ## Handle asynchronous requests
-tokio = { version = "=1.49.0", features = ["rt", "rt-multi-thread", "macros", "io-std", "io-util"] }
+tokio = { version = "=1.49.0", features = ["rt", "rt-multi-thread", "macros", "io-std", "io-util", "sync"] }
 tokio-util = { version = "=0.7.18", features = ["compat"] }
 futures-util = { version = "0.3.31", features = ["sink", "alloc"] }
 ## Smaller, faster and more flexible implementation of RwLock and Mutex. Used everywhere.

README.md

@@ -209,9 +209,9 @@ cargo run -- -h
 | - Ping                                                                         | :white_check_mark:                                                                  | :x:                  | :x:                      |
 | - Pong                                                                         | :white_check_mark:                                                                  | :x:                  | :x:                      |
 | - Close                                                                        | :white_check_mark:                                                                  | :x:                  | :x:                      |
+| **MQTT Client** (3.1.1 & 5, TCP & TLS)                                         | :white_check_mark:                                                                  | :white_check_mark:   | :x:                      |
 | **GraphQL**                                                                    | :x: :soon:                                                                          | :white_check_mark:   | :white_check_mark:       |
 | **gRPC**                                                                       | :x: :soon:                                                                          | :white_check_mark:   | :white_check_mark:       |
-| **MQTT**                                                                       | :x: :soon:                                                                          | :white_check_mark:   | :x:                      |
 | **Free**                                                                       | :white_check_mark:                                                                  | Depends              | Depends                  |
 | **Lightweight, fast and efficient**                                            | :white_check_mark:                                                                  | :x:                  | :x:                      |
 | **Data storage**                                                               | Your own committable, readable, versioned and retro-compatible files (JSON or YAML) | Tied to your account | Tied to your account     |
@@ -245,7 +245,6 @@ cargo run -- -h
 
 - **To add**
   - Maybe GraphQL requests
-  - Maybe MQTT requests
   - Maybe gRPC requests
 
 - **To improve**
@@ -309,6 +308,8 @@ You can read more about it here: https://github.com/Julien-cpsn/ATAC/releases/ta
 | [reqwest](https://github.com/seanmonstar/reqwest)                                | 0.13.2      | Send requests                                                                                             |
 | [reqwest-middleware](https://github.com/TrueLayer/reqwest-middleware)            | 0.5.1       | Wrapper around reqwest to allow for client middleware chains                                              |
 | [reqwest-websocket](https://github.com/jgraef/reqwest-websocket)                 | 0.6.0       | Wrapper around reqwest to handle web-socket requests                                                      |
+| [rumqttc](https://github.com/bytebeamio/rumqtt)                                  | 0.25.1      | MQTT 3.1.1 and 5 client                                                                                   |
+| [rustls-native-certs](https://github.com/rustls/rustls-native-certs)             | 0.8.3       | Load the platform root certificates for MQTT over TLS                                                     |
 | **TUI**                                                                          |             |                                                                                                           |
 | [ratatui](https://github.com/ratatui/ratatui)                                    | 0.30.0      | Terminal UI framework                                                                                     |
 | [crokey](https://github.com/Canop/crokey)                                        | 1.4.0       | Used to parse, use key bindings files and some utilities                                                  |

example_resources/collections/mqtt.json

@@ -0,0 +1,93 @@
+{
+  "name": "MQTT",
+  "last_position": null,
+  "requests": [
+    {
+      "name": "Test",
+      "url": "mqtts://test.mosquitto.org:8886",
+      "params": [],
+      "headers": [],
+      "auth": "no_auth",
+      "scripts": {
+        "pre_request_script": null,
+        "post_request_script": null
+      },
+      "settings": {
+        "use_config_proxy": true,
+        "allow_redirects": true,
+        "timeout": 30000,
+        "store_received_cookies": true,
+        "pretty_print_response_content": true,
+        "accept_invalid_certs": false,
+        "accept_invalid_hostnames": false
+      },
+      "protocol": {
+        "type": "mqtt",
+        "version": "5",
+        "client_id": "",
+        "clean_session": true,
+        "session_expiry_interval": 3600,
+        "keep_alive": 60,
+        "max_packet_size": 1048576,
+        "subscriptions": [
+          {
+            "enabled": true,
+            "topic": "atac/example/#",
+            "qos": "at_least_once"
+          }
+        ],
+        "publish": {
+          "topic": "atac/example/hello",
+          "qos": "at_least_once",
+          "retain": false
+        }
+      }
+    },
+    {
+      "name": "Sensors",
+      "url": "mqtt://broker.emqx.io",
+      "params": [],
+      "headers": [],
+      "auth": "no_auth",
+      "scripts": {
+        "pre_request_script": null,
+        "post_request_script": null
+      },
+      "settings": {
+        "use_config_proxy": true,
+        "allow_redirects": true,
+        "timeout": 30000,
+        "store_received_cookies": true,
+        "pretty_print_response_content": true,
+        "accept_invalid_certs": false,
+        "accept_invalid_hostnames": false
+      },
+      "protocol": {
+        "type": "mqtt",
+        "version": "3.1.1",
+        "client_id": "",
+        "clean_session": true,
+        "session_expiry_interval": 3600,
+        "keep_alive": 60,
+        "max_packet_size": 1048576,
+        "subscriptions": [
+          {
+            "enabled": true,
+            "topic": "atac/example/sensors/+",
+            "qos": "at_most_once"
+          },
+          {
+            "enabled": true,
+            "topic": "atac/example/alerts",
+            "qos": "exactly_once"
+          }
+        ],
+        "publish": {
+          "topic": "atac/example/sensors/temperature",
+          "qos": "at_most_once",
+          "retain": false
+        }
+      }
+    }
+  ]
+}

src/app/app.rs

@@ -125,6 +125,13 @@ pub struct App<'a> {
 
     pub message_text_area: TextInput,
 
+    /* MQTT */
+
+    /// Selected field in the MQTT connection and publish tabs
+    pub mqtt_form_selection: TextInputSelection,
+    pub mqtt_form_text_input: TextInput,
+    pub mqtt_subscriptions_table: StatefulCustomTable<'a>,
+
     /* Settings */
 
     pub request_settings_popup: SettingsPopup,
@@ -281,6 +288,20 @@ impl App<'_> {
 
             message_text_area: TextInput::new(None),
 
+            /* MQTT */
+
+            mqtt_form_selection: TextInputSelection::default(),
+            mqtt_form_text_input: TextInput::new(None),
+            mqtt_subscriptions_table: StatefulCustomTable::new(
+                vec![
+                    Line::default(),
+                    Line::from("No subscriptions").fg(THEME.read().ui.font_color),
+                    Line::from("(Add one with n)").fg(THEME.read().ui.secondary_foreground_color)
+                ],
+                "Topic",
+                "QoS (Enter to change)"
+            ),
+
             /* Settings */
 
             request_settings_popup: SettingsPopup::default(),

src/app/business_logic/collection.rs

@@ -7,6 +7,7 @@ use tracing::info;
 use crate::app::app::App;
 use crate::app::business_logic::collection::CollectionError::{CollectionNameAlreadyExists, CollectionNameIsEmpty};
 use crate::app::business_logic::collection::RequestError::RequestNameIsEmpty;
+use crate::app::business_logic::request::mqtt::send::mqtt_disconnect;
 use crate::cli::args::ARGS;
 use crate::models::collection::Collection;
 use crate::models::request::Request;
@@ -92,13 +93,20 @@ impl App<'_> {
         info!("Collection deleted");
 
         let collection = self.collections.remove(collection_index);
+
+        for request in &collection.requests {
+            disconnect_mqtt_request(request);
+        }
+
         self.delete_collection_file(collection);
     }
 
     pub fn delete_request(&mut self, collection_index: usize, request_index: usize) -> anyhow::Result<()> {
         info!("Request deleted");
 
-        self.collections[collection_index].requests.remove(request_index);
+        let request = self.collections[collection_index].requests.remove(request_index);
+        disconnect_mqtt_request(&request);
+
         self.save_collection_to_file(collection_index);
 
         Ok(())
@@ -166,6 +174,13 @@ impl App<'_> {
             info!("Request \"{}\" duplicated", selected_request.name);
 
             selected_request.name = format!("{} copy", selected_request.name);
+
+            // The copy does not share the broker connection
+            if let Ok(mqtt_request) = selected_request.get_mqtt_request_mut() {
+                mqtt_request.connection = None;
+                mqtt_request.is_connected = false;
+            }
+
             self.collections[collection_index].requests.insert(request_index + 1, Arc::new(RwLock::new(selected_request)));
         }
 
@@ -182,3 +197,10 @@ impl App<'_> {
         }
     }
 }
+
+/// A deleted request would otherwise keep its broker connection open in the background
+fn disconnect_mqtt_request(request: &Arc<RwLock<Request>>) {
+    if let Ok(mqtt_request) = request.write().get_mqtt_request_mut() {
+        mqtt_disconnect(mqtt_request);
+    }
+}