feat: Implement XMPP Client Connection Manager and Refactor Build System

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<int64_t>("c2s", "port").value_or(5222);
     uint16_t listeningPort = static_cast<uint16_t>(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();
 

include/xtrpg/interface/Observable.hpp

@@ -2,6 +2,7 @@
 
 #include <iostream>
 #include <memory>
+#include <mutex>
 #include <string_view>
 #include <vector>
 
@@ -18,18 +19,24 @@ template <typename TContext> class Observable {
     }
   };
 
-  void setObserver(Observer<TContext> *ptr) { this->_ptrObserver = ptr; }
+  void setObserver(Observer<TContext> *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<TContext> *_ptrObserver;
+  Observer<TContext> *_ptrObserver = nullptr;
+  std::mutex _observerMutex;
 };
 } // namespace xtrpg::interface

include/xtrpg/interface/Observer.hpp

@@ -8,6 +8,6 @@ template <typename TContext> class Observer {
   virtual ~Observer() = default;
 
   // Generic handler callback
-  virtual void onObservation(TContext &ctx) = 0;
+  virtual void onObservation(TContext ctx) = 0;
 };
 } // namespace xtrpg::interface

include/xtrpg/network/SocketConnectionListener.hpp

@@ -11,14 +11,13 @@
 #include "xtrpg/network/TcpConnection.hpp"
 
 namespace xtrpg::network {
-class SocketConnectionListener
-    : public interface::Observable<std::shared_ptr<TcpConnection>> {
+class SocketConnectionListener : public interface::Observable<TcpConnection *> {
 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<asio::ip::tcp::acceptor> _ipv4Acceptor;
   std::optional<asio::ip::tcp::acceptor> _ipv6Acceptor;

include/xtrpg/network/TcpConnection.hpp

@@ -2,9 +2,12 @@
 
 #include <asio.hpp>
 #include <asio/ssl.hpp>
+#include <atomic>
+#include <functional>
 #include <iostream>
 #include <memory>
 #include <optional>
+#include <vector>
 
 #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,19 +60,31 @@ 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<void(std::istream &)> callback);
+  void
+  read(std::function<void(const std::error_code &, std::istream &)> callback);
+
+  /** Cancels the currently pending read operation, if any. */
+  void cancelRead();
 
   /**
    * Writes data to the connection. If the connection is closed, it will throw
    * an exception.
    */
   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<void()> callback = {});
 
   /**
    * Checks if the connection is secure (SSL/TLS).
@@ -112,11 +127,29 @@ class TcpConnection {
     return *this;
   }
 
+  void
+  appendStateChangeCallback(std::function<void(ConnectionState)> callback) {
+    std::cout << "[TcpConnection] Append State Change Callback." << std::endl;
+    this->_stateChangeCallbacks.push_back(callback);
+  }
+
 private:
+  std::vector<std::function<void(ConnectionState)>> _stateChangeCallbacks;
+  std::vector<std::function<void()>> _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<ConnectionState> _state{ConnectionState::INSECURE};
 
   /**
    * The underlying TCP socket used for the connection.

include/xtrpg/xmpp/ClientConnectionManager.hpp

@@ -1,37 +1,56 @@
 #pragma once
 
+#include <algorithm>
 #include <asio.hpp>
-#include <memory>
+#include <cstdint>
+#include <mutex>
+#include <shared_mutex>
 #include <vector>
 
 #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<std::shared_ptr<network::TcpConnection>> {
+      public interface::Observer<network::TcpConnection *> {
 public:
+  /** Creates an inactive manager without a connection listener. */
   ClientConnectionManager() = default;
 
+  /** Stops accepting connections and releases all owned client sessions. */
   ~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<network::TcpConnection> &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 client-to-server listener configuration schema. */
   config::ModuleConfig getConfigSchema() const {
     return {.name = "c2s",
             .description = "",
@@ -42,10 +61,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<int>(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<session::ClientSession *> _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);

include/xtrpg/xmpp/Jid.hpp

@@ -1,6 +1,5 @@
 #pragma once
 
-#include <functional>
 #include <iostream>
 #include <optional>
 #include <string>