Open streams working for all stream types + some bug fixes

src/context.cpp

@@ -504,7 +504,7 @@ namespace librealsense
         rs2_video_stream to_rs2_video_stream( std::shared_ptr< realdds::dds_video_stream_profile > const & profile )
         {
             rs2_video_stream prof;
-            prof.type = RS2_STREAM_ANY;
+            prof.type = static_cast< rs2_stream >(profile->type());
             prof.index = profile->uid().index;
             prof.uid = profile->uid().sid;
             prof.width = profile->width();
@@ -519,7 +519,7 @@ namespace librealsense
         rs2_motion_stream to_rs2_motion_stream( std::shared_ptr< realdds::dds_motion_stream_profile > const & profile )
         {
             rs2_motion_stream prof;
-            prof.type = RS2_STREAM_ANY;
+            prof.type = static_cast< rs2_stream >( profile->type() );
             prof.index = profile->uid().index;
             prof.uid = profile->uid().sid;
             prof.fps = profile->frequency();

third-party/realdds/include/realdds/dds-control-server.h

@@ -0,0 +1,47 @@
+// License: Apache 2.0. See LICENSE file in root directory.
+// Copyright(c) 2022 Intel Corporation. All Rights Reserved.
+
+#pragma once
+
+
+#include <realdds/topics/control/controlPubSubTypes.h>  // raw::device::control
+#include <librealsense2/utilities/concurrency/concurrency.h>
+
+#include <memory>
+#include <string>
+
+
+namespace realdds {
+
+
+namespace topics {
+namespace raw {
+namespace device {
+class control;
+}
+}  // namespace raw
+}  // namespace topics
+
+
+class dds_subscriber;
+class dds_topic_reader;
+
+
+// A control is a command sent from a client (camera user) and the server (this class) receives and interprets.
+//
+class dds_control_server
+{
+public:
+    dds_control_server( std::shared_ptr< dds_subscriber > const & subscriber, const std::string & topic_name );
+    ~dds_control_server();
+
+    typedef std::function< void() > on_control_message_received_callback;
+    void on_control_message_received( on_control_message_received_callback callback );
+
+private:
+    std::shared_ptr< dds_subscriber > _subscriber;
+    std::shared_ptr< dds_topic_reader > _reader;
+};
+
+
+}

third-party/realdds/include/realdds/dds-device-server.h

@@ -18,6 +18,7 @@ namespace raw {
 class device_info;
 namespace device {
 class notification;
+class control;
 }  // namespace device
 }  // namespace raw
 class device_info;
@@ -26,8 +27,10 @@ class device_info;
 
 class dds_participant;
 class dds_publisher;
+class dds_subscriber;
 class dds_stream_server;
 class dds_notification_server;
+class dds_control_server;
 struct image_header;
 
 
@@ -65,10 +68,14 @@ class dds_device_server
     void publish_notification( topics::raw::device::notification&& notification );
 
 private:
+    void on_control_message_received();
+
     std::shared_ptr< dds_publisher > _publisher;
+    std::shared_ptr< dds_subscriber > _subscriber;
     std::string _topic_root;
     std::unordered_map<std::string, std::shared_ptr<dds_stream_server>> _stream_name_to_server;
     std::shared_ptr< dds_notification_server > _notification_server;
+    std::shared_ptr< dds_control_server > _control_server;
 };  // class dds_device_server
 
 

third-party/realdds/include/realdds/dds-stream-profile.h

@@ -83,18 +83,20 @@ class dds_stream_profile
 {
     dds_stream_uid _uid;
     dds_stream_format _format;
-    int16_t _frequency;  // "Frames" per second
+    int16_t _frequency; // "Frames" per second
+    int8_t _inner_type; // User defined type
 
     std::weak_ptr< dds_stream_base > _stream;
 
 public:
     virtual ~dds_stream_profile() {}
 
 protected:
-    dds_stream_profile( dds_stream_uid uid, dds_stream_format format, int16_t frequency )
+    dds_stream_profile( dds_stream_uid uid, dds_stream_format format, int16_t frequency, int8_t inner_type = 0 )
         : _uid( uid )
         , _format( format )
         , _frequency( frequency )
+        , _inner_type( inner_type )
     {
     }
     dds_stream_profile( dds_stream_profile && ) = default;
@@ -107,6 +109,7 @@ class dds_stream_profile
     dds_stream_uid uid() const { return _uid; }
     dds_stream_format format() const { return _format; }
     int16_t frequency() const { return _frequency; }
+    int8_t type() const { return _inner_type; }
 
     // These are for debugging - not functional
     virtual std::string to_string() const;
@@ -125,8 +128,9 @@ class dds_video_stream_profile : public dds_stream_profile
     uint8_t _bytes_per_pixel;
 
 public:
-    dds_video_stream_profile( dds_stream_uid uid, dds_stream_format format, int16_t frequency, uint16_t width, uint16_t height, uint8_t bytes_per_pixel )
-        : dds_stream_profile( uid, format, frequency )
+    dds_video_stream_profile( dds_stream_uid uid, dds_stream_format format, int16_t frequency, uint16_t width, uint16_t height,
+                              uint8_t bytes_per_pixel, int8_t inner_type = 0 )
+        : dds_stream_profile( uid, format, frequency, inner_type )
         , _width( width )
         , _height( height )
         , _bytes_per_pixel( bytes_per_pixel )
@@ -146,8 +150,8 @@ class dds_video_stream_profile : public dds_stream_profile
 class dds_motion_stream_profile : public dds_stream_profile
 {
 public:
-    dds_motion_stream_profile( dds_stream_uid uid, dds_stream_format format, int16_t frequency )
-        : dds_stream_profile( uid, format, frequency )
+    dds_motion_stream_profile( dds_stream_uid uid, dds_stream_format format, int16_t frequency, int8_t inner_type = 0 )
+        : dds_stream_profile( uid, format, frequency, inner_type )
     {
     }
     dds_motion_stream_profile( dds_motion_stream_profile && ) = default;

third-party/realdds/include/realdds/dds-subscriber.h

@@ -0,0 +1,43 @@
+// License: Apache 2.0. See LICENSE file in root directory.
+// Copyright(c) 2022 Intel Corporation. All Rights Reserved.
+
+#pragma once
+
+#include <memory>
+
+
+namespace eprosima {
+namespace fastdds {
+namespace dds {
+class Subscriber;
+}  // namespace dds
+}  // namespace fastdds
+}  // namespace eprosima
+
+
+namespace realdds {
+
+
+class dds_participant;
+
+
+// The Subscriber manages the activities of several dds_topic_reader (DataReader) entities
+//
+class dds_subscriber
+{
+    std::shared_ptr< dds_participant > _participant;
+
+    eprosima::fastdds::dds::Subscriber * _subscriber;
+
+public:
+    dds_subscriber( std::shared_ptr< dds_participant > const & participant );
+    ~dds_subscriber();
+
+    eprosima::fastdds::dds::Subscriber * get() const { return _subscriber; }
+    eprosima::fastdds::dds::Subscriber * operator->() const { return get(); }
+
+    std::shared_ptr< dds_participant > const & get_participant() const { return _participant; }
+};
+
+
+}  // namespace realdds

third-party/realdds/include/realdds/dds-topic-reader.h

@@ -23,20 +23,24 @@ namespace realdds {
 
 
 class dds_topic;
-
+class dds_subscriber;
 
 // The 'reader' is the entity used to subscribe to updated values of data in a topic. It is bound at creation to this
 // topic.
+// 
+// You may choose to create one via a 'subscriber' that manages the activities of several readers.
+// on_data_available callback will be called when a sample is received.
 //
 class dds_topic_reader : public eprosima::fastdds::dds::DataReaderListener
 {
     std::shared_ptr< dds_topic > const _topic;
+    std::shared_ptr < dds_subscriber > const _subscriber;
 
-    eprosima::fastdds::dds::Subscriber * _subscriber = nullptr;
     eprosima::fastdds::dds::DataReader * _reader = nullptr;
 
 public:
     dds_topic_reader( std::shared_ptr< dds_topic > const & topic );
+    dds_topic_reader( std::shared_ptr< dds_topic > const & topic, std::shared_ptr< dds_subscriber > const & subscriber );
     ~dds_topic_reader();
 
     eprosima::fastdds::dds::DataReader * get() const { return _reader; }

third-party/realdds/src/dds-control-server.cpp

@@ -0,0 +1,56 @@
+// License: Apache 2.0. See LICENSE file in root directory.
+// Copyright(c) 2022 Intel Corporation. All Rights Reserved.
+
+#include <realdds/dds-control-server.h>
+
+#include <realdds/dds-participant.h>
+#include <realdds/dds-subscriber.h>
+#include <realdds/dds-stream-server.h>
+#include <realdds/dds-utilities.h>
+#include <realdds/topics/dds-topics.h>
+#include <realdds/dds-topic.h>
+#include <realdds/dds-topic-reader.h>
+
+#include <fastdds/dds/topic/Topic.hpp>
+
+#include <fastdds/dds/domain/DomainParticipant.hpp>
+#include <fastdds/dds/subscriber/Subscriber.hpp>
+#include <fastdds/dds/subscriber/DataReader.hpp>
+#include <fastdds/dds/subscriber/DataReaderListener.hpp>
+#include <fastdds/dds/subscriber/qos/SubscriberQos.hpp>
+#include <librealsense2/utilities/concurrency/concurrency.h>
+
+
+using namespace eprosima::fastdds::dds;
+
+namespace realdds {
+
+
+dds_control_server::dds_control_server( std::shared_ptr< dds_subscriber > const & subscriber, const std::string & topic_name )
+    : _subscriber( subscriber )
+{
+    auto topic = topics::device::control::create_topic( subscriber->get_participant(), topic_name.c_str() );
+    _reader = std::make_shared< dds_topic_reader >( topic, subscriber );
+
+    dds_topic_reader::qos rqos( RELIABLE_RELIABILITY_QOS );
+    rqos.history().depth = 10;                                                                      // default is 1
+    rqos.endpoint().history_memory_policy = eprosima::fastrtps::rtps::DYNAMIC_RESERVE_MEMORY_MODE;
+    _reader->run( rqos );
+}
+
+
+dds_control_server::~dds_control_server()
+{
+}
+
+
+void dds_control_server::on_control_message_received( on_control_message_received_callback callback )
+{
+    if ( !_reader )
+        DDS_THROW( runtime_error, "setting callback when reader is not created" );
+
+    _reader->on_data_available( callback );
+}
+
+
+}  // namespace realdds

third-party/realdds/src/dds-device-impl.cpp

@@ -60,7 +60,8 @@ to_realdds_profile( const topics::device::notification::video_stream_profile & p
                                                               profile.framerate,
                                                               profile.width,
                                                               profile.height,
-                                                              0 );  // TODO - bpp
+                                                              0, // TODO - bpp
+                                                              profile.type);
     // TODO - add intrinsics
 
     return prof;
@@ -71,7 +72,8 @@ to_realdds_profile( const topics::device::notification::motion_stream_profile &
 {
     auto prof = std::make_shared< dds_motion_stream_profile >( dds_stream_uid( profile.uid, profile.stream_index ),
                                                                dds_stream_format::from_rs2( profile.format ),
-                                                               profile.framerate );
+                                                               profile.framerate,
+                                                               profile.type );
     // TODO - add intrinsics
 
     return prof;