Given the next C++ code:```#include <iostream>#include <string>#include <thread>#include <vector>#include <boost/log/trivial.hpp>#include <boost/program_options.hpp>#include "broker.h"std::string PORT = "8080";int NUM_THREADS = 1;int main(int argc, char* argv[]){    // Get the number of available cores    int num_cores = std::thread::hardware_concurrency();    // Parse command line arguments    boost::program_options::options_description desc("Allowed options");    desc.add_options()        ("help,h", "produce help message")        ("port,p", boost::program_options::value<std::string>()->default_value(PORT), "broker port")        ("threads,t", boost::program_options::value<int>()->default_value(num_cores), "number of threads");    boost::program_options::variables_map vm;    boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);    boost::program_options::notify(vm);    if (vm.count("help"))    {        std::cout << desc << "\n";        return 0;    }    if (vm.count("port"))    {        PORT = vm["port"].as<std::string>();        BOOST_LOG_TRIVIAL(info) << "Broker port: " << PORT;    }    if (vm.count("threads"))    {        NUM_THREADS = vm["threads"].as<int>();        BOOST_LOG_TRIVIAL(info) << "Number of threads: " << NUM_THREADS;    }    // The io_context is required for all I/O    boost::asio::io_context io_context;    // Create the broker    std::shared_ptr<broker> broker_instance = broker::create(io_context, PORT);    // Run the broker    broker_instance->run();    // Create a pool of threads to run all of the io_contexts    std::vector<std::thread> threads;    for(int i = 0; i < NUM_THREADS; ++i) {        threads.emplace_back([&io_context](){            io_context.run();        });    }    // Wait for all threads in the pool to exit    for(std::thread& t : threads) {        t.join();    }    return 0;}``````#ifndef BROKER_SESSION_H#define BROKER_SESSION_H#ifdef MULTITHREADED    #include <mutex>#endif#include <websocket/websocket_session.h>#include "broker.h"#include "states/state_base.h"class broker_session : public websocket_session{    friend class state_base<broker_session>;    friend class broker_factory;public:    broker_session(boost::asio::io_context &io_context, boost::asio::ip::tcp::socket socket);    void set_broker(std::weak_ptr<broker> broker);        const std::string& id() const;    protected:    void on_connect() override;    void on_message(const std::string &msg);    void on_failure();    void on_close() override;    void set_state(std::unique_ptr<state_base<broker_session>> state);    std::unique_ptr<state_base<broker_session>> state_;    std::string id_;private:    std::weak_ptr<broker> broker_;    #ifdef MULTITHREADED        std::mutex state_mutex_;    #endif};#endif``````#include <boost/uuid/uuid.hpp>#include <boost/uuid/uuid_generators.hpp>#include <boost/uuid/uuid_io.hpp>#include "states/initial_state.h"#include "broker_session.h"broker_session::broker_session(    boost::asio::io_context &io_context,    boost::asio::ip::tcp::socket socket) : websocket_session(io_context, std::move(socket)),    broker_(std::weak_ptr<broker>()){    // Set the initial state    state_ = std::make_unique<initial_state>();    // Generate a unique id for the broker session    boost::uuids::random_generator generator;    boost::uuids::uuid uuid = generator();    id_ = to_string(uuid);}void broker_session::set_broker(std::weak_ptr<broker> broker){    // We don't need to lock the mutex here because this method    // is called from the factory. It is guaranteed that the    // factory is only called from a single thread at a time.    broker_ = broker;}const std::string& broker_session::id() const{    return id_;}void broker_session::on_connect(){    if (state_)    {        state_->on_connect(*this);    }    else    {        std::cerr << "No state set for client connection." << std::endl;    }}void broker_session::on_message(const std::string &msg){    if (state_)    {        state_->on_message(*this, msg);    }    else    {        std::cerr << "No state set for client connection." << std::endl;    }}void broker_session::on_failure(){    if (state_)    {        state_->on_failure(*this);    }    else    {        std::cerr << "No state set for client connection." << std::endl;    }    // Close the connection    close();}void broker_session::on_close(){    std::shared_ptr<broker_session> self = std::static_pointer_cast<broker_session>(shared_from_this());    // Check if the broker is set and still being valid    if (!broker_.expired())    {        broker_.lock()->remove_relay(self);        broker_.lock()->remove_session(id_);    } else {        std::cerr << "Broker is not set for client connection." << std::endl;    }}void broker_session::set_state(std::unique_ptr<state_base<broker_session>> state){    #ifdef MULTITHREADED        std::lock_guard<std::mutex> lock(state_mutex_);    #endif    state_ = std::move(state);}``````#ifndef BROKER_FACTORY_H#define BROKER_FACTORY_H#include <websocket/websocket_session_factory.h>class broker;class broker_factory : public websocket_session_factory {public:    broker_factory(std::weak_ptr<broker> broker);    std::shared_ptr<websocket_session> create_session(boost::asio::io_context& io_context, boost::asio::ip::tcp::socket socket) override;private:    std::weak_ptr<broker> broker_instance_;};#endif``````#include "broker.h"#include "broker_session.h"#include "broker_factory.h"broker_factory::broker_factory(std::weak_ptr<broker> broker) :    broker_instance_(broker){}std::shared_ptr<websocket_session> broker_factory::create_session(    boost::asio::io_context& io_context,    boost::asio::ip::tcp::socket socket){    auto session = std::make_shared<broker_session>(io_context, std::move(socket));    session->set_broker(broker_instance_);    std::shared_ptr<broker> broker_instance = broker_instance_.lock();    if (!broker_instance) {        throw std::runtime_error("Broker instance no longer exists.");    }        broker_instance->add_session(session->id_, session);    return session;}``````#ifndef BROKER_H#define BROKER_H#ifdef MULTITHREADED    #include <shared_mutex>#endif#define BOOST_STACKTRACE_GNU_SOURCE_NOT_REQUIRED#include <boost/asio.hpp>#include <boost/beast.hpp>#include <boost/json.hpp>#include <websocket/websocket_server.h>#include "broker_factory.h"class broker_session;class broker : public std::enable_shared_from_this<broker> {public:    static std::shared_ptr<broker> create(boost::asio::io_context& io_context, const std::string& port);    ~broker();    void run();    void stop();    void add_session(const std::string& id, std::shared_ptr<broker_session> session);    void remove_session(const std::string& id);    std::shared_ptr<broker_session> get_session(const std::string& id);    void add_rpc_relay(const std::string& name, std::shared_ptr<broker_session> relay);    void remove_relay(std::shared_ptr<broker_session> relay);    std::shared_ptr<broker_session> get_rpc_relay(const std::string& name);protected:    boost::asio::io_context& io_context_;    std::shared_ptr<broker_factory> factory_;    std::unique_ptr<websocket_server> server_;    // List of total broker sessions connected    std::unordered_map<std::string, std::shared_ptr<broker_session>> sessions_;    // List of broker sessions that are relays for RPC calls    std::unordered_map<std::string, std::vector<std::shared_ptr<broker_session>>> rpc_relays_;    std::unordered_map<std::string, size_t> rpc_relay_indices_;#ifdef MULTITHREADED    std::shared_mutex sessions_mutex_;    std::shared_mutex rpc_relays_mutex_;#endif    // Private constructor to force the use of the create function    broker(boost::asio::io_context& io_context);    void initialize(const std::string& port);};#endif``````#include <memory>#include <boost/log/trivial.hpp>#include "broker.h"#include "broker_session.h"std::shared_ptr<broker> broker::create(boost::asio::io_context& io_context, const std::string& port) {    std::shared_ptr<broker> instance(new broker(io_context));    instance->initialize(port);    return instance;}void broker::initialize(const std::string &port) {    this->factory_ = std::make_shared<broker_factory>(std::weak_ptr<broker>(shared_from_this()));    this->server_ = std::make_unique<websocket_server>(io_context_, port, *factory_);}broker::broker(boost::asio::io_context& io_context) :    io_context_(io_context),    factory_(nullptr),    server_(nullptr){}broker::~broker(){    BOOST_LOG_TRIVIAL(trace) << "[Broker] Destroying broker";}void broker::run(){    server_->run();}void broker::stop(){    server_->stop();}void broker::add_session(const std::string& id, std::shared_ptr<broker_session> session){    #ifdef MULTITHREADED        std::unique_lock<std::shared_mutex> lock(sessions_mutex_);    #endif    BOOST_LOG_TRIVIAL(debug) << "[Broker] Adding session with id: " << id;    sessions_[id] = session;}void broker::remove_session(const std::string& id){    #ifdef MULTITHREADED        std::unique_lock<std::shared_mutex> lock(sessions_mutex_);    #endif    BOOST_LOG_TRIVIAL(debug) << "[Broker] Removing session with id: " << id;    sessions_.erase(id);}std::shared_ptr<broker_session> broker::get_session(const std::string& id){    #ifdef MULTITHREADED        std::shared_lock<std::shared_mutex> lock(sessions_mutex_);    #endif    auto it = sessions_.find(id);    if (it != sessions_.end())    {        return it->second;    }    return nullptr;}void broker::add_rpc_relay(const std::string& name, std::shared_ptr<broker_session> relay){    #ifdef MULTITHREADED        std::unique_lock<std::shared_mutex> lock(rpc_relays_mutex_);    #endif    if (rpc_relays_.find(name) == rpc_relays_.end())    {        rpc_relays_[name] = std::vector<std::shared_ptr<broker_session>>();    }    auto& relays = rpc_relays_[name];    relays.push_back(relay);    // Initialize the index for round-robin if this is the first relay for the RPC name    if (relays.size() == 1) {        rpc_relay_indices_[name] = 0;    }    BOOST_LOG_TRIVIAL(debug) << "[Broker] Added RPC relay for: " << name << " with session ID: " << relay->id();}void broker::remove_relay(std::shared_ptr<broker_session> relay) {    #ifdef MULTITHREADED        std::unique_lock<std::shared_mutex> lock(rpc_relays_mutex_);    #endif    for (auto& pair : rpc_relays_) {        auto& relays = pair.second;        relays.erase(std::remove(relays.begin(), relays.end(), relay), relays.end());    }    BOOST_LOG_TRIVIAL(debug) << "[Broker] Removed RPC relay with session ID: " << relay->id();}std::shared_ptr<broker_session> broker::get_rpc_relay(const std::string& name){    #ifdef MULTITHREADED        std::shared_lock<std::shared_mutex> lock(rpc_relays_mutex_);    #endif    auto it = rpc_relays_.find(name);    if (it != rpc_relays_.end() && !it->second.empty()) {        // Round-robin selection        size_t& index = rpc_relay_indices_[name];        auto relay = it->second[index];        index = (index + 1) % it->second.size();        BOOST_LOG_TRIVIAL(trace) << "[Broker] Selected RPC relay for: " << name << " with session ID: " << relay->id();        return relay;    }    return nullptr;}```I want to add too the capability of "build a cluster" of brokers (connecting a broker with other broker) and share its states, implementing too something like a distributed memory (so it's a new functionality of the broker - clients could save in the distributed memory key-values objects). So if one broker get down, the "shared" memory is keeped in the other brokers, something like a consensus.Make it DRY, performant, well-written, C++17, extensible, modular, etc.New peers could be added at runtime and a configuration file should be saved for rerunning. So when broker starts it would try to read this config file and check if the rest of "brokers" are alive.It should impelment mechanism of what happen when broker get down, if they're desynced, etc.Your code modifications/additions should be completly functional and covering all this.Show final code, only modifications in the code, not the full code.