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;
    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;
    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;
    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;
    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_