Adds immediate migration results

2026-08-18 22:12:24 +02:00
1 parent ba4121728b
commit 6a2a0667ab
@@ -0,0 +1,524 @@
+/*
+ * (c) Copyright, Real-Time Innovations, 2017.
+ * All rights reserved.
+ *
+ * No duplications, whole or partial, manual or electronic, may be made
+ * without express written permission.  Any such copies, or
+ * revisions thereof, must display this notice unaltered.
+ * This code contains trade secrets of Real-Time Innovations, Inc.
+ */
+
+#ifndef RTI_ROUTING_ADAPTER_DETAIL_STREAM_READER_FORWARDER_HPP_
+#define RTI_ROUTING_ADAPTER_DETAIL_STREAM_READER_FORWARDER_HPP_
+
+#include <assert.h>
+#include <list>
+#include <rti/core/Exception.hpp>
+
+#include <rti/apputils/ServiceException.hpp>
+
+#include <rti/routing/adapter/StreamReader.hpp>
+#include <rti/routing/StreamInfo.hpp>
+#include <rti/routing/detail/UpdatableEntityForwarder.hpp>
+#include <rti/routing/detail/ForwarderUtils.hpp>
+#include <rti/routing/adapter/detail/DiscoveryStreamReaderForwarder.hpp>
+
+
+namespace rti { namespace routing { namespace adapter { namespace detail {
+
+template <bool ReadOrTake, bool HasSelector> struct read_or_take;
+
+
+template<>
+struct read_or_take<false, false> {
+
+    static void read(
+            rti::routing::adapter::StreamReader& stream_reader,
+            std::vector<StreamReader::SamplePtr>& sample_seq,
+            std::vector<StreamReader::InfoPtr>& info_seq,
+            const struct RTI_RoutingServiceSelectorState *)
+    {
+        stream_reader.take(sample_seq, info_seq);
+    }
+};
+
+template<>
+struct read_or_take<true, false> {
+
+    static void read(
+            rti::routing::adapter::StreamReader& stream_reader,
+            std::vector<StreamReader::SamplePtr>& sample_seq,
+            std::vector<StreamReader::InfoPtr>& info_seq,
+            const struct RTI_RoutingServiceSelectorState *)
+    {
+        stream_reader.read(sample_seq, info_seq);
+    }
+};
+
+template<>
+struct read_or_take<false, true> {
+
+    static void read(
+            rti::routing::adapter::StreamReader& stream_reader,
+            std::vector<StreamReader::SamplePtr>& sample_seq,
+            std::vector<StreamReader::InfoPtr>& info_seq,
+            const struct RTI_RoutingServiceSelectorState *native_selector)
+    {
+        stream_reader.take(
+                sample_seq,
+                info_seq,
+                SelectorState(*native_selector));
+    }
+};
+
+
+template<>
+struct read_or_take<true, true> {
+
+    static void read(
+            rti::routing::adapter::StreamReader& stream_reader,
+            std::vector<StreamReader::SamplePtr>& sample_seq,
+            std::vector<StreamReader::InfoPtr>& info_seq,
+            const struct RTI_RoutingServiceSelectorState *native_selector)
+    {
+        stream_reader.read(
+                sample_seq,
+                info_seq,
+                SelectorState(*native_selector));
+    }
+};
+
+
+struct SamplesHolder {
+public:
+
+    SamplesHolder()
+    {
+    }
+
+    std::vector<StreamReader::SamplePtr> sample_seq_;
+    std::vector<StreamReader::InfoPtr> info_seq_;
+};
+
+class StreamReaderForwarder {
+public:
+    static RTI_RoutingServiceStreamReaderExt* create_native(
+            Connection *connection,
+            Session *session,
+            const struct RTI_RoutingServiceStreamInfo *native_stream_info,
+            const struct RTI_RoutingServiceProperties *native_properties,
+            const struct RTI_RoutingServiceStreamReaderListenerExt *native_listener,
+            RTI_RoutingServiceEnvironment *environment)
+    {
+
+        try {
+            using rti::routing::detail::ScopedForwarder;
+
+            StreamInfo stream_info(*native_stream_info);
+
+            std::map<std::string, std::string> properties;
+            rti::routing::PropertyAdapter::add_properties_from_native(
+                    properties,
+                    native_properties);
+
+            StreamReaderForwarder *forwarder = new StreamReaderForwarder(
+                    native_listener);
+            ScopedForwarder<Connection, StreamReaderForwarder> scoped(
+                    connection,
+                    forwarder,
+                    environment);
+            forwarder->stream_reader_ = connection->create_stream_reader(
+                    session,
+                    stream_info,
+                    properties,
+                    &forwarder->listener_);
+            RTI_ROUTING_THROW_ON_NULL(forwarder->stream_reader_);
+
+            scoped.release();
+
+            // TRUST-134: the forwarder is not leaked here. It might be leaked
+            // if delete_native() were not to be called, but Routing Service
+            // will make sure that is not the case
+
+            /* coverity[leaked_storage : FALSE] */
+            return forwarder->native();
+        } catch(const rti::apputils::FatalServiceException& ex) {
+            RTI_RoutingServiceEnvironment_fatal_error(
+                    environment,
+                    ex.what());
+            return NULL;
+        } catch(const std::exception& ex) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    ex.what());
+            return NULL;
+        } catch (...) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    "unexpected exception");
+            return NULL;
+        }
+
+
+    }
+
+    static void delete_native(
+            Connection *connection,
+            RTI_RoutingServiceStreamReaderExt *native_stream_reader,
+            RTI_RoutingServiceEnvironment *environment)
+    {
+        StreamReaderForwarder *stream_reader_forwarder =
+                from_native(native_stream_reader);
+        try {
+            if (stream_reader_forwarder->stream_reader_ != NULL) {
+                connection->delete_stream_reader(
+                        stream_reader_forwarder->stream_reader_);
+                stream_reader_forwarder->stream_reader_ = NULL;
+            }
+        } catch(const rti::apputils::FatalServiceException& ex) {
+            RTI_RoutingServiceEnvironment_fatal_error(
+                    environment,
+                    ex.what());
+        } catch(const std::exception& ex) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    ex.what());
+        } catch (...) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    "unexpected exception");
+        }
+
+        delete stream_reader_forwarder;
+    }
+
+
+    RTI_RoutingServiceStreamReaderExt* native()
+    {
+        return &this->native_;
+    }
+
+
+private:
+
+    StreamReaderForwarder(
+            const RTI_RoutingServiceStreamReaderListenerExt *native_listener) :
+            stream_reader_(NULL),
+            listener_(native_listener)
+    {
+        RTIOsapiMemory_zero(&native_, sizeof(native_));
+        native_.stream_reader_data =
+                static_cast<void *>(this);
+        native_.take =
+                StreamReaderForwarder::take;
+        native_.read =
+                StreamReaderForwarder::read;
+        native_.take_w_selector =
+                StreamReaderForwarder::take_with_selector;
+        native_.read_w_selector =
+                StreamReaderForwarder::read_with_selector;
+        native_.return_loan =
+                StreamReaderForwarder::return_loan;
+        native_.create_content_query =
+                StreamReaderForwarder::create_content_query;
+        native_.delete_content_query =
+                StreamReaderForwarder::delete_content_query;
+        native_.update =
+                StreamReaderForwarder::update;
+    }
+
+    ~StreamReaderForwarder()
+    {
+        // delete holders
+        for (std::list<SamplesHolder*>::iterator it = holder_pool_.begin();
+             it != holder_pool_.end();
+             ++it) {
+            delete (*it);
+        }
+    }
+
+
+    static void take(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSample **sample_array,
+        RTI_RoutingServiceSampleInfo **sample_info_array,
+        int *array_length,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        proxy_read<false, false>(
+                native_stream_reader_data,
+                sample_array,
+                sample_info_array,
+                array_length,
+                NULL,
+                environment);
+    }
+
+    static void read(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSample **sample_array,
+        RTI_RoutingServiceSampleInfo **sample_info_array,
+        int *array_length,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        proxy_read<true, false>(
+                native_stream_reader_data,
+                sample_array,
+                sample_info_array,
+                array_length,
+                NULL,
+                environment);
+    }
+
+    static void take_with_selector(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSample **sample_array,
+        RTI_RoutingServiceSampleInfo **sample_info_array,
+        int *array_length,
+        const struct RTI_RoutingServiceSelectorState *native_selector,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        proxy_read<false, true>(
+                native_stream_reader_data,
+                sample_array,
+                sample_info_array,
+                array_length,
+                native_selector,
+                environment);
+    }
+
+    static void read_with_selector(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSample **sample_array,
+        RTI_RoutingServiceSampleInfo **sample_info_array,
+        int *array_length,
+        const struct RTI_RoutingServiceSelectorState *native_selector,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        proxy_read<true, true>(
+                native_stream_reader_data,
+                sample_array,
+                sample_info_array,
+                array_length,
+                native_selector,
+                environment);
+    }
+
+    static void return_loan(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSample *native_samples,
+        RTI_RoutingServiceSampleInfo *,
+        int count,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        StreamReaderForwarder *forwarder =
+                static_cast<StreamReaderForwarder*>(native_stream_reader_data);
+        assert(forwarder != NULL);
+
+        SamplesHolder *holder = static_cast<SamplesHolder*>(
+                native_samples[count]);
+        assert(holder != NULL);
+
+
+        try {
+            forwarder->stream_reader_->return_loan(
+                    holder->sample_seq_,
+                    holder->info_seq_);
+        } catch (const rti::apputils::FatalServiceException& ex) {
+            RTI_RoutingServiceEnvironment_fatal_error(
+                    environment,
+                    ex.what());
+        } catch (const std::exception& ex) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    ex.what());
+        } catch (...) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    "unexpected exception");
+        }
+
+        forwarder->return_holder(holder);
+    }
+
+    static void update(
+            void *native_stream_reader_data,
+            const struct RTI_RoutingServiceProperties * native_properties,
+            RTI_RoutingServiceEnvironment * environment)
+    {
+
+        StreamReaderForwarder *stream_reader_forwarder =
+                static_cast<StreamReaderForwarder*>(native_stream_reader_data);
+
+        rti::routing::detail::UpdatableEntityForwarder::update(
+                stream_reader_forwarder->stream_reader_,
+                native_properties,
+                environment);
+    }
+
+    static void* create_content_query(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSelectorStateQueryData old_query_data,
+        const struct RTI_RoutingServiceSelectorContent *content,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        StreamReaderForwarder *forwarder =
+                static_cast<StreamReaderForwarder*>(native_stream_reader_data);
+
+        void *query_data = NULL;
+        try {
+            dds::topic::Filter filter(
+                    content->expression == NULL ? "" : content->expression,
+                    rti::core::native_conversions::from_native<std::string>(
+                            content->expression_parameters));
+            query_data = forwarder->stream_reader_->create_content_query(
+                    old_query_data,
+                    filter);
+        } catch (const rti::apputils::FatalServiceException& ex) {
+            RTI_RoutingServiceEnvironment_fatal_error(
+                    environment,
+                    ex.what());
+        } catch (const std::exception& ex) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    ex.what());
+        } catch (...) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    "unexpected exception");
+        }
+
+        return query_data;
+    }
+
+    static void delete_content_query(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSelectorStateQueryData query_data,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        StreamReaderForwarder *forwarder =
+                static_cast<StreamReaderForwarder*>(native_stream_reader_data);
+        try {
+            forwarder->stream_reader_->delete_content_query(query_data);
+        } catch (const std::exception& ex) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    "%s",
+                    ex.what());
+        } catch (...) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    "unexpected exception");
+        }
+    }
+
+    static StreamReaderForwarder* from_native(
+            RTI_RoutingServiceStreamReaderExt *native)
+    {
+        return static_cast<StreamReaderForwarder*>(native->stream_reader_data);
+    }
+
+    template <bool ReadOrTake, bool HasSelector>
+    static void proxy_read(
+        void *native_stream_reader_data,
+        RTI_RoutingServiceSample **sample_array,
+        RTI_RoutingServiceSampleInfo **sample_info_array,
+        int *array_length,
+        const struct RTI_RoutingServiceSelectorState *native_selector,
+        RTI_RoutingServiceEnvironment *environment)
+    {
+        StreamReaderForwarder *forwarder =
+                static_cast<StreamReaderForwarder*>(native_stream_reader_data);
+        assert(forwarder != NULL);
+
+        SamplesHolder *holder = forwarder->get_holder();
+
+        *array_length = 0;
+        try {
+            read_or_take<ReadOrTake, HasSelector>::read(
+                    *forwarder->stream_reader_,
+                    holder->sample_seq_,
+                    holder->info_seq_,
+                    native_selector);
+            if (!holder->info_seq_.empty()
+                    && (holder->info_seq_.size() != holder->sample_seq_.size())) {
+                throw dds::core::PreconditionNotMetError(
+                        "sample and info sequences length mismatch");
+            }
+            /*
+             * We reserve a hidden slot in the sample_seq to keep the original
+             * loaned returned from the native StreamReader. This allows us
+             * to quickly access it on return_loan(), instead of having to search
+             * in a list.
+             *
+             * TRUST-1466: reserving just 1 more memory spot can be inefficient,
+             * and we're doing this for every read/take operation called in C++.
+             */
+            if (holder->sample_seq_.size() + 1
+                    >= holder->sample_seq_.capacity()) {
+                holder->sample_seq_.reserve(
+                        holder->sample_seq_.capacity() == 0
+                                ? 1
+                                : 2 * holder->sample_seq_.capacity());
+            }
+            *(holder->sample_seq_.data() + holder->sample_seq_.size()) = holder;
+        } catch (const rti::apputils::FatalServiceException& ex) {
+            RTI_RoutingServiceEnvironment_fatal_error(
+                    environment,
+                    ex.what());
+            forwarder->return_holder(holder);
+            return;
+        } catch (const std::exception& ex) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    ex.what());
+            forwarder->return_holder(holder);
+            return;
+        } catch (...) {
+            RTI_RoutingServiceEnvironment_set_error(
+                    environment,
+                    "unexpected exception");
+            forwarder->return_holder(holder);
+            return;
+        }
+
+        // direct mapping
+        *array_length = static_cast<int>(holder->sample_seq_.size());
+        *sample_array = holder->sample_seq_.data();
+        if (holder->info_seq_.empty()) {
+            *sample_info_array = NULL;
+        } else {
+            *sample_info_array = &holder->info_seq_[0];
+        }
+    }
+
+    SamplesHolder* get_holder()
+    {
+        if (holder_pool_.size() == 0) {
+            return new SamplesHolder();
+        }
+
+        SamplesHolder *holder = holder_pool_.front();
+        holder_pool_.pop_front();
+
+        return holder;
+    }
+
+    void return_holder(SamplesHolder *holder)
+    {
+        holder->sample_seq_.clear();
+        holder->info_seq_.clear();
+        holder_pool_.push_front(holder);
+    }
+
+private:
+
+    RTI_RoutingServiceStreamReaderExt native_;
+    StreamReader *stream_reader_;
+    StreamReaderListener listener_;
+    std::list<SamplesHolder*> holder_pool_;
+};
+
+}}}}
+
+#endif // RTI_ROUTING_ADAPTER_DETAIL_STREAM_READER_FORWARDER_HPP_