TarsCpp/unit-test/rpc/test_registry.cpp


#include "hello_test.h"
#include "servant/CommunicatorEpoll.h"
#include "servant/ObjectProxy.h"
#include "server/framework/DbHandle.h"
#include "server/FrameworkServer.h"
#include "QueryF.h"

TEST_F(HelloTest, registryQuery)
{
	shared_ptr<Communicator> c = getCommunicator();

	TC_Config conf = CLIENT_CONFIG();

	c->setProperty(conf);

	TC_Config fconf = FRAMEWORK_CONFIG();

	CDbHandle::cleanEndPoint();
	CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9990, 1);
	CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9991, 1);

	FrameworkServer fs;
	startServer(fs, FRAMEWORK_CONFIG());

	string obj = getObj(fconf, "RegistryAdapter");

	QueryFPrx qPrx = c->stringToProxy<QueryFPrx>(obj);

	{
		std::vector<EndpointF> endpoints = qPrx->findObjectById("TestApp.RpcServer.HelloObj");

		ASSERT_TRUE(endpoints.size() == 2);
		ASSERT_TRUE(endpoints[0].port == 9990);
		ASSERT_TRUE(endpoints[1].port == 9991);
	}

	{
		CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9992, 1);

		TC_Common::sleep(6);

		qPrx->getEndpoint();

		TC_Common::msleep(100);

		{
			std::vector<EndpointF> endpoints = qPrx->findObjectById("TestApp.RpcServer.HelloObj");

			ASSERT_TRUE(endpoints.size() == 3);
			ASSERT_TRUE(endpoints[0].port == 9990);
			ASSERT_TRUE(endpoints[1].port == 9991);
			ASSERT_TRUE(endpoints[2].port == 9992);
		}
	}

	{
		CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9993, 1);

		TC_Common::sleep(6);

		qPrx->getEndpoint();

		TC_Common::msleep(100);

		{
			std::vector<EndpointF> endpoints = qPrx->findObjectById("TestApp.RpcServer.HelloObj");

			ASSERT_TRUE(endpoints.size() == 4);
			ASSERT_TRUE(endpoints[0].port == 9990);
			ASSERT_TRUE(endpoints[1].port == 9991);
			ASSERT_TRUE(endpoints[2].port == 9992);
			ASSERT_TRUE(endpoints[3].port == 9993);
		}
	}

	stopServer(fs);

}

#define START_FRAMEWORK_SERVER_1_2 \
CDbHandle::cleanEndPoint(); \
CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9990, 1); \
CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9991, 1); \
FrameworkServer fs; \
startServer(fs, FRAMEWORK_CONFIG()); \
RpcServer rpc1Server; \
startServer(rpc1Server, RPC1_CONFIG()); \
RpcServer rpc2Server; \
startServer(rpc2Server, RPC2_CONFIG());


#define START_FRAMEWORK_HTTP_SERVER_1_2 \
CDbHandle::cleanEndPoint(); \
CDbHandle::addActiveEndPoint("TestApp.RpcServer.HttpObj", 8180, 1); \
CDbHandle::addActiveEndPoint("TestApp.RpcServer.HttpObj", 8181, 1); \
FrameworkServer fs; \
startServer(fs, FRAMEWORK_CONFIG()); \
RpcServer rpc1Server; \
startServer(rpc1Server, RPC1_CONFIG()); \
RpcServer rpc2Server; \
startServer(rpc2Server, RPC2_CONFIG());

#define STOP_FRAMEWORK_SERVER stopServer(fs);

#define HELLO_CALL {\
HelloPrx qPrx = comm->stringToProxy<HelloPrx>("TestApp.RpcServer.HelloObj"); \
int ret; \
string out; \
ret = qPrx->testHello(0, _buffer, out); \
ASSERT_TRUE(ret == 0); \
ASSERT_TRUE(out == _buffer); }

#define CHECK_REGISTRY_UPDATE { \
LOG_CONSOLE_DEBUG << endl;  \
stopServer(rpc1Server);         \
LOG_CONSOLE_DEBUG << endl;  \
CDbHandle::cleanEndPoint(); \
CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9992, 1); \
RpcServer rpc3Server; \
startServer(rpc3Server, RPC3_CONFIG()); \
LOG_CONSOLE_DEBUG << endl;  \
wait(6000);                    \
LOG_CONSOLE_DEBUG << endl;  \
HelloPrx qPrx = comm->stringToProxy<HelloPrx>("TestApp.RpcServer.HelloObj"); \
string out = ""; \
LOG_CONSOLE_DEBUG << endl;  \
int ret = qPrx->testHello(0, _buffer, out); \
LOG_CONSOLE_DEBUG << endl;  \
ASSERT_TRUE(ret == 0); \
ASSERT_TRUE(out == _buffer); \
stopServer(rpc2Server); \
LOG_CONSOLE_DEBUG << endl;  \
wait(100); \
LOG_CONSOLE_DEBUG << endl;  \
out = ""; \
ret = qPrx->testHello(0, _buffer, out); \
LOG_CONSOLE_DEBUG << endl;  \
ASSERT_TRUE(ret == 0); \
ASSERT_TRUE(out == _buffer); \
stopServer(rpc3Server); }

#define CREATE_COR \
{ \
	auto scheduler = TC_CoroutineScheduler::create(); \
	ServantProxyThreadData::getData()->_sched = scheduler; \
	scheduler->go([&]() \
	{	\
		{ HELLO_CALL; wait(6500); HELLO_CALL }   \
	}); \
	scheduler->run();	\
}

#define HTTP_CALL {\
ServantPrx prx = comm->stringToProxy<ServantPrx>("TestApp.RpcServer.HttpObj"); \
prx->tars_set_protocol(ServantProxy::PROTOCOL_HTTP1, 5); \
string buff = _buffer + "-" + TC_Common::tostr(TC_Thread::CURRENT_THREADID()); \
shared_ptr<TC_HttpRequest> req = std::make_shared<TC_HttpRequest>(); \
req->setPostRequest("http://tars.com/hello", buff, true); \
req->setHeader("Connection", "close"); \
shared_ptr<TC_HttpResponse> rsp;                                               \
int count = 10;     \
while(count-- > 0) { prx->http_call("hello", req, rsp); } }

#define HTTP_CREATE_COR \
{ \
	auto scheduler = TC_CoroutineScheduler::create(); \
	ServantProxyThreadData::getData()->_sched = scheduler; \
	scheduler->go([&]() \
	{	\
		{ HTTP_CALL; wait(6200); HTTP_CALL }   \
	}); \
	scheduler->run();	\
}

TEST_F(HelloTest, registryRpc)
{
//	shared_ptr<Communicator> c = getCommunicator();

	START_FRAMEWORK_SERVER_1_2

	shared_ptr<Communicator> c = getCommunicator();

	rpcFromRegistry(c.get());
	LOG_CONSOLE_DEBUG << endl;

	funcInCoroutine([&](){
		rpcFromRegistry(c.get());
	});

	funcInCoroutine([&](){
		shared_ptr<Communicator> comm = getCommunicator();

		rpcFromRegistry(comm.get());
	});

	LOG_CONSOLE_DEBUG << endl;
	stopServer(rpc1Server);
	stopServer(rpc2Server);
	stopServer(fs);
}


//一致hash测试
TEST_F(HelloTest, registryRpcConHashInvoke)
{
	START_FRAMEWORK_SERVER_1_2
        CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9989, 1);
	CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9992, 1);

	shared_ptr<Communicator> c = getCommunicator();

	rpcConHashFromRegistry(c.get());

	stopServer(rpc1Server);
	stopServer(rpc2Server);
        STOP_FRAMEWORK_SERVER;
}

TEST_F(HelloTest, registryRpcUpdateList)
{
	shared_ptr<Communicator> comm = getCommunicator();
	comm->setProperty("refresh-endpoint-interval", "5000");

	START_FRAMEWORK_SERVER_1_2

	HELLO_CALL

	CHECK_REGISTRY_UPDATE

	STOP_FRAMEWORK_SERVER
}


TEST_F(HelloTest, registryRpcUpdateListInCoroutine1)
{
	shared_ptr<Communicator> comm = getCommunicator();
	comm->setProperty("refresh-endpoint-interval", "5000");

	START_FRAMEWORK_SERVER_1_2

	funcInCoroutine([&]()
			{
				HELLO_CALL

				CHECK_REGISTRY_UPDATE
			}

	);

	STOP_FRAMEWORK_SERVER
}

TEST_F(HelloTest, registryRpcUpdateListInCoroutine2)
{
	shared_ptr<Communicator> comm = getCommunicator();
	comm->setProperty("refresh-endpoint-interval", "5000");

	START_FRAMEWORK_SERVER_1_2

	HELLO_CALL

	funcInCoroutine([&]()
	{
		CHECK_REGISTRY_UPDATE
	});

	STOP_FRAMEWORK_SERVER
}

//
//TEST_F(HelloTest, registryRpcCheckUpdateList)
//{
//	shared_ptr<Communicator> comm = getCommunicator();
//	comm->setProperty("refresh-endpoint-interval", "5000");
//
//	START_FRAMEWORK_SERVER_1_2
//
//	//发起网路调用
//	HELLO_CALL
//
//	HELLO_CALL
//
//	//构建多个线程, 都完成hello的调用
//	std::thread cor1([&]()
//	{
//		CREATE_COR
//	});
//
//	std::thread cor2([&]()
//	{
//		CREATE_COR
//	});
//
//	std::thread cor3([&]()
//	{
//		CREATE_COR
//	});
//
//	TC_Common::msleep(200);
//
//	vector<shared_ptr<CommunicatorEpoll>> v = comm->getAllCommunicatorEpoll();
//
//	//5个网络通信器(2个公有, 3个私有)
//	ASSERT_TRUE(v.size() == 5);
//	ASSERT_FALSE(v[0]->isSchedCommunicatorEpoll());
//	ASSERT_FALSE(v[1]->isSchedCommunicatorEpoll());
//	ASSERT_TRUE(v[2]->isSchedCommunicatorEpoll());
//	ASSERT_TRUE(v[3]->isSchedCommunicatorEpoll());
//	ASSERT_TRUE(v[4]->isSchedCommunicatorEpoll());
//
//	vector<ObjectProxy*> vop;
//
//	for (auto ce : v)
//	{
//		ObjectProxy* op = ce->hasObjectProxy("TestApp.RpcServer.HelloObj");
//
//		if (op)
//		{
//			vop.push_back(op);
//		}
//	}
//
//	//网络通信器都有对象
//	ASSERT_TRUE(vop.size() == 5);
//
//	for (auto op : vop)
//	{
//		vector<AdapterProxy*> adapters = op->getAdapters();
//
//		ASSERT_TRUE(adapters.size() == 2);
//		ASSERT_TRUE(adapters[0]->trans()->getConnectEndpoint().getPort() == 9990);
//		ASSERT_TRUE(adapters[1]->trans()->getConnectEndpoint().getPort() == 9991);
//	}
//
//	stopServer(rpc1Server);
//
//	//更新主控ip list
//	CDbHandle::cleanEndPoint();
//	CDbHandle::addActiveEndPoint("TestApp.RpcServer.HelloObj", 9992, 1);
//
//	RpcServer rpc3Server;
//	startServer(rpc3Server, RPC3_CONFIG());
//	//等待主控更新时间
//	wait(6000);
//
//	//调用两次, 这样两个公有网路通信器都会更新ip list
//	HELLO_CALL
//
//	HELLO_CALL
//
//	stopServer(rpc2Server);
//
//	wait(1000);
//
//	for (auto op : vop)
//	{
//		vector<AdapterProxy*> adapters = op->getAdapters();
//
//		for(auto adapter : adapters)
//		{
//			if(adapter->trans()->getConnectEndpoint().getPort() == 9992)
//			{
//				ASSERT_TRUE(adapter->isActiveInReg());
//			}
//			else
//			{
//				ASSERT_FALSE(adapter->isActiveInReg());
//			}
//		}
//	}
//
//	//第三个服务也停掉
//	stopServer(rpc3Server);
//	cor1.detach();
//	cor2.detach();
//	cor3.detach();
//	STOP_FRAMEWORK_SERVER
//}
//
//
//TEST_F(HelloTest, registryHttpRpcCheckUpdateList)
//{
//	shared_ptr<Communicator> comm = getCommunicator();
//	comm->setProperty("refresh-endpoint-interval", "5000");
//
//	START_FRAMEWORK_HTTP_SERVER_1_2
//
//	//发起网路调用
//	HTTP_CALL
//
//	//构建多个线程, 都完成hello的调用
//	std::thread cor1([&]()
//	{
//		//注意http call里面会wait一下在调用, 等待ip list更新
//		HTTP_CREATE_COR
//	});
//
//	TC_Common::msleep(200);
//
//	vector<shared_ptr<CommunicatorEpoll>> v = comm->getAllCommunicatorEpoll();
//
//	//3个网络通信器(2个公有, 1个私有)
//	ASSERT_TRUE(v.size() == 3);
//	ASSERT_FALSE(v[0]->isSchedCommunicatorEpoll());
//	ASSERT_FALSE(v[1]->isSchedCommunicatorEpoll());
//	ASSERT_TRUE(v[2]->isSchedCommunicatorEpoll());
//
//	ServantPrx prx = comm->stringToProxy<ServantPrx>("TestApp.RpcServer.HttpObj"); \
//
//	vector<ObjectProxy*> vop = 	prx->getObjectProxys();
//	//并行连接数*网络通信器个数
//	ASSERT_TRUE(vop.size() == 5*v.size());
//
//	for (auto op : vop)
//	{
//		vector<AdapterProxy*> adapters = op->getAdapters();
//
//		ASSERT_TRUE(adapters.size() == 2);
//		ASSERT_TRUE(adapters[0]->trans()->getConnectEndpoint().getPort() == 8180);
//		ASSERT_TRUE(adapters[1]->trans()->getConnectEndpoint().getPort() == 8181);
//	}
//
//	stopServer(rpc1Server);
//
//	//更新主控ip list
//	CDbHandle::cleanEndPoint();
//	CDbHandle::addActiveEndPoint("TestApp.RpcServer.HttpObj", 8182, 1);
//
//	RpcServer rpc3Server;
//	startServer(rpc3Server, RPC3_CONFIG());
//	//等待主控更新时间
//	wait(6000);
//	//调用, 触发ip list更新
//	HTTP_CALL
//	stopServer(rpc2Server);
//	wait(100);
//	stopServer(rpc3Server);
//
//	for (auto op : vop)
//	{
//		vector<AdapterProxy*> adapters = op->getAdapters();
//
//		for(auto adapter : adapters)
//		{
//			if(adapter->trans()->getConnectEndpoint().getPort() == 8182)
//			{
//				ASSERT_TRUE(adapter->isActiveInReg());
//			}
//			else
//			{
//				ASSERT_FALSE(adapter->isActiveInReg());
//			}
//		}
//	}
//
//	cor1.detach();
//	STOP_FRAMEWORK_SERVER
//}

TEST_F(HelloTest, registryRpcHashTagInvoke)
{
	START_FRAMEWORK_SERVER_1_2
	CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9989, 1);
	CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9992, 1);

	shared_ptr<Communicator> c = getCommunicator();

	string obj = "TestApp.RpcServer.HelloObj#9999";

	HelloPrx prx = c->stringToProxy<HelloPrx>(obj);

	string out;

	int co = prx->testHello(0, _buffer, out);

	ASSERT_TRUE(co == 0);

	stopServer(rpc1Server);
	stopServer(rpc2Server);
	STOP_FRAMEWORK_SERVER;
}

TEST_F(HelloTest, registryRpcMultiHashTagInvoke)
{
	START_FRAMEWORK_SERVER_1_2
	CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9989, 1);
	CDbHandle::addInactiveEndPoint("TestApp.RpcServer.HelloObj", 9992, 1);

	shared_ptr<Communicator> c = getCommunicator();

	string obj = "TestApp.RpcServer.HelloObj";

	HelloPrx prx = c->stringToProxy<HelloPrx>(obj);

	string out;

	int co = prx->testHello(0, _buffer, out);

	ASSERT_TRUE(co == 0);

	stopServer(rpc1Server);
	stopServer(rpc2Server);
	STOP_FRAMEWORK_SERVER;
}


TEST_F(HelloTest, registryPush)
{
	string obj = "TestApp.RpcServer.HelloObj";

	START_FRAMEWORK_SERVER_1_2

	shared_ptr<Communicator> c = getCommunicator();

	HelloPrx prx = c->stringToProxy<HelloPrx>(obj);

	vector<EndpointInfo> activeEndPoint;
	vector<EndpointInfo> inactiveEndPoint;

	int co;
	int i = 20;
	while(--i)
	{
		co = prx->testPort();
		ASSERT_TRUE(co == 9990 || co == 9991);
	}

	CDbHandle::cleanEndPoint();
	CDbHandle::addActiveEndPoint(obj, 9990, 1);
	CDbHandle::addInactiveEndPoint(obj, 9991, 1);
	CDbHandle::push();

	TC_Common::sleep(1);

	co = prx->testPort();

	prx->tars_endpoints(activeEndPoint, inactiveEndPoint);

	for(auto e : activeEndPoint)
	{
		LOG_CONSOLE_DEBUG << e.getEndpoint().toString() << endl;
	}

	TC_Common::sleep(1);

	i = 10;
	while(--i>0)
	{
		co = prx->testPort();
		LOG_CONSOLE_DEBUG << co << endl;
		ASSERT_TRUE(co == 9990);
	}

	stopServer(rpc1Server);
	stopServer(rpc2Server);
	STOP_FRAMEWORK_SERVER;
}