modify ipc and image buffer

src/ImageBuffer.cpp

@@ -54,6 +54,7 @@ void ImageBuffer::PushHumanImage(std::vector<uint8> data, uint32 width, uint32 h
     human_.width = width;
     human_.height = height;
     human_.comp = comp;
+    human_.updated = true;
 }
 
 void ImageBuffer::UpdateHuman(Texture2D* texture) {
@@ -62,6 +63,10 @@ void ImageBuffer::UpdateHuman(Texture2D* texture) {
     }
 
     ScopeLock<CriticalSection> lock(&human_cs_);
+    if (!human_.updated) {
+        return;
+    }
+    human_.updated = false;
     texture->Update(human_.width, human_.height, human_.comp, human_.data.data());
 }
 

src/ImageBuffer.h

@@ -41,6 +41,7 @@ private:
         uint32 width;
         uint32 height;
         uint32 comp;
+        bool updated{false};
     };
     CriticalSection human_cs_;
     Image human_;

src/Ipc/Ipc.cpp

@@ -26,7 +26,7 @@ Ipc::~Ipc() {
 }
 
 bool Ipc::listen() {
-	if (reciver_stop_.load(std::memory_order_acquire)) {
+	if (!reciver_stop_.load(std::memory_order_acquire)) {
 		return false;
 	}
 
@@ -38,7 +38,7 @@ bool Ipc::listen() {
 }
 
 void Ipc::stop() {
-	if (!reciver_stop_.load(std::memory_order_acquire)) {
+	if (reciver_stop_.load(std::memory_order_acquire)) {
 		return;
 	}
 
@@ -60,6 +60,23 @@ bool Ipc::send(const char* data, unsigned int size) {
 	return sender_->send(data, size);
 }
 
+
+bool Ipc::isConnected() {
+   /* if (receiver_) {
+        return receiver_->is_connected();
+    }*/
+	return false;
+}
+
+
+void Ipc::reConnect() {
+	sender_->reconnect(ipc::sender);
+	if (receiver_) {
+		receiver_->disconnect();
+		receiver_->reconnect(ipc::receiver);
+	}
+}
+
 void Ipc::registReadCallback(const std::shared_ptr<IReaderCallback>& reader) {
 	std::lock_guard<std::mutex> lock(mutex_);
 
@@ -88,9 +105,9 @@ void Ipc::doReciver(std::weak_ptr<Ipc> wThis) {
 
 	receiver_ = std::make_unique<ipc::channel>(writer_name_.c_str(), ipc::receiver);
 	while (!reciver_stop_.load(std::memory_order_acquire)) {
-		ipc::buff_t buffer = receiver_->recv();
+		ipc::buff_t buffer = receiver_->recv(5);
 		if (buffer.empty()) {
-			break;
+			continue;
 		}
 		const char* data = buffer.get<const char*>();
 		onReciveer(data, static_cast<unsigned int>(buffer.size()));

src/Ipc/Ipc.h

@@ -19,6 +19,8 @@ public:
 	bool listen();
 	void stop();
 	bool send(const char* data, unsigned int size);
+	bool isConnected();
+	void reConnect();
 
 	void registReadCallback(const std::shared_ptr<IReaderCallback>& reader);
 	void unregistReadCallback(const std::shared_ptr<IReaderCallback>& reader);
@@ -39,7 +41,7 @@ private:
 	std::unique_ptr<ipc::channel> sender_;
 	std::unique_ptr<ipc::channel> receiver_;
 	std::thread reciver_thread_;
-	std::atomic_bool reciver_stop_{false};
+	std::atomic_bool reciver_stop_{true};
 
 	std::mutex mutex_;
 	using ReaderCallbackList = std::vector<std::shared_ptr<IReaderCallback>>;

src/Ipc/IpcMoudle.cpp

@@ -45,11 +45,10 @@ void OnParseImageData(const char* data, size_t size) {
     } else if (0x03 == identifier) {
         ImageBuffer::Get()->PushImage(ImageBuffer::IBType::FaceMask, std::move(img_data), width, height, bit_depth);
     }
-    // Further processing of img_bytes can be done here
 }
 
 bool IpcMoudle::Initialize() {
-    if (!initialize("ipc_sender", "ipc_sender")) {
+    if (!initialize("human_render", "human_product")) {
         ERRORLOG("ipc initialize failed");
         return false;
     }
@@ -64,9 +63,28 @@ bool IpcMoudle::Initialize() {
         ERRORLOG("ipc listen failed");
         return false;
     }
+    lastHeartbeatTime_ = std::chrono::steady_clock::now();
     return true;
 }
 
 void IpcMoudle::Uninitialize() {
     uninitialize();
 }
+
+bool IpcMoudle::Send(const char* data, unsigned int size) {
+    return send(data, size);
+}
+
+void IpcMoudle::OnFrame() {
+    auto now = std::chrono::steady_clock::now();
+    auto duration = std::chrono::duration_cast<std::chrono::seconds>(now - lastHeartbeatTime_);
+
+    if (duration.count() >= 2) {
+        constexpr char heartbeat[] = "heartbeat";
+        if (!send("heartbeat", sizeof(heartbeat) / sizeof(heartbeat[0]))) {
+            ERRORLOG("send heartbeat failed");
+            reConnect();
+        }
+        lastHeartbeatTime_ = now;
+    }
+}

src/Ipc/IpcMoudle.h

@@ -1,5 +1,7 @@
 #pragma once
 
+#include <chrono>
+
 #include "Core/Singleton.h"
 #include "Core/Constant.h"
 
@@ -11,5 +13,13 @@ public:
     virtual ~IpcMoudle() = default;
     bool Initialize() override;
     void Uninitialize() override;
+
+    bool Send(const char* data, unsigned int size);
+
+    void OnFrame();
+
+private:
+    std::chrono::time_point<std::chrono::steady_clock> lastHeartbeatTime_;
+
 };
 

src/Ipc/ipclib.cpp

@@ -67,3 +67,14 @@ bool __stdcall listen() {
 	 g_readCallback->setCallback(callback);
 	 return true;
  }
+
+
+ void __stdcall reConnect() {
+     assert(g_ipc);
+     return g_ipc->reConnect();
+ }
+
+ bool __stdcall isConnect() {
+	 assert(g_ipc);
+	 return g_ipc->isConnected();
+ }

src/Ipc/ipclib.h

@@ -16,6 +16,10 @@ extern "C" {
     bool __stdcall send(const char* data, unsigned int size);
     bool __stdcall setReaderCallback(ReaderCallbackFunc callback);
 
+    void __stdcall reConnect();
+
+    bool __stdcall isConnect();
+
 #ifdef __cplusplus
 }
 #endif // __cplusplus

src/Main.cpp

@@ -153,6 +153,8 @@ static int APIENTRY wWinMain(_In_ HINSTANCE hInstance, _In_opt_ HINSTANCE hPrevI
         view = glm::lookAt(glm::vec3(0, 0, 10), glm::vec3(0, 0, 0), glm::vec3(0, 1, 0));
         projection = glm::perspective(glm::radians(60.f), ratio, 1.f, 1000.f);
 
+        IpcMoudle::Get()->OnFrame();
+
         ImageBuffer::Get()->Update(ImageBuffer::IBType::Human, &huaman);
 
         glBindVertexArray(VAO);
@@ -170,6 +172,9 @@ static int APIENTRY wWinMain(_In_ HINSTANCE hInstance, _In_opt_ HINSTANCE hPrevI
     //glDrawArrays(GL_TRIANGLES, 0, 6);
     //glDrawElements(GL_TRIANGLES, 6, GL_UNSIGNED_INT, 0);
 
+    constexpr char quit[] = "quit";
+    IpcMoudle::Get()->Send(quit, sizeof(quit)/ sizeof(quit[0]));
+
     glfwDestroyWindow(window);
     glfwTerminate();