From 98e3961e18a6548193ebc0374d1e8fa34e1251d5 Mon Sep 17 00:00:00 2001 From: Viacheslav Demydiuk Date: Fri, 12 Jan 2024 21:28:35 +0200 Subject: [PATCH] Added MtNotification message for sending client ready. Moved parsing message to MtMessage source code --- MtService/MtMessage.cpp | 63 ++++++++++++ MtService/MtMessage.h | 27 ++++- MtService/MtServer.cpp | 152 ++++++++++++---------------- MtService/MtServer.h | 4 +- MtService/MtService.cpp | 26 +++-- MtService/MtService.vcxproj | 1 + MtService/MtService.vcxproj.filters | 3 + 7 files changed, 174 insertions(+), 102 deletions(-) create mode 100755 MtService/MtMessage.cpp diff --git a/MtService/MtMessage.cpp b/MtService/MtMessage.cpp new file mode 100755 index 00000000..712b3720 --- /dev/null +++ b/MtService/MtMessage.cpp @@ -0,0 +1,63 @@ +#include "pch.h" +#include "MtMessage.h" + +#include +#include +#include + +std::unique_ptr MtCommand::Parse(const std::string& msg) +{ + std::string::size_type pos = msg.find(';'); + + using namespace boost::spirit; + using rule_s = qi::rule; + using rule = qi::rule; + + std::string message_type; + std::string expert_handle; + std::string command_id; + std::string command_type; + std::string payload; + + rule_s word_p = qi::as_string[+(qi::char_ - qi::char_(';'))]; + rule message_type_p = word_p[boost::phoenix::ref(message_type) = qi::_1]; + rule expert_handle_p = +qi::char_(';') >> word_p[boost::phoenix::ref(expert_handle) = qi::_1]; + rule command_id_p = +qi::char_(';') >> word_p[boost::phoenix::ref(command_id) = qi::_1]; + rule command_type_p = +qi::char_(';') >> word_p[boost::phoenix::ref(command_type) = qi::_1]; + rule payload_p = +qi::char_(';') >> word_p[boost::phoenix::ref(payload) = qi::_1]; + + std::unique_ptr command; + + bool ok = qi::parse(msg.begin(), msg.end(), message_type_p >> -expert_handle_p >> -command_id_p >> -command_type_p >> -payload_p); + if (ok) + { + command = std::make_unique(std::stoi(expert_handle), + std::stoi(command_id), std::stoi(command_type), std::move(payload)); + } + + return command; +} + +std::unique_ptr MtNotification::Parse(const std::string& msg) +{ + std::string::size_type pos = msg.find(';'); + + using namespace boost::spirit; + using rule_s = qi::rule; + using rule = qi::rule; + + std::string message_type; + std::string notification_type; + + rule_s word_p = qi::as_string[+(qi::char_ - qi::char_(';'))]; + rule message_type_p = word_p[boost::phoenix::ref(message_type) = qi::_1]; + rule notification_type_p = +qi::char_(';') >> word_p[boost::phoenix::ref(notification_type) = qi::_1]; + + std::unique_ptr command; + + bool ok = qi::parse(msg.begin(), msg.end(), message_type_p >> -notification_type_p); + if (ok) + command = std::make_unique((NotificationType)std::stoi(notification_type)); + + return command; +} \ No newline at end of file diff --git a/MtService/MtMessage.h b/MtService/MtMessage.h index 11641097..3152ed28 100755 --- a/MtService/MtMessage.h +++ b/MtService/MtMessage.h @@ -2,6 +2,7 @@ #include #include +#include const std::string MT_MESSAGE_DELIMETER{ ";" }; @@ -13,7 +14,12 @@ enum MessageType EXPERT_LIST = 3, EXPERT_ADDED = 4, EXPERT_REMOVED = 5, - CLIENT_READY = 6 + NOTIFICATION = 6 +}; + +enum NotificationType +{ + CLIENT_READY = 0 }; class MtMessage @@ -83,6 +89,8 @@ public: std::string getPayload() const { return payload_; } + static std::unique_ptr Parse(const std::string& msg); + private: MessageType GetType() const override { @@ -103,23 +111,33 @@ private: std::string payload_; }; -class MtClientReady : public MtMessage +class MtNotification : public MtMessage { public: - MtClientReady() + MtNotification(NotificationType type) + : notification_type_(type) { } + NotificationType GetNotificationType() const + { + return notification_type_; + } + + static std::unique_ptr Parse(const std::string& msg); + private: MessageType GetType() const override { - return MessageType::CLIENT_READY; + return MessageType::NOTIFICATION; } std::string GetBody() const override { return ""; } + + NotificationType notification_type_; }; class MtEvent : public MtMessage @@ -176,7 +194,6 @@ private: return ss.str(); } - int expert_handle_; std::vector experts_; }; diff --git a/MtService/MtServer.cpp b/MtService/MtServer.cpp index 85113454..d497d4a6 100755 --- a/MtService/MtServer.cpp +++ b/MtService/MtServer.cpp @@ -3,9 +3,6 @@ #include #include -#include -#include -#include #include "MtConnection.h" #include "MtExpert.h" @@ -165,59 +162,15 @@ void MtServer::OnAccept(boost::beast::error_code ec, } }); connection->OnMessageReceived.connect([con = connection->weak_from_this(), this](const std::string& msg) { - log_.Trace("MtServer::OnMessageReceived: %s", msg.c_str()); - - std::string::size_type pos = msg.find(';'); - if (pos == std::string::npos) + try { - log_.Warning("MtServer::OnMessageReceived: invalid message type."); - return; + ProcessMessage(msg, con); + } + catch (std::exception e) + { + log_.Error("MtServer::OnMessageReceived: Failed to process message: %s", msg.c_str()); } - std::string msg_type_str = msg.substr(0, pos);; - auto msg_type = std::atoi(msg_type_str.c_str()); - if (msg_type == MessageType::COMMAND) - { - auto command = ParseCommand(msg); - if (command) - { - auto expert_handle = command->getExpertHandle(); - if (experts_.count(expert_handle) > 0) - { - experts_[expert_handle]->expert->Process(std::move(command), [&](const MtResponse& response) { - auto connection = con.lock(); - if (connection) - connection->Send(response.Serialize()); - else - log_.Warning("MtServer::OnMessageReceived: connection is not valid (null)"); - }); - } - else - { - log_.Warning("MtServer::OnMessageReceived: Expert is not found with handle %d", command->getExpertHandle()); - } - } - else - { - log_.Warning("MtServer::OnMessageReceived: Failed to parse command from message: %s", msg.c_str()); - } - } - else if (msg_type == MessageType::CLIENT_READY) - { - std::vector expert_list; - for (const auto& e : experts_) - expert_list.push_back(e.first); - - MtExpertListMsg msg(std::move(expert_list)); - auto connection = con.lock(); - if (connection) - connection->Send(msg.Serialize()); - } - else - { - log_.Warning("MtServer::OnMessageReceived: unsupported message type %d", msg_type); - return; - } }); pending_connections_.push_back(connection); connection->Accept(); @@ -227,39 +180,6 @@ void MtServer::OnAccept(boost::beast::error_code ec, } } -std::unique_ptr MtServer::ParseCommand(const std::string& msg) -{ - std::string::size_type pos = msg.find(';'); - - using namespace boost::spirit; - using rule_s = qi::rule; - using rule = qi::rule; - - std::string message_type; - std::string expert_handle; - std::string command_id; - std::string command_type; - std::string payload; - - rule_s word_p = qi::as_string[+(qi::char_ - qi::char_(';'))]; - rule message_type_p = word_p[boost::phoenix::ref(message_type) = qi::_1]; - rule expert_handle_p = +qi::char_(';') >> word_p[boost::phoenix::ref(expert_handle) = qi::_1]; - rule command_id_p = +qi::char_(';') >> word_p[boost::phoenix::ref(command_id) = qi::_1]; - rule command_type_p = +qi::char_(';') >> word_p[boost::phoenix::ref(command_type) = qi::_1]; - rule payload_p = +qi::char_(';') >> word_p[boost::phoenix::ref(payload) = qi::_1]; - - std::unique_ptr command; - - bool ok = qi::parse(msg.begin(), msg.end(), message_type_p >> -expert_handle_p >> -command_id_p >> -command_type_p >> -payload_p); - if (ok) - { - command = std::make_unique(std::stoi(expert_handle), - std::stoi(command_id), std::stoi(command_type), std::move(payload)); - } - - return command; -} - // Start accepting incoming connections void MtServer::Start() { @@ -318,4 +238,64 @@ void MtServer::Send(const MtMessage& message) log_.Trace("%s: %s", __FUNCTION__, msg.c_str()); for (auto& c : connections_) c->Send(msg); +} + +void MtServer::ProcessMessage(const std::string& msg, std::weak_ptr con) +{ + log_.Trace("%s: %s", __FUNCTION__, msg.c_str()); + + std::string::size_type pos = msg.find(';'); + if (pos == std::string::npos) + { + log_.Warning("%s: : invalid message type.", __FUNCTION__); + return; + } + + std::string msg_type_str = msg.substr(0, pos);; + auto msg_type = std::atoi(msg_type_str.c_str()); + if (msg_type == MessageType::COMMAND) + { + auto command = MtCommand::Parse(msg); + if (command) + { + auto expert_handle = command->getExpertHandle(); + if (experts_.count(expert_handle) > 0) + { + experts_[expert_handle]->expert->Process(std::move(command), [con = con, this](const MtResponse& response) { + auto connection = con.lock(); + if (connection) + connection->Send(response.Serialize()); + }); + } + else + log_.Warning("%s: Expert is not found with handle %d", __FUNCTION__, command->getExpertHandle()); + } + else + log_.Warning("%s: Failed to parse command from message: %s", __FUNCTION__, msg.c_str()); + } + else if (msg_type == MessageType::NOTIFICATION) + { + auto notification = MtNotification::Parse(msg); + if (notification) + { + if (notification->GetNotificationType() == NotificationType::CLIENT_READY) + { + std::vector expert_list; + for (const auto& e : experts_) + expert_list.push_back(e.first); + + MtExpertListMsg msg(std::move(expert_list)); + auto connection = con.lock(); + if (connection) + connection->Send(msg.Serialize()); + } + } + else + log_.Warning("%s: Failed to parse notification from message: %s", __FUNCTION__, msg.c_str()); + } + else + { + log_.Warning("%s: unsupported message type %d", __FUNCTION__, msg_type); + return; + } } \ No newline at end of file diff --git a/MtService/MtServer.h b/MtService/MtServer.h index 229a1889..87bf5355 100755 --- a/MtService/MtServer.h +++ b/MtService/MtServer.h @@ -48,13 +48,13 @@ private: void OnAccept(boost::beast::error_code ec, boost::asio::ip::tcp::socket socket); - std::unique_ptr ParseCommand(const std::string& msg); - void OnEvent(const MtEvent& event); void OnDeinitExpert(int handle); void Send(const MtMessage& message); + void ProcessMessage(const std::string& msg, std::weak_ptr con); + Logger log_{ "MtServer" }; boost::asio::io_context& context_; unsigned short port_; diff --git a/MtService/MtService.cpp b/MtService/MtService.cpp index acec9d60..40413376 100755 --- a/MtService/MtService.cpp +++ b/MtService/MtService.cpp @@ -126,26 +126,30 @@ void MtServiceImpl::SendResponse(int handle, const std::string& payload) int MtServiceImpl::GetCommandType(int handle) { - log_.Trace("%s: handle = %d", __FUNCTION__, handle); - std::packaged_task task([handle, this]() { return (experts_.count(handle) > 0) ? experts_[handle]->GetCommandType() : 0; }); auto f = task.get_future(); boost::asio::post(context_, std::bind(std::move(task))); - return f.get(); + + auto command_type = f.get(); + log_.Trace("%s: handle = %d, command_type = %d", __FUNCTION__, handle, command_type); + + return command_type; } std::string MtServiceImpl::GetCommandPayload(int handle) { - log_.Trace("%s: handle = %d", __FUNCTION__, handle); - std::packaged_task task([handle, this]() { - return (experts_.count(handle) > 0) ? experts_[handle]->GetCommandPayload() : 0; + return (experts_.count(handle) > 0) ? experts_[handle]->GetCommandPayload() : ""; }); auto f = task.get_future(); boost::asio::post(context_, std::bind(std::move(task))); - return f.get(); + + auto payload = f.get(); + log_.Trace("%s: handle = %d, payload = %s", __FUNCTION__, handle, payload.c_str()); + + return payload; } void MtServiceImpl::LogError(const std::string& error) @@ -163,9 +167,13 @@ void MtServiceImpl::OnServerStopped(unsigned short port) void MtServiceImpl::ThreadProc() { - log_.Debug("%s: started", __FUNCTION__); + std::thread::id this_id = std::this_thread::get_id(); + std::stringstream ss; + ss << this_id; + + log_.Debug("%s: thread %s started", __FUNCTION__, ss.str().c_str()); context_.run(); - log_.Debug("%s: stopped", __FUNCTION__); + log_.Debug("%s: thread %s stopped", __FUNCTION__, ss.str().c_str()); } //-------------------------------------------------------------------------------- diff --git a/MtService/MtService.vcxproj b/MtService/MtService.vcxproj index c815bc55..96272c3b 100755 --- a/MtService/MtService.vcxproj +++ b/MtService/MtService.vcxproj @@ -159,6 +159,7 @@ + diff --git a/MtService/MtService.vcxproj.filters b/MtService/MtService.vcxproj.filters index bada8a25..9ad7d084 100755 --- a/MtService/MtService.vcxproj.filters +++ b/MtService/MtService.vcxproj.filters @@ -74,5 +74,8 @@ Source Files + + Source Files + \ No newline at end of file