Merge remote-tracking branch 'upstream/main' into ros2-add-multiple-topics

dmbuck32 · dmbuck32 · commit 786b4edb4101 · 2023-09-17T16:26:10.000+12:00
diff --git a/.github/workflows/docker-ros.yml b/.github/workflows/docker-ros.yml
@@ -14,7 +14,7 @@ jobs:
           command: roslaunch mqtt_client standalone.launch
           platform: amd64,arm64
           target: run
-  
+
   ros2:
     runs-on: ubuntu-latest
     steps:
@@ -26,4 +26,3 @@ jobs:
           platform: amd64,arm64
           target: run
           enable-push-as-latest: 'true'
-          
diff --git a/README.md b/README.md
@@ -254,6 +254,9 @@ client:
     certificate:          # client certificate file (only needed if broker requires client certificates; relative to ROS_HOME)
     key:                  # client private key file (relative to ROS_HOME)
     password:             # client private key password
+    version:              # TLS version (https://github.com/eclipse/paho.mqtt.cpp/blob/master/src/mqtt/ssl_options.h#L305)
+    verify:               # verify the client should conduct post-connect checks.
+    alpn_protos:          # list of ALPN protocols (https://www.openssl.org/docs/man1.1.1/man3/SSL_CTX_set_alpn_protos.html)
 ```
 
 #### Bridge Parameters
diff --git a/mqtt_client/CMakeLists.txt b/mqtt_client/CMakeLists.txt
@@ -84,6 +84,7 @@ elseif(${ROS_VERSION} EQUAL 1)
     mqtt_client_interfaces
     nodelet
     roscpp
+    rosfmt ## For fmt::format.
     std_msgs
     topic_tools
   )
diff --git a/mqtt_client/config/params.aws.yaml b/mqtt_client/config/params.aws.yaml
@@ -0,0 +1,49 @@
+# YAML alias for MQTT SSL version.
+# The supported values are defined at:
+# https://github.com/eclipse/paho.mqtt.cpp/blob/master/src/mqtt/ssl_options.h#L305
+tls_1_2: &tls_1_2 3
+
+broker:
+  # YOU MUST CHANGE THIS ENDPOINT
+  # Can be found by executing: aws iot describe-endpoint
+  host: not-a-real-endpoint-please-use-your-own.us-west-2.amazonaws.com
+  port: 8883
+  tls:
+    enabled: true
+    # Available at https://www.amazontrust.com/repository/AmazonRootCA1.pem
+    ca_certificate: path/to/AmazonRootCA1.pem
+client:
+  id: user
+  # Whether or not to start a clean session with each reconnect.
+  # If True, the server will forget all subscriptions with each reconnect.
+  # Set False to request that the server resume an existing session or start
+  # a new session that may be resumed after a connection loss.
+  clean_session: false
+  # The keep alive value, in seconds, to send in CONNECT packet.
+  keep_alive_interval: 6.0
+  tls:
+    # The certificate generated by the AWS IoT Core service
+    certificate: path/to/a-certificate.pem
+    # The private key generated by the AWS IoT Core service
+    key: path/to/a-private-key.pem
+    # AWS uses TLS v1.2 to encrypt all communication.
+    version: *tls_1_2
+    verify: true
+    # https://www.openssl.org/docs/man1.1.1/man3/SSL_CTX_set_alpn_protos.html
+    alpn_protos:
+      # MQTT over AWS IOT requires an ALPN protocol negotiation.
+      # https://docs.aws.amazon.com/iot/latest/developerguide/protocols.html
+      - x-amzn-mqtt-ca
+  buffer:
+    enabled: true
+bridge:
+  # NOTE: It seems that AWS IOT only supports primitive topics. Using non-primitive
+  # types results in the error message `Connection to broker lost, will try to reconnect...`
+  ros2mqtt:
+    - ros_topic: /ping/primitive
+      mqtt_topic: pingpong/primitive
+      primitive: true
+  mqtt2ros:
+    - mqtt_topic: pingpong/primitive
+      ros_topic: /pong/primitive
+      primitive: true
diff --git a/mqtt_client/config/params.ros2.aws.yaml b/mqtt_client/config/params.ros2.aws.yaml
@@ -0,0 +1,48 @@
+mqtt_client:
+  ros__parameters:
+    broker:
+      # YOU MUST CHANGE THIS ENDPOINT
+      # Can be found by executing: aws iot describe-endpoint
+      host: not-a-real-endpoint-please-use-your-own.us-west-2.amazonaws.com
+      port: 8883
+      tls:
+        enabled: true
+        # Available at https://www.amazontrust.com/repository/AmazonRootCA1.pem
+        ca_certificate: path/to/AmazonRootCA1.pem
+    client:
+      id: user
+      # Whether or not to start a clean session with each reconnect.
+      # If True, the server will forget all subscriptions with each reconnect.
+      # Set False to request that the server resume an existing session or start
+      # a new session that may be resumed after a connection loss.
+      clean_session: false
+      # The keep alive value, in seconds, to send in CONNECT packet.
+      keep_alive_interval: 6.0
+      tls:
+        # The certificate generated by the AWS IoT Core service
+        certificate: path/to/a-certificate.pem
+        # The private key generated by the AWS IoT Core service
+        key: path/to/a-private-key.pem
+        # AWS uses TLS v1.2 to encrypt all communication.
+        # The supported values are defined at:
+        # https://github.com/eclipse/paho.mqtt.cpp/blob/master/src/mqtt/ssl_options.h#L305
+        version: 3 # TLS v1.2
+        verify: true
+        # https://www.openssl.org/docs/man1.1.1/man3/SSL_CTX_set_alpn_protos.html
+        alpn_protos:
+          # MQTT over AWS IOT requires an ALPN protocol negotiation.
+          # https://docs.aws.amazon.com/iot/latest/developerguide/protocols.html
+          - x-amzn-mqtt-ca
+      buffer:
+        enabled: true
+    bridge:
+      # NOTE: It seems that AWS IOT only supports primitive topics. Using non-primitive
+      # types results in the error message `Connection to broker lost, will try to reconnect...`
+      ros2mqtt:
+        ros_topic: /ping/primitive
+        mqtt_topic: pingpong/primitive
+        primitive: true
+      mqtt2ros:
+        mqtt_topic: pingpong/primitive
+        ros_topic: /pong/primitive
+        primitive: true
diff --git a/mqtt_client/include/mqtt_client/MqttClient.h b/mqtt_client/include/mqtt_client/MqttClient.h
@@ -32,10 +32,11 @@ SOFTWARE.
 #include <memory>
 #include <string>
 
-#include <mqtt/async_client.h>
 #include <mqtt_client_interfaces/IsConnected.h>
+#include <mqtt/async_client.h>
 #include <nodelet/nodelet.h>
 #include <ros/ros.h>
+#include <rosfmt/full.h> // fmt::format, fmt::join
 #include <topic_tools/shape_shifter.h>
 
 
@@ -126,6 +127,36 @@ class MqttClient : public nodelet::Nodelet,
   template <typename T>
   bool loadParameter(const std::string& key, T& value, const T& default_value);
 
+  /**
+   * @brief Loads requested ROS parameter from parameter server.
+   *
+   * @tparam  T            type (one of int, double, bool)
+   *
+   * @param[in]   key      parameter name
+   * @param[out]  value    variable where to store the retrieved parameter
+   *
+   * @return  true         if parameter was successfully retrieved
+   * @return  false        if parameter was not found
+   */
+  template <typename T>
+  bool loadParameter(const std::string& key, std::vector<T>& value);
+
+  /**
+   * @brief Loads requested ROS parameter from parameter server, allows default
+   * value.
+   *
+   * @tparam  T            type (one of int, double, bool)
+   *
+   * @param[in]   key            parameter name
+   * @param[out]  value          variable where to store the retrieved parameter
+   * @param[in]   default_value  default value
+   *
+   * @return  true         if parameter was successfully retrieved
+   * @return  false        if parameter was not found or default was used
+   */
+  template <typename T>
+  bool loadParameter(const std::string& key, std::vector<T>& value, const std::vector<T>& default_value);
+
   /**
    * @brief Converts a string to a path object resolving paths relative to
    * ROS_HOME.
@@ -330,9 +361,13 @@ class MqttClient : public nodelet::Nodelet,
     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
+      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
   };
 
@@ -468,6 +503,32 @@ bool MqttClient::loadParameter(const std::string& key, T& value,
 }
 
 
+template <typename T>
+bool MqttClient::loadParameter(const std::string& key, std::vector<T>& value)
+{
+  const bool found = private_node_handle_.getParam(key, value);
+  if (found)
+    NODELET_DEBUG("Retrieved parameter '%s' = '[%s]'", key.c_str(),
+                  fmt::format("{}", fmt::join(value, ", ")).c_str());
+  return found;
+}
+
+
+template <typename T>
+bool MqttClient::loadParameter(const std::string& key, std::vector<T>& value,
+                               const std::vector<T>& default_value)
+{
+  const bool found = private_node_handle_.param<T>(key, value, default_value);
+  if (!found)
+    NODELET_WARN("Parameter '%s' not set, defaulting to '%s'", key.c_str(),
+                  fmt::format("{}", fmt::join(value, ", ")).c_str());
+  if (found)
+    NODELET_DEBUG("Retrieved parameter '%s' = '%s'", key.c_str(),
+                  fmt::format("{}", fmt::join(value, ", ")).c_str());
+  return found;
+}
+
+
 /**
  * Serializes a ROS message to a buffer.
  *
diff --git a/mqtt_client/include/mqtt_client/MqttClient.ros2.hpp b/mqtt_client/include/mqtt_client/MqttClient.ros2.hpp
@@ -127,6 +127,36 @@ class MqttClient : public rclcpp::Node,
   template <typename T>
   bool loadParameter(const std::string& key, T& value, const T& default_value);
 
+  /**
+   * @brief Loads requested ROS parameter from parameter server.
+   *
+   * @tparam  T            type (one of int, double, bool)
+   *
+   * @param[in]   key      parameter name
+   * @param[out]  value    variable where to store the retrieved parameter
+   *
+   * @return  true         if parameter was successfully retrieved
+   * @return  false        if parameter was not found
+   */
+  template <typename T>
+  bool loadParameter(const std::string& key, std::vector<T>& value);
+
+  /**
+   * @brief Loads requested ROS parameter from parameter server, allows default
+   * value.
+   *
+   * @tparam  T            type (one of int, double, bool)
+   *
+   * @param[in]   key            parameter name
+   * @param[out]  value          variable where to store the retrieved parameter
+   * @param[in]   default_value  default value
+   *
+   * @return  true         if parameter was successfully retrieved
+   * @return  false        if parameter was not found or default was used
+   */
+  template <typename T>
+  bool loadParameter(const std::string& key, std::vector<T>& value, const std::vector<T>& default_value);
+
   /**
    * @brief Converts a string to a path object resolving paths relative to
    * ROS_HOME.
@@ -323,9 +353,13 @@ 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
+      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
   };
 
@@ -462,6 +496,32 @@ bool MqttClient::loadParameter(const std::string& key, T& value,
 }
 
 
+template <typename T>
+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());
+  return found;
+}
+
+
+template <typename T>
+bool MqttClient::loadParameter(const std::string& key, std::vector<T>& value,
+                               const std::vector<T>& default_value)
+{
+  const bool found = get_parameter_or(key, value, default_value);
+  if (!found)
+    RCLCPP_WARN(get_logger(), "Parameter '%s' not set, defaulting to '%s'",
+                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());
+  return found;
+}
+
+
 /**
  * Serializes a ROS message.
  *
diff --git a/mqtt_client/package.xml b/mqtt_client/package.xml
@@ -34,6 +34,7 @@
   <depend condition="$ROS_VERSION == 1">nodelet</depend>
   <depend condition="$ROS_VERSION == 1">paho-mqtt-cpp</depend>
   <depend condition="$ROS_VERSION == 1">roscpp</depend>
+  <depend condition="$ROS_VERSION == 1">rosfmt</depend> <!-- Needed for fmt library. -->
   <depend condition="$ROS_VERSION == 1">topic_tools</depend>
 
   <export>
diff --git a/mqtt_client/src/MqttClient.cpp b/mqtt_client/src/MqttClient.cpp
@@ -194,6 +194,9 @@ void MqttClient::loadParameters() {
     if (loadParameter("client/tls/certificate", client_tls_certificate)) {
       loadParameter("client/tls/key", client_tls_key);
       loadParameter("client/tls/password", client_config_.tls.password);
+      loadParameter("client/tls/version", client_config_.tls.version);
+      loadParameter("client/tls/verify", client_config_.tls.verify);
+      loadParameter("client/tls/alpn_protos", client_config_.tls.alpn_protos);
     }
   }
 
@@ -437,13 +440,16 @@ void MqttClient::setupClient() {
       if (!client_config_.tls.password.empty())
         ssl.set_private_key_password(client_config_.tls.password);
     }
+    ssl.set_ssl_version(client_config_.tls.version);
+    ssl.set_verify(client_config_.tls.verify);
+    ssl.set_alpn_protos(client_config_.tls.alpn_protos);
     connect_options_.set_ssl(ssl);
   }
 
   // create MQTT client
-  std::string protocol = broker_config_.tls.enabled ? "ssl" : "tcp";
-  std::string uri = protocol + std::string("://") + broker_config_.host +
-                    std::string(":") + std::to_string(broker_config_.port);
+  const std::string protocol = broker_config_.tls.enabled ? "ssl" : "tcp";
+  const std::string uri = fmt::format("{}://{}:{}", protocol, broker_config_.host,
+                                      broker_config_.port);
   try {
     if (client_config_.buffer.enabled) {
       client_ = std::shared_ptr<mqtt::async_client>(new mqtt::async_client(
diff --git a/mqtt_client/src/MqttClient.ros2.cpp b/mqtt_client/src/MqttClient.ros2.cpp
@@ -261,6 +261,9 @@ void MqttClient::loadParameters() {
     if (loadParameter("client.tls.certificate", client_tls_certificate)) {
       loadParameter("client.tls.key", client_tls_key);
       loadParameter("client.tls.password", client_config_.tls.password);
+      loadParameter("client.tls.version", client_config_.tls.version);
+      loadParameter("client.tls.verify", client_config_.tls.verify);
+      loadParameter("client.tls.alpn_protos", client_config_.tls.alpn_protos);
     }
   }
 
@@ -517,13 +520,16 @@ void MqttClient::setupClient() {
       if (!client_config_.tls.password.empty())
         ssl.set_private_key_password(client_config_.tls.password);
     }
+    ssl.set_ssl_version(client_config_.tls.version);
+    ssl.set_verify(client_config_.tls.verify);
+    ssl.set_alpn_protos(client_config_.tls.alpn_protos);
     connect_options_.set_ssl(ssl);
   }
 
   // create MQTT client
-  std::string protocol = broker_config_.tls.enabled ? "ssl" : "tcp";
-  std::string uri = protocol + std::string("://") + broker_config_.host +
-                    std::string(":") + std::to_string(broker_config_.port);
+  const std::string protocol = broker_config_.tls.enabled ? "ssl" : "tcp";
+  const std::string uri = fmt::format("{}://{}:{}", protocol, broker_config_.host,
+                                      broker_config_.port);
   try {
     if (client_config_.buffer.enabled) {
       client_ = std::shared_ptr<mqtt::async_client>(new mqtt::async_client(

Original file line number	Diff line number	Diff line change
`@@ -84,6 +84,7 @@ elseif(${ROS_VERSION} EQUAL 1)`
`84`	`84`	`mqtt_client_interfaces`
`85`	`85`	`nodelet`
`86`	`86`	`roscpp`
	`87`	`+ rosfmt ## For fmt::format.`
`87`	`88`	`std_msgs`
`88`	`89`	`topic_tools`
`89`	`90`	`)`