StreamReader.hpp

/*
 * (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_STREAM_READER_HPP_
#define RTI_ROUTING_ADAPTER_STREAM_READER_HPP_
 
#include <vector>
 
#include <dds/core/SafeEnumeration.hpp>
#include <dds/sub/status/DataState.hpp>
#include <dds/core/InstanceHandle.hpp>
#include <dds/core/types.hpp>
#include <dds/topic/Filter.hpp>
#include <dds/core/xtypes/DynamicData.hpp>
#include <dds/sub/SampleInfo.hpp>
#include <rti/core/NativeValueType.hpp>
#include <rti/routing/UpdatableEntity.hpp>
#include <rti/routing/detail/ForwarderUtils.hpp>
 
namespace rti { namespace routing { namespace adapter {
 
class SelectorState;
 
class StreamReader : public UpdatableEntity{
 
public:
    typedef void* SamplePtr;
    typedef void* InfoPtr;
 
    virtual void take(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) = 0;
 
    virtual void read(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) = 0;
 
    virtual void take(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq,
            const SelectorState& selector_state) = 0;
 
    virtual void read(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq,
            const SelectorState& selector_state) = 0;
 
    virtual void return_loan(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) = 0;
 
    virtual void* create_content_query(
            void *old_query_data,
            const dds::topic::Filter& filter) = 0;
 
    virtual void delete_content_query(void* query_data) = 0;
 
    virtual ~StreamReader()
    {
    }
 
};
 
 
template <typename Data, typename Info>
class TStreamReader : public StreamReader {
 
public:
 
    typedef Data DataRep;
    typedef Info InfoRep;
 
    using StreamReader::take;
    using StreamReader::read;
    using StreamReader::return_loan;
 
    void take(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) RTI_FINAL
    {
        take(sample_seq_, info_seq_);
        convert_samples(sample_seq, info_seq);
    }
 
    void read(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) RTI_FINAL
    {
        read(sample_seq_, info_seq_);
        convert_samples(sample_seq, info_seq);
    }
 
    void take(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq,
            const SelectorState& selector_state) RTI_FINAL
    {
        take(sample_seq_, info_seq_, selector_state);
        convert_samples(sample_seq, info_seq);
    }
 
    void read(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq,
            const SelectorState& selector_state) RTI_FINAL
    {
        read(sample_seq_, info_seq_, selector_state);
        convert_samples(sample_seq, info_seq);
    }
 
    void return_loan(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) RTI_FINAL
    {
        RTI_ROUTING_SAMPLE_VECTOR_COPY_PTRS(sample_seq_, sample_seq);
        RTI_ROUTING_SAMPLE_VECTOR_COPY_PTRS(info_seq_, info_seq);
        return_loan(sample_seq_, info_seq_);
        sample_seq_.clear();
        info_seq_.clear();
    }
 
    virtual void take(
            std::vector<Data*>& sample_seq,
            std::vector<Info*>& info_seq) = 0;
 
    virtual void read(
            std::vector<Data*>& sample_seq,
            std::vector<Info*>& info_seq) = 0;
 
    virtual void take(
            std::vector<Data*>& sample_seq,
            std::vector<Info*>& info_seq,
            const SelectorState& selector_state) = 0;
 
    virtual void read(
            std::vector<Data*>& sample_seq,
            std::vector<Info*>& info_seq,
            const SelectorState& selector_state) = 0;
 
    virtual void return_loan(
            std::vector<Data*>& sample_seq,
            std::vector<Info*>& info_seq) = 0;
 
    /*
     * @brief Virtual destructor
     */
    virtual ~TStreamReader()
    {
 
    }
 
private:
    void convert_samples(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq)
    {
        RTI_ROUTING_SAMPLE_VECTOR_COPY_PTRS(sample_seq, sample_seq_);
        RTI_ROUTING_SAMPLE_VECTOR_COPY_PTRS(info_seq, info_seq_);
    }
 
private:
    std::vector<Data*> sample_seq_;
    std::vector<Info*> info_seq_;
 
};
 
template <typename Data, typename Info>
class NoOpStreamReader : public TStreamReader<Data, Info> {
 
public:
    virtual void take(
            std::vector<Data*>&,
            std::vector<Info*>&) RTI_OVERRIDE
    {
    }
 
    virtual void read(
            std::vector<Data*>&,
            std::vector<Info*>&) RTI_OVERRIDE
    {
    }
 
    virtual void take(
            std::vector<Data*>&,
            std::vector<Info*>&,
            const SelectorState&) RTI_OVERRIDE
    {
    }
 
    virtual void read(
            std::vector<Data*>&,
            std::vector<Info*>&,
            const SelectorState&) RTI_OVERRIDE
    {
    }
 
    virtual void return_loan(
            std::vector<Data*>&,
            std::vector<Info*>&) RTI_OVERRIDE
    {
    }
 
    virtual void* create_content_query(
            void *,
            const dds::topic::Filter&) RTI_OVERRIDE
    {
        return NULL;
    }
 
 
    virtual void delete_content_query(void*) RTI_OVERRIDE
    {
 
    }
 
 
 
    virtual ~NoOpStreamReader()
    {
    }
 
};
 
typedef NoOpStreamReader<dds::core::xtypes::DynamicData, dds::sub::SampleInfo>
        DynamicDataStreamReader;
 
 
 
class SelectorStateAdapter {
public:
    typedef RTI_RoutingServiceSelectorState native_type;
 
    static void initialize(native_type& native_value)
    {
        RTI_RoutingServiceSelectorState native_state =
                RTI_RoutingServiceSelectorState_INITIALIZER;
        memcpy(&native_value, &native_state, sizeof(native_state));
    }
 
    static void finalize(native_type& native_value)
    {
        if (native_value.content.expression != NULL) {
            DDS_String_free(native_value.content.expression);
        }
        DDS_StringSeq_finalize(&native_value.content.expression_parameters);
    }
 
    static void copy(native_type& destination, const native_type& source)
    {
        destination.sample_state = source.sample_state;
        destination.view_state = source.view_state;
        destination.instance_state = source.instance_state;
        destination.instance_handle = source.instance_handle;
        destination.instance_selection = source.instance_selection;
        destination.sample_count_max = source.sample_count_max;
        destination.query_data = source.query_data;
        char *result = DDS_String_replace(
                &destination.content.expression,
                source.content.expression);
        if (source.content.expression != NULL && result == NULL) {
            throw std::bad_alloc();
        }
        if (DDS_StringSeq_copy(
                &destination.content.expression_parameters,
                &source.content.expression_parameters) == NULL) {
            throw std::bad_alloc();
        }
    }
 
    static bool equals(const native_type& first, const native_type& second)
    {
        if (first.sample_state != second.sample_state) {
            return false;
        }
        if (first.view_state != second.view_state) {
            return false;
        }
        if (first.instance_state != second.instance_state) {
            return false;
        }
        if (first.instance_selection != second.instance_selection) {
            return false;
        }
        if (first.sample_count_max != second.sample_count_max) {
            return false;
        }
        if (first.query_data != second.query_data) {
            return false;
        }
        if (first.content.expression != NULL
                && second.content.expression != NULL
                && strcmp(first.content.expression, second.content.expression) != 0) {
            return false;
        }
        return DDS_StringSeq_equals(
                &first.content.expression_parameters,
                &second.content.expression_parameters);
    }
 
};
 
} }
 
// native_type_traits needs to be defined in rti::core
namespace core {
 
template <>
struct native_type_traits<rti::routing::adapter::SelectorState> {
    typedef rti::routing::adapter::SelectorStateAdapter adapter_type;
    typedef RTI_RoutingServiceSelectorState native_type;
};
 
}
 
namespace routing { namespace adapter {
 
class SelectorState : public rti::core::NativeValueType<SelectorState> {
public:
    RTI_NATIVE_VALUE_TYPE_DEFINE_DEFAULT_MOVE_OPERATIONS(SelectorState)
 
    typedef rti::core::NativeValueType<SelectorState> Base;
public:
    SelectorState()
    {
 
    }
 
    SelectorState(const RTI_RoutingServiceSelectorState &native)
            : Base(native)
    {
    }
 
    dds::sub::status::DataState state() const
    {
        return dds::sub::status::DataState(
                dds::sub::status::SampleState(native().sample_state),
                dds::sub::status::ViewState(native().view_state),
                dds::sub::status::InstanceState(native().instance_state));
    }
 
    SelectorState& state(const dds::sub::status::DataState& state)
    {
         native().sample_state = static_cast<DDS_SampleStateMask>(
                 state.sample_state().to_ulong());
         native().view_state = static_cast<DDS_ViewStateMask>(
                 state.view_state().to_ulong());
         native().instance_state = static_cast<DDS_InstanceStateMask>(
                 state.instance_state().to_ulong());
         return *this;
    }
 
    int32_t max_samples() const
    {
       return native().sample_count_max;
    }
 
    SelectorState& max_samples(const int32_t count)
    {
        native().sample_count_max = count;
        return *this;
    }
 
    dds::core::InstanceHandle instance() const
    {
        if (native().instance_selection
                == RTI_ROUTING_SERVICE_THIS_INSTANCE_SELECTION) {
            return rti::core::native_conversions::from_native_handle(
                        native().instance_handle);
        }
 
        return dds::core::InstanceHandle::nil();
    }
 
    SelectorState& instance(const dds::core::InstanceHandle& handle)
    {
        native().instance_handle =
                static_cast<DDS_InstanceHandle_t>(handle->native());
        native().instance_selection = RTI_ROUTING_SERVICE_THIS_INSTANCE_SELECTION;
        return *this;
    }
 
    dds::core::InstanceHandle next_instance() const
    {
        if (native().instance_selection
                == RTI_ROUTING_SERVICE_NEXT_INSTANCE_SELECTION) {
            return rti::core::native_conversions::from_native_handle(
                        native().instance_handle);
        }
 
        return dds::core::InstanceHandle::nil();
    }
 
    SelectorState& next_instance(
            const dds::core::InstanceHandle& handle)
    {
        native().instance_handle =
                static_cast<DDS_InstanceHandle_t>(handle->native());
        native().instance_selection = RTI_ROUTING_SERVICE_NEXT_INSTANCE_SELECTION;
        return *this;
    }
 
    void* content()
    {
        return native().query_data;
    }
 
    SelectorState& content(void *query_data)
    {
        native().query_data = query_data;
        return *this;
    }
 
    dds::topic::Filter filter() const
    {
        return dds::topic::Filter(
                native().content.expression == NULL ? "" : native().content.expression,
                rti::core::native_conversions::from_native<std::string>(
                        native().content.expression_parameters));
    }
 
    SelectorState& filter(const dds::topic::Filter& filter)
    {
        DDS_String_replace(
                &native().content.expression,
                filter.expression().c_str());
        rti::core::native_conversions::to_native(
                native().content.expression_parameters,
                filter->parameters());
        return *this;
    }
 
};
 
}}}
 
#endif // RTI_ROUTING_ADAPTER_STREAM_READER_HPP_