datarecovery_impl.cpp

oracle下调用数据泵接口进行远程数据导出和导入（数据恢复）的CORBA服务

//      DataRecovery_impl.cpp
//      DataRecovery_impl.h的实现
//
#include "DataRecovery_impl.h"
#include "table_struct.h"
#include <string.h>

#include <fstream>

#define LOCAL_SERVER_PORT 1500
#define MAX_MSG 3000
char buf[MAX_MSG];

#define  DEFAULT_FILE_SIZE   100    //单位为M
#define  DEFAULT_TIME_OUT    300    //单位为秒
#define  NOT_FINISH_GETING_RESULT 101  //代表还需要继续取结果
#define  FINISH_GETING_RESULT     102  //代表已经取完数据了
#define  NOT_ENOUGH_ROOM_FOR_RECOVERYFILE  -103    //用户提供的供恢复的文件大小
#define  ERR_SQSERVER_QUERY  	-2
#define  RECOVERY_FILE_PATH   "/home/oracle/pdb_rx/DataRecovery/DataRecovery/RECOVERED_DATA"

DataRecovery_impl::DataRecovery_impl(PortableServer::POA_ptr poa, TimeOutThreadHandle ttTimeThread, CORBA::Short timeOut)
	 :m_poa(PortableServer::POA::_duplicate(poa))
{
//	//^
	my_id = 0;         //北京
	
		//注意，还要控制线程的stop时机
	m_pRecoveryThread = NULL;
	
	m_pDdmServer = new CDataDistributionManager();
	
	memset(m_blockId, 0, DXX_BLOCK_ID_LENGTH);
	memset(m_strProvinceCode, 0, MAX_PROVINCE_ID_STR_LENGTH);
	memset(m_TableName, 0, MAX_TABLE_NAME_LENGTH);
	memset(m_strStatement, 0, MAX_STATEMENT_LEN);
	
	//默认超时时间
	m_TimeThread = ttTimeThread;
	m_maxIdleTime = timeOut;
	
	int argc = 1;
	char * argv = "" ;
	
	//这句话有问题，会报：
	//non-compliant application detected:
	//object was deleted without an object reference count of zero
	//assertion failed: ref_ == 0
	//file /home/oracle/starbus/release/include/STAR/Common.h, line 151
	//if (AflsConfCheck.nCheckConf() != SUCCESS)
		//return ERR_SQSERVER_COMMON;
	CConfCheck AflsConfCheck("OUTER");
	AflsConfCheck.nCheckConf();

	m_AflsConfig = AflsConfCheck.sGetAflsConf();

	m_orb = CORBA::ORB_init(argc, &argv);

	aflsobj = m_orb->string_to_object(m_AflsConfig);
	cout << "655" << endl;
	afls	=Afls::_narrow(aflsobj);
	cout << "666" << endl;
	assert(!CORBA::is_nil(afls));
	cout << "677" << endl;	
}

DataRecovery_impl::~DataRecovery_impl()
{
        //^仅仅在这里释放，够不够？
        if(m_pDdmServer!=NULL)
              delete m_pDdmServer;
		
		m_TimeThread -> delNodebyPointer(this);
        
        m_result->destroy();
}
	    
PortableServer::POA_ptr  DataRecovery_impl::_default_POA()
{
      return PortableServer::POA::_duplicate(m_poa);    
}

CORBA::Short DataRecovery_impl::loseData(const char* blockId,
						const char* beginTime,
                          const char* endTime,
                           const char* strProvinceCode,
                              const char* strTableName)throw(CORBA::SystemException)
{			 
			 //报警
	     int rc;
	     int i;
	     //同步？
       	 MTLSynchronized syn(*this);
	     
	     strcpy(m_blockId , blockId);
	     strcpy(m_time.BeginTime , beginTime);
	     strcpy(m_time.EndTime   , endTime);
	     strcpy(m_strProvinceCode ,strProvinceCode);
	     strcpy(m_TableName , strTableName);
	     //报警
	     //^ 暂定为向屏幕输出警告
	     cout<<"WARNING:: Data Lost!"<<endl;
	     cout<<"--------------------"<<endl;
	     cout<<"BLOCK_ID: "<<m_blockId<<endl;
	     //cout<<"TIME INTERVAL: "<<m_beginTime<<" ~ "<< m_endTime<<endl;
	     cout<<"THE LOST DXX FROM : "<<m_strProvinceCode<<endl;
	     cout<<"THE LOST DXX WILL BE LOADED INTO TABLE : "<<m_TableName<<endl;
	     cout<<"--------------------"<<endl;
	     
	     //启动线程，执行具体的数据恢复过程

//			 if (m_pRecoveryThread->nInit(this) != SUCCESS)     	   //这里是不是这么使用this的呢？
//			 		return ERR_DRecovery_NEW_THREAD;

	  if ( strcmp("DZFX_DXX", strTableName) == 0 || strcmp("dzfx_dxx", strTableName) == 0 )
		{
				m_nTableType = 1;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  
	  else if ( strcmp("DZFD_DXX", strTableName) == 0 || strcmp("dzfd_dxx", strTableName) == 0 )
		{
				m_nTableType = 2;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  
	  else if ( strcmp("NRFD_DXX", strTableName) == 0 || strcmp("nrfd_dxx", strTableName) == 0 )
		{
				m_nTableType = 3;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  
	  else if ( strcmp("DZCT_DXX", strTableName) == 0 || strcmp("dzct_dxx", strTableName) == 0 )
		{
				m_nTableType = 4;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  
	  else if ( strcmp("NRCT_DXX", strTableName) == 0 || strcmp("nrct_dxx", strTableName) == 0 )
		{
				m_nTableType = 5;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  
	  else if ( strcmp("NRJC_DXX", strTableName) == 0 || strcmp("nrjc_dxx", strTableName) == 0 )
		{
				m_nTableType = 6;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  
	  else if ( strcmp("DZJS_DXX", strTableName) == 0 || strcmp("dzjs_dxx", strTableName) == 0 )
		{
				m_nTableType = 7;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  
	  else if ( strcmp("NRJS_DXX", strTableName) == 0 || strcmp("nrjs_dxx", strTableName) == 0 )
		{
				m_nTableType = 8;
				m_nDXX_TABLE_FIELD = 25;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nCOMMON_DXX[i];				
		}	  

		else if( strcmp("JM_DXX", strTableName) == 0 || strcmp("jm_dxx", strTableName) == 0 )
		{		
				m_nTableType = 9;				
				m_nDXX_TABLE_FIELD = 23;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nJM_DXX[i];
		}
		else if( strcmp("BF_DXX", strTableName) == 0 || strcmp("bf_dxx", strTableName) == 0 )
		{
				m_nTableType = 10;
				m_nDXX_TABLE_FIELD = 17;
				for (i=0; i<m_nDXX_TABLE_FIELD; i++)
						m_DXX_TABLE[i] = nBF_DXX[i];
		}
		//这里得到的是基本记录本身的长度值
		for (i=0;i<m_nDXX_TABLE_FIELD;i++)
				m_nSDataLen += (m_DXX_TABLE[i]-1);            //由于在table_struct中，都算了一位字符串结束符，所以要减1
		m_nSDataLen ++;                                   //最后还得把整行记录的结束符给算上
		cout << "m_nSDataLen= "<< m_nSDataLen <<endl;	   


//	     m_pRecoveryThread = new CRecoveryThread(this);
//			 m_pRecoveryThread->start();
	     
}


//根据接口参数形成select语句
int DataRecovery_impl::nFormSelectStatement(const char* blockId,
                               								 const TimeInterval timeInter,
                               								 const char* strProvinceCode,
                               								 const char* TableName)
{
			int i,j;
			int rc = SUCCESS;
			int strLen = 0;
			int provinceCount = 0;
			int provinceID[31];
			int tempCount = 0;
			char tempProvince[6];
			provinceCount = 0;
			strLen = strlen(strProvinceCode);
  		for (i=0; i<=strLen; i++)         //扫描每一个字符
  		{
  				tempCount = 0;
  				memset (tempProvince, 0, 6);
 
  				if (strProvinceCode[i] != ' ')   //跳过空格
  				{   
  						for (j=0; j<6+1; j++)        //一定要扫描到","，所以至少要走6次(5个number加1个",")，为了纠错,
							{                            //即走了6步却还没有碰到","的情况，再多走一次，以便于判断
  								if (strProvinceCode[i+j] == '\0')    //如果已经到结束符，跳出循环。而且一个provinceID中间不许有空格！
  										break;
  								if (strProvinceCode[i+j] != '\,')    
  										tempCount ++;
 									else break;
 							}
 							if (tempCount == 7)       //如果走了7次，代表正常情况下没有碰到","，可能是输入的字符串有问题                          
 							{
 									rc = ERR_DRecovery_PROVICE_ID;
 									return rc;
 							}
 							
 							strncpy(tempProvince, strProvinceCode+i, tempCount);
 							provinceID[provinceCount]= atoi(tempProvince);
  						provinceCount ++;

  						i = i+j;                 //注意，在if结束之后，i还会自动加1
  				}
  		}

			//如果字符串为空呢？怎么判断？
			
			int templen;
			sprintf(m_strStatement, "select * from %s where ", TableName);
			templen = strlen(m_strStatement);
			if (strcmp(blockId, "00000000000000")!=0 )      //若dxx_block_id为全零（14个0），则不需要考虑dxx_block_id的内容
			{
					sprintf(m_strStatement+templen, "dxx_block_id = \'%s\' and ", blockId);
					templen = strlen(m_strStatement);
			}
			sprintf(m_strStatement+templen, "fssj >= to_date(\'%s\', \'yyyy-mm-dd hh24:mi:ss\') ",timeInter.BeginTime);
			templen = strlen(m_strStatement); 
			sprintf(m_strStatement+templen, "and fssj <= to_date(\'%s\', \'yyyy-mm-dd hh24:mi:ss\') and ",timeInter.EndTime);
			templen = strlen(m_strStatement);
			sprintf(m_strStatement+templen, "( ");
			
			if (provinceCount == 1)            
			{
				  templen = strlen(m_strStatement);
				  sprintf(m_strStatement+templen, "dxzx_id = %d )",provinceID[provinceCount-1]);
			}
			else if (provinceCount > 1)
			{
				  for (i=0; i<provinceCount-1; i++)			//provinceCount如果为i，则province应该有i+1个！
				  {
							templen = strlen(m_strStatement);
					    sprintf(m_strStatement+templen, "dxzx_id = %d or ",provinceID[i]);
				  }
				  templen = strlen(m_strStatement);
				  sprintf(m_strStatement+templen, "dxzx_id = %d )",provinceID[i]);
			}
			else rc = ERR_DRecovery_PROVICE_ID;

      return rc;
}

CORBA::Short DataRecovery_impl::nBeginGetData(const char* blockId,
							const char* beginTime,
                             const char* endTime,
                              const char* strProvinceCode,
                               const char* strTableName,
                                const char* strFileName,
                                 const char* strPathName,
                                  CORBA::Short nFileSize,
                                   CORBA::Short timeOut,
                                   	CORBA::Short nTableType,
                                   	 CORBA::Long_out lRealFileSize,
                                      CORBA::Short_out nExpdTimes)throw(CORBA::SystemException)
{
	  //查数据源
	  //由于nBeginGetData是给其他服务方调用的接口，所以不应该将其调用信息变为内部的存储变量
	  //结构内部应该维持的变量因该是我自身需要恢复时需要的信息，也就是loseData接口所传进来的信息
    //strcpy(m_blockId , blockId);
    //strcpy(m_strProvinceCode ,strProvinceCode);
    //strcpy(m_TableName , strTableName);	  
	UpdateTime();
    //设置timeOut时间
    if (timeOut >0)
    	m_maxIdleTime = timeOut;
    		
    int rc;
    int i;
    TimeInterval time;
    strcpy(time.BeginTime , beginTime);
    strcpy(time.EndTime   , endTime);
		char ExpdpStatament [MAX_STATEMENT_LEN];    
    int DXX_TABLE[25];
    int nDXX_TABLE_FIELD;
    int nSDataLen=0;
    Records_var values;
	

    
    //不同类型的表处理方式不同
		switch (nTableType)
		{
				case 1:         //日志表（短信息表）

						//构造select语句
						rc = nFormSelectStatement(blockId, time, strProvinceCode, strTableName);
						//cout << "statement =" << m_strStatement << endl;		
						if (rc != SUCCESS)
						{
								cerr << "error in form select statement\n error code is:" << rc <<endl;
								return rc;
						}
				    try
					  {				
								if (CORBA::is_nil(m_SServer))
								{
									// Get "SServer" object
								    CORBA::Object_var ssobj =afls->alloc_obj("SServer.RAO");
								    assert(!CORBA::is_nil(ssobj));		   	    	
								    m_SServer = IQueryServer::_narrow(ssobj);
								    assert(!CORBA::is_nil(m_SServer));
								}
						}
						catch(const CORBA::Exception& ex)
						{
								m_SServer =NULL;
								cout <<"perhaps QueryServer have breakdown !"<<endl<<"please  check it "<<endl;
						}		
						
						//获得查询服务的引用，进行查询
						try
						{		
								m_result = m_SServer->ResConnect("rao",0);	
								CORBA::String_var 	strErrMsg;
								if(CORBA::is_nil(m_result))
								{
										cerr<<endl;
										cerr<< "Error to exec statment:" << m_strStatement << endl;
										cerr<<"please type help to get more help information\n";
										return ERR_SQSERVER_QUERY;
								}
								rc = m_result->nExec(m_strStatement,0,0);
								if (m_result->bIsBad())
								{
										cout <<"error code : "<< rc <<endl;
										cout <<"error in nSelect() " <<endl;
										CORBA::String_var str=m_result->strErrorMessage(rc);
										cout <<str.in()<<endl;