Method end_coherent_changes

dds/DCPS/DataWriterImpl.cpp:2389–2441

}

void
DataWriterImpl::end_coherent_changes(const GroupCoherentSamples& group_samples)
{
  // PublisherImpl::pi_lock_ should be held.
  ACE_GUARD(ACE_Recursive_Thread_Mutex,
            guard,
            get_lock());

  CoherentChangeControl end_msg;
  end_msg.coherent_samples_.num_samples_ = this->coherent_samples_;
  end_msg.coherent_samples_.last_sample_ = get_max_sn();

  RcHandle<PublisherImpl> publisher = this->publisher_servant_.lock();

  if (publisher) {
    end_msg.group_coherent_
      = publisher->qos_.presentation.access_scope == DDS::GROUP_PRESENTATION_QOS;
  }

  if (publisher && end_msg.group_coherent_) {
    end_msg.publisher_id_ = publisher->publisher_id_;
    end_msg.group_coherent_samples_ = group_samples;
  }

  Message_Block_Ptr data(
    new ACE_Message_Block(
      end_msg.get_max_serialized_size(),
      ACE_Message_Block::MB_DATA,
      0, // cont
      0, // data
      0, // alloc_strategy
      get_db_lock()));

  Serializer serializer(data.get(), Encoding::KIND_UNALIGNED_CDR,
    this->swap_bytes());

  serializer << end_msg;

  DataSampleHeader header;
  Message_Block_Ptr control(
    create_control_message(END_COHERENT_CHANGES, header, OPENDDS_MOVE_NS::move(data),
      SystemTimePoint::now().to_idl_struct()));

  this->coherent_ = false;
  this->coherent_samples_ = 0;

  guard.release();
  if (this->send_control(header, OPENDDS_MOVE_NS::move(control)) == SEND_CONTROL_ERROR) {
    ACE_ERROR((LM_ERROR,
               ACE_TEXT("(%P|%t) ERROR: DataWriterImpl::end_coherent_changes:")
               ACE_TEXT(" unable to send END_COHERENT_CHANGES control message!\n")));
  }
}

#endif // OPENDDS_NO_OBJECT_MODEL_PROFILE