参考：http://wiki.ros.org/ROS/Master_API

http://wiki.ros.org/ROS/Connection Header

说明

ROS本质上就是一个松耦合的通信框架，通信模式包括：远程调用（service-client）、订阅发布（topic）、持续通信（action）和全局参数（参数服务器），这四种模式基本已经能够涵盖百分之九十的应用场景了

本次针对订阅发布模式，探究一下ROS通信中的具体通信协议，读完本文后，你可以在不依赖ROS的情况下和ROS通信

本次通信采用从机订阅主机数据，通过wireshark抓包，得到具体xmlrpc协议数据内容，根据xmlrpc协议格式，找到对应代码

（因为时间有限，部分协议可能有跳过的地方）

1、registerPublisher

从机创建节点的第一步

这个方法用于注册一个发布者的caller

<?xml version="1.0"?>
<methodCall>
    <methodName>registerPublisher</methodName>
    <params>
        <param>
            <value>/test_sub</value>
        </param>
        <param>
            <value>/rosout</value>
        </param>
        <param>
            <value>rosgraph_msgs/Log</value>
        </param>
        <param>
            <value>http://192.168.1.150:40209</value>
        </param>
    </params>
</methodCall>


<?xml version='1.0'?>
<methodResponse>
    <params>
        <param>
            <value>
                <array>
                    <data>
                        <value>
                            <int>1</int>
                        </value>
                        <value>
                            <string>Registered [/test_sub] as publisher of [/rosout]</string>
                        </value>
                        <value>
                            <array>
                                <data>
                                    <value>
                                        <string>http://sherlock:35861/</string>
                                    </value>
                                </data>
                            </array>
                        </value>
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodResponse>

先说结论：

ROS有一个日志相关的topic，名称是 /rosout，所有节点的日志信息都会通过这个 topic 发布出来

ROS还有一个日志相关的节点，名称是 /rosout，负责订阅 /rosout数据，然后使用名称为 /rosout_agg 的topic发布出来， /rosout_agg 的订阅者是rqt等调试工具

所以，结合上面的xml内容，我们可以大致推断，创建一个新的节点，那么这个节点必定会发布一个topic，就是/rosout，所以上面的XMLRPC协议内容，就是网rosmaster内注册一个publisher，用于发布/rosout

整体来说，就是调用接口

registerPublisher("/test_sub", "/rosout", "rosgraph_msgs/Log", "http://192.168.1.150:40209")

返回值是：

1
Registered [/test_sub] as publisher of [/rosout]
http://sherlock:35861/

1是固定数据

第二行是message

最后一个返回值表示订阅者的uri列表，这里因为只有一个订阅者，所有只有一个uri

再看代码：

函数声明如下：

registerPublisher(caller_id, topic, topic_type, caller_api)
    Register the caller as a publisher the topic.
    参数
        caller_id (str)
            ROS caller ID
        topic (str)
            Fully-qualified name of topic to register.
        topic_type (str)
            Datatype for topic. Must be a package-resource name, i.e. the .msg name.
        caller_api (str)
            API URI of publisher to register.
    返回值(int, str, [str])
        (code, statusMessage, subscriberApis)
        List of current subscribers of topic in the form of XMLRPC URIs.

找到 registerPublisher 接接口，位于ros_comm/rosmaster 包中，文件为：master_api.py（ROS主从机制利用python实现，拿掉python则无法实现主从）

@apivalidate([], ( is_topic('topic'), valid_type_name('topic_type'), is_api('caller_api')))
def registerPublisher(self, caller_id, topic, topic_type, caller_api):
    """
    Register the caller as a publisher the topic.
    @param caller_id: ROS caller id
    @type  caller_id: str
    @param topic: Fully-qualified name of topic to register.
    @type  topic: str
    @param topic_type: Datatype for topic. Must be a
    package-resource name, i.e. the .msg name.
    @type  topic_type: str
    @param caller_api str: ROS caller XML-RPC API URI
    @type  caller_api: str
    @return: (code, statusMessage, subscriberApis).
    List of current subscribers of topic in the form of XMLRPC URIs.
    @rtype: (int, str, [str])
    """
    #NOTE: we need topic_type for getPublishedTopics.
    try:
        self.ps_lock.acquire()
        self.reg_manager.register_publisher(topic, caller_id, caller_api)
        # don't let '*' type squash valid typing
        if topic_type != rosgraph.names.ANYTYPE or not topic in self.topics_types:
            self.topics_types[topic] = topic_type
        pub_uris = self.publishers.get_apis(topic)
        sub_uris = self.subscribers.get_apis(topic)
        self._notify_topic_subscribers(topic, pub_uris, sub_uris)
        mloginfo("+PUB [%s] %s %s",topic, caller_id, caller_api)
        sub_uris = self.subscribers.get_apis(topic)
    finally:
        self.ps_lock.release()
    return 1, "Registered [%s] as publisher of [%s]"%(caller_id, topic), sub_uris

registerPublisher 接口的注释：Register the caller as a publisher the topic，将调用者注册为一个topic发布者

可以对应xmlrpc中对应参数，加上猜测：

caller_id：调用者，可以认为是节点，/test_sub，从及创建的节点

topic：发布的topic name，/rosout

topic_type：发布的topic数据类型，rosgraph_msgs/Log

caller_api：调用者发布数据的API接口，http://192.168.1.150:40209

总上，我们大概有几点猜测：

1. 接口在rosmaster中，接口是registerPublisher，表示，这是注册节点的
2. 告诉master节点，我创建了一个节点，节点名是/test_sub
3. 告诉master，这个节点需要发布topic，topic名是/rosout，数据类型是rosgraph_msgs/Log

registerPublisher 接口中有三个地方需要注意：

1. register_publisher接口调用
2. _notify_topic_subscribers接口调用，告知当前所有的subscriber，有新的publisher，他们需要再次到新的publisher中去订阅数据
3. return 内容，最后会拼接成xmlrpc的报文，response 回去，这也就顺便解释了第二条xmlrpc报文（response）

先看 register_publisher，代码在rosmaster中的registrations.py文件中

def register_publisher(self, topic, caller_id, caller_api):
    """
    Register topic publisher
    @return: None
    """
    self._register(self.publishers, topic, caller_id, caller_api)

_register 接口，这个节点做了三件事

1. 调用内部接口保存节点信息

2. 如果这个节点之前已经存在，就表明它是在更新，则发布数据的接口改变，且之前已经有订阅，则此时所有订阅该接口的所有subscriber解除订阅

3. 调用register接口保存

   def _register(self, r, key, caller_id, caller_api, service_api=None):
   # update node information
   node_ref, changed = self._register_node_api(caller_id, caller_api)
   node_ref.add(r.type, key)
   # update pub/sub/service indicies
   if changed:
   self.publishers.unregister_all(caller_id)
   self.subscribers.unregister_all(caller_id)
   self.services.unregister_all(caller_id)
   self.param_subscribers.unregister_all(caller_id)
   r.register(key, caller_id, caller_api, service_api)

_register_node_api 接口，我们可以看到，它主要做两件事

1. 更新master中节点信息（节点名、节点发布数据的接口）

2. 检查这个节点是不是已经存在，如果是，则告诉调用者

   def _register_node_api(self, caller_id, caller_api):
   """
   @param caller_id: caller_id of provider
   @type caller_id: str
   @param caller_api: caller_api of provider
   @type caller_api: str
   @return: (registration_information, changed_registration). changed_registration is true if
   caller_api is differet than the one registered with caller_id
   @rtype: (NodeRef, bool)
   """
   node_ref = self.nodes.get(caller_id, None)

   bumped_api = None
   if node_ref is not None:
       if node_ref.api == caller_api:
           return node_ref, False
       else:
           bumped_api = node_ref.api
           self.thread_pool.queue_task(bumped_api, shutdown_node_task,
                                       (bumped_api, caller_id, "new node registered with same name"))
   
   node_ref = NodeRef(caller_id, caller_api)
   self.nodes[caller_id] = node_ref
   return (node_ref, bumped_api != None)

_notify_topic_subscribers 代码，根据注释说明，接口的作用就是通知所有的subscriber，有新的publisher

def _notify_topic_subscribers(self, topic, pub_uris, sub_uris):
    """
    Notify subscribers with new publisher list
    @param topic: name of topic
    @type  topic: str
    @param pub_uris: list of URIs of publishers.
    @type  pub_uris: [str]
    """
    self._notify(self.subscribers, publisher_update_task, topic, pub_uris, sub_uris)

_notify 代码，将更新的通知任务（publisher_update_task）放进事件队列中，等待执行：

def _notify(self, registrations, task, key, value, node_apis):
    """
    Generic implementation of callback notification
    @param registrations: Registrations
    @type  registrations: L{Registrations}
    @param task: task to queue
    @type  task: fn
    @param key: registration key
    @type  key: str
    @param value: value to pass to task
    @type  value: Any
    """
    # cache thread_pool for thread safety
    thread_pool = self.thread_pool
    if not thread_pool:
        return

    try:
        for node_api in node_apis:
            # use the api as a marker so that we limit one thread per subscriber
            thread_pool.queue_task(node_api, task, (node_api, key, value))
    except KeyError:
        _logger.warn('subscriber data stale (key [%s], listener [%s]): node API unknown'%(key, s))

publisher_update_task 代码，传入的三个参数分别是：新节点的接口、topic名称、订阅者的接口：

def publisher_update_task(api, topic, pub_uris):
    """
    Contact api.publisherUpdate with specified parameters
    @param api: XML-RPC URI of node to contact
    @type  api: str
    @param topic: Topic name to send to node
    @type  topic: str
    @param pub_uris: list of publisher APIs to send to node
    @type  pub_uris: [str]
    """
    msg = "publisherUpdate[%s] -> %s %s" % (topic, api, pub_uris)
    mloginfo(msg)
    start_sec = time.time()
    try:
        #TODO: check return value for errors so we can unsubscribe if stale
        ret = xmlrpcapi(api).publisherUpdate('/master', topic, pub_uris)
        msg_suffix = "result=%s" % ret
    except Exception as ex:
        msg_suffix = "exception=%s" % ex
        raise
    finally:
        delta_sec = time.time() - start_sec
        mloginfo("%s: sec=%0.2f, %s", msg, delta_sec, msg_suffix)

publisherUpdate 接口在 rospy 模块的 masterslave.py 文件中，猜测是使用XMLRPC协议，通知所有的订阅者节点，发布者更新了

@apivalidate(-1, (is_topic('topic'), is_publishers_list('publishers')))
def publisherUpdate(self, caller_id, topic, publishers):
    """
    Callback from master of current publisher list for specified topic.
    @param caller_id: ROS caller id
    @type  caller_id: str
    @param topic str: topic name
    @type  topic: str
    @param publishers: list of current publishers for topic in the form of XMLRPC URIs
    @type  publishers: [str]
    @return: [code, status, ignore]
    @rtype: [int, str, int]
    """
    if self.reg_man:
        for uri in publishers:
            self.reg_man.publisher_update(topic, publishers)
    return 1, "", 0

2、hasParam

<?xml version='1.0'?>
<methodResponse>
    <params>
        <param>
            <value>
                <array>
                    <data>
                        <value>
                            <int>1</int>
                        </value>
                        <value>
                            <string>/use_sim_time</string>
                        </value>
                        <value>
                            <boolean>0</boolean>
                        </value>
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodResponse>


<?xml version="1.0"?>
<methodCall>
    <methodName>requestTopic</methodName>
    <params>
        <param>
            <value>/rosout</value>
        </param>
        <param>
            <value>/rosout</value>
        </param>
        <param>
            <value>
                <array>
                    <data>
                        <value>
                            <array>
                                <data>
                                    <value>TCPROS</value>
                                </data>
                            </array>
                        </value>
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodCall>

先说结论：

调用 hasParam 接口，检查参数服务器是否有参数 /test_sub/use_sim_time

返回值是 [1, /use_sim_time, 0]，表示没有

再看代码：

有了上面的分析经验，我们可以很轻松地的出结论，这是在调用 hasParam 接口

我们可以很轻松地找到这个方法的代码，在rosmaster模块的master_api.py文件中

@apivalidate(False, (non_empty_str('key'),))
def hasParam(self, caller_id, key):
    """
    Check if parameter is stored on server.
    @param caller_id str: ROS caller id
    @type  caller_id: str
    @param key: parameter to check
    @type  key: str
    @return: [code, statusMessage, hasParam]
    @rtype: [int, str, bool]
    """
    key = resolve_name(key, caller_id)
    if self.param_server.has_param(key):
        return 1, key, True
    else:
        return 1, key, False

根据协议

1. caller_id 传参是 /test_sub
2. key 传参是 /use_sim_time

根据注释和代码，我们可以确认，这个就口就是在检查，参数服务器是否有参数 /test_sub/use_sim_time

resolve_name 接口接收两个参数，根据调用：

1. name是/use_sim_time
2. namespace_是/test_sub

所以，才能确认上面的全局参数 /test_sub/use_sim_time

hasParam 接口的返回值有三个

1. code，整型，这里无论有没有，都返回1，可以忽略
2. key，这里就是 /use_sim_time
3. hasParam，表示是否有这个参数，True/False

根据 response 报文，这里应该返回 [1, /use_sim_time, 0]，表示没有这个参数

想要理解为什么要调用这个接口，就要理解 use_sim_time 参数的作用

use_sim_time是一个重要的参数，它默认值为false，可以配合Rosbag使用，是一个很重要的离线调试工具

我们都知道，ROS 中的时间有两种：

1. ROS::Time()
2. ROS::WallTime()

ROS::Time()和ROS::WallTime()

表示ROS网络中的时间，如果当时在非仿真环境里运行，那它就是当前的时间。但是假设去回放当时的情况，那就需要把当时的时间录下来

以控制为例，很多的数据处理需要知道当时某一个时刻发生了什么。Wall Time可以理解为墙上时间，墙上挂着的时间没有人改变的了，永远在往前走；ROS Time可以被人为修改，你可以暂停它，可以加速，可以减速，但是Wall Time不可以。

在开启一个Node之前，当把use_sim_time设置为true时，这个节点会从clock Topic获得时间。所以操作这个clock的发布者，可以实现一个让Node中得到ROS Time暂停、加速、减速的效果。同时下面这些方面都是跟Node透明的，所以非常适合离线的调试方式。当把ROSbag记下来以后重新play出来时，加两个横杠，--clock，它就会发布出这个消息

3、registerSubscriber

先看报文：

<?xml version="1.0"?>
<methodCall>
    <methodName>registerSubscriber</methodName>
    <params>
        <param>
            <value>/test_sub</value>
        </param>
        <param>
            <value>/ros_message</value>
        </param>
        <param>
            <value>my_package/MessageDefine</value>
        </param>
        <param>
            <value>http://192.168.1.150:43597</value>
        </param>
    </params>
</methodCall>


<?xml version='1.0'?>
<methodResponse>
    <params>
        <param>
            <value>
                <array>
                    <data>
                        <value>
                            <int>1</int>
                        </value>
                        <value>
                            <string>Subscribed to [/ros_message]</string>
                        </value>
                        <value>
                            <array>
                                <data>
                                    <value>
                                        <string>http://sherlock:41689/</string>
                                    </value>
                                </data>
                            </array>
                        </value>
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodResponse>

registerSubscriber 代码在rosmaster包中的master_api.py文件中，如下：

@apivalidate([], ( is_topic('topic'), valid_type_name('topic_type'), is_api('caller_api')))
def registerSubscriber(self, caller_id, topic, topic_type, caller_api):
    """
    Subscribe the caller to the specified topic. In addition to receiving
    a list of current publishers, the subscriber will also receive notifications
    of new publishers via the publisherUpdate API.
    @param caller_id: ROS caller id
    @type  caller_id: str
    @param topic str: Fully-qualified name of topic to subscribe to.
    @param topic_type: Datatype for topic. Must be a package-resource name, i.e. the .msg name.
    @type  topic_type: str
    @param caller_api: XML-RPC URI of caller node for new publisher notifications
    @type  caller_api: str
    @return: (code, message, publishers). Publishers is a list of XMLRPC API URIs
       for nodes currently publishing the specified topic.
    @rtype: (int, str, [str])
    """
    #NOTE: subscribers do not get to set topic type
    try:
        self.ps_lock.acquire()
        self.reg_manager.register_subscriber(topic, caller_id, caller_api)

        # ROS 1.1: subscriber can now set type if it is not already set
        #  - don't let '*' type squash valid typing
        if not topic in self.topics_types and topic_type != rosgraph.names.ANYTYPE:
            self.topics_types[topic] = topic_type

        mloginfo("+SUB [%s] %s %s",topic, caller_id, caller_api)
        pub_uris = self.publishers.get_apis(topic)
    finally:
        self.ps_lock.release()
    return 1, "Subscribed to [%s]"%topic, pub_uris

根据协议往来，我们可以看到调用过程

registerSubscriber("/test_sub", "/ros_message", "my_package/MessageDefine", "http://192.168.1.150:43597")

入参有4个：

1. 订阅节点名：/test_sub
2. 需要订阅的topic 名称：/ros_message
3. topic的数据类型：my_package/MessageDefine
4. 订阅节点自己的uri，即发布者通知时的发送目标

返回值有3个：

1. code，这里固定是1

2. message，这里是 Subscribed to [/ros_message]

3. publisher 的订阅URI 列表，因为这里只有一个publisher，所以只有一个 http://sherlock:41689/，首先，这可能是主机里面某个几点的uri，需要从机去订阅

代码说明：

和第一条，注册publisher相反，这里是注册subscriber

有几个关键代码

register_subscriber 代码，位于rosmaster包中的registerations.py文件中：

def register_subscriber(self, topic, caller_id, caller_api):
    """
    Register topic subscriber
    @return: None
    """
    self._register(self.subscribers, topic, caller_id, caller_api)

_register 代码，调用 _register_node_api 接口更新节点信息，如果之前有该节点的注册信息，则先删除：

def _register(self, r, key, caller_id, caller_api, service_api=None):
    # update node information
    node_ref, changed = self._register_node_api(caller_id, caller_api)
    node_ref.add(r.type, key)
    # update pub/sub/service indicies
    if changed:
        self.publishers.unregister_all(caller_id)
        self.subscribers.unregister_all(caller_id)
        self.services.unregister_all(caller_id)
        self.param_subscribers.unregister_all(caller_id)
    r.register(key, caller_id, caller_api, service_api)

_register_node_api 代码：

def _register_node_api(self, caller_id, caller_api):
    """
    @param caller_id: caller_id of provider
    @type  caller_id: str
    @param caller_api: caller_api of provider
    @type  caller_api: str
    @return: (registration_information, changed_registration). changed_registration is true if
    caller_api is differet than the one registered with caller_id
    @rtype: (NodeRef, bool)
    """
    node_ref = self.nodes.get(caller_id, None)

    bumped_api = None
    if node_ref is not None:
        if node_ref.api == caller_api:
            return node_ref, False
        else:
            bumped_api = node_ref.api
            self.thread_pool.queue_task(bumped_api, shutdown_node_task,
                                        (bumped_api, caller_id, "new node registered with same name"))

    node_ref = NodeRef(caller_id, caller_api)
    self.nodes[caller_id] = node_ref
    return (node_ref, bumped_api != None)

shutdown_node_task 代码，如果订阅节点退出了，则需要通知：

def shutdown_node_task(api, caller_id, reason):
    """
    Method to shutdown another ROS node. Generally invoked within a
    separate thread as this is used to cleanup hung nodes.

    @param api: XML-RPC API of node to shutdown
    @type  api: str
    @param caller_id: name of node being shutdown
    @type  caller_id: str
    @param reason: human-readable reason why node is being shutdown
    @type  reason: str
    """
    try:
        xmlrpcapi(api).shutdown('/master', "[{}] Reason: {}".format(caller_id, reason))
    except:
        pass #expected in many common cases
    remove_server_proxy(api)

4、requestTopic

request报文body，如下，我们可以看到，xmlrpc发送的host是sherlock:41689，是上一步骤，收到的publisher的uri，如果有多个publisher，要request多次

POST /RPC2 HTTP/1.1
Host: sherlock:41689
User-Agent: Go-http-client/1.1
Content-Length: 307
Content-Type: text/xml
Accept-Encoding: gzip

<?xml version="1.0"?>
<methodCall>
    <methodName>requestTopic</methodName>
    <params>
        <param>
            <value>/test_sub</value>
        </param>
        <param>
            <value>/ros_message</value>
        </param>
        <param>
            <value>
                <array>
                    <data>
                        <value>
                            <array>
                                <data>
                                    <value>TCPROS</value>
                                </data>
                            </array>
                        </value>
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodCall>

response报文

<?xml version="1.0"?>
<methodResponse>
    <params>
        <param>
            <value>
                <array>
                    <data>
                        <value>
                            <i4>1</i4>
                        </value>
                        <value></value>
                        <value>
                            <array>
                                <data>
                                    <value>TCPROS</value>
                                    <value>sherlock</value>
                                    <value>
                                        <i4>33173</i4>
                                    </value>
                                </data>
                            </array>
                        </value>
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodResponse>

根据协议，调用过程如下：

requestTopic("/test_sub", "/ros_message", ["TCPROS"])

告诉publisher，subscriber准备好了，可以发数据了

_remap_table['requestTopic'] = [0] # remap topic
@apivalidate([], (is_topic('topic'), non_empty('protocols')))
def requestTopic(self, caller_id, topic, protocols):
    """
    Publisher node API method called by a subscriber node.

    Request that source allocate a channel for communication. Subscriber provides
    a list of desired protocols for communication. Publisher returns the
    selected protocol along with any additional params required for
    establishing connection. For example, for a TCP/IP-based connection,
    the source node may return a port number of TCP/IP server.
    @param caller_id str: ROS caller id
    @type  caller_id: str
    @param topic: topic name
    @type  topic: str
    @param protocols: list of desired
     protocols for communication in order of preference. Each
     protocol is a list of the form [ProtocolName,
     ProtocolParam1, ProtocolParam2...N]
    @type  protocols: [[str, XmlRpcLegalValue*]]
    @return: [code, msg, protocolParams]. protocolParams may be an
    empty list if there are no compatible protocols.
    @rtype: [int, str, [str, XmlRpcLegalValue*]]
    """
    if not get_topic_manager().has_publication(topic):
        return -1, "Not a publisher of [%s]"%topic, []
    for protocol in protocols: #simple for now: select first implementation
        protocol_id = protocol[0]
        for h in self.protocol_handlers:
            if h.supports(protocol_id):
                _logger.debug("requestTopic[%s]: choosing protocol %s", topic, protocol_id)
                return h.init_publisher(topic, protocol)
    return 0, "no supported protocol implementations", []

init_publisher 代码

def init_publisher(self, resolved_name, protocol):
    """
    Initialize this node to receive an inbound TCP connection,
    i.e. startup a TCP server if one is not already running.

    @param resolved_name: topic name
    @type  resolved__name: str

    @param protocol: negotiated protocol
      parameters. protocol[0] must be the string 'TCPROS'
    @type  protocol: [str, value*]
    @return: (code, msg, [TCPROS, addr, port])
    @rtype: (int, str, list)
    """
    if protocol[0] != TCPROS:
        return 0, "Internal error: protocol does not match TCPROS: %s"%protocol, []
    start_tcpros_server()
    addr, port = get_tcpros_server_address()
    return 1, "ready on %s:%s"%(addr, port), [TCPROS, addr, port]

publisher 检查，是否支持指定协议，如果不支持，则返回1，否则返回0

返回值的第二个参数有三个值，分别是协议类型、ip地址 和 端口

5、unregisterSubscriber

<?xml version="1.0"?>
<methodCall>
    <methodName>unregisterSubscriber</methodName>
    <params>
        <param>
            <value>/test_sub</value>
        </param>
        <param>
            <value>/ros_message</value>
        </param>
        <param>
            <value>http://192.168.1.150:43597</value>
        </param>
    </params>
</methodCall>


<?xml version='1.0'?>
<methodResponse>
    <params>
        <param>
            <value>
                <array>
                    <data>
                        <value>
                            <int>1</int>
                        </value>
                        <value>
                            <string>Unregistered [/test_sub] as provider of [/ros_message]</string>
                        </value>
                        <value>
                            <int>1</int>
                        </value>
                    </data>
                </array>
            </value>
        </param>
    </params>
</methodResponse>


@apivalidate(0, (is_topic('topic'), is_api('caller_api')))
def unregisterSubscriber(self, caller_id, topic, caller_api):
    """
    Unregister the caller as a subscriber of the topic.
    @param caller_id: ROS caller id
    @type  caller_id: str
    @param topic: Fully-qualified name of topic to unregister.
    @type  topic: str
    @param caller_api: API URI of service to unregister. Unregistration will only occur if current
       registration matches.
    @type  caller_api: str
    @return: (code, statusMessage, numUnsubscribed).
      If numUnsubscribed is zero it means that the caller was not registered as a subscriber.
      The call still succeeds as the intended final state is reached.
    @rtype: (int, str, int)
    """
    try:
        self.ps_lock.acquire()
        retval = self.reg_manager.unregister_subscriber(topic, caller_id, caller_api)
        mloginfo("-SUB [%s] %s %s",topic, caller_id, caller_api)
        return retval
    finally:
        self.ps_lock.release()

取消订阅，固定返回1

6、TCP数据私有协议

首先保证主从机的数据类型一致，包括字段的顺序，实际ROS框架内是通过md5检测，保证数据类型一致的

数据传输前提：

1. 数据类型一致
2. 字段名一致
3. 字段顺序一致

数据传输模式：小端hex

数据域长度

数据域

4 byte

n byte

数据域长度固定4byte，长度不包括自身

数据域根据字段类型解析，ros 通信的内置数据类型有：

原始类型

字节数

bool

1

int8

1

uint8

1

int16

2

uint16

2

int32

4

uint32

4

int64

8

uint64

8

float32

4

float64

8

string

n(n > 4)

time

8

duration

8

数组

n(n > 4)

其中，除 string、time、duration 和 数组 类型外的其余类型，直接根据字节数读取即可

string

字符串类型，也可认为是字符数组（则可以和数组类型复用），因为是不定长度，所以需要知道字符串的长度，ROS中使用uint32类型表示长度/数组元素数量，即4byte

所以，如果出现字符串类型，则数据域为：

字符串长度

字符

4 byte

n byte

数组

数组类型，因为是不定长度，所以需要知道数组的元素数量，和string同理，ROS 中使用uint32类型表示数组的元素数量，再结合数组元素的类型，即可得到总长度

所以，出现数组类型，则数据域为：

数组元素数量

数组数据

4 byte

n byte

如果数组类型是 int32，则数组数据占 4 * n byte，其余类型以此类推

time

ROS 中把 time 单独提取作为基本数据类型，对应 ROS 中的 ros::Time 类，因为我们可以认为是嵌套类型

ros::Time 有两个字段：

1. sec： uint32
2. nsec： uint32

所以，time 类型在数据域占8byte，如果出现 time 类型，则数据域为：

sec

nsec

4 byte

4 byte

duration

duration 类型和 time 相同，在 ROS 中对应 ros::Duration 类，可以认为是嵌套类型

ros::Duration 有两个字段：

1. sec： uint32
2. nsec： uint32

所以，duration 类型在数据域中占8byte，如果出现 duration 类型，则数据域为：

sec

nsec

4 byte

4 byte

嵌套类型

嵌套类型可以认为是数据域的组合，如果发现字段类型不是内置数据类型，则可认为是嵌套类型，嵌套类型按照类型的字段，递归处理即可

示例1：

.msg 文件为：

int8 shutdown_time
string text

主机发出数据为：

shutdown_time = 123
text = abc

从机收到数据为：

08 00 00 00 7b 03 00 00 00 61 62 63

分析如下：

1. 包头4 byte表示数据与长度

   08 00 00 00，表示数据域长度为8，即后续数据总长度为8

2. 字段1为shutdown_time，类型是int8，1byte

   7b转10进制，为123

3. 字段2为text，类型是字符串 (4+n)byte

   4byte 长度：03 00 00 00，表示字符串长度为3，后面3byte 为字符串内容：61 62 63，ASCII转换为：abc）

示例2：

.msg 文件为：

Header header
int8 shutdown_time
int32 shutdown_time2
string text
float32 num
string text2
int8[] data
int16[] data2

Header的数据类型为：

uint32 seq
time stamp
string frame_id

主机发出数据为：

//header一般由ROS系统自己处理，这里写出来是为了方便观察
header.seq = 29;
header.time.sec = 0;
header.time.nsec = 0;
header.frame_id = "";
shutdown_time = 123;
shutdown_time2 = 987654;
text2 = "lmn";
text = "abc";
num = 23.4;
data = [1, 2, 4, 89];
data2 = [11, 22, 908]

从机收到的数据为：

39 00 00 00 1d 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 7b 06 12 0f 00 03 00 00 00 61 62 63 33 33 bb 41 03 00 00 00 6c 6d 6e 04 00 00 00 01 02 04 59 03 00 00 00 0b 00 16 00 8c 03

分析如下：

1. 包头4byte表示数据域长度

   0x39 00 00 00，10进制57，表明后续数据域长度57byte

2. 字段1，Header 类型，可以认为是嵌套类型，Header字段如下：

3. 字段2，shutdown_time，int8，1byte，数据为：0x7b，十进制123；

4. 字段3，shutdown_time2，int32，4byte，数据为：0x06 12 0f 00，十进制：987654；

5. 字段4，text，字符串：

6. 字段5，num，flota32，4byte，数据为：33 33 bb 41，十进制：23.4；

7. 字段6：text2，字符串：

8. 字段7，data，int8数组：

9. 字段8，data2，int16数组：

7、小结

宗上，整体的从机订阅时序图如下：