drserver.cpp

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

#include <STAR/CORBA.h>

#include <STAR/AFLSInitializer.h>
#include <STAR/Properties.h>
#include <MTL/MTL.h>
#include <STAR/ATF_IIOP.h>
#include <STAR/AssistAdaptorManager.h>

#include "DRServer_impl.h"

#include <checkConfig.h>
#include <STAR/Properties.h>

#ifdef HAS_DEBUG_LEVEL
#include "LogClient.h"
LogClient * g_pLogWriter = NULL;
#endif

#ifdef __UNIX__
#include <sys/time.h>
#endif

#ifdef _WIN32
#include <time.h>
#endif

#include <STAR/ATF_IIOP.h>
#include <argParser.h>

//SuLiang modify for Interceptor **************
#include "ServerInterceptor.h"
#include "ManageHelper.h"
//SuLiang modify for Interceptor **************

#ifdef HAVE_FSTREAM
#   include <fstream>
#else
#   include <fstream.h>
#endif

#ifdef HAVE_STD_IOSTREAM
using namespace std;
#endif

CORBA::ORB_var orb;

IDRecoveryServer_var * cssImpl;

char * sAflsConfig;

int g_nDebugLevel = 0;



//added by baixiaobo 20041227
class CloseCallback_impl : public ATF::CloseCallback
{
	TimeOutThreadHandle m_TimeThread;
public:
	CloseCallback_impl(TimeOutThreadHandle ttTimeThread)
	{
		m_TimeThread = ttTimeThread;
	}

    virtual void close_callback(ATF::TransportInfo_ptr transport_info)//连接断开时的回调函数
    {
        ATF::IIOP::TransportInfo_var iiopInfo =
            ATF::IIOP::TransportInfo::_narrow(transport_info);
	char ipport[50]="";
        if(!CORBA::is_nil(iiopInfo))
        {
            ATF::IIOP::InetAddr_var remoteAddr = iiopInfo -> remote_addr();
            CORBA::UShort remotePort = iiopInfo -> remote_port();


		    sprintf(ipport,"%d.%d.%d.%d:%d",(int)remoteAddr[0],
						(int)remoteAddr[1],
						(int)remoteAddr[2],
						(int)remoteAddr[3],remotePort);
		    m_TimeThread->delNodebyIp(ipport);
        }
    }
};


class AccCallback_impl : public ATF::AccCallback
                         //public CORBA::LocalObject
{
	TimeOutThreadHandle m_TimeThread;
public:

	AccCallback_impl(TimeOutThreadHandle ttTimeThread)
	{
		m_TimeThread = ttTimeThread;
	}
    virtual void acc_callback(ATF::TransportInfo_ptr transport_info)//接受连接时的回调函数
    {
        ATF::IIOP::TransportInfo_var iiopInfo =
            ATF::IIOP::TransportInfo::_narrow(transport_info);

        if(!CORBA::is_nil(iiopInfo))
        {
            //
            // Add the close callback object
            //

            ATF::IIOP::InetAddr_var remoteAddr = iiopInfo -> remote_addr();
            CORBA::UShort remotePort = iiopInfo -> remote_port();

            //cout << "get connection from: "
            //     << (int)remoteAddr[0] << '.' << (int)remoteAddr[1] << '.'
			//     << (int)remoteAddr[2] << '.' << (int)remoteAddr[3]
			//     << ":" << remotePort << endl;
			//cout<<"get connection from outer"<<endl;
        }
    }
};

//end added by baixiaobo 20041227

int
//run(CORBA::ORB_ptr orb, int argc, char* argv[])
run(int argc, char* argv[])
{
	//create afls object
	CORBA::Object_var obj	=
	orb->string_to_object(sAflsConfig);
	
	Afls_var afls =Afls::_narrow(obj);
	assert(!CORBA::is_nil(afls));

	//
    // Resolve Root POA
    //
    CORBA::Object_var poaObj = orb -> resolve_initial_references("RootPOA");
    PortableServer::POA_var rootPOA = PortableServer::POA::_narrow(poaObj);
    
    //
    // Get a reference to the POA manager
    //
    PortableServer::POAManager_var manager = rootPOA -> the_POAManager();
    
    
    CORBA::PolicyList policies;
    policies.length(2);
    policies[0] =
        rootPOA -> create_id_assignment_policy(PortableServer::USER_ID);
    policies[1] =
        rootPOA -> create_lifespan_policy(PortableServer::PERSISTENT);
 
    PortableServer::POA_var poa;
    try
    {
        poa = rootPOA -> create_POA("DRPOA", manager, policies);
    }
    catch(const PortableServer::POA::AdapterAlreadyExists&)
    {
    }
    assert(!CORBA::is_nil(poa));
    	
	//SuLiang modify for Interceptor **************
	ManageHelper::createMonitor(orb, rootPOA, manager);
	ManageHelper::setAfls(afls);
	//SuLiang modify for Interceptor **************
    
	//---------------------------------------------------
	STARPortableServer::POAManager_var starManager =
        STARPortableServer::POAManager::_narrow(manager);//必须narrow
       	obj = orb -> resolve_initial_references("ATFCurrent");
   	ATF::Current_var atfCurrent = ATF::Current::_narrow(obj);
   	assert(!CORBA::is_nil(atfCurrent));
   	
	IDRecoveryServer_impl* cssImpl = new IDRecoveryServer_impl(rootPOA,atfCurrent);
	if (cssImpl == NULL)
	{
		cerr<<"error in new IDRecoveryServer!"<<endl;	
		return -1;
	}
	PortableServer::ServantBase_var servant = cssImpl;

	IDRecoveryServer_var css = cssImpl -> _this();

	LoadProvider_var lpv = LoadProvider_init(orb);

	afls->register_obj("DRServer.RAO",css.in(), lpv.in());

	cout << "DRServer is available!" << endl;
   	
   	ATF::AcceptorSeq_var acceptors = starManager -> get_acceptors();

	for(CORBA::ULong i = 0 ; i < acceptors -> length() ; i++)
	{
	    ATF::AcceptorInfo_var info = acceptors[i] -> get_info();
	    ATF::IIOP::AcceptorInfo_var iiopInfo =
	        ATF::IIOP::AcceptorInfo::_narrow(info);
	
	    if(!CORBA::is_nil(iiopInfo))
	    {
	        CORBA::StringSeq_var hosts = iiopInfo -> hosts();
	        CORBA::UShort port = iiopInfo -> port();
	
	        //cout << "*** Running on host `" << hosts[0]
	        //     << "', port number " << port << endl;
	        ATF::AccCallback_var acceptorCB = new AccCallback_impl(cssImpl->getTimeOutThread());
	    	    iiopInfo -> add_acc_callback(acceptorCB);
	    }
	}
	//---------------------------------------------

	manager -> activate();

	#ifdef HAS_DEBUG_LEVEL
	STARTINFO(1,"start normally")
	#endif

	orb -> run();
	
	return 0;
}

int
main(int argc, char* argv[], char*[])
{

	int status = 0;

	CConfCheck AflsConfCheck("OUTER");

	if (AflsConfCheck.nCheckConf() != 0)
	return -2;

	sAflsConfig = AflsConfCheck.sGetAflsConf();

	AFLS_init();	

	//SuLiang modify for Interceptor **************
	const char * monitoredOperations[] = {
	    "ResGet"
	};
	
	//operationNum是上面monitoredOperations的个数
	int operationNum = 1;
	ServerInterceptor::init(operationNum, monitoredOperations);	//加入截获器
	//SuLiang modify for Interceptor **************

	//CORBA::ORB_var orb;
	MTLInitialize initialize;
	try
	{
		STAR::Properties_var props = STAR::Properties::getDefaultProperties();
		STAR::ParseArgs(argc, argv, props);
		
		orb = STARCORBA::ORB_init(argc, argv, props);
		
		//status = run(orb, argc, argv);

		//--初始化日志对象
		#ifdef HAS_DEBUG_LEVEL
		
		g_pLogWriter = new LogClient(argv[0],orb);
		if (g_pLogWriter == NULL)
		{
			cerr <<"cannot new LogClient" <<endl;
			exit(-1);
		}
		#endif	

		status = run(argc, argv);
	}
	catch(const CORBA::Exception& ex)
	{
		cerr << ex << endl;
		
		#ifdef HAS_DEBUG_LEVEL
		STARTINFO(4,"exit abnormally")
		#endif
		
		status = -1;
	}
	if(!CORBA::is_nil(orb))
	{
		try
		{
    		orb -> destroy();
		}
		catch(const CORBA::Exception& ex)
		{
		    cerr << ex << endl;
		    status = -1;
		}
	}
	
	return status;
}