diff --git a/CMakeLists.txt b/CMakeLists.txt
index 70c4ecc..b2f5093 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,27 @@ 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
+    src/xml/tokenizer/XmlStreamTokenizer.cpp
+    src/xmpp/ClientConnectionManager.cpp
+    src/xmpp/session/ClientSession.cpp
+    src/xmpp/Jid.cpp
 )
-target_include_directories(xtrpg_network PUBLIC include)
-target_link_libraries(xtrpg_network PUBLIC
+
+
+target_link_libraries(xtrpg_cpp_server PRIVATE
     asio::asio
     OpenSSL::SSL
     OpenSSL::Crypto
@@ -94,27 +86,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})
diff --git a/apps/main.cpp b/apps/main.cpp
index c693556..1725096 100644
--- a/apps/main.cpp
+++ b/apps/main.cpp
@@ -1,14 +1,36 @@
+#include <atomic>
 #include <fstream>
 #include <iostream>
+#include <memory>
+#include <signal.h>
+#include <thread>
+#include <vector>
+
+#include <asio.hpp>
 
 #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 <windows.h>
 #endif
 
+// Global flag for signal handling
+std::atomic<bool> 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
+    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();
+
+    // 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<std::thread> 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();
+    listener.setObserver(nullptr);
+
+    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;
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 <typename TContext> class Observable {
 public:
-  virtual ~Observable() = default;
-
-  void addObserver(const std::shared_ptr<Observer<TContext>> &ptr) {
-    if (ptr) {
-      this->_observers.push_back(ptr);
-    }
-  }
-
-  void eraseObserver(const Observer<TContext> *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<Observer<TContext>> &wp) {
-                    auto sp = wp.lock();
-                    return !sp || sp.get() == pTarget;
-                  });
-  }
-
-  void clearObservers() { this->_observers.clear(); }
+  void setObserver(Observer<TContext> *ptr) { this->_ptrObserver = ptr; }
 
 protected:
   void dispatchObservation(TContext &ctx) {
-    std::erase_if(this->_observers,
-                  [&ctx](const std::weak_ptr<Observer<TContext>> &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<std::weak_ptr<Observer<TContext>>> _observers;
+  Observer<TContext> *_ptrObserver;
 };
-} // namespace xtrpg
\ No newline at end of file
+} // namespace xtrpg::interface
\ No newline at end of file
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<TcpConnection> {
+class TcpConnection {
 
 public:
   /**
@@ -53,6 +53,12 @@ class TcpConnection : public std::enable_shared_from_this<TcpConnection> {
    */
   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<void(std::istream &)> 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 <vector>
 
 #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..1bf9f03 100644
--- a/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp
+++ b/include/xtrpg/xml/tokenizer/XmlStreamTokenizer.hpp
@@ -7,6 +7,7 @@
 #include <vector>
 
 #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
@@ -20,19 +21,25 @@ namespace xtrpg::xml::tokenizer {
  */
 class XmlStreamTokenizer {
 public:
+  ~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);
 
-  /**
-   * Defines a instance to act as the listener for this class.
-   */
-  void setListener(std::shared_ptr<XmlTokenListener> listener) {
-    _listener = listener;
-  }
+  void setObserver(XmlTokenListener *ptr) { this->_ptrObserver = ptr; }
 
 private:
+  XmlTokenListener *_ptrObserver = nullptr;
+
   enum class State {
     TEXT,
     AFTER_OPEN,
@@ -69,7 +76,10 @@ class XmlStreamTokenizer {
 
   TokenizationError _error{TokenizationError::NONE};
 
-  std::weak_ptr<XmlTokenListener> _listener;
+  /**
+   * 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 <string>
+#include <unordered_map>
+
+namespace xtrpg::xml::tokenizer {
+
+enum class TokenType {
+  /**
+   * Represents an OPEN XML Tag with attributes (eg: "<myTag attr='value'>").
+   * The content field will equal the name of the tag (eg: "myTag").
+   */
+  OPEN_TAG,
+
+  /**
+   * Represents a CLOSE XML Tag (eg "</myTag>").
+   * 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: "<myTag attr='value'
+   * />"). 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<std::string, std::string> attributes;
+};
+} // namespace xtrpg::xml::tokenizer
diff --git a/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp b/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp
index 1bd3938..ac7b901 100644
--- a/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp
+++ b/include/xtrpg/xml/tokenizer/XmlTokenListener.hpp
@@ -13,47 +13,8 @@ class XmlTokenListener {
 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;
+  virtual void onXmlToken(const xml::tokenizer::XmlToken &xmlToken) = 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;
-
-  /**
-   * 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;
-
-  /**
-   * 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/ClientConnectionManager.hpp b/include/xtrpg/xmpp/ClientConnectionManager.hpp
new file mode 100644
index 0000000..779e740
--- /dev/null
+++ b/include/xtrpg/xmpp/ClientConnectionManager.hpp
@@ -0,0 +1,53 @@
+#pragma once
+
+#include <asio.hpp>
+#include <memory>
+#include <vector>
+
+#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,
+      public interface::Observer<std::shared_ptr<network::TcpConnection>> {
+public:
+  ClientConnectionManager() = default;
+
+  ~ClientConnectionManager();
+
+  explicit ClientConnectionManager(asio::io_context &ioContext)
+      : _ioContext(&ioContext) {}
+
+  /**
+   * A new Tcp Connection is created.
+   */
+  void onObservation(std::shared_ptr<network::TcpConnection> &ctx);
+
+  /**
+   *
+   */
+  config::ModuleConfig getConfigSchema() const {
+    return {.name = "c2s",
+            .description = "",
+            .options = {{.key = "port",
+                         .defaultValue = 5222,
+                         .description =
+                             "<value> Port number that this server will listen "
+                             "on for Client (or C2S) connections."}}};
+  }
+
+private:
+  asio::io_context *_ioContext = nullptr;
+
+  std::vector<session::ClientSession *> _clientSessionPtrs;
+};
+
+REGISTER_MODULE_CONFIG(ClientConnectionManager);
+
+} // namespace xtrpg::xmpp
diff --git a/include/xtrpg/xmpp/session/ClientSession.hpp b/include/xtrpg/xmpp/session/ClientSession.hpp
new file mode 100644
index 0000000..7bb15c9
--- /dev/null
+++ b/include/xtrpg/xmpp/session/ClientSession.hpp
@@ -0,0 +1,59 @@
+#pragma once
+
+#include <atomic>
+#include <memory>
+
+#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/XmlToken.hpp"
+#include "xtrpg/xml/tokenizer/XmlTokenListener.hpp"
+
+namespace xtrpg::xmpp::session {
+
+class ClientSession : public xml::tokenizer::XmlTokenListener,
+                      public std::enable_shared_from_this<ClientSession> {
+
+public:
+  ClientSession(network::TcpConnection tcpConnection)
+      : _tcpConnection(std::move(tcpConnection)) {
+    std::cout << "[ClientSession] New Instance created." << std::endl;
+    this->_tokenizer.setObserver(this);
+  }
+  ~ClientSession();
+
+  // Session Control
+  void start();
+  void stop();
+  void shutdown();
+  void process();
+
+  // Transport Control
+  void sendRaw(std::string_view data);
+
+  // Tokenizer Calls
+  void onXmlToken(const xml::tokenizer::XmlToken &xmlToken);
+  void onTokenizationError(const 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<bool> _isStopped{true};
+
+  /**
+   * Boolean flag that indicates whether the underlying connection has been
+   * terminated.
+   */
+  std::atomic<bool> _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 <sstream>
+
 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<void(std::istream &)> callback) {
+  std::cout << "[TcpConnection] Requesting to read." << std::endl;
+  auto buffer = std::make_shared<std::vector<char>>(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<std::string>(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..1e84266 100644
--- a/src/xml/tokenizer/XmlStreamTokenizer.cpp
+++ b/src/xml/tokenizer/XmlStreamTokenizer.cpp
@@ -18,50 +18,34 @@ 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 (const auto listener = this->_listener.lock()) {
-      listener->onError(this->_error);
-    }
     return;
   }
 
-  const auto listener = this->_listener.lock();
-  const auto fail = [&](const TokenizationError error) {
+  const auto fail = [&](const TokenizationError &error) {
     this->_error = error;
-    if (listener) {
-      listener->onError(error);
-    }
-  };
-  const auto appendText = [&](const std::string_view text) {
-    if (!text.empty() && listener) {
-      listener->appendText(text);
-    }
+    this->_ptrObserver->onTokenizationError(error);
   };
-  const auto openStartTag = [&]() {
-    if (this->_buffer.empty()) {
-      fail(TokenizationError::MALFORMED_INPUT);
-      return;
-    }
-    if (listener) {
-      listener->openTag(this->_buffer);
-    }
-    this->_buffer.clear();
+
+  const auto emitToken = [&](const XmlToken &token) {
+    std::cout << "[XmlStreamTokenizer] Dispatching XML Token: " << token.content
+              << std::endl;
+    this->_ptrObserver->onXmlToken(token);
   };
-  const auto openDeclaration = [&]() {
-    if (this->_buffer.empty()) {
-      fail(TokenizationError::MALFORMED_INPUT);
-      return;
-    }
-    if (listener) {
-      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);
@@ -76,29 +60,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);
@@ -111,14 +99,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);
         }
@@ -128,6 +132,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;
@@ -180,9 +185,7 @@ void XmlStreamTokenizer::process(std::istream &stream) {
         break;
       case State::ATTRIBUTE_VALUE:
         if (character == this->_quote) {
-          if (listener) {
-            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;
@@ -200,15 +203,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 (listener) {
-              listener->closeTag();
-            }
+            this->_currentToken.content = this->_buffer;
             this->_buffer.clear();
+            emitToken(this->_currentToken);
             this->_state = State::TEXT;
           }
         } else {
@@ -219,11 +223,8 @@ void XmlStreamTokenizer::process(std::istream &stream) {
         if (isWhitespace(character)) {
           break;
         }
-        if (character == '>' && !this->_buffer.empty()) {
-          if (listener) {
-            listener->closeTag();
-          }
-          this->_buffer.clear();
+        if (character == '>') {
+          emitToken(this->_currentToken);
           this->_state = State::TEXT;
         } else {
           fail(TokenizationError::MALFORMED_INPUT);
@@ -236,11 +237,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);
         }
@@ -300,9 +311,7 @@ void XmlStreamTokenizer::process(std::istream &stream) {
         break;
       case State::DECLARATION_ATTRIBUTE_VALUE:
         if (character == this->_quote) {
-          if (listener) {
-            listener->setAttribute(this->_attributeName, this->_buffer);
-          }
+          this->_currentToken.attributes[this->_attributeName] = this->_buffer;
           this->_attributeName.clear();
           this->_buffer.clear();
           this->_state = State::DECLARATION_BODY;
@@ -315,9 +324,7 @@ void XmlStreamTokenizer::process(std::istream &stream) {
         break;
       case State::DECLARATION_QUESTION:
         if (character == '>') {
-          if (listener) {
-            listener->closeDeclaration();
-          }
+          emitToken(this->_currentToken);
           this->_state = State::TEXT;
         } else {
           fail(TokenizationError::MALFORMED_INPUT);
@@ -325,9 +332,8 @@ void XmlStreamTokenizer::process(std::istream &stream) {
         break;
       case State::SELF_CLOSING:
         if (character == '>') {
-          if (listener) {
-            listener->closeTag();
-          }
+          this->_currentToken.type = TokenType::EMPTY_TAG;
+          emitToken(this->_currentToken);
           this->_state = State::TEXT;
         } else {
           fail(TokenizationError::MALFORMED_INPUT);
@@ -352,6 +358,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() >
@@ -363,13 +373,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/ClientConnectionManager.cpp b/src/xmpp/ClientConnectionManager.cpp
new file mode 100644
index 0000000..33afe8f
--- /dev/null
+++ b/src/xmpp/ClientConnectionManager.cpp
@@ -0,0 +1,36 @@
+#include "xtrpg/xmpp/ClientConnectionManager.hpp"
+
+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<network::TcpConnection> &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..2b28b89
--- /dev/null
+++ b/src/xmpp/session/ClientSession.cpp
@@ -0,0 +1,98 @@
+#include "xtrpg/xmpp/session/ClientSession.hpp"
+
+#include <iostream>
+
+#include "xtrpg/xml/tokenizer/XmlToken.hpp"
+
+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.setObserver(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::onXmlToken(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(
+        "<stream:stream xmlns='jabber:client' "
+        "xmlns:stream='http://etherx.jabber.org/streams' id='err-1' "
+        "from='example.com' version='1.0'><stream:error><policy-violation "
+        "xmlns='urn:ietf:params:xml:ns:xmpp-streams'/><text "
+        "xmlns='urn:ietf:params:xml:ns:xmpp-streams' xml:lang='en'>Stanza size "
+        "limit of 64KB exceeded.</text></stream:error></stream:stream>");
+  }
+}
+void ClientSession::onTokenizationError(
+    const xml::tokenizer::TokenizationError &error) {}
+
+} // namespace xtrpg::xmpp::session
\ No newline at end of file