datarecovery_impl.cpp

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

										m_result->destroy();
										return -1;
								}
								
								m_nFieldCount = m_result -> nGetFieldCount();
						}
						catch(const CORBA::Exception& ex)
						{ 
								cout<<"CORBA Exception, error code is :"<<endl;
								cout << ex <<endl;
						}
						
					break;
					
				case 2:				//配置表
						//对于配置表而言，如果导出的数据大于nFileSize的大小，就要报错
						//执行expdp语句，其中要将strTableName和strPathName的信息用好。
						//system ();
						//等system系统调用结束之后，拿到刚刚生成的.dmp文件的大小
						struct stat FileStat;

						sprintf(ExpdpStatament, "expdp pdb_cuser/dbc DIRECTORY=dump_dir DUMPFILE=tab.dmp ESTIMATE=STATISTICS CONTENT=DATA_ONLY TABLES=%s",  strTableName);
						rc = system ("rm /home/oracle/pdb/DataRecovery/EXPDATA/tab.tmp" );
						rc = system (ExpdpStatament);
						//将文件名和路径名拼在一起。
						//sprintf(m_strPathName, "/%s", strFileName);
						stat ("/home/oracle/pdb/DataRecovery/EXPDATA/tab.tmp",&FileStat);
						
						lRealFileSize = FileStat.st_size;         //nRealFileSize这个东西需要传到CBJ中，以便于CBJ返回给用户
																											//让用户能够知道应该设定nFileSize为多大才符合要求。
						if (nFileSize * 1024 * 1024 < lRealFileSize)   //nFileSize是以MB为单位。而lRealFileSize是以B为单位
								rc = 	NOT_ENOUGH_ROOM_FOR_RECOVERYFILE;
						else
						{
							rc =  SUCCESS;
						}
					break;
				case 3:       //统计表
					
					//注意，这里有个格式转换的问题。
					lRealFileSize = (fEstimateFileSize(strTableName));
					//nExpdTimes = lRealFileSize / nFileSize + 1;
					cout << "lRealFileSize =" << lRealFileSize << endl;
					char strStatement[MAX_STATEMENT_LEN];
					sprintf(strStatement,"select MIN(SJ), MAX(SJ)-MIN(SJ) from %s", strTableName);
					cout << "strStatement =" << strStatement << endl;
					float fInterval;
					cout << "11" << endl;
					try
					{				
							if (CORBA::is_nil(m_SServer))
							{
								// Get "SServer" object
							    cout << "111" << endl;
							    CORBA::Object_var ssobj =afls->alloc_obj("SServer.RAO");
							    cout << "112" << endl;	
							    assert(!CORBA::is_nil(ssobj));		   	    	
							    m_SServer = IQueryServer::_narrow(ssobj);
							    cout << "113" << endl;	
							    assert(!CORBA::is_nil(m_SServer));
							}
					}
					catch(const CORBA::Exception& ex)
					{
							m_SServer =NULL;
							cout <<"perhaps QueryServer have breakdown !"<<endl<<"please  check it "<<endl;
					}		
					cout << "22" << 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(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;
									m_result->destroy();
									return -1;
							}							
							rc = m_result->nGetRecords(1, values);
							//values[0][0]
							strcpy (m_MinDate, values[0][0]);
							fInterval = atof(values[0][1]);
							
							//这里缩小了每次执行的间隔
							m_nInterval = fInterval / (1 + lRealFileSize / (nFileSize*1024*1024));
							
							if (m_nInterval < 1)
									m_nInterval=1;
							
							if (lRealFileSize <= nFileSize*1024*1024)
									nExpdTimes = 1;
							else
								  nExpdTimes = fInterval / (fInterval / (1 + lRealFileSize / (nFileSize*1024*1024)));

							cout << "nExpdTimes=" << nExpdTimes<< endl;
							
					}
					catch(const CORBA::Exception& ex)
					{ 
							cout<<"CORBA Exception, error code is :"<<endl;
							cout << ex <<endl;
					}					
					cout << "33" << endl;
					break;

				default:
					cerr << "Table dis type error"<< endl;
					return -1;
			
		}

		
		return rc;
}

//在服务方的GetNextData只是负责1次传输过程
CORBA::Short DataRecovery_impl::nGetNextData(Records_out values)throw(CORBA::SystemException)
{
		UpdateTime();	    
	    //
		int 	nRecordCount = 0;
		int   rc;
		//cout << "hi" << endl;
		
		//作为out类型的参数，一定要先分配内存！
		values = new Records;
		values -> length(0);		

		if(!m_result->bIsEOF())
		{
				//cout << "en1" << endl;
				//对于out类型的参数，一定要先分配内存！
				rc = m_result->nGetRecords(100,values);
				if (rc != SUCCESS)
				{
						cerr << "nGetRecords error!" << endl;
						return -1;
				}
				if (values -> length() < 100 )  //本次数据取得的结果小于100，说明是最后一次拿数据了
						rc = FINISH_GETING_RESULT;
				else rc = NOT_FINISH_GETING_RESULT;
		}
		else  //对于out类型的参数，一定要先分配内存！，如果是走这里的话，就没有分配内存，当函数体结束之后，starbus自动将values析构，
		      //肯定要报错！
		{
				rc = IS_EOF;
				//cout << "en2" << endl;
	  }
	  //测试时使用
//	  if (rc == 0)
//	  		nLoadIn();
	  
	  //cout << "rc=" << rc <<endl;
		return rc;
}
    
CORBA::Short DataRecovery_impl::nCancel()throw(CORBA::SystemException)
{
	     //^
	     UpdateTime();
}
   
char* DataRecovery_impl::strErrorMessage(CORBA::Short nErrorId) throw(CORBA::SystemException)
{
//	UpdateTime();
//	NONIDLMETHODENTER("strErrorMessage()",nErrorId,1)
	CORBA::String_var str;
	switch (nErrorId)
	{
		case ERR_SQSERVER_COMMON:
			str = CORBA::string_dup("common error:can not know which type error.");
			break;
		case ERR_SQSERVER_GET_CONNPOOL:
			str = CORBA::string_dup("constructor the connectionpool error:parameter error or not enough memory.");
			break;
		case ERR_SQSERVER_GET_DIC:
			str = CORBA::string_dup("constructor the dictionary error:not enough memory.");
			break;
		//case ERR_SQSERVER_OPEN_DB:
		//	str = CORBA::string_dup("constructor the dictionary error or constructor the connectionpool error");
		//	break;	
		case ERR_SQSERVER_NOTSUPPORTED_FUNCTION:
			str = CORBA::string_dup("the function is not supported now.");
			break;
		case ERR_SQSERVER_OPENFILEINDATAQUERY:
			str = CORBA::string_dup("cannot open file in DataQurey module.");
			break;
		case ERR_SQSERVER_GETRESULTLIST	:
			str = CORBA::string_dup("can not get result list from timeoutthread: narrow() error");
			break;
		case ERR_SQSERVER_GET_CONN_FROM_POOL:
			str = CORBA::string_dup("can not get connection from connectionpool:input parameter error or connection is full");
			break;
		case ERR_SQSERVER_NEW_DQS:
			str = CORBA::string_dup("can not initialize the DataQuery object:not enough memory");
			break;
		case ERR_SQSERVER_NEW_THREAD:
			str = CORBA::string_dup("can not initialize thread");
			break;
		case ERR_SQSERVER_MTL_Exception:
			str = CORBA::string_dup("MTL thread exception");
			break;
		case ERR_SQSERVER_GET_DOMAIN:
			str = CORBA::string_dup("get domain from dictionary error");
			break;
		case ERR_SQSERVER_DOMAIN_INFO:
			str = CORBA::string_dup("get domain from dictionary error");
			break;
		case ERR_SQSERVER_GET_DISNODE:
			str = CORBA::string_dup("get distribute node from dictionary error");
			break;
		case ERR_SQSERVER_GET_DIS_TYPE:
			str = CORBA::string_dup("get distribute type from dictionary error");
			break;
		case ERR_SQSERVER_LOAD_DIC_TABLE:
			str = CORBA::string_dup("read dictionary table from db error");
			break;
		case ERR_SQSERVER_DOMAIN_DISMATCH:
			str = CORBA::string_dup("two or more table in a statement that is not belong to the same domain");
			break;
		case ERR_SQSERVER_NOT_ENOUGH_MEMORY:
			str = CORBA::string_dup("not enough memory");
			break;
		case ERR_SQSERVER_NO_AVAILABLE_DB:
			str = CORBA::string_dup("not available connection to Database");
			break;
		case ERR_SQSERVER_OCI:
			str = CORBA::string_dup("OCI common error");
			break;
		case NODATA:
			str = CORBA::string_dup("can not get more data from DB");
			break;
		case IS_EOF:
			str = CORBA::string_dup("the cursor is at the end of record list");
			break;		
		case IS_BOF:
			str = CORBA::string_dup("the cursor is at the top of record list");
			break;
		case ERR_SQSERVER_OCIHandleAlloc:
			str = CORBA::string_dup("OCIHandleAlloc statement execute error");
			break;
		case ERR_SQSERVER_OCIStmtPrepare:
			str = CORBA::string_dup("OCIStmtPrepare statement execute error");
			break;
//		case ERR_SQSERVER_OCIStmtExecute:
//			str = CORBA::string_dup(m_OCIErrMsg);
			break;
		case ERR_SQSERVER_SYNTAX:
			str = CORBA::string_dup("query statement\'s syntax error");
			break;
		case ERR_SQSERVER_DDL:
			str = CORBA::string_dup("DDL statement\'s syntax error");
			break;
		case ERR_SQSERVER_CORBA:
			str = CORBA::string_dup("CORBA error");
			break;
		case ERR_SQSERVER_NOT_SUPPORT_SQL:
			str = CORBA::string_dup("This is a not support statement");
			break;
		case ERR_SQSERVER_MIN_MAX:
			str = CORBA::string_dup("Error in min and max function,could not comare value");
			break;
		case ERR_SQSERVER_GET_VALUE_BY_INDEX:
			str = CORBA::string_dup("can not get value by the index which is user given");
			break;
		case ERR_SQSERVER_CONFLICT:
			str = CORBA::string_dup("this SQL has some field conflict !");
			break;
		case ERR_SQSERVER_FIELD_NAME:
			str = CORBA::string_dup("no  field in the table !");
			break;
//		case ERR_SQSERVER_NEW_CRESULT:
//			str = CORBA::string_dup("error in call new methods to create IResult, maybe there is no enough memory");
//			break;
		case ERR_SQSERVER_INVALID_RESULT:
			str = CORBA::string_dup("The IResult object is invalid");
			break;
		case ERR_SQSERVER_TABLE_NOT_EXIST:
			str = CORBA::string_dup("this table dose not exist");
			break;
		case ERR_SQSERVER_FIELD_NUM:
			str = CORBA::string_dup("the field index beyond range");
			break;
		case ERR_SQSERVER_PARAM:
			str = CORBA::string_dup("dictionary initialized error in table name or field name or error db connect param.");
			break;
		case ERR_SQSERVER_NOT_EXIST:
			str = CORBA::string_dup("domain or table does not exist in data dictionary");
			break;
		//Top-k Query
		case ERR_SQSERVER_WRITEFILEFAILED:
			str = CORBA::string_dup("cannot write the temp file");
			break;
		case ERR_SQSERVER_DIRECTPATHLOADFAILED:
			str = CORBA::string_dup("load data by OCI DirectPathLoad failed");
			break;
		//Top-k Query
		default:
			str = CORBA::string_dup("no such error code");
			break;
	}
	//NONIDLMETHODLEAVE("strErrorMessage()")
	return CORBA::string_dup(str);
}

//--------------以下函数都为非接口函数------------------
//判断是否需要进行恢复操作
int DataRecovery_impl::nNeedRecovery()
{
		int rc = 1;
		return rc;
}

int DataRecovery_impl::nRecoveryProcess()
{
		int rc;
		int i,j,nRecordCount;
		int SourceLocationID = -1;
//		SourceLocationID = m_pDdmServer -> getDataSource(m_blockId,
//																										 m_time,  
//																										 m_strProvinceCode,	
//																										 m_TableName); 
    //测试时使用
    SourceLocationID = 0;
    
    switch(SourceLocationID)
    {
			 //BJ本地是不能够这样恢复的，这里是测试时用的
   	   case 0:  //北京
//   	   	         CDataRecovery   bjSrv = new CDataRecovery();
//   	   	         rr = bjSrv->nBeginGetData(m_blockId,ti,m_strProvinceCode,m_TableName, \
//   	   	                                    m_blockId,"/LostData/",DEFAULT_FILE_SIZE,DEFAULT_TIME_OUT);
			 
// 	  	 rc = nBeginGetData( m_blockId,
//								            m_time.BeginTime,
//                             m_time.EndTime,
//                              m_strProvinceCode,
//                               m_TableName,
//                                "",
//                                 "",
//                                  DEFAULT_FILE_SIZE,
//                                   DEFAULT_TIME_OUT );
// 
//      if (rc == SUCCESS)
//      {
//      		while (rc == SUCCESS)
//      		{
//      				rc = nGetNextData(test_values);
//      				if (rc == SUCCESS)
//      				{
//		      				for (i=0; i<test_values->length(); i++)