ros_subscriber_node.hpp

// Copyright 2024 Robin Müller
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
 
#pragma once
 
#include <algorithm>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
#include <vector>
 
#include "auto_apms_behavior_tree_core/exceptions.hpp"
#include "auto_apms_behavior_tree_core/node/base/ros_condition_node.hpp"
#include "auto_apms_util/logging.hpp"
#include "rclcpp/executors.hpp"
#include "rclcpp/qos.hpp"
 
namespace auto_apms_behavior_tree::core
{

template <class MessageT>
class RosSubscriberNode : public RosConditionNode
{
  using Subscriber = typename rclcpp::Subscription<MessageT>;
 
  struct SubscriberInstance
  {
    SubscriberInstance(
      rclcpp::Node::SharedPtr node, rclcpp::CallbackGroup::SharedPtr group, const std::string & topic_name,
      const rclcpp::QoS & qos);
 
    std::shared_ptr<Subscriber> subscriber;
    std::vector<std::pair<const void *, std::function<void(const std::shared_ptr<MessageT>)>>> callbacks;
    std::shared_ptr<MessageT> last_msg;
    std::string name;
 
    void addCallback(
      const void * callback_owner, const std::function<void(const std::shared_ptr<MessageT>)> & callback);
    void removeCallback(const void * callback_owner);
    void broadcast(const std::shared_ptr<MessageT> & msg);
  };
 
public:
  using MessageType = MessageT;
  using Config = BT::NodeConfig;
  using Context = RosNodeContext;

  explicit RosSubscriberNode(
    const std::string & instance_name, const Config & config, Context context, const rclcpp::QoS & qos = {10});
 
  virtual ~RosSubscriberNode()
  {
    if (sub_instance_) {
      sub_instance_->removeCallback(this);
    }
  }

  static BT::PortsList providedBasicPorts(BT::PortsList addition)
  {
    BT::PortsList basic = {BT::InputPort<std::string>("topic", "Name of the ROS 2 topic to subscribe to.")};
    basic.insert(addition.begin(), addition.end());
    return basic;
  }

  static BT::PortsList providedPorts() { return providedBasicPorts({}); }

  virtual BT::NodeStatus onTick(const std::shared_ptr<MessageT> & last_msg_ptr);

  virtual BT::NodeStatus onMessageReceived(const MessageT & msg);

  bool createSubscriber(const std::string & topic_name);

  std::string getTopicName() const;
 
protected:
  BT::NodeStatus tick() override final;
 
private:
  const rclcpp::QoS qos_;
  std::string topic_name_;
  bool dynamic_client_instance_ = false;
  std::shared_ptr<SubscriberInstance> sub_instance_;
  std::shared_ptr<MessageT> last_msg_;
};
 
// #####################################################################################################################
// ################################              DEFINITIONS              ##############################################
// #####################################################################################################################
 
template <class MessageT>
inline RosSubscriberNode<MessageT>::SubscriberInstance::SubscriberInstance(
  rclcpp::Node::SharedPtr node, rclcpp::CallbackGroup::SharedPtr group, const std::string & topic_name,
  const rclcpp::QoS & qos)
{
  rclcpp::SubscriptionOptions option;
  option.callback_group = group;
 
  // The callback will broadcast to all the instances of RosSubscriberNode<MessageT>
  auto callback = [this](const std::shared_ptr<MessageT> msg) {
    this->last_msg = msg;
    this->broadcast(msg);
  };
  subscriber = node->create_subscription<MessageT>(topic_name, qos, callback, option);
  name = topic_name;
}
 
template <class MessageT>
inline void RosSubscriberNode<MessageT>::SubscriberInstance::addCallback(
  const void * callback_owner, const std::function<void(const std::shared_ptr<MessageT>)> & callback)
{
  callbacks.emplace_back(callback_owner, callback);
}
 
template <class MessageT>
inline void RosSubscriberNode<MessageT>::SubscriberInstance::removeCallback(const void * callback_owner)
{
  callbacks.erase(
    std::remove_if(
      callbacks.begin(), callbacks.end(), [callback_owner](const auto & pair) { return pair.first == callback_owner; }),
    callbacks.end());
}
 
template <class MessageT>
inline void RosSubscriberNode<MessageT>::SubscriberInstance::broadcast(const std::shared_ptr<MessageT> & msg)
{
  for (auto & callback_pair : callbacks) {
    callback_pair.second(msg);
  }
}
 
template <class MessageT>
inline RosSubscriberNode<MessageT>::RosSubscriberNode(
  const std::string & instance_name, const Config & config, Context context, const rclcpp::QoS & qos)
: RosConditionNode{instance_name, config, context}, qos_{qos}
{
  // Port aliasing is already applied by RosConditionNode ctor.
 
  if (const BT::Expected<std::string> expected_name = context_.getTopicName(this)) {
    createSubscriber(expected_name.value());
  } else {
    // We assume that determining the topic name requires a blackboard pointer, which cannot be evaluated at
    // construction time. The expression will be evaluated each time before the node is ticked the first time after
    // successful execution.
    dynamic_client_instance_ = true;
  }
}
 
template <class MessageT>
inline bool RosSubscriberNode<MessageT>::createSubscriber(const std::string & topic_name)
{
  if (topic_name.empty()) {
    throw exceptions::RosNodeError(
      context_.getFullyQualifiedTreeNodeName(this) + " - Argument topic_name is empty when trying to create a client.");
  }
 
  // Check if the subscriber with given name is already set up
  if (sub_instance_ && topic_name == sub_instance_->name) return true;
 
  rclcpp::Node::SharedPtr node = context_.getRosNode();
  rclcpp::CallbackGroup::SharedPtr group = context_.getWaitablesCallbackGroup();
  if (!group) {
    throw exceptions::RosNodeError(
      context_.getFullyQualifiedTreeNodeName(this) +
      " - The weak pointer to the ROS 2 callback group expired. The tree node doesn't "
      "take ownership of it.");
  }
 
  // Reuse a subscriber shared across tree nodes with the same topic name, or create one on first use. The shared
  // instance broadcasts every received message to all the nodes that registered a callback below.
  sub_instance_ = this->template getSharedEntity<SubscriberInstance>(
    topic_name, [&] { return std::make_shared<SubscriberInstance>(node, group, topic_name, qos_); });
 
  // Check if there was a message received before the creation of this subscriber action
  if (sub_instance_->last_msg) {
    last_msg_ = sub_instance_->last_msg;
  }
 
  // add "this" as received of the broadcaster
  sub_instance_->addCallback(this, [this](const std::shared_ptr<MessageT> msg) { last_msg_ = msg; });
 
  return true;
}
 
template <class MessageT>
inline BT::NodeStatus RosSubscriberNode<MessageT>::tick()
{
  if (!rclcpp::ok()) {
    halt();
    return BT::NodeStatus::FAILURE;
  }
 
  // If client has been set up in derived constructor, event though this constructor couldn't, we discard the intention
  // of dynamically creating the client
  if (dynamic_client_instance_ && sub_instance_) {
    dynamic_client_instance_ = false;
  }
 
  // Try again to create the client on first tick if this was not possible during construction or if client should be
  // created from a blackboard entry on the start of every iteration
  if (status() == BT::NodeStatus::IDLE && dynamic_client_instance_) {
    const BT::Expected<std::string> expected_name = context_.getTopicName(this);
    if (expected_name) {
      createSubscriber(expected_name.value());
    } else {
      throw exceptions::RosNodeError(
        context_.getFullyQualifiedTreeNodeName(this) +
        " - Cannot create the subscriber because the topic name couldn't be resolved using "
        "the expression specified by the node's registration parameters (" +
        NodeRegistrationOptions::PARAM_NAME_ROS2TOPIC + ": " + context_.getRegistrationOptions().topic +
        "). Error message: " + expected_name.error());
    }
  }
 
  if (!sub_instance_) {
    throw exceptions::RosNodeError(context_.getFullyQualifiedTreeNodeName(this) + " - sub_instance_ is nullptr.");
  }
 
  auto check_status = [this](BT::NodeStatus status) {
    if (!isStatusCompleted(status)) {
      throw exceptions::RosNodeError(
        context_.getFullyQualifiedTreeNodeName(this) + " - The callback must return either SUCCESS or FAILURE.");
    }
    return status;
  };
  auto status = check_status(onTick(last_msg_));
  last_msg_.reset();
  return status;
}
 
template <class MessageT>
inline BT::NodeStatus RosSubscriberNode<MessageT>::onTick(const std::shared_ptr<MessageT> & last_msg_ptr)
{
  if (!last_msg_ptr) return BT::NodeStatus::FAILURE;
  return onMessageReceived(*last_msg_ptr);
}
 
template <class MessageT>
inline BT::NodeStatus RosSubscriberNode<MessageT>::onMessageReceived(const MessageT & /*msg*/)
{
  return BT::NodeStatus::SUCCESS;
}
 
template <class MessageT>
inline std::string RosSubscriberNode<MessageT>::getTopicName() const
{
  if (sub_instance_) return sub_instance_->name;
  return "unknown";
}
 
}  // namespace auto_apms_behavior_tree::core