Multi-uORB support changes - part 1

This adds support for a dynamic build for QuRT and initial
Multi-uORB changes to enable communication between the DSP and
the application processor.

This part of the changes do not affect the POSIX build. This is
enablement for the QuRT build using Multi-uORB. The second part
of the changes will be added in a new module under src/modules.

Signed-off-by: Mark Charlebois <charlebm@gmail.com>

src/modules/uORB/module.mk

@@ -46,9 +46,13 @@ SRCS			= uORBDevices_nuttx.cpp \
  
 else
 SRCS			= uORBDevices_posix.cpp \
-			  uORBTest_UnitTest.cpp \
 			  uORBManager_posix.cpp
 endif
+
+ifeq ($(PX4_TARGET_OS),posix)
+SRCS += uORBTest_UnitTest.cpp
+endif
+
 SRCS	+= 		  objects_common.cpp \
 			  Publication.cpp \
 			  Subscription.cpp \

src/modules/uORB/uORBCommunicator.hpp

@@ -0,0 +1,167 @@
+/****************************************************************************
+ *
+ *   Copyright (c) 2012-2015 PX4 Development Team. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. 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.
+ * 3. Neither the name PX4 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 _uORBCommunicator_hpp_
+#define _uORBCommunicator_hpp_
+
+#include <stdint.h>
+#include <string>
+
+namespace uORBCommunicator
+{
+	class IChannel;
+	class IChannelRxHandler;
+}
+
+/**
+ * Interface to enable remote subscriptions.  The implementor of this interface
+ * shall manage the communication channel. It can be fastRPC or tcp or ip.
+ */
+
+class uORBCommunicator::IChannel
+{
+public:
+
+	//=========================================================================
+	//     INTERFACES FOR Control messages over a channel.
+	//=========================================================================
+
+	/**
+	 * @brief Interface to notify the remote entity of interest of a
+	 * subscription for a message.
+	 *
+	 * @param messageName
+	 * 	This represents the uORB message name; This message name should be
+	 * 	globally unique.
+	 * @param msgRate
+	 * 	The max rate at which the subscriber can accept the messages.
+	 * @return
+	 * 	0 = success; This means the messages is successfully sent to the receiver
+	 * 		Note: This does not mean that the receiver as received it.
+	 *  otherwise = failure.
+	 */
+	virtual int16_t add_subscription( const std::string& messageName, int32_t msgRateInHz ) = 0;
+
+
+	/**
+	 * @brief Interface to notify the remote entity of removal of a subscription
+	 *
+	 * @param messageName
+	 * 	This represents the uORB message name; This message name should be
+	 * 	globally unique.
+	 * @return
+	 * 	0 = success; This means the messages is successfully sent to the receiver
+	 * 		Note: This does not necessarily mean that the receiver as received it.
+	 *  otherwise = failure.
+	 */
+	virtual int16_t remove_subscription( const std::string& messageName ) = 0;
+
+	/**
+	 * Register Message Handler.  This is internal for the IChannel implementer*
+	 */
+	virtual int16_t register_handler( uORBCommunicator::IChannelRxHandler* handler ) = 0;
+
+
+	//=========================================================================
+	//     INTERFACES FOR Data messages
+	//=========================================================================
+
+	/**
+	 * @brief Sends the data message over the communication link.
+	 * @param messageName
+	 * 	This represents the uORB message name; This message name should be
+	 * 	globally unique.
+	 * @param length
+	 * 	The length of the data buffer to be sent.
+	 * @param data
+	 * 	The actual data to be sent.
+	 * @return
+	 *  0 = success; This means the messages is successfully sent to the receiver
+	 * 		Note: This does not mean that the receiver as received it.
+	 *  otherwise = failure.
+	 */
+	virtual int16_t send_message( const std::string& messageName, int32_t length, uint8_t* data) = 0;
+};
+
+/**
+ * Class passed to the communication link implement to provide callback for received
+ * messages over a channel.
+ */
+class uORBCommunicator::IChannelRxHandler
+{
+public:
+
+	/**
+	 * Interface to process a received AddSubscription from remote.
+	 * @param messageName
+	 * 	This represents the uORB message Name; This message Name should be
+	 * 	globally unique.
+	 * @param msgRate
+	 * 	The max rate at which the subscriber can accept the messages.
+	 * @return
+	 *  0 = success; This means the messages is successfully handled in the
+	 *  	handler.
+	 *  otherwise = failure.
+	 */
+	virtual int16_t process_add_subscription( const std::string& messageName, int32_t msgRateInHz ) = 0;
+
+	/**
+	 * Interface to process a received control msg to remove subscription
+	 * @param messageName
+	 * 	This represents the uORB message Name; This message Name should be
+	 * 	globally unique.
+	 * @return
+	 *  0 = success; This means the messages is successfully handled in the
+	 *  	handler.
+	 *  otherwise = failure.
+	 */
+	virtual int16_t process_remove_subscription( const std::string& messageName ) = 0;
+
+	/**
+	 * Interface to process the received data message.
+	 * @param messageName
+	 * 	This represents the uORB message Name; This message Name should be
+	 * 	globally unique.
+	 * @param length
+	 * 	The length of the data buffer to be sent.
+	 * @param data
+	 * 	The actual data to be sent.
+	 * @return
+	 *  0 = success; This means the messages is successfully handled in the
+	 *  	handler.
+	 *  otherwise = failure.
+	 */
+	virtual int16_t process_received_message( const std::string& messageName, int32_t length, uint8_t* data ) = 0;
+};
+
+#endif /* _uORBCommunicator_hpp_ */

src/modules/uORB/uORBDevices_nuttx.cpp

@@ -36,10 +36,15 @@
 #include <fcntl.h>
 #include <errno.h>
 #include <nuttx/arch.h>
+#include <algorithm>
 #include "uORBDevices_nuttx.hpp"
 #include "uORBUtils.hpp"
+#include "uORBManager.hpp"
+#include "uORBCommunicator.hpp"
 #include <stdlib.h>
 
+std::map<std::string, uORB::DeviceNode*> uORB::DeviceMaster::_node_map;
+
 uORB::DeviceNode::DeviceNode
 (
     const struct orb_metadata *meta,
@@ -53,7 +58,9 @@ uORB::DeviceNode::DeviceNode
   _last_update(0),
   _generation(0),
   _publisher(0),
-  _priority(priority)
+  _priority(priority),
+  _IsRemoteSubscriberPresent( false ),
+  _subscriber_count(0)
 {
   // enable debug() calls
   _debug_enabled = true;
@@ -120,6 +127,8 @@ uORB::DeviceNode::open(struct file *filp)
 
     ret = CDev::open(filp);
 
+    add_internal_subscriber();
+
     if (ret != OK)
       delete sd;
 
@@ -142,7 +151,9 @@ uORB::DeviceNode::close(struct file *filp)
 
     if (sd != nullptr) {
       hrt_cancel(&sd->update_call);
+      remove_internal_subscriber();
       delete sd;
+      sd = nullptr;
     }
   }
 
@@ -295,7 +306,19 @@ uORB::DeviceNode::publish
     errno = EIO;
     return ERROR;
   }
-
+  /*
+   * if the write is successful, send the data over the Multi-ORB link
+   */
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( ch != nullptr )
+  {
+    if( ch->send_message( meta->o_name, meta->o_size, (uint8_t*)data ) != 0 )
+    {
+      warnx( "[uORB::DeviceNode::publish(%d)]: Error Sending [%s] topic data over comm_channel",
+             __LINE__, meta->o_name );
+      return ERROR;
+    }
+  }
   return OK;
 }
 
@@ -421,6 +444,79 @@ uORB::DeviceNode::update_deferred_trampoline(void *arg)
   node->update_deferred();
 }
 
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+void uORB::DeviceNode::add_internal_subscriber()
+{
+  _subscriber_count++;
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( ch != nullptr && _subscriber_count > 0 )
+  {
+    ch->add_subscription( _meta->o_name, 1 );
+  }
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+void uORB::DeviceNode::remove_internal_subscriber()
+{
+  _subscriber_count--;
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( ch != nullptr && _subscriber_count == 0 )
+  {
+    ch->remove_subscription( _meta->o_name );
+  }
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::DeviceNode::process_add_subscription(int32_t rateInHz )
+{
+  // if there is already data in the node, send this out to
+  // the remote entity.
+  // send the data to the remote entity.
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( _data != nullptr && ch != nullptr ) // _data will not be null if there is a publisher.
+  {
+    ch->send_message( _meta->o_name, _meta->o_size, _data);
+  }
+
+  return OK;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::DeviceNode::process_remove_subscription()
+{
+  return OK;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::DeviceNode::process_received_message(int32_t length, uint8_t* data)
+{
+  int16_t ret = -1;
+  if( length != (int32_t)(_meta->o_size) )
+  {
+    warnx( "[uORB::DeviceNode::process_received_message(%d)]Error:[%s] Received DataLength[%d] != ExpectedLen[%d]",
+           __LINE__, _meta->o_name, (int)length, (int)_meta->o_size );
+    return ERROR;
+  }
+
+  /* call the devnode write method with no file pointer */
+  ret = write(nullptr, (const char *)data, _meta->o_size);
+
+  if (ret < 0)
+    return ERROR;
+
+  if (ret != (int)_meta->o_size) {
+    errno = EIO;
+    return ERROR;
+  }
+
+  return OK;
+}
+
 uORB::DeviceMaster::DeviceMaster(Flavor f) :
   CDev((f == PUBSUB) ? "obj_master" : "param_master",
        (f == PUBSUB) ? TOPIC_MASTER_DEVICE_PATH : PARAM_MASTER_DEVICE_PATH),
@@ -509,6 +605,11 @@ uORB::DeviceMaster::ioctl(struct file *filp, int cmd, unsigned long arg)
           free((void *)objname);
           free((void *)devpath);
         }
+        else
+        {
+          // add to the node map;.
+          _node_map[std::string(nodepath)] = node;
+        }
 
         group_tries++;
 
@@ -529,3 +630,14 @@ uORB::DeviceMaster::ioctl(struct file *filp, int cmd, unsigned long arg)
     return CDev::ioctl(filp, cmd, arg);
   }
 }
+
+uORB::DeviceNode* uORB::DeviceMaster::GetDeviceNode( const std::string& nodepath )
+{
+  uORB::DeviceNode* rc = nullptr;
+  if( _node_map.find( nodepath ) != _node_map.end() )
+  {
+    rc = _node_map[nodepath];
+  }
+  return rc;
+}
+

src/modules/uORB/uORBDevices_nuttx.hpp

@@ -35,6 +35,8 @@
 #define _uORBDevices_nuttx_hpp_
 
 #include <stdint.h>
+#include <string>
+#include <map>
 #include "uORBCommon.hpp"
 
 
@@ -119,6 +121,43 @@ public:
       const void *data
   );
 
+  /**
+   * processes a request for add subscription from remote
+   * @param rateInHz
+   *   Specifies the desired rate for the message.
+   * @return
+   *   0 = success
+   *   otherwise failure.
+   */
+  int16_t process_add_subscription( int32_t rateInHz );
+
+  /**
+   * processes a request to remove a subscription from remote.
+   */
+  int16_t process_remove_subscription();
+
+  /**
+   * processed the received data message from remote.
+   */
+  int16_t process_received_message( int32_t length, uint8_t* data );
+
+  /**
+    * Add the subscriber to the node's list of subscriber.  If there is
+    * remote proxy to which this subscription needs to be sent, it will
+    * done via uORBCommunicator::IChannel interface.
+    * @param sd
+    *   the subscriber to be added.
+    */
+   void add_internal_subscriber();
+
+   /**
+    * Removes the subscriber from the list.  Also notifies the remote
+    * if there a uORBCommunicator::IChannel instance.
+    * @param sd
+    *   the Subscriber to be removed.
+    */
+   void remove_internal_subscriber();
+
 protected:
   virtual pollevent_t poll_state(struct file *filp);
   virtual void poll_notify_one(struct pollfd *fds, pollevent_t events);
@@ -147,6 +186,9 @@ private: // private class methods.
     return sd;
   }
 
+  bool    _IsRemoteSubscriberPresent;
+  int32_t _subscriber_count;
+
   /**
    * Perform a deferred update for a rate-limited subscriber.
    */
@@ -166,6 +208,10 @@ private: // private class methods.
    * @return    True if the topic should appear updated to the subscriber
    */
   bool      appears_updated(SubscriberData *sd);
+
+  // disable copy and assignment operators
+  DeviceNode( const DeviceNode& );
+  DeviceNode& operator=( const DeviceNode& );
 };
 
 /**
@@ -180,9 +226,11 @@ public:
   DeviceMaster(Flavor f);
   virtual ~DeviceMaster();
 
+  static uORB::DeviceNode* GetDeviceNode( const std::string& node_name );
   virtual int   ioctl(struct file *filp, int cmd, unsigned long arg);
 private:
   Flavor      _flavor;
+  static std::map<std::string, uORB::DeviceNode*> _node_map;
 };
 
 

src/modules/uORB/uORBDevices_posix.cpp

@@ -35,11 +35,16 @@
 #include <string.h>
 #include <fcntl.h>
 #include <errno.h>
+#include <algorithm>
 
 #include "uORBDevices_posix.hpp"
 #include "uORBUtils.hpp"
+#include "uORBManager.hpp"
+#include "uORBCommunicator.hpp"
 #include <stdlib.h>
 
+std::map<std::string, uORB::DeviceNode*> uORB::DeviceMaster::_node_map;
+
 
 uORB::DeviceNode::SubscriberData  *uORB::DeviceNode::filp_to_sd(device::file_t *filp)
 {
@@ -60,7 +65,8 @@ uORB::DeviceNode::DeviceNode(const struct orb_metadata *meta, const char *name,
   _last_update(0),
   _generation(0),
   _publisher(0),
-  _priority(priority)
+  _priority(priority),
+  _subscriber_count(0)
 {
   // enable debug() calls
   //_debug_enabled = true;
@@ -127,6 +133,8 @@ uORB::DeviceNode::open(device::file_t *filp)
 
     ret = VDev::open(filp);
 
+    add_internal_subscriber();
+
     if (ret != PX4_OK) {
       warnx("ERROR: VDev::open failed\n");
       delete sd;
@@ -153,7 +161,9 @@ uORB::DeviceNode::close(device::file_t *filp)
 
     if (sd != nullptr) {
       hrt_cancel(&sd->update_call);
+      remove_internal_subscriber();
       delete sd;
+      sd = nullptr;
     }
   }
 
@@ -280,7 +290,7 @@ uORB::DeviceNode::ioctl(device::file_t *filp, int cmd, unsigned long arg)
 }
 
 ssize_t
-uORB::DeviceNode::publish(const orb_metadata *meta, orb_advert_t handle, const void *data)
+uORB::DeviceNode::publish(const orb_metadata *meta, orb_advert_t handle, const void *data )
 {
   //warnx("uORB::DeviceNode::publish meta = %p", meta);
 
@@ -310,6 +320,19 @@ uORB::DeviceNode::publish(const orb_metadata *meta, orb_advert_t handle, const v
     return ERROR;
   }
 
+  /*
+   * if the write is successful, send the data over the Multi-ORB link
+   */
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( ch != nullptr )
+  {
+    if( ch->send_message( meta->o_name, meta->o_size, (uint8_t*)data ) != 0 )
+    {
+      warnx( "[uORB::DeviceNode::publish(%d)]: Error Sending [%s] topic data over comm_channel",
+             __LINE__, meta->o_name );
+      return ERROR;
+    }
+  }
   return PX4_OK;
 }
 
@@ -437,6 +460,80 @@ uORB::DeviceNode::update_deferred_trampoline(void *arg)
   node->update_deferred();
 }
 
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+void uORB::DeviceNode::add_internal_subscriber()
+{
+  _subscriber_count++;
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( ch != nullptr && _subscriber_count > 0 )
+  {
+    ch->add_subscription( _meta->o_name, 1 );
+  }
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+void uORB::DeviceNode::remove_internal_subscriber()
+{
+  _subscriber_count--;
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( ch != nullptr && _subscriber_count == 0 )
+  {
+    ch->remove_subscription( _meta->o_name );
+  }
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::DeviceNode::process_add_subscription(int32_t rateInHz )
+{
+  // if there is already data in the node, send this out to
+  // the remote entity.
+  // send the data to the remote entity.
+  uORBCommunicator::IChannel* ch = uORB::Manager::get_instance()->get_uorb_communicator();
+  if( _data != nullptr && ch != nullptr ) // _data will not be null if there is a publisher.
+  {
+    ch->send_message( _meta->o_name, _meta->o_size, _data);
+  }
+
+  return 0;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::DeviceNode::process_remove_subscription()
+{
+  return 0;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::DeviceNode::process_received_message(int32_t length, uint8_t* data)
+{
+  int16_t ret = -1;
+  if( length != (int32_t)(_meta->o_size) )
+  {
+    warnx( "[uORB::DeviceNode::process_received_message(%d)]Error:[%s] Received DataLength[%d] != ExpectedLen[%d]",
+           __LINE__, _meta->o_name, (int)length, (int)_meta->o_size );
+    return ERROR;
+  }
+
+  /* call the devnode write method with no file pointer */
+  ret = write(nullptr, (const char *)data, _meta->o_size);
+
+  if (ret < 0)
+    return ERROR;
+
+  if (ret != (int)_meta->o_size) {
+    errno = EIO;
+    return ERROR;
+  }
+
+  return PX4_OK;
+}
+
+
 uORB::DeviceMaster::DeviceMaster(Flavor f) :
   VDev((f == PUBSUB) ? "obj_master" : "param_master",
        (f == PUBSUB) ? TOPIC_MASTER_DEVICE_PATH : PARAM_MASTER_DEVICE_PATH),
@@ -528,6 +625,12 @@ uORB::DeviceMaster::ioctl(device::file_t *filp, int cmd, unsigned long arg)
           free((void *)objname);
           free((void *)devpath);
         }
+        else
+        {
+          // add to the node map;.
+          _node_map[std::string(nodepath)] = node;
+        }
+
 
         group_tries++;
 
@@ -549,3 +652,12 @@ uORB::DeviceMaster::ioctl(device::file_t *filp, int cmd, unsigned long arg)
   }
 }
 
+uORB::DeviceNode* uORB::DeviceMaster::GetDeviceNode( const std::string& nodepath )
+{
+  uORB::DeviceNode* rc = nullptr;
+  if( _node_map.find( nodepath ) != _node_map.end() )
+  {
+    rc = _node_map[nodepath];
+  }
+  return rc;
+}

src/modules/uORB/uORBDevices_posix.hpp

@@ -35,9 +35,10 @@
 #define _uORBDevices_posix_hpp_
 
 #include <stdint.h>
+#include <string>
+#include <map>
 #include "uORBCommon.hpp"
 
-
 namespace uORB
 {
    class DeviceNode;
@@ -56,7 +57,44 @@ public:
   virtual ssize_t   write(device::file_t *filp, const char *buffer, size_t buflen);
   virtual int   ioctl(device::file_t *filp, int cmd, unsigned long arg);
 
-  static ssize_t    publish(const orb_metadata *meta, orb_advert_t handle, const void *data);
+  static ssize_t    publish(const orb_metadata *meta, orb_advert_t handle, const void *data );
+
+  /**
+   * processes a request for add subscription from remote
+   * @param rateInHz
+   *   Specifies the desired rate for the message.
+   * @return
+   *   0 = success
+   *   otherwise failure.
+   */
+  int16_t process_add_subscription( int32_t rateInHz );
+
+  /**
+   * processes a request to remove a subscription from remote.
+   */
+  int16_t process_remove_subscription();
+
+  /**
+   * processed the received data message from remote.
+   */
+  int16_t process_received_message( int32_t length, uint8_t* data );
+
+  /**
+    * Add the subscriber to the node's list of subscriber.  If there is
+    * remote proxy to which this subscription needs to be sent, it will
+    * done via uORBCommunicator::IChannel interface.
+    * @param sd
+    *   the subscriber to be added.
+    */
+   void add_internal_subscriber();
+
+   /**
+    * Removes the subscriber from the list.  Also notifies the remote
+    * if there a uORBCommunicator::IChannel instance.
+    * @param sd
+    *   the Subscriber to be removed.
+    */
+   void remove_internal_subscriber();
 
 protected:
   virtual pollevent_t poll_state(device::file_t *filp);
@@ -81,6 +119,8 @@ private:
 
   SubscriberData    *filp_to_sd(device::file_t *filp);
 
+  int32_t _subscriber_count;
+
   /**
    * Perform a deferred update for a rate-limited subscriber.
    */
@@ -100,6 +140,11 @@ private:
    * @return    True if the topic should appear updated to the subscriber
    */
   bool      appears_updated(SubscriberData *sd);
+
+
+  // disable copy and assignment operators
+  DeviceNode( const DeviceNode& );
+  DeviceNode& operator=( const DeviceNode& );
 };
 
 /**
@@ -114,9 +159,12 @@ public:
   DeviceMaster(Flavor f);
   ~DeviceMaster();
 
+  static uORB::DeviceNode* GetDeviceNode( const std::string& node_name );
+
   virtual int   ioctl(device::file_t *filp, int cmd, unsigned long arg);
 private:
   Flavor      _flavor;
+  static std::map<std::string, uORB::DeviceNode*> _node_map;
 };
 
 #endif /* _uORBDeviceNode_posix.hpp */

src/modules/uORB/uORBMain.cpp

@@ -34,9 +34,12 @@
 #include <string.h>
 #include "uORBDevices.hpp"
 #include "uORB.h"
-#include "uORBTest_UnitTest.hpp"
 #include "uORBCommon.hpp"
 
+#ifndef __PX4_QURT
+#include "uORBTest_UnitTest.hpp"
+#endif
+
 extern "C" { __EXPORT int uorb_main(int argc, char *argv[]); }
 
 static uORB::DeviceMaster *g_dev = nullptr;
@@ -85,6 +88,7 @@ uorb_main(int argc, char *argv[])
     return OK;
   }
 
+#ifndef __PX4_QURT
   /*
    * Test the driver/device.
    */
@@ -108,6 +112,7 @@ uorb_main(int argc, char *argv[])
       return t.latency_test<struct orb_test>(ORB_ID(orb_test), true);
     }
   }
+#endif
 
   /*
    * Print driver information.

src/modules/uORB/uORBManager.hpp

@@ -35,7 +35,13 @@
 #define _uORBManager_hpp_
 
 #include "uORBCommon.hpp"
+#include "uORBDevices.hpp"
+#include <map>
+#include <string>
 #include <stdint.h>
+#include <set>
+
+#include "uORBCommunicator.hpp"
 
 namespace uORB
 {
@@ -47,7 +53,7 @@ namespace uORB
  * uORB nodes for each uORB topics and also implements the behavor of the
  * uORB Api's.
  */
-class uORB::Manager
+class uORB::Manager  : public uORBCommunicator::IChannelRxHandler
 {
  public:
   // public interfaces for this class.
@@ -281,6 +287,27 @@ class uORB::Manager
    */
   int  orb_set_interval(int handle, unsigned interval) ;
 
+  /**
+   * Method to set the uORBCommunicator::IChannel instance.
+   * @param comm_channel
+   *  The IChannel instance to talk to remote proxies.
+   * @note:
+   *  Currently this call only supports the use of one IChannel
+   *  Future extensions may include more than one IChannel's.
+   */
+  void set_uorb_communicator(uORBCommunicator::IChannel* comm_channel);
+
+  /**
+   * Gets the uORB Communicator instance.
+   */
+  uORBCommunicator::IChannel* get_uorb_communicator( void );
+
+  /**
+   * Utility method to check if there is a remote subscriber present
+   * for a given topic
+   */
+  bool is_remote_subscriber_present( const std::string& messageName );
+
  private: // class methods
   /**
    * Advertise a node; don't consider it an error if the node has
@@ -316,10 +343,57 @@ class uORB::Manager
 
  private: // data members
   static Manager _Instance;
+  // the communicator channel instance.
+  uORBCommunicator::IChannel* _comm_channel;
+  std::set<std::string> _remote_subscriber_topics;
 
  private: //class methods
   Manager();
 
+  /**
+     * Interface to process a received AddSubscription from remote.
+     * @param messageName
+     *  This represents the uORB message Name; This message Name should be
+     *  globally unique.
+     * @param msgRate
+     *  The max rate at which the subscriber can accept the messages.
+     * @return
+     *  0 = success; This means the messages is successfully handled in the
+     *    handler.
+     *  otherwise = failure.
+     */
+    virtual int16_t process_add_subscription(const std::string& messageName,
+                                             int32_t msgRateInHz);
+
+    /**
+     * Interface to process a received control msg to remove subscription
+     * @param messageName
+     *  This represents the uORB message Name; This message Name should be
+     *  globally unique.
+     * @return
+     *  0 = success; This means the messages is successfully handled in the
+     *    handler.
+     *  otherwise = failure.
+     */
+    virtual int16_t process_remove_subscription(const std::string& messageName);
+
+    /**
+     * Interface to process the received data message.
+     * @param messageName
+     *  This represents the uORB message Name; This message Name should be
+     *  globally unique.
+     * @param length
+     *  The length of the data buffer to be sent.
+     * @param data
+     *  The actual data to be sent.
+     * @return
+     *  0 = success; This means the messages is successfully handled in the
+     *    handler.
+     *  otherwise = failure.
+     */
+    virtual int16_t process_received_message(const std::string& messageName,
+                                             int32_t length, uint8_t* data);
+
 };
 
 #endif /* _uORBManager_hpp_ */

src/modules/uORB/uORBManager_nuttx.cpp

@@ -39,7 +39,6 @@
 #include <errno.h>
 #include "uORBUtils.hpp"
 #include "uORBManager.hpp"
-#include "uORBDevices.hpp"
 
 
 //=========================  Static initializations =================
@@ -55,6 +54,7 @@ uORB::Manager* uORB::Manager::get_instance()
 //-----------------------------------------------------------------------------
 //-----------------------------------------------------------------------------
 uORB::Manager::Manager()
+: _comm_channel( nullptr )
 {
 }
 
@@ -280,3 +280,103 @@ int uORB::Manager::node_open
   /* everything has been OK, we can return the handle now */
   return fd;
 }
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+void uORB::Manager::set_uorb_communicator( uORBCommunicator::IChannel* channel)
+{
+  _comm_channel = channel;
+  if (_comm_channel != nullptr) {
+    _comm_channel->register_handler(this);
+  }
+}
+
+uORBCommunicator::IChannel* uORB::Manager::get_uorb_communicator( void )
+{
+  return _comm_channel;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::Manager::process_add_subscription(const std::string& messageName,
+                                                int32_t msgRateInHz)
+{
+  warnx("[posix-uORB::Manager::process_add_subscription(%d)] entering Manager_process_add_subscription: name: %s",
+          __LINE__, messageName.c_str() );
+  int16_t rc = 0;
+  _remote_subscriber_topics.insert( messageName );
+  char nodepath[orb_maxpath];
+  int ret = uORB::Utils::node_mkpath(nodepath, PUBSUB, messageName );
+  if (ret == OK) {
+    // get the node name.
+    uORB::DeviceNode* node = uORB::DeviceMaster::GetDeviceNode( nodepath );
+    if ( node == nullptr) {
+      warnx( "[posix-uORB::Manager::process_add_subscription(%d)]DeviceNode(%s) not created yet",
+             __LINE__, messageName.c_str() );
+    }
+    else{
+      // node is present.
+        node->process_add_subscription(msgRateInHz);
+    }
+  } else {
+    rc = -1;
+  }
+  return rc;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::Manager::process_remove_subscription(
+    const std::string& messageName)
+{
+  warnx("[posix-uORB::Manager::process_remove_subscription(%d)] Enter: name: %s",
+          __LINE__, messageName.c_str() );
+  int16_t rc = -1;
+  _remote_subscriber_topics.erase( messageName );
+  char nodepath[orb_maxpath];
+  int ret = uORB::Utils::node_mkpath(nodepath, PUBSUB, messageName );
+  if (ret == OK) {
+    uORB::DeviceNode* node = uORB::DeviceMaster::GetDeviceNode( nodepath );
+    // get the node name.
+    if ( node == nullptr) {
+      warnx("[posix-uORB::Manager::process_remove_subscription(%d)]Error No existing subscriber found for message: [%s]",
+              __LINE__, messageName.c_str());
+    } else {
+      // node is present.
+      node->process_remove_subscription();
+      rc = 0;
+    }
+  }
+  return rc;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::Manager::process_received_message(const std::string& messageName,
+                                                int32_t length, uint8_t* data)
+{
+  //warnx("[uORB::Manager::process_received_message(%d)] Enter name: %s", __LINE__, messageName.c_str() );
+
+  int16_t rc = -1;
+  char nodepath[orb_maxpath];
+  int ret = uORB::Utils::node_mkpath(nodepath, PUBSUB, messageName );
+  if (ret == OK) {
+    uORB::DeviceNode* node = uORB::DeviceMaster::GetDeviceNode( nodepath );
+    // get the node name.
+    if ( node == nullptr) {
+      warnx("[uORB::Manager::process_received_message(%d)]Error No existing subscriber found for message: [%s] nodepath:[%s]",
+              __LINE__, messageName.c_str(), nodepath );
+
+    } else {
+      // node is present.
+      node->process_received_message( length, data );
+      rc = 0;
+    }
+  }
+  return rc;
+}
+
+bool uORB::Manager::is_remote_subscriber_present( const std::string& messageName )
+{
+  return ( _remote_subscriber_topics.find( messageName ) != _remote_subscriber_topics.end() );
+}

src/modules/uORB/uORBManager_posix.cpp

@@ -39,6 +39,7 @@
 #include <px4_posix.h>
 #include "uORBUtils.hpp"
 #include "uORBManager.hpp"
+#include "px4_config.h"
 #include "uORBDevices.hpp"
 
 
@@ -55,6 +56,7 @@ uORB::Manager* uORB::Manager::get_instance()
 //-----------------------------------------------------------------------------
 //-----------------------------------------------------------------------------
 uORB::Manager::Manager()
+: _comm_channel( nullptr )
 {
 }
 
@@ -292,3 +294,103 @@ int uORB::Manager::node_open
   /* everything has been OK, we can return the handle now */
   return fd;
 }
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+void uORB::Manager::set_uorb_communicator( uORBCommunicator::IChannel* channel)
+{
+  _comm_channel = channel;
+  if (_comm_channel != nullptr) {
+    _comm_channel->register_handler(this);
+  }
+}
+
+uORBCommunicator::IChannel* uORB::Manager::get_uorb_communicator( void )
+{
+  return _comm_channel;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::Manager::process_add_subscription(const std::string& messageName,
+                                                int32_t msgRateInHz)
+{
+  warnx("[posix-uORB::Manager::process_add_subscription(%d)] entering Manager_process_add_subscription: name: %s",
+          __LINE__, messageName.c_str() );
+  int16_t rc = 0;
+  _remote_subscriber_topics.insert( messageName );
+  char nodepath[orb_maxpath];
+  int ret = uORB::Utils::node_mkpath(nodepath, PUBSUB, messageName );
+  if (ret == OK) {
+    // get the node name.
+    uORB::DeviceNode* node = uORB::DeviceMaster::GetDeviceNode( nodepath );
+    if ( node == nullptr) {
+      warnx( "[posix-uORB::Manager::process_add_subscription(%d)]DeviceNode(%s) not created yet",
+             __LINE__, messageName.c_str() );
+    }
+    else{
+      // node is present.
+        node->process_add_subscription(msgRateInHz);
+    }
+  } else {
+    rc = -1;
+  }
+  return rc;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::Manager::process_remove_subscription(
+    const std::string& messageName)
+{
+  warnx("[posix-uORB::Manager::process_remove_subscription(%d)] Enter: name: %s",
+          __LINE__, messageName.c_str() );
+  int16_t rc = -1;
+  _remote_subscriber_topics.erase( messageName );
+  char nodepath[orb_maxpath];
+  int ret = uORB::Utils::node_mkpath(nodepath, PUBSUB, messageName );
+  if (ret == OK) {
+    uORB::DeviceNode* node = uORB::DeviceMaster::GetDeviceNode( nodepath );
+    // get the node name.
+    if ( node == nullptr) {
+      warnx("[posix-uORB::Manager::process_remove_subscription(%d)]Error No existing subscriber found for message: [%s]",
+              __LINE__, messageName.c_str());
+    } else {
+      // node is present.
+      node->process_remove_subscription();
+      rc = 0;
+    }
+  }
+  return rc;
+}
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int16_t uORB::Manager::process_received_message(const std::string& messageName,
+                                                int32_t length, uint8_t* data)
+{
+  //warnx("[uORB::Manager::process_received_message(%d)] Enter name: %s", __LINE__, messageName.c_str() );
+
+  int16_t rc = -1;
+  char nodepath[orb_maxpath];
+  int ret = uORB::Utils::node_mkpath(nodepath, PUBSUB, messageName );
+  if (ret == OK) {
+    uORB::DeviceNode* node = uORB::DeviceMaster::GetDeviceNode( nodepath );
+    // get the node name.
+    if ( node == nullptr) {
+      warnx("[uORB::Manager::process_received_message(%d)]Error No existing subscriber found for message: [%s] nodepath:[%s]",
+              __LINE__, messageName.c_str(), nodepath );
+
+    } else {
+      // node is present.
+      node->process_received_message( length, data );
+      rc = 0;
+    }
+  }
+  return rc;
+}
+
+bool uORB::Manager::is_remote_subscriber_present( const std::string& messageName )
+{
+  return ( _remote_subscriber_topics.find( messageName ) != _remote_subscriber_topics.end() );
+}

src/modules/uORB/uORBUtils.cpp

@@ -61,3 +61,21 @@ int uORB::Utils::node_mkpath
 
   return OK;
 }
+
+//-----------------------------------------------------------------------------
+//-----------------------------------------------------------------------------
+int uORB::Utils::node_mkpath(char *buf, Flavor f,
+                               const std::string& orbMsgName )
+{
+  unsigned len;
+
+  unsigned index = 0;
+
+  len = snprintf(buf, orb_maxpath, "/%s/%s%d", (f == PUBSUB) ? "obj" : "param",
+                 orbMsgName.c_str(), index );
+
+  if (len >= orb_maxpath)
+    return -ENAMETOOLONG;
+
+  return OK;
+}

src/modules/uORB/uORBUtils.hpp

@@ -34,6 +34,7 @@
 #define _uORBUtils_hpp_
 
 #include "uORBCommon.hpp"
+#include <string>
 
 namespace uORB
 {
@@ -50,6 +51,12 @@ public:
       const struct orb_metadata *meta,
       int *instance = nullptr
   );
+
+  /**
+   * same as above except this generators the path based on the string.
+   */
+  static int node_mkpath(char *buf, Flavor f, const std::string& orbMsgName);
+
 };
 
 #endif // _uORBUtils_hpp_