Input.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_PROCESSOR_INPUT_HPP_
#define RTI_ROUTING_PROCESSOR_INPUT_HPP_
 
#include <dds/core/Value.hpp>
#include <dds/core/SafeEnumeration.hpp>
#include <rti/core/NativeValueType.hpp>
#include <rti/core/Entity.hpp>
#include <dds/sub/DataReader.hpp>
#include <dds/core/xtypes/DynamicData.hpp>
 
#include "routingservice/routingservice_processor.h"
#include "routingservice/routingservice_adapter_new.h"
#include <rti/routing/StreamInfo.hpp>
#include <rti/routing/adapter/StreamReader.hpp>
#include <rti/routing/processor/LoanedSamples.hpp>
#include <rti/routing/processor/Query.hpp>
 
namespace rti { namespace routing { namespace processor {
 
class Route;
 
template <typename Data, typename Info> class TypedInput;
 
template <typename Data, typename Info = dds::sub::SampleInfo>
class TypedInput;
 
class Input {
public:
 
    /*i
     */
    Input(RTI_RoutingServiceInput *native,
          int32_t index,
          RTI_RoutingServiceRoute *native_route)
        : native_(native),
          index_(index),
        native_route_(native_route),
        stream_info_(*RTI_RoutingServiceInput_get_stream_info(native)),
        name_(RTI_RoutingServiceInput_get_name(native))
    {
    }
 
    /*i
     *
     */
    const rti::routing::StreamInfo& stream_info() const
    {
        return stream_info_;
    }
 
     /*i
     *
     */
    const std::string& name() const
    {
        return name_;
    }
 
    /*i
     *
     */
    int32_t index()
    {
        return index_;
    }
 
    template <typename Data, typename Info>
    TypedInput<Data, Info> get()
    {
        return this;
    }
 
    template <typename Data>
    TypedInput<Data, dds::sub::SampleInfo> get()
    {
        return this;
    }
 
    /*i
     *
     */
    RTI_RoutingServiceInput* native()
    {
        return native_;
    }
 
    // Implementation details
private:
 
    template <typename Data, typename Info> friend class TypedInput;
    friend class Route;
    friend class std::allocator<Input>;
    RTI_RoutingServiceInput *native_;
    int32_t index_;
    RTI_RoutingServiceRoute *native_route_;
    rti::routing::StreamInfo stream_info_;
    std::string name_;
    DDS_Entity *native_topic_;
};
 
template <typename Data, typename Info = dds::sub::SampleInfo>
class Selector;
 
template <typename Data, typename Info>
class TypedInput{
public:
 
    typedef rti::routing::processor::Selector<Data, Info> Selector;
 
    TypedInput(Input *input);
 
    const rti::routing::StreamInfo& stream_info() const;
 
    const std::string& name() const;
 
    LoanedSamples<Data, Info> take();
 
    LoanedSamples<Data, Info> read();
 
    Selector select();
 
    rti::routing::processor::Query query(
            const dds::topic::Filter& filter);
 
    bool active();
 
    Data create_data();
 
    /*i
     *
     */
    TypedInput<Data, Info>* operator->()
    {
        return this;
    }
 
 
    dds::sub::DataReader<Data> dds_data_reader()
    {
        using rti::core::detail::create_from_native_entity;
 
        DDS_DataReader *native_reader =
                RTI_RoutingServiceInput_get_dds_reader(input_->native_);
        if (native_reader == NULL) {
            throw dds::core::InvalidArgumentError(
                    "invalid argument: input does not hold a DDS StreamReader");
        }
 
        typedef dds::sub::DataReader<Data> data_reader_type;
        return rti::core::detail::create_from_native_entity<data_reader_type>(
                native_reader);
    }
 
private:
    TypedInput();
 
    friend class rti::routing::processor::Selector<Data, Info>;
 
    LoanedSamples<Data, Info> take(
            const rti::routing::adapter::SelectorState& selector_state);
 
    LoanedSamples<Data, Info> read(
            const rti::routing::adapter::SelectorState& selector_state);
 
    friend class rti::routing::processor::Route;
    Input *input_;
};
 
 
template <typename Data, typename Info>
class Selector {
public:
 
    Selector(const rti::routing::processor::TypedInput<Data, Info> input)
        : typed_input_(input), query_(dds::core::null)
    {
 
    }
 
    Selector(const Selector& other)
        : typed_input_(other.typed_input_),
          state_(other.state_),
          query_(other.query_)
    {
 
    }
 
    Selector& state(
            const dds::sub::status::DataState& the_state)
    {
        state_.state(the_state);
        return *this;
    }
 
    Selector& max_samples(
            const int32_t count)
    {
        state_.max_samples(count);
        return *this;
    }
 
    Selector& instance(
            const dds::core::InstanceHandle& the_handle)
    {
        state_.instance(the_handle);
        return *this;
    }
 
    Selector& next_instance(
            const dds::core::InstanceHandle& the_handle)
    {
        state_.next_instance(the_handle);
        return *this;
    }
 
    Selector& filter(
            const dds::topic::Filter& the_filter)
    {
        state_.filter(the_filter);
        return *this;
    }
 
 
    Selector& query(const rti::routing::processor::Query& the_query)
    {
        query_ = the_query;
        state_.content(the_query.delegate().get()->query_data_);
        return *this;
    }
 
    LoanedSamples<Data, Info> take()
    {
        return typed_input_.take(state_);
    }
 
    LoanedSamples<Data, Info> read()
    {
        return typed_input_.read(state_);
    }
 
private:
    rti::routing::processor::TypedInput<Data, Info> typed_input_;
    rti::routing::adapter::SelectorState state_;
    rti::routing::processor::Query query_;
};
 
 
template <typename Data, typename Info>
struct create_data_from_input {
 
    static Data get(TypedInput<Data, Info>& )
    {
        return Data();
    }
};
 
template <typename Info>
struct create_data_from_input<dds::core::xtypes::DynamicData, Info> {
 
    static dds::core::xtypes::DynamicData get(
            TypedInput<dds::core::xtypes::DynamicData, Info>& input)
    {
        if (input->stream_info().type_info().type_representation_kind()
                != TypeRepresentationKind::DYNAMIC_TYPE) {
            throw dds::core::PreconditionNotMetError(
                    "inconsistent data representation kind");
        }
        dds::core::xtypes::DynamicType *type_code =
                static_cast<dds::core::xtypes::DynamicType *> (
                input->stream_info().type_info().type_representation());
        return dds::core::xtypes::DynamicData(*type_code);
    }
};
 
template <typename Data, typename Info>
TypedInput<Data, Info>::TypedInput(Input *input)
: input_(input)
{
}
 
template <typename Data, typename Info>
const rti::routing::StreamInfo& TypedInput<Data, Info>::stream_info() const
{
    return input_->stream_info_;
}
 
template <typename Data, typename Info>
const std::string& TypedInput<Data, Info>::name() const
{
    return input_->name_;
}
 
template <typename Data, typename Info>
LoanedSamples<Data, Info> TypedInput<Data, Info>::take()
{
    RTI_RoutingServiceLoanedSamples native_samples =
            RTI_RoutingServiceLoanedSamples_INITIALIZER;
    if (!RTI_RoutingServiceInput_take(
            input_->native_,
            &native_samples)) {
        throw dds::core::Error("error taking samples from native input");
    }
    return LoanedSamples<Data, Info>(input_->native_, native_samples);
}
 
template <typename Data, typename Info>
LoanedSamples<Data, Info> TypedInput<Data, Info>::read()
{
    RTI_RoutingServiceLoanedSamples native_samples =
            RTI_RoutingServiceLoanedSamples_INITIALIZER;
    if (!RTI_RoutingServiceInput_read(
            input_->native_,
            &native_samples)) {
        throw dds::core::Error("error reading samples from native input");
    }
    return LoanedSamples<Data, Info>(input_->native_, native_samples);
}
 
template <typename Data, typename Info>
rti::routing::processor::Selector<Data, Info>
TypedInput<Data, Info>::select()
{
    return rti::routing::processor::Selector<Data, Info>(*this);
}
 
template <typename Data, typename Info>
rti::routing::processor::Query TypedInput<Data, Info>::query(
        const dds::topic::Filter& filter)
{
    return rti::routing::processor::Query(input_->native_, filter);
 
}
 
template <typename Data, typename Info>
LoanedSamples<Data, Info> TypedInput<Data, Info>::take(
        const rti::routing::adapter::SelectorState& selector_state)
{
    RTI_RoutingServiceLoanedSamples native_samples =
            RTI_RoutingServiceLoanedSamples_INITIALIZER;
    if (!RTI_RoutingServiceInput_take_w_selector(
            input_->native_,
            &native_samples,
            &selector_state.native())) {
        throw dds::core::Error("error taking samples with selector from native input");
    }
    return LoanedSamples<Data, Info>(input_->native_, native_samples);
}
 
template <typename Data, typename Info>
LoanedSamples<Data, Info> TypedInput<Data, Info>::read(
        const rti::routing::adapter::SelectorState& selector_state)
{
    RTI_RoutingServiceLoanedSamples native_samples =
            RTI_RoutingServiceLoanedSamples_INITIALIZER;
    if (!RTI_RoutingServiceInput_read_w_selector(
            input_->native_,
            &native_samples,
            &selector_state.native())) {
        throw dds::core::Error("error reading samples with selector from native input");
    }
    return LoanedSamples<Data, Info>(input_->native_, native_samples);
}
 
template <typename Data, typename Info>
bool TypedInput<Data, Info>::active()
{
    return RTI_RoutingServiceInput_is_active(input_->native_) ? true : false;
}
 
template <typename Data, typename Info>
Data TypedInput<Data,Info>::create_data()
{
    return create_data_from_input<Data, Info>::get(*this);
}
 
typedef TypedInput<dds::core::xtypes::DynamicData> DynamicDataInput;
typedef dds::sub::DataReader<dds::core::xtypes::DynamicData> DynamicDataReader;
 
} } }
 
 
#endif // RTI_ROUTING_PROCESSOR_INPUT_HPP_