<commit_before>/*******************************************************************************
* ALMA - Atacama Large Millimiter Array
* (c) European Southern Observatory, 2011
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
*
* "@(#) $Id: bulkDataNTStream.cpp,v 1.49 2013/02/06 23:12:01 bjeram Exp $"
*
* who       when      what
* --------  --------  ----------------------------------------------
* bjeram  2011-04-19  created
*/
#include "bulkDataNTStream.h"
#include <iostream>
using namespace ACS_BD_Errors;
using namespace ACSErrTypeCommon;
using namespace ACS_DDS_Errors;

using namespace AcsBulkdata;

unsigned int BulkDataNTStream::factoryRefCount_m = 0;
DDS::DomainParticipantFactory *BulkDataNTStream::factory_m=0;

unsigned int BulkDataNTStream::globalParticipantRefCount_m = 0;
DDS::DomainParticipant* BulkDataNTStream::globalParticipant_m=0;

ACE_Recursive_Thread_Mutex BulkDataNTStream::globalPartMutex_m;
ACE_Recursive_Thread_Mutex BulkDataNTStream::DDSFactoryMutex_m;

BulkDataNTStream::BulkDataNTStream(const char* name, const StreamConfiguration &cfg) :
	    streamName_m(name), configuration_m(cfg), participant_m(0)
{
  AUTO_TRACE(__PRETTY_FUNCTION__);

  try
  {

	  //TBD: Maybe factory and global DDS stream should be treated with singleton, but at least for global stream there is no default ctor
	  {//critical section
		  // Maybe we should wait with timeout here
		  ACE_GUARD (ACE_Recursive_Thread_Mutex, protDDSFActory, DDSFactoryMutex_m); // protection

		  if (factory_m==0)	  createDDSFactory();  //it is enough to have one factory
		  factoryRefCount_m++;
	  }//end of critical section

	  addDDSQoSProfile(configuration_m, name);  //add QoS profile for stream (and flows)

	  if (configuration_m.isParticipantPerStream())
	  {
		  participant_m = createDDSParticipant(); //we have a participant per stream
	  }
	  else
	  {
		  {//Beginning of critical section
			  // Maybe we should wait with timeout here
			  ACE_GUARD (ACE_Recursive_Thread_Mutex, protGlobalPart, globalPartMutex_m); // protection
			  // for performance reason (number of threads) is better to have just one participant, but it might be good to have possibility to create
			  // also a participant per stream
			  if (BulkDataNTStream::globalParticipantRefCount_m==0 && BulkDataNTStream::globalParticipant_m==0)
			  {
				  participant_m = BulkDataNTStream::globalParticipant_m = createDDSParticipant();
			  }
			  else
			  {
				  participant_m = BulkDataNTStream::globalParticipant_m;
			  }
			  BulkDataNTStream::globalParticipantRefCount_m++;   // but we need to know how many users of participant do we have
		  }//end critical section
		  ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "Going to use global participant for stream: %s.", streamName_m.c_str()));
	  }//if-else (configuration_m.participantPerStream)

	  ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "Stream: %s has been created.", streamName_m.c_str()));
  }catch(const ACSErr::ACSbaseExImpl &e)
  {
	  destroyDDSFactory();
	  StreamCreateProblemExImpl ex (e, __FILE__, __LINE__, __PRETTY_FUNCTION__);
	  ex.setStreamName(name);
	  throw ex;
  }//try-catch
}//BulkDataNTStream


BulkDataNTStream::~BulkDataNTStream()
{
  AUTO_TRACE(__PRETTY_FUNCTION__);

  if (configuration_m.participantPerStream)
  {
	  destroyDDSParticipant(participant_m);
  }else
  {
	  ACE_GUARD (ACE_Recursive_Thread_Mutex, protGlobalPart, globalPartMutex_m); // protection
	  BulkDataNTStream::globalParticipantRefCount_m--;
	  if (BulkDataNTStream::globalParticipantRefCount_m==0)
	  {
		  ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "Going to destroy global participant."));
		  destroyDDSParticipant(BulkDataNTStream::globalParticipant_m);
		  BulkDataNTStream::globalParticipant_m=0;
	  }
  }//if-else

  removeDDSQoSProfile(configuration_m);  //remove QoS profile for stream

  participant_m=0;
  {//critical section
	  // Maybe we should wait with timeout here
	  ACE_GUARD (ACE_Recursive_Thread_Mutex, protDDSFActory, DDSFactoryMutex_m); // protection
	  destroyDDSFactory(); // we do not need to call, but .... it just decrease the counter

  }//end of critical section

}//~BulkDataNTStream


void BulkDataNTStream::createDDSFactory()
{
	AUTO_TRACE(__PRETTY_FUNCTION__);
	DDS::ReturnCode_t ret;
	DDS::DomainParticipantFactoryQos factory_qos;

	factory_m = DDS::DomainParticipantFactory::get_instance();
	//factoryRefCount_m++; //anyway we have to increase the counter that we know when to call finalize_instance,
	// ... so it must be right after call get_instance call

	// needed by RTI only
	ret = factory_m->get_qos(factory_qos);
	if (ret!=DDS::RETCODE_OK)
	{
		DDSQoSSetProblemExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setDDSTypeCode(ret);
		ex.setQoS("factory_m->get_qos");
		throw ex;
	}//if


	factory_qos.entity_factory.autoenable_created_entities = DDS_BOOLEAN_FALSE;

	// if both values are true (default) we prevent that default values are taken from default places
	factory_qos.profile.ignore_user_profile = DDSConfiguration::ignoreUserProfileQoS;
	factory_qos.profile.ignore_environment_profile = DDSConfiguration::ignoreEnvironmentProfileQoS;
	factory_qos.resource_limits.max_objects_per_thread = 4096; //if we want to have more than 10 participants (at the moment can not be set in XML)

	if (DDSConfiguration::urlProfileQoS.length()>0)
	{
		factory_qos.profile.url_profile.ensure_length(1,1);
		factory_qos.profile.url_profile[0] = DDS_String_dup(DDSConfiguration::urlProfileQoS.c_str());
	}//if

	ret = factory_m->set_qos(factory_qos);
	if (ret!=DDS::RETCODE_OK)
	{
		DDSQoSSetProblemExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setDDSTypeCode(ret);
		ex.setQoS("factory_m->set_qos");
		throw ex;
	}//if

/*	DDS_StringSeq profile_names;
	ret = factory_m->get_qos_profiles(profile_names, "BulkDataQoSLibrary");
	printf("profile length: %d\n", profile_names.length());
*/

}//createDDSFactory


void BulkDataNTStream::destroyDDSFactory()
{
  // if factory_m is not NULL means that get_instance was successful , and that factoryRefCount_m was also increased
  if(factory_m!=0)
    {
      factoryRefCount_m--;
     //TBD: cause problem if all streams are deleted and new is created   if (factoryRefCount_m==0)  DDS::DomainParticipantFactory::finalize_instance();
    }
}//destroyDDSFactory

DDS::DomainParticipant* BulkDataNTStream::createDDSParticipant()
{
	AUTO_TRACE(__PRETTY_FUNCTION__);
	DDS::ReturnCode_t ret;
	DDS::DomainParticipantQos participant_qos;
	int domainID=0; //TBD: where to get domain ID
	DDS::DomainParticipant* domainParticipant;

	if (factory_m==NULL)
	{
		NullPointerExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setVariable("factory_m");
		throw ex;
	}

	// If the profileQoS set on the configuration corresponds to a default stream profile,
	// then we use the default library as the library to search for it.
	// Otherwise, it means that the user specified it, so it must belong
	// to the specified library too
	const char *library = 0;
	if( configuration_m.libraryQos.compare((const char*)DDSConfiguration::DEFAULT_LIBRARY) == 0 &&
		(configuration_m.profileQos.compare((const char*)DDSConfiguration::DEFAULT_RECEIVER_STREAM_PROFILE) == 0 ||
	    configuration_m.profileQos.compare((const char*)DDSConfiguration::DEFAULT_SENDER_STREAM_PROFILE) == 0)
		)
		library = DDSConfiguration::DEFAULT_LIBRARY;
	else
		library = configuration_m.libraryQos.c_str();

	ret = factory_m->get_participant_qos_from_profile(participant_qos, library, configuration_m.profileQos.c_str());
	if (ret!=DDS::RETCODE_OK)
	{
		DDSQoSSetProblemExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setDDSTypeCode(ret);
		ex.setQoS("get_participant_qos_from_profile");<commit_msg>ICT-9064: applied patch provided by Kris<commit_after>/*******************************************************************************
* ALMA - Atacama Large Millimiter Array
* (c) European Southern Observatory, 2011
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 2.1 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
*
* "@(#) $Id: bulkDataNTStream.cpp,v 1.49 2013/02/06 23:12:01 bjeram Exp $"
*
* who       when      what
* --------  --------  ----------------------------------------------
* bjeram  2011-04-19  created
*/
#include "bulkDataNTStream.h"
#include <iostream>
using namespace ACS_BD_Errors;
using namespace ACSErrTypeCommon;
using namespace ACS_DDS_Errors;

using namespace AcsBulkdata;

unsigned int BulkDataNTStream::factoryRefCount_m = 0;
DDS::DomainParticipantFactory *BulkDataNTStream::factory_m=0;

unsigned int BulkDataNTStream::globalParticipantRefCount_m = 0;
DDS::DomainParticipant* BulkDataNTStream::globalParticipant_m=0;

ACE_Recursive_Thread_Mutex BulkDataNTStream::globalPartMutex_m;
ACE_Recursive_Thread_Mutex BulkDataNTStream::DDSFactoryMutex_m;

BulkDataNTStream::BulkDataNTStream(const char* name, const StreamConfiguration &cfg) :
	    streamName_m(name), configuration_m(cfg), participant_m(0)
{
  AUTO_TRACE(__PRETTY_FUNCTION__);

  try
  {

	  //TBD: Maybe factory and global DDS stream should be treated with singleton, but at least for global stream there is no default ctor
	  {//critical section
		  // Maybe we should wait with timeout here
		  ACE_GUARD (ACE_Recursive_Thread_Mutex, protDDSFActory, DDSFactoryMutex_m); // protection

		  if (factory_m==0)	  createDDSFactory();  //it is enough to have one factory
		  factoryRefCount_m++;
	  }//end of critical section

	  addDDSQoSProfile(configuration_m, name);  //add QoS profile for stream (and flows)

	  if (configuration_m.isParticipantPerStream())
	  {
		  participant_m = createDDSParticipant(); //we have a participant per stream
	  }
	  else
	  {
		  {//Beginning of critical section
			  // Maybe we should wait with timeout here
			  ACE_GUARD (ACE_Recursive_Thread_Mutex, protGlobalPart, globalPartMutex_m); // protection
			  // for performance reason (number of threads) is better to have just one participant, but it might be good to have possibility to create
			  // also a participant per stream
			  if (BulkDataNTStream::globalParticipantRefCount_m==0 && BulkDataNTStream::globalParticipant_m==0)
			  {
				  participant_m = BulkDataNTStream::globalParticipant_m = createDDSParticipant();
			  }
			  else
			  {
				  participant_m = BulkDataNTStream::globalParticipant_m;
			  }
			  BulkDataNTStream::globalParticipantRefCount_m++;   // but we need to know how many users of participant do we have
		  }//end critical section
		  ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "Going to use global participant for stream: %s.", streamName_m.c_str()));
	  }//if-else (configuration_m.participantPerStream)

	  ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "Stream: %s has been created.", streamName_m.c_str()));
  }catch(const ACSErr::ACSbaseExImpl &e)
  {
	  destroyDDSFactory();
	  StreamCreateProblemExImpl ex (e, __FILE__, __LINE__, __PRETTY_FUNCTION__);
	  ex.setStreamName(name);
	  throw ex;
  }//try-catch
}//BulkDataNTStream


BulkDataNTStream::~BulkDataNTStream()
{
  AUTO_TRACE(__PRETTY_FUNCTION__);

  if (configuration_m.participantPerStream)
  {
	  destroyDDSParticipant(participant_m);
  }else
  {
	  ACE_GUARD (ACE_Recursive_Thread_Mutex, protGlobalPart, globalPartMutex_m); // protection
	  BulkDataNTStream::globalParticipantRefCount_m--;
	  if (BulkDataNTStream::globalParticipantRefCount_m==0)
	  {
		  ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "Going to destroy global participant."));
		  destroyDDSParticipant(BulkDataNTStream::globalParticipant_m);
		  BulkDataNTStream::globalParticipant_m=0;
	  }
  }//if-else

  removeDDSQoSProfile(configuration_m);  //remove QoS profile for stream

  participant_m=0;
  {//critical section
	  // Maybe we should wait with timeout here
	  ACE_GUARD (ACE_Recursive_Thread_Mutex, protDDSFActory, DDSFactoryMutex_m); // protection
	  destroyDDSFactory(); // we do not need to call, but .... it just decrease the counter

  }//end of critical section

}//~BulkDataNTStream


void BulkDataNTStream::createDDSFactory()
{
	AUTO_TRACE(__PRETTY_FUNCTION__);
	DDS::ReturnCode_t ret;
	DDS::DomainParticipantFactoryQos factory_qos;

	factory_m = DDS::DomainParticipantFactory::get_instance();
	//factoryRefCount_m++; //anyway we have to increase the counter that we know when to call finalize_instance,
	// ... so it must be right after call get_instance call

	// needed by RTI only
	ret = factory_m->get_qos(factory_qos);
	if (ret!=DDS::RETCODE_OK)
	{
		DDSQoSSetProblemExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setDDSTypeCode(ret);
		ex.setQoS("factory_m->get_qos");
		throw ex;
	}//if


	factory_qos.entity_factory.autoenable_created_entities = DDS_BOOLEAN_FALSE;

	// if both values are true (default) we prevent that default values are taken from default places
	factory_qos.profile.ignore_user_profile = DDSConfiguration::ignoreUserProfileQoS;
	factory_qos.profile.ignore_environment_profile = DDSConfiguration::ignoreEnvironmentProfileQoS;
	factory_qos.resource_limits.max_objects_per_thread = 4096; //if we want to have more than 10 participants (at the moment can not be set in XML)

	if (DDSConfiguration::urlProfileQoS.length()>0)
	{
		factory_qos.profile.url_profile.ensure_length(1,1);
		factory_qos.profile.url_profile[0] = DDS_String_dup(DDSConfiguration::urlProfileQoS.c_str());
	}//if

	ret = factory_m->set_qos(factory_qos);
	if (ret!=DDS::RETCODE_OK)
	{
		DDSQoSSetProblemExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setDDSTypeCode(ret);
		ex.setQoS("factory_m->set_qos");
		throw ex;
	}//if

/*	DDS_StringSeq profile_names;
	ret = factory_m->get_qos_profiles(profile_names, "BulkDataQoSLibrary");
	printf("profile length: %d\n", profile_names.length());
*/

}//createDDSFactory


void BulkDataNTStream::destroyDDSFactory()
{
  // if factory_m is not NULL means that get_instance was successful , and that factoryRefCount_m was also increased
  if(factory_m!=0)
    {
      factoryRefCount_m--;
     //TBD: cause problem if all streams are deleted and new is created   if (factoryRefCount_m==0)  DDS::DomainParticipantFactory::finalize_instance();
    }
}//destroyDDSFactory

DDS::DomainParticipant* BulkDataNTStream::createDDSParticipant()
{
	AUTO_TRACE(__PRETTY_FUNCTION__);
	DDS::ReturnCode_t ret;
	DDS::DomainParticipantQos participant_qos;
	int domainID=0; //TBD: where to get domain ID
	DDS::DomainParticipant* domainParticipant;

	// In test environments where multiple developers are operating on the same subnet simultaneously,
	// it is important to ensure that DDS communications do not interfere with one another.
	// The mechanism by which DDS provides this isolation is through Domains.
	// See : https://community.rti.com/glossary/domain
	//       And while not for our exact version of DDS; still useful:
	//       https://community.rti.com/static/documentation/connext-dds/5.2.0/doc/manuals/connext_dds/html_files/RTI_ConnextDDS_CoreLibraries_UsersManual/Content/UsersManual/ChoosingDomainID.htm
	// A similar mechanism to isolate ACS instances is available via the $ACS_INSTANCE environment variable
	// I (Kris) propose that we incorporate the $ACS_INSTANCE into the domainID to provide consistent isolation between developers.
	char *acsInstance = std::getenv("ACS_INSTANCE");
	if (acsInstance)
	{
		domainID = (int)strtol(acsInstance, NULL, 10);
		ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "Found ACS_INSTANCE %s, using it as DDS domainID : %d",
		                                           acsInstance, domainID));
	}
	else
	{
		ACS_LOG(LM_RUNTIME_CONTEXT, __FUNCTION__, (LM_DEBUG, "No ACS_INSTANCE was found, DDS domainID = %d", domainID)); 
	}

	if (factory_m==NULL)
	{
		NullPointerExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setVariable("factory_m");
		throw ex;
	}

	// If the profileQoS set on the configuration corresponds to a default stream profile,
	// then we use the default library as the library to search for it.
	// Otherwise, it means that the user specified it, so it must belong
	// to the specified library too
	const char *library = 0;
	if( configuration_m.libraryQos.compare((const char*)DDSConfiguration::DEFAULT_LIBRARY) == 0 &&
		(configuration_m.profileQos.compare((const char*)DDSConfiguration::DEFAULT_RECEIVER_STREAM_PROFILE) == 0 ||
	    configuration_m.profileQos.compare((const char*)DDSConfiguration::DEFAULT_SENDER_STREAM_PROFILE) == 0)
		)
		library = DDSConfiguration::DEFAULT_LIBRARY;
	else
		library = configuration_m.libraryQos.c_str();

	ret = factory_m->get_participant_qos_from_profile(participant_qos, library, configuration_m.profileQos.c_str());
	if (ret!=DDS::RETCODE_OK)
	{
		DDSQoSSetProblemExImpl ex(__FILE__, __LINE__, __PRETTY_FUNCTION__);
		ex.setDDSTypeCode(ret);
		ex.setQoS("get_participant_qos_from_profile");
{'fraction_non_alphanumeric': 0.10786591437105875, 'fraction_numerical': 0.00857395729616108, 'mean_word_length': 4.830054821025476, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 19, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3867733', 'n_tokens_mistral': 6147, 'n_tokens_neox': 5705, 'n_words': 1490}