TarsCpp/unit-test/rpc/test_communicator.cpp


#include "hello_test.h"
#include "servant/CommunicatorEpoll.h"

//
//TEST_F(HelloTest, testNotifyCommunicatorEpoll)
//{
//	shared_ptr<Communicator> c = getCommunicator();
//
//	int netThread = TC_Common::strto<int>(c->getProperty("netthread"));
//
//	HelloServer server;
//	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE) 0);
//
//	auto v1 = c->getAllCommunicatorEpoll();
//
//	ASSERT_TRUE(v1.size() == 0);
//
//	HelloPrx prx = getObj<HelloPrx>(c.get(), "HelloAdapter");
//
//	std::thread th1([&]()
//					{ checkSyncOnce(prx); });
//
//	TC_Common::msleep(100);
//
//	auto v2 = c->getAllCommunicatorEpoll();
//	//两个公有通信器
//	ASSERT_TRUE(v2.size() == (size_t)netThread);
//
//	std::thread th2([&]()
//					{ checkSyncOnce(prx); });
//	TC_Common::msleep(100);
//
//	auto v3 = c->getAllCommunicatorEpoll();
//	//通信器个数不变
//	ASSERT_TRUE(v3.size() == (size_t)netThread);
//
//	funcInCoroutine([&]()
//					{
//		checkSyncOnce(prx);
//		auto v4 = c->getAllCommunicatorEpoll();
//		//为设置协程的情况下, 仍然是两个通信器
//		ASSERT_TRUE(v4.size() == (size_t)netThread);
//					},
//					false);
//
//	funcInCoroutine([&]()
//					{
//		checkSyncOnce(prx);
//		auto v5 = c->getAllCommunicatorEpoll();
//		//设置了协程, 会增加一个私有协程网络通信器
//		ASSERT_TRUE(v5.size() == (size_t)(netThread + 1));
//					},
//					true);
//
//	auto v6 = c->getAllCommunicatorEpoll();
//	//之前的业务线程释放了, 会减少一个私有协程网络通信器
//	ASSERT_TRUE(v6.size() == (size_t)netThread);
//
//	funcInCoroutine([&]()
//					{
//		checkASyncOnce(prx);
//		//设置协程, 会增加一个私有网路通信器, 同时异步回调在私有网络通信器中
//		auto v7 = c->getAllCommunicatorEpoll();
//		ASSERT_TRUE(v7.size() == (size_t)(netThread + 1));
//					});
//
//	th1.join();
//	th2.join();
//
//	stopServer(server);
//
//}
//
//int getNotify(shared_ptr<Communicator> c)
//{
//	if(c->getCommunicatorEpollNum() <= 0)
//	{
//		return 0;
//	}
//	auto n = c->getCommunicatorEpoll(0)->getNotify();
//
//	int sum = 0;
//	for(size_t i = 0; i< MAX_CLIENT_NOTIFYEVENT_NUM; i++)
//	{
//		if(n[i] != NULL)
//		{
//			++sum;
//		}
//	}
//	return sum;
//}

//
////测试notify句柄不泄露!
//TEST_F(HelloTest, testNotifyBussThreadQuitOnce)
//{
//	shared_ptr<Communicator> c = getCommunicator();
//
//	HelloPrx prx = getObj<HelloPrx>(c.get(), "HelloAdapter");
//
//	HelloServer server;
//	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE) 0);
//
//	//先把服务器端的强制都创建好
//	for (int i = 0; i < 20; i++)
//	{
//		checkSyncOnce(prx);
//	}
//
//	for (int i = 0; i < 3; i++)
//	{
//		size_t sp_count = ServantProxyThreadData::g_immortal->getList().size();
//		size_t notify_count = getNotify(c);
//
//		{
//			std::thread th1([&]()
//							{ checkSyncOnce(prx); });
//
//			th1.join();
//		}
//		TC_Common::msleep(10);
//
//		//私有线程数据释放了, 通知句柄数不增加
//		ASSERT_TRUE(ServantProxyThreadData::g_immortal->getList().size() == sp_count);
//		ASSERT_TRUE(getNotify(c) == (int)notify_count);
//	}
//
//	stopServer(server);
//
//}
//
////测试notify句柄不泄露!
//TEST_F(HelloTest, testNotifyBussThreadQuit)
//{
//	shared_ptr<Communicator> c = getCommunicator();
//	HelloPrx prx = getObj<HelloPrx>(c.get(), "HelloAdapter");
//
//	HelloServer server;
//	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE) 0);
//
////	for(int i = 0; i < 5000; i++)
//	for(int i = 0; i < 1000; i++)
//	{
//		{
//			std::thread th1([&]()
//							{
//								checkSyncOnce(prx);
//							});
//
//			th1.join();
//		}
//		TC_Common::msleep(10);
//
//		int num = getNotify(c);
//		ASSERT_TRUE(num < 5);
//	}
//
//	stopServer(server);
//
//}
//
//TEST_F(HelloTest, testNotifyCommunicatorQuit)
//{
//	HelloServer server;
//	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE) 0);
//
//	for(int i = 0; i < 1000; i++)
//	{
//		shared_ptr<Communicator> c = getCommunicator();
//		HelloPrx prx = getObj<HelloPrx>(c.get(), "HelloAdapter");
//
//		{
//
//			std::thread th1([&]()
//			{
//				try
//				{
//					checkSyncOnce(prx);
//				}
//				catch(exception &ex)
//				{
//					LOG_CONSOLE_DEBUG << i << ", " << ex.what() << endl;
//				}
//
//				c.reset();
//			});
//
//			th1.join();
//		}
//
//	}
//
//	stopServer(server);
//
//}

TEST_F(HelloTest, testNotifyProxyQuit)
{
	HelloServer server;
	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE)0);

	shared_ptr<Communicator> c = getCommunicator();
	HelloPrx prx = getObj<HelloPrx>(c.get(), "HelloAdapter");

	int count = 10;
	while (count-- > 0)
	{
		vector<std::thread*> vt;

		for (int i = 0; i < 1; i++)
		{
			vt.push_back(new std::thread([&]()
			{
				checkSyncOnce(prx);
			}));
		}

		for (auto v: vt)
		{
			v->join();
			delete v;
		}

//		TC_Common::sleep(3);
	}

	
	stopServer(server);
}


TEST_F(HelloTest, testCommunicatorGetResourcesInfo)
{
	HelloServer server;
	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE)0);

	shared_ptr<Communicator> c = getCommunicator();
	HelloPrx prx = getObj<HelloPrx>(c.get(), "HelloAdapter");

	int count = 10;
	while (count-- > 0)
	{
		vector<std::thread*> vt;

		for (int i = 0; i < 1; i++)
		{
			vt.push_back(new std::thread([&]()
			{
				checkSyncOnce(prx);
			}));
		}

		for (auto v: vt)
		{
			v->join();
			delete v;
		}

	}

	string buf = c->getResourcesInfo();
	LOG_CONSOLE_DEBUG << buf << endl;


	stopServer(server);
}
#if 0

TEST_F(HelloTest, testNotifyCtrlC)
{
	std::thread th([](){
		TC_Common::sleep(2);
		kill(getpid(), SIGINT);
	});

	HelloServer server;
	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE) 0);

	HelloPrx prx = getObj<HelloPrx>(server.getCommunicator(), "HelloAdapter");

	for(int i = 0; i < 10000; i++)
	{
		checkSyncOnce(prx);
	}

	stopServer(server);
}

TEST_F(HelloTest, testNotifyCtrlCGlobalCommunicator)
{
	std::thread th([](){
		TC_Common::sleep(2);
		kill(getpid(), SIGINT);
	});

	shared_ptr<Communicator> c = std::make_shared<Communicator>();

	HelloServer server;
	startServer(server, (TC_EpollServer::SERVER_OPEN_COROUTINE) 0);
	for(int i = 0; i < 10000; i++)
	{
		checkSync(c.get());
	}

	stopServer(server);
}

#endif