Use message_info in SubscriptionTopicStatistics instead of typed mess…

…age (#2337)

* Use message_info in SubscriptionTopicStatistics instead of typed message

- Untemplatize the rclcpp::topic_statistics::SubscriptionTopicStatistics
class. Now we will be using message_info instead of typed deserialized
messages in the handle_message callbacks.

* Fix test_receive_stats_include_window_reset by using publisher emulator

- Emulate publishing messages by directly calling
rclcpp::Subscription::handle_message(msg_shared_ptr, message_info) and
settling up needed message_info.source_timestamp

Signed-off-by: Michael Orlov <michael.orlov@apex.ai>
Co-authored-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

rclcpp/include/rclcpp/create_subscription.hpp

@@ -50,8 +50,8 @@ template<
   typename SubscriptionT,
   typename MessageMemoryStrategyT,
   typename NodeParametersT,
-  typename NodeTopicsT,
-  typename ROSMessageType = typename SubscriptionT::ROSMessageType>
+  typename NodeTopicsT
+>
 typename std::shared_ptr<SubscriptionT>
 create_subscription(
   NodeParametersT & node_parameters,
@@ -70,7 +70,7 @@ create_subscription(
   using rclcpp::node_interfaces::get_node_topics_interface;
   auto node_topics_interface = get_node_topics_interface(node_topics);
 
-  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>
+  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>
   subscription_topic_stats = nullptr;
 
   if (rclcpp::detail::resolve_enable_topic_statistics(
@@ -80,8 +80,7 @@ create_subscription(
     if (options.topic_stats_options.publish_period <= std::chrono::milliseconds(0)) {
       throw std::invalid_argument(
               "topic_stats_options.publish_period must be greater than 0, specified value of " +
-              std::to_string(options.topic_stats_options.publish_period.count()) +
-              " ms");
+              std::to_string(options.topic_stats_options.publish_period.count()) + " ms");
     }
 
     std::shared_ptr<Publisher<statistics_msgs::msg::MetricsMessage>>
@@ -91,12 +90,12 @@ create_subscription(
       options.topic_stats_options.publish_topic,
       qos);
 
-    subscription_topic_stats = std::make_shared<
-      rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>
-      >(node_topics_interface->get_node_base_interface()->get_name(), publisher);
+    subscription_topic_stats =
+      std::make_shared<rclcpp::topic_statistics::SubscriptionTopicStatistics>(
+      node_topics_interface->get_node_base_interface()->get_name(), publisher);
 
     std::weak_ptr<
-      rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>
+      rclcpp::topic_statistics::SubscriptionTopicStatistics
     > weak_subscription_topic_stats(subscription_topic_stats);
     auto sub_call_back = [weak_subscription_topic_stats]() {
         auto subscription_topic_stats = weak_subscription_topic_stats.lock();

rclcpp/include/rclcpp/subscription.hpp

@@ -104,7 +104,7 @@ class Subscription : public SubscriptionBase
 
 private:
   using SubscriptionTopicStatisticsSharedPtr =
-    std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>;
+    std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>;
 
 public:
   RCLCPP_SMART_PTR_DEFINITIONS(Subscription)
@@ -316,7 +316,7 @@ class Subscription : public SubscriptionBase
     if (subscription_topic_statistics_) {
       const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
       const auto time = rclcpp::Time(nanos.time_since_epoch().count());
-      subscription_topic_statistics_->handle_message(*typed_message, time);
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
     }
   }
 
@@ -357,7 +357,7 @@ class Subscription : public SubscriptionBase
     if (subscription_topic_statistics_) {
       const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(now);
       const auto time = rclcpp::Time(nanos.time_since_epoch().count());
-      subscription_topic_statistics_->handle_message(*typed_message, time);
+      subscription_topic_statistics_->handle_message(message_info.get_rmw_message_info(), time);
     }
   }
 

rclcpp/include/rclcpp/subscription_factory.hpp

@@ -75,15 +75,14 @@ template<
   typename CallbackT,
   typename AllocatorT,
   typename SubscriptionT = rclcpp::Subscription<MessageT, AllocatorT>,
-  typename MessageMemoryStrategyT = typename SubscriptionT::MessageMemoryStrategyType,
-  typename ROSMessageType = typename SubscriptionT::ROSMessageType
+  typename MessageMemoryStrategyT = typename SubscriptionT::MessageMemoryStrategyType
 >
 SubscriptionFactory
 create_subscription_factory(
   CallbackT && callback,
   const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options,
   typename MessageMemoryStrategyT::SharedPtr msg_mem_strat,
-  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics<ROSMessageType>>
+  std::shared_ptr<rclcpp::topic_statistics::SubscriptionTopicStatistics>
   subscription_topic_stats = nullptr
 )
 {

rclcpp/include/rclcpp/topic_statistics/subscription_topic_statistics.hpp

@@ -48,21 +48,12 @@ using libstatistics_collector::moving_average_statistics::StatisticData;
 /**
  * Class used to collect, measure, and publish topic statistics data. Current statistics
  * supported for subscribers are received message age and received message period.
- *
- * \tparam CallbackMessageT the subscribed message type
- */
-template<typename CallbackMessageT>
+  */
 class SubscriptionTopicStatistics
 {
-  using TopicStatsCollector =
-    libstatistics_collector::topic_statistics_collector::TopicStatisticsCollector<
-    CallbackMessageT>;
-  using ReceivedMessageAge =
-    libstatistics_collector::topic_statistics_collector::ReceivedMessageAgeCollector<
-    CallbackMessageT>;
-  using ReceivedMessagePeriod =
-    libstatistics_collector::topic_statistics_collector::ReceivedMessagePeriodCollector<
-    CallbackMessageT>;
+  using TopicStatsCollector = libstatistics_collector::TopicStatisticsCollector;
+  using ReceivedMessageAge = libstatistics_collector::ReceivedMessageAgeCollector;
+  using ReceivedMessagePeriod = libstatistics_collector::ReceivedMessagePeriodCollector;
 
 public:
   /// Construct a SubscriptionTopicStatistics object.
@@ -101,16 +92,16 @@ class SubscriptionTopicStatistics
   /**
    * This method acquires a lock to prevent race conditions to collectors list.
    *
-   * \param received_message the message received by the subscription
+   * \param message_info the message info corresponding to the received message
    * \param now_nanoseconds current time in nanoseconds
    */
   virtual void handle_message(
-    const CallbackMessageT & received_message,
+    const rmw_message_info_t & message_info,
     const rclcpp::Time now_nanoseconds) const
   {
     std::lock_guard<std::mutex> lock(mutex_);
     for (const auto & collector : subscriber_statistics_collectors_) {
-      collector->OnMessageReceived(received_message, now_nanoseconds.nanoseconds());
+      collector->OnMessageReceived(message_info, now_nanoseconds.nanoseconds());
     }
   }