replace device-watcher dispatcher with reader-thread

maloel · Mar 16, 2023 · 74d5a35 · 74d5a35
1 parent c2bb6da
commit 74d5a35
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 18 deletions.
diff --git a/third-party/realdds/include/realdds/dds-device-watcher.h b/third-party/realdds/include/realdds/dds-device-watcher.h
@@ -5,10 +5,9 @@
 
 #include "dds-participant.h"
 
-#include <rsutils/concurrency/concurrency.h>
-
 #include <map>
 #include <memory>
+#include <mutex>
 
 
 namespace realdds {
@@ -43,7 +42,7 @@ class dds_device_watcher
 
     void start();
     void stop();
-    bool is_stopped() const { return ! _active_object.is_active(); }
+    bool is_stopped() const;
 
     bool foreach_device( std::function< bool( std::shared_ptr< dds_device > const & ) > ) const;
 
@@ -59,7 +58,6 @@ class dds_device_watcher
     std::shared_ptr< dds_participant::listener > _listener;
     std::shared_ptr< dds_topic_reader > _device_info_topic;
 
-    active_object<> _active_object;
     on_device_change_callback _on_device_added;
     on_device_change_callback _on_device_removed;
     std::map< dds_guid, std::shared_ptr< dds_device > > _dds_devices;

diff --git a/third-party/realdds/src/dds-device-watcher.cpp b/third-party/realdds/src/dds-device-watcher.cpp
@@ -3,7 +3,7 @@
 
 #include <realdds/dds-device-watcher.h>
 #include <realdds/dds-topic.h>
-#include <realdds/dds-topic-reader.h>
+#include <realdds/dds-topic-reader-thread.h>
 #include <realdds/dds-device.h>
 #include <realdds/dds-utilities.h>
 #include <realdds/dds-guid.h>
@@ -23,26 +23,25 @@ using namespace realdds;
 
 dds_device_watcher::dds_device_watcher( std::shared_ptr< dds_participant > const & participant )
     : _device_info_topic(
-        new dds_topic_reader( topics::flexible_msg::create_topic( participant, topics::DEVICE_INFO_TOPIC_NAME ) ) )
+        new dds_topic_reader_thread( topics::flexible_msg::create_topic( participant, topics::DEVICE_INFO_TOPIC_NAME ) ) )
     , _participant( participant )
-    , _active_object( [this]( dispatcher::cancellable_timer timer ) {
-
-        eprosima::fastrtps::Duration_t const one_second = { 1, 0 };
-        if( _device_info_topic->get()->wait_for_unread_message( one_second ) )
+{
+    _device_info_topic->on_data_available(
+        [this]()
         {
             topics::flexible_msg msg;
             eprosima::fastdds::dds::SampleInfo info;
             while( topics::flexible_msg::take_next( *_device_info_topic, &msg, &info ) )
             {
-                if( !msg.is_valid() )
+                if( ! msg.is_valid() )
                     continue;
 
                 topics::device_info device_info = topics::device_info::from_json( msg.json_data() );
 
                 eprosima::fastrtps::rtps::GUID_t guid;
                 eprosima::fastrtps::rtps::iHandle2GUID( guid, info.publication_handle );
 
-                LOG_DEBUG( "DDS device (" << _participant->print(guid) << ") detected:"
+                LOG_DEBUG( "DDS device (" << _participant->print( guid ) << ") detected:"
                                           << "\n\tName: " << device_info.name
                                           << "\n\tSerial: " << device_info.serial
                                           << "\n\tProduct line: " << device_info.product_line
@@ -62,13 +61,19 @@ dds_device_watcher::dds_device_watcher( std::shared_ptr< dds_participant > const
                 if( _on_device_added )
                     _on_device_added( device );
             }
-        }
-    } )
-{
+        } );
+
     if( ! _participant->is_valid() )
         DDS_THROW( runtime_error, "participant was not initialized" );
 }
 
+
+bool dds_device_watcher::is_stopped() const
+{
+    return ! _device_info_topic->is_running();
+}
+
+
 void dds_device_watcher::start()
 {
     stop();
@@ -79,7 +84,6 @@ void dds_device_watcher::start()
         // it takes time and keeps the 'dds_device_server' busy
         init();
     }
-    _active_object.start();
     LOG_DEBUG( "DDS device watcher started on '" << _participant->get()->get_qos().name() << "' "
                                                  << realdds::print( _participant->guid() ) );
 }
@@ -88,7 +92,7 @@ void dds_device_watcher::stop()
 {
     if( ! is_stopped() )
     {
-        _active_object.stop();
+        _device_info_topic->stop();
         //_callback_inflight.wait_until_empty();
         LOG_DEBUG( "DDS device watcher stopped" );
     }
@@ -127,7 +131,7 @@ void dds_device_watcher::init()
     if( ! _device_info_topic->is_running() )
         _device_info_topic->run( dds_topic_reader::qos() );
 
-    LOG_DEBUG( "DDS device watcher initialized successfully" );
+    LOG_DEBUG( "DDS device watcher is running" );
 }