From 1e50648415031d56d79a791c577fdfe1a79c066a Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Mon, 24 Aug 2026 18:28:38 +1000 Subject: [PATCH 1/9] Add XMPP client connection manager Introduce a new ClientConnectionManager for XMPP client-to-server connections. It provides a module config schema for the C2S listener port (default 5222) and registers the module for config discovery, preparing the server-side connection management scaffolding. --- .../xtrpg/xmpp/ClientConnectionManager.hpp | 37 +++++++++++++++++++ src/xmpp/ClientConnectionManager.cpp | 3 ++ 2 files changed, 40 insertions(+) create mode 100644 include/xtrpg/xmpp/ClientConnectionManager.hpp create mode 100644 src/xmpp/ClientConnectionManager.cpp diff --git a/include/xtrpg/xmpp/ClientConnectionManager.hpp b/include/xtrpg/xmpp/ClientConnectionManager.hpp new file mode 100644 index 0000000..3f58ea6 --- /dev/null +++ b/include/xtrpg/xmpp/ClientConnectionManager.hpp @@ -0,0 +1,37 @@ +#pragma once + +#include + +#include "xtrpg/config/ConfigManager.hpp" + +namespace xtrpg::xmpp { +/** + * + */ +class ClientConnectionManager : public config::IModuleConfigProvider { +public: + ClientConnectionManager() = default; + + explicit ClientConnectionManager(asio::io_context &ioContext) + : _ioContext(&ioContext) {} + + /** + * + */ + config::ModuleConfig getConfigSchema() const { + return {.name = "c2s", + .description = "", + .options = {{.key = "port", + .defaultValue = 5222, + .description = + " Port number that this server will listen " + "on for Client (or C2S) connections."}}}; + } + +private: + asio::io_context *_ioContext = nullptr; +}; + +REGISTER_MODULE_CONFIG(ClientConnectionManager); + +} // namespace xtrpg::xmpp diff --git a/src/xmpp/ClientConnectionManager.cpp b/src/xmpp/ClientConnectionManager.cpp new file mode 100644 index 0000000..f95ad3c --- /dev/null +++ b/src/xmpp/ClientConnectionManager.cpp @@ -0,0 +1,3 @@ +#include "xtrpg/xmpp/ClientConnectionManager.hpp" + +namespace xtrpg::xmpp {} \ No newline at end of file From f231952faf244cffdc0d05ef803fbf2bc3e8a63d Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Mon, 24 Aug 2026 18:29:02 +1000 Subject: [PATCH 2/9] Consolidate CMake build to single executable Refactor build configuration to use a single executable target instead of multiple intermediate libraries (xtrpg_config, xtrpg_network, xtrpg_xmpp). Sources are now compiled directly into xtrpg_cpp_server. Changes: - Remove separate library targets and consolidate into executable - Add ENABLE_XMPP_JID build option - Change include/link visibility from PUBLIC to PRIVATE - Remove commented-out legacy storage backend code --- CMakeLists.txt | 60 ++++++++++++++++---------------------------------- 1 file changed, 19 insertions(+), 41 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 70c4ecc..f8d26bd 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -25,6 +25,8 @@ configure_file("generated/version.hpp.in" ${CMAKE_BINARY_DIR}/generated/version. set(CMAKE_CXX_STANDARD 20) set(CMAKE_CXX_STANDARD_REQUIRED ON) +option(ENABLE_XMPP_JID "Build XMPP JID support" ON) + # Find Packages - vcpkg provides standard CMake target configs find_package(asio CONFIG REQUIRED) find_package(OpenSSL REQUIRED) @@ -43,24 +45,6 @@ find_package(UserModules QUIET) # option(ENABLE_STORAGE_SQLITE "Include SQLite storage backend" OFF) # option(ENABLE_STORAGE_POSTGRES "Include PostgreSQL storage backend" OFF) -# Core Server Target -# add_library(xtrpg_core STATIC -# ) -# target_include_directories(xtrpg_core PUBLIC include) - -# Conditional Storage Compilation -# if(ENABLE_STORAGE_SQLITE) -# find_package(SQLite3 REQUIRED) -# target_sources(xtrpg_core PRIVATE src/storage/SQLiteBackend.cpp) -# target_compile_definitions(xtrpg_core PUBLIC HAS_STORAGE_SQLITE) -# target_link_libraries(xtrpg_core PRIVATE SQLite::SQLite3) -# endif() - -# if(ENABLE_STORAGE_MEMORY) -# target_sources(xtrpg_core PRIVATE src/storage/MemoryBackend.cpp) -# target_compile_definitions(xtrpg_core PUBLIC HAS_STORAGE_MEMORY) -# endif() - # Include discovered modules into the build if(UserModules_FOUND) message(STATUS "Integrating User Modules:") @@ -73,19 +57,24 @@ if(UserModules_FOUND) endforeach() endif() -# Config LIB -add_library(xtrpg_config STATIC - src/config/ConfigManager.cpp +# Single executable target +add_executable(xtrpg_cpp_server + apps/main.cpp +) +target_include_directories(xtrpg_cpp_server PRIVATE + include + ${CMAKE_BINARY_DIR}/generated ) -target_include_directories(xtrpg_config PUBLIC include) -# Network LIB -add_library(xtrpg_network STATIC +target_sources(xtrpg_cpp_server PRIVATE + src/config/ConfigManager.cpp src/network/SocketConnectionListener.cpp src/network/TcpConnection.cpp -) -target_include_directories(xtrpg_network PUBLIC include) -target_link_libraries(xtrpg_network PUBLIC + src/xmpp/ClientConnectionManager.cpp + src/xmpp/Jid.cpp) + + +target_link_libraries(xtrpg_cpp_server PRIVATE asio::asio OpenSSL::SSL OpenSSL::Crypto @@ -94,27 +83,16 @@ if(WIN32) # Windows-specific OS network primitives for raw sockets # Define minimum Windows version (0x0601 = Windows 7, 0x0A00 = Windows 10) # 0x0A00 unlocks modern Windows socket features for Asio - target_compile_definitions(xtrpg_network PUBLIC + target_compile_definitions(xtrpg_cpp_server PRIVATE _WIN32_WINNT=0x0A00 WINVER=0x0A00 ) - target_link_libraries(xtrpg_network PUBLIC ws2_32 wsock32 crypt32) + target_link_libraries(xtrpg_cpp_server PRIVATE ws2_32 wsock32 crypt32) elseif(UNIX AND NOT APPLE) # Platform-specific OS network primitives for raw sockets - target_link_libraries(xtrpg_network PUBLIC pthread dl) + target_link_libraries(xtrpg_cpp_server PRIVATE pthread dl) endif() -# XMPP LIB -add_library(xtrpg_xmpp STATIC - src/xmpp/Jid.cpp -) -target_include_directories(xtrpg_xmpp PUBLIC include) - -# Executable Target -add_executable(xtrpg_cpp_server apps/main.cpp) -target_link_libraries(xtrpg_cpp_server PRIVATE xtrpg_config xtrpg_network xtrpg_xmpp) -target_include_directories(xtrpg_cpp_server PRIVATE ${CMAKE_BINARY_DIR}/generated) - # If modules produce targets registered via xmpp_register_user_module if(USER_MODULE_TARGETS) target_link_libraries(xtrpg_cpp_server PRIVATE ${USER_MODULE_TARGETS}) From 53a6aaf2981b0205048460070d415522b2ff28d5 Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Sat, 29 Aug 2026 16:46:50 +1000 Subject: [PATCH 3/9] Add XML tokenizer and XMPP sessions This change introduces XML stream tokenization and a dedicated XMPP client session lifecycle. TcpConnection now supports async reads without relying on shared_from_this, and the tokenizer listener contract was simplified to a raw pointer for direct callback handling. ClientConnectionManager now tracks active sessions, creates a session for each new TCP connection, and starts processing the inbound XML stream. A new ClientSession class reads socket data, passes it through the tokenizer, and handles XML events for the XMPP client flow. --- CMakeLists.txt | 5 +- include/xtrpg/network/TcpConnection.hpp | 8 +- include/xtrpg/xml/node/TagNode.hpp | 8 +- .../xml/tokenizer/XmlStreamTokenizer.hpp | 5 +- .../xtrpg/xmpp/ClientConnectionManager.hpp | 18 +++- include/xtrpg/xmpp/session/ClientSession.hpp | 63 ++++++++++++ src/network/TcpConnection.cpp | 83 ++++++++++++---- src/xml/tokenizer/XmlStreamTokenizer.cpp | 45 +++++---- src/xmpp/ClientConnectionManager.cpp | 35 ++++++- src/xmpp/session/ClientSession.cpp | 96 +++++++++++++++++++ 10 files changed, 313 insertions(+), 53 deletions(-) create mode 100644 include/xtrpg/xmpp/session/ClientSession.hpp create mode 100644 src/xmpp/session/ClientSession.cpp diff --git a/CMakeLists.txt b/CMakeLists.txt index f8d26bd..b2f5093 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -70,8 +70,11 @@ target_sources(xtrpg_cpp_server PRIVATE src/config/ConfigManager.cpp src/network/SocketConnectionListener.cpp src/network/TcpConnection.cpp + src/xml/tokenizer/XmlStreamTokenizer.cpp src/xmpp/ClientConnectionManager.cpp - src/xmpp/Jid.cpp) + src/xmpp/session/ClientSession.cpp + src/xmpp/Jid.cpp +) target_link_libraries(xtrpg_cpp_server PRIVATE diff --git a/include/xtrpg/network/TcpConnection.hpp b/include/xtrpg/network/TcpConnection.hpp index d38e618..23cd814 100644 --- a/include/xtrpg/network/TcpConnection.hpp +++ b/include/xtrpg/network/TcpConnection.hpp @@ -36,7 +36,7 @@ enum class ConnectionState { * Represents a TCP connection that can be upgraded to TLS. It provides methods * to write data to the connection, read data from the connection and close it. */ -class TcpConnection : public std::enable_shared_from_this { +class TcpConnection { public: /** @@ -53,6 +53,12 @@ class TcpConnection : public std::enable_shared_from_this { */ void upgrade(asio::ssl::context &ssl_ctx); + /** + * Async read from the underlying tcp connection, calling the provided lambda + * function with a new istream of the incoming stream data. + */ + void read(std::function callback); + /** * Writes data to the connection. If the connection is closed, it will throw * an exception. diff --git a/include/xtrpg/xml/node/TagNode.hpp b/include/xtrpg/xml/node/TagNode.hpp index 01a9bcf..d38d0ea 100644 --- a/include/xtrpg/xml/node/TagNode.hpp +++ b/include/xtrpg/xml/node/TagNode.hpp @@ -12,7 +12,6 @@ #include #include "xtrpg/xml/node/IAttributes.hpp" -#include "xtrpg/xml/node/INode.hpp" #include "xtrpg/xml/node/ITagname.hpp" #include "xtrpg/xml/node/NodeContainer.hpp" #include "xtrpg/xml/node/NodeType.hpp" @@ -23,16 +22,13 @@ namespace xtrpg::xml::node { /** * Represents an XML element with a tag name, attributes, and child nodes. */ -class TagNode : public INode, - public ITagname, - public IAttributes, - public NodeContainer { +class TagNode : public ITagname, public IAttributes, public NodeContainer { public: /** * Inline constructor that accepts a tag name. */ explicit TagNode(std::string name) - : INode(NodeType::TAG), ITagname(name), IAttributes(), NodeContainer() {} + : ITagname(name), IAttributes(), NodeContainer(NodeType::TAG) {} /** * Explicitly defaulted copy constructor. diff --git a/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp b/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp index 70a9747..e5877d6 100644 --- a/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp +++ b/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp @@ -28,7 +28,8 @@ class XmlStreamTokenizer { /** * Defines a instance to act as the listener for this class. */ - void setListener(std::shared_ptr listener) { + void setListener(XmlTokenListener *listener) { + std::cout << "[XmlStreamTokenizer] Assign listener." << std::endl; _listener = listener; } @@ -69,7 +70,7 @@ class XmlStreamTokenizer { TokenizationError _error{TokenizationError::NONE}; - std::weak_ptr _listener; + XmlTokenListener *_listener = nullptr; }; } // namespace xtrpg::xml::tokenizer \ No newline at end of file diff --git a/include/xtrpg/xmpp/ClientConnectionManager.hpp b/include/xtrpg/xmpp/ClientConnectionManager.hpp index 3f58ea6..779e740 100644 --- a/include/xtrpg/xmpp/ClientConnectionManager.hpp +++ b/include/xtrpg/xmpp/ClientConnectionManager.hpp @@ -1,20 +1,34 @@ #pragma once #include +#include +#include #include "xtrpg/config/ConfigManager.hpp" +#include "xtrpg/interface/Observer.hpp" +#include "xtrpg/network/TcpConnection.hpp" +#include "xtrpg/xmpp/session/ClientSession.hpp" namespace xtrpg::xmpp { /** * */ -class ClientConnectionManager : public config::IModuleConfigProvider { +class ClientConnectionManager + : public config::IModuleConfigProvider, + public interface::Observer> { public: ClientConnectionManager() = default; + ~ClientConnectionManager(); + explicit ClientConnectionManager(asio::io_context &ioContext) : _ioContext(&ioContext) {} + /** + * A new Tcp Connection is created. + */ + void onObservation(std::shared_ptr &ctx); + /** * */ @@ -30,6 +44,8 @@ class ClientConnectionManager : public config::IModuleConfigProvider { private: asio::io_context *_ioContext = nullptr; + + std::vector _clientSessionPtrs; }; REGISTER_MODULE_CONFIG(ClientConnectionManager); diff --git a/include/xtrpg/xmpp/session/ClientSession.hpp b/include/xtrpg/xmpp/session/ClientSession.hpp new file mode 100644 index 0000000..b2cab2b --- /dev/null +++ b/include/xtrpg/xmpp/session/ClientSession.hpp @@ -0,0 +1,63 @@ +#pragma once + +#include +#include + +#include "xtrpg/network/TcpConnection.hpp" +#include "xtrpg/xml/node/DeclarationNode.hpp" +#include "xtrpg/xml/node/TagNode.hpp" +#include "xtrpg/xml/tokenizer/TokenizationError.hpp" +#include "xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp" +#include "xtrpg/xml/tokenizer/XmlTokenListener.hpp" + +namespace xtrpg::xmpp::session { + +class ClientSession : public xml::tokenizer::XmlTokenListener, + public std::enable_shared_from_this { + +public: + ClientSession(network::TcpConnection tcpConnection) + : _tcpConnection(std::move(tcpConnection)) { + std::cout << "[ClientSession] New Instance created." << std::endl; + this->_tokenizer.setListener(this); + } + ~ClientSession(); + + // Session Control + void start(); + void stop(); + void shutdown(); + void process(); + + // Transport Control + void sendRaw(std::string_view data); + + // Tokenizer Calls + void openTag(std::string_view tagname); + void closeTag(); + void openDeclaration(std::string_view tagname); + void closeDeclaration(); + void setAttribute(std::string_view name, std::string_view value); + void appendText(std::string_view content); + void onError(xml::tokenizer::TokenizationError error); + +private: + network::TcpConnection _tcpConnection; + xml::tokenizer::XmlStreamTokenizer _tokenizer; + + xml::node::DeclarationNode *_ptrDeclarationNode = nullptr; + xml::node::TagNode *_ptrRootStreamNode = nullptr; + + /** + * Boolean flag that indicates whether the session is actively processing + * data to/from the underling connection. + */ + std::atomic _isStopped{true}; + + /** + * Boolean flag that indicates whether the underlying connection has been + * terminated. + */ + std::atomic _isShutdown{false}; +}; +} // namespace xtrpg::xmpp::session \ No newline at end of file diff --git a/src/network/TcpConnection.cpp b/src/network/TcpConnection.cpp index f7437c3..2665037 100644 --- a/src/network/TcpConnection.cpp +++ b/src/network/TcpConnection.cpp @@ -1,10 +1,12 @@ #include "xtrpg/network/TcpConnection.hpp" +#include + namespace xtrpg::network { void TcpConnection::upgrade(asio::ssl::context &ssl_ctx) { - auto self = shared_from_this(); - asio::post(*this->_strand, [this, self, &ssl_ctx]() { + + asio::post(*this->_strand, [this, &ssl_ctx]() { if (this->isClosed() || this->isClosing() || this->isSecure()) { return; } @@ -12,7 +14,7 @@ void TcpConnection::upgrade(asio::ssl::context &ssl_ctx) { this->_sslStream.emplace(std::move(this->_tcpSocket), ssl_ctx); this->_sslStream->async_handshake( - asio::ssl::stream_base::server, [this, self](std::error_code ec) { + asio::ssl::stream_base::server, [this](std::error_code ec) { if (ec) { this->_state = ConnectionState::CLOSED; if (this->_sslStream) { @@ -25,18 +27,64 @@ void TcpConnection::upgrade(asio::ssl::context &ssl_ctx) { }); } +void TcpConnection::read(std::function callback) { + std::cout << "[TcpConnection] Requesting to read." << std::endl; + auto buffer = std::make_shared>(4096); + + asio::post(*this->_strand, [this, buffer, callback]() { + std::cout << "[TcpConnection] asio::post." << std::endl; + if (!this->isOpen()) { + std::cout << "[TcpConnection] Stream not open." << std::endl; + throw exception::ConnectionClosed(); + } + + if (this->isSecure() && this->_sslStream) { + this->_sslStream->async_read_some( + asio::buffer(*buffer), + [this, buffer, callback](std::error_code ec, + std::size_t bytes_transferred) { + if (ec) { + this->_state = ConnectionState::CLOSED; + if (this->_sslStream) { + this->_sslStream->lowest_layer().close(); + } + return; + } + std::string data(buffer->data(), bytes_transferred); + std::istringstream stream(data); + callback(stream); + }); + + return; + } + + this->_tcpSocket.async_read_some( + asio::buffer(*buffer), + [this, buffer, callback](std::error_code ec, + std::size_t bytes_transferred) { + if (ec) { + this->_state = ConnectionState::CLOSED; + this->_tcpSocket.close(); + return; + } + std::string data(buffer->data(), bytes_transferred); + std::istringstream stream(data); + callback(stream); + }); + }); +} + void TcpConnection::write(std::string_view data) { auto payload = std::make_shared(data); - auto self = shared_from_this(); - asio::post(*this->_strand, [this, self, payload]() { + asio::post(*this->_strand, [this, payload]() { if (!this->isOpen()) { throw exception::ConnectionClosed(); } if (this->isSecure() && this->_sslStream) { asio::async_write(*this->_sslStream, asio::buffer(*payload), - [this, self, payload](std::error_code ec, std::size_t) { + [this, payload](std::error_code ec, std::size_t) { if (ec) { this->_state = ConnectionState::CLOSED; if (this->_sslStream) { @@ -48,7 +96,7 @@ void TcpConnection::write(std::string_view data) { } asio::async_write(this->_tcpSocket, asio::buffer(*payload), - [this, self, payload](std::error_code ec, std::size_t) { + [this, payload](std::error_code ec, std::size_t) { if (ec) { this->_state = ConnectionState::CLOSED; this->_tcpSocket.close(); @@ -58,8 +106,8 @@ void TcpConnection::write(std::string_view data) { } void TcpConnection::close() { - auto self = shared_from_this(); - asio::post(*this->_strand, [this, self]() { + + asio::post(*this->_strand, [this]() { if (this->isClosed() || this->isClosing()) { return; } @@ -69,16 +117,15 @@ void TcpConnection::close() { if (this->isSecure() && this->_sslStream) { this->_sslStream->lowest_layer().cancel(); - this->_sslStream->async_shutdown( - [this, self](const asio::error_code &ec) { - if (this->_sslStream) { - this->_sslStream->lowest_layer().shutdown( - asio::ip::tcp::socket::shutdown_both); - this->_sslStream->lowest_layer().close(); - } + this->_sslStream->async_shutdown([this](const asio::error_code &ec) { + if (this->_sslStream) { + this->_sslStream->lowest_layer().shutdown( + asio::ip::tcp::socket::shutdown_both); + this->_sslStream->lowest_layer().close(); + } - this->_state = ConnectionState::CLOSED; - }); + this->_state = ConnectionState::CLOSED; + }); return; } diff --git a/src/xml/tokenizer/XmlStreamTokenizer.cpp b/src/xml/tokenizer/XmlStreamTokenizer.cpp index 157cea1..6628b39 100644 --- a/src/xml/tokenizer/XmlStreamTokenizer.cpp +++ b/src/xml/tokenizer/XmlStreamTokenizer.cpp @@ -20,22 +20,21 @@ namespace xtrpg::xml::tokenizer { void XmlStreamTokenizer::process(std::istream &stream) { // If the tokenizer is already in an error state then re-issue the same error. if (TokenizationError::NONE != this->_error) { - if (const auto listener = this->_listener.lock()) { - listener->onError(this->_error); + if (nullptr == this->_listener) { + this->_listener->onError(this->_error); } return; } - const auto listener = this->_listener.lock(); const auto fail = [&](const TokenizationError error) { this->_error = error; - if (listener) { - listener->onError(error); + if (nullptr != this->_listener) { + this->_listener->onError(error); } }; const auto appendText = [&](const std::string_view text) { - if (!text.empty() && listener) { - listener->appendText(text); + if (!text.empty() && nullptr != this->_listener) { + this->_listener->appendText(text); } }; const auto openStartTag = [&]() { @@ -43,8 +42,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { fail(TokenizationError::MALFORMED_INPUT); return; } - if (listener) { - listener->openTag(this->_buffer); + if (nullptr != this->_listener) { + this->_listener->openTag(this->_buffer); } this->_buffer.clear(); }; @@ -53,8 +52,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { fail(TokenizationError::MALFORMED_INPUT); return; } - if (listener) { - listener->openDeclaration(this->_buffer); + if (nullptr != this->_listener) { + this->_listener->openDeclaration(this->_buffer); } this->_buffer.clear(); }; @@ -180,8 +179,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::ATTRIBUTE_VALUE: if (character == this->_quote) { - if (listener) { - listener->setAttribute(this->_attributeName, this->_buffer); + if (nullptr != this->_listener) { + this->_listener->setAttribute(this->_attributeName, this->_buffer); } this->_attributeName.clear(); this->_buffer.clear(); @@ -205,8 +204,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { if (this->_buffer.empty()) { fail(TokenizationError::MALFORMED_INPUT); } else { - if (listener) { - listener->closeTag(); + if (nullptr != this->_listener) { + this->_listener->closeTag(); } this->_buffer.clear(); this->_state = State::TEXT; @@ -220,8 +219,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; } if (character == '>' && !this->_buffer.empty()) { - if (listener) { - listener->closeTag(); + if (nullptr != this->_listener) { + this->_listener->closeTag(); } this->_buffer.clear(); this->_state = State::TEXT; @@ -300,8 +299,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::DECLARATION_ATTRIBUTE_VALUE: if (character == this->_quote) { - if (listener) { - listener->setAttribute(this->_attributeName, this->_buffer); + if (nullptr != this->_listener) { + this->_listener->setAttribute(this->_attributeName, this->_buffer); } this->_attributeName.clear(); this->_buffer.clear(); @@ -315,8 +314,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::DECLARATION_QUESTION: if (character == '>') { - if (listener) { - listener->closeDeclaration(); + if (nullptr != this->_listener) { + this->_listener->closeDeclaration(); } this->_state = State::TEXT; } else { @@ -325,8 +324,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::SELF_CLOSING: if (character == '>') { - if (listener) { - listener->closeTag(); + if (nullptr != this->_listener) { + this->_listener->closeTag(); } this->_state = State::TEXT; } else { diff --git a/src/xmpp/ClientConnectionManager.cpp b/src/xmpp/ClientConnectionManager.cpp index f95ad3c..33afe8f 100644 --- a/src/xmpp/ClientConnectionManager.cpp +++ b/src/xmpp/ClientConnectionManager.cpp @@ -1,3 +1,36 @@ #include "xtrpg/xmpp/ClientConnectionManager.hpp" -namespace xtrpg::xmpp {} \ No newline at end of file +namespace xtrpg::xmpp { + +ClientConnectionManager::~ClientConnectionManager() { + // Loop over the `this->_clientSessionPtrs` vector, shut them down and delete + // the instances + for (auto ptrSession : this->_clientSessionPtrs) { + if (ptrSession) { + ptrSession->shutdown(); + delete ptrSession; + } + } + this->_clientSessionPtrs.clear(); +} + +/** + * A new Tcp Connection is created. + */ +void ClientConnectionManager::onObservation( + std::shared_ptr &ctx) { + std::cout << "[ClientConnectionManager] New client connection received" + << std::endl; + + // Create a ClientSession for this connection + auto clientSession = new session::ClientSession(std::move(*ctx)); + this->_clientSessionPtrs.push_back(clientSession); + + // Start the session + clientSession->start(); + + std::cout << "[ClientConnectionManager] ClientSession created and started" + << std::endl; +} + +} // namespace xtrpg::xmpp \ No newline at end of file diff --git a/src/xmpp/session/ClientSession.cpp b/src/xmpp/session/ClientSession.cpp new file mode 100644 index 0000000..5e5f554 --- /dev/null +++ b/src/xmpp/session/ClientSession.cpp @@ -0,0 +1,96 @@ +#include "xtrpg/xmpp/session/ClientSession.hpp" + +#include + +namespace xtrpg::xmpp::session { + +ClientSession::~ClientSession() { + // destroy the root stream node + if (this->_ptrRootStreamNode != nullptr) { + delete this->_ptrRootStreamNode; + this->_ptrRootStreamNode = nullptr; + } + + // destroy the xml declaration node + if (this->_ptrDeclarationNode != nullptr) { + delete this->_ptrDeclarationNode; + this->_ptrDeclarationNode = nullptr; + } + + // remove myself from the tokenizer + this->_tokenizer.setListener(nullptr); +} + +void ClientSession::start() { + std::cout << "[ClientSession] Requesting to start." << std::endl; + if (this->_isShutdown) { + std::cout << "[ClientSession] Failed to start, already shutdown." + << std::endl; + return; + } + + this->_isStopped.exchange(false); + this->process(); +} + +void ClientSession::stop() { this->_isStopped.exchange(true); } + +void ClientSession::shutdown() { + this->stop(); + if (this->_isShutdown.exchange(true)) { + return; + } + + // shutdown the TCP connection + this->_tcpConnection.close(); +} + +void ClientSession::sendRaw(std::string_view data) { + if (this->_isShutdown) { + return; + } + + this->_tcpConnection << data; +} + +void ClientSession::process() { + + std::cout << "[ClientSession] Requesting to process." << std::endl; + if (this->_isStopped) { + std::cout << "[ClientSession] Session is stopped." << std::endl; + return; + } + + // calls the _tcpConnect to request the next chunk of data + // the lambda function + this->_tcpConnection.read([this](std::istream &is) { + std::cout << "[Client Session] Passing input stream to the tokenizer." + << std::endl; + this->_tokenizer.process(is); + this->process(); + }); +} + +void ClientSession::openTag(std::string_view tagname) { + std::cout << "[Client Session] Receive tag: " << tagname << std::endl; + + this->sendRaw( + "Stanza size " + "limit of 64KB exceeded."); +} +void ClientSession::closeTag() {} +void ClientSession::openDeclaration(std::string_view tagname) { + // ignore declaration tags +} +void ClientSession::closeDeclaration() { + // ignore declaration tags +} +void ClientSession::setAttribute(std::string_view name, + std::string_view value) {} +void ClientSession::appendText(std::string_view content) {} +void ClientSession::onError(xml::tokenizer::TokenizationError error) {} +} // namespace xtrpg::xmpp::session \ No newline at end of file From 8bb7babd4e942dfed857c17fd52cbe95943541ca Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Sat, 29 Aug 2026 16:47:17 +1000 Subject: [PATCH 4/9] Add async IO listener and graceful shutdown Initialize Asio io_context and start a SocketConnectionListener in main. Adds includes, a global atomic shutdown flag and signal handler (SIGINT/SIGTERM/SIGHUP) to support graceful termination. Creates a ClientConnectionManager, registers it as an observer with the listener, reads the C2S port (default 5222) from config, and runs io_context on a thread pool. On shutdown the listener is stopped, io_context is stopped and all IO threads are joined. Also adds a catch-all exception handler and informational logging. --- apps/main.cpp | 90 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) diff --git a/apps/main.cpp b/apps/main.cpp index c693556..4ac0c76 100644 --- a/apps/main.cpp +++ b/apps/main.cpp @@ -1,14 +1,36 @@ +#include #include #include +#include +#include +#include +#include + +#include #include "version.hpp" #include "xtrpg/config/ConfigManager.hpp" +#include "xtrpg/interface/Observer.hpp" +#include "xtrpg/network/SocketConnectionListener.hpp" +#include "xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp" +#include "xtrpg/xmpp/ClientConnectionManager.hpp" +#include "xtrpg/xmpp/session/ClientSession.hpp" #ifdef _WIN32 #define _WINSOCKAPI_ #include #endif +// Global flag for signal handling +std::atomic g_shouldShutdown{false}; + +// Signal handler for graceful shutdown +void signalHandler(int signal) { + std::cout << "\nReceived signal " << signal + << ", initiating graceful shutdown..." << std::endl; + g_shouldShutdown = true; +} + int main(int argc, char *argv[]) { #ifdef _WIN32 // Set console codepages to UTF-8 (65001) @@ -45,11 +67,79 @@ int main(int argc, char *argv[]) { } configManager.parseCLI(argc, argv); + // Initialize Asio IO context for async I/O operations + asio::io_context ioContext; + + // Create temporary connection handler to listen for incoming connections + // and create ClientSession instances + auto connectionManager = + std::make_shared(ioContext); + + // Get the port from configuration (default 5222 for XMPP C2S) + auto portValue = configManager.get("c2s", "port").value_or(5222); + uint16_t listeningPort = static_cast(portValue); + + // Initialize socket connection listener + std::cout << "[INFO] Starting XMPP Client-to-Server (C2S) listener on port " + << listeningPort << std::endl; + xtrpg::network::SocketConnectionListener listener(ioContext, listeningPort); + + // Register connection handler as observer for incoming connections + listener.addObserver(connectionManager); + + // Start accepting connections + listener.start(); + + // Set up signal handlers for graceful shutdown (SIGINT and SIGTERM) +#ifdef _WIN32 + signal(SIGINT, signalHandler); + signal(SIGTERM, signalHandler); +#else + signal(SIGINT, signalHandler); + signal(SIGTERM, signalHandler); + signal(SIGHUP, signalHandler); +#endif + + std::cout << "[INFO] Server running. Press Ctrl+C to shutdown." + << std::endl; + + // Run IO context in a thread pool for handling async operations + std::vector ioThreads; + const size_t threadCount = std::thread::hardware_concurrency(); + for (size_t i = 0; i < threadCount; ++i) { + ioThreads.emplace_back([&ioContext]() { ioContext.run(); }); + } + + // Main thread: wait for shutdown signal + while (!g_shouldShutdown) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + // Graceful shutdown: stop listener and wait for pending operations + std::cout << "[INFO] Stopping listener..." << std::endl; + listener.stop(); + + std::cout << "[INFO] Shutting down IO context..." << std::endl; + ioContext.stop(); + + // Wait for all IO threads to complete + for (auto &thread : ioThreads) { + if (thread.joinable()) { + thread.join(); + } + } + + std::cout << "[INFO] Server shutdown complete." << std::endl; + } catch (const std::exception &e) { std::cerr << "EXCEPTION OCCURRED" << std::endl << "Application closing die to \"" << e.what() << "\"." << std ::endl; return 1; + } catch (...) { + std::cerr << "UNEXPECTED EXCEPTION OCCURRED" << std::endl + << "Application closing." << std ::endl; + return 1; } return 0; From 2ed4a71d6a642e0c1b0863dd61a6a5fe05dc655e Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Sat, 29 Aug 2026 17:03:50 +1000 Subject: [PATCH 5/9] Add debug logging to ClientSession XML handlers Add std::cout debug output to ClientSession's XML handler methods (openDeclaration, closeDeclaration, setAttribute, appendText) to aid troubleshooting of XML tokenization/parsing. This is a non-functional change that only emits diagnostic logs when those hooks are invoked. --- src/xmpp/session/ClientSession.cpp | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/xmpp/session/ClientSession.cpp b/src/xmpp/session/ClientSession.cpp index 5e5f554..f54f586 100644 --- a/src/xmpp/session/ClientSession.cpp +++ b/src/xmpp/session/ClientSession.cpp @@ -84,13 +84,20 @@ void ClientSession::openTag(std::string_view tagname) { } void ClientSession::closeTag() {} void ClientSession::openDeclaration(std::string_view tagname) { + std::cout << "[ClientSession] Open Declaration: " << tagname << std::endl; // ignore declaration tags } void ClientSession::closeDeclaration() { + std::cout << "[ClientSession] Close Declaration. " << std::endl; // ignore declaration tags } void ClientSession::setAttribute(std::string_view name, - std::string_view value) {} -void ClientSession::appendText(std::string_view content) {} + std::string_view value) { + std::cout << "[ClientSession] Set Attribute: " << name << "=" << value + << std::endl; +} +void ClientSession::appendText(std::string_view content) { + std::cout << "[ClientSession] Append Text: `" << content << "`" << std::endl; +} void ClientSession::onError(xml::tokenizer::TokenizationError error) {} } // namespace xtrpg::xmpp::session \ No newline at end of file From 9e71fb90ecdb42c0a50130fb5304bd150ceb8c0e Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Sat, 29 Aug 2026 18:10:15 +1000 Subject: [PATCH 6/9] Refactor tokenizer observer model; add XmlToken MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the old multi-observer pattern with a single raw-pointer observer API (Observable::setObserver) and simplify dispatching. Introduce XmlToken and TokenType to represent parsed XML tokens. XmlStreamTokenizer now builds and emits XmlToken instances (and text/comments) via the Observable interface instead of calling listener methods. XmlTokenListener was converted to observer interfaces for XmlToken and TokenizationError. Update ClientSession and main to use the new setObserver/onObservation API and adjust includes. This is an API-level refactor to centralize token events and streamline tokenizer → session communication. --- apps/main.cpp | 6 +- include/xtrpg/interface/Observable.hpp | 48 ++---- .../xml/tokenizer/XmlStreamTokenizer.hpp | 19 ++- include/xtrpg/xml/tokenizer/XmlToken.hpp | 53 +++++++ .../xtrpg/xml/tokenizer/XmlTokenListener.hpp | 43 +----- include/xtrpg/xmpp/session/ClientSession.hpp | 12 +- src/xml/tokenizer/XmlStreamTokenizer.cpp | 138 ++++++++++-------- src/xmpp/session/ClientSession.cpp | 50 +++---- 8 files changed, 186 insertions(+), 183 deletions(-) create mode 100644 include/xtrpg/xml/tokenizer/XmlToken.hpp diff --git a/apps/main.cpp b/apps/main.cpp index 4ac0c76..1725096 100644 --- a/apps/main.cpp +++ b/apps/main.cpp @@ -72,8 +72,7 @@ int main(int argc, char *argv[]) { // Create temporary connection handler to listen for incoming connections // and create ClientSession instances - auto connectionManager = - std::make_shared(ioContext); + xtrpg::xmpp::ClientConnectionManager connectionManager(ioContext); // Get the port from configuration (default 5222 for XMPP C2S) auto portValue = configManager.get("c2s", "port").value_or(5222); @@ -85,7 +84,7 @@ int main(int argc, char *argv[]) { xtrpg::network::SocketConnectionListener listener(ioContext, listeningPort); // Register connection handler as observer for incoming connections - listener.addObserver(connectionManager); + listener.setObserver(&connectionManager); // Start accepting connections listener.start(); @@ -118,6 +117,7 @@ int main(int argc, char *argv[]) { // Graceful shutdown: stop listener and wait for pending operations std::cout << "[INFO] Stopping listener..." << std::endl; listener.stop(); + listener.setObserver(nullptr); std::cout << "[INFO] Shutting down IO context..." << std::endl; ioContext.stop(); diff --git a/include/xtrpg/interface/Observable.hpp b/include/xtrpg/interface/Observable.hpp index 1ac61c0..d65c38d 100644 --- a/include/xtrpg/interface/Observable.hpp +++ b/include/xtrpg/interface/Observable.hpp @@ -10,48 +10,26 @@ namespace xtrpg::interface { template class Observable { public: - virtual ~Observable() = default; - - void addObserver(const std::shared_ptr> &ptr) { - if (ptr) { - this->_observers.push_back(ptr); - } - } - - void eraseObserver(const Observer *pTarget) { - if (!pTarget) { - return; + virtual ~Observable() { + if (nullptr != this->_ptrObserver) { + std::cerr << "Observable not removed from an instance. This may lead to " + "memory leaks." + << std::endl; } + }; - std::erase_if(this->_observers, - [pTarget](const std::weak_ptr> &wp) { - auto sp = wp.lock(); - return !sp || sp.get() == pTarget; - }); - } - - void clearObservers() { this->_observers.clear(); } + void setObserver(Observer *ptr) { this->_ptrObserver = ptr; } protected: void dispatchObservation(TContext &ctx) { - std::erase_if(this->_observers, - [&ctx](const std::weak_ptr> &wp) { - // Attempt to gain temporary ownership of the current - // observer. - if (auto observer = wp.lock()) { - // Dispatch the observation. - observer->onObservation(ctx); - - // Pointer is still valid, keep it in the vector. - return false; - } + if (nullptr == this->_ptrObserver) { + return; + } - // Pointer has expired, remove it from the observers vector. - return true; - }); + this->_ptrObserver->onObservation(ctx); } private: - std::vector>> _observers; + Observer *_ptrObserver; }; -} // namespace xtrpg \ No newline at end of file +} // namespace xtrpg::interface \ No newline at end of file diff --git a/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp b/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp index e5877d6..56d8323 100644 --- a/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp +++ b/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp @@ -6,7 +6,9 @@ #include #include +#include "xtrpg/interface/Observable.hpp" #include "xtrpg/xml/tokenizer/TokenizationError.hpp" +#include "xtrpg/xml/tokenizer/XmlToken.hpp" #include "xtrpg/xml/tokenizer/XmlTokenListener.hpp" #ifndef __TOKENIZER_MAX_BUFFER_SIZE_IN_CHARS @@ -18,21 +20,15 @@ namespace xtrpg::xml::tokenizer { * A processor class that processing a stream of XML data and fires off * tokenization events to a registered listener. */ -class XmlStreamTokenizer { +class XmlStreamTokenizer : public interface::Observable { public: + ~XmlStreamTokenizer() = default; + /** * Consumes the data on the provided stream until it's exhausted. */ void process(std::istream &stream); - /** - * Defines a instance to act as the listener for this class. - */ - void setListener(XmlTokenListener *listener) { - std::cout << "[XmlStreamTokenizer] Assign listener." << std::endl; - _listener = listener; - } - private: enum class State { TEXT, @@ -70,7 +66,10 @@ class XmlStreamTokenizer { TokenizationError _error{TokenizationError::NONE}; - XmlTokenListener *_listener = nullptr; + /** + * The current token being parsed. + */ + XmlToken _currentToken{}; }; } // namespace xtrpg::xml::tokenizer \ No newline at end of file diff --git a/include/xtrpg/xml/tokenizer/XmlToken.hpp b/include/xtrpg/xml/tokenizer/XmlToken.hpp new file mode 100644 index 0000000..235e0b4 --- /dev/null +++ b/include/xtrpg/xml/tokenizer/XmlToken.hpp @@ -0,0 +1,53 @@ +#pragma once + +#include +#include + +namespace xtrpg::xml::tokenizer { + +enum class TokenType { + /** + * Represents an OPEN XML Tag with attributes (eg: ""). + * The content field will equal the name of the tag (eg: "myTag"). + */ + OPEN_TAG, + + /** + * Represents a CLOSE XML Tag (eg ""). + * The content field will equal the name of the tag (eg: "myTag"). The + * attributes field must be empty. + */ + CLOSE_TAG, + + /** + * Represents an empty (or self-closing) XML Tag (eg: ""). The content field will equal the name of the tag (eg: "myTag"). + */ + EMPTY_TAG, + + /** + * Represents a declaration tag (eg: "<%xml version='1.0' %>"). + * The content field will equal the name of the tag (eg: "xml"). + */ + DECLARATION, + + /** + * Represents the raw text conent inside a XML tag. The raw text up to the + * close tag or the next child element. The content field will equal the + * contents of the text node. Attributes will generally be empty. + */ + TEXT_CONTENT, + + /** + * Represents an XML comment. The content field will equal the comment message + * itself. The attributes field must be empty. + */ + COMMENT, +}; + +struct XmlToken { + TokenType type; + std::string content; + std::unordered_map attributes; +}; +} // namespace xtrpg::xml::tokenizer diff --git a/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp b/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp index 1bd3938..1c33f78 100644 --- a/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp +++ b/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp @@ -2,6 +2,7 @@ #include +#include "xtrpg/interface/Observer.hpp" #include "xtrpg/xml/tokenizer/TokenizationError.hpp" namespace xtrpg::xml::tokenizer { @@ -9,47 +10,15 @@ namespace xtrpg::xml::tokenizer { * Represents a class that is capable of processing a stream of XML Token * events. */ -class XmlTokenListener { +class XmlTokenListener + : public interface::Observer, + public interface::Observer { public: virtual ~XmlTokenListener() = default; - /** - * Indication to the listener instance that it should begin processing a new - * XML Tag Node, with the given tagname. If the listener already has an open - * tag then it should assign the existing tag to be the parent of this new - * tag. - */ - virtual void openTag(std::string_view tagname) = 0; - - /** - * Indication that the current tag has finished processing and focus should be - * returned to it's parent. - */ - virtual void closeTag() = 0; - - /** - * Indication to the listener that it should begin processing a new XML - * Declaration Tag Node, with the given tagname. - */ - virtual void openDeclaration(std::string_view tagname) = 0; + virtual void onObservation(const xml::tokenizer::XmlToken &xmlToken) = 0; - /** - * Indication that the current declaration tag has finished processing and - * focus should be returned to it's parent. - */ - virtual void closeDeclaration() = 0; - - /** - * Defines an attribute (key/value pair) that should be assigned to the - * current tag or declaration tag. - */ - virtual void setAttribute(std::string_view name, std::string_view value) = 0; - - /** - * Defines a block of text that should be applied as a Raw Text Child Node of - * the current tag. - */ - virtual void appendText(std::string_view content) = 0; + virtual void onObservation(const TokenizationError &error) = 0; /** * Defines an error state of the tokenizer. diff --git a/include/xtrpg/xmpp/session/ClientSession.hpp b/include/xtrpg/xmpp/session/ClientSession.hpp index b2cab2b..28caea5 100644 --- a/include/xtrpg/xmpp/session/ClientSession.hpp +++ b/include/xtrpg/xmpp/session/ClientSession.hpp @@ -8,6 +8,7 @@ #include "xtrpg/xml/node/TagNode.hpp" #include "xtrpg/xml/tokenizer/TokenizationError.hpp" #include "xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp" +#include "xtrpg/xml/tokenizer/XmlToken.hpp" #include "xtrpg/xml/tokenizer/XmlTokenListener.hpp" namespace xtrpg::xmpp::session { @@ -19,7 +20,7 @@ class ClientSession : public xml::tokenizer::XmlTokenListener, ClientSession(network::TcpConnection tcpConnection) : _tcpConnection(std::move(tcpConnection)) { std::cout << "[ClientSession] New Instance created." << std::endl; - this->_tokenizer.setListener(this); + this->_tokenizer.setObserver(this); } ~ClientSession(); @@ -33,12 +34,9 @@ class ClientSession : public xml::tokenizer::XmlTokenListener, void sendRaw(std::string_view data); // Tokenizer Calls - void openTag(std::string_view tagname); - void closeTag(); - void openDeclaration(std::string_view tagname); - void closeDeclaration(); - void setAttribute(std::string_view name, std::string_view value); - void appendText(std::string_view content); + void onObservation(const xml::tokenizer::XmlToken &xmlToken); + void onObservation(const xml::tokenizer::TokenizationError &error); + void onError(xml::tokenizer::TokenizationError error); private: diff --git a/src/xml/tokenizer/XmlStreamTokenizer.cpp b/src/xml/tokenizer/XmlStreamTokenizer.cpp index 6628b39..53d39f1 100644 --- a/src/xml/tokenizer/XmlStreamTokenizer.cpp +++ b/src/xml/tokenizer/XmlStreamTokenizer.cpp @@ -18,49 +18,33 @@ bool isWhitespace(const char character) { namespace xtrpg::xml::tokenizer { void XmlStreamTokenizer::process(std::istream &stream) { - // If the tokenizer is already in an error state then re-issue the same error. + // If the tokenizer is already in an error state then cease processing. if (TokenizationError::NONE != this->_error) { - if (nullptr == this->_listener) { - this->_listener->onError(this->_error); - } return; } const auto fail = [&](const TokenizationError error) { this->_error = error; - if (nullptr != this->_listener) { - this->_listener->onError(error); - } - }; - const auto appendText = [&](const std::string_view text) { - if (!text.empty() && nullptr != this->_listener) { - this->_listener->appendText(text); - } }; - const auto openStartTag = [&]() { - if (this->_buffer.empty()) { - fail(TokenizationError::MALFORMED_INPUT); - return; - } - if (nullptr != this->_listener) { - this->_listener->openTag(this->_buffer); - } - this->_buffer.clear(); + + const auto emitToken = [&](const XmlToken &token) { + std::cout << "[XmlStreamTokenizer] Dispatching XML Token: " << token.content + << std::endl; + dispatchObservation(token); }; - const auto openDeclaration = [&]() { - if (this->_buffer.empty()) { - fail(TokenizationError::MALFORMED_INPUT); - return; - } - if (nullptr != this->_listener) { - this->_listener->openDeclaration(this->_buffer); + + const auto emitText = [&](const std::string_view text) { + if (!text.empty()) { + XmlToken token; + token.type = TokenType::TEXT_CONTENT; + token.content = std::string(text); + emitToken(token); } - this->_buffer.clear(); }; + const auto bufferExceeded = [&]() { fail(TokenizationError::BUFFER_SIZE_EXHAUSTED); }; - this->_buffer.reserve(__TOKENIZER_MAX_BUFFER_SIZE_IN_CHARS); this->_attributeName.reserve(128); this->_specialPrefix.reserve(7); @@ -75,29 +59,33 @@ void XmlStreamTokenizer::process(std::istream &stream) { switch (this->_state) { case State::TEXT: if (character == '<') { - appendText(this->_buffer); + emitText(this->_buffer); this->_buffer.clear(); this->_state = State::AFTER_OPEN; } else { this->_buffer += character; if (this->_buffer.size() >= __TOKENIZER_MAX_BUFFER_SIZE_IN_CHARS) { - appendText(this->_buffer); + emitText(this->_buffer); this->_buffer.clear(); } } break; case State::AFTER_OPEN: + this->_currentToken = XmlToken{}; if (character == '/') { this->_buffer.clear(); + this->_currentToken.type = TokenType::CLOSE_TAG; this->_state = State::END_TAG_NAME; } else if (character == '?') { this->_buffer.clear(); + this->_currentToken.type = TokenType::DECLARATION; this->_state = State::DECLARATION_NAME; } else if (character == '!') { this->_specialPrefix.clear(); this->_state = State::SPECIAL; } else if (isNameCharacter(character)) { this->_buffer = character; + this->_currentToken.type = TokenType::OPEN_TAG; this->_state = State::START_TAG_NAME; } else { fail(TokenizationError::MALFORMED_INPUT); @@ -110,14 +98,30 @@ void XmlStreamTokenizer::process(std::istream &stream) { bufferExceeded(); } } else if (isWhitespace(character)) { - openStartTag(); - this->_state = State::START_TAG_BODY; + if (this->_buffer.empty()) { + fail(TokenizationError::MALFORMED_INPUT); + } else { + this->_currentToken.content = this->_buffer; + this->_buffer.clear(); + this->_state = State::START_TAG_BODY; + } } else if (character == '>') { - openStartTag(); - this->_state = State::TEXT; + if (this->_buffer.empty()) { + fail(TokenizationError::MALFORMED_INPUT); + } else { + this->_currentToken.content = this->_buffer; + this->_buffer.clear(); + emitToken(this->_currentToken); + this->_state = State::TEXT; + } } else if (character == '/') { - openStartTag(); - this->_state = State::SELF_CLOSING; + if (this->_buffer.empty()) { + fail(TokenizationError::MALFORMED_INPUT); + } else { + this->_currentToken.content = this->_buffer; + this->_buffer.clear(); + this->_state = State::SELF_CLOSING; + } } else { fail(TokenizationError::MALFORMED_INPUT); } @@ -127,6 +131,7 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; } if (character == '>') { + emitToken(this->_currentToken); this->_state = State::TEXT; } else if (character == '/') { this->_state = State::SELF_CLOSING; @@ -179,9 +184,7 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::ATTRIBUTE_VALUE: if (character == this->_quote) { - if (nullptr != this->_listener) { - this->_listener->setAttribute(this->_attributeName, this->_buffer); - } + this->_currentToken.attributes[this->_attributeName] = this->_buffer; this->_attributeName.clear(); this->_buffer.clear(); this->_state = State::START_TAG_BODY; @@ -199,15 +202,16 @@ void XmlStreamTokenizer::process(std::istream &stream) { bufferExceeded(); } } else if (isWhitespace(character)) { + this->_currentToken.content = this->_buffer; + this->_buffer.clear(); this->_state = State::END_TAG_BODY; } else if (character == '>') { if (this->_buffer.empty()) { fail(TokenizationError::MALFORMED_INPUT); } else { - if (nullptr != this->_listener) { - this->_listener->closeTag(); - } + this->_currentToken.content = this->_buffer; this->_buffer.clear(); + emitToken(this->_currentToken); this->_state = State::TEXT; } } else { @@ -218,11 +222,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { if (isWhitespace(character)) { break; } - if (character == '>' && !this->_buffer.empty()) { - if (nullptr != this->_listener) { - this->_listener->closeTag(); - } - this->_buffer.clear(); + if (character == '>') { + emitToken(this->_currentToken); this->_state = State::TEXT; } else { fail(TokenizationError::MALFORMED_INPUT); @@ -235,11 +236,21 @@ void XmlStreamTokenizer::process(std::istream &stream) { bufferExceeded(); } } else if (isWhitespace(character)) { - openDeclaration(); - this->_state = State::DECLARATION_BODY; + if (this->_buffer.empty()) { + fail(TokenizationError::MALFORMED_INPUT); + } else { + this->_currentToken.content = this->_buffer; + this->_buffer.clear(); + this->_state = State::DECLARATION_BODY; + } } else if (character == '?') { - openDeclaration(); - this->_state = State::DECLARATION_QUESTION; + if (this->_buffer.empty()) { + fail(TokenizationError::MALFORMED_INPUT); + } else { + this->_currentToken.content = this->_buffer; + this->_buffer.clear(); + this->_state = State::DECLARATION_QUESTION; + } } else { fail(TokenizationError::MALFORMED_INPUT); } @@ -299,9 +310,7 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::DECLARATION_ATTRIBUTE_VALUE: if (character == this->_quote) { - if (nullptr != this->_listener) { - this->_listener->setAttribute(this->_attributeName, this->_buffer); - } + this->_currentToken.attributes[this->_attributeName] = this->_buffer; this->_attributeName.clear(); this->_buffer.clear(); this->_state = State::DECLARATION_BODY; @@ -314,9 +323,7 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::DECLARATION_QUESTION: if (character == '>') { - if (nullptr != this->_listener) { - this->_listener->closeDeclaration(); - } + emitToken(this->_currentToken); this->_state = State::TEXT; } else { fail(TokenizationError::MALFORMED_INPUT); @@ -324,9 +331,8 @@ void XmlStreamTokenizer::process(std::istream &stream) { break; case State::SELF_CLOSING: if (character == '>') { - if (nullptr != this->_listener) { - this->_listener->closeTag(); - } + this->_currentToken.type = TokenType::EMPTY_TAG; + emitToken(this->_currentToken); this->_state = State::TEXT; } else { fail(TokenizationError::MALFORMED_INPUT); @@ -351,6 +357,10 @@ void XmlStreamTokenizer::process(std::istream &stream) { case State::COMMENT: this->_buffer += character; if (this->_buffer.size() >= 3 && this->_buffer.ends_with("-->")) { + XmlToken token; + token.type = TokenType::COMMENT; + token.content = this->_buffer.substr(0, this->_buffer.size() - 3); + emitToken(token); this->_buffer.clear(); this->_state = State::TEXT; } else if (this->_buffer.size() > @@ -362,13 +372,13 @@ void XmlStreamTokenizer::process(std::istream &stream) { this->_buffer += character; if (this->_buffer.size() >= 3 && this->_buffer.ends_with("]]>")) { this->_buffer.resize(this->_buffer.size() - 3); - appendText(this->_buffer); + emitText(this->_buffer); this->_buffer.clear(); this->_state = State::TEXT; } else if (this->_buffer.size() >= __TOKENIZER_MAX_BUFFER_SIZE_IN_CHARS) { const auto textSize = this->_buffer.size() - 2; - appendText(std::string_view(this->_buffer.data(), textSize)); + emitText(std::string_view(this->_buffer.data(), textSize)); const char penultimate = this->_buffer[this->_buffer.size() - 2]; const char last = this->_buffer[this->_buffer.size() - 1]; this->_buffer.clear(); diff --git a/src/xmpp/session/ClientSession.cpp b/src/xmpp/session/ClientSession.cpp index f54f586..ec49fe1 100644 --- a/src/xmpp/session/ClientSession.cpp +++ b/src/xmpp/session/ClientSession.cpp @@ -2,6 +2,8 @@ #include +#include "xtrpg/xml/tokenizer/XmlToken.hpp" + namespace xtrpg::xmpp::session { ClientSession::~ClientSession() { @@ -18,7 +20,7 @@ ClientSession::~ClientSession() { } // remove myself from the tokenizer - this->_tokenizer.setListener(nullptr); + this->_tokenizer.setObserver(nullptr); } void ClientSession::start() { @@ -71,33 +73,27 @@ void ClientSession::process() { }); } -void ClientSession::openTag(std::string_view tagname) { - std::cout << "[Client Session] Receive tag: " << tagname << std::endl; - - this->sendRaw( - "Stanza size " - "limit of 64KB exceeded."); -} -void ClientSession::closeTag() {} -void ClientSession::openDeclaration(std::string_view tagname) { - std::cout << "[ClientSession] Open Declaration: " << tagname << std::endl; - // ignore declaration tags -} -void ClientSession::closeDeclaration() { - std::cout << "[ClientSession] Close Declaration. " << std::endl; - // ignore declaration tags -} -void ClientSession::setAttribute(std::string_view name, - std::string_view value) { - std::cout << "[ClientSession] Set Attribute: " << name << "=" << value +void ClientSession::onObservation(const xml::tokenizer::XmlToken &xmlToken) { + std::cout << "[ClientSession] Observed XML Token: " << xmlToken.content << std::endl; + + for (const auto &[key, value] : xmlToken.attributes) { + std::cout << " - " << key << ": " << value << std::endl; + } + + if (xml::tokenizer::TokenType::OPEN_TAG == xmlToken.type && + "stream:stream" == xmlToken.content) { + this->sendRaw( + "Stanza size " + "limit of 64KB exceeded."); + } } -void ClientSession::appendText(std::string_view content) { - std::cout << "[ClientSession] Append Text: `" << content << "`" << std::endl; -} +void ClientSession::onObservation( + const xml::tokenizer::TokenizationError &error) {} + void ClientSession::onError(xml::tokenizer::TokenizationError error) {} } // namespace xtrpg::xmpp::session \ No newline at end of file From 6ac89ee159a3c65c411a7a44b611290267fb96bd Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Sat, 29 Aug 2026 18:25:34 +1000 Subject: [PATCH 7/9] fix: remove type erasure issue --- .../xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp | 16 +++++++++++++--- include/xtrpg/xml/tokenizer/XmlTokenListener.hpp | 14 +++----------- include/xtrpg/xmpp/session/ClientSession.hpp | 6 ++---- src/xml/tokenizer/XmlStreamTokenizer.cpp | 5 +++-- src/xmpp/session/ClientSession.cpp | 5 ++--- 5 files changed, 23 insertions(+), 23 deletions(-) diff --git a/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp b/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp index 56d8323..1bf9f03 100644 --- a/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp +++ b/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp @@ -6,7 +6,6 @@ #include #include -#include "xtrpg/interface/Observable.hpp" #include "xtrpg/xml/tokenizer/TokenizationError.hpp" #include "xtrpg/xml/tokenizer/XmlToken.hpp" #include "xtrpg/xml/tokenizer/XmlTokenListener.hpp" @@ -20,16 +19,27 @@ namespace xtrpg::xml::tokenizer { * A processor class that processing a stream of XML data and fires off * tokenization events to a registered listener. */ -class XmlStreamTokenizer : public interface::Observable { +class XmlStreamTokenizer { public: - ~XmlStreamTokenizer() = default; + ~XmlStreamTokenizer() { + if (nullptr != this->_ptrObserver) { + std::cerr << "Observable not removed from an instance of " + "XmlStreamTokenizer. This may lead to " + "memory leaks." + << std::endl; + } + }; /** * Consumes the data on the provided stream until it's exhausted. */ void process(std::istream &stream); + void setObserver(XmlTokenListener *ptr) { this->_ptrObserver = ptr; } + private: + XmlTokenListener *_ptrObserver = nullptr; + enum class State { TEXT, AFTER_OPEN, diff --git a/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp b/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp index 1c33f78..ac7b901 100644 --- a/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp +++ b/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp @@ -2,7 +2,6 @@ #include -#include "xtrpg/interface/Observer.hpp" #include "xtrpg/xml/tokenizer/TokenizationError.hpp" namespace xtrpg::xml::tokenizer { @@ -10,19 +9,12 @@ namespace xtrpg::xml::tokenizer { * Represents a class that is capable of processing a stream of XML Token * events. */ -class XmlTokenListener - : public interface::Observer, - public interface::Observer { +class XmlTokenListener { public: virtual ~XmlTokenListener() = default; - virtual void onObservation(const xml::tokenizer::XmlToken &xmlToken) = 0; + virtual void onXmlToken(const xml::tokenizer::XmlToken &xmlToken) = 0; - virtual void onObservation(const TokenizationError &error) = 0; - - /** - * Defines an error state of the tokenizer. - */ - virtual void onError(TokenizationError error) = 0; + virtual void onTokenizationError(const TokenizationError &error) = 0; }; } // namespace xtrpg::xml::tokenizer \ No newline at end of file diff --git a/include/xtrpg/xmpp/session/ClientSession.hpp b/include/xtrpg/xmpp/session/ClientSession.hpp index 28caea5..7bb15c9 100644 --- a/include/xtrpg/xmpp/session/ClientSession.hpp +++ b/include/xtrpg/xmpp/session/ClientSession.hpp @@ -34,10 +34,8 @@ class ClientSession : public xml::tokenizer::XmlTokenListener, void sendRaw(std::string_view data); // Tokenizer Calls - void onObservation(const xml::tokenizer::XmlToken &xmlToken); - void onObservation(const xml::tokenizer::TokenizationError &error); - - void onError(xml::tokenizer::TokenizationError error); + void onXmlToken(const xml::tokenizer::XmlToken &xmlToken); + void onTokenizationError(const xml::tokenizer::TokenizationError &error); private: network::TcpConnection _tcpConnection; diff --git a/src/xml/tokenizer/XmlStreamTokenizer.cpp b/src/xml/tokenizer/XmlStreamTokenizer.cpp index 53d39f1..1e84266 100644 --- a/src/xml/tokenizer/XmlStreamTokenizer.cpp +++ b/src/xml/tokenizer/XmlStreamTokenizer.cpp @@ -23,14 +23,15 @@ void XmlStreamTokenizer::process(std::istream &stream) { return; } - const auto fail = [&](const TokenizationError error) { + const auto fail = [&](const TokenizationError &error) { this->_error = error; + this->_ptrObserver->onTokenizationError(error); }; const auto emitToken = [&](const XmlToken &token) { std::cout << "[XmlStreamTokenizer] Dispatching XML Token: " << token.content << std::endl; - dispatchObservation(token); + this->_ptrObserver->onXmlToken(token); }; const auto emitText = [&](const std::string_view text) { diff --git a/src/xmpp/session/ClientSession.cpp b/src/xmpp/session/ClientSession.cpp index ec49fe1..2b28b89 100644 --- a/src/xmpp/session/ClientSession.cpp +++ b/src/xmpp/session/ClientSession.cpp @@ -73,7 +73,7 @@ void ClientSession::process() { }); } -void ClientSession::onObservation(const xml::tokenizer::XmlToken &xmlToken) { +void ClientSession::onXmlToken(const xml::tokenizer::XmlToken &xmlToken) { std::cout << "[ClientSession] Observed XML Token: " << xmlToken.content << std::endl; @@ -92,8 +92,7 @@ void ClientSession::onObservation(const xml::tokenizer::XmlToken &xmlToken) { "limit of 64KB exceeded."); } } -void ClientSession::onObservation( +void ClientSession::onTokenizationError( const xml::tokenizer::TokenizationError &error) {} -void ClientSession::onError(xml::tokenizer::TokenizationError error) {} } // namespace xtrpg::xmpp::session \ No newline at end of file From 37f6053e5eba0e06c3f822971b85a484cccd0f7c Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Fri, 4 Sep 2026 19:12:42 +1000 Subject: [PATCH 8/9] Refactor connection/session lifecycle and observers Introduce robust lifecycle management for TCP connections and client sessions plus observer API and threading improvements. Main changes: - Make Observable thread-safe; dispatchObservation returns bool and Observer takes context by value. - SocketConnectionListener now accepts/io_context pointer, emits raw TcpConnection* and only keeps connections when observed. - TcpConnection: new read(error_code, stream) callback, cancelRead(), close(callback), state-change and close callbacks, atomic state, safer async/error handling. - ClientConnectionManager: owns listener, manages sessions with mutexes, graceful shutdown, onObservation(TcpConnection*), countConnections(). - ClientSession: owns TcpConnection*, updated start/stop/shutdown and completion notification. - apps/main updated to construct manager with port and print connection count. Overall: safer shutdown, clearer ownership, and synchronized callbacks. --- apps/main.cpp | 34 ++--- include/xtrpg/interface/Observable.hpp | 15 ++- include/xtrpg/interface/Observer.hpp | 2 +- .../network/SocketConnectionListener.hpp | 9 +- include/xtrpg/network/TcpConnection.hpp | 41 +++++- .../xtrpg/xmpp/ClientConnectionManager.hpp | 59 +++++++- include/xtrpg/xmpp/Jid.hpp | 1 - include/xtrpg/xmpp/session/ClientSession.hpp | 82 ++++++++++-- src/network/SocketConnectionListener.cpp | 16 ++- src/network/TcpConnection.cpp | 126 +++++++++++++----- src/xmpp/ClientConnectionManager.cpp | 124 +++++++++++++++-- src/xmpp/session/ClientSession.cpp | 68 ++++++---- 12 files changed, 449 insertions(+), 128 deletions(-) diff --git a/apps/main.cpp b/apps/main.cpp index 1725096..0e5bacd 100644 --- a/apps/main.cpp +++ b/apps/main.cpp @@ -70,32 +70,19 @@ int main(int argc, char *argv[]) { // Initialize Asio IO context for async I/O operations asio::io_context ioContext; - // Create temporary connection handler to listen for incoming connections - // and create ClientSession instances - xtrpg::xmpp::ClientConnectionManager connectionManager(ioContext); - // Get the port from configuration (default 5222 for XMPP C2S) auto portValue = configManager.get("c2s", "port").value_or(5222); uint16_t listeningPort = static_cast(portValue); - // Initialize socket connection listener - std::cout << "[INFO] Starting XMPP Client-to-Server (C2S) listener on port " - << listeningPort << std::endl; - xtrpg::network::SocketConnectionListener listener(ioContext, listeningPort); - - // Register connection handler as observer for incoming connections - listener.setObserver(&connectionManager); - - // Start accepting connections - listener.start(); + // Create temporary connection handler to listen for incoming connections + // and create ClientSession instances + xtrpg::xmpp::ClientConnectionManager connectionManager(ioContext, + listeningPort); // Set up signal handlers for graceful shutdown (SIGINT and SIGTERM) -#ifdef _WIN32 - signal(SIGINT, signalHandler); - signal(SIGTERM, signalHandler); -#else signal(SIGINT, signalHandler); signal(SIGTERM, signalHandler); +#ifndef _WIN32 signal(SIGHUP, signalHandler); #endif @@ -110,15 +97,16 @@ int main(int argc, char *argv[]) { } // Main thread: wait for shutdown signal + int previousCount = -1; while (!g_shouldShutdown) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); + if (previousCount != connectionManager.countConnections()) { + std::cout << "[main] Connection Count: " + << connectionManager.countConnections() << std::endl; + previousCount = connectionManager.countConnections(); + } } - // Graceful shutdown: stop listener and wait for pending operations - std::cout << "[INFO] Stopping listener..." << std::endl; - listener.stop(); - listener.setObserver(nullptr); - std::cout << "[INFO] Shutting down IO context..." << std::endl; ioContext.stop(); diff --git a/include/xtrpg/interface/Observable.hpp b/include/xtrpg/interface/Observable.hpp index d65c38d..201ecdd 100644 --- a/include/xtrpg/interface/Observable.hpp +++ b/include/xtrpg/interface/Observable.hpp @@ -2,6 +2,7 @@ #include #include +#include #include #include @@ -18,18 +19,24 @@ template class Observable { } }; - void setObserver(Observer *ptr) { this->_ptrObserver = ptr; } + void setObserver(Observer *ptr) { + std::lock_guard lock(this->_observerMutex); + this->_ptrObserver = ptr; + } protected: - void dispatchObservation(TContext &ctx) { + bool dispatchObservation(TContext &ctx) { + std::lock_guard lock(this->_observerMutex); if (nullptr == this->_ptrObserver) { - return; + return false; } this->_ptrObserver->onObservation(ctx); + return true; } private: - Observer *_ptrObserver; + Observer *_ptrObserver = nullptr; + std::mutex _observerMutex; }; } // namespace xtrpg::interface \ No newline at end of file diff --git a/include/xtrpg/interface/Observer.hpp b/include/xtrpg/interface/Observer.hpp index b6e9234..1250cfd 100644 --- a/include/xtrpg/interface/Observer.hpp +++ b/include/xtrpg/interface/Observer.hpp @@ -8,6 +8,6 @@ template class Observer { virtual ~Observer() = default; // Generic handler callback - virtual void onObservation(TContext &ctx) = 0; + virtual void onObservation(TContext ctx) = 0; }; } // namespace xtrpg::interface \ No newline at end of file diff --git a/include/xtrpg/network/SocketConnectionListener.hpp b/include/xtrpg/network/SocketConnectionListener.hpp index 885e9ea..0acfa22 100644 --- a/include/xtrpg/network/SocketConnectionListener.hpp +++ b/include/xtrpg/network/SocketConnectionListener.hpp @@ -11,14 +11,13 @@ #include "xtrpg/network/TcpConnection.hpp" namespace xtrpg::network { -class SocketConnectionListener - : public interface::Observable> { +class SocketConnectionListener : public interface::Observable { public: /** * Instantiates a new listener instance. */ - SocketConnectionListener(asio::io_context &ioContext, uint16_t port) - : _ioContext(ioContext), _port(port), _isStopped(true) { + SocketConnectionListener(asio::io_context *ptrIoContext, uint16_t port) + : _ptrIoContext(ptrIoContext), _port(port), _isStopped(true) { this->initializeAcceptors(); } @@ -32,7 +31,7 @@ class SocketConnectionListener void acceptIPv4Connections(); void acceptIPv6Connections(); - asio::io_context &_ioContext; + asio::io_context *_ptrIoContext; uint16_t _port; std::optional _ipv4Acceptor; std::optional _ipv6Acceptor; diff --git a/include/xtrpg/network/TcpConnection.hpp b/include/xtrpg/network/TcpConnection.hpp index 23cd814..e585652 100644 --- a/include/xtrpg/network/TcpConnection.hpp +++ b/include/xtrpg/network/TcpConnection.hpp @@ -2,9 +2,12 @@ #include #include +#include +#include #include #include #include +#include #include "xtrpg/network/exception/ConnectionClosed.hpp" @@ -42,7 +45,7 @@ class TcpConnection { /** * Constructs a TcpConnection with the given TCP socket. */ - explicit TcpConnection(asio::ip::tcp::socket tcpSocket) + explicit TcpConnection(asio::ip::tcp::socket &tcpSocket) : _tcpSocket(std::move(tcpSocket)) { this->_strand.emplace(asio::make_strand(this->_tcpSocket.get_executor())); } @@ -57,7 +60,11 @@ class TcpConnection { * Async read from the underlying tcp connection, calling the provided lambda * function with a new istream of the incoming stream data. */ - void read(std::function callback); + void + read(std::function callback); + + /** Cancels the currently pending read operation, if any. */ + void cancelRead(); /** * Writes data to the connection. If the connection is closed, it will throw @@ -65,11 +72,19 @@ class TcpConnection { */ void write(std::string_view data); + /** + * Returns whether the current state of the TCP Connection matches the + * provided argument. + */ + bool is(ConnectionState connectionState) const { + return connectionState == this->_state; + } + /** * Closes the connection. If the connection is already closed, it will do * nothing. */ - void close(); + void close(std::function callback = {}); /** * Checks if the connection is secure (SSL/TLS). @@ -112,11 +127,29 @@ class TcpConnection { return *this; } + void + appendStateChangeCallback(std::function callback) { + std::cout << "[TcpConnection] Append State Change Callback." << std::endl; + this->_stateChangeCallbacks.push_back(callback); + } + private: + std::vector> _stateChangeCallbacks; + std::vector> _closeCallbacks; + + void dispatchStateChange(ConnectionState newState) { + this->_state = newState; + for (auto &callback : this->_stateChangeCallbacks) { + callback(newState); + } + } + + void dispatchCloseCallbacks(); + /** * The current state of the connection. */ - ConnectionState _state{ConnectionState::INSECURE}; + std::atomic _state{ConnectionState::INSECURE}; /** * The underlying TCP socket used for the connection. diff --git a/include/xtrpg/xmpp/ClientConnectionManager.hpp b/include/xtrpg/xmpp/ClientConnectionManager.hpp index 779e740..a0836a6 100644 --- a/include/xtrpg/xmpp/ClientConnectionManager.hpp +++ b/include/xtrpg/xmpp/ClientConnectionManager.hpp @@ -1,36 +1,63 @@ #pragma once +#include #include -#include +#include +#include +#include #include #include "xtrpg/config/ConfigManager.hpp" #include "xtrpg/interface/Observer.hpp" +#include "xtrpg/network/SocketConnectionListener.hpp" #include "xtrpg/network/TcpConnection.hpp" #include "xtrpg/xmpp/session/ClientSession.hpp" namespace xtrpg::xmpp { /** + * Owns active XMPP client sessions and accepts new client connections. * + * The manager owns each session created from an accepted TCP connection and + * removes it when the session completes. Listener and session callbacks are + * synchronized with destruction so the manager remains valid while callbacks + * are in flight. */ class ClientConnectionManager : public config::IModuleConfigProvider, - public interface::Observer> { + public interface::Observer { public: + /** Creates an inactive manager without a connection listener. */ ClientConnectionManager() = default; + /** + * Stops accepting connections and releases all owned client sessions. + * + * The owner must stop and join every thread calling run() on the supplied + * I/O context before destroying this manager. The destructor drains pending + * handlers itself and cannot synchronize with handlers running concurrently + * on an I/O-context thread it does not own. + */ ~ClientConnectionManager(); - explicit ClientConnectionManager(asio::io_context &ioContext) - : _ioContext(&ioContext) {} - /** - * A new Tcp Connection is created. + * Starts listening for client connections on the supplied I/O context. + * + * @param ioContext I/O context used for asynchronous network operations + * @param port TCP port on which to accept client connections */ - void onObservation(std::shared_ptr &ctx); + explicit ClientConnectionManager(asio::io_context &ioContext, + uint16_t port = 5222); /** + * Takes ownership of a newly accepted TCP connection and starts its session. * + * @param ctx newly accepted connection; ownership is transferred to the + * session manager + */ + void onObservation(network::TcpConnection *ctx) override; + + /** + * Returns the schema for the client-to-server listener configuration. */ config::ModuleConfig getConfigSchema() const { return {.name = "c2s", @@ -42,10 +69,28 @@ class ClientConnectionManager "on for Client (or C2S) connections."}}}; } + /** Returns the number of currently registered client sessions. */ + int countConnections() const { + std::shared_lock lock(this->_clientSessionsVectorMutex); + return static_cast(this->_clientSessionPtrs.size()); + } + private: + /** I/O context used by the listener and client sessions. */ asio::io_context *_ioContext = nullptr; + /** Listener that accepts client TCP connections. */ + network::SocketConnectionListener *_ptrSocketConnectionListener = nullptr; + + /** Registry of sessions currently owned by this manager. */ + mutable std::shared_mutex _clientSessionsVectorMutex; std::vector _clientSessionPtrs; + + /** Serializes manager callbacks with shutdown and destruction. */ + mutable std::recursive_mutex _callbackMutex; + + /** Prevents callbacks from entering the manager during destruction. */ + bool _isShuttingDown = false; }; REGISTER_MODULE_CONFIG(ClientConnectionManager); diff --git a/include/xtrpg/xmpp/Jid.hpp b/include/xtrpg/xmpp/Jid.hpp index 9c4d693..fe7810f 100644 --- a/include/xtrpg/xmpp/Jid.hpp +++ b/include/xtrpg/xmpp/Jid.hpp @@ -1,6 +1,5 @@ #pragma once -#include #include #include #include diff --git a/include/xtrpg/xmpp/session/ClientSession.hpp b/include/xtrpg/xmpp/session/ClientSession.hpp index 7bb15c9..eae65b9 100644 --- a/include/xtrpg/xmpp/session/ClientSession.hpp +++ b/include/xtrpg/xmpp/session/ClientSession.hpp @@ -1,7 +1,10 @@ #pragma once #include -#include +#include +#include +#include +#include #include "xtrpg/network/TcpConnection.hpp" #include "xtrpg/xml/node/DeclarationNode.hpp" @@ -13,35 +16,87 @@ namespace xtrpg::xmpp::session { -class ClientSession : public xml::tokenizer::XmlTokenListener, - public std::enable_shared_from_this { +/** + * Coordinates XML tokenization and transport I/O for one XMPP client. + * + * The session does not own the execution context used by the asynchronous + * connection. The caller transfers ownership of @p tcpConnection to this + * object, which deletes it during destruction. The owner must keep the + * session alive until all connection callbacks have completed. + */ +class ClientSession : public xml::tokenizer::XmlTokenListener { public: - ClientSession(network::TcpConnection tcpConnection) - : _tcpConnection(std::move(tcpConnection)) { - std::cout << "[ClientSession] New Instance created." << std::endl; + /** + * Creates a session for an already accepted TCP connection. + * + * @param tcpConnection connection transferred to the new session; must not + * be null + */ + ClientSession(network::TcpConnection *tcpConnection) + : _ptrTcpConnection(tcpConnection) { this->_tokenizer.setObserver(this); } + + /** + * Stops token processing and releases the owned connection and XML state. + * + * The session must not be destroyed while an asynchronous connection + * callback can still invoke it. + */ ~ClientSession(); // Session Control + /** Starts reading and processing data from the client. */ void start(); + + /** Prevents the session from scheduling another read. */ void stop(); + + /** Registers a callback invoked once the pending read has completed. */ + void setCompletionCallback(std::function callback) { + std::lock_guard lock(this->_completionCallbackMutex); + this->_completionCallback = std::move(callback); + } + + /** Stops processing and closes the underlying connection. */ void shutdown(); + + /** Schedules the next asynchronous read when the session is active. */ void process(); - // Transport Control + /** + * Returns whether the owned connection has reached the closed state. + * + * @pre The connection passed to the constructor is still owned by the + * session. + */ + bool isClosed() const { return this->_ptrTcpConnection->isClosed(); } + + /** + * Asynchronously writes raw XML or other protocol data to the client. + * + * @param data bytes to send; the data is copied by the connection layer + */ void sendRaw(std::string_view data); - // Tokenizer Calls + /** Handles one token emitted by the XML stream tokenizer. */ void onXmlToken(const xml::tokenizer::XmlToken &xmlToken); + + /** Handles a tokenizer error reported for this client stream. */ void onTokenizationError(const xml::tokenizer::TokenizationError &error); private: - network::TcpConnection _tcpConnection; + /** TCP connection owned by this session. */ + network::TcpConnection *_ptrTcpConnection; + + /** Stateful tokenizer that retains XML data between network reads. */ xml::tokenizer::XmlStreamTokenizer _tokenizer; + /** Parsed XML declaration, when one is retained by the session. */ xml::node::DeclarationNode *_ptrDeclarationNode = nullptr; + + /** Root XMPP stream node, when one is retained by the session. */ xml::node::TagNode *_ptrRootStreamNode = nullptr; /** @@ -55,5 +110,14 @@ class ClientSession : public xml::tokenizer::XmlTokenListener, * terminated. */ std::atomic _isShutdown{false}; + + /** Called once the session's pending read has completed after shutdown. */ + mutable std::mutex _completionCallbackMutex; + std::function _completionCallback; + + /** Ensures completion is reported at most once. */ + std::atomic _completionNotified{false}; + + void notifyCompletion(); }; } // namespace xtrpg::xmpp::session \ No newline at end of file diff --git a/src/network/SocketConnectionListener.cpp b/src/network/SocketConnectionListener.cpp index 5f0f414..42ef434 100644 --- a/src/network/SocketConnectionListener.cpp +++ b/src/network/SocketConnectionListener.cpp @@ -12,7 +12,7 @@ bool isListenerShutdownError(const std::error_code &ec) { void SocketConnectionListener::initializeAcceptors() { try { asio::ip::tcp::acceptor ipv6Acceptor( - this->_ioContext, + *this->_ptrIoContext, asio::ip::tcp::endpoint(asio::ip::tcp::v6(), this->_port)); asio::ip::v6_only option(false); @@ -32,7 +32,7 @@ void SocketConnectionListener::initializeAcceptors() { try { this->_ipv4Acceptor.emplace( - this->_ioContext, + *this->_ptrIoContext, asio::ip::tcp::endpoint(asio::ip::tcp::v4(), this->_port)); } catch (const std::exception &ex) { std::cerr @@ -99,8 +99,10 @@ void SocketConnectionListener::acceptIPv4Connections() { std::cout << "[SocketConnectionListener] New incoming IPv4 connection." << std::endl; - auto tcpConnection = std::make_shared(std::move(socket)); - this->dispatchObservation(tcpConnection); + TcpConnection *ptrTcpConnection = new TcpConnection(socket); + if (!this->dispatchObservation(ptrTcpConnection)) { + delete ptrTcpConnection; + } } else if (!isListenerShutdownError(ec)) { std::cerr << "[SocketConnectionListener] IPv4 accept failed: " << ec.message() << std::endl; @@ -123,8 +125,10 @@ void SocketConnectionListener::acceptIPv6Connections() { std::cout << "[SocketConnectionListener] New incoming IPv6 connection." << std::endl; - auto tcpConnection = std::make_shared(std::move(socket)); - this->dispatchObservation(tcpConnection); + TcpConnection *ptrTcpConnection = new TcpConnection(socket); + if (!this->dispatchObservation(ptrTcpConnection)) { + delete ptrTcpConnection; + } } else if (!isListenerShutdownError(ec)) { std::cerr << "[SocketConnectionListener] IPv6 accept failed: " << ec.message() << std::endl; diff --git a/src/network/TcpConnection.cpp b/src/network/TcpConnection.cpp index 2665037..cb9f113 100644 --- a/src/network/TcpConnection.cpp +++ b/src/network/TcpConnection.cpp @@ -4,7 +4,22 @@ namespace xtrpg::network { +void TcpConnection::dispatchCloseCallbacks() { + auto callbacks = std::move(this->_closeCallbacks); + for (auto &callback : callbacks) { + if (callback) { + callback(); + } + } +} + void TcpConnection::upgrade(asio::ssl::context &ssl_ctx) { + if (!this->isOpen()) { + std::cout + << "[TcpConnection] Unable to upgrade as TCP Connection is not open." + << std::endl; + return; + } asio::post(*this->_strand, [this, &ssl_ctx]() { if (this->isClosed() || this->isClosing() || this->isSecure()) { @@ -16,26 +31,36 @@ void TcpConnection::upgrade(asio::ssl::context &ssl_ctx) { this->_sslStream->async_handshake( asio::ssl::stream_base::server, [this](std::error_code ec) { if (ec) { - this->_state = ConnectionState::CLOSED; - if (this->_sslStream) { - this->_sslStream->lowest_layer().close(); - } + this->close(); return; } - this->_state = ConnectionState::SECURE; + + this->dispatchStateChange(ConnectionState::SECURE); }); }); } -void TcpConnection::read(std::function callback) { +void TcpConnection::read( + std::function callback) { + if (!this->isOpen()) { + std::cout << "[TcpConnection] Unable to read as TCP Connection is not open." + << std::endl; + // Preserve read's callback contract even when the socket closed before + // the asynchronous operation could be posted. + std::istringstream stream; + callback(asio::error::operation_aborted, stream); + return; + } std::cout << "[TcpConnection] Requesting to read." << std::endl; auto buffer = std::make_shared>(4096); asio::post(*this->_strand, [this, buffer, callback]() { - std::cout << "[TcpConnection] asio::post." << std::endl; if (!this->isOpen()) { - std::cout << "[TcpConnection] Stream not open." << std::endl; - throw exception::ConnectionClosed(); + // The connection may close after the caller's initial state check but + // before this strand handler begins executing. + std::istringstream stream; + callback(asio::error::operation_aborted, stream); + return; } if (this->isSecure() && this->_sslStream) { @@ -43,16 +68,17 @@ void TcpConnection::read(std::function callback) { asio::buffer(*buffer), [this, buffer, callback](std::error_code ec, std::size_t bytes_transferred) { + std::istringstream stream; if (ec) { - this->_state = ConnectionState::CLOSED; - if (this->_sslStream) { - this->_sslStream->lowest_layer().close(); - } + this->close([callback, ec]() { + std::istringstream errorStream; + callback(ec, errorStream); + }); return; } std::string data(buffer->data(), bytes_transferred); - std::istringstream stream(data); - callback(stream); + stream.str(data); + callback(ec, stream); }); return; @@ -62,34 +88,52 @@ void TcpConnection::read(std::function callback) { asio::buffer(*buffer), [this, buffer, callback](std::error_code ec, std::size_t bytes_transferred) { + std::istringstream stream; if (ec) { - this->_state = ConnectionState::CLOSED; - this->_tcpSocket.close(); + this->close([callback, ec]() { + std::istringstream errorStream; + callback(ec, errorStream); + }); return; } std::string data(buffer->data(), bytes_transferred); - std::istringstream stream(data); - callback(stream); + stream.str(data); + callback(ec, stream); }); }); } +void TcpConnection::cancelRead() { + asio::post(*this->_strand, [this]() { + if (this->isOpen()) { + if (this->isSecure() && this->_sslStream) { + this->_sslStream->lowest_layer().cancel(); + } else { + this->_tcpSocket.cancel(); + } + } + }); +} + void TcpConnection::write(std::string_view data) { + if (!this->isOpen()) { + std::cout + << "[TcpConnection] Unable to write as TCP Connection is not open." + << std::endl; + return; + } auto payload = std::make_shared(data); asio::post(*this->_strand, [this, payload]() { if (!this->isOpen()) { - throw exception::ConnectionClosed(); + return; } if (this->isSecure() && this->_sslStream) { asio::async_write(*this->_sslStream, asio::buffer(*payload), [this, payload](std::error_code ec, std::size_t) { if (ec) { - this->_state = ConnectionState::CLOSED; - if (this->_sslStream) { - this->_sslStream->lowest_layer().close(); - } + this->close(); } }); return; @@ -98,22 +142,36 @@ void TcpConnection::write(std::string_view data) { asio::async_write(this->_tcpSocket, asio::buffer(*payload), [this, payload](std::error_code ec, std::size_t) { if (ec) { - this->_state = ConnectionState::CLOSED; - this->_tcpSocket.close(); + this->close(); } }); }); } -void TcpConnection::close() { +void TcpConnection::close(std::function callback) { - asio::post(*this->_strand, [this]() { - if (this->isClosed() || this->isClosing()) { - return; + if (this->is(ConnectionState::CLOSED) || this->is(ConnectionState::CLOSING)) { + std::cout << "[TcpConnection] Connection is already closed or in the " + "process of being closed." + << std::endl; + if (this->isClosed() && callback) { + callback(); + } else if (this->isClosing() && callback) { + this->_closeCallbacks.push_back(std::move(callback)); } + return; + } - this->_state = ConnectionState::CLOSING; + if (callback) { + this->_closeCallbacks.push_back(std::move(callback)); + } + // Set the state to closing. + std::cout << "[TcpConnection] Request Close." << std::endl; + this->dispatchStateChange(ConnectionState::CLOSING); + + // Serialize transport shutdown with reads and writes on the strand. + asio::post(*this->_strand, [this]() { if (this->isSecure() && this->_sslStream) { this->_sslStream->lowest_layer().cancel(); @@ -124,7 +182,8 @@ void TcpConnection::close() { this->_sslStream->lowest_layer().close(); } - this->_state = ConnectionState::CLOSED; + this->dispatchStateChange(ConnectionState::CLOSED); + this->dispatchCloseCallbacks(); }); return; } @@ -132,7 +191,8 @@ void TcpConnection::close() { std::error_code ec; this->_tcpSocket.shutdown(asio::ip::tcp::socket::shutdown_both, ec); this->_tcpSocket.close(); - this->_state = ConnectionState::CLOSED; + this->dispatchStateChange(ConnectionState::CLOSED); + this->dispatchCloseCallbacks(); }); } diff --git a/src/xmpp/ClientConnectionManager.cpp b/src/xmpp/ClientConnectionManager.cpp index 33afe8f..0273b07 100644 --- a/src/xmpp/ClientConnectionManager.cpp +++ b/src/xmpp/ClientConnectionManager.cpp @@ -2,31 +2,133 @@ namespace xtrpg::xmpp { +ClientConnectionManager::ClientConnectionManager(asio::io_context &ioContext, + uint16_t port) + : _ioContext(&ioContext) { + + // Create the listener that accepts incoming client TCP connections. + this->_ptrSocketConnectionListener = + new network::SocketConnectionListener(this->_ioContext, port); + + // Route newly accepted connections to this manager. + this->_ptrSocketConnectionListener->setObserver(this); + + // Begin accepting client connections on the configured port. + this->_ptrSocketConnectionListener->start(); +} + ClientConnectionManager::~ClientConnectionManager() { - // Loop over the `this->_clientSessionPtrs` vector, shut them down and delete - // the instances - for (auto ptrSession : this->_clientSessionPtrs) { + + // Stop accepting new connections and detach the observer before destruction. + if (nullptr != this->_ptrSocketConnectionListener) { + // Prevent new accept completions from entering the manager while it is + // dismantling the sessions it already owns. + this->_ptrSocketConnectionListener->stop(); + this->_ptrSocketConnectionListener->setObserver(nullptr); + } + + // Move the sessions out while holding the mutex so no concurrent operation + // can access the manager's vector during shutdown. + { + std::lock_guard lock(this->_callbackMutex); + this->_isShuttingDown = true; + } + std::vector sessions; + { + std::unique_lock lock(this->_clientSessionsVectorMutex); + sessions.swap(this->_clientSessionPtrs); + } + + // Shut down and delete sessions after releasing the mutex because shutdown + // may trigger callbacks that access the manager. + for (auto *ptrSession : sessions) { if (ptrSession) { + // Completion callbacks must not mutate the registry while the manager + // is being destroyed; the destructor owns cleanup from this point on. + ptrSession->setCompletionCallback(nullptr); ptrSession->shutdown(); - delete ptrSession; } } - this->_clientSessionPtrs.clear(); + if (this->_ioContext != nullptr) { + // Run cancellation and close handlers while the manager and sessions are + // still alive. This drains callbacks that capture their raw addresses. + this->_ioContext->restart(); + this->_ioContext->run(); + } + for (auto *ptrSession : sessions) { + // All handlers have completed, so deleting the sessions is now safe. + delete ptrSession; + } + + delete this->_ptrSocketConnectionListener; + this->_ptrSocketConnectionListener = nullptr; } /** * A new Tcp Connection is created. */ -void ClientConnectionManager::onObservation( - std::shared_ptr &ctx) { +void ClientConnectionManager::onObservation(network::TcpConnection *ctx) { + std::lock_guard callbackLock(this->_callbackMutex); + if (this->_isShuttingDown) { + delete ctx; + return; + } + std::cout << "[ClientConnectionManager] New client connection received" << std::endl; - // Create a ClientSession for this connection - auto clientSession = new session::ClientSession(std::move(*ctx)); - this->_clientSessionPtrs.push_back(clientSession); + // Wrap the new TCP connection in a session owned by this manager. + auto clientSession = new session::ClientSession(ctx); + { + // Protect the session registry from concurrent connection callbacks. + std::unique_lock lock(this->_clientSessionsVectorMutex); + this->_clientSessionPtrs.push_back(clientSession); + } + + // Stop the session when its underlying connection closes. Destruction is + // deferred until the pending read handler has completed. + ctx->appendStateChangeCallback( + [this, clientSession](network::ConnectionState state) { + std::lock_guard lock(this->_callbackMutex); + if (this->_isShuttingDown) { + return; + } + std::cout << "[ClientConnectionManager] State Change: " + << (network::ConnectionState::CLOSED == state ? "CLOSED" + : state == network::ConnectionState::CLOSING ? "CLOSING" + : state == network::ConnectionState::SECURE ? "SECURE" + : "INSECURE") + << std::endl; + if (network::ConnectionState::CLOSED == state) { + // Stop scheduling reads; completion still comes from the pending + // read or close callback and removes the session from the registry. + clientSession->stop(); + } + }); + + clientSession->setCompletionCallback([this](session::ClientSession *session) { + std::lock_guard callbackLock(this->_callbackMutex); + // Shutdown suppresses registry mutation because the destructor already + // removed every session from the registry before draining the context. + if (this->_isShuttingDown) { + return; + } + + std::unique_lock lock(this->_clientSessionsVectorMutex); + const auto sessionIt = std::find(this->_clientSessionPtrs.begin(), + this->_clientSessionPtrs.end(), session); + if (sessionIt == this->_clientSessionPtrs.end()) { + return; + } + + this->_clientSessionPtrs.erase(sessionIt); + lock.unlock(); + // Let the transport callback return before destroying the session and its + // owned connection; TcpConnection may still be dispatching close callbacks. + asio::post(*this->_ioContext, [session]() { delete session; }); + }); - // Start the session + // Start reading and processing data for the new client. clientSession->start(); std::cout << "[ClientConnectionManager] ClientSession created and started" diff --git a/src/xmpp/session/ClientSession.cpp b/src/xmpp/session/ClientSession.cpp index 2b28b89..af953fb 100644 --- a/src/xmpp/session/ClientSession.cpp +++ b/src/xmpp/session/ClientSession.cpp @@ -21,13 +21,14 @@ ClientSession::~ClientSession() { // remove myself from the tokenizer this->_tokenizer.setObserver(nullptr); + + // destroy the TCP connection + delete this->_ptrTcpConnection; + this->_ptrTcpConnection = nullptr; } void ClientSession::start() { - std::cout << "[ClientSession] Requesting to start." << std::endl; if (this->_isShutdown) { - std::cout << "[ClientSession] Failed to start, already shutdown." - << std::endl; return; } @@ -35,7 +36,11 @@ void ClientSession::start() { this->process(); } -void ClientSession::stop() { this->_isStopped.exchange(true); } +void ClientSession::stop() { + if (!this->_isStopped.exchange(true)) { + this->_ptrTcpConnection->cancelRead(); + } +} void ClientSession::shutdown() { this->stop(); @@ -43,8 +48,9 @@ void ClientSession::shutdown() { return; } - // shutdown the TCP connection - this->_tcpConnection.close(); + // Notify the manager after the connection's close work has completed. This + // also covers shutdown initiated from a successful read callback. + this->_ptrTcpConnection->close([this]() { this->notifyCompletion(); }); } void ClientSession::sendRaw(std::string_view data) { @@ -52,35 +58,47 @@ void ClientSession::sendRaw(std::string_view data) { return; } - this->_tcpConnection << data; + *this->_ptrTcpConnection << data; } void ClientSession::process() { - - std::cout << "[ClientSession] Requesting to process." << std::endl; - if (this->_isStopped) { - std::cout << "[ClientSession] Session is stopped." << std::endl; + if (this->_isStopped || this->_isShutdown) { return; } - // calls the _tcpConnect to request the next chunk of data - // the lambda function - this->_tcpConnection.read([this](std::istream &is) { - std::cout << "[Client Session] Passing input stream to the tokenizer." - << std::endl; - this->_tokenizer.process(is); - this->process(); - }); + // Each read either schedules the next read or reports completion, so a + // closed connection cannot leave the session registered indefinitely. + this->_ptrTcpConnection->read( + [this](const std::error_code &error, std::istream &stream) { + if (error || this->_isStopped || this->_isShutdown) { + // Errors, cancellation, and explicit shutdown all end the session. + this->notifyCompletion(); + return; + } + this->_tokenizer.process(stream); + this->process(); + }); } -void ClientSession::onXmlToken(const xml::tokenizer::XmlToken &xmlToken) { - std::cout << "[ClientSession] Observed XML Token: " << xmlToken.content - << std::endl; +void ClientSession::notifyCompletion() { + // Multiple terminal events can race; only the first one may notify the + // manager and trigger deletion. + if (this->_completionNotified.exchange(true)) { + return; + } + + std::function completionCallback; + { + std::lock_guard lock(this->_completionCallbackMutex); + completionCallback = this->_completionCallback; + } - for (const auto &[key, value] : xmlToken.attributes) { - std::cout << " - " << key << ": " << value << std::endl; + if (completionCallback) { + completionCallback(this); } +} +void ClientSession::onXmlToken(const xml::tokenizer::XmlToken &xmlToken) { if (xml::tokenizer::TokenType::OPEN_TAG == xmlToken.type && "stream:stream" == xmlToken.content) { this->sendRaw( @@ -90,6 +108,8 @@ void ClientSession::onXmlToken(const xml::tokenizer::XmlToken &xmlToken) { "xmlns='urn:ietf:params:xml:ns:xmpp-streams'/>Stanza size " "limit of 64KB exceeded."); + + this->shutdown(); } } void ClientSession::onTokenizationError( From c2fb160d9c03d657c01d4e90163a6888c123510b Mon Sep 17 00:00:00 2001 From: Xeno Snow Fox Date: Fri, 4 Sep 2026 19:23:50 +1000 Subject: [PATCH 9/9] Format state-change callback Tidies the callback lambda in ClientConnectionManager::onObservation for readability and consistent formatting. The connection-state shutdown logic remains unchanged: sessions are still stopped when the TCP connection closes and shutdown is not in progress. --- src/xmpp/ClientConnectionManager.cpp | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/src/xmpp/ClientConnectionManager.cpp b/src/xmpp/ClientConnectionManager.cpp index edddc98..8972623 100644 --- a/src/xmpp/ClientConnectionManager.cpp +++ b/src/xmpp/ClientConnectionManager.cpp @@ -61,14 +61,13 @@ void ClientConnectionManager::onObservation(network::TcpConnection *ctx) { this->_clientSessionPtrs.push_back(clientSession); } - ctx->appendStateChangeCallback( - [this, clientSession](network::ConnectionState state) { - std::lock_guard lock(this->_callbackMutex); - if (!this->_isShuttingDown && - state == network::ConnectionState::CLOSED) { - clientSession->stop(); - } - }); + ctx->appendStateChangeCallback([this, clientSession]( + network::ConnectionState state) { + std::lock_guard lock(this->_callbackMutex); + if (!this->_isShuttingDown && state == network::ConnectionState::CLOSED) { + clientSession->stop(); + } + }); clientSession->setCompletionCallback([this](session::ClientSession *session) { std::lock_guard callbackLock(this->_callbackMutex);