Unit tests for PIClientServer

libs/client_server/piclientserver_client.cpp

@@ -50,5 +50,4 @@ void PIClientServer::Client::connect(PINetworkAddress addr) {
 	config.apply(this);
 	tcp->connect(addr, true);
 	tcp->startThreadedRead();
-	piCout << "Connect to" << addr.toString();
 }

libs/client_server/piclientserver_server.cpp

@@ -123,7 +123,6 @@ void PIClientServer::Server::newClient(ServerClient * c) {
 	config.apply(c);
 	c->tcp->startThreadedRead();
 	c->connected();
-	piCout << "New client";
 }
 
 

tests/CMakeLists.txt

@@ -34,3 +34,4 @@ endmacro()
 pip_test(math)
 pip_test(core)
 pip_test(piobject)
+pip_test(client_server pip_client_server)

tests/client_server/client_server_test.cpp

@@ -0,0 +1,210 @@
+#include "pitime.h"
+#include "test_client_helper.h"
+
+#include "gtest/gtest.h"
+
+template<bool WithConnectPing = false, bool WithPong = false, ullong WriteSize = 64_KiB>
+PIClientServer::Server * createServer() {
+	auto s = new PIClientServer::Server();
+	s->setClientFactory([] { return new TestServerClient<WithConnectPing, WithPong, WriteSize>(); });
+	s->listenAll(12345);
+	return s;
+}
+
+bool waitLoop(std::function<bool()> exit_loop, const PISystemTime & timeout) {
+	PITimeMeasurer tm;
+	while (!exit_loop() && (tm.elapsed() < timeout)) {
+		piMinSleep();
+	}
+	return tm.elapsed() < timeout;
+}
+
+template<typename Client>
+Client * createAndConnectClient() {
+	auto c = new Client();
+	c->connect(PINetworkAddress::resolve("127.0.0.1", 12345));
+	return c;
+}
+
+TEST(ClientServer, OneClient) {
+	auto const loop_timeout = 1000_ms;
+	auto s                  = createServer<false, true>();
+	auto c                  = createAndConnectClient<TestClient<false, false>>();
+
+	waitLoop([s]() { return s->clientsCount() > 0; }, loop_timeout);
+	EXPECT_EQ(1, s->clientsCount());
+
+	c->ping();
+	waitLoop([c]() { return c->pongCnt() > 0; }, loop_timeout);
+	EXPECT_EQ(1, c->pongCnt());
+
+	s->forEachClient([](PIClientServer::ServerClient * sc) { static_cast<TestServerClient<> *>(sc)->ping(); });
+	waitLoop([c]() { return c->pongCnt() > 1; }, loop_timeout);
+	EXPECT_EQ(2, c->pongCnt());
+
+	EXPECT_TRUE(c->getTCP()->isConnected());
+
+	delete c;
+	waitLoop([s]() { return s->clientsCount() == 0; }, loop_timeout);
+	EXPECT_EQ(0, s->clientsCount());
+	delete s;
+}
+
+
+class ClientSendThread {
+	using ClientType = TestClient<false, false, 100_KiB>;
+
+public:
+	ClientSendThread() { client = createAndConnectClient<ClientType>(); }
+
+	~ClientSendThread() {
+		sendThread.stopAndWait();
+		delete client;
+	}
+
+	void startSend() {
+		auto c = client;
+		sendThread.start([c] { c->ping(); }, 100._Hz);
+	}
+    
+    void sendOnce() {client->ping();}
+
+	ClientType * client = nullptr;
+	PIThread sendThread;
+};
+
+
+int getServerPongs(PIClientServer::Server * s) {
+    int pongs = 0;
+    s->forEachClient([&pongs](PIClientServer::ServerClient * sc){
+        const auto c = static_cast<TestServerClient<> *>(sc);
+        pongs += c->pongCnt();
+    });
+    return pongs;
+}
+
+int getClientsPongs(const PIVector<ClientSendThread *> & clients) {
+    int pongs = 0;
+    clients.forEach([&pongs](ClientSendThread * c){
+        pongs += c->client->pongCnt();
+    });
+    return pongs;
+}
+
+int getClientsPings(const PIVector<ClientSendThread *> & clients) {
+    int pings = 0;
+    clients.forEach([&pings](ClientSendThread * c){
+        pings += c->client->pingCnt();
+    });
+    return pings;
+}
+
+
+TEST(ClientServer, ManyClients) {
+	auto const loop_timeout     = 1000_ms;
+	constexpr int clients_count = 20;
+	PIVector<ClientSendThread *> clients;
+	auto s = createServer<false, false, 100_KiB>();
+
+	piForTimes(clients_count) {
+		clients.append(new ClientSendThread());
+	}
+	EXPECT_EQ(clients_count, clients.size_s());
+
+	waitLoop([s]() { return s->clientsCount() == clients_count; }, loop_timeout);
+	EXPECT_EQ(clients_count, s->clientsCount());
+    
+    EXPECT_EQ(0, getServerPongs(s));
+    
+    EXPECT_EQ(getClientsPings(clients), 0);
+    
+    for (const auto c : clients) {
+        c->sendOnce();
+    }
+    
+    EXPECT_EQ(getClientsPings(clients), clients_count);
+    EXPECT_TRUE(clients.every([](ClientSendThread * c){return c->client->pingCnt() == 1;}));
+    EXPECT_TRUE(clients.every([](ClientSendThread * c){return c->client->pongCnt() == 0;}));
+    waitLoop([s]() { return getServerPongs(s) >= clients_count; }, loop_timeout);
+    EXPECT_EQ(clients_count, getServerPongs(s));
+    
+    s->forEachClient([](PIClientServer::ServerClient * sc) { static_cast<TestServerClient<> *>(sc)->ping(); });
+    const auto clientCheckPong = [&clients](){return clients.every([](ClientSendThread * c){return c->client->pongCnt() == 1;});};
+    waitLoop([&clientCheckPong]() { return clientCheckPong(); }, loop_timeout);
+    EXPECT_TRUE(clientCheckPong());
+    
+    for (const auto c : clients) {
+        c->startSend();
+    }
+    (100_ms).sleep();
+    EXPECT_TRUE(getClientsPings(clients) > clients_count*2);
+    EXPECT_TRUE(getServerPongs(s) > clients_count*2);
+    piDeleteAllAndClear(clients);
+    waitLoop([s]() { return s->clientsCount() == 0; }, loop_timeout);
+    EXPECT_EQ(0, s->clientsCount());
+    delete s;
+}
+
+TEST(ClientServer, DynamicClients) {
+    auto const loop_timeout     = 1000_ms;
+	constexpr int clients_count = 20;
+    PIVector<ClientSendThread *> clients;
+    PIMutex clients_mutex;
+	auto s = createServer<true, true, 100_KiB>();
+    
+    const auto spawnClient = [&clients, &clients_mutex]() {
+        auto c = new ClientSendThread();
+        c->startSend();
+        clients_mutex.lock();
+        clients << c;
+        clients_mutex.unlock();
+        piCout << "new client" << clients.size();
+     };
+    
+    piForTimes(clients_count) {
+		spawnClient();
+	}
+    
+    PIThread spawnThread;
+    PIThread deleteThread;
+    
+    spawnThread.start([&spawnClient](){
+        const int new_cnt = randomi() % 10;
+        piForTimes(new_cnt) {
+            spawnClient();
+        }
+    }, 12_Hz);
+    
+    deleteThread.start([&clients, &clients_mutex](){
+        const int rm_cnt = randomi() % 10;
+        piForTimes(rm_cnt) {
+            ClientSendThread * c = nullptr;
+            clients_mutex.lock();
+            if (clients.size() > 10) {
+                c = clients.take_back();
+            }
+            clients_mutex.unlock();
+            if (c) {
+                delete c;
+                piCout << "remove client" << clients.size();
+            }
+        }
+    }, 13_Hz);
+    
+    (1_s).sleep();
+    
+    EXPECT_GE(s->clientsCount(), 10);
+    
+    piCout << "now clients" << clients.size();
+    
+    //delete s;
+    
+    deleteThread.stopAndWait();
+    spawnThread.stopAndWait();
+    
+    piCout << "total clients" << clients.size();
+    
+    piDeleteAllAndClear(clients);
+    waitLoop([s]() { return s->clientsCount() == 0; }, loop_timeout);
+    EXPECT_EQ(0, s->clientsCount());
+}

tests/client_server/test_client_helper.h

@@ -0,0 +1,77 @@
+#ifndef TEST_CLIENT_HELPER_H
+#define TEST_CLIENT_HELPER_H
+
+#include "piclientserver_client.h"
+#include "piclientserver_server.h"
+#include "piethernet.h"
+#include "piliterals.h"
+#include "pithread.h"
+
+
+template<ullong WriteSize = 64_KiB, bool WithPong = false>
+class TestClientBase {
+public:
+	int pongCnt() const { return pong_cnt; }
+
+	int pingCnt() const { return ping_cnt; }
+
+	void ping() {
+        const auto data = PIByteArray(WriteSize);
+		writeInternal(data);
+		ping_cnt++;
+	}
+
+protected:
+	virtual void writeInternal(const PIByteArray & ba) = 0;
+
+	void readInternal(const PIByteArray & ba) {
+		last_read_size = ba.size();
+		pong_cnt++;
+		if (WithPong) ping();
+	}
+
+private:
+	int pong_cnt          = 0;
+	int ping_cnt          = 0;
+	ullong last_read_size = 0;
+};
+
+
+template<bool WithConnectPing = false, bool WithPong = false, ullong WriteSize = 64_KiB>
+class TestServerClient
+	: public PIClientServer::ServerClient
+	, public TestClientBase<WriteSize, WithPong> {
+	using Base = TestClientBase<WriteSize, WithPong>;
+
+private:
+	void readed(PIByteArray data) override { Base::readInternal(data); }
+
+	void connected() override {
+		if (WithConnectPing) {
+			Base::ping();
+		}
+	}
+
+	void writeInternal(const PIByteArray & ba) override { write(ba); }
+};
+
+
+template<bool WithConnectPing = false, bool WithPong = false, ullong WriteSize = 64_KiB>
+class TestClient
+	: public PIClientServer::Client
+	, public TestClientBase<WriteSize, WithPong> {
+	using Base = TestClientBase<WriteSize, WithPong>;
+
+private:
+	void readed(PIByteArray data) override { Base::readInternal(data); }
+
+	void connected() override {
+		if (WithConnectPing) {
+			Base::ping();
+		}
+	}
+
+	void writeInternal(const PIByteArray & ba) override { write(ba); }
+};
+
+#endif // TEST_CLIENT_HELPER_H