cdrcombination.cpp

Mysql的处理

			//printf("cGCdrCom->RecordLocRecSeqNumToList in Combination\n");
			/* 合并Local Record Sequence Number */
			cGCdrCom->RecordLocRecSeqNumToList(pGCdrCg->LocalRecordSequeceNumnber,
						  	   pGCdr->LocalRecordSequeceNumnber);
	
			//printf("cGCdrCom->RecordExtensionInfo in Combination\n");
			/* 合并Record Extensions字段 */
			for(int cv = 0; cv < NUM_OF_CONTENT_INFO_NODE; cv++)
			{
				//cGCdrCom->RecordExtensionInfo(pGCdrCg->RecordExtensions,
				//			      pGCdr->RecordExtensions);
				cGCdrCom->RecordExtensionInfo(pGCdrCg->ContentInfo,
							      pGCdr->ContentInfo[cv]);
			}
			//pGCdrCg->ContentInfoLength += NUM_OF_CONTENT_INFO_NODE*SIZE_S_CONTENT_INFO_C;
			
			long pGCdrRecordSequenceNumber = atol(pGCdr->RecordSequenceNumber);
			long pGCdrCgRecordSequenceNumberMin = atol(pGCdrCg->RecordSequenceNumberMin);
			long pGCdrCgRecordSequenceNumberMax = atol(pGCdrCg->RecordSequenceNumberMax);
			if(pGCdrCgRecordSequenceNumberMin > pGCdrRecordSequenceNumber)
			{
				cGCdrCom->SaveFirstValueForGCdrCg(pGCdrCg, pGCdr);
				if ( !cGCdrCom->IsRecSeqNumNull(pGCdrCg->RecordSequenceNumber) )
				{
					/* 将SGSN地址重新赋值到Record Sequence Number链表头中 */
					pGCdrCg->RecordSequenceNumber[0] = pGCdr->GGSNAddress;
					
					//cSCdrCom->RecordSgsnAddrToRecSeqNumListHead(pSCdrCg->RecordSequenceNumber,
					//						 pSCdr->SGSNIPAddress);
				}

				if ( !cGCdrCom->IsLocalRecSeqNumNull(pGCdrCg->LocalRecordSequeceNumnber) )
				{
					//cSCdrCom->RecordGgsnAddrToLocalRecSeqNumListHead(pSCdrCg->LocalRecordSequeceNumnber,
											 //pSCdr->GGSNAddress);
					//						 pSCdr->SGSNIPAddress);
					/* 将SGSN地址重新赋值到Local Record Sequence Number链表头中 */
					pGCdrCg->LocalRecordSequeceNumnber[0] = pGCdr->GGSNAddress;
				}
				memcpy(pGCdrCg->RecordSequenceNumberMin,
					pGCdr->RecordSequenceNumber,
					sizeof(pGCdrCg->RecordSequenceNumberMin));
			}
			/* 如果RECORD SEQ NUM比前一张大，则重新赋值 */
			if (pGCdrCgRecordSequenceNumberMax < pGCdrRecordSequenceNumber)
			{
				/* Diagnostics数值进行重新赋值，即取最后一张小话单的数值 */
				//pGCdrCg->Diagnostics=pGCdr->Diagnostics;
				memcpy(pGCdrCg->Diagnostics,
					pGCdr->Diagnostics,
					sizeof(pGCdrCg->Diagnostics));
				
				memcpy(pGCdrCg->RecordSequenceNumberMax,
					pGCdr->RecordSequenceNumber,
					sizeof(pGCdrCg->RecordSequenceNumberMax));	
				memcpy(pGCdrCg->RecordOpenTimeLast,
					pGCdr->RecordOpenTime,
					sizeof(pGCdrCg->RecordOpenTimeLast));

				memcpy(pGCdrCg->CallDurationLast,
					pGCdr->CallDuration,
					sizeof(pGCdrCg->CallDurationLast));
			}			

			/* 拷贝和排序REC SEQ NUM列表、REC OPEN TIME列表和DURATION列表 */
			cGCdrCom->AddSortedRecSeqNum(pGCdrCg,
						     pGCdr->RecordSequenceNumber,
						     pGCdr->RecordOpenTime,
						     pGCdr->CallDuration);
			
			pGCdrCg->SeqId.push_back(pGCdr->SeqId);			   
			pGCdrCg->TableName.push_back(pGCdr->TableName);			   

			//printf("pGCdrCg->TableName.push_back in Combination\n");

			
			/* 判断话单的关闭原因 */
			/*if (!cGCdrCom->IsCauseForRecClosingNormalRelease(pGCdr))
			{
				continue;
			}*/
	
			/* 判断Record Sequence Number是否达到最大数值 */
			if (cGCdrCom->HaveMaxRecSeqNum(pGCdrCg))
			{
				//cout<<"###################caused by HaveMaxRecSeqNum"<<endl;
				sprintf(pGCdrCg->ConsolidationResult, "%d", 
					C_GCdrCombination::CONSOLIDATION_RESULT_NORMAL);
				//pGCdrCg->Diagnostics=pGCdr->Diagnostics;
				cGCdrCom->SaveCdr(pGCdrCg);
				cGCdrCom->EraseGCdrCg(pGCdrCg);
				pthread_mutex_unlock( &g_mutex );
				
				continue;
	
				//return true;
			}
			else
			{
				sprintf(pGCdrCg->ConsolidationResult, "%d", 
					C_GCdrCombination::CONSOLIDATION_RESULT_ABNORMAL);
					
				/* 如果话单合并门限时间到，则关闭此话单，合并操作结束 */
//				TODO: time limit control
				if (cGCdrCom->IsTimeOfCombinedReached(pGCdrCg))
				{
					//cout<<"###################caused by IsTimeOfCombinedReached"<<endl;
					cGCdrCom->SaveCdr(pGCdrCg);
					cGCdrCom->EraseGCdrCg(pGCdrCg);
					
					pthread_mutex_unlock( &g_mutex );
					continue;
				}				
//				if (false)
	//			{
		//		}
				else
				{
					/* 如果话单合并门限数量到，则关闭此话单，合并操作结束 */
					if (cGCdrCom->IsNumOfCombinedReached(pGCdrCg))
					{
					//cout<<"###################caused by IsNumOfCombinedReached"<<endl;
						//cout<<"ggsnaddr:"<<pGCdrCg->GGSNAddress<<endl;
						//cout<<"ChargingID:"<<pGCdrCg->ChargingID<<endl;
						//cout<<endl;
						
						sprintf(pGCdrCg->ConsolidationResult, "%d", 
							C_GCdrCombination::CONSOLIDATION_RESULT_REACHLIMIT);
						//pGCdrCg->Diagnostics=pGCdr->Diagnostics;
						//cout<<"before save"<<endl;
						cGCdrCom->SaveCdr(pGCdrCg);
						//cout<<"after save"<<endl;
						cGCdrCom->EraseGCdrCg(pGCdrCg);
						//cout<<"after erase"<<endl;
						pthread_mutex_unlock( &g_mutex );
						
						continue;
						//return true;
					}
					else
					{
						//cGCdrCom->GetCdr();
						//printf("Before pthread_mutex_unlock in Combination\n");
						pthread_mutex_unlock( &g_mutex );
						continue;
					}
				}
			}
		}//ELSE: read data from SHM

	}//while loop
	
	//Do NOT forget to free memory!
	if (NULL != pGCdr)
	{
		delete pGCdr;
		pGCdr = NULL;
	}
   	return NULL; //meaningless, a value must be returned to meet the requirement of this function!
	//return ;
};

S_OGCDR2 * C_GCdrCombination::SetArrayNext(S_OGCDR2 *pGCdr)
{
	pGCdr++;
	return pGCdr;
}



/* 静态功能函数 */
/* 队列的重复判断功能函数 */
/* 
 * Determine whether some specified item exists in the vector
 * If exists, return true; return false otherwise
 */
static bool containItem(vector<string>& v, const char *item) 
{
	bool bContain = false;
	if  ( find(v.begin(), v.end(), item) != v.end( ) ) 
	{
		bContain = true;
	}
	return bContain;
}

/* 
 * Determine whether some specified item exists in the vector
 * If exists, return true; return false otherwise
 */
static bool containItem(deque<string>& d, const char *item) 
{
	bool bContain = false;
	if  ( find(d.begin(), d.end(), item) != d.end( ) ) 
	{
		bContain = true;
	}
	return bContain;
}
/* 话单合并基类的定义 */

//const  map<CidAndGgsnAddrStruct, S_CGCDR_C*>::size_type C_GCdrCombination::R99GCDRCG_MAX_SIZE=5;
/*
 * Sort the incoming cdr(small cdr) into the CdrCg(big cdr).
 * The sort criteria is charging id and ggsn addr.
 * The function returns true if sorting succeeds; 
 * otherwise return false.
 */
S_CGCDR_C * C_GCdrCombination::GcdrSort (S_OGCDR2 * pGCdr, unsigned long ulCounterFlag, 
//	int iTriggerType = C_GCdrCombination::TRIGGER_TYPE_NORMAL)
	int iTriggerType)
{
	//Ensure the pointer to cdr is not null
	if ( NULL == pGCdr)
	{
		return NULL;
	}
	
	//Indicate whether the small CDR should be sorted
	bool bSort = true;
	if( C_GCdrCombination::TRIGGER_TYPE_TIME == iTriggerType ) 
	{
		string sROTSub = string(pGCdr->RecordOpenTime).substr(0, 12);
		if( ! (C_GCdrCombination::TRIGGER_TIME_START <=  sROTSub
			&& C_GCdrCombination::TRIGGER_TIME_END >= sROTSub) )
		{
				cout<<"ROT invalid, RSN:"<<pGCdr->RecordSequenceNumber<<",ROT:"<<pGCdr->RecordOpenTime<<endl;
				bSort = false;
		}
	}
	
 	//if the current CDR can be merged, set the flag to PROGRAM_VACANCY;
	//else set the flag to PROGRAM_GCDR for later use
	unsigned short iFlag = PROGRAM_VACANCY;
		
	S_CGCDR_C *pGCdrCg = NULL;
	
	if (bSort)
	{
		//Charging id and ggsn addr structure, acts as key of the map
		CidAndGgsnAddrStruct sortStruct;	
		memcpy(sortStruct.ChargingID, pGCdr->ChargingID, sizeof(pGCdr->ChargingID));
		memcpy(sortStruct.GGSNAddress, pGCdr->GGSNAddress, sizeof(pGCdr->GGSNAddress));
		
		//The map contains pairs of above structure and corresopnding CdrCg
	//	static map<CidAndGgsnAddrStruct, S_CGCDR_C*> map_GCdrCg;
	
		//Search in the map according to Charging Id and ggsn addr
		map<CidAndGgsnAddrStruct, S_CGCDR_C*>::iterator map_it;
		map_it = map_GCdrCg.find(sortStruct);
		
		//Get the CdrCg from map if found		
		if ( map_it != map_GCdrCg.end() )
		{
			pGCdrCg = map_it -> second;
			//fprintf(fptr2,"found CG\n");
		}
		//If not found and not reach size limit, create a new CdrCg and insert a new pair into the map
		else 
		{	
			if (GCDRCG_MAX_SIZE != map_GCdrCg.size())
			{
				pGCdrCg = new S_CGCDR_C;
				if (NULL == pGCdrCg)
				{
					cout<<"No enough memory when allocating memory for "
						<<"S_CGCDR_C"<<" in "<<"C_GCdrCombination::GcdrSort"<<" !"
						<<endl;
					//delete pGCdr;
					//delete this;
					//ClearGCdrMap();
					//exit(0);
					WriteCombinedGCdrErrorInfoToFile("GcdrSort:No enough memory when allocating memory for S_CGCDR_C");
					iFlag = PROGRAM_GCDR;
					//return NULL;
				}
	
				InitGCdrCg(pGCdrCg, pGCdr);
				//memcpy(pGCdrCg->ChargingCharacteristics,pGCdr->ChargingCharacteristics,sizeof(pGCdrCg->ChargingCharacteristics));
				//pGCdrCg->AnonyAccessIndicator=pGCdr->AnonyAccessIndicator;
				//memcpy(pGCdrCg->RemotePdpAddr,pGCdr->RemotePdpAddr,sizeof(pGCdrCg->RemotePdpAddr));
	   			map_GCdrCg.insert ( Pair_GCdr_Sort (sortStruct, pGCdrCg) );
	   		
	   		//fprintf(fptr2,"in sort, not reach limit, chargingid:%s,ggsn:%s,rsn:%s\n",\
	   			//pGCdr->ChargingID,pGCdr->GGSNAddress,pGCdr->RecordSequenceNumber);
	
			}
			else
			{
	   			iFlag = PROGRAM_GCDR;
			}
	
		}
		
	}
	long lReturn = this->ShareMemory_GCDR.UpdateNextFlag(ulCounterFlag,iFlag);
	if(lReturn < 0)
	{
	    cerr<<"Update Next Flag "<<iFlag<<" Error! Return:"<<lReturn<<endl;
	}
	else
	{
	    //cout<<"Update Next Flag "<<iFlag<<" Successful! Return:"<<lReturn<<endl;
	}

	return pGCdrCg;
}

static long GetCurrentTicks()
{
	struct timeval tp; 
	gettimeofday(&tp,NULL);
	return 1000000 * tp.tv_sec + tp.tv_usec;//in microseconds
}

void C_GCdrCombination::SaveFirstValueForGCdrCg(S_CGCDR_C *pGCdrCg, S_OGCDR2 *pGCdr)
{
	pGCdrCg->RecordType=pGCdr->RecordType;
	pGCdrCg->NetworkInitiatedPDPContext=pGCdr->NetworkInitiatedPDPContext;
	memcpy(pGCdrCg->ServedIMSI,pGCdr->ServedIMSI,sizeof(pGCdrCg->ServedIMSI));
	memcpy(pGCdrCg->ServedMSISDN,pGCdr->ServedMSISDN,sizeof(pGCdrCg->ServedMSISDN));
	memcpy(pGCdrCg->AccessPointNameNI,pGCdr->AccessPointNameNI,sizeof(pGCdrCg->AccessPointNameNI));
	pGCdrCg->APNSelectionMode=pGCdr->APNSelectionMode;
	memcpy(pGCdrCg->PDPTypeNumber,pGCdr->PDPTypeNumber,sizeof(pGCdrCg->PDPTypeNumber));
	memcpy(pGCdrCg->ServedPDPAddress,pGCdr->ServedPDPAddress,sizeof(pGCdrCg->ServedPDPAddress));
	pGCdrCg->DynamicAddressFlag=pGCdr->DynamicAddressFlag;
	memcpy(pGCdrCg->SgsnPlmnIdentifier,pGCdr->SgsnPlmnIdentifier,sizeof(pGCdrCg->SgsnPlmnIdentifier));
	memcpy(pGCdrCg->NodeID, pGCdr->NodeID, sizeof(pGCdrCg->NodeID));
	//pGCdrCg->Diagnostics = pGCdr->Diagnostics;
	memcpy(pGCdrCg->Diagnostics,
		pGCdr->Diagnostics,
		sizeof(pGCdrCg->Diagnostics));
	
	memcpy(pGCdrCg->ChargingCharacteristics,pGCdr->ChargingCharacteristics,sizeof(pGCdrCg->ChargingCharacteristics));
	pGCdrCg->AnonyAccessIndicator=pGCdr->AnonyAccessIndicator;
	memcpy(pGCdrCg->RemotePdpAddr,pGCdr->RemotePdpAddr,sizeof(pGCdrCg->RemotePdpAddr));
	pGCdrCg->ExtensionType = pGCdr->ExtensionType;
}

void C_GCdrCombination::InitGCdrCg(S_CGCDR_C *pGCdrCg, S_OGCDR2 *pGCdr)
{

	pGCdrCg->ListOfTrafficDataVolumes.clear();
	pGCdrCg->CauseForRecClosing.clear();
	pGCdrCg->SGSNIPAddress.clear();
	pGCdrCg->RecordSequenceNumber.clear();
	pGCdrCg->LocalRecordSequeceNumnber.clear();
	//pGCdrCg->RecordExtensions.clear();
	pGCdrCg->ContentInfo.clear();
	pGCdrCg->SeqId.clear();
	pGCdrCg->TableName.clear();
	
	pGCdrCg->StartTimeOfCombined = GetCurrentTicks();
	memcpy(pGCdrCg->ChargingID, 
	   pGCdr->ChargingID,
	   sizeof(pGCdrCg->ChargingID));
	memcpy(pGCdrCg->GGSNAddress, 
	   pGCdr->GGSNAddress,
	   sizeof(pGCdrCg->GGSNAddress));
	pGCdrCg->CallDuration[0]='\0';
	pGCdrCg->RecordOpenTime[0]='\0';
	pGCdrCg->ConsolidationResult[0]='\0';
	pGCdrCg->NumOfCombined = 0;

	/* Record Extensions信息的初步处理 */
	//pGCdrCg->ExtensionType = pGCdr->ExtensionType;
	pGCdrCg->ContentInfoLength = 0;
 
	pGCdrCg->ChargingCharSelMode = '\0';

		
	memcpy(pGCdrCg->RecordSequenceNumberMin, 
		pGCdr->RecordSequenceNumber, 
		sizeof(pGCdrCg->RecordSequenceNumberMin));
			
	memcpy(pGCdrCg->RecordSequenceNumberMax, 
		pGCdr->RecordSequenceNumber, 
		sizeof(pGCdrCg->RecordSequenceNumberMax));
	memcpy(pGCdrCg->RecordOpenTimeLast,
		pGCdr->RecordOpenTime,
		sizeof(pGCdrCg->RecordOpenTimeLast));

	memcpy(pGCdrCg->CallDurationLast,
		pGCdr->CallDuration,
		sizeof(pGCdrCg->CallDurationLast));
		
	/* SortedRecSeqNum：这里作初始化操作 */
	/* 话单合并时需要进行排序和赋值，话单关闭时需要进行相应的释放 */
	pGCdrCg->SortedRecSeqNum = NULL;
			
	/* 需要取第一章话单的相关内容 */
	SaveFirstValueForGCdrCg(pGCdrCg, pGCdr);
/*	
	pGCdrCg->RecordType=pGCdr->RecordType;
	pGCdrCg->NetworkInitiatedPDPContext=pGCdr->NetworkInitiatedPDPContext;
	memcpy(pGCdrCg->ServedIMSI,pGCdr->ServedIMSI,sizeof(pGCdrCg->ServedIMSI));
	memcpy(pGCdrCg->ServedMSISDN,pGCdr->ServedMSISDN,sizeof(pGCdrCg->ServedMSISDN));
	memcpy(pGCdrCg->AccessPointNameNI,pGCdr->AccessPointNameNI,sizeof(pGCdrCg->AccessPointNameNI));
	pGCdrCg->APNSelectionMode=pGCdr->APNSelectionMode;
	memcpy(pGCdrCg->PDPTypeNumber,pGCdr->PDPTypeNumber,sizeof(pGCdrCg->PDPTypeNumber));
	memcpy(pGCdrCg->ServedPDPAddress,pGCdr->ServedPDPAddress,sizeof(pGCdrCg->ServedPDPAddress));
	pGCdrCg->DynamicAddressFlag=pGCdr->DynamicAddressFlag;
	memcpy(pGCdrCg->SgsnPlmnIdentifier,pGCdr->SgsnPlmnIdentifier,sizeof(pGCdrCg->SgsnPlmnIdentifier));
	memcpy(pGCdrCg->NodeID, pGCdr->NodeID, sizeof(pGCdrCg->NodeID));
	pGCdrCg->Diagnostics = pGCdr->Diagnostics;
	
	memcpy(pGCdrCg->ChargingCharacteristics,pGCdr->ChargingCharacteristics,sizeof(pGCdrCg->ChargingCharacteristics));
	pGCdrCg->AnonyAccessIndicator=pGCdr->AnonyAccessIndicator;
	memcpy(pGCdrCg->RemotePdpAddr,pGCdr->RemotePdpAddr,sizeof(pGCdrCg->RemotePdpAddr));
*/	
}

int C_CdrCombination::SaveCdr(S_CGCDR_C * pCg)
{
	//Connect to db
	//C_CDR cdr;
//	if(!cdr.Connect(C_GCdrCombination::DB_USER.c_str(), C_GCdrCombination::DB_PASSWORD.c_str(), 
//			C_GCdrCombination::DB_DBNAME.c_str()) )
//	{
		//fprintf(fptr2,"Connect failed\n");