Method coherent_changes_completed

dds/DCPS/DataReaderImpl.cpp:2852–2907

void
DataReaderImpl::coherent_changes_completed(DataReaderImpl* reader)
{
  RcHandle<SubscriberImpl> subscriber = get_subscriber_servant();
  if (!subscriber) {
    return;
  }

  subscriber->set_status_changed_flag(::DDS::DATA_ON_READERS_STATUS, true);
  this->set_status_changed_flag(::DDS::DATA_AVAILABLE_STATUS, true);

  ::DDS::SubscriberListener_var sub_listener =
      subscriber->listener_for(::DDS::DATA_ON_READERS_STATUS);
  if (!CORBA::is_nil(sub_listener.in()))
  {
    if (!is_bit()) {
      this->set_status_changed_flag(::DDS::DATA_AVAILABLE_STATUS, false);
      subscriber->set_status_changed_flag(::DDS::DATA_ON_READERS_STATUS, false);
      if (reader == this) {
        // Release the sample_lock before listener callback.
        ACE_GUARD(Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
        sub_listener->on_data_on_readers(subscriber.in());
      }
    } else {
      TheServiceParticipant->job_queue()->enqueue(make_rch<OnDataOnReaders>(subscriber, sub_listener, rchandle_from(this), reader == this, true));
    }
  }
  else
  {
    subscriber->notify_status_condition();

    ::DDS::DataReaderListener_var listener =
        this->listener_for (::DDS::DATA_AVAILABLE_STATUS);

    if (!CORBA::is_nil(listener.in()))
    {
      if (!is_bit()) {
        set_status_changed_flag(::DDS::DATA_AVAILABLE_STATUS, false);
        subscriber->set_status_changed_flag(::DDS::DATA_ON_READERS_STATUS, false);
        if (reader == this) {
          // Release the sample_lock before listener callback.
          ACE_GUARD(Reverse_Lock_t, unlock_guard, reverse_sample_lock_);
          listener->on_data_available(this);
        } else {
          listener->on_data_available(this);
        }
      } else {
        TheServiceParticipant->job_queue()->enqueue(make_rch<OnDataAvailable>(listener, rchandle_from(this), reader == this, true, true));
      }
    }
    else
    {
      this->notify_status_condition();
    }
  }
}