Review changes. Stream writer and reader improvements

Author: iutrilla-rti

examples/routing_service/udp_socket_adapter_dynamic/RsSocketAdapter.xml

@@ -14,12 +14,12 @@
     <routing_service name="SocketAdapterToDDS">
         <domain_route name="SocketBridge">
             <connection name="SocketConnection" plugin_name="AdapterLib::SocketAdapter">
-                <register_typename="PingType" type_name="PingType" />
+                <register_type name="PingType" type_ref="PingType" />
             </connection>
             <participant name="DDSConnection">
                 <domain_id>1</domain_id>
                 <domain_participant_qos base_name="BuiltinQosLib::Generic.Common" />
-                <register_typename="PingType" type_name="PingType" />
+                <register_type name="PingType" type_ref="PingType" />
             </participant>
             <session name="session">
                 <route>
@@ -57,12 +57,12 @@
     <routing_service name="DDSToSocketAdapter">
         <domain_route name="SocketBridge">
             <connection name="SocketConnection" plugin_name="AdapterLib::SocketAdapter">
-                <register_typename="PingType" type_name="PingType" />
+                <register_type name="PingType" type_ref="PingType" />
             </connection>
             <participant name="DDSConnection">
                 <domain_id>0</domain_id>
                 <domain_participant_qos base_name="BuiltinQosLib::Generic.Common" />
-                <register_typename="PingType" type_name="PingType" />
+                <register_type name="PingType" type_ref="PingType" />
             </participant>
             <session>
                 <route>

examples/routing_service/udp_socket_adapter_dynamic/src/SocketStreamReader.cxx

@@ -15,6 +15,7 @@
 #include <chrono>
 #include <sstream>
 #include <thread>
+#include <queue>
 
 #include <cstring>
 #include <iostream>
@@ -55,7 +56,10 @@ void SocketStreamReader::socket_reading_thread()
             continue;
         }
 
-        received_bytes_ = received_bytes;
+        {
+            std::lock_guard<std::mutex> lock(buffer_mutex_);
+            received_buffers_.emplace(received_buffer_, received_buffer_ + received_bytes);
+        }
 
         reader_listener_->on_data_available(this);
     }
@@ -95,10 +99,21 @@ void SocketStreamReader::take(
         std::vector<dds::core::xtypes::DynamicData *> &samples,
         std::vector<dds::sub::SampleInfo *> &infos)
 {
+    std::vector<char> buffer;
+    {
+        std::unique_lock<std::mutex> lock(buffer_mutex_);
+        if (received_buffers_.empty()) {
+            // No data available
+            samples.clear();
+            infos.clear();
+            return;
+        }
+        buffer = std::move(received_buffers_.front());
+        received_buffers_.pop();
+    }
+
     dds::core::xtypes::DynamicData deserialized_sample(*adapter_type_);
-    std::vector<char> received_buffer = std::vector<char>(
-        received_buffer_, received_buffer_ + received_bytes_);
-    rti::core::xtypes::from_cdr_buffer(deserialized_sample, received_buffer);
+    rti::core::xtypes::from_cdr_buffer(deserialized_sample, buffer);
 
     samples.resize(1);
     infos.resize(1);

examples/routing_service/udp_socket_adapter_dynamic/src/SocketStreamReader.hpp

@@ -16,6 +16,7 @@
 #include <fstream>
 #include <iostream>
 #include <thread>
+#include <queue>
 
 #include "SocketConnection.hpp"
 #include "UdpSocket.hpp"
@@ -65,8 +66,8 @@ class SocketStreamReader : public rti::routing::adapter::DynamicDataStreamReader
     std::ifstream input_socket_stream_;
     std::string receive_address_;
     int receive_port_;
-    char received_buffer_[BUFFER_MAX_SIZE]; // Value that's high enough
-    int received_bytes_;
+    char received_buffer_[BUFFER_MAX_SIZE];
+    std::queue<std::vector<char>> received_buffers_;
     std::mutex buffer_mutex_;
 
     rti::routing::StreamInfo stream_info_;

examples/routing_service/udp_socket_adapter_dynamic/src/SocketStreamWriter.cxx

@@ -83,41 +83,19 @@ int SocketStreamWriter::write(
 {
     size_t len = 0;
     for (const auto sample : samples) {
-        std::vector<char> buffer;
-        rti::core::xtypes::to_cdr_buffer(buffer, *sample);
+        serialization_buffer_.clear();
+        rti::core::xtypes::to_cdr_buffer(serialization_buffer_, *sample);
 	 // Send the serialized data
         len = socket->send_data(
-            buffer.data(),
-            buffer.size(),
+            serialization_buffer_.data(),
+            serialization_buffer_.size(),
             dest_address_.c_str(),
             dest_port_);
     }
 
     return len;
 }
 
-int SocketStreamWriter::write(
-        const std::vector<dds::core::xtypes::DynamicData *> &samples,
-        const std::vector<dds::sub::SampleInfo *> &infos,
-        const SelectorState &selector_state)
-{
-    int len;
-    len = write(samples, infos);
-    return len;
-}
-
-void SocketStreamWriter::return_loan(
-        std::vector<dds::core::xtypes::DynamicData *> &samples,
-        std::vector<dds::sub::SampleInfo *> &infos)
-{
-    for (int i = 0; i < samples.size(); ++i) {
-        delete samples[i];
-        delete infos[i];
-    }
-    samples.clear();
-    infos.clear();
-}
-
 SocketStreamWriter::~SocketStreamWriter()
 {
 }

examples/routing_service/udp_socket_adapter_dynamic/src/SocketStreamWriter.hpp

@@ -38,18 +38,9 @@ class SocketStreamWriter : public rti::routing::adapter::DynamicDataStreamWriter
             const rti::routing::PropertySet &
             );
 
-    virtual int
-            write(const std::vector<dds::core::xtypes::DynamicData *> &,
-                  const std::vector<dds::sub::SampleInfo *> &) final;
-
     virtual int write(
             const std::vector<dds::core::xtypes::DynamicData *> &,
-            const std::vector<dds::sub::SampleInfo *> &,
-            const rti::routing::adapter::SelectorState &selector_state) final;
-
-    virtual void return_loan(
-            std::vector<dds::core::xtypes::DynamicData *> &,
-            std::vector<dds::sub::SampleInfo *> &) final;
+            const std::vector<dds::sub::SampleInfo *> &) final;
 
     ~SocketStreamWriter();
 
@@ -61,7 +52,7 @@ class SocketStreamWriter : public rti::routing::adapter::DynamicDataStreamWriter
      */
 
     SocketConnection *socket_connection_;
-
+    std::vector<char> serialization_buffer_;
     std::unique_ptr<UdpSocket> socket;
 
     int send_port_;