ROS源码学习分享_TopicManager::start()

        在上一章节中，我们讲解了NodeHandle节点创建后的一些背后行为。其最重要的行为是启动了全部的管理节点。在本章中，我们将看一看TopicManager节点在启动之后发生了什么。(以下内容，属于个人观看源码后得出的理解，可能有错，仅用于自我复习，请批判的看待）

        TopicManager::start()的源码如下：

void TopicManager::start()
{
  boost::mutex::scoped_lock shutdown_lock(shutting_down_mutex_);
  shutting_down_ = false;
  poll_manager_ = PollManager::instance();
  connection_manager_ = ConnectionManager::instance();
  xmlrpc_manager_ = XMLRPCManager::instance();

  xmlrpc_manager_->bind("publisherUpdate", boost::bind(&TopicManager::pubUpdateCallback, this, boost::placeholders::_1, boost::placeholders::_2));
  xmlrpc_manager_->bind("requestTopic", boost::bind(&TopicManager::requestTopicCallback, this, boost::placeholders::_1, boost::placeholders::_2));
  xmlrpc_manager_->bind("getBusStats", boost::bind(&TopicManager::getBusStatsCallback, this, boost::placeholders::_1, boost::placeholders::_2));
  xmlrpc_manager_->bind("getBusInfo", boost::bind(&TopicManager::getBusInfoCallback, this, boost::placeholders::_1, boost::placeholders::_2));
  xmlrpc_manager_->bind("getSubscriptions", boost::bind(&TopicManager::getSubscriptionsCallback, this, boost::placeholders::_1, boost::placeholders::_2));
  xmlrpc_manager_->bind("getPublications", boost::bind(&TopicManager::getPublicationsCallback, this, boost::placeholders::_1, boost::placeholders::_2));
   
  poll_manager_->addPollThreadListener(boost::bind(&TopicManager::processPublishQueues, this));
}

        他先给xmlrpc_manager注册了一系列查询Topic相关信息的回调函数。在第一章，我们有提到，对于master节点的话题注册订阅发布是由其下的xml_rpc进行完成的。那么如果接受到远程调用，xmlrpc就会调用这里的回调函数。

        我们先看看publisherUpdate的一个实现。我们先来看看xmlrpc是如何进行绑定的:

bool XMLRPCManager::bind(const std::string& function_name, const XMLRPCFunc& cb)
{
  boost::mutex::scoped_lock lock(functions_mutex_);
  if (functions_.find(function_name) != functions_.end())
  {
    return false;
  }

  FunctionInfo info;
  info.name = function_name;
  info.function = cb;
  info.wrapper.reset(new XMLRPCCallWrapper(function_name, cb, &server_));
  functions_[function_name] = info;

  return true;
}

functions_其实就是M_StringToFunInfo,而M_StringToFunInfo是一个键对值为string,FunctionInfo的map。

  typedef std::map<std::string, FunctionInfo> M_StringToFuncInfo;

  struct FunctionInfo
  {
    std::string name;
    XMLRPCFunc function;
    XMLRPCCallWrapperPtr wrapper;
  };

而FunctionInfo则是一个拥有回调函数信息的结构体。

这里面warpper则是这段代码中的核心，他的作用是包装一个 XML-RPC 函数，使其能够在 XmlRpcServer 上被调用，它的主要功能是将一个回调函数 (XMLRPCFunc) 与一个 XML-RPC 函数名称绑定，并在接收到相应的 XML-RPC 请求时调用该回调函数。

让我们看看这段代码：

class XMLRPCCallWrapper : public XmlRpcServerMethod
{
public:
  XMLRPCCallWrapper(const std::string& function_name, const XMLRPCFunc& cb, XmlRpcServer *s)
  : XmlRpcServerMethod(function_name, s)
  , name_(function_name)
  , func_(cb)
  { }

  void execute(XmlRpcValue &params, XmlRpcValue &result)
  {
    func_(params, result);
  }

private:
  std::string name_;
  XMLRPCFunc func_;
};

他先给用传入的回调函数给func_进行了赋值，然后设置了一个execute执行函数去执行回调。那么还有构建一个XmlRpcServerMethod，在看这个函数之前，我们需要看一下XmlRpcServer这个类也就是传入参数s。

        XmlRpcServer类提供了一个框架来管理和处理XML-RPC请求，包括添加和移除方法、处理客户端连接、支持内省、以及处理事件和连接等。由于XmlRpcServer类代码众多，我们在此仅列举出与XMLRPCCallWrapper相关的一些函数

        类代码如下：

class XmlRpcValue;


  //! A class to handle XML RPC requests
  class XMLRPCPP_DECL XmlRpcServer : public XmlRpcSource {
  public:
    //! Create a server object.
    XmlRpcServer();
    //! Destructor.
    virtual ~XmlRpcServer();

    //! Specify whether introspection is enabled or not. Default is not enabled.
    void enableIntrospection(bool enabled=true);

    //! Add a command to the RPC server
    void addMethod(XmlRpcServerMethod* method);

    //! Remove a command from the RPC server
    void removeMethod(XmlRpcServerMethod* method);

    //! Remove a command from the RPC server by name
    void removeMethod(const std::string& methodName);

        我们来看看addMethod的源码:

void XmlRpcServer::addMethod(XmlRpcServerMethod* method) {
  _methods[method->name()] = method;
}

        他又回到了XmlRpcServerMethod* ,再看看xmlRpcServerMethod的构造函数：

typedef std::map< std::string, XmlRpcServerMethod* > MethodMap;
MethodMap _methods;

XmlRpcServerMethod::XmlRpcServerMethod(const std::string& function_name, XmlRpcServer* s)
: _functionName(function_name), _server(s)
{
  if (s)
  {
    s->addMethod(this); // 将当前方法注册到 XmlRpcServer
  }
}

       看了这么多代码可能有点疑惑，那么在这里梳理一下，首先xmlrpc_manager调用了bind函数，将回调函数进行绑定，在这个绑定函数中先创建了拥有回调函数信息的结构体info,info包含回调函数的函数名，函数本身以及warpper指针，并将其映射到一个map表中保存。warpper将一个回调函数 (XMLRPCFunc) 与一个 XML-RPC 函数名称绑定，并在接收到相应的 XML-RPC 请求时调用该回调函数。在warpper代码中，他设置了一个execute执行函数去执行回调，由于其继承了XmlRpcServerMethod，其构建了一个XmlRpcServerMethod的基类。基类传了一个XmlRpcServer的指针用于调用其中的addMethod函数将当前方法注册到XmlRpcServer.值得注意的是，请不要这里注册的是父类XmlRpcServerMethod，并不是子类warpper，构建由内而外，先构建父类，这个this指针就指向父类本身。他将method方法与对应的回调函数绑定起来进行注册，后面他调用warpper是使用了多态。将当前方法对象注册到服务器的意义在于允许服务器能够识别和调用这些方法，以响应客户端的 XML-RPC 请求。

        对于两个存储map大家可能也有疑问他们有什么区别，这里说一下。

• functions_ 用于 XMLRPCManager 内部追踪和管理方法信息，可以查询和操作每个方法的详细信息，包括方法的回调函数和 XMLRPCCallWrapper 对象的指针。
• _methods 用于 XmlRpcServer 处理 XML-RPC 请求，服务器接收到请求后，会根据方法名在 _methods 中查找相应的 XmlRpcServerMethod 对象，并调用其 execute 方法来处理请求。

这几个回调函数就不看了(偷懒)。简单说一下几个函数的功能：

• pubUpdateCallback

  • 功能： 处理发布者更新的回调函数。
  • 详细解释： 当 XML-RPC 请求名称为 "publisherUpdate" 时，服务器将调用 pubUpdateCallback 函数。这个函数通常用于更新发布者的信息或状态。在 ROS 中，发布者发布消息到话题时，它们的状态（例如频率、主题名称等）可能会改变，因此需要一个机制来通知其他节点或系统进行相应的更新。pubUpdateCallback 会检查发布者的状态或参数，并且在需要时更新相关的信息。

• requestTopicCallback

  • 功能： 处理请求主题的回调函数。
  • 详细解释： 当 XML-RPC 请求名称为 "requestTopic" 时，服务器将调用 requestTopicCallback 函数。这个函数通常用于处理节点或客户端请求订阅特定主题的操作。在 ROS 中，节点可能会使用此功能来获取关于某个主题的信息（如消息类型、发布者信息等），以便进行订阅。requestTopicCallback 会根据请求提供相关的主题信息或者协商订阅参数。

• getBusStatsCallback

  • 功能： 获取总线统计信息的回调函数。
  • 详细解释： 当 XML-RPC 请求名称为 "getBusStats" 时，服务器将调用 getBusStatsCallback 函数。这个函数通常用于获取 ROS 总线（如通信总线）的统计信息。ROS 中的总线负责节点之间的通信和消息传递，因此了解总线的负载、延迟等统计信息对系统的监控和优化非常重要。getBusStatsCallback 会查询总线的实时数据，并将这些数据返回给请求的节点或客户端。

• getBusInfoCallback

  • 功能： 获取总线信息的回调函数。
  • 详细解释： 当 XML-RPC 请求名称为 "getBusInfo" 时，服务器将调用 getBusInfoCallback 函数。这个函数通常用于获取有关 ROS 总线（通信基础设施）的详细信息。这些信息可能包括总线的拓扑结构、节点连接情况、通信协议等。getBusInfoCallback 会返回总线的静态信息或者实时状态，以便用户了解和分析 ROS 系统的通信配置和状态。

• getSubscriptionsCallback

  • 功能： 获取订阅信息的回调函数。
  • 详细解释： 当 XML-RPC 请求名称为 "getSubscriptions" 时，服务器将调用 getSubscriptionsCallback 函数。这个函数通常用于获取当前节点（或系统）中所有订阅的主题信息。在 ROS 中，节点通过订阅主题接收其他节点发布的消息。getSubscriptionsCallback 会列出当前节点订阅的所有主题及其相关信息，如主题名称、消息类型、发布者信息等。

• getPublicationsCallback

  • 功能： 获取发布信息的回调函数。
  • 详细解释： 当 XML-RPC 请求名称为 "getPublications" 时，服务器将调用 getPublicationsCallback 函数。这个函数通常用于获取当前节点（或系统）中所有发布的主题信息。在 ROS 中，节点通过发布主题向其他节点传递消息。getPublicationsCallback 会列出当前节点发布的所有主题及其相关信息，如主题名称、消息类型、订阅者信息等。

讲了这么多，其实topic_manager还有一个功能

poll_manager_->addPollThreadListener(boost::bind(&TopicManager::processPublishQueues, this));

他是将发布队列注册函数给poll_manager。

内部函数如下：

boost::signals2::connection PollManager::addPollThreadListener(const VoidFunc& func)
{
  boost::recursive_mutex::scoped_lock lock(signal_mutex_);
  return poll_signal_.connect(func);
}

其实是把这个发布队列函数与poll_signal_这个信号进行绑定，给到信号槽进行监听，如果有数据传输他就会调用这个函数。经过上面的绑定在poll线程中会定期地执行processPublishQueues函数这个函数处理发布队列。简单的说他绑定func之后，在发送信号poll_signal_()，其所连接的槽函数都会被调用。这就是异步调用的概念，一边生产者把要发布的消息添加到发布队列中，另一边调用处理发布队列。

总结：在此篇文章中我们观看了ros源代码中，xml_rpc是如何绑定回调函数的机制以及如何异步处理发布队列是如何实现的

截止目前的pub底层时序图：