Method init

dds/DCPS/MultiTopicDataReaderBase.cpp:91–219

namespace DCPS {

void MultiTopicDataReaderBase::init(const DDS::DataReaderQos& dr_qos,
  DDS::DataReaderListener_ptr a_listener, DDS::StatusMask mask,
  SubscriberImpl* parent, MultiTopicImpl* multitopic)
{
  using namespace std;
  DDS::DataReader_var dr = multitopic->get_type_support()->create_datareader();
  resulting_reader_ = DataReaderEx::_narrow(dr);
  DataReaderImpl* resulting_impl =
    dynamic_cast<DataReaderImpl*>(resulting_reader_.in());

  if (!resulting_impl) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: MultiTopicDataReaderBase::init: ")
      ACE_TEXT("Failed to get DataReaderImpl.\n")));
    return;
  }

  resulting_impl->enable_multi_topic(multitopic);
  resulting_impl->raw_latency_buffer_size() = parent->raw_latency_buffer_size();
  resulting_impl->raw_latency_buffer_type() = parent->raw_latency_buffer_type();

  DDS::DomainParticipant_var participant = parent->get_participant();
  DomainParticipantImpl* dpi = dynamic_cast<DomainParticipantImpl*>(participant.in());
  if (!dpi) {
    ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: MultiTopicDataReaderBase::init: ")
      ACE_TEXT("Failed to get DomainParticipantImpl.\n")));
    return;
  }
  resulting_impl->init(multitopic, dr_qos, a_listener, mask, dpi, parent);

  init_typed(resulting_reader_);

  std::map<OPENDDS_STRING, OPENDDS_STRING> fieldToTopic;

  // key: name of field that's a key for the 'join'
  // mapped: set of topicNames that have this key in common
  std::map<OPENDDS_STRING, set<OPENDDS_STRING> > joinKeys;

  listener_.reset(new Listener(this));

  const vector<OPENDDS_STRING>& selection = multitopic->get_selection();
  for (size_t i = 0; i < selection.size(); ++i) {

    const DDS::Duration_t no_wait = {0, 0};
    DDS::Topic_var t = participant->find_topic(selection[i].c_str(), no_wait);
    if (!t.in()) {
      throw runtime_error("Topic: " + selection[i] + " not found.");
    }


    QueryPlan& qp = query_plans_[selection[i]];
    {
      ACE_WRITE_GUARD(ACE_RW_Thread_Mutex, write_guard, qp_lock_);
      qp.data_reader_ =
        parent->create_datareader(t, DATAREADER_QOS_USE_TOPIC_QOS,
                                  listener_.get(), ALL_STATUS_MASK);
    }
    if (!qp.data_reader_.in()) {
      throw runtime_error("Could not create incoming DataReader "