Method data_received

dds/DCPS/DataReaderImpl.cpp:1343–1632

}

void
DataReaderImpl::data_received(const ReceivedDataSample& sample)
{
  DBG_ENTRY_LVL("DataReaderImpl","data_received",6);

  DDS::InstanceHandle_t publication_handle = DDS::HANDLE_NIL;
  {
    ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, publication_handle_lock_);
    RepoIdToHandleMap::const_iterator pos = publication_id_to_handle_map_.find(sample.header_.publication_id_);
    if (pos != publication_id_to_handle_map_.end()) {
      publication_handle = pos->second;
    }
  }

  // ensure some other thread is not changing the sample container
  // or statuses related to samples.
  ACE_GUARD(ACE_Recursive_Thread_Mutex, guard, this->sample_lock_);

  if (get_deleted()) return;

  if (DCPS_debug_level > 9) {
    ACE_DEBUG((LM_DEBUG,
        ACE_TEXT("(%P|%t) DataReaderImpl::data_received: ")
        ACE_TEXT("%C received sample: %C.\n"),
        LogGuid(get_guid()).c_str(),
        to_string(sample.header_).c_str()));
  }

  switch (sample.header_.message_id_) {
  case SAMPLE_DATA:
  case INSTANCE_REGISTRATION: {
    SubscriptionInstance_rch instance;
    if (!check_historic(sample)) break;

    DataSampleHeader const & header = sample.header_;

    this->writer_activity(header);

    // Verify data has not exceeded its lifespan.
    if (this->filter_sample(header)) break;

    // This adds the reader to the set/list of readers with data.
    RcHandle<SubscriberImpl> subscriber = get_subscriber_servant();
    if (subscriber) {
      subscriber->data_received(this);
    }

    // Only gather statistics about real samples, not registration data, etc.
    if (header.message_id_ == SAMPLE_DATA) {
      this->process_latency(sample);
    }

    // This also adds to the sample container and makes any callbacks
    // and condition modifications.

    bool is_new_instance = false;
    bool filtered = false;
    dds_demarshal(sample, publication_handle, instance, is_new_instance, filtered,