OpenDDS::DCPS::RakeResults< MessageType > Class Template Reference

#include <RakeResults_T.h>

Collaboration diagram for OpenDDS::DCPS::RakeResults< MessageType >:

Classes
struct	InstanceData
class	SortedSetCmp

Public Member Functions
	RakeResults (DataReaderImpl *reader, SampleSeq &received_data, DDS::SampleInfoSeq &info_seq, CORBA::Long max_samples, DDS::PresentationQosPolicy presentation, DDS::QueryCondition_ptr cond, Operation_t oper)
bool	insert_sample (ReceivedDataElement *sample, ReceivedDataElementList *rdel, SubscriptionInstance_rch instance, size_t index_in_instance)
bool	copy_to_user ()

Private Types
typedef DDSTraits< MessageType >::MessageSequenceType	SampleSeq
typedef DDSTraits< MessageType >::MessageSequenceAdapterType	MessageSequenceAdapterType

Private Member Functions
template<class FwdIter >
bool	copy_into (FwdIter begin, FwdIter end, MessageSequenceAdapterType &received_data_p)
	RakeResults (const RakeResults &)
RakeResults &	operator= (const RakeResults &)
typedef	OPENDDS_MULTISET_CMP (RakeData, SortedSetCmp) SortedSet
	OPENDDS_VECTOR (RakeData) unsorted_
typedef	OPENDDS_VECTOR (CORBA::ULong) IndexList

Private Attributes
DataReaderImpl *	reader_
SampleSeq &	received_data_
DDS::SampleInfoSeq &	info_seq_
CORBA::ULong	max_samples_
DDS::QueryCondition_ptr	cond_
Operation_t	oper_
bool	do_sort_
bool	do_filter_
SortedSet	sorted_

Detailed Description

template<class MessageType>
class OpenDDS::DCPS::RakeResults< MessageType >

Rake is an abbreviation for "read or take". This class manages the results from a read() or take() operation, which are the received_data and the info_seq sequences passed in by-reference from the user.

Definition at line 34 of file RakeResults_T.h.

Member Typedef Documentation

MessageSequenceAdapterType

template<class MessageType>

typedef DDSTraits<MessageType>::MessageSequenceAdapterType OpenDDS::DCPS::RakeResults< MessageType >::MessageSequenceAdapterType

private

Definition at line 37 of file RakeResults_T.h.

SampleSeq

template<class MessageType>

typedef DDSTraits<MessageType>::MessageSequenceType OpenDDS::DCPS::RakeResults< MessageType >::SampleSeq

private

Definition at line 36 of file RakeResults_T.h.

Constructor & Destructor Documentation

RakeResults() [1/2]

template<class MessageType >

OpenDDS::DCPS::RakeResults< MessageType >::RakeResults	(	DataReaderImpl *	reader,
		SampleSeq &	received_data,
		DDS::SampleInfoSeq &	info_seq,
		CORBA::Long	max_samples,
		DDS::PresentationQosPolicy	presentation,
		DDS::QueryCondition_ptr	cond,
		Operation_t	oper
	)

Definition at line 22 of file RakeResults_T.cpp.

References DDS::PresentationQosPolicy::access_scope, ACE_ERROR, ACE_TEXT(), OpenDDS::DCPS::RakeResults< MessageType >::cond_, OpenDDS::DCPS::RakeResults< MessageType >::do_filter_, OpenDDS::DCPS::RakeResults< MessageType >::do_sort_, OpenDDS::DCPS::QueryConditionImpl::getOrderBys(), OpenDDS::DCPS::QueryConditionImpl::hasFilter(), LM_ERROR, DDS::PresentationQosPolicy::ordered_access, OpenDDS::DCPS::RakeResults< MessageType >::sorted_, and DDS::TOPIC_PRESENTATION_QOS.

  : reader_(reader)
  , received_data_(received_data)
  , info_seq_(info_seq)
  , max_samples_(max_samples)
#ifndef OPENDDS_NO_QUERY_CONDITION
  , cond_(cond)
#endif
  , oper_(oper)
  , do_sort_(false)
  , do_filter_(false)
{
#ifndef OPENDDS_NO_QUERY_CONDITION

  if (cond_) {
    const QueryConditionImpl* qci = dynamic_cast<QueryConditionImpl*>(cond_);
    if (!qci) {
      ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: RakeResults(): ")
        ACE_TEXT("failed to obtain QueryConditionImpl\n")));
      return;
    }
    do_filter_ = qci->hasFilter();
    std::vector<String> order_bys = qci->getOrderBys();
    do_sort_ = order_bys.size() > 0;

    if (do_sort_) {
      ComparatorBase::Ptr cmp;

      // Iterate in reverse over the comma-separated fields so that the
      // top-level comparison is the leftmost.  The others will be chained.
      for (size_t i = order_bys.size(); i > 0; --i) {
        const String& fieldspec = order_bys[i - 1];
        //FUTURE: handle ASC / DESC as an extension to the DDS spec?
        cmp = getMetaStruct<MessageType>().create_qc_comparator(fieldspec.c_str(), cmp);
      }

      SortedSetCmp comparator(cmp);
      SortedSet actual_sort(comparator);
      sorted_.swap(actual_sort);
    }

  } else {
#endif
    // PRESENTATION ordered access (TOPIC)
    do_sort_ = presentation.ordered_access && presentation.access_scope == DDS::TOPIC_PRESENTATION_QOS;
#ifndef OPENDDS_NO_QUERY_CONDITION
  }

#endif
}

RakeResults() [2/2]

template<class MessageType>

OpenDDS::DCPS::RakeResults< MessageType >::RakeResults ( const RakeResults< MessageType > &  )

private

Member Function Documentation

copy_into()

template<class MessageType >

template<class FwdIter >

bool OpenDDS::DCPS::RakeResults< MessageType >::copy_into ( FwdIter  begin, FwdIter  end, MessageSequenceAdapterType &  received_data_p  )

private

Definition at line 121 of file RakeResults_T.cpp.

References DDS::SampleInfo::absolute_generation_rank, OpenDDS::DCPS::InstanceState::accessed(), OpenDDS::DCPS::DDS_OPERATION_TAKE, OpenDDS::DCPS::ReceivedDataElement::dec_ref(), OpenDDS::DCPS::ReceivedDataElement::disposed_generation_count_, DDS::SampleInfo::generation_rank, OpenDDS::DCPS::RakeResults< MessageType >::info_seq_, OpenDDS::DCPS::SubscriptionInstance::instance_state_, OpenDDS::DCPS::ReceivedDataElementList::mark_read(), OpenDDS::DCPS::RakeResults< MessageType >::max_samples_, OpenDDS::DCPS::InstanceState::most_recent_generation(), OpenDDS::DCPS::ReceivedDataElement::no_writers_generation_count_, OpenDDS::DCPS::OPENDDS_MAP(), OpenDDS::DCPS::OPENDDS_SET(), OpenDDS::DCPS::RakeResults< MessageType >::oper_, OpenDDS::DCPS::ReceivedDataElementList::peek_tail(), OpenDDS::DCPS::SubscriptionInstance::rcvd_samples_, OpenDDS::DCPS::RakeResults< MessageType >::received_data_, OpenDDS::DCPS::ReceivedDataElement::registered_data_, OpenDDS::DCPS::ReceivedDataElementList::remove(), OpenDDS::DCPS::InstanceState::sample_info(), and DDS::SampleInfo::sample_rank.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_to_user().

{
  typedef OPENDDS_MAP(SubscriptionInstance*, InstanceData) InstanceMap;
  InstanceMap inst_map;

  typedef OPENDDS_SET(SubscriptionInstance*) InstanceSet;
  InstanceSet released_instances;

  for (CORBA::ULong idx = 0; iter != end && idx < max_samples_; ++idx, ++iter) {
    // 1. Populate the Received Data sequence
    ReceivedDataElement* rde = iter->rde_;
    ReceivedDataElementList* rdel = iter->rdel_;

    if (received_data_.maximum() != 0) {
      if (rde->registered_data_ == 0) {
        received_data_p.assign_sample(idx, MessageType());

      } else {
        received_data_p.assign_sample(idx,
                                      *static_cast<MessageType*>(rde->registered_data_));
      }

    } else {
      received_data_p.assign_ptr(idx, rde);
    }

    // 2. Per-sample SampleInfo (not the three *_rank variables) and state
    SubscriptionInstance& inst = *iter->si_;
    inst.instance_state_->sample_info(info_seq_[idx], rde);
    rdel->mark_read(rde);

    // 3. Record some info about per-instance SampleInfo (*_rank) so that
    //    we can fill in the ranks after the loop has completed
    std::pair<typename InstanceMap::iterator, bool> result =
      inst_map.insert(std::make_pair(&inst, InstanceData()));
    InstanceData& id = result.first->second;

    if (result.second) { // first time we've seen this Instance
      const ReceivedDataElement& mrs = *inst.rcvd_samples_.peek_tail();
      id.MRS_disposed_gc_ =
        static_cast<CORBA::Long>(mrs.disposed_generation_count_);
      id.MRS_nowriters_gc_ =
        static_cast<CORBA::Long>(mrs.no_writers_generation_count_);
    }

    if (iter->index_in_instance_ >= id.MRSIC_index_) {
      id.MRSIC_index_ = iter->index_in_instance_;
      id.MRSIC_disposed_gc_ =
        static_cast<CORBA::Long>(rde->disposed_generation_count_);
      id.MRSIC_nowriters_gc_ =
        static_cast<CORBA::Long>(rde->no_writers_generation_count_);
    }

    if (!id.most_recent_generation_) {
      id.most_recent_generation_ =
        inst.instance_state_->most_recent_generation(rde);
    }

    id.sampleinfo_positions_.push_back(idx);

    // 4. Take
    if (oper_ == DDS_OPERATION_TAKE) {
      // If removing the sample releases it
      if (inst.rcvd_samples_.remove(rde)) {
        // Prevent access of the SampleInfo, below
        released_instances.insert(&inst);
      }
      rde->dec_ref();
    }
  }

  // Fill in the *_ranks in the SampleInfo, and set instance state (mrg)
  for (typename InstanceMap::iterator i_iter(inst_map.begin()),
       i_end(inst_map.end()); i_iter != i_end; ++i_iter) {

    InstanceData& id = i_iter->second;
    {  // Danger, limit visibility of inst
      SubscriptionInstance& inst = *i_iter->first;
      // If this instance has not been released
      if (released_instances.find(&inst) == released_instances.end()) {
        if (id.most_recent_generation_) {
          inst.instance_state_->accessed();
        }
      }
    }

    CORBA::Long sample_rank =
      static_cast<CORBA::Long>(id.sampleinfo_positions_.size());

    for (IndexList::iterator s_iter(id.sampleinfo_positions_.begin()),
         s_end(id.sampleinfo_positions_.end()); s_iter != s_end; ++s_iter) {
      DDS::SampleInfo& si = info_seq_[*s_iter];
      si.sample_rank = --sample_rank;
      si.generation_rank = id.MRSIC_disposed_gc_
                           + id.MRSIC_nowriters_gc_ - si.generation_rank;
      si.absolute_generation_rank = id.MRS_disposed_gc_ +
                                    id.MRS_nowriters_gc_ - si.absolute_generation_rank;
    }
  }

  return true;
}

copy_to_user()

template<class MessageType >

bool OpenDDS::DCPS::RakeResults< MessageType >::copy_to_user

(

)

Definition at line 226 of file RakeResults_T.cpp.

References OpenDDS::DCPS::RakeResults< MessageType >::copy_into(), OpenDDS::DCPS::RakeResults< MessageType >::do_sort_, OpenDDS::DCPS::RakeResults< MessageType >::info_seq_, OpenDDS::DCPS::RakeResults< MessageType >::max_samples_, OPENDDS_END_VERSIONED_NAMESPACE_DECL, OpenDDS::DCPS::RakeResults< MessageType >::received_data_, and OpenDDS::DCPS::RakeResults< MessageType >::sorted_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_i(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_instance_i(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_i(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_instance_i().

{
  MessageSequenceAdapterType received_data_p(received_data_);

  if (do_sort_) {
    size_t len = std::min(static_cast<size_t>(sorted_.size()),
                          static_cast<size_t>(max_samples_));
    received_data_p.internal_set_length(static_cast<CORBA::ULong>(len));
    info_seq_.length(static_cast<CORBA::ULong>(len));
    return copy_into(sorted_.begin(), sorted_.end(), received_data_p);

  } else {
    size_t len = unsorted_.size(); //can't be larger than max_samples_
    received_data_p.internal_set_length(static_cast<CORBA::ULong>(len));
    info_seq_.length(static_cast<CORBA::ULong>(len));
    return copy_into(unsorted_.begin(), unsorted_.end(), received_data_p);
  }
}

insert_sample()

template<class MessageType >

bool OpenDDS::DCPS::RakeResults< MessageType >::insert_sample	(	ReceivedDataElement *	sample,
		ReceivedDataElementList *	rdel,
		SubscriptionInstance_rch	instance,
		size_t	index_in_instance
	)

Returns false if the sample will definitely not be part of the resulting dataset, however if this returns true it still may be excluded (due to sorting and max_samples).

Definition at line 82 of file RakeResults_T.cpp.

References OpenDDS::DCPS::RakeResults< MessageType >::cond_, OpenDDS::DCPS::RakeResults< MessageType >::do_filter_, OpenDDS::DCPS::RakeResults< MessageType >::do_sort_, OpenDDS::DCPS::QueryConditionImpl::filter(), OpenDDS::DCPS::RakeResults< MessageType >::max_samples_, OpenDDS::DCPS::ReceivedDataElement::registered_data_, OpenDDS::DCPS::RakeResults< MessageType >::sorted_, and OpenDDS::DCPS::ReceivedDataElement::valid_data_.

Referenced by OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_i(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::read_instance_i(), OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_i(), and OpenDDS::DCPS::DataReaderImpl_T< DynamicSample >::take_instance_i().

{
#ifndef OPENDDS_NO_QUERY_CONDITION

  if (do_filter_) {
    const QueryConditionImpl* qci = dynamic_cast<QueryConditionImpl*>(cond_);
    const MessageType* typed_sample = static_cast<MessageType*>(sample->registered_data_);
    if (!qci || !typed_sample || !qci->filter(*typed_sample, !sample->valid_data_)) {
      return false;
    }
  }

#endif

  if (do_sort_) {
    // N.B. Until a better heuristic is found, non-valid
    // samples are elided when sorting by QueryCondition.
#ifndef OPENDDS_NO_QUERY_CONDITION
    if (cond_ && !sample->registered_data_) return false;
#endif

    RakeData rd = {sample, rdel, instance, index_in_instance};
    sorted_.insert(rd);

  } else {
    if (unsorted_.size() == max_samples_) return false;

    RakeData rd = {sample, rdel, instance, index_in_instance};
    unsorted_.push_back(rd);
  }

  return true;
}

OPENDDS_MULTISET_CMP()

template<class MessageType>

typedef OpenDDS::DCPS::RakeResults< MessageType >::OPENDDS_MULTISET_CMP ( RakeData  , SortedSetCmp    )

private

OPENDDS_VECTOR() [1/2]

template<class MessageType>

OpenDDS::DCPS::RakeResults< MessageType >::OPENDDS_VECTOR ( RakeData  )

private

OPENDDS_VECTOR() [2/2]

template<class MessageType>

typedef OpenDDS::DCPS::RakeResults< MessageType >::OPENDDS_VECTOR ( CORBA::ULong  )

private

operator=()

template<class MessageType>

RakeResults& OpenDDS::DCPS::RakeResults< MessageType >::operator= ( const RakeResults< MessageType > &  )

private

Member Data Documentation

cond_

template<class MessageType>

DDS::QueryCondition_ptr OpenDDS::DCPS::RakeResults< MessageType >::cond_

private

Definition at line 73 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::insert_sample(), and OpenDDS::DCPS::RakeResults< MessageType >::RakeResults().

do_filter_

template<class MessageType>

bool OpenDDS::DCPS::RakeResults< MessageType >::do_filter_

private

Definition at line 97 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::insert_sample(), and OpenDDS::DCPS::RakeResults< MessageType >::RakeResults().

do_sort_

template<class MessageType>

bool OpenDDS::DCPS::RakeResults< MessageType >::do_sort_

private

Definition at line 97 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_to_user(), OpenDDS::DCPS::RakeResults< MessageType >::insert_sample(), and OpenDDS::DCPS::RakeResults< MessageType >::RakeResults().

info_seq_

template<class MessageType>

DDS::SampleInfoSeq& OpenDDS::DCPS::RakeResults< MessageType >::info_seq_

private

Definition at line 70 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_into(), and OpenDDS::DCPS::RakeResults< MessageType >::copy_to_user().

max_samples_

template<class MessageType>

CORBA::ULong OpenDDS::DCPS::RakeResults< MessageType >::max_samples_

private

Definition at line 71 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_into(), OpenDDS::DCPS::RakeResults< MessageType >::copy_to_user(), and OpenDDS::DCPS::RakeResults< MessageType >::insert_sample().

oper_

template<class MessageType>

Operation_t OpenDDS::DCPS::RakeResults< MessageType >::oper_

private

Definition at line 75 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_into().

reader_

template<class MessageType>

DataReaderImpl* OpenDDS::DCPS::RakeResults< MessageType >::reader_

private

Definition at line 68 of file RakeResults_T.h.

received_data_

template<class MessageType>

SampleSeq& OpenDDS::DCPS::RakeResults< MessageType >::received_data_

private

Definition at line 69 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_into(), and OpenDDS::DCPS::RakeResults< MessageType >::copy_to_user().

sorted_

template<class MessageType>

SortedSet OpenDDS::DCPS::RakeResults< MessageType >::sorted_

private

Definition at line 101 of file RakeResults_T.h.

Referenced by OpenDDS::DCPS::RakeResults< MessageType >::copy_to_user(), OpenDDS::DCPS::RakeResults< MessageType >::insert_sample(), and OpenDDS::DCPS::RakeResults< MessageType >::RakeResults().

---

The documentation for this class was generated from the following files:

• RakeResults_T.h
• RakeResults_T.cpp