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;

inline
rti::routing::StreamInfo from_native_input(
        RTI_RoutingServiceStreamReaderExt *native,
        RTI_RoutingServiceRoute *native_route)
{
    return rti::routing::StreamInfo(
            *RTI_RoutingServiceRoute_get_input_stream_info(native_route, native));
}

inline
std::string name_from_native_input(
        RTI_RoutingServiceStreamReaderExt *native,
        RTI_RoutingServiceRoute *native_route)
{
    return RTI_RoutingServiceRoute_get_input_name(
            native_route,
            native);
}

template <typename Data, typename Info = dds::sub::SampleInfo>
class TypedInput;

class Input : private rti::routing::adapter::StreamReader {
public:

    /*i
     */
    Input(RTI_RoutingServiceStreamReaderExt *native,
          int32_t index,
          RTI_RoutingServiceRoute *native_route,
          RTI_RoutingServiceEnvironment *native_env)
        : native_(native),
          index_(index),
        native_route_(native_route),
        native_env_(native_env),
        stream_info_(from_native_input(native, native_route)),
        name_(name_from_native_input(native, native_route))
    {
    }

    /*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;
    }


    /*
     * --- StreamReader Implementation ----------------------------------------
     */
private:
    /*i
     * @brief Returns the underlying rti::routing::adapter::StreamReader.
     *
     * @warning The returned object is a wrapper to an implementation of the
     *          StreamReader, which may be an object returned by any of the
     *          possible adapter API bindings.
     *          The use of this operation is not recommended unless your Processor
     *          implementation strictly needs to perform operations on the
     *          StreamReader.
     */
    rti::routing::adapter::StreamReader& stream_reader()
    {
        return *this;
    }

    void take(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) RTI_FINAL
    {
        using namespace rti::routing::adapter;

        detail::NativeSamples native_samples;
        native_->take(
                native_->stream_reader_data,
                &native_samples.sample_array_,
                &native_samples.info_array_,
                &native_samples.length_,
                native_env_);
        RTI_ROUTING_THROW_ON_ENV_ERROR(native_env_);
        get_samples_from_native(sample_seq, info_seq, native_samples);
    }

    void read(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) RTI_FINAL
    {
        using namespace rti::routing::adapter;

        detail::NativeSamples native_samples;
        native_->take(
                native_->stream_reader_data,
                &native_samples.sample_array_,
                &native_samples.info_array_,
                &native_samples.length_,
                native_env_);
        RTI_ROUTING_THROW_ON_ENV_ERROR(native_env_);
        get_samples_from_native(sample_seq, info_seq, native_samples);
    }

    void take(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq,
            const rti::routing::adapter::SelectorState& selector_state) RTI_FINAL
    {
        detail::NativeSamples native_samples;
        native_->take_w_selector(
                native_->stream_reader_data,
                &native_samples.sample_array_,
                &native_samples.info_array_,
                &native_samples.length_,
                &selector_state.native(),
                native_env_);
        RTI_ROUTING_THROW_ON_ENV_ERROR(native_env_);
        get_samples_from_native(sample_seq, info_seq, native_samples);
    }

    void read(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq,
            const rti::routing::adapter::SelectorState& selector_state) RTI_FINAL
    {
        detail::NativeSamples native_samples;
        native_->read_w_selector(
                native_->stream_reader_data,
                &native_samples.sample_array_,
                &native_samples.info_array_,
                &native_samples.length_,
                &selector_state.native(),
                native_env_);
        RTI_ROUTING_THROW_ON_ENV_ERROR(native_env_);
        get_samples_from_native(sample_seq, info_seq, native_samples);
    }

    void return_loan(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq) RTI_FINAL
    {
        detail::NativeSamples native_samples;
        native_samples.length_ = sample_seq.size();
        native_samples.sample_array_ = static_cast<RTI_RoutingServiceSample*>
                (sample_seq[native_samples.length_]);
        if (info_seq.size() > 0) {
            native_samples.info_array_ = static_cast<RTI_RoutingServiceSample*>
                (info_seq[info_seq.size()]);
        }

        native_->return_loan(
                native_->stream_reader_data,
                native_samples.sample_array_,
                native_samples.info_array_,
                native_samples.length_,
                native_env_);
        RTI_ROUTING_THROW_ON_ENV_ERROR(native_env_);
        sample_seq.clear();
        info_seq.clear();
    }


    void* create_content_query(
            void* old_query_data,
            const dds::topic::Filter& filter) RTI_FINAL
    {
        if (native_->create_content_query == NULL) {
            throw dds::core::UnsupportedError("Unsupported operation: create_content_query");
        }

        RTI_RoutingServiceSelectorContent content =
                RTI_RoutingServiceSelectorContent_INITIALIZER;
        content.expression = (char *) filter.expression().c_str();
        rti::core::native_conversions::to_native(
                content.expression_parameters,
                filter->parameters());
        void* query_data = native_->create_content_query(
                native_->stream_reader_data,
                old_query_data,
                &content,
                native_env_);
        DDS_StringSeq_finalize(&content.expression_parameters);
        RTI_ROUTING_THROW_ON_ENV_ERROR(native_env_);
        return query_data;
    }

    void delete_content_query(void* query_data) RTI_FINAL
    {
        if (native_->delete_content_query == NULL) {
            throw dds::core::UnsupportedError("delete_content_query");
        }

        native_->delete_content_query(
                native_->stream_reader_data,
                query_data,
                native_env_);
        RTI_ROUTING_THROW_ON_ENV_ERROR(native_env_);
    }


    virtual void update(const std::map<std::string, std::string>&) RTI_FINAL
    {
        throw dds::core::PreconditionNotMetError(
                "update not applicable within a Processor notification context");
    }

    // Implementation details
private:
    void get_samples_from_native(
            std::vector<SamplePtr>& sample_seq,
            std::vector<InfoPtr>& info_seq,
            detail::NativeSamples& native_samples)
    {
        sample_seq.reserve(native_samples.length_ + 1);
        RTI_ROUTING_SAMPLE_VECTOR_COPY_FROM_NATIVE(
                sample_seq,
                native_samples.sample_array_,
                native_samples.length_);
        sample_seq[native_samples.length_] =
                native_samples.sample_array_;
        if (native_samples.info_array_ != NULL) {
            info_seq.reserve(native_samples.length_ + 1);
            RTI_ROUTING_SAMPLE_VECTOR_COPY_FROM_NATIVE(
                    info_seq,
                    native_samples.info_array_,
                    native_samples.length_);
            info_seq[info_seq.size()] =
                    native_samples.info_array_;
        } else {
            info_seq.clear();
        }
    }

    RTI_RoutingServiceStreamReaderExt* native()
    {
        return native_;
    }

    template <typename Data, typename Info> friend class TypedInput;
    friend class Route;
    friend class std::allocator<Input>;
    RTI_RoutingServiceStreamReaderExt *native_;
    int32_t index_;
    RTI_RoutingServiceRoute *native_route_;
    RTI_RoutingServiceEnvironment *native_env_;
    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_RoutingServiceRoute_get_dds_reader(
                input_->native_route_,
                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()
{
    detail::NativeSamples native_samples;
    input_->native_->take(
            input_->native_->stream_reader_data,
            &native_samples.sample_array_,
            &native_samples.info_array_,
            &native_samples.length_,
            input_->native_env_);
    RTI_ROUTING_THROW_ON_ENV_ERROR(input_->native_env_);
    return LoanedSamples<Data, Info>(
            input_->native_,
            native_samples,
            input_->native_env_);
}

template <typename Data, typename Info>
LoanedSamples<Data, Info> TypedInput<Data, Info>::read()
{
    detail::NativeSamples native_samples;
    input_->native_->read(
            input_->native_->stream_reader_data,
            &native_samples.sample_array_,
            &native_samples.info_array_,
            &native_samples.length_,
            input_->native_env_);
    RTI_ROUTING_THROW_ON_ENV_ERROR(input_->native_env_);
    return LoanedSamples<Data, Info>(
            input_->native_,
            native_samples,
            input_->native_env_);
}

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_->stream_reader(),
            filter);

}

template <typename Data, typename Info>
LoanedSamples<Data, Info> TypedInput<Data, Info>::take(
        const rti::routing::adapter::SelectorState& selector_state)
{
    detail::NativeSamples native_samples;
    input_->native_->take_w_selector(
            input_->native_->stream_reader_data,
            &native_samples.sample_array_,
            &native_samples.info_array_,
            &native_samples.length_,
            &selector_state.native(),
            input_->native_env_);
    RTI_ROUTING_THROW_ON_ENV_ERROR(input_->native_env_);
    return LoanedSamples<Data, Info>(
            input_->native_,
            native_samples,
            input_->native_env_);
}

template <typename Data, typename Info>
LoanedSamples<Data, Info> TypedInput<Data, Info>::read(
        const rti::routing::adapter::SelectorState& selector_state)
{
    detail::NativeSamples native_samples;
    input_->native_->read_w_selector(
            input_->native_->stream_reader_data,
            &native_samples.sample_array_,
            &native_samples.info_array_,
            &native_samples.length_,
            &selector_state.native(),
            input_->native_env_);
    RTI_ROUTING_THROW_ON_ENV_ERROR(input_->native_env_);
    return LoanedSamples<Data, Info>(
            input_->native_,
            native_samples,
            input_->native_env_);
}

template <typename Data, typename Info>
bool TypedInput<Data, Info>::active()
{
    return RTI_RoutingServiceRoute_is_input_active(
            input_->native_route_,
            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_