Support TCP/IP and UDS(Unix Domain Socket) hybrid transport for roscpp and rospy

Signed-off-by: Barry Xu <barry.xu@sony.com>

Author: Barry-Xu-2018

clients/roscpp/CMakeLists.txt

@@ -75,7 +75,13 @@ CHECK_CXX_SYMBOL_EXISTS(epoll_wait "sys/epoll.h" HAVE_EPOLL)
 configure_file(${CMAKE_CURRENT_SOURCE_DIR}/src/libros/config.h.in ${CMAKE_CURRENT_BINARY_DIR}/config.h)
 include_directories(${CMAKE_CURRENT_BINARY_DIR})
 
-add_library(roscpp 
+set(UDS_FEATURE_SRC
+  src/libros/transport/transport_uds.cpp
+  src/libros/transport/transport_uds_stream.cpp
+  src/libros/transport/transport_uds_datagram.cpp
+)
+
+add_library(roscpp
   src/libros/master.cpp
   src/libros/network.cpp
   src/libros/subscriber.cpp
@@ -124,6 +130,7 @@ add_library(roscpp
   src/libros/service.cpp
   src/libros/this_node.cpp
   src/libros/steady_timer.cpp
+  ${UDS_FEATURE_SRC}
   )
 
 if(WIN32)

clients/roscpp/include/ros/connection_manager.h

@@ -64,10 +64,17 @@ class ROSCPP_DECL ConnectionManager
   uint32_t getTCPPort();
   uint32_t getUDPPort();
 
+  const std::string getUDSStreamPath();
+  const std::string getUDSDatagramPath();
+
   const TransportTCPPtr& getTCPServerTransport() { return tcpserver_transport_; }
   const TransportUDPPtr& getUDPServerTransport() { return udpserver_transport_; }
 
+  const TransportUDSStreamPtr& getUDSStreamServerTransport() { return uds_stream_server_transport_; }
+  const TransportUDSDatagramPtr& getUDSDatagramServerTransport() { return uds_datagram_server_transport_; }
+
   void udprosIncomingConnection(const TransportUDPPtr& transport, Header& header);
+  void udprosIncomingConnection(const TransportUDSDatagramPtr& transport, Header& header);
 
   void start();
   void shutdown();
@@ -81,6 +88,7 @@ class ROSCPP_DECL ConnectionManager
 
   bool onConnectionHeaderReceived(const ConnectionPtr& conn, const Header& header);
   void tcprosAcceptConnection(const TransportTCPPtr& transport);
+  void tcprosAcceptConnection(const TransportUDSStreamPtr& transport);
 
   PollManagerPtr poll_manager_;
 
@@ -99,6 +107,10 @@ class ROSCPP_DECL ConnectionManager
   TransportTCPPtr tcpserver_transport_;
   TransportUDPPtr udpserver_transport_;
 
+  // Unix Domain Socket (stream, datagram)
+  TransportUDSStreamPtr uds_stream_server_transport_;
+  TransportUDSDatagramPtr uds_datagram_server_transport_;
+
   const static int MAX_TCPROS_CONN_QUEUE = 100; // magic
 };
 

clients/roscpp/include/ros/forwards.h

@@ -43,6 +43,7 @@
 #include <ros/macros.h>
 #include "exceptions.h"
 #include "ros/datatypes.h"
+#include "ros/common.h"
 
 namespace ros
 {
@@ -59,6 +60,10 @@ class TransportTCP;
 typedef boost::shared_ptr<TransportTCP> TransportTCPPtr;
 class TransportUDP;
 typedef boost::shared_ptr<TransportUDP> TransportUDPPtr;
+class TransportUDSStream;
+typedef boost::shared_ptr<TransportUDSStream> TransportUDSStreamPtr;
+class TransportUDSDatagram;
+typedef boost::shared_ptr<TransportUDSDatagram> TransportUDSDatagramPtr;
 class Connection;
 typedef boost::shared_ptr<Connection> ConnectionPtr;
 typedef std::set<ConnectionPtr> S_Connection;

clients/roscpp/include/ros/network.h

@@ -41,6 +41,8 @@ namespace network
 {
 
 ROSCPP_DECL bool splitURI(const std::string& uri, std::string& host, uint32_t& port);
+ROSCPP_DECL bool splitURI(const std::string& uri, std::string& uds_path);
+ROSCPP_DECL bool isInternal(const std::string& uds_path, const std::string& hostname);
 ROSCPP_DECL const std::string& getHost();
 ROSCPP_DECL uint16_t getTCPROSPort();
 

clients/roscpp/include/ros/service_manager.h

@@ -99,6 +99,15 @@ class ROSCPP_DECL ServiceManager
    */
   bool lookupService(const std::string& name, std::string& serv_host, uint32_t& serv_port);
 
+  /** @brief Lookup the host/port, uds_path of a service.
+   *
+   * @param name The name of the service
+   * @param serv_host OUT -- The host of the service
+   * @param serv_port OUT -- The port of the service
+   * @param serv_uds_path OUT -- The UDS path of the service
+   */
+  bool lookupServiceExt(const std::string& name, std::string& serv_host, uint32_t& serv_port, std::string& serv_uds_path);
+
   /** @brief Unadvertise a service.
    *
    * This call unadvertises a service, which must have been previously

clients/roscpp/include/ros/subscription.h

@@ -119,7 +119,7 @@ class ROSCPP_DECL Subscription : public boost::enable_shared_from_this<Subscript
   uint32_t getNumCallbacks() const { return callbacks_.size(); }
   uint32_t getNumPublishers();
 
-  // We'll keep a list of these objects, representing in-progress XMLRPC 
+  // We'll keep a list of these objects, representing in-progress XMLRPC
   // connections to other nodes.
   class ROSCPP_DECL PendingConnection : public ASyncXMLRPCConnection
   {
@@ -131,13 +131,24 @@ class ROSCPP_DECL Subscription : public boost::enable_shared_from_this<Subscript
       , remote_uri_(remote_uri)
       {}
 
+      PendingConnection(XmlRpc::XmlRpcClient* client, TransportUDPPtr udp_transport, TransportUDSDatagramPtr uds_datagram_transport, const SubscriptionWPtr& parent, const std::string& remote_uri)
+      : client_(client)
+      , udp_transport_(udp_transport)
+      , uds_datagram_transport_(uds_datagram_transport)
+      , parent_(parent)
+      , remote_uri_(remote_uri)
+      {}
+
       ~PendingConnection()
       {
         delete client_;
       }
 
       XmlRpc::XmlRpcClient* getClient() const { return client_; }
       TransportUDPPtr getUDPTransport() const { return udp_transport_; }
+      TransportUDSDatagramPtr getUDSDatagramTransport() const { return uds_datagram_transport_; }
+      const std::string& getUDSPath() const { return uds_path_; }
+      void setUDSPath(const std::string& uds_path) { uds_path_ = uds_path; }
 
       virtual void addToDispatch(XmlRpc::XmlRpcDispatch* disp)
       {
@@ -172,11 +183,77 @@ class ROSCPP_DECL Subscription : public boost::enable_shared_from_this<Subscript
     private:
       XmlRpc::XmlRpcClient* client_;
       TransportUDPPtr udp_transport_;
+      TransportUDSDatagramPtr uds_datagram_transport_;
+      std::string uds_path_;
       SubscriptionWPtr parent_;
       std::string remote_uri_;
   };
   typedef boost::shared_ptr<PendingConnection> PendingConnectionPtr;
 
+  class ROSCPP_DECL PendingConnectionUDS : public ASyncXMLRPCConnection
+  {
+    public:
+      PendingConnectionUDS(XmlRpc::XmlRpcClient* client, const SubscriptionWPtr& parent, const std::string& remote_uri, PendingConnectionPtr pending_connection)
+      : client_(client)
+      , parent_(parent)
+      , remote_uri_(remote_uri)
+      , pending_connection_(pending_connection)
+      {}
+
+      ~PendingConnectionUDS()
+      {
+        delete client_;
+      }
+
+      virtual void addToDispatch(XmlRpc::XmlRpcDispatch* disp)
+      {
+        disp->addSource(client_, XmlRpc::XmlRpcDispatch::WritableEvent | XmlRpc::XmlRpcDispatch::Exception);
+      }
+
+      virtual void removeFromDispatch(XmlRpc::XmlRpcDispatch* disp)
+      {
+        disp->removeSource(client_);
+      }
+
+      virtual bool check()
+      {
+        SubscriptionPtr parent = parent_.lock();
+        if (!parent)
+        {
+          return true;
+        }
+
+        XmlRpc::XmlRpcValue result;
+        if (client_->executeCheckDone(result))
+        {
+
+          XmlRpc::XmlRpcValue proto;
+          // validate the requestTopicUds response data
+          if (XMLRPCManager::instance()->validateXmlrpcResponse("requestTopicUds", result, proto)) {
+            // protocol data contains (TCPROS, uds_path) or (UDPROS, uds_path, ...)
+            std::string proto_name = proto[0];
+            if (proto_name == std::string("TCPROS") ||
+                proto_name == std::string("UDPROS")) {
+              std::string uds_path = std::string(proto[1]);
+              pending_connection_->setUDSPath(uds_path);
+            }
+          }
+
+          XMLRPCManager::instance()->addASyncConnection(pending_connection_);
+          return true;
+        }
+
+        return false;
+      }
+
+    private:
+      XmlRpc::XmlRpcClient* client_;
+      SubscriptionWPtr parent_;
+      std::string remote_uri_;
+      PendingConnectionPtr pending_connection_;
+  };
+  typedef boost::shared_ptr<PendingConnectionUDS> PendingConnectionUDSPtr;
+
   void pendingConnectionDone(const PendingConnectionPtr& pending_conn, XmlRpc::XmlRpcValue& result);
 
   void getPublishTypes(bool& ser, bool& nocopy, const std::type_info& ti);

clients/roscpp/include/ros/topic_manager.h

@@ -152,7 +152,9 @@ class ROSCPP_DECL TopicManager
    *
    * @todo Consider making this private
    */
-  bool requestTopic(const std::string &topic, XmlRpc::XmlRpcValue &protos, XmlRpc::XmlRpcValue &ret);
+  bool requestTopic(const std::string &topic, XmlRpc::XmlRpcValue &protos, XmlRpc::XmlRpcValue &ret
+                    , bool uds_flag=false
+                    );
 
   // Must lock the advertised topics mutex before calling this function
   bool isTopicAdvertised(const std::string& topic);
@@ -211,6 +213,7 @@ class ROSCPP_DECL TopicManager
 
   void pubUpdateCallback(XmlRpc::XmlRpcValue& params, XmlRpc::XmlRpcValue& result);
   void requestTopicCallback(XmlRpc::XmlRpcValue& params, XmlRpc::XmlRpcValue& result);
+  void requestTopicUdsCallback(XmlRpc::XmlRpcValue& params, XmlRpc::XmlRpcValue& result);
   void getBusStatsCallback(XmlRpc::XmlRpcValue& params, XmlRpc::XmlRpcValue& result);
   void getBusInfoCallback(XmlRpc::XmlRpcValue& params, XmlRpc::XmlRpcValue& result);
   void getSubscriptionsCallback(XmlRpc::XmlRpcValue& params, XmlRpc::XmlRpcValue& result);

clients/roscpp/include/ros/transport/transport_uds.h

@@ -0,0 +1,81 @@
+/*
+ * Software License Agreement (BSD License)
+ *
+ *  Copyright (c) 2018, Open Source Robotics Foundation, Inc.
+ *  All rights reserved.
+ *
+ *  Redistribution and use in source and binary forms, with or without
+ *  modification, are permitted provided that the following conditions
+ *  are met:
+ *
+ *   * Redistributions of source code must retain the above copyright
+ *     notice, this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above
+ *     copyright notice, this list of conditions and the following
+ *     disclaimer in the documentation and/or other materials provided
+ *     with the distribution.
+ *   * Neither the name of Willow Garage, Inc. nor the names of its
+ *     contributors may be used to endorse or promote products derived
+ *     from this software without specific prior written permission.
+ *
+ *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
+ *  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
+ *  COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
+ *  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
+ *  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ *  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+ *  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ *  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
+ *  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ *  POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#ifndef ROSCPP_TRANSPORT_UDS_H
+#define ROSCPP_TRANSPORT_UDS_H
+
+#include <ros/transport/transport.h>
+
+/*! @name UDS Feature Bit Field
+    Unix Domain Socket Feature Bit Field controlled by environmental value from user.
+*/
+/* @{ */
+#define ROS_UDS_EXT_ABSTRACT_SOCK_NAME   0x00000001    /*!< enable abstract named socket */
+/* @} */
+
+#define ROS_UDS_EXT_IS_ENABLE(feature)   (TransportUDS::s_uds_feature_ & feature)
+
+namespace ros
+{
+
+/**
+ * \brief Abstract base class that allows abstraction of the transport type, eg. Unix domain socket(stream or datagram)
+ */
+class TransportUDS : public Transport
+{
+public:
+  static uint32_t s_uds_feature_;
+
+  /**
+   * \brief Returns the server UDS path this transport is using with
+   */
+  const std::string getServerUDSPath() { return server_uds_path_; }
+
+protected:
+  /**
+   * \brief Generate abstrace named socket".
+   */
+  const std::string generateServerUDSPath();
+  /**
+   * \brief Generate a string of UDS path like "${TMP}/ros-uds-[stream|datagram]-${PID}-${COUNTER}", the default value of ${TMP} is "/tmp".
+   */
+  const std::string generateServerUDSPath(uint32_t counter);
+
+  std::string server_type_;
+  std::string server_uds_path_;
+};
+
+}
+
+#endif // ROSCPP_TRANSPORT_UDS_H