cdrcombination.cpp

Mysql的处理

		pGCdrCg->SortedRecSeqNum->prev = NULL;
		
		memcpy(pGCdrCg->SortedRecSeqNum->RecordSequenceNumber,
		       RecordSequenceNumber,
		       sizeof(pGCdrCg->SortedRecSeqNum->RecordSequenceNumber));
		       
		memcpy(pGCdrCg->SortedRecSeqNum->RecordOpenTime,
		       RecordOpenTime,
		       sizeof(pGCdrCg->SortedRecSeqNum->RecordOpenTime));
		       
		memcpy(pGCdrCg->SortedRecSeqNum->CallDuration,
		       CallDuration,
		       sizeof(pGCdrCg->SortedRecSeqNum->CallDuration));
		return ;
	}
printf("%d\n", __LINE__);	
	/* 3.如果链表不为空，则插入到链表中，并进行排序 */
	/* (1)待插入节点的赋值 */
	SortedRecSeqNumList * SortedRecSeqNumTemp = NULL;
	SortedRecSeqNumTemp = (SortedRecSeqNumList * )malloc(sizeof(SortedRecSeqNumList));
	memset(SortedRecSeqNumTemp,
	       0x00,
	       sizeof(SortedRecSeqNumList));
	       
	SortedRecSeqNumTemp->next = NULL;
	SortedRecSeqNumTemp->prev = NULL;
	
	memcpy(SortedRecSeqNumTemp->RecordSequenceNumber,
	       RecordSequenceNumber,
	       sizeof(SortedRecSeqNumTemp->RecordSequenceNumber));
	       
	memcpy(SortedRecSeqNumTemp->RecordOpenTime,
	       RecordOpenTime,
	       sizeof(SortedRecSeqNumTemp->RecordOpenTime));
	       
	memcpy(SortedRecSeqNumTemp->CallDuration,
	       CallDuration,
	       sizeof(SortedRecSeqNumTemp->CallDuration));
	
printf("%d\n", __LINE__);	
	/* (2)如果待插入节点的REC SEQ NUM比列表头还小，直接作为列表头 */
	SortedRecSeqNumList * pSortedRecSeqNum = pGCdrCg->SortedRecSeqNum;
printf("%d\n", __LINE__);	
	printf("%s\n", SortedRecSeqNumTemp->RecordSequenceNumber);
printf("%d\n", __LINE__);	
	printf("%d\n", atol(SortedRecSeqNumTemp->RecordSequenceNumber));
printf("%d\n", __LINE__);	
	printf("%s\n", pSortedRecSeqNum->RecordSequenceNumber);
printf("%d\n", __LINE__);	
	printf("%d\n", atol(pSortedRecSeqNum->RecordSequenceNumber));

	if ( atol(pSortedRecSeqNum->RecordSequenceNumber) > 
	     atol(SortedRecSeqNumTemp->RecordSequenceNumber))
	{
printf("%d\n", __LINE__);	
		SortedRecSeqNumTemp->next = pSortedRecSeqNum;
printf("%d\n", __LINE__);	
		pSortedRecSeqNum->prev = SortedRecSeqNumTemp;
printf("%d\n", __LINE__);	
		pGCdrCg->SortedRecSeqNum = SortedRecSeqNumTemp;
printf("%d\n", __LINE__);	
		return ;
	}
printf("%d\n", __LINE__);	
	pSortedRecSeqNum = pSortedRecSeqNum->next;
printf("%d\n", __LINE__);	
	SortedRecSeqNumList * pSortedRecSeqNumPrev = pSortedRecSeqNum;
printf("%d\n", __LINE__);		
	/* (3)如果只有一个节点 */
	if (NULL == pSortedRecSeqNum)
	{
printf("%d\n", __LINE__);	
		SortedRecSeqNumTemp->prev = pGCdrCg->SortedRecSeqNum;
printf("%d\n", __LINE__);	
		pGCdrCg->SortedRecSeqNum->next = SortedRecSeqNumTemp;
printf("%d\n", __LINE__);	
		return ;
	}
	
	/* (4)否则，找到插入的地点，并插到合适的地方 */
	//while ( NULL != pSortedRecSeqNum->next )
	while ( NULL != pSortedRecSeqNum )
	{
		pSortedRecSeqNumPrev = pSortedRecSeqNum;
		if ( atol(pSortedRecSeqNum->RecordSequenceNumber) < 
		     atol(SortedRecSeqNumTemp->RecordSequenceNumber))
		{
			pSortedRecSeqNum = pSortedRecSeqNum->next;
			continue ;
		}
		
		/* 找到插入点，则进行插入操作 */
		SortedRecSeqNumTemp->next = pSortedRecSeqNum;
		SortedRecSeqNumTemp->prev = pSortedRecSeqNum->prev;
		pSortedRecSeqNum->prev->next = SortedRecSeqNumTemp;
		pSortedRecSeqNum->prev = SortedRecSeqNumTemp;
		return ;
	}
printf("%d\n", __LINE__);	
	/* (5)是否可以插入到链表尾部  */
	if (NULL == pSortedRecSeqNum)
	{
printf("%d\n", __LINE__);
		if ( atol(pSortedRecSeqNumPrev->RecordSequenceNumber) < 
		     atol(SortedRecSeqNumTemp->RecordSequenceNumber))		
		{
printf("%d\n", __LINE__);
			pSortedRecSeqNumPrev->next = SortedRecSeqNumTemp;
printf("%d\n", __LINE__);
			SortedRecSeqNumTemp->prev = pSortedRecSeqNumPrev;
printf("%d\n", __LINE__);
		}
	}
printf("%d\n", __LINE__);	
	return ;
}

void C_GCdrCombination::SetFlags4TimeControl(S_CGCDR_C * pGCdrCg)
{
//  	pthread_mutex_lock( &g_mutex );
//	if (pGCdrCg == NULL)
	//	return false;
	for(map<CidAndGgsnAddrStruct, S_CGCDR_C*>::iterator map_it = map_GCdrCg.begin(); 
		map_it!=map_GCdrCg.end();map_it++)
	{
		map_it->second->bHandlingForTimeControl = false;
	}
	pGCdrCg->bHandlingForTimeControl = true;
	
//   	pthread_mutex_unlock( &g_mutex );	
}
void * ControlTimeLimit(void * pGCdrCom)
{
	for(;;)
	{
	  	pthread_mutex_lock( &g_mutex );
	  	//printf("After pthread_mutex_lock in ControlTimeLimit\n");
	  	C_GCdrCombination *cGCdrCom = static_cast<C_GCdrCombination*> (pGCdrCom); 
	  
		for(map<CidAndGgsnAddrStruct, S_CGCDR_C*>::iterator map_it = map_GCdrCg.begin(); 
			map_it!=map_GCdrCg.end();map_it++)
		{
			S_CGCDR_C * pGCdrCg = map_it->second;
			//?
			//if( ! pGCdrCg->bHandlingForTimeControl )
			//{
				//cout<<"%%%%%%%%%%%%%%%%%%%%%%%%%
				if ( cGCdrCom->IsTimeOfCombinedReached(pGCdrCg) )
				{
					//pthread_mutex_lock( &g_mutex );
					sprintf(pGCdrCg->ConsolidationResult, "%d", 
						C_GCdrCombination::CONSOLIDATION_RESULT_ABNORMAL);					cout<<"%%%%%%%%%%%%%%%%%%%% In ControlTimeLimit, before save"<<endl;
					cGCdrCom->SaveCdr(pGCdrCg);
					cout<<"%%%%%%%%%%%%%%%%%%%% In ControlTimeLimit, after save"<<endl;
					cGCdrCom->EraseGCdrCg(pGCdrCg);
					printf("After SaveCdr in ControlTimeLimit\n");
					//pthread_mutex_unlock( &g_mutex );
					//continue;	
					break;	
				}
			//}
		}
	
	   	pthread_mutex_unlock( &g_mutex );	
	   	//printf("After pthread_mutex_unlock in ControlTimeLimit\n");
	}
   	return NULL; //meaningless, a value must be returned to meet the requirement of this function!
}

//void CombinationForGcdr(C_GCdrCombination *pGCdrCom)
//void CombinationForGcdr(C_GCdrCombination *cGCdrCom)
void * CombinationForGcdr(void *pGCdrCom)
{
    C_GCdrCombination *cGCdrCom = static_cast<C_GCdrCombination*> (pGCdrCom);
    
    long lReturn;
    


	S_OGCDR2 *pGCdr = new S_OGCDR2;
	if (NULL == pGCdr)
	{
		//fprintf(fptr2,"No enough memory when allocating memory for S_R4OGCDR_C in CombinationForGcdr\n");
		if (NULL != cGCdrCom)
		{
			delete cGCdrCom;
			cGCdrCom = NULL;
		}
		WriteCombinedGCdrErrorInfoToFile("CombinationForGcdr:No enough memory when allocating memory for S_CGCDR_C");
		//exit(-1);
		return NULL;
	}

	
	
	int x = 0;
	for( ; ; )//while(x++<50)
	{

		//Set to "0" before read data?
		memset(pGCdr,0,sizeof(S_OGCDR2));
	        //lReturn = ShareMemory_GCDR.ReadData(pGCdr,sizeof(S_OGCDR2),PROGRAM_VACANCY);
	        unsigned long ulCounterFlag;
	        lReturn = cGCdrCom->ShareMemory_GCDR.ReadDataUpdate(pGCdr,sizeof(S_OGCDR2),ulCounterFlag);
	        
	        if(lReturn < 0)
	        {
		        //fprintf(fptr2, "Read AREA_GCDR Error! Return:%d\n",lReturn);;
			//delete pGCdr;
			//delete cGCdrCom;
			//exit(-1);
			WriteCombinedGCdrErrorInfoToFile("CombinationForGcdr:ShareMemory_GCDR.ReadDataUpdate Error");
			continue ;
	        }
	        else if(lReturn == 0)
	        {
		        //cout<<"AREA_GCDR"<<AREA_GCDR<<" is empty! read nothing! Return:"<<lReturn<<endl;
			//sleep(5);
			continue;
	        }    
	        else
	        {
			//fprintf(fptr2,"read AREA_GCDR Successful!\n");
			//fprintf(fptr2, "ChargingID:%s\n", pGCdr->ChargingID);
			//fprintf(fptr2, "GGSNAddress:%s\n", pGCdr->GGSNAddress);
			//fprintf(fptr2, "RecordSequenceNumber:%s\n", pGCdr->RecordSequenceNumber);
	
			cout<<"Origin Cdr:"<<endl;
			cout<<"ggsnaddr:"<<pGCdr->GGSNAddress<<endl;
			cout<<"ChargingID:"<<pGCdr->ChargingID<<endl;
			//cout<<"DataVolumeGPRSUp:"<<pGCdr->ListOfTrafficVolumn[2].DataVolumeGPRSUp<<endl;
			//cout<<"DataVolumeGPRSDown:"<<pGCdr->ListOfTrafficVolumn[3].DataVolumeGPRSDown<<endl;
			cout<<"SGSNIPAddress:"<<pGCdr->SGSNIPAddress<<endl;
			cout<<"CallDuration:"<<pGCdr->CallDuration<<endl;
			cout<<"RecordSequenceNumber:"<<pGCdr->RecordSequenceNumber<<endl;
			cout<<"RecordOpenTime:"<<pGCdr->RecordOpenTime<<endl;
			cout<<"TableName:"<<pGCdr->TableName<<endl;
			cout<<"SeqId:"<<pGCdr->SeqId<<endl;
			cout<<endl;
			if(strcmp("", pGCdr->ChargingID) == 0)
			{
				continue;
			}
	
	
			/* 创建GCDR话单 */
		//	S_OGCDR2 *pGCdr = (S_OGCDR2 *)cGCdrCom->GetCdr();
			S_OGCDR2 *pGCdrOld = pGCdr;
	
			/* 对话单进行简单地过滤处理 */
			S_CGCDR_C *pGCdrCg;
			pthread_mutex_lock( &g_mutex );//get the lock
			//printf("Before call GcdrSort in Combination\n");
			if (C_GCdrCombination::TRIGGER_TYPE == C_GCdrCombination::TRIGGER_TYPE_NORMAL )
			{
				pGCdrCg = cGCdrCom->GcdrSort(pGCdr, ulCounterFlag);
			}
			else
			{
				pGCdrCg = cGCdrCom->GcdrSort(pGCdr, ulCounterFlag, C_GCdrCombination::TRIGGER_TYPE_TIME);
			}
	
			//printf("Before call pthread_mutex_lock  001 in Combination\n");
  			//pthread_mutex_lock( &g_mutex );//get the lock
			//printf("After pthread_mutex_lock 001 in Combination\n");

			//if( ! pGCdrCg) {
			if( NULL == pGCdrCg) {
			//printf("Before call pthread_mutex_unlock  002 in Combination\n");
				pthread_mutex_unlock( &g_mutex );
			//printf("After call pthread_mutex_unlock  002 in Combination\n");
				continue;
			}
  			
			//printf("NULL != pGCdrCg\n");
			//Set time-control flags for all big CDRs
			//cGCdrCom->SetFlags4TimeControl(pGCdrCg);
	
			/* 判断是否是重复话单 */
			//fprintf(fptr2,"pGCdr->RecordSequenceNumber:%s\n",pGCdr->RecordSequenceNumber);
			//fprintf(fptr2,"pGCdrCg->RecordSequenceNumber:\n");
			if (cGCdrCom->IsRecSeqNumExisted(pGCdrCg->RecordSequenceNumber,
							pGCdr->RecordSequenceNumber))
			{
				printf("Repeat Cdr\n");
				string sUpdateSql;
				sUpdateSql.append("UPDATE ").append(pGCdr->TableName).append(" SET ")
     					.append("flagrepeat=").append("2").append(" where seqid='")
     					.append(pGCdr->SeqId).append("'");
     				cout<<"update sql:"<<sUpdateSql<<endl;
				//g_cdr.Update(sUpdateSql.c_str());


				string sUpdateSql1;
				sUpdateSql1.append("UPDATE ").append(pGCdr->TableName).append(" SET ")
     					.append("flagread=").append("2").append(" where seqid='")
     					.append(pGCdr->SeqId).append("'");
				cout<<"update sql:"<<sUpdateSql1<<endl;
				//g_cdr.Update(sUpdateSql1.c_str());

				//remove the cdr from shared memory
				pthread_mutex_unlock( &g_mutex );
				continue;
			}

			if( false == cGCdrCom->IsCallDurationCorrect(pGCdr->CallDuration))
			{
				printf("Error Cdr\n");
				printf("pGCdr->SGSNIPAddress is %s\n",
					pGCdr->SGSNIPAddress);
				printf("pGCdr->RecordSequenceNumber is %s\n",
					pGCdr->RecordSequenceNumber);
				printf("pGCdr->CallDuration is %s\n",
					pGCdr->CallDuration);
				//回写数据库

				string sUpdateSql;
				sUpdateSql.append("UPDATE ").append(pGCdr->TableName).append(" SET ")
     					.append("flagrepeat=").append("3").append(" where seqid='")
     					.append(pGCdr->SeqId).append("'");
				cout<<"update sql:"<<sUpdateSql<<endl;
				//g_cdr.Update(sUpdateSql.c_str());


				string sUpdateSql1;
				sUpdateSql1.append("UPDATE ").append(pGCdr->TableName).append(" SET ")
     					.append("flagread=").append("2").append(" where seqid='")
     					.append(pGCdr->SeqId).append("'");
				cout<<"update sql:"<<sUpdateSql1<<endl;
				//g_cdr.Update(sUpdateSql1.c_str());


				pthread_mutex_unlock( &g_mutex );
				continue;
			}

	
			//printf("Before pGCdrCg->NumOfCombined++ in Combination\n");
			pGCdrCg->NumOfCombined++;
	
			/* Diagnostics进行重新赋值 */
			//pGCdrCg->Diagnostics=pGCdr->Diagnostics;
			
			if ( cGCdrCom->IsRecSeqNumNull(pGCdrCg->RecordSequenceNumber) )
			{
				/* 将SGSN地址加入到Record Sequence Number链表头中 */
				cGCdrCom->RecordSgsnAddrToRecSeqNumListHead(pGCdrCg->RecordSequenceNumber,
									    pGCdr->GGSNAddress);
			}
			//printf("cGCdrCom->RecordRecSeqNumToList in Combination\n");
			//printf("%d\n", __LINE__);
			/* 设置待合并话单中的记录打开时间 */
			cGCdrCom->RecordRecSeqNumToList(pGCdrCg->RecordSequenceNumber, 
							pGCdr->RecordSequenceNumber);
	
			//printf("%d\n", __LINE__);
			long recOpenTime_cg = atol(pGCdrCg->RecordOpenTime);
			//printf("%d\n", __LINE__);
			long recOpenTime = atol(pGCdr->RecordOpenTime);
			//printf("%d\n", __LINE__);
			if ( recOpenTime_cg == 0 || recOpenTime < recOpenTime_cg )
			{
			//printf("%d\n", __LINE__);
				strncpy(pGCdrCg->RecordOpenTime, pGCdr->RecordOpenTime, sizeof(pGCdrCg->RecordOpenTime));
			}
			//printf("%d\n", __LINE__);
			/* 设置待合并话单中的DURATION数值 */
			long total_duration = atol(pGCdrCg->CallDuration);
			//printf("%d\n", __LINE__);
			long duration = atol(pGCdr->CallDuration);
			//printf("%d\n", __LINE__);
			total_duration += duration;
		//	int radix = 10;
			//lltostr(total_duration, pGCdrCg->CallDuration);
			//printf("%d\n", __LINE__);
			sprintf(pGCdrCg->CallDuration, "%d", total_duration);
			//printf("%d\n", __LINE__);
	
			/* 合并话单中的List of Traffic Data */
			//?
			long SortId = atol(pGCdr->RecordSequenceNumber);
			for(int cv = 0; cv < NUM_OF_VOL_NODE; cv++)
			{
				sprintf(pGCdr->ListOfTrafficVolumn[cv].SortId,
				//sprintf(pGCdrCg->ListOfTrafficDataVolumes[cv].SortId,
				       "%d",
				       SortId + cv);
				cGCdrCom->RecordListOfTrafficDataToList(pGCdrCg->ListOfTrafficDataVolumes, 
					pGCdr->ListOfTrafficVolumn[cv]);
			}
	
			//printf("%d\n", __LINE__);
			//printf("cGCdrCom->RecordCauseForRecClosingToList in Combination\n");
			/* 合并话单中的Cause for Record Closing */
			cGCdrCom->RecordCauseForRecClosingToList(pGCdrCg->CauseForRecClosing, 
								 pGCdr->CauseForRecClosing);
	
			//printf("%d\n", __LINE__);
			//printf("cGCdrCom->IsSgsnAddrExisted in Combination\n");
			/* 合并话单中SGSN地址信息 */
			/* 首先判断SGSN地址是否已经有记录 */
			if (! cGCdrCom->IsSgsnAddrExisted(pGCdrCg->SGSNIPAddress,
							  pGCdr->SGSNIPAddress))
			{
			//printf("%d\n", __LINE__);
				cGCdrCom->RecordSgsnAddrToList(pGCdrCg->SGSNIPAddress,
							       pGCdr->SGSNIPAddress);
			}
	
			//printf("%d\n", __LINE__);
			//printf("cGCdrCom->IsLocalRecSeqNumNull in Combination\n");
			/* 合并话单中的Local Record Sequence Number字段 */
			/* 首先判断话单中Local Record Sequence Number链表是否为空 */
			if ( cGCdrCom->IsLocalRecSeqNumNull(pGCdrCg->LocalRecordSequeceNumnber) )
			{
				//printf("%d\n", __LINE__);
			/* 将GGSN地址加入到Local Record Sequence Number链表头中 */
				cGCdrCom->RecordGgsnAddrToLocalRecSeqNumListHead(pGCdrCg->LocalRecordSequeceNumnber,
										 pGCdr->GGSNAddress);
			}
				//printf("%d\n", __LINE__);