OpenDDS::FaceTSS Namespace Reference

Classes
class	Entities
class	Listener
Namespaces
namespace	config
Enumerations
enum	{ NSEC_PER_SEC = 1000000000 }
Functions
FACE::RETURN_CODE_TYPE	update_status (FACE::CONNECTION_ID_TYPE connection_id, DDS::ReturnCode_t retcode)
DDS::Duration_t	convertTimeout (FACE::TIMEOUT_TYPE timeout)
FACE::SYSTEM_TIME_TYPE	convertDuration (const DDS::Duration_t &duration)
FACE::SYSTEM_TIME_TYPE	convertTime (const DDS::Time_t &timestamp)
FACE::MESSAGE_INSTANCE_GUID	create_message_instance_guid (const OpenDDS::DCPS::RepoId &pub, const CORBA::LongLong &orig_seq)
void	populate_header_received (const FACE::CONNECTION_ID_TYPE &connection_id, const DDS::DomainParticipant_var part, const DDS::SampleInfo &sinfo, FACE::RETURN_CODE_TYPE &return_code)
template<typename Msg>
void	receive_message (FACE::CONNECTION_ID_TYPE connection_id, FACE::TIMEOUT_TYPE timeout, FACE::TRANSACTION_ID_TYPE &transaction_id, Msg &message, FACE::MESSAGE_SIZE_TYPE message_size, FACE::RETURN_CODE_TYPE &return_code)
template<typename Msg>
void	send_message (FACE::CONNECTION_ID_TYPE connection_id, FACE::TIMEOUT_TYPE timeout, FACE::TRANSACTION_ID_TYPE &, const Msg &message, FACE::MESSAGE_SIZE_TYPE message_size, FACE::RETURN_CODE_TYPE &return_code)
template<typename Msg>
void	register_callback (FACE::CONNECTION_ID_TYPE connection_id, const FACE::WAITSET_TYPE, void(*callback)(FACE::TRANSACTION_ID_TYPE, Msg &, FACE::MESSAGE_TYPE_GUID, FACE::MESSAGE_SIZE_TYPE, const FACE::WAITSET_TYPE, FACE::RETURN_CODE_TYPE &), FACE::MESSAGE_SIZE_TYPE max_message_size, FACE::RETURN_CODE_TYPE &return_code)

---

Enumeration Type Documentation

anonymous enum

Enumerator:

NSEC_PER_SEC

Definition at line 711 of file FaceTSS.cpp.

{ NSEC_PER_SEC = 1000000000 };

---

Function Documentation

OpenDDS_FACE_Export FACE::RETURN_CODE_TYPE OpenDDS::FaceTSS::update_status	(	FACE::CONNECTION_ID_TYPE	connection_id,
		DDS::ReturnCode_t	retcode
	)

Definition at line 664 of file FaceTSS.cpp.

References OpenDDS::FaceTSS::Entities::instance(), DDS::RETCODE_ALREADY_DELETED, DDS::RETCODE_BAD_PARAMETER, DDS::RETCODE_ERROR, DDS::RETCODE_ILLEGAL_OPERATION, DDS::RETCODE_IMMUTABLE_POLICY, DDS::RETCODE_INCONSISTENT_POLICY, DDS::RETCODE_NO_DATA, DDS::RETCODE_NOT_ENABLED, DDS::RETCODE_OK, DDS::RETCODE_OUT_OF_RESOURCES, DDS::RETCODE_PRECONDITION_NOT_MET, DDS::RETCODE_TIMEOUT, and DDS::RETCODE_UNSUPPORTED.

Referenced by OpenDDS::FaceTSS::Listener< Msg >::on_data_available(), FACE::TS::receive_header(), receive_message(), and send_message().

{
  FACE::TRANSPORT_CONNECTION_STATUS_TYPE& status =
    Entities::instance()->connections_[connection_id].connection_status;
  FACE::RETURN_CODE_TYPE rc = FACE::INVALID_PARAM;

  switch (retcode) {
  case DDS::RETCODE_OK:
    status.LAST_MSG_VALIDITY = FACE::VALID;
    return FACE::RC_NO_ERROR;

  case DDS::RETCODE_ERROR:
    rc = FACE::CONNECTION_CLOSED; break;

  case DDS::RETCODE_BAD_PARAMETER:
    rc = FACE::INVALID_PARAM; break;

  case DDS::RETCODE_OUT_OF_RESOURCES:
    rc = FACE::DATA_BUFFER_TOO_SMALL; break;

  case DDS::RETCODE_PRECONDITION_NOT_MET:
  case DDS::RETCODE_NOT_ENABLED:
    rc = FACE::INVALID_MODE; break;

  case DDS::RETCODE_IMMUTABLE_POLICY:
  case DDS::RETCODE_INCONSISTENT_POLICY:
    rc = FACE::INVALID_CONFIG; break;

  case DDS::RETCODE_ALREADY_DELETED:
    rc = FACE::CONNECTION_CLOSED; break;

  case DDS::RETCODE_TIMEOUT:
    rc = FACE::TIMED_OUT; break;

  case DDS::RETCODE_UNSUPPORTED:
  case DDS::RETCODE_NO_DATA:
    rc = FACE::NOT_AVAILABLE; break;

  case DDS::RETCODE_ILLEGAL_OPERATION:
    rc = FACE::PERMISSION_DENIED; break;
  }

  status.LAST_MSG_VALIDITY = FACE::INVALID;
  return rc;
}

OpenDDS_FACE_Export DDS::Duration_t OpenDDS::FaceTSS::convertTimeout

(

FACE::TIMEOUT_TYPE

timeout

)

Definition at line 713 of file FaceTSS.cpp.

References DDS::DURATION_INFINITE_NSEC, DDS::DURATION_INFINITE_SEC, and NSEC_PER_SEC.

Referenced by receive_message().

{
  if (timeout == FACE::INF_TIME_VALUE) {
    static const DDS::Duration_t dds_inf = {DDS::DURATION_INFINITE_SEC,
                                            DDS::DURATION_INFINITE_NSEC};
    return dds_inf;
  }

  DDS::Duration_t dur = {static_cast<int>(timeout / NSEC_PER_SEC),
                         static_cast<unsigned int>(timeout % NSEC_PER_SEC)};
  return dur;
}

OpenDDS_FACE_Export FACE::SYSTEM_TIME_TYPE OpenDDS::FaceTSS::convertDuration

(

const DDS::Duration_t &

duration

)

Definition at line 726 of file FaceTSS.cpp.

References DDS::DURATION_INFINITE_NSEC, DDS::DURATION_INFINITE_SEC, DDS::Duration_t::nanosec, NSEC_PER_SEC, and DDS::Duration_t::sec.

Referenced by FACE::TS::Create_Connection(), and send_message().

{
  if (duration.sec == DDS::DURATION_INFINITE_SEC
      && duration.nanosec == DDS::DURATION_INFINITE_NSEC) {
    return FACE::INF_TIME_VALUE;
  }
  return duration.nanosec +
    duration.sec * static_cast<FACE::SYSTEM_TIME_TYPE>(NSEC_PER_SEC);
}

OpenDDS_FACE_Export FACE::SYSTEM_TIME_TYPE OpenDDS::FaceTSS::convertTime

(

const DDS::Time_t &

timestamp

)

Definition at line 736 of file FaceTSS.cpp.

References DDS::Time_t::nanosec, NSEC_PER_SEC, and DDS::Time_t::sec.

Referenced by populate_header_received().

{
  return timestamp.nanosec +
      timestamp.sec * static_cast<FACE::SYSTEM_TIME_TYPE>(NSEC_PER_SEC);
}

OpenDDS_FACE_Export FACE::MESSAGE_INSTANCE_GUID OpenDDS::FaceTSS::create_message_instance_guid	(	const OpenDDS::DCPS::RepoId &	pub,
		const CORBA::LongLong &	orig_seq
	)

Definition at line 743 of file FaceTSS.cpp.

Referenced by populate_header_received().

{
  OpenDDS::DCPS::GuidConverter writer(pub);

  FACE::MESSAGE_INSTANCE_GUID message_instance_guid;
  FACE::LongLong mig_low;
  FACE::LongLong masked_seq;

  //Until MESSAGE_INSTANCE_GUID becomes 128 bit GUID, use checksum to represent Prefix
  FACE::Long prefix_representation = ACE::crc32(reinterpret_cast<const void*>(&pub), sizeof(pub));
  masked_seq = orig_seq >> 32;

  if (masked_seq) {
    ACE_DEBUG((LM_WARNING, "(%P|%t) WARNING: create_message_instance_guid - seq does not fit in FACE::Long, truncating high bits to fit\n"));
  }
  mig_low = orig_seq & 0xFFFFFFFF;
  message_instance_guid = (((FACE::LongLong) prefix_representation) << 32 ) | ((FACE::LongLong) mig_low);
/*
  //TODO: This is initial work toward defining how a 128 bit guid could be created
  //in the future for a Message Instance Guid when supported.
  // 13 byte prefix contains identifying pieces of guid
  // 3 byte seq - truncated sequence from the sample
  typedef CORBA::Octet MsgInstGuidPrefix_t[13];
  typedef CORBA::Octet MsgInstGuidSeq_t[3];
  MsgInstGuidPrefix_t migPrefix;
  MsgInstGuidSeq_t migSeq;
  ACE_OS::memcpy(&migPrefix[0], &pub.guidPrefix[2], 10);
  ACE_OS::memcpy(&migPrefix[10], &pub.entityId.entityKey[0], 3);
  masked_seq = orig_seq >> 24;

  if (masked_seq) {
    ACE_DEBUG((LM_WARNING, "(%P|%t) WARNING: create_message_instance_guid - seq does not fit in 3 bytes, truncating high bits to fit\n"));
  }
  FACE::LongLong masked = orig_seq & 0xFFFFFF;

#ifdef ACE_BIG_ENDIAN
  masked <<= 8 * (sizeof(FACE::LongLong)-3);
#endif
  ACE_OS::memcpy(&migSeq[0], &masked, 3);
*/
  return message_instance_guid;
}

OpenDDS_FACE_Export void OpenDDS::FaceTSS::populate_header_received	(	const FACE::CONNECTION_ID_TYPE &	connection_id,
		const DDS::DomainParticipant_var	part,
		const DDS::SampleInfo &	sinfo,
		FACE::RETURN_CODE_TYPE &	return_code
	)

Definition at line 786 of file FaceTSS.cpp.

References DDS::ALIVE_INSTANCE_STATE, DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, OpenDDS::DCPS::BUILT_IN_PUBLICATION_TOPIC, convertTime(), create_message_instance_guid(), OpenDDS::DCPS::DCPS_debug_level, DDS::DURATION_INFINITE_NSEC, DDS::DURATION_INFINITE_SEC, header, OpenDDS::FaceTSS::Entities::instance(), DDS::Duration_t::nanosec, DDS::SampleInfo::opendds_reserved_publication_seq, DDS::SampleInfo::publication_handle, OpenDDS::FaceTSS::Entities::receivers_, DDS::RETCODE_NO_DATA, DDS::RETCODE_OK, DDS::Duration_t::sec, DDS::SampleInfo::source_timestamp, OpenDDS::DCPS::time_to_time_value(), and OpenDDS::DCPS::time_value_to_time().

Referenced by OpenDDS::FaceTSS::Listener< Msg >::on_data_available(), and receive_message().

{
  Entities::ConnIdToReceiverMap& readers = Entities::instance()->receivers_;
  if (!readers.count(connection_id)) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  FACE::TS::MessageHeader& header = readers[connection_id]->last_msg_header;

  header.platform_view_guid = Entities::instance()->connections_[connection_id].platform_view_guid;

  DDS::Subscriber_var temp_sub = readers[connection_id]->dr->get_subscriber();
  DDS::DomainParticipant_var temp_dp = temp_sub->get_participant();
  const OpenDDS::DCPS::RepoId pub = dynamic_cast<OpenDDS::DCPS::DomainParticipantImpl*>(temp_dp.in())->get_repoid(sinfo.publication_handle);
  header.message_instance_guid = create_message_instance_guid(pub, sinfo.opendds_reserved_publication_seq);

  header.message_timestamp = convertTime(sinfo.source_timestamp);
  ACE_Time_Value now(ACE_OS::gettimeofday());

  readers[connection_id]->sum_recvd_msgs_latency += (convertTime(OpenDDS::DCPS::time_value_to_time(now)) - header.message_timestamp);
  ++readers[connection_id]->total_msgs_recvd;

  if (OpenDDS::DCPS::DCPS_debug_level > 8) {
    ACE_DEBUG((LM_DEBUG, "(%P|%t) populate_header_received: Latency is now (tot_latency %d / tot_msgs_recvd %d): %d\n",
        readers[connection_id]->sum_recvd_msgs_latency,
        readers[connection_id]->total_msgs_recvd,
        readers[connection_id]->sum_recvd_msgs_latency/readers[connection_id]->total_msgs_recvd));
  }
  ::DDS::Subscriber_var bit_subscriber
   = part->get_builtin_subscriber () ;

  ::DDS::DataReader_var reader
   = bit_subscriber->lookup_datareader (OpenDDS::DCPS::BUILT_IN_PUBLICATION_TOPIC) ;
  ::DDS::PublicationBuiltinTopicDataDataReader_var pub_reader
   = ::DDS::PublicationBuiltinTopicDataDataReader::_narrow (reader.in ());
  if (CORBA::is_nil (pub_reader.in ())) {
    ACE_ERROR((LM_ERROR, "(%P|%t) populate_header_received: failed to get BUILT_IN_PUBLICATION_TOPIC datareader.\n"));
    return_code = FACE::NOT_AVAILABLE;
    return;
  }

  ::DDS::ReturnCode_t ret;
  ::DDS::SampleInfoSeq pubinfos(1);
  ::DDS::PublicationBuiltinTopicDataSeq pubdata(1);
  ret = pub_reader->read_instance(pubdata,
                                  pubinfos,
                                  1,
                                  sinfo.publication_handle,
                                  ::DDS::ANY_SAMPLE_STATE,
                                  ::DDS::ANY_VIEW_STATE,
                                  ::DDS::ALIVE_INSTANCE_STATE);

  if (ret != ::DDS::RETCODE_OK && ret != ::DDS::RETCODE_NO_DATA) {
    ACE_ERROR((LM_ERROR,
        "(%P|%t) populate_header_received:  failed to read BIT publication data.\n"));
    return_code = FACE::NOT_AVAILABLE;
    return;
  }

  const CORBA::ULong i = 0;

  header.message_source_guid =
    (pubdata[i].user_data.value[0] << 0) |
    (pubdata[i].user_data.value[1] << 8) |
    (pubdata[i].user_data.value[2] << 16);

//  ACE_DEBUG((LM_DEBUG, "(%P|%t) populate_header_received: DW lifespan qos value: sec: %d nanosec: %d\n",
//                         pubdata[i].lifespan.duration.sec, pubdata[i].lifespan.duration.nanosec));

  DDS::Duration_t lifespan = pubdata[i].lifespan.duration;
  if (lifespan.sec != DDS::DURATION_INFINITE_SEC &&
      lifespan.nanosec != DDS::DURATION_INFINITE_NSEC) {
    // Finite lifespan.  Check if data has expired.

    DDS::Time_t const tmp = {
      sinfo.source_timestamp.sec + lifespan.sec,
      sinfo.source_timestamp.nanosec + lifespan.nanosec
    };

    // We assume that the publisher host's clock and subscriber host's
    // clock are synchronized (allowed by the spec).
    ACE_Time_Value const expiration_time(
        OpenDDS::DCPS::time_to_time_value(tmp));

    if (now >= expiration_time) {
//      ACE_DEBUG((LM_DEBUG, "(%P|%t) populate_header_received: Last message expired, setting message_validity to INVALID\n"));
      header.message_validity = FACE::INVALID;
      return_code = FACE::RC_NO_ERROR;
      return;
    }
  }
//  ACE_DEBUG((LM_DEBUG, "(%P|%t) populate_header_received: Setting message_validity to VALID\n"));
  header.message_validity = FACE::VALID;
  return_code = FACE::RC_NO_ERROR;
}

template<typename Msg>

void OpenDDS::FaceTSS::receive_message	(	FACE::CONNECTION_ID_TYPE	connection_id,
		FACE::TIMEOUT_TYPE	timeout,
		FACE::TRANSACTION_ID_TYPE &	transaction_id,
		Msg &	message,
		FACE::MESSAGE_SIZE_TYPE	message_size,
		FACE::RETURN_CODE_TYPE &	return_code
	)

Definition at line 143 of file FaceTSS.h.

References DDS::ALIVE_INSTANCE_STATE, DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, convertTimeout(), DataReader, OpenDDS::FaceTSS::Entities::instance(), populate_header_received(), OpenDDS::FaceTSS::Entities::receivers_, DDS::RETCODE_BAD_PARAMETER, DDS::RETCODE_NO_DATA, DDS::RETCODE_OK, DDS::RETCODE_TIMEOUT, OpenDDS::FaceTSS::Entities::DDSAdapter::status_valid, and update_status().

{
  Entities::ConnIdToReceiverMap& readers = Entities::instance()->receivers_;
  if (!readers.count(connection_id)) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  if(!Entities::instance()->connections_.count(connection_id)) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  FACE::TRANSPORT_CONNECTION_STATUS_TYPE status =
    Entities::instance()->connections_[connection_id].connection_status;
  if (message_size < status.MAX_MESSAGE_SIZE) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  typedef typename DCPS::DDSTraits<Msg>::DataReaderType DataReader;
  const typename DataReader::_var_type typedReader =
    DataReader::_narrow(readers[connection_id]->dr);
  if (!typedReader) {
    return_code = update_status(connection_id, DDS::RETCODE_BAD_PARAMETER);
    return;
  }
  if (readers[connection_id]->status_valid != FACE::VALID) {
    Entities::FaceReceiver* tmp = readers[connection_id];
    readers[connection_id] = new Entities::DDSTypedAdapter<Msg>(*readers[connection_id]);
    delete tmp;
  }
  readers[connection_id]->status_valid = FACE::VALID;

  const DDS::ReadCondition_var rc =
    typedReader->create_readcondition(DDS::ANY_SAMPLE_STATE,
                                      DDS::ANY_VIEW_STATE,
                                      DDS::ALIVE_INSTANCE_STATE);
  const DDS::WaitSet_var ws = new DDS::WaitSet;
  ws->attach_condition(rc);

  DDS::ConditionSeq active;
  const DDS::Duration_t ddsTimeout = convertTimeout(timeout);
  DDS::ReturnCode_t ret = ws->wait(active, ddsTimeout);
  ws->detach_condition(rc);

  if (ret == DDS::RETCODE_TIMEOUT) {
    return_code = update_status(connection_id, ret);
    return;
  }

  typename DCPS::DDSTraits<Msg>::MessageSequenceType seq;
  DDS::SampleInfoSeq sinfo;
  ret = typedReader->take_w_condition(seq, sinfo, 1 /*max*/, rc);
  if (ret == DDS::RETCODE_OK && sinfo[0].valid_data) {
    DDS::DomainParticipant_var participant = typedReader->get_subscriber()->get_participant();
    FACE::RETURN_CODE_TYPE ret_code;
    populate_header_received(connection_id, participant, sinfo[0], ret_code);
    if (ret_code != FACE::RC_NO_ERROR) {
      return_code = update_status(connection_id, ret_code);
      return;
    }

    transaction_id = ++readers[connection_id]->last_msg_tid;

    message = seq[0];
    return_code = update_status(connection_id, ret);
    return;
  }
  return_code = update_status(connection_id, DDS::RETCODE_NO_DATA);
}

template<typename Msg>

void OpenDDS::FaceTSS::send_message	(	FACE::CONNECTION_ID_TYPE	connection_id,
		FACE::TIMEOUT_TYPE	timeout,
		FACE::TRANSACTION_ID_TYPE &	,
		const Msg &	message,
		FACE::MESSAGE_SIZE_TYPE	message_size,
		FACE::RETURN_CODE_TYPE &	return_code
	)

Definition at line 218 of file FaceTSS.h.

References convertDuration(), DataWriter, DDS::HANDLE_NIL, OpenDDS::FaceTSS::Entities::instance(), DDS::DataWriterQos::reliability, DDS::RELIABLE_RELIABILITY_QOS, DDS::RETCODE_BAD_PARAMETER, OpenDDS::FaceTSS::Entities::senders_, and update_status().

{
  if(!Entities::instance()->connections_.count(connection_id)) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  FACE::TRANSPORT_CONNECTION_STATUS_TYPE status =
    Entities::instance()->connections_[connection_id].connection_status;
  if (message_size < status.MAX_MESSAGE_SIZE) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  Entities::ConnIdToSenderMap& writers = Entities::instance()->senders_;
  if (!writers.count(connection_id)) {
    return_code = FACE::INVALID_PARAM;
    return;
  }

  typedef typename DCPS::DDSTraits<Msg>::DataWriterType DataWriter;
  const typename DataWriter::_var_type typedWriter =
    DataWriter::_narrow(writers[connection_id].dw);
  if (!typedWriter) {
    return_code = update_status(connection_id, DDS::RETCODE_BAD_PARAMETER);
    return;
  }
  writers[connection_id].status_valid = FACE::VALID;

  DDS::DataWriterQos dw_qos;
  typedWriter->get_qos(dw_qos);
  FACE::SYSTEM_TIME_TYPE max_blocking_time = convertDuration(dw_qos.reliability.max_blocking_time);
  if (dw_qos.reliability.kind == DDS::RELIABLE_RELIABILITY_QOS &&
      timeout != FACE::INF_TIME_VALUE &&
      ((max_blocking_time == FACE::INF_TIME_VALUE) || (timeout < max_blocking_time))) {
    return_code = update_status(connection_id, DDS::RETCODE_BAD_PARAMETER);
    return;
  }

  return_code = update_status(connection_id, typedWriter->write(message, DDS::HANDLE_NIL));
}

template<typename Msg>

void OpenDDS::FaceTSS::register_callback	(	FACE::CONNECTION_ID_TYPE	connection_id,
		const FACE::WAITSET_TYPE	,
		void(*)(FACE::TRANSACTION_ID_TYPE, Msg &, FACE::MESSAGE_TYPE_GUID, FACE::MESSAGE_SIZE_TYPE, const FACE::WAITSET_TYPE, FACE::RETURN_CODE_TYPE &)	callback,
		FACE::MESSAGE_SIZE_TYPE	max_message_size,
		FACE::RETURN_CODE_TYPE &	return_code
	)

Definition at line 350 of file FaceTSS.h.

References OpenDDS::FaceTSS::Listener< Msg >::add_callback(), DDS::DATA_AVAILABLE_STATUS, OpenDDS::FaceTSS::Entities::instance(), OpenDDS::FaceTSS::Entities::receivers_, and OpenDDS::FaceTSS::Entities::DDSAdapter::status_valid.

{
  Entities::ConnIdToReceiverMap& readers = Entities::instance()->receivers_;
  if (!readers.count(connection_id)) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  if(!Entities::instance()->connections_.count(connection_id)) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  FACE::TRANSPORT_CONNECTION_STATUS_TYPE status =
    Entities::instance()->connections_[connection_id].connection_status;
  if (max_message_size < status.MAX_MESSAGE_SIZE) {
    return_code = FACE::INVALID_PARAM;
    return;
  }
  DDS::DataReaderListener_ptr existing_listener = readers[connection_id]->dr->get_listener();
  if (existing_listener) {
    Listener<Msg>* typedListener = dynamic_cast<Listener<Msg>*>(existing_listener);
    typedListener->add_callback(callback);
  } else {
    DDS::DataReaderListener_var listener = new Listener<Msg>(callback, connection_id);
    readers[connection_id]->dr->set_listener(listener, DDS::DATA_AVAILABLE_STATUS);
  }
  if (readers[connection_id]->status_valid != FACE::VALID) {
    Entities::FaceReceiver* tmp = readers[connection_id];
    readers[connection_id] = new Entities::DDSTypedAdapter<Msg>(*readers[connection_id]);
    delete tmp;
  }
  readers[connection_id]->status_valid = FACE::VALID;

  return_code = FACE::RC_NO_ERROR;
}

---