DataReaderImpl.hpp

Go to the documentation of this file.

/*
 *                         Vortex OpenSplice
 *
 *   This software and documentation are Copyright 2006 to 2021 ADLINK
 *   Technology Limited, its affiliated companies and licensors. All rights
 *   reserved.
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 *
 */
#ifndef OSPL_DDS_SUB_TDATAREADER_IMPL_HPP_
#define OSPL_DDS_SUB_TDATAREADER_IMPL_HPP_

/*
 * OMG PSM class declaration
 */
#include <dds/sub/detail/DataReader.hpp>
#include <dds/sub/Query.hpp>
#include <dds/sub/detail/SamplesHolder.hpp>




/***************************************************************************
 *
 * dds/sub/DataReader<> WRAPPER implementation.
 * Declaration can be found in dds/sub/DataReader.hpp
 *
 ***************************************************************************/

// Implementation

namespace dds
{
namespace sub
{

//--------------------------------------------------------------------------------
//  DATAREADER
//--------------------------------------------------------------------------------

template <typename T>
DataReader<T>::Selector::Selector(DataReader& dr) : impl_(dr.delegate())
{
}

template <typename T>
typename DataReader<T>::Selector&
DataReader<T>::Selector::instance(const dds::core::InstanceHandle& h)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.instance(h);
    return *this;
}


template <typename T>
typename DataReader<T>::Selector&
DataReader<T>::Selector::next_instance(const dds::core::InstanceHandle& h)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.next_instance(h);
    return *this;
}


template <typename T>
typename DataReader<T>::Selector&
DataReader<T>::Selector::state(const dds::sub::status::DataState& s)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.filter_state(s);
    return *this;
}

template <typename T>
typename DataReader<T>::Selector&
DataReader<T>::Selector::content(const dds::sub::Query& query)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.filter_content(query);
    return *this;
}

template <typename T>
typename DataReader<T>::Selector&
DataReader<T>::Selector::max_samples(uint32_t n)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.max_samples(n);
    return *this;
}

template <typename T>
dds::sub::LoanedSamples<T>
DataReader<T>::Selector::read()
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    return impl_.read();
}

template <typename T>
dds::sub::LoanedSamples<T>
DataReader<T>::Selector::take()
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    return impl_.take();
}

template <typename T>
template <typename SamplesFWIterator>
uint32_t
DataReader<T>::Selector::read(SamplesFWIterator sfit, uint32_t max_samples)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    return impl_.read(sfit, max_samples);
}

template <typename T>
template <typename SamplesFWIterator>
uint32_t
DataReader<T>::Selector::take(SamplesFWIterator sfit,    uint32_t max_samples)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    return impl_.take(sfit, max_samples);
}

template <typename T>
template <typename SamplesBIIterator>
uint32_t
DataReader<T>::Selector::read(SamplesBIIterator sbit)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    return impl_.read(sbit);
}

template <typename T>
template <typename SamplesBIIterator>
uint32_t
DataReader<T>::Selector::take(SamplesBIIterator sbit)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    return impl_.take(sbit);
}

//--------------------------------------------------------------------------------
//  DATAREADER::MANIPULATORSELECTOR
//--------------------------------------------------------------------------------
template <typename T>
DataReader<T>::ManipulatorSelector::
ManipulatorSelector(DataReader& dr) : impl_(dr.delegate()) {}

template <typename T>
bool
DataReader<T>::ManipulatorSelector::read_mode()
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    return impl_.read_mode();
}

template <typename T>
void
DataReader<T>::ManipulatorSelector::read_mode(bool b)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.read_mode(b);
}

template <typename T>
typename DataReader<T>::ManipulatorSelector&
DataReader<T>::ManipulatorSelector::instance(const dds::core::InstanceHandle& h)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.instance(h);
    return *this;
}

template <typename T>
typename DataReader<T>::ManipulatorSelector&
DataReader<T>::ManipulatorSelector::next_instance(const dds::core::InstanceHandle& h)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.next_instance(h);
    return *this;
}

template <typename T>
typename DataReader<T>::ManipulatorSelector&
DataReader<T>::ManipulatorSelector::operator >>(dds::sub::LoanedSamples<T>& samples)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_ >> samples;
    return *this;
}

template <typename T>
typename DataReader<T>::ManipulatorSelector&
DataReader<T>::ManipulatorSelector::operator >> (ManipulatorSelector & (manipulator)(ManipulatorSelector&))
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    manipulator(*this);
    return *this;
}

template <typename T>
template <typename Functor>
typename DataReader<T>::ManipulatorSelector
DataReader<T>::ManipulatorSelector::operator >> (Functor f)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    f(*this);
    return *this;
}
template <typename T>
typename DataReader<T>::ManipulatorSelector&
DataReader<T>::ManipulatorSelector::state(const dds::sub::status::DataState& s)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.filter_state(s);
    return *this;
}

template <typename T>
typename DataReader<T>::ManipulatorSelector&
DataReader<T>::ManipulatorSelector::content(const dds::sub::Query& query)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.filter_content(query);
    return *this;
}

template <typename T>
typename DataReader<T>::ManipulatorSelector&
DataReader<T>::ManipulatorSelector::max_samples(uint32_t n)
{
    ISOCPP_REPORT_STACK_DELEGATE_BEGIN(this->impl_.get_reader());

    impl_.max_samples(n);
    return *this;
}

template <typename T>
DataReader<T>::DataReader(
    const dds::sub::Subscriber& sub,
    const dds::topic::Topic<T>& topic):
        ::dds::core::Reference(new DELEGATE<T>(sub, topic, sub.is_nil() ? dds::sub::qos::DataReaderQos() : sub->default_datareader_qos()))
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(sub);

    this->delegate()->init(this->impl_);
}

template <typename T>
DataReader<T>::DataReader(
    const dds::sub::Subscriber& sub,
    const ::dds::topic::Topic<T>& topic,
    const dds::sub::qos::DataReaderQos& qos,
    dds::sub::DataReaderListener<T>* listener,
    const dds::core::status::StatusMask& mask) :
        ::dds::core::Reference(new DELEGATE<T>(sub, topic, qos, listener, mask))
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(sub);

    this->delegate()->init(this->impl_);
}

#ifdef OMG_DDS_CONTENT_SUBSCRIPTION_SUPPORT
template <typename T>
DataReader<T>::DataReader(
    const dds::sub::Subscriber& sub,
    const dds::topic::ContentFilteredTopic<T>& topic) :
        ::dds::core::Reference(new DELEGATE<T>(sub, topic, sub.default_datareader_qos()))
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(sub);

    this->delegate()->init(this->impl_);
}

template <typename T>
DataReader<T>::DataReader(
    const dds::sub::Subscriber& sub,
    const ::dds::topic::ContentFilteredTopic<T>& topic,
    const dds::sub::qos::DataReaderQos& qos,
    dds::sub::DataReaderListener<T>* listener,
    const dds::core::status::StatusMask& mask) :
        ::dds::core::Reference(new DELEGATE<T>(sub, topic, qos, listener, mask))
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(sub);

    this->delegate()->init(this->impl_);
}
#endif /* OMG_DDS_CONTENT_SUBSCRIPTION_SUPPORT */

#ifdef OMG_DDS_MULTI_TOPIC_SUPPORT
template <typename T>
DataReader<T>::DataReader(
    const dds::sub::Subscriber& sub,
    const dds::topic::MultiTopic<T>& topic) :
        ::dds::core::Reference(new DELEGATE<T>(sub, topic))
{
    this->delegate()->init(this->impl_);
}

template <typename T>
DataReader<T>::DataReader(
    const dds::sub::Subscriber& sub,
    const ::dds::topic::MultiTopic<T>& topic,
    const dds::sub::qos::DataReaderQos& qos,
    dds::sub::DataReaderListener<T>* listener,
    const dds::core::status::StatusMask& mask) :
    ::dds::core::Reference(new DELEGATE<T>(sub, topic, qos, listener, mask))
{
    this->delegate()->init(this->impl_);
}
#endif /* OMG_DDS_MULTI_TOPIC_SUPPORT */

template <typename T>
DataReader<T>::~DataReader() { }

template <typename T>
dds::sub::status::DataState
DataReader<T>::default_filter_state()
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->default_filter_state();
}

template <typename T>
DataReader<T>& DataReader<T>::default_filter_state(const dds::sub::status::DataState& status)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    this->delegate()->default_filter_state(status);
    return *this;
}

template <typename T>
DataReader<T>& DataReader<T>::operator >>(dds::sub::LoanedSamples<T>& ls)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    ls = this->read();
    return *this;
}

template <typename T>
typename DataReader<T>::ManipulatorSelector
DataReader<T>::operator >> (ManipulatorSelector& (manipulator)(ManipulatorSelector&))
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    ManipulatorSelector selector(*this);
    manipulator(selector);
    return selector;
}

template <typename T>
template <typename Functor>
typename DataReader<T>::ManipulatorSelector
DataReader<T>::operator >> (Functor f)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    ManipulatorSelector selector(*this);
    f(selector);
    return selector;
}
template <typename T>
LoanedSamples<T>
DataReader<T>::read()
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->read();
}

template <typename T>
LoanedSamples<T>
DataReader<T>::take()
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->take();
}


template <typename T>
template <typename SamplesFWIterator>
uint32_t
DataReader<T>::read(SamplesFWIterator sfit, uint32_t max_samples)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->read(sfit, max_samples);
}

template <typename T>
template <typename SamplesFWIterator>
uint32_t
DataReader<T>::take(SamplesFWIterator sfit, uint32_t max_samples)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->take(sfit, max_samples);
}

template <typename T>
template <typename SamplesBIIterator>
uint32_t
DataReader<T>::read(SamplesBIIterator sbit)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->read(sbit);
}

template <typename T>
template <typename SamplesBIIterator>
uint32_t
DataReader<T>::take(SamplesBIIterator sbit)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->take(sbit);
}

template <typename T>
typename DataReader<T>::Selector
DataReader<T>::select()
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    Selector selector(*this);
    return selector;
}

template <typename T>
dds::topic::TopicInstance<T>
DataReader<T>::key_value(const dds::core::InstanceHandle& h)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->key_value(h);
}

template <typename T>
T&
DataReader<T>::key_value(T& sample, const dds::core::InstanceHandle& h)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->key_value(sample, h);
}

template <typename T>
const dds::core::InstanceHandle
DataReader<T>::lookup_instance(const T& key) const
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->lookup_instance(key);
}

template <typename T>
void
DataReader<T>::listener(
    Listener* listener,
    const dds::core::status::StatusMask& event_mask)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    this->delegate()->listener(listener, event_mask);
}

template <typename T>
typename DataReader<T>::Listener*
DataReader<T>::listener() const
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(*this);

    return this->delegate()->listener();
}

}
}




/***************************************************************************
 *
 * dds/sub/detail/DataReader<> DELEGATE implementation.
 * Declaration can be found in dds/sub/detail/DataReader.hpp
 *
 * Implementation and declaration have been separated because some circular
 * dependencies, like with DataReaderListener and AnyDataReader.
 *
 ***************************************************************************/

#include <dds/sub/AnyDataReader.hpp>
#include <dds/sub/DataReaderListener.hpp>
#include <dds/topic/Topic.hpp>
#include <dds/topic/ContentFilteredTopic.hpp>
#include <org/opensplice/sub/AnyDataReaderDelegate.hpp>


template <typename T>
dds::sub::detail::DataReader<T>::DataReader(const dds::sub::Subscriber& sub,
           const dds::topic::Topic<T>& topic,
           const dds::sub::qos::DataReaderQos& qos,
           dds::sub::DataReaderListener<T>* listener,
           const dds::core::status::StatusMask& mask)
    : ::org::opensplice::sub::AnyDataReaderDelegate(qos, topic), sub_(sub)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(topic);

    /* Create a implicit subscriber with the topic participant when needed. */
    if (sub_.is_nil()) {
        sub_ = dds::sub::Subscriber(topic->domain_participant());
    }

    /* Merge the topic QoS implicitly when needed. */
    if (topic.qos()->force_merge()) {
        qos_ = topic.qos();
    }

    common_constructor(listener, mask);
}

template <typename T>
dds::sub::detail::DataReader<T>::DataReader(const dds::sub::Subscriber& sub,
           const dds::topic::ContentFilteredTopic<T, dds::topic::detail::ContentFilteredTopic>& topic,
           const dds::sub::qos::DataReaderQos& qos,
           dds::sub::DataReaderListener<T>* listener,
           const dds::core::status::StatusMask& mask)
    : ::org::opensplice::sub::AnyDataReaderDelegate(qos, topic), sub_(sub)
{
    ISOCPP_REPORT_STACK_DDS_BEGIN(topic);

    /* Create a implicit subscriber with the topic participant when needed. */
    if (sub_.is_nil()) {
        sub_ = dds::sub::Subscriber(topic->domain_participant());
    }

    /* Merge the topic QoS implicitly when needed. */
    if (topic.topic().qos()->force_merge()) {
        qos_ = topic.topic().qos();
    }

    common_constructor(listener, mask);
}

template <typename T>
void
dds::sub::detail::DataReader<T>::common_constructor(
            dds::sub::DataReaderListener<T>* listener,
            const dds::core::status::StatusMask& mask)
{
    if (dds::topic::is_topic_type<T>::value == 0) {
        ISOCPP_THROW_EXCEPTION(ISOCPP_PRECONDITION_NOT_MET_ERROR, "DataReader cannot be created, topic information not found");
    }

    org::opensplice::sub::qos::DataReaderQosDelegate drQos = qos_.delegate();

    // get and validate the kernel qos
    drQos.check();
    u_readerQos uQos = drQos.u_qos();

    u_subscriber uSubscriber = (u_subscriber)(sub_.delegate()->get_user_handle());

    std::string expression = this->AnyDataReaderDelegate::td_.delegate()->reader_expression();
    std::vector<c_value> params = this->AnyDataReaderDelegate::td_.delegate()->reader_parameters();

    std::string name = "reader <" + this->AnyDataReaderDelegate::td_.name() + ">";
    u_dataReader uReader = u_dataReaderNew(uSubscriber, name.c_str(), expression.c_str(),
            params.empty() ? NULL : &params[0], (os_uint32)params.size(), uQos);
    u_readerQosFree(uQos);

    if (!uReader) {
        ISOCPP_THROW_EXCEPTION(ISOCPP_ERROR, "Failed to create DataReader");
    } else {
        this->AnyDataReaderDelegate::td_.delegate()->incrNrDependents();
    }

    this->AnyDataReaderDelegate::setCopyOut(org::opensplice::topic::TopicTraits<T>::getCopyOut());
    this->AnyDataReaderDelegate::setCopyIn(org::opensplice::topic::TopicTraits<T>::getCopyIn());

    this->userHandle = u_object(uReader);
    this->listener_set((void*)listener, mask);
    this->set_domain_id(this->sub_.delegate()->get_domain_id());
}

template <typename T>
dds::sub::detail::DataReader<T>::~DataReader()
{
    if (!this->closed) {
        try {
            close();
        } catch (...) {

        }
    }
}

template <typename T>
void
dds::sub::detail::DataReader<T>::init(ObjectDelegate::weak_ref_type weak_ref)
{
    /* Set weak_ref before passing ourselves to other isocpp objects. */
    this->set_weak_ref(weak_ref);
    /* Register writer at publisher. */
    this->sub_.delegate()->add_datareader(*this);
    /* Use listener dispatcher from the publisher. */
    this->listener_dispatcher_set(this->sub_.delegate()->listener_dispatcher_get());
    /* This only starts listening when the status mask shows interest. */
    this->listener_enable();
    /* Enable when needed. */
    if (this->sub_.delegate()->is_enabled() && this->sub_.delegate()->is_auto_enable()) {
        this->enable();
    }
}

template <typename T>
dds::sub::status::DataState
dds::sub::detail::DataReader<T>::default_filter_state()
{
    org::opensplice::core::ScopedObjectLock scopedLock(*this);

    dds::sub::status::DataState state = this->status_filter_;

    scopedLock.unlock();

    return state;
}

template <typename T>
void
dds::sub::detail::DataReader<T>::default_filter_state(const dds::sub::status::DataState& state)
{
    org::opensplice::core::ScopedObjectLock scopedLock(*this);

    this->status_filter_ = state;

    scopedLock.unlock();
}


template <typename T>
dds::sub::LoanedSamples<T>
dds::sub::detail::DataReader<T>::read()
{
    dds::sub::LoanedSamples<T> samples;
    dds::sub::detail::LoanedSamplesHolder<T> holder(samples);

    this->AnyDataReaderDelegate::read((u_dataReader)(this->userHandle), this->status_filter_, holder, -1);

    return samples;
}

template <typename T>
dds::sub::LoanedSamples<T>
dds::sub::detail::DataReader<T>::take()
{
    dds::sub::LoanedSamples<T> samples;
    dds::sub::detail::LoanedSamplesHolder<T> holder(samples);

    this->AnyDataReaderDelegate::take((u_dataReader)(this->userHandle), this->status_filter_, holder, -1);

    return samples;
}

template <typename T>
template<typename SamplesFWIterator>
uint32_t
dds::sub::detail::DataReader<T>::read(SamplesFWIterator samples, uint32_t max_samples)
{
    dds::sub::detail::SamplesFWInteratorHolder<T, SamplesFWIterator> holder(samples);

    this->AnyDataReaderDelegate::read((u_dataReader)(this->userHandle), this->status_filter_, holder, max_samples);

    return holder.get_length();
}

template <typename T>
template<typename SamplesFWIterator>
uint32_t
dds::sub::detail::DataReader<T>::take(SamplesFWIterator samples, uint32_t max_samples)
{
    dds::sub::detail::SamplesFWInteratorHolder<T, SamplesFWIterator> holder(samples);

    this->AnyDataReaderDelegate::take((u_dataReader)(this->userHandle), this->status_filter_, holder, max_samples);

    return holder.get_length();
}

template <typename T>
template<typename SamplesBIIterator>
uint32_t
dds::sub::detail::DataReader<T>::read(SamplesBIIterator samples)
{
    dds::sub::detail::SamplesBIIteratorHolder<T, SamplesBIIterator> holder(samples);

    this->AnyDataReaderDelegate::read((u_dataReader)(this->userHandle), this->status_filter_, holder, -1);

    return holder.get_length();
}

template <typename T>
template<typename SamplesBIIterator>
uint32_t
dds::sub::detail::DataReader<T>::take(SamplesBIIterator samples)
{
    dds::sub::detail::SamplesBIIteratorHolder<T, SamplesBIIterator> holder(samples);

    this->AnyDataReaderDelegate::take((u_dataReader)(this->userHandle), this->status_filter_, holder, -1);

    return holder.get_length();
}

template <typename T>
dds::topic::TopicInstance<T>
dds::sub::detail::DataReader<T>::key_value(const dds::core::InstanceHandle& h)
{
    T key_holder;

    this->AnyDataReaderDelegate::get_key_value((u_dataReader)(this->userHandle), h, &key_holder);

    return dds::topic::TopicInstance<T>(h, key_holder);
}

template <typename T>
T&
dds::sub::detail::DataReader<T>::key_value(T& key, const dds::core::InstanceHandle& h)
{
    this->AnyDataReaderDelegate::get_key_value((u_dataReader)(this->userHandle), h, &key);

    return key;
}

template <typename T>
const dds::core::InstanceHandle
dds::sub::detail::DataReader<T>::lookup_instance(const T& key) const
{
    dds::core::InstanceHandle handle(this->AnyDataReaderDelegate::lookup_instance((u_dataReader)(this->userHandle), &key));

    return handle;
}

template <typename T>
const dds::sub::Subscriber&
dds::sub::detail::DataReader<T>::subscriber() const
{
    this->check();

    return sub_;
}

template <typename T>
void
dds::sub::detail::DataReader<T>::close()
{
    this->listener(NULL, dds::core::status::StatusMask::none());
    this->listener_dispatcher_reset();

    org::opensplice::core::ScopedObjectLock scopedLock(*this);

    this->sub_.delegate()->remove_datareader(*this);

    // Remove our dependency on the topicdescription, and drop our reference to it,
    // so that it can become garbage collected.
    // It is important that we also drop our reference to the topicdescription, since
    // subsequent dependencies between for example ContentFilteredTopic to Topic can
    // only be dropped by the destructor of the ContentFilteredTopic.
    this->AnyDataReaderDelegate::td_.delegate()->decrNrDependents();
    this->AnyDataReaderDelegate::td_ = dds::topic::TopicDescription(dds::core::null);

    org::opensplice::sub::AnyDataReaderDelegate::close();

    scopedLock.unlock();
}


template <typename T>
dds::sub::DataReaderListener<T>*
dds::sub::detail::DataReader<T>::listener()
{
    return reinterpret_cast<dds::sub::DataReaderListener<T>*>(this->listener_get());
}

template <typename T>
void
dds::sub::detail::DataReader<T>::listener(
        dds::sub::DataReaderListener<T>* l,
        const dds::core::status::StatusMask& event_mask)
{
    /* EntityDelegate takes care of thread safety. */
    this->listener_set((void*)l, event_mask);
    this->listener_enable();
}

template <typename T>
dds::sub::DataReader<T, dds::sub::detail::DataReader>
dds::sub::detail::DataReader<T>::wrapper()
{
    typename DataReader::ref_type ref =
            OSPL_CXX11_STD_MODULE::dynamic_pointer_cast<DataReader<T> >(this->get_strong_ref());
    dds::sub::DataReader<T, dds::sub::detail::DataReader> reader(ref);

    return reader;
}

template <typename T>
void
dds::sub::detail::DataReader<T>::listener_notify(
        ObjectDelegate::ref_type source,
        uint32_t                 triggerMask,
        void                    *eventData,
        void                    *l)
{
    /* The EntityDelegate takes care of the thread safety and always
     * provides a listener and source. */
    dds::sub::DataReaderListener<T>* listener =
            reinterpret_cast<dds::sub::DataReaderListener<T>*>(l);
    assert(listener);

    /* Get DataWriter wrapper from given source EntityDelegate. */
    typename DataReader::ref_type ref =
            OSPL_CXX11_STD_MODULE::dynamic_pointer_cast<DataReader<T> >(source);
    dds::sub::DataReader<T, dds::sub::detail::DataReader> reader(ref->wrapper());


    if (triggerMask & V_EVENT_DATA_AVAILABLE) {
        ref->reset_data_available();
        listener->on_data_available(reader);
    }

    if (triggerMask & V_EVENT_SAMPLE_REJECTED) {
        dds::core::status::SampleRejectedStatus status;
        status.delegate().v_status(((v_readerStatus)eventData)->sampleRejected);
        listener->on_sample_rejected(reader, status);
    }

    if (triggerMask & V_EVENT_LIVELINESS_CHANGED) {
        dds::core::status::LivelinessChangedStatus status;
        status.delegate().v_status(((v_readerStatus)eventData)->livelinessChanged);
        listener->on_liveliness_changed(reader, status);
    }

    if (triggerMask & V_EVENT_REQUESTED_DEADLINE_MISSED) {
        dds::core::status::RequestedDeadlineMissedStatus status;
        status.delegate().v_status(((v_readerStatus)eventData)->deadlineMissed);
        listener->on_requested_deadline_missed(reader, status);
    }

    if (triggerMask & V_EVENT_REQUESTED_INCOMPATIBLE_QOS) {
        dds::core::status::RequestedIncompatibleQosStatus status;
        status.delegate().v_status(((v_readerStatus)eventData)->incompatibleQos);
        listener->on_requested_incompatible_qos(reader, status);
    }

    if (triggerMask & V_EVENT_SAMPLE_LOST) {
        dds::core::status::SampleLostStatus status;
        status.delegate().v_status(((v_readerStatus)eventData)->sampleLost);
        listener->on_sample_lost(reader, status);
    }

    if (triggerMask & V_EVENT_SUBSCRIPTION_MATCHED) {
        dds::core::status::SubscriptionMatchedStatus status;
        status.delegate().v_status(((v_readerStatus)eventData)->subscriptionMatch);
        listener->on_subscription_matched(reader, status);
    }
}

template <typename T>
dds::sub::detail::DataReader<T>::Selector::Selector(typename DataReader<T>::ref_type dr)
    : mode(SELECT_MODE_READ), reader(dr), state_filter_is_set_(false),
      max_samples_((uint32_t)-1), query_(dds::core::null)
{
}

template <typename T>
typename dds::sub::detail::DataReader<T>::Selector&
dds::sub::detail::DataReader<T>::Selector::instance(const dds::core::InstanceHandle& h)
{
    this->handle = h;
    switch (this->mode) {
    case SELECT_MODE_READ:
    case SELECT_MODE_READ_INSTANCE:
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->mode = SELECT_MODE_READ_INSTANCE;
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        this->mode = SELECT_MODE_READ_INSTANCE_WITH_CONDITION;
        break;
    }

    return *this;
}

template <typename T>
typename dds::sub::detail::DataReader<T>::Selector&
dds::sub::detail::DataReader<T>::Selector::next_instance(const dds::core::InstanceHandle& h)
{
    this->handle = h;
    switch (this->mode) {
    case SELECT_MODE_READ:
    case SELECT_MODE_READ_INSTANCE:
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->mode = SELECT_MODE_READ_NEXT_INSTANCE;
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        this->mode = SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION;
        break;
    }

    return *this;
}

template <typename T>
typename dds::sub::detail::DataReader<T>::Selector&
dds::sub::detail::DataReader<T>::Selector::filter_state(const dds::sub::status::DataState& s)
{
    this->state_filter_ = s;
    this->state_filter_is_set_ = true;
    if ((this->mode == SELECT_MODE_READ_WITH_CONDITION) ||
        (this->mode == SELECT_MODE_READ_INSTANCE_WITH_CONDITION) ||
        (this->mode == SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION)) {
        if (!this->query_.delegate()->modify_state_filter(this->state_filter_)) {
            dds::sub::Query q(this->query_.data_reader(), this->query_.expression(), this->query_.delegate()->parameters());
            q.delegate()->state_filter(this->state_filter_);
            this->query_ = q;
        }
    }
    return *this;
}

template <typename T>
typename dds::sub::detail::DataReader<T>::Selector&
dds::sub::detail::DataReader<T>::Selector::max_samples(uint32_t n)
{
    this->max_samples_ = n;
    return *this;
}

template <typename T>
typename dds::sub::detail::DataReader<T>::Selector&
dds::sub::detail::DataReader<T>::Selector::filter_content(
    const dds::sub::Query& query)
{
    if (this->state_filter_is_set_) {
        if (!query.delegate()->modify_state_filter(this->state_filter_)) {
            dds::sub::Query q(query.data_reader(), query.expression(), query.delegate()->parameters());
            q.delegate()->state_filter(this->state_filter_);
            this->query_ = q;
        } else {
            this->query_ = query;
        }
    } else {
        this->query_ = query;
    }

    switch (mode) {
    case SELECT_MODE_READ:
        mode = SELECT_MODE_READ_WITH_CONDITION;
        break;
    case SELECT_MODE_READ_INSTANCE:
        mode = SELECT_MODE_READ_INSTANCE_WITH_CONDITION;
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE:
        mode = SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION;
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        break;
    default:
        break;
    }

    return *this;
}

template <typename T>
dds::sub::LoanedSamples<T>
dds::sub::detail::DataReader<T>::Selector::read()
{
    return this->reader->read(*this);
}

template <typename T>
dds::sub::LoanedSamples<T>
dds::sub::detail::DataReader<T>::Selector::take()
{
    return this->reader->take(*this);
}

// --- Forward Iterators: --- //

template <typename T>
template<typename SamplesFWIterator>
uint32_t
dds::sub::detail::DataReader<T>::Selector::read(SamplesFWIterator sfit, uint32_t max_samples)
{
    return this->reader->read(sfit, max_samples, *this);
}

template <typename T>
template<typename SamplesFWIterator>
uint32_t
dds::sub::detail::DataReader<T>::Selector::take(SamplesFWIterator sfit, uint32_t max_samples)
{
    return this->reader->take(sfit, max_samples, *this);
}

// --- Back-Inserting Iterators: --- //

template <typename T>
template<typename SamplesBIIterator>
uint32_t
dds::sub::detail::DataReader<T>::Selector::read(SamplesBIIterator sbit)
{
    return this->reader->read(sbit, *this);
}

template <typename T>
template<typename SamplesBIIterator>
uint32_t
dds::sub::detail::DataReader<T>::Selector::take(SamplesBIIterator sbit)
{
    return this->reader->take(sbit, *this);
}

template <typename T>
typename dds::sub::detail::DataReader<T>::SelectMode
dds::sub::detail::DataReader<T>::Selector::get_mode() const
{
    return this->mode;
}

template <typename T>
dds::sub::detail::DataReader<T>::ManipulatorSelector::ManipulatorSelector(typename DataReader<T>::ref_type dr) :
      Selector(dr), read_mode_(true)
{
}

template <typename T>
bool
dds::sub::detail::DataReader<T>::ManipulatorSelector::read_mode()
{
    return read_mode_;
}

template <typename T>
void
dds::sub::detail::DataReader<T>::ManipulatorSelector::read_mode(bool b)
{
    read_mode_ = b;
}

template <typename T>
typename dds::sub::detail::DataReader<T>::ManipulatorSelector&
dds::sub::detail::DataReader<T>::ManipulatorSelector::operator >>(dds::sub::LoanedSamples<T>& samples)
{
    if(read_mode_)
    {
        samples = this->Selector::read();
    }
    else
    {
        samples = this->Selector::take();
    }
    return *this;
}


template <typename T>
dds::sub::LoanedSamples<T>
dds::sub::detail::DataReader<T>::read(const Selector& selector)
{
    u_query uQuery;

    dds::sub::LoanedSamples<T> samples;
    dds::sub::detail::LoanedSamplesHolder<T> holder(samples);

    switch(selector.mode) {
    case SELECT_MODE_READ:
        this->AnyDataReaderDelegate::read((u_dataReader)(userHandle), selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE:
        this->AnyDataReaderDelegate::read_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->AnyDataReaderDelegate::read_next_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_w_condition(uQuery, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_next_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    }

    return samples;
}

template <typename T>
dds::sub::LoanedSamples<T>
dds::sub::detail::DataReader<T>::take(const Selector& selector)
{
    u_query uQuery;

    dds::sub::LoanedSamples<T> samples;
    dds::sub::detail::LoanedSamplesHolder<T> holder(samples);

    switch(selector.mode) {
    case SELECT_MODE_READ:
        this->AnyDataReaderDelegate::take((u_dataReader)(userHandle), selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE:
        this->AnyDataReaderDelegate::take_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->AnyDataReaderDelegate::take_next_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_w_condition(uQuery, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_next_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    }

    return samples;
}

// --- Forward Iterators: --- //

template <typename T>
template<typename SamplesFWIterator>
uint32_t
dds::sub::detail::DataReader<T>::read(SamplesFWIterator samples,
              uint32_t max_samples, const Selector& selector)
{
    u_query uQuery;

    dds::sub::detail::SamplesFWInteratorHolder<T, SamplesFWIterator> holder(samples);

    switch(selector.mode) {
    case SELECT_MODE_READ:
        this->AnyDataReaderDelegate::read((u_dataReader)(userHandle), selector.state_filter_, holder, max_samples);
        break;
    case SELECT_MODE_READ_INSTANCE:
        this->AnyDataReaderDelegate::read_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, max_samples);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->AnyDataReaderDelegate::read_next_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, max_samples);
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_w_condition(uQuery, holder, max_samples);
        break;
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_instance_w_condition(uQuery, selector.handle, holder, max_samples);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_next_instance_w_condition(uQuery, selector.handle, holder, max_samples);
        break;
    }

    return holder.get_length();
}

template <typename T>
template<typename SamplesFWIterator>
uint32_t
dds::sub::detail::DataReader<T>::take(SamplesFWIterator samples,
              uint32_t max_samples, const Selector& selector)
{
    u_query uQuery;

    dds::sub::detail::SamplesFWInteratorHolder<T, SamplesFWIterator> holder(samples);

    switch(selector.mode) {
    case SELECT_MODE_READ:
        this->AnyDataReaderDelegate::take((u_dataReader)(userHandle), selector.state_filter_, holder, max_samples);
        break;
    case SELECT_MODE_READ_INSTANCE:
        this->AnyDataReaderDelegate::take_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, max_samples);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->AnyDataReaderDelegate::take_next_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, max_samples);
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_w_condition(uQuery, holder, max_samples);
        break;
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_instance_w_condition(uQuery, selector.handle, holder, max_samples);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_next_instance_w_condition(uQuery, selector.handle, holder, max_samples);
        break;
    }

    return holder.get_length();
}

// --- Back-Inserting Iterators: --- //

template <typename T>
template<typename SamplesBIIterator>
uint32_t
dds::sub::detail::DataReader<T>::read(SamplesBIIterator samples, const Selector& selector)
{
    u_query uQuery;

    dds::sub::detail::SamplesBIIteratorHolder<T, SamplesBIIterator> holder(samples);

    switch(selector.mode) {
    case SELECT_MODE_READ:
        this->AnyDataReaderDelegate::read((u_dataReader)(userHandle), selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE:
        this->AnyDataReaderDelegate::read_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->AnyDataReaderDelegate::read_next_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_w_condition(uQuery, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::read_next_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    }

    return holder.get_length();
}

template <typename T>
template<typename SamplesBIIterator>
uint32_t
dds::sub::detail::DataReader<T>::take(SamplesBIIterator samples, const Selector& selector)
{
    u_query uQuery;

    dds::sub::detail::SamplesBIIteratorHolder<T, SamplesBIIterator> holder(samples);

    switch(selector.mode) {
    case SELECT_MODE_READ:
        this->AnyDataReaderDelegate::take((u_dataReader)(userHandle), selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE:
        this->AnyDataReaderDelegate::take_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE:
        this->AnyDataReaderDelegate::take_next_instance((u_dataReader)(userHandle), selector.handle, selector.state_filter_, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_w_condition(uQuery, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    case SELECT_MODE_READ_NEXT_INSTANCE_WITH_CONDITION:
        uQuery = selector.query_.delegate()->get_user_query();
        this->AnyDataReaderDelegate::take_next_instance_w_condition(uQuery, selector.handle, holder, selector.max_samples_);
        break;
    }

    return holder.get_length();
}


namespace dds
{
namespace sub
{

template <typename SELECTOR>
SELECTOR& read(SELECTOR& selector)
{
    selector.read_mode(true);
    return selector;
}

template <typename SELECTOR>
SELECTOR& take(SELECTOR& selector)
{
    selector.read_mode(false);
    return selector;
}

inline dds::sub::functors::MaxSamplesManipulatorFunctor
max_samples(uint32_t n)
{
    return dds::sub::functors::MaxSamplesManipulatorFunctor(n);
}

inline dds::sub::functors::ContentFilterManipulatorFunctor
content(const dds::sub::Query& query)
{
    return dds::sub::functors::ContentFilterManipulatorFunctor(query);
}


inline dds::sub::functors::StateFilterManipulatorFunctor
state(const dds::sub::status::DataState& s)
{
    return dds::sub::functors::StateFilterManipulatorFunctor(s);
}

inline dds::sub::functors::InstanceManipulatorFunctor
instance(const dds::core::InstanceHandle& h)
{
    return dds::sub::functors::InstanceManipulatorFunctor(h);
}

inline dds::sub::functors::NextInstanceManipulatorFunctor
next_instance(const dds::core::InstanceHandle& h)
{
    return dds::sub::functors::NextInstanceManipulatorFunctor(h);
}

}
}

// End of implementation

#endif /* OSPL_DDS_SUB_TDATAREADER_IMPL_HPP_ */