add exit command.

.gitea/workflows/deploy.yaml

@@ -50,9 +50,10 @@ jobs:
           script: |
             cd /root/Server
             source /etc/profile
-            echo "这里需要先停止原来的HttpServer, 然后再将/tmp/HttpServer拷贝至/root/Server"
-            # pgrep -f HttpServer | xargs -r kill -s 9
-            # nohup /root/Server/HttpServer > /dev/null 2>&1 &
+            /root/Server/HttpServer --exit
+            sleep 10
+            cp /tmp/HttpServer /root/Server/HttpServer
+            /root/Server/HttpServer &
             exit 0
       - name: Notify-End
         if: ${{ always() }}

CMakeLists.txt

@@ -16,12 +16,18 @@ set(WT_INCLUDE_DIR ${WT_ROOT}/include)
 set(WT_LIBRARY_DIRS ${WT_ROOT}/lib)
 set(WT_LIBRARIES wt wttest wthttp wtdbo wtdbosqlite3)
 
+set(NNG_ROOT ${Libraries_ROOT}/nng-1.9.0)
+set(NNG_INCLUDE_DIR ${NNG_ROOT}/include)
+set(NNG_LIBRARY_DIRS ${NNG_ROOT}/lib)
+
 set(OPENSSL_LIBRARIES ssl crypto)
 
 include(FetchContent)
 FetchContent_Declare(Kylin
 GIT_REPOSITORY https://amass.fun/gitea/amass/Kylin.git
 )
+set(KYLIN_WITH_NNG ON)
+# add_subdirectory(/mnt/e/Projects/Kylin Kylin)
 FetchContent_MakeAvailable(Kylin)
 
 add_subdirectory(MediaServer)

Server/Application.cpp

@@ -3,12 +3,17 @@
 #include "DateTime.h"
 #include "HttpSession.h"
 #include "IoContext.h"
+#include "Nng/SocketAisoWrapper.h"
 #include "ServiceLogic.h"
 #include "ServiceManager.h"
 #include "SystemUsage.h"
 #include "WeChatContext/CorporationContext.h"
+#include <boost/json/object.hpp>
+#include <boost/json/serialize.hpp>
 #include <boost/stacktrace.hpp>
 
+constexpr auto IpcUrl = "ipc:///tmp/nng_ipc_server";
+
 Application::Application(const std::string &path)
     : ApplicationSettings(path), m_router{std::make_shared<boost::urls::router<RequestHandler>>()} {
 
@@ -232,6 +237,9 @@ Application::Application(const std::string &path)
     m_ioContext = Amass::Singleton<IoContext>::instance<Amass::Construct>(getThreads());
     m_timer = std::make_shared<boost::asio::system_timer>(*m_ioContext->ioContext());
 
+    m_replyer = std::make_shared<Nng::Asio::Socket>(*m_ioContext->ioContext(), Nng::Reply);
+    m_replyer->listen(IpcUrl);
+
     m_systemUsage = std::make_shared<SystemUsage>(*m_ioContext->ioContext(), getHomeAssistantAccessToken());
     m_systemUsage->start();
 
@@ -259,9 +267,10 @@ void Application::insertUrl(std::string_view url, RequestHandler &&handler) {
 }
 
 int Application::exec() {
+    startAcceptRequest();
     LOG(info) << "application start successful ...";
     startCheckInterval(*m_ioContext->ioContext(), 2);
-    m_ioContext->run<IoContext::Mode::Synchronous>();
+    m_ioContext->run();
     LOG(info) << "application exit successful ...";
     return m_status;
 }
@@ -313,3 +322,39 @@ void Application::alarmTask() {
         alarmTask();
     });
 }
+
+void Application::startAcceptRequest() {
+    m_replyer->asyncReceive([ptr{weak_from_this()}](const boost::system::error_code &error, const Nng::Buffer &buffer) {
+        if (error) {
+            LOG(error) << error.message();
+        }
+        LOG(info) << buffer.data();
+        if (ptr.expired()) return;
+        auto self = ptr.lock();
+        auto value = boost::json::parse(buffer.data());
+        auto &request = value.as_object();
+        if (request.at("command").as_string() == "exit") {
+            boost::json::object reply;
+            reply["status"] = 0;
+            reply["message"] = "will exit.";
+            auto txt = boost::json::serialize(reply);
+            self->m_replyer->send(txt.data(), txt.size());
+            std::raise(SIGUSR1); // 发送自定义信号
+        }
+        self->startAcceptRequest();
+    });
+}
+
+void Application::requetExit() {
+    nng_log_set_logger(nng_stderr_logger);
+    LOG(info) << "send exit request to program.";
+    Nng::Socket request(Nng::Request);
+    request.dial(IpcUrl);
+
+    boost::json::object object;
+    object["command"] = "exit";
+    auto text = boost::json::serialize(object);
+    request.send(text.data(), text.size());
+    auto buffer = request.recv();
+    LOG(info) << buffer.data<char>();
+}

Server/Application.h

@@ -13,6 +13,12 @@ class ChatRoom;
 class SystemUsage;
 class IoContext;
 
+namespace Nng {
+namespace Asio {
+class Socket;
+}
+} // namespace Nng
+
 class Application : public ApplicationSettings<Application>, public std::enable_shared_from_this<Application> {
 public:
     using Pointer = std::shared_ptr<Application>;
@@ -35,9 +41,11 @@ public:
     const RequestHandler *find(boost::urls::segments_encoded_view path,
                                boost::urls::matches_base &matches) const noexcept;
     void insertUrl(std::string_view url, RequestHandler &&handler);
+    static void requetExit();
 
 protected:
     void alarmTask();
+    void startAcceptRequest();
 
 private:
     int m_status = 0;
@@ -46,6 +54,7 @@ private:
     std::shared_ptr<boost::asio::system_timer> m_timer;
     std::shared_ptr<ChatRoom> m_charRoom;
     std::shared_ptr<SystemUsage> m_systemUsage;
+    std::shared_ptr<Nng::Asio::Socket> m_replyer;
 };
 
 #endif // __SETTINGS_H__

Server/CMakeLists.txt

@@ -24,6 +24,7 @@ target_link_libraries(Server
     PRIVATE Database
     PRIVATE MediaServer
     PRIVATE WebApplication
+    PRIVATE Nng
     PRIVATE ${Boost_LIBRARIES}
 )
 

Server/WeChatContext/CorporationContext.cpp

@@ -11,7 +11,6 @@
 
 CorporationContext::CorporationContext(boost::asio::io_context &ioContext)
     : m_ioContext(ioContext), m_timer(ioContext) {
-    boost::asio::defer([this]() { updateAccessToken(); });
 
     auto manager = Amass::Singleton<ServiceManager>::instance();
     if (manager)
@@ -47,6 +46,17 @@ void CorporationContext::sendMessage(MessageType type, const std::string &messag
     LOG(info) << response;
 }
 
+void CorporationContext::start() {
+    boost::asio::defer(m_ioContext, [ptr{weak_from_this()}]() {
+        if (ptr.expired()) {
+            LOG(error) << "CorporationContext instance was expired";
+            return;
+        }
+        auto self = ptr.lock();
+        self->updateAccessToken();
+    });
+}
+
 void CorporationContext::notify(const RequestType &request) {
     boost::system::error_code error;
     auto json = boost::json::parse(request.body(), error);

Server/WeChatContext/CorporationContext.h

@@ -1,10 +1,13 @@
 #ifndef __CORPORATIONCONTEXT_H__
 #define __CORPORATIONCONTEXT_H__
 
+#include "Singleton.h"
 #include <boost/asio/steady_timer.hpp>
 #include <boost/beast/http/string_body.hpp>
 
-class CorporationContext {
+class CorporationContext : public std::enable_shared_from_this<CorporationContext> {
+    friend class Amass::Singleton<CorporationContext>;
+
 public:
     enum MessageType {
         Text,
@@ -12,8 +15,8 @@ public:
     };
     using RequestType = boost::beast::http::request<boost::beast::http::string_body>;
 
-    CorporationContext(boost::asio::io_context &ioContext);
     void sendMessage(MessageType type, const std::string &message);
+    void start();
 
     /**
      * @brief
@@ -25,6 +28,7 @@ public:
     void notify(const RequestType &request);
 
 protected:
+    CorporationContext(boost::asio::io_context &ioContext);
     void updateAccessToken();
 
 private:

Server/WeChatContext/WeChatContext.cpp

@@ -60,8 +60,7 @@ std::string WeChatContext::reply(const std::string &body) {
 
 WeChatContext::WeChatContext(boost::asio::io_context &ioContext)
     : m_ioContext(ioContext), m_timer(ioContext), m_sessionsExpireTimer(ioContext) {
-
-    boost::asio::defer([this]() { updateAccessToken(); });
+    boost::asio::defer(m_ioContext, [this]() { updateAccessToken(); });
 }
 
 void WeChatContext::updateAccessToken() {

Server/main.cpp

@@ -25,6 +25,7 @@ int main(int argc, char const *argv[]) {
     // clang-format off
     description.add_options()
             ("help,h", "produce help message.")
+            ("exit,e", "signal program to exit.")
             ("prefix", boost::program_options::value<std::string>(),"set prefix path (default: ${pwd} )");
     // clang-format on
     boost::program_options::variables_map values;
@@ -39,6 +40,9 @@ int main(int argc, char const *argv[]) {
     if (values.count("help")) {
         std::cout << description << std::endl;
         std::exit(0);
+    } else if (values.count("exit")) {
+        Application::requetExit();
+        std::exit(0);
     }
 
     std::error_code error;
@@ -76,6 +80,7 @@ int main(int argc, char const *argv[]) {
 
     auto wechatContext = Singleton<WeChatContext>::instance<Construct>(application->ioContext());
     auto corpContext = Singleton<CorporationContext>::instance<Construct>(application->ioContext());
+    corpContext->start();
     auto live2d = std::make_shared<Live2dBackend>();
 
     LOG(info) << "hardware_concurrency: " << std::thread::hardware_concurrency()
@@ -90,6 +95,7 @@ int main(int argc, char const *argv[]) {
 #else
     boost::asio::signal_set signals(application->ioContext(), SIGINT, SIGTERM);
 #endif
+    signals.add(SIGUSR1);
     signals.async_wait([&application](boost::system::error_code const &, int signal) {
         // Stop the io_context. This will cause run()
         // to return immediately, eventually destroying the

WebApplication/WebApplication.cpp

@@ -12,7 +12,8 @@ static std::unique_ptr<Wt::WApplication> createApplication(const Wt::WEnvironmen
     return std::make_unique<Hello>(env, false);
 }
 
-std::unique_ptr<Wt::WApplication> createBlogApplication(const Wt::WEnvironment &env, Wt::Dbo::SqlConnectionPool *blogDb) {
+std::unique_ptr<Wt::WApplication> createBlogApplication(const Wt::WEnvironment &env,
+                                                        Wt::Dbo::SqlConnectionPool *blogDb) {
     return std::make_unique<BlogApplication>(env, *blogDb);
 }
 
@@ -32,31 +33,22 @@ WebApplication::WebApplication() {
 
         BlogSession::configureAuth();
         m_blogSqlConnectionPool = BlogSession::createConnectionPool(m_server->appRoot() + "database.sqlite");
-        m_server->addEntryPoint(Wt::EntryPointType::Application, std::bind(&createBlogApplication, std::placeholders::_1, m_blogSqlConnectionPool.get()),
+        m_server->addEntryPoint(Wt::EntryPointType::Application,
+                                std::bind(&createBlogApplication, std::placeholders::_1, m_blogSqlConnectionPool.get()),
                                 "/blog");
         m_server->addEntryPoint(Wt::EntryPointType::WidgetSet, createWidgetSet, "/gui/hello.js");
         Session::configureAuth();
         m_server->addResource(std::make_shared<JsonResource>(), "/json");
         m_server->addResource(std::make_shared<PlaintextResource>(), "/plaintext");
         m_server->addResource(std::make_shared<DbResource>("database.sqlite"), "/db");
-        m_thread = std::thread(&WebApplication::run, this);
 
+           m_server->start();
     } catch (const std::exception &e) {
         LOG(error) << e.what();
     }
 }
 
 WebApplication::~WebApplication() {
-    if (m_thread.joinable()) {
-        m_thread.join();
-    }
+
 }
 
-void WebApplication::run() {
-    try {
-
-        m_server->run();
-    } catch (const std::exception &e) {
-        LOG(error) << e.what();
-    }
-}

WebApplication/WebApplication.h

@@ -17,12 +17,8 @@ public:
     WebApplication();
     ~WebApplication();
 
-protected:
-    void run();
-
 private:
     std::unique_ptr<Wt::WServer> m_server;
-    std::thread m_thread;
     std::unique_ptr<Wt::Dbo::SqlConnectionPool> m_blogSqlConnectionPool;
 };