Added the ability to explicitly set type names and some QoS settings for Mqtt -> ROS

Author: JayHerpin

mqtt_client/config/params.ros2.fixed.yaml

@@ -0,0 +1,20 @@
+/**/*:
+  ros__parameters:
+    broker:
+      host: localhost
+      port: 1883
+    bridge:
+      ros2mqtt:
+        ros_topics:
+          - /ping/primitive
+        /ping/primitive:
+          mqtt_topic: pingpong/primitive
+          primitive: true
+          ros_type: std_msgs/msg/Bool
+      mqtt2ros:
+        mqtt_topics:
+          - pingpong/primitive
+        pingpong/primitive:
+          ros_topic: /pong/primitive
+          primitive: true
+          ros_type: std_msgs/msg/Bool

mqtt_client/include/mqtt_client/MqttClient.ros2.hpp

@@ -34,12 +34,13 @@ SOFTWARE.
 
 #define FMT_HEADER_ONLY
 #include <fmt/format.h>
+#include <mqtt/async_client.h>
 #include <mqtt_client_interfaces/srv/is_connected.hpp>
 #include <mqtt_client_interfaces/srv/new_mqtt2_ros_bridge.hpp>
 #include <mqtt_client_interfaces/srv/new_ros2_mqtt_bridge.hpp>
-#include <mqtt/async_client.h>
 #include <rclcpp/rclcpp.hpp>
 #include <rclcpp/serialization.hpp>
+#include <rclcpp/qos.hpp>
 #include <std_msgs/msg/float64.hpp>
 
 
@@ -98,8 +99,7 @@ class MqttClient : public rclcpp::Node,
    * @return  true         if parameter was successfully retrieved
    * @return  false        if parameter was not found or default was used
    */
-  bool loadParameter(const std::string& key, std::string& value,
-                     const std::string& default_value);
+  bool loadParameter(const std::string& key, std::string& value, const std::string& default_value);
 
   /**
    * @brief Loads requested ROS parameter from parameter server.
@@ -242,6 +242,15 @@ class MqttClient : public rclcpp::Node,
    */
   void mqtt2primitive(mqtt::const_message_ptr mqtt_msg);
 
+  /**
+   * @brief Publishes a primitive message received via MQTT to ROS.
+   *
+   * Currently not implemented.
+   *
+   * @param   mqtt_msg     MQTT message
+   */
+  void mqtt2fixed(mqtt::const_message_ptr mqtt_msg);
+
   /**
    * @brief Callback for when the client has successfully connected to the
    * broker.
@@ -351,10 +360,9 @@ class MqttClient : public rclcpp::Node,
     std::string user;  ///< username
     std::string pass;  ///< password
     struct {
-      bool enabled;  ///< whether to connect via SSL/TLS
-      std::filesystem::path
-        ca_certificate;  ///< public CA certificate trusted by client
-    } tls;               ///< SSL/TLS-related variables
+      bool enabled;                          ///< whether to connect via SSL/TLS
+      std::filesystem::path ca_certificate;  ///< public CA certificate trusted by client
+    } tls;                                   ///< SSL/TLS-related variables
   };
 
   /**
@@ -377,14 +385,14 @@ class MqttClient : public rclcpp::Node,
     double keep_alive_interval;  ///< keep-alive interval
     int max_inflight;            ///< maximum number of inflight messages
     struct {
-      std::filesystem::path certificate;    ///< client certificate
-      std::filesystem::path key;            ///< client private keyfile
-      std::string password;                 ///< decryption password for private key
-      int version;                          ///< TLS version (https://github.com/eclipse/paho.mqtt.cpp/blob/master/src/mqtt/ssl_options.h#L305)
-      bool verify;                          ///< Verify the client should conduct
-                                            ///< post-connect checks
-      std::vector<std::string> alpn_protos; ///< list of ALPN protocols
-    } tls;                   ///< SSL/TLS-related variables
+      std::filesystem::path certificate;     ///< client certificate
+      std::filesystem::path key;             ///< client private keyfile
+      std::string password;                  ///< decryption password for private key
+      int version;                           ///< TLS version (https://github.com/eclipse/paho.mqtt.cpp/blob/master/src/mqtt/ssl_options.h#L305)
+      bool verify;                           ///< Verify the client should conduct
+                                             ///< post-connect checks
+      std::vector<std::string> alpn_protos;  ///< list of ALPN protocols
+    } tls;                                   ///< SSL/TLS-related variables
   };
 
   /**
@@ -403,6 +411,7 @@ class MqttClient : public rclcpp::Node,
       int qos = 0;            ///< MQTT QoS value
       bool retained = false;  ///< whether to retain MQTT message
     } mqtt;                   ///< MQTT-related variables
+    bool fixed_type = false;  ///< whether the published message type is specified explicitly
     bool primitive = false;   ///< whether to publish as primitive message
     bool stamped = false;     ///< whether to inject timestamp in MQTT message
   };
@@ -412,18 +421,23 @@ class MqttClient : public rclcpp::Node,
    */
   struct Mqtt2RosInterface {
     struct {
-      int qos = 0;  ///< MQTT QoS value
-    } mqtt;         ///< MQTT-related variables
+      int qos = 0;      ///< MQTT QoS value
+    } mqtt;             ///< MQTT-related variables
     struct {
       std::string topic;     ///< ROS topic
       std::string msg_type;  ///< message type of publisher
       rclcpp::GenericPublisher::SharedPtr publisher;  ///< generic ROS publisher
       rclcpp::Publisher<std_msgs::msg::Float64>::SharedPtr
         latency_publisher;   ///< ROS publisher for latency
       int queue_size = 1;    ///< ROS publisher queue size
+      struct {
+        rclcpp::ReliabilityPolicy reliability = rclcpp::ReliabilityPolicy::SystemDefault;
+        rclcpp::DurabilityPolicy durability = rclcpp::DurabilityPolicy::SystemDefault;
+      } qos;
       bool latched = false;  ///< whether to latch ROS message
       bool is_stale = false; ///< whether a new generic publisher/subscriber is required
-    } ros;                   ///< ROS-related variables
+    } ros;      ///< ROS-related variables
+    bool fixed_type = false;   ///< whether the published ros message type is specified explicitly
     bool primitive = false;  ///< whether to publish as primitive message (if
                              ///< coming from non-ROS MQTT client)
     bool stamped = false;    ///< whether timestamp is injected
@@ -511,7 +525,8 @@ class MqttClient : public rclcpp::Node,
 
 
 template <typename T>
-bool MqttClient::loadParameter(const std::string& key, T& value) {
+bool MqttClient::loadParameter(const std::string& key, T& value)
+{
   bool found = get_parameter(key, value);
   if (found)
     RCLCPP_DEBUG(get_logger(), "Retrieved parameter '%s' = '%s'", key.c_str(),
@@ -522,7 +537,8 @@ bool MqttClient::loadParameter(const std::string& key, T& value) {
 
 template <typename T>
 bool MqttClient::loadParameter(const std::string& key, T& value,
-                               const T& default_value) {
+                               const T& default_value)
+{
   bool found = get_parameter_or(key, value, default_value);
   if (!found)
     RCLCPP_WARN(get_logger(), "Parameter '%s' not set, defaulting to '%s'",
@@ -540,7 +556,7 @@ bool MqttClient::loadParameter(const std::string& key, std::vector<T>& value)
   const bool found = get_parameter(key, value);
   if (found)
     RCLCPP_WARN(get_logger(), "Retrieved parameter '%s' = '[%s]'", key.c_str(),
-                  fmt::format("{}", fmt::join(value, ", ")).c_str());
+                fmt::format("{}", fmt::join(value, ", ")).c_str());
   return found;
 }
 
@@ -555,7 +571,7 @@ bool MqttClient::loadParameter(const std::string& key, std::vector<T>& value,
                 key.c_str(), fmt::format("{}", fmt::join(value, ", ")).c_str());
   if (found)
     RCLCPP_DEBUG(get_logger(), "Retrieved parameter '%s' = '%s'", key.c_str(),
-                  fmt::format("{}", fmt::join(value, ", ")).c_str());
+                 fmt::format("{}", fmt::join(value, ", ")).c_str());
   return found;
 }
 
@@ -570,7 +586,8 @@ bool MqttClient::loadParameter(const std::string& key, std::vector<T>& value,
  */
 template <typename T>
 void serializeRosMessage(const T& msg,
-                         rclcpp::SerializedMessage& serialized_msg) {
+                         rclcpp::SerializedMessage& serialized_msg)
+{
 
   rclcpp::Serialization<T> serializer;
   serializer.serialize_message(&msg, &serialized_msg);
@@ -587,7 +604,8 @@ void serializeRosMessage(const T& msg,
  */
 template <typename T>
 void deserializeRosMessage(const rclcpp::SerializedMessage& serialized_msg,
-                           T& msg) {
+                           T& msg)
+{
 
   rclcpp::Serialization<T> serializer;
   serializer.deserialize_message(&serialized_msg, &msg);