fix_server.h

// DISCLAIMER_PLACEHOLDER
#pragma once
 
#include "common.h"
 
#include <cassert>
#include <cstdint>
#include <cstddef>
#include <string>
#include <string_view>
#include <cstdio>
#include <vector>
#include <unordered_map>
#include <functional>
#include <mutex> // for std::lock_guard
#include <new>
 
#include "core/compiler/hints_branch_predictor.h"
#include "core/compiler/hints_hot_code.h"
#include "core/compiler/unused.h"
#include "core/os/vdso.h"
#include "core/os/thread_local_storage.h"
 
#include "core/utilities/configuration.h"
#include "core/utilities/logger.h"
#include "core/utilities/object_cache.h"
#include "core/utilities/std_string_utilities.h"
#include "core/utilities/userspace_spinlock.h"
 
#include "electronic_trading/common/message_persist_plugin.h"
#include "electronic_trading/managed_instance/modifying_admin_command.h"
#include "electronic_trading/managed_instance/managed_instance.h"
 
#include "fix_constants.h"
#include "fix_string_view.h"
#include "fix_string.h"
#include "fix_session_settings.h"
#include "fix_utilities.h"
#include "incoming_fix_message.h"
#include "outgoing_fix_message.h"
#include "fix_parser_error_codes.h"
#include "fix_session.h"
#include "fix_server_settings.h"
 
#include "core/utilities/tcp_reactor_options.h"
 
namespace llfix
{
 
// Thread local
struct UnknownSessionContext
{
    IncomingFixMessage m_incoming_fix_message;
    ObjectCache<FixStringView> m_fix_string_view_cache;
};
 
class FixServerConnectors
{
    public:
        FixServerConnectors()
        {
            m_tables_lock.initialise();
            m_peer_index_session_table.reserve(1024);
            m_session_peer_index_table.reserve(1024);
        }
 
        bool has_peer(std::size_t peer_index) const
        {
            const std::lock_guard<UserspaceSpinlock<>> lock(m_tables_lock);
 
            if (m_peer_index_session_table.find(peer_index) == m_peer_index_session_table.end())
            {
                return false;
            }
 
            return true;
        }
 
        void update(std::size_t peer_index, FixSession* session)
        {
            const std::lock_guard<UserspaceSpinlock<>> lock(m_tables_lock);
 
            // In concurrent scenarios, disconnection for an existing item can happen before its connection as they will happen on different threads
            if (m_session_peer_index_table.find(session) != m_session_peer_index_table.end())
            {
                auto existing_peer_index = m_session_peer_index_table[session];
 
                if (existing_peer_index != peer_index)
                {
                    m_peer_index_session_table.erase(existing_peer_index);
                }
            }
 
            m_peer_index_session_table[peer_index] = session;
            m_session_peer_index_table[session] = peer_index;
        }
 
        FixSession* get_session(std::size_t peer_index)
        {
            const std::lock_guard<UserspaceSpinlock<>> lock(m_tables_lock);
 
            if (m_peer_index_session_table.find(peer_index) == m_peer_index_session_table.end())
            {
                return nullptr;
            }
 
            return m_peer_index_session_table[peer_index];
        }
 
        std::size_t get_peer_index(FixSession* session)
        {
            const std::lock_guard<UserspaceSpinlock<>> lock(m_tables_lock);
 
            if (m_session_peer_index_table.find(session) == m_session_peer_index_table.end())
            {
                return static_cast<std::size_t>(-1);
            }
 
            return m_session_peer_index_table[session];
        }
 
        void remove(std::size_t peer_index, FixSession* session)
        {
            const std::lock_guard<UserspaceSpinlock<>> lock(m_tables_lock);
 
            m_peer_index_session_table.erase(peer_index);
            m_session_peer_index_table.erase(session);
        }
 
    private:
        std::unordered_map<std::size_t, FixSession*> m_peer_index_session_table;
        std::unordered_map<FixSession*, std::size_t> m_session_peer_index_table;
        mutable UserspaceSpinlock<> m_tables_lock;
};
 
template<typename Transport>
class FixServer : public Transport, public ManagedInstance
{
    public:
        FixServer() = default;
 
        virtual ~FixServer()
        {
            for (auto& entry : m_sessions)
            {
                delete entry.second;
            }
        }
 
        [[nodiscard]] bool create(const std::string& server_name, const std::string server_config_file_path)
        {
            m_name = server_name;
 
            if (m_settings.load_from_config_file(server_config_file_path, server_name) == false)
            {
                LLFIX_LOG_ERROR("Loading settings for server " + server_name + " failed : " + m_settings.config_load_error);
                return false;
            }
 
            if (m_settings.validate() == false)
            {
                LLFIX_LOG_ERROR("FixServerSettings for " + m_name + " validation failed : " + m_settings.validation_error);
                return false;
            }
 
            m_is_ha_primary = m_settings.starts_as_primary_instance;
            m_unknown_session_context_lock.initialise();
 
            TCPReactorOptions reactor_options;
 
            reactor_options.m_accept_timeout_seconds = m_settings.accept_timeout_seconds;
            reactor_options.m_async_io_timeout_nanoseconds = m_settings.async_io_timeout_nanoseconds;
            reactor_options.m_busy_poll_microseconds = m_settings.busy_poll_microseconds;
            reactor_options.m_cpu_core_id = m_settings.cpu_core_id;
            reactor_options.m_worker_thread_count = m_settings.worker_thread_count;
            reactor_options.m_send_try_count = m_settings.send_try_count;
            reactor_options.m_disable_nagle = m_settings.disable_nagle;
            reactor_options.m_enable_quick_ack = m_settings.quick_ack;
            reactor_options.m_max_poll_events = m_settings.max_poll_events;
            reactor_options.m_nic_interface_ip = m_settings.nic_address;
            reactor_options.m_nic_interface_name = m_settings.nic_name;
            reactor_options.m_nic_ringbuffer_rx_size = m_settings.nic_ringbuffer_rx_size;
            reactor_options.m_nic_ringbuffer_tx_size = m_settings.nic_ringbuffer_tx_size;
            reactor_options.m_pending_connection_queue_size = m_settings.pending_connection_queue_size;
            reactor_options.m_port = m_settings.accept_port;
            reactor_options.m_rx_buffer_capacity = m_settings.rx_buffer_capacity;
            reactor_options.m_receive_size = m_settings.receive_size;
            reactor_options.m_socket_rx_size = m_settings.socket_rx_size;
            reactor_options.m_socket_tx_size = m_settings.socket_tx_size;
            reactor_options.m_spin_count = m_settings.spin_count;
            #ifdef LLFIX_ENABLE_OPENSSL
            reactor_options.m_use_ssl = m_settings.use_ssl;
            reactor_options.m_ssl_verify_peer = m_settings.ssl_verify_peer;
            reactor_options.m_ssl_ca_pem_file = m_settings.ssl_ca_pem_file;
            reactor_options.m_ssl_cert_pem_file = m_settings.ssl_certificate_pem_file;
            reactor_options.m_ssl_private_key_pem_file = m_settings.ssl_private_key_pem_file;
            reactor_options.m_ssl_private_key_password = m_settings.ssl_private_key_password;
            reactor_options.m_ssl_version = m_settings.ssl_version;
            reactor_options.m_ssl_cipher_suite = m_settings.ssl_cipher_suite;
            reactor_options.m_ssl_crl_path = m_settings.ssl_crl_path;
            #endif
 
            this->set_params(reactor_options);
 
            if constexpr (Transport::is_multithreaded())
            {
                this->register_acceptor_callback(
                    [this](std::size_t peer_index)
                    {
                        LLFIX_UNUSED(peer_index);
                        this->process_admin_commands_of_all_non_live_sessions();
                    });
 
                this->register_acceptor_termination_callback(
                    [this](std::size_t peer_index)
                    {
                        LLFIX_UNUSED(peer_index);
                        this->destroy_thread_local_unknown_session_context();
                    });
 
                this->register_worker_callback(
                    [this](std::size_t peer_index)
                    {
                        this->process_session(peer_index);
                    });
 
                this->register_worker_termination_callback(
                    [this](std::size_t peer_index)
                    {
                        LLFIX_UNUSED(peer_index);
                        this->destroy_thread_local_unknown_session_context();
                    });
            }
            else
            {
                this->register_callback(std::bind(&FixServer::process, this));
                this->register_termination_callback(std::bind(&FixServer::destroy_thread_local_unknown_session_context, this));
            }
 
            LLFIX_LOG_INFO(m_name + " : Loaded server config =>\n" + m_settings.to_string());
            LLFIX_LOG_INFO("FixServer " + m_name + " creation success");
 
            return true;
        }
 
        [[nodiscard]] bool add_session(const std::string& session_name, FixSessionSettings& session_settings)
        {
            return internal_add_session(session_name, session_settings);
        }
 
        [[nodiscard]] bool add_session(const std::string& session_config_file_path, const std::string& session_name)
        {
            FixSessionSettings session_settings;
 
            if (session_settings.load_from_config_file(session_config_file_path, session_name) == false)
            {
                LLFIX_LOG_ERROR("Loading settings for session " + session_name + " failed : " + session_settings.config_load_error);
                return false;
            }
 
            return internal_add_session(session_name, session_settings);
        }
 
        [[nodiscard]] bool add_sessions_from(const std::string& session_config_file_path)
        {
            Configuration config_file;
            std::string config_load_error;
 
            if (config_file.load_from_file(session_config_file_path, config_load_error) == false)
            {
                return false;
            }
 
            std::vector<std::string> session_names;
            config_file.get_group_names(session_names);
 
            int added_session_count{0};
 
            for (const auto& session_name : session_names)
            {
                auto lowered_session_name = StringUtilities::to_lower(session_name);
 
                if (StringUtilities::contains(lowered_session_name, "session"))
                {
                    FixSessionSettings session_settings;
 
                    if (session_settings.load_from_config_file(session_config_file_path, session_name) == false)
                    {
                        LLFIX_LOG_ERROR("Loading settings for session " + session_name + " failed : " + session_settings.config_load_error);
                        return false;
                    }
 
                    if (internal_add_session(session_name, session_settings) == false)
                    {
                        return false;
                    }
 
                    added_session_count++;
                }
            }
 
            if(added_session_count == 0)
            {
                LLFIX_LOG_ERROR(m_name + " : Could not find any session in " + session_config_file_path + ". Make sure they have 'SESSION' in their names.");
                return false;
            }
 
            return true;
        }
 
        std::size_t get_session_count() const
        {
            return m_sessions.size();
        }
 
        bool is_instance_ha_primary() const override
        {
            return m_is_ha_primary;
        }
 
        FixSession* get_session(const std::string& session_name) override
        {
            if (has_session(session_name) == false)
            {
                return nullptr;
            }
 
            return m_sessions[session_name];
        }
 
        std::string get_session_name(FixSession* session)
        {
            assert(session);
 
            for (const auto& session_entry : m_sessions)
            {
                if(session_entry.second == session)
                {
                    return session_entry.first;
                }
            }
 
            return "";
        }
 
        void get_session_names(std::vector<std::string>& target) override
        {
            for (const auto& iter : m_sessions)
            {
                target.push_back(iter.first);
            }
        }
 
        std::string get_name() const override
        {
            return m_name;
        }
 
        template<typename... Args>
        void specify_repeating_group(Args... args)
        {
            FixSession::get_repeating_group_specs().specify_repeating_group(args...);
        }
 
        #ifdef LLFIX_ENABLE_BINARY_FIELDS
 
        void specify_binary_field(const std::string& message_type, uint32_t tag_length, uint32_t tag_data)
        {
            FixSession::get_binary_field_specs().specify_binary_field(message_type, tag_length, tag_data);
        }
        #endif
 
        OutgoingFixMessage* outgoing_message_instance(FixSession* session)
        {
            return session->get_outgoing_fix_message();
        }
 
        virtual bool send_outgoing_message(FixSession* session, OutgoingFixMessage* message)
        {
            std::size_t encoded_length = 0;
            auto int_sequence_no = session->get_sequence_store()->get_outgoing_seq_no() + 1;
 
            message->encode(session->get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, int_sequence_no, encoded_length);
            return send_bytes<true>(session, session->get_tx_encode_buffer(), encoded_length);
        }
 
        void push_admin_command(const std::string& session_name, ModifyingAdminCommandType type, uint32_t arg = 0) override
        {
            if(llfix_likely(session_name != "*"))
            {
                push_admin_command_internal(session_name, type, arg);
            }
            else
            {
                for (auto& session : m_sessions)
                {
                    push_admin_command_internal(session.first, type, arg);
                }
            }
        }
 
        #ifdef LLFIX_AUTOMATION
        // Used for testing outgoing msg resends  and gap fills
        bool send_fake_outgoing_message(FixSession* session, OutgoingFixMessage* message)
        {
            std::size_t encoded_length = 0;
            auto int_sequence_no = session->get_sequence_store()->get_outgoing_seq_no() + 1;
 
            message->encode(session->get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, int_sequence_no, encoded_length);
 
            auto sequence_store = session->get_sequence_store();
            sequence_store->increment_outgoing_seq_no();
 
            if(session->serialisation_enabled())
                session->get_outgoing_message_serialiser()->write(reinterpret_cast<const void*>(session->get_tx_encode_buffer()), encoded_length, true, sequence_store->get_outgoing_seq_no());
 
            session->set_last_sent_message_timestamp_nanoseconds(VDSO::nanoseconds_monotonic());
 
            return true;
        }
 
        void set_replay_messages_on_incoming_resend_request_for_all_sessions(bool b)
        {
            for (auto& entry : m_sessions)
            {
                entry.second->set_replay_messages_on_incoming_resend_request(b);
            }
        }
        #endif
 
        void shutdown() { this->stop(); }
        // TCP connections
        virtual void on_async_io_error(int error_code, int event_result) override
        {
            LLFIX_UNUSED(error_code);
            LLFIX_UNUSED(event_result);
        }
 
        virtual void on_socket_error(int error_code, int event_result) override
        {
            LLFIX_UNUSED(error_code);
            LLFIX_UNUSED(event_result);
        }
        // Application level messages
        virtual void on_new_order(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        virtual void on_cancel_order(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        virtual void on_replace_order(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        virtual void on_application_level_reject(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        virtual void on_session_level_reject(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        virtual void on_custom_message(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        // Session/admin messages
        virtual void on_logon_request(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);}
        virtual void on_logout_request(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);}
        virtual void on_client_resend_request(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        virtual void on_client_test_request(FixSession* session, const IncomingFixMessage* message) {LLFIX_UNUSED(session); LLFIX_UNUSED(message);};
        virtual void on_client_heartbeat(FixSession* session) {LLFIX_UNUSED(session);};
        // Others
        virtual bool process_incoming_throttling(FixSession* session, const IncomingFixMessage* incoming_fix_message)
        {
            if(session->settings()->throttle_limit == 0)
            {
                return true;
            }
 
            session->throttler()->update();
 
            if(session->throttler()->reached_limit() )
            {
                session->increment_incoming_throttler_exceed_count();
 
                if(session->settings()->throttle_action == IncomingThrottlerAction::WAIT)
                {
                    session->throttler()->wait();
                }
                else if(session->settings()->throttle_action == IncomingThrottlerAction::DISCONNECT)
                {
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " : terminating session " + session->get_name() + " as the other end exceeded the throttle rate : " + std::to_string(session->settings()->throttle_limit));
                    this->process_connection_closure(m_fix_connectors.get_peer_index(session));
                    return false;
                }
                else if(session->settings()->throttle_action == IncomingThrottlerAction::REJECT)
                {
                    send_throttle_reject_message(session, incoming_fix_message);
                    return false;
                }
            }
 
            return true;
        }
        virtual void process_schedule_validator(FixSession* session)
        {
            if (session->is_now_valid_session_datetime() == false)
            {
                LLFIX_LOG_DEBUG("FixServer " + m_name + " : terminating session " + session->get_name() + " due to schedule settings");
                this->process_connection_closure(m_fix_connectors.get_peer_index(session), true);
            }
        }
 
        virtual FixSession* accept_session(std::size_t peer_index, const IncomingFixMessage* incoming_fix_message, const char* buffer, std::size_t buffer_length, uint32_t parser_reject_code)
        {
            if (parser_reject_code != static_cast<uint32_t>(-1))
            {
                std::string message;
                if (parser_reject_code > FixConstants::FIX_MAX_ERROR_CODE)
                {
                    message = FixParserErrorCodes::get_internal_parser_error_description(parser_reject_code) + " : " + FixUtilities::fix_to_human_readible(buffer, buffer_length);
                }
                else
                {
                    char temp_buf[512];
                    std::size_t length{ 0 };
                    FixUtilities::get_reject_reason_text(temp_buf, length, parser_reject_code);
                    message = temp_buf;
                    message += " , message : " + FixUtilities::fix_to_human_readible(buffer, buffer_length);
                }
                process_invalid_logon(peer_index, message);
                return nullptr;
            }
 
            FixSession* session = nullptr;
 
            if (!(incoming_fix_message->has_tag(FixConstants::TAG_BEGIN_STRING) && incoming_fix_message->has_tag(FixConstants::TAG_MSG_SEQ_NUM) && incoming_fix_message->has_tag(FixConstants::TAG_SENDER_COMP_ID) && incoming_fix_message->has_tag(FixConstants::TAG_TARGET_COMP_ID) && incoming_fix_message->has_tag(FixConstants::TAG_MSG_TYPE)))
            {
                process_invalid_logon(peer_index, "One of following required tags is missing : 8,34,49,56,35 , message : " + FixUtilities::fix_to_human_readible(buffer, buffer_length));
                return nullptr;
            }
            else
            {
                auto message_type = incoming_fix_message->get_tag_value_as<char>(FixConstants::TAG_MSG_TYPE);
 
                if (message_type == FixConstants::MSG_TYPE_LOGON)
                {
                    auto incoming_begin_string = incoming_fix_message->get_tag_value_as<std::string>(FixConstants::TAG_BEGIN_STRING);
                    auto incoming_comp_id = incoming_fix_message->get_tag_value_as<std::string>(FixConstants::TAG_SENDER_COMP_ID);
                    auto incoming_target_comp_id = incoming_fix_message->get_tag_value_as<std::string>(FixConstants::TAG_TARGET_COMP_ID);
 
                    session = find_session(incoming_begin_string, incoming_target_comp_id, incoming_comp_id);
 
                    if (session != nullptr)
                    {
                        auto session_state = session->get_state();
 
                        if (is_state_live(session_state)) // This check is the most important part of multithreaded FixServer
                        {
                            std::string message = "Incoming logon message for server " + m_name + " is invalid as the client is already logged on, session : " + session->get_name();
                            process_invalid_logon(peer_index, message);
                            return nullptr;
                        }
 
                        if(session_state == SessionState::DISABLED)
                        {
                            std::string message = "Cannot accept connection attempt as session " + session->get_name() + " is disabled. Closing connection.";
                            process_invalid_logon(peer_index, message);
                            return nullptr;
                        }
 
                        if (session->is_now_valid_session_datetime() == false)
                        {
                            std::string message = "Cannot accept connection attempt for session " + session->get_name() + " due to schedule settings. Closing connection.";
                            process_invalid_logon(peer_index, message);
                            return nullptr;
                        }
 
                        m_fix_connectors.update(peer_index, session);
 
                        if constexpr(Transport::is_multithreaded())
                        {
                            this->mark_connector_as_ready_for_worker_thread_dispatch(peer_index);
                        }
                        session->reset_flags();
                        session->get_incoming_fix_message()->reset();
                        session->get_incoming_fix_message()->copy_non_dirty_tag_values_from(*incoming_fix_message);
 
                        if(incoming_fix_message->has_tag(FixConstants::TAG_RESET_SEQ_NUM_FLAG))
                        {
                            if(incoming_fix_message->get_tag_value_as<char>(FixConstants::TAG_RESET_SEQ_NUM_FLAG) == FixConstants::FIX_BOOLEAN_TRUE)
                            {
                                session->get_sequence_store()->reset_numbers();
                            }
                        }
 
                        session->set_state(SessionState::PENDING_LOGON);
                        return session;
                    }
                    else
                    {
                        std::string message = "Could not find a session for incoming message : 8=" + incoming_begin_string + " 49=" + incoming_comp_id + " 56=" + incoming_target_comp_id + " , message : " + FixUtilities::fix_to_human_readible(buffer, buffer_length);
                        process_invalid_logon(peer_index, message);
                        return nullptr;
                    }
                }
                else
                {
                    std::string message = std::string("Received a message with a msgtype different than A (") + message_type + ") while expecting a logon message : " + FixUtilities::fix_to_human_readible(buffer, buffer_length);
                    process_invalid_logon(peer_index, message);
                    return nullptr;
                }
            }
        }
 
        virtual void process_logon_request(FixSession* session, std::size_t peer_index, const IncomingFixMessage* message)
        {
            if (validate_logon_request(session, peer_index, message) == false)
            {
                process_invalid_logon(peer_index, "");
                return;
            }
 
            if (authenticate_logon_request(session, message) == false)
            {
                std::string message = "Authentication failed for incoming logon message for server " + m_name + " for session " + session->get_name();
                process_invalid_logon(peer_index, message);
                return;
            }
 
            auto requested_heartbeat_interval = message->get_tag_value_as<uint32_t>(FixConstants::TAG_HEART_BT_INT);
            auto requested_heartbeat_interval_nanoseconds = static_cast<uint64_t>(requested_heartbeat_interval) * static_cast<uint64_t>(1'000'000'000);
 
            session->set_heartbeart_interval_in_nanoseconds(requested_heartbeat_interval_nanoseconds);
            session->set_outgoing_test_request_interval_in_nanoseconds(static_cast<uint64_t>(requested_heartbeat_interval_nanoseconds * session->settings()->outgoing_test_request_interval_multiplier));
 
            send_logon_response(session, message);
            session->set_state(SessionState::LOGGED_ON);
 
            LLFIX_LOG_DEBUG(m_name + ", session logged on : " + session->get_name() + " , peer index : " + std::to_string(peer_index));
 
            do_post_logon_sequence_number_check(session);
 
            on_logon_request(session, message);
        }
 
        virtual bool authenticate_logon_request(FixSession* session, const IncomingFixMessage* message)
        {
            LLFIX_UNUSED(session);
            LLFIX_UNUSED(message);
            return true;
        }
    public:
        void on_data_ready(std::size_t peer_index) override
        {
            auto read = this->receive(peer_index);
 
            if (read > 0 && read <= static_cast<int>(this->m_options.m_rx_buffer_capacity))
            {
                auto buffer_size = this->get_rx_buffer_size(peer_index);
                if(buffer_size >0 && buffer_size != static_cast<std::size_t>(-1))
                    process_rx_buffer(peer_index, this->get_rx_buffer(peer_index), buffer_size);
            }
 
            this->receive_done(peer_index);
        }
 
        void on_client_connected(std::size_t peer_index) override
        {
            LLFIX_LOG_DEBUG("FixServer " + m_name + " : new client connected , peer index : " + std::to_string(peer_index));
        }
 
        void on_client_disconnected(std::size_t peer_index) override
        {
            Transport::on_client_disconnected(peer_index);
 
            if (m_fix_connectors.has_peer(peer_index) == false)
                return;
 
            // A logged on client
            auto session = m_fix_connectors.get_session(peer_index);
 
            if(session != nullptr)
            {
                LLFIX_LOG_DEBUG("FixServer " + m_name + " : session " + session->get_name() + " disconnected , peer index : " + std::to_string(peer_index));
                on_client_disconnected(session);
            }
 
            // Remove entries
            m_fix_connectors.remove(peer_index, session);
        }
 
        void on_client_disconnected(FixSession* session)
        {
            session->set_state(SessionState::DISCONNECTED);
        }
 
        FixServerSettings& get_settings() { return m_settings; }
 
        std::string get_settings_as_string(const std::string& delimiter) override
        {
            return m_settings.to_string(delimiter);
        }
 
        void set_message_persist_plugin(MessagePersistPlugin* plugin)
        {
            assert(plugin);
            m_message_persist_plugin = plugin;
        }
 
    protected:
        // ADMIN
        virtual bool send_heartbeat(FixSession* session, FixString* test_request_id)
        {
            session->build_heartbeat_message(outgoing_message_instance(session), test_request_id);
            return send_outgoing_message(session, outgoing_message_instance(session));
        }
 
        virtual bool send_logon_response(FixSession* session, const IncomingFixMessage* message)
        {
            LLFIX_UNUSED(message);
            auto response = outgoing_message_instance(session);
            response->set_msg_type(FixConstants::MSG_TYPE_LOGON);
 
            // TAG 98 ENCRYPTION METHOD
            response->set_tag(FixConstants::TAG_ENCRYPT_METHOD, 0);
 
            // TAG 108 HEARTBEAT INTERVAL
            response->set_tag(FixConstants::TAG_HEART_BT_INT, static_cast<uint32_t>(session->get_heartbeart_interval_in_nanoseconds() / 1'000'000'000));
 
            // TAG 1137 DEFAULT APP VER ID
            auto default_app_ver_id = session->get_default_app_ver_id();
 
            if (default_app_ver_id.length() > 0)
            {
                response->set_tag(FixConstants::TAG_DEFAULT_APPL_VER_ID, default_app_ver_id);
            }
 
            return send_outgoing_message(session, response);
        }
 
        virtual bool send_logout_message(FixSession* session, const std::string& reason_text = "")
        {
            session->build_logout_message(outgoing_message_instance(session), reason_text);
            return send_outgoing_message(session, outgoing_message_instance(session));
        }
 
        virtual bool send_test_request(FixSession* session)
        {
            session->build_test_request_message(outgoing_message_instance(session));
            return send_outgoing_message(session, outgoing_message_instance(session));
        }
 
        virtual bool send_resend_request(FixSession* session)
        {
            session->build_resend_request_message(outgoing_message_instance(session), "0");
            return send_outgoing_message(session, outgoing_message_instance(session));
        }
 
        virtual bool send_sequence_reset_message(FixSession* session, uint32_t desired_sequence_no)
        {
            /*
                This message is a "hard" gap fill that doesn't respond to an incoming 35=2/resend request but to an internal admin request
                Therefore we can't set 123=Y
            */
            auto message = outgoing_message_instance(session);
            session->build_sequence_reset_message(message, desired_sequence_no);
 
            // ENCODE
            std::size_t encoded_length = 0;
            message->encode(session->get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, desired_sequence_no, encoded_length);
 
            // SEND
            return send_bytes<false>(session, session->get_tx_encode_buffer(), encoded_length); // false is for not incrementing seq no for this message
        }
 
        virtual bool send_gap_fill_message(FixSession* session)
        {
            /*
                This message is a gap fill that responds to an incoming 35=2/resend request.
                For "hard" gap fills that don't set 123=Y, see send_sequence_reset_message
            */
            auto message = outgoing_message_instance(session);
            session->build_gap_fill_message(message);
 
            // ENCODE
            std::size_t encoded_length = 0;
            message->encode(session->get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, session->get_incoming_resend_request_begin_no(), encoded_length); // We need to encode with seq no that the peer expects hence using m_incoming_resend_request_begin_no
 
            // SEND
            return send_bytes<false>(session, session->get_tx_encode_buffer(), encoded_length); // false is for not incrementing seq no for this message
        }
 
        virtual void resend_messages_to_client(FixSession* session, uint32_t begining_seq_no, uint32_t end_seq_no)
        {
            for (uint32_t i = begining_seq_no; i <= end_seq_no; i++)
            {
                std::size_t message_length = 0;
                session->get_outgoing_message_serialiser()->read_message(i, session->get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, message_length);
 
                auto message = outgoing_message_instance(session);
                message->load_from_buffer(session->get_tx_encode_buffer(), message_length); // t122(orig sending time) will be handled by load_from_buffer
 
                message->template set_tag<FixMessageComponent::HEADER>(FixConstants::TAG_POSS_DUP_FLAG, FixConstants::FIX_BOOLEAN_TRUE);
 
                if (session->include_t97_during_resends())
                {
                    message->template set_tag<FixMessageComponent::HEADER>(FixConstants::TAG_POSS_RESEND, FixConstants::FIX_BOOLEAN_TRUE);
                }
 
                std::size_t encoded_length = 0;
                message->encode(session->get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, i, encoded_length);
                send_bytes<false>(session, session->get_tx_encode_buffer(), encoded_length); // false is for not incrementing seq no for this message
            }
        }
 
        virtual bool send_reject_message(FixSession* session, const IncomingFixMessage* incoming_message, uint32_t reject_reason_code, const char* buffer_message, std::size_t buffer_message_length, uint32_t error_tag=0)
        {
            LLFIX_UNUSED(incoming_message);
            char reject_reason_text[256];
 
            std::size_t text_length{ 0 };
            FixUtilities::get_reject_reason_text(reject_reason_text, text_length, reject_reason_code);
 
            if (error_tag == 0)
            {
                LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " received an invalid message : " + std::string(reject_reason_text) + " ,message : " + FixUtilities::fix_to_human_readible(buffer_message, buffer_message_length));
            }
            else
            {
                LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " received an invalid message : " + std::string(reject_reason_text) + " ,tag : " + std::to_string(error_tag) + " ,message : " + FixUtilities::fix_to_human_readible(buffer_message, buffer_message_length));
            }
 
            session->build_session_level_reject_message(outgoing_message_instance(session), reject_reason_code, reject_reason_text, error_tag);
            return send_outgoing_message(session, outgoing_message_instance(session));
        }
 
        virtual bool send_throttle_reject_message(FixSession* session, const IncomingFixMessage* incoming_message)
        {
            auto reject_message = outgoing_message_instance(session);
            reject_message->set_msg_type(FixConstants::MSG_TYPE_BUSINESS_REJECT);
 
            reject_message->set_tag(FixConstants::TAG_TEXT, session->settings()->throttler_reject_message);
 
            if (incoming_message->has_tag(FixConstants::TAG_MSG_SEQ_NUM))
            {
                reject_message->set_tag(FixConstants::TAG_REF_SEQ_NUM, incoming_message->get_tag_value_as<uint32_t>(FixConstants::TAG_MSG_SEQ_NUM));
            }
 
            if (incoming_message->has_tag(FixConstants::TAG_MSG_TYPE))
            {
                reject_message->set_tag(FixConstants::TAG_REF_MSG_TYPE, incoming_message->get_tag_value_as<std::string>(FixConstants::TAG_MSG_TYPE));
            }
 
            return send_outgoing_message(session, reject_message);
        }
 
        std::unordered_map<std::string, FixSession*> m_sessions;
 
    private:
        static inline constexpr std::size_t MINIMUM_REQUIRED_INITIAL_BUFFER_FOR_PARSING = 4; // 10=<DELIMITER>
        std::string m_name;
        bool m_is_ha_primary = true;
 
        FixServerSettings m_settings;
 
        FixServerConnectors m_fix_connectors;
 
        MessagePersistPlugin* m_message_persist_plugin = nullptr;
 
        UserspaceSpinlock<> m_unknown_session_context_lock;
 
        bool has_session(const std::string& session_name)
        {
            if (m_sessions.find(session_name) == m_sessions.end())
            {
                return false;
            }
 
            return true;
        }
 
        FixSession* find_session(const std::string& begin_string, const std::string& sender_comp_id, const std::string& target_comp_id)
        {
            FixSession* ret = nullptr;
 
            for (const auto& session_entry : m_sessions)
            {
                if (session_entry.second->settings()->sender_comp_id == sender_comp_id && session_entry.second->settings()->target_comp_id == target_comp_id && session_entry.second->settings()->begin_string == begin_string)
                {
                    return session_entry.second;
                }
            }
 
            return ret;
        }
 
        bool does_any_session_use_path(const std::string& path)
        {
            for (const auto& session_entry : m_sessions)
            {
                if (session_entry.second->settings()->sequence_store_file_path == path)
                {
                    return true;
                }
 
                if (session_entry.second->settings()->incoming_message_serialisation_path == path)
                {
                    return true;
                }
 
                if (session_entry.second->settings()->outgoing_message_serialisation_path == path)
                {
                    return true;
                }
            }
            return false;
        }
 
        bool does_any_session_use_compids(const std::string& sender_comp_id, const std::string& target_comp_id)
        {
            for (const auto& session_entry : m_sessions)
            {
                if (session_entry.second->settings()->sender_comp_id == sender_comp_id && session_entry.second->settings()->target_comp_id == target_comp_id)
                {
                    return true;
                }
            }
            return false;
        }
 
        bool internal_add_session(const std::string& session_name, const FixSessionSettings& session_settings)
        {
            if (has_session(session_name))
            {
                LLFIX_LOG_ERROR(m_name + " : Already loaded a session with name " + session_name + " . You have to specify unique names for sessions.");
                return false;
            }
 
            if (does_any_session_use_path(session_settings.sequence_store_file_path))
            {
                LLFIX_LOG_ERROR(m_name + " : session " + session_name + "'s sequence_store_file_path already being used in another session.");
                return false;
            }
 
            if (does_any_session_use_path(session_settings.incoming_message_serialisation_path))
            {
                LLFIX_LOG_ERROR(m_name + " : session " + session_name + "'s incoming_message_serialisation_path already being used in another session.");
                return false;
            }
 
            if (does_any_session_use_path(session_settings.outgoing_message_serialisation_path))
            {
                LLFIX_LOG_ERROR(m_name + " : session " + session_name + "'s outgoing_message_serialisation_path already being used in another session.");
                return false;
            }
 
            if (does_any_session_use_compids(session_settings.sender_comp_id, session_settings.target_comp_id))
            {
                std::string warning_message = "WARNING (" + m_name + "): session " + session_name + "'s sender and target comp ids are already being used in another session.";
                LLFIX_LOG_WARNING(warning_message);
                fprintf(stderr, "%s", warning_message.c_str());
            }
 
            FixSession* current_session = new (std::nothrow) FixSession;
 
            if (current_session == nullptr)
            {
                LLFIX_LOG_ERROR(m_name + " : Failed to allocate fix session for " + session_name);
                return false;
            }
 
            session_settings.is_server = true;
            session_settings.tx_encode_buffer_capacity = m_settings.tx_encode_buffer_capacity; // Propagate
 
            if (current_session->initialise(session_name, session_settings) == false)
            {
                delete current_session;
                return false;
            }
 
            m_sessions[session_name] = current_session;
 
            if (m_settings.starts_as_primary_instance == false)
            {
                disable_session(current_session);
            }
 
            return true;
        }
 
        void process_session(std::size_t peer_index)
        {
            static_assert(Transport::is_multithreaded());
 
            FixSession* sess = m_fix_connectors.get_session(peer_index);
 
            if (sess)
            {
                sess->lock();
                process_session(sess);
                sess->unlock();
            }
        }
 
        void process_session(FixSession* session)
        {
            static_assert(Transport::is_multithreaded());
 
            if (llfix_likely(this->m_is_stopping.load() == false))
            {
                process_admin_commands(session);
 
                auto current_state = session->get_state();
 
                if (is_state_live(current_state))
                {
                    process_outgoing_resend_request_if_necessary(session, VDSO::nanoseconds_monotonic());
                    process_outgoing_test_request_if_necessary(session, VDSO::nanoseconds_monotonic());
                    respond_to_resend_request_if_necessary(session);
                    respond_to_test_request_if_necessary(session);
                    send_heartbeat_if_necessary(session, VDSO::nanoseconds_monotonic());
 
                    process_schedule_validator(session);
                }
            }
            else
            {
                send_logout_message(session);
                on_client_disconnected(session);
            }
        }
 
        void process_admin_commands_of_all_non_live_sessions()
        {
            static_assert(Transport::is_multithreaded());
 
            for (auto& iter : m_sessions)
            {
                auto state = iter.second->get_state();
 
                if (is_state_live(state) == false)
                {
                    iter.second->lock();
                    process_admin_commands(iter.second);
                    iter.second->unlock();
                }
            }
 
        }
 
        void process()
        {
            static_assert(!Transport::is_multithreaded());
 
            if (llfix_likely(this->m_is_stopping.load() == false))
            {
                for (const auto& session_entry : m_sessions)
                {
                    FixSession* current_session = session_entry.second;
 
                    process_admin_commands(current_session);
 
                    auto current_state = current_session->get_state();
 
                    if (is_state_live(current_state))
                    {
                        process_outgoing_resend_request_if_necessary(current_session, VDSO::nanoseconds_monotonic());
                        process_outgoing_test_request_if_necessary(current_session, VDSO::nanoseconds_monotonic());
                        respond_to_resend_request_if_necessary(current_session);
                        respond_to_test_request_if_necessary(current_session);
                        send_heartbeat_if_necessary(current_session, VDSO::nanoseconds_monotonic());
 
                        process_schedule_validator(current_session);
                    }
                }
            }
            else
            {
                send_logout_messages_to_all_sessions();
            }
        }
 
        void push_admin_command_internal(const std::string& session_name, ModifyingAdminCommandType type, uint32_t arg = 0)
        {
            auto admin_command = new (std::nothrow) ModifyingAdminCommand;
 
            if (llfix_likely(admin_command))
            {
                admin_command->type = type;
                admin_command->arg = arg;
                m_sessions[session_name]->get_admin_commands()->push(admin_command); // Session name existence is checked in management server layer
            }
            else
            {
                LLFIX_LOG_ERROR("Failed to process setter admin command for FixServer " + m_name);
            }
        }
 
        void process_admin_commands(FixSession* session)
        {
            ModifyingAdminCommand* admin_command{ nullptr };
 
            if (session->get_admin_commands()->try_pop(&admin_command) == true)
            {
                switch (admin_command->type)
                {
                    case ModifyingAdminCommandType::SET_INCOMING_SEQUENCE_NUMBER:
                    {
                        session->get_sequence_store()->set_incoming_seq_no(admin_command->arg);
                        LLFIX_LOG_DEBUG(m_name + " : processed set incoming sequence number admin command , session name : " + session->get_name() + " , new seq no : " + std::to_string(admin_command->arg) );
                        break;
                    }
 
                    case ModifyingAdminCommandType::SET_OUTGOING_SEQUENCE_NUMBER:
                    {
                        session->get_sequence_store()->set_outgoing_seq_no(admin_command->arg);
                        LLFIX_LOG_DEBUG(m_name + " : processed set outgoing sequence number admin command , session name : " + session->get_name() + " , new seq no : " + std::to_string(admin_command->arg) );
                        break;
                    }
 
                    case ModifyingAdminCommandType::SEND_SEQUENCE_RESET:
                    {
                        if( session->get_state() == SessionState::LOGGED_ON )
                        {
                            send_sequence_reset_message(session, admin_command->arg);
                            LLFIX_LOG_DEBUG(m_name + " : processed send sequence reset admin command , session name : " + session->get_name() + " , new seq no : " + std::to_string(admin_command->arg) );
                        }
                        else
                        {
                            LLFIX_LOG_ERROR(m_name + " : dropping send sequence reset admin command as not logged on , session name : " + session->get_name());
                        }
 
                        break;
                    }
 
                    case ModifyingAdminCommandType::DISABLE_SESSION:
                    {
                        disable_session(session);
                        LLFIX_LOG_DEBUG(m_name + " : processed disable session admin command for session " + session->get_name() + " , new session state : Disabled");
                        break;
                    }
 
                    case ModifyingAdminCommandType::ENABLE_SESSION:
                    {
                        if(m_is_ha_primary == true)
                        {
                            enable_session(session);
                            LLFIX_LOG_DEBUG(m_name + " : processed enable session admin command for " + session->get_name());
                        }
                        else
                        {
                            LLFIX_LOG_ERROR(m_name + " ignoring enable session admin command for " + session->get_name() + " since not a primary instance" );
                        }
                        break;
                    }
 
                    case ModifyingAdminCommandType::SET_IS_HA_PRIMARY_INSTANCE:
                    {
                        m_is_ha_primary = admin_command->arg == 1 ? true : false;
 
                        if(admin_command->arg == 1)
                        {
                            enable_session(session);
                            LLFIX_LOG_DEBUG(m_name + " : processed set is ha primary instance 1 admin command for " + session->get_name());
                        }
                        else
                        {
                            disable_session(session);
                            LLFIX_LOG_DEBUG(m_name + " : processed set is ha primary instance 0 admin command for " + session->get_name());
                        }
 
                        break;
                    }
 
 
                    default: break;
                }
 
                delete admin_command;
            }
        }
 
        void disable_session(FixSession* session)
        {
            if (is_state_live(session->get_state()))
            {
                this->process_connection_closure(m_fix_connectors.get_peer_index(session), true);
            }
            session->set_state(SessionState::DISABLED);
        }
 
        void enable_session(FixSession* session)
        {
            if(session->get_state() == SessionState::DISABLED)
            {
                session->set_state(SessionState::DISCONNECTED);
                if(m_settings.refresh_resend_cache_during_promotion)
                    session->reinitialise_outgoing_serialiser();
            }
        }
 
        // ADMIN UTILITY
        void process_outgoing_resend_request_if_necessary(FixSession* session, uint64_t current_timestamp_nanoseconds)
        {
            if (session->get_state() == SessionState::IN_RETRANSMISSION_INITIATED_BY_SELF)
            {
                // TERMINATE SESSION IF NECESSARY
                uint64_t delta_nanoseconds = current_timestamp_nanoseconds - session->outgoing_resend_request_timestamp_nanoseconds();
                uint64_t required_delta_nanoseconds = static_cast<uint64_t>(session->outgoing_resend_request_expire_secs()) * 1'000'000'000;
 
                if (delta_nanoseconds >= required_delta_nanoseconds)
                {
                    // We need to terminate the session as the other side didn't respond properly to our outgoing resend request
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " : terminating session " + session->get_name() + " as the other end didn't respond to 35=2/resend request in pre-configured timeout (outgoing_resend_request_expire_secs) : " + std::to_string(session->outgoing_resend_request_expire_secs()));
                    this->process_connection_closure(m_fix_connectors.get_peer_index(session));
                }
            }
            else
            {
                // SEND IF NECESSARY
                if (session->needs_to_send_resend_request())
                {
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " peer " + std::to_string(m_fix_connectors.get_peer_index(session)) + " : sending resend request , begin no : " + std::to_string(session->get_outgoing_resend_request_begin_no()));
                    send_resend_request(session);
                    session->set_outgoing_resend_request_timestamp_nanoseconds(current_timestamp_nanoseconds);
                    session->set_state(llfix::SessionState::IN_RETRANSMISSION_INITIATED_BY_SELF);
                    session->set_needs_to_send_resend_request(false);
                }
            }
        }
 
        void process_outgoing_test_request_if_necessary(FixSession* session, uint64_t current_timestamp_nanoseconds)
        {
            if (session->expecting_response_for_outgoing_test_request() == false)
            {
                // SEND IF NECESSARY
                auto delta_nanoseconds = current_timestamp_nanoseconds - session->last_received_message_timestamp_nanoseconds();
 
                if (delta_nanoseconds >= (session->get_outgoing_test_request_interval_in_nanoseconds()))
                {
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " : sending test request");
                    send_test_request(session);
                    session->set_outgoing_test_request_timestamp_nanoseconds(current_timestamp_nanoseconds);
                    session->set_expecting_response_for_outgoing_test_request(true);
                }
            }
            else
            {
                // TERMINATE SESSION IF NECESSARY
                auto delta_nanoseconds = current_timestamp_nanoseconds - session->outgoing_test_request_timestamp_nanoseconds();
 
                if (delta_nanoseconds >= (session->get_heartbeart_interval_in_nanoseconds() * static_cast<uint64_t>(2)))
                {
                    session->set_expecting_response_for_outgoing_test_request(false);
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " : terminating session " + session->get_name() + " as the other end didn't respond to 35=1/test request");
                    this->process_connection_closure(m_fix_connectors.get_peer_index(session));
                }
            }
        }
 
        void respond_to_resend_request_if_necessary(FixSession* session)
        {
            if (session->needs_responding_to_incoming_resend_request())
            {
                const auto last_outgoing_seq_no = session->get_sequence_store()->get_outgoing_seq_no();
 
                // Partial replays not supported therefore incoming end no should be either 0 or same as last outgoing seq no
                bool will_replay = (session->serialisation_enabled()) && (session->replay_messages_on_incoming_resend_request() && (session->get_incoming_resend_request_end_no() == 0
                                   || session->get_incoming_resend_request_end_no() == last_outgoing_seq_no) && session->get_incoming_resend_request_begin_no()<=last_outgoing_seq_no);
 
                if(last_outgoing_seq_no > session->get_incoming_resend_request_begin_no())
                    if(last_outgoing_seq_no-session->get_incoming_resend_request_begin_no()+1 > session->max_resend_range())
                        will_replay = false;
 
                if(will_replay)
                {
                    for (uint32_t i = session->get_incoming_resend_request_begin_no(); i <= last_outgoing_seq_no; i++)
                    {
                        if (session->get_outgoing_message_serialiser()->has_message_in_memory(i) == false)
                        {
                            will_replay = false;
                            break;
                        }
                    }
                }
 
                if(will_replay)
                {
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " : replaying messages , begin seq no : " + std::to_string(session->get_incoming_resend_request_begin_no()) + " , end seq no : " + std::to_string(last_outgoing_seq_no));
                    resend_messages_to_client(session, session->get_incoming_resend_request_begin_no(), last_outgoing_seq_no);
                }
                else
                {
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " : sending gap fill message");
                    send_gap_fill_message(session);
                }
 
                session->set_needs_responding_to_incoming_resend_request(false);
                session->set_state(llfix::SessionState::LOGGED_ON);
            }
        }
 
        void respond_to_test_request_if_necessary(FixSession* session)
        {
            if (session->needs_responding_to_incoming_test_request())
            {
                if (session->get_incoming_test_request_id()->length() > 0)
                {
                    send_heartbeat(session , session->get_incoming_test_request_id());
                    session->get_incoming_test_request_id()->set_length(0);
                }
                else
                {
                    // We should not hit here , but better than risking disconnection
                    send_heartbeat(session, nullptr);
                }
 
                LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " : responded to incoming test request");
                session->set_needs_responding_to_incoming_test_request(false);
            }
        }
 
        void send_heartbeat_if_necessary(FixSession* session, uint64_t current_timestamp_nanoseconds)
        {
            auto delta_nanoseconds = current_timestamp_nanoseconds - session->last_sent_message_timestamp_nanoseconds();
 
            if (delta_nanoseconds >= (session->get_heartbeart_interval_in_nanoseconds() * static_cast<uint64_t>(9) / static_cast<uint64_t>(10))) // 0.9 for safety
            {
                send_heartbeat(session, nullptr);
            }
        }
 
        void send_logout_messages_to_all_sessions()
        {
            for (const auto& session_entry : m_sessions)
            {
                FixSession* current_session = session_entry.second;
 
                if (current_session->get_state() != SessionState::DISCONNECTED)
                {
                    send_logout_message(current_session);
                    on_client_disconnected(current_session);
                }
            }
        }
        // TO CLIENT
        template <bool increment_outgoing_seq_no>
        LLFIX_FORCE_INLINE bool send_bytes(FixSession* session, const char* buffer, std::size_t buffer_size)
        {
            std::size_t peer_index = m_fix_connectors.get_peer_index(session);
 
            if (peer_index == static_cast<std::size_t>(-1))
                return false;
 
            auto sequence_store = session->get_sequence_store();
 
            auto ret = this->send(peer_index, buffer, buffer_size);
 
            if constexpr (increment_outgoing_seq_no == true)
            {
                if(ret==true)
                    sequence_store->increment_outgoing_seq_no();
            }
 
            if(session->serialisation_enabled())
                session->get_outgoing_message_serialiser()->write(reinterpret_cast<const void*>(buffer), buffer_size, ret, sequence_store->get_outgoing_seq_no());
 
            if (m_message_persist_plugin)
                m_message_persist_plugin->persist_outgoing_message(session->get_name(), sequence_store->get_outgoing_seq_no(), buffer, buffer_size, ret);
 
            session->set_last_sent_message_timestamp_nanoseconds(VDSO::nanoseconds_monotonic());
 
            return ret;
        }
 
        // THREAD LOCAL UNKNOWN SESSION CONTEXT
        UnknownSessionContext* get_thread_local_unknown_session_context()
        {
            const std::lock_guard<UserspaceSpinlock<>> lock(m_unknown_session_context_lock);
 
            UnknownSessionContext* unknown_session_context = nullptr;
            unknown_session_context = reinterpret_cast<UnknownSessionContext*>(ThreadLocalStorage::get_instance().get());
 
            if (unknown_session_context == nullptr)
            {
                unknown_session_context = new(std::nothrow) UnknownSessionContext;
                ThreadLocalStorage::get_instance().set(unknown_session_context);
            }
 
            if (unknown_session_context)
            {
                if (unknown_session_context->m_incoming_fix_message.initialise() == false)
                {
                    LLFIX_LOG_ERROR(m_name + " : Failed to initialise unknown session context's incoming_fix_message");
                    return nullptr;
                }
 
                if (unknown_session_context->m_fix_string_view_cache.create(256) == false)
                {
                    LLFIX_LOG_ERROR(m_name + " : Failed to create  unknown session context's fix string view cache");
                    return nullptr;
                }
            }
 
            return unknown_session_context;
        }
 
        void destroy_thread_local_unknown_session_context()
        {
            const std::lock_guard<UserspaceSpinlock<>> lock(m_unknown_session_context_lock);
 
            UnknownSessionContext* unknown_session_context{ nullptr };
            unknown_session_context = reinterpret_cast<UnknownSessionContext*>(ThreadLocalStorage::get_instance().get());
 
            if (unknown_session_context)
            {
                delete unknown_session_context;
                ThreadLocalStorage::get_instance().set(nullptr);
            }
        }
 
        // FROM CLIENT
        LLFIX_HOT void process_rx_buffer(std::size_t peer_index, char* buffer, std::size_t buffer_size)
        {
            std::size_t buffer_read_index = 0;
 
            if (llfix_unlikely(buffer_size < MINIMUM_REQUIRED_INITIAL_BUFFER_FOR_PARSING))
            {
                this->set_incomplete_buffer(peer_index, buffer, buffer_size);
                return;
            }
 
            FixSession* session = m_fix_connectors.get_session(peer_index);
 
            // FIND OUT THE END
            int final_tag10_delimiter_index{ -1 };
            int current_index = static_cast<int>(buffer_size - 1);
 
            if(llfix_unlikely(FixUtilities::find_delimiter_from_end(buffer, buffer_size, current_index) == false))
            {
                // We don't have the entire message
                this->set_incomplete_buffer(peer_index, buffer, buffer_size);
                return;
            }
 
            FixUtilities::find_tag10_start_from_end(buffer, buffer_size, current_index, final_tag10_delimiter_index);
 
            if (final_tag10_delimiter_index == -1)
            {
                // We don't have the entire message
                this->set_incomplete_buffer(peer_index, buffer, buffer_size);
                return;
            }
            // AT THIS POINT WE HAVE AT LEAST ONE COMPLETE MESSAGE
            LLFIX_ALIGN_CODE_32;
            while (true)
            {
                // FIND OUT THE BEGIN STRING
                int begin_string_offset{-1};
                FixUtilities::find_begin_string_position(buffer+buffer_read_index, buffer_size-buffer_read_index, begin_string_offset);
 
                if(llfix_likely(begin_string_offset>=0))
                {
                    buffer_read_index += begin_string_offset;
                }
                else
                {
                    LLFIX_LOG_ERROR(m_name + " received a message with no begin string : " + FixUtilities::fix_to_human_readible(buffer+buffer_read_index, buffer_size-buffer_read_index));
                    return;
                }
                IncomingFixMessage* incoming_fix_message{ nullptr };
                ObjectCache<FixStringView>* fix_string_view_cache{ nullptr };
 
                if (llfix_likely(session != nullptr))
                {
                    incoming_fix_message = session->get_incoming_fix_message();
                    fix_string_view_cache = session->get_fix_string_view_cache();
                }
                else
                {
                    UnknownSessionContext* unknown_session_context = get_thread_local_unknown_session_context();
 
                    if (unknown_session_context)
                    {
                        incoming_fix_message = &(unknown_session_context->m_incoming_fix_message);
                        fix_string_view_cache = &(unknown_session_context->m_fix_string_view_cache);
                    }
                    else
                    {
                        this->set_incomplete_buffer(peer_index, buffer, buffer_size);
                        return;
                    }
                }
 
                incoming_fix_message->reset();
                fix_string_view_cache->reset_pointer();
 
                current_index = static_cast<int>(buffer_read_index);
                bool looking_for_equals = true;
 
                int current_tag_start = static_cast<int>(buffer_read_index);
                int current_tag_len{ 0 };
 
                int current_value_start = static_cast<int>(buffer_read_index);
                int current_value_len{ 0 };
 
                int current_tag_10_delimiter_index = -1;
                int current_tag_35_tag_start_index = -1;
 
                uint32_t parser_reject_code = static_cast<uint32_t>(-1);
                uint32_t current_tag_index = 0;
 
                uint32_t encoded_current_message_type = static_cast<uint32_t>(-1);
 
                bool in_a_repeating_group{ false };
                uint32_t current_rg_count_tag{ 0 };
 
                #ifdef LLFIX_ENABLE_BINARY_FIELDS
                int binary_field_length = 0;
                int binary_field_counter = 0;
                #endif
 
                LLFIX_ALIGN_CODE_32;
                while (true)
                {
                    char current_char = buffer[current_index];
 
                    if (looking_for_equals)
                    {
                        if (current_char == FixConstants::FIX_EQUALS)
                        {
                            current_tag_len = current_index - current_tag_start;
                            current_value_start = current_index + 1;
                            looking_for_equals = false;
                        }
                        else if (llfix_unlikely(current_char == FixConstants::FIX_DELIMITER))
                        {
                            current_tag_start = current_index + 1;
                            parser_reject_code = FixParserErrorCodes::NO_EQUALS_SIGN;
                        }
                    }
                    #ifdef LLFIX_ENABLE_BINARY_FIELDS
                    else
                    {
                        if(llfix_unlikely(binary_field_length>0))
                            if(binary_field_counter < binary_field_length)
                                binary_field_counter++;
                    }
 
                    if (looking_for_equals == false && current_char == FixConstants::FIX_DELIMITER)
                    {
                        bool reached_value_end = true;
 
                        if(llfix_unlikely(binary_field_length>0))
                        {
                            if(binary_field_length>binary_field_counter)
                            {
                                reached_value_end = false;
                            }
                            else if(binary_field_length == binary_field_counter)
                            {
                                binary_field_length=0;
                                binary_field_counter=0;
                            }
                        }
 
                        if(llfix_likely(reached_value_end))
                        {
 
                    #else
                    else if (current_char == FixConstants::FIX_DELIMITER)
                    {
                    #endif
                            current_value_len = current_index - current_value_start;
 
                            if (llfix_unlikely(current_value_len == 0))
                            {
                                parser_reject_code = FixConstants::FIX_ERROR_CODE_TAG_WITHOUT_VALUE;
                            }
 
                            bool is_current_tag_numeric{ true };
                            FixSession::validate_tag_format(buffer + current_tag_start, static_cast<std::size_t>(current_tag_len), is_current_tag_numeric, parser_reject_code);
 
                            if (llfix_likely(is_current_tag_numeric))
                            {
                                // RECORD THE CURRENT TAG VALUE PAIR
                                uint32_t tag = Converters::chars_to_unsigned_int<uint32_t>(buffer + current_tag_start, static_cast<std::size_t>(current_tag_len));
 
                                if(llfix_unlikely(tag == 0))
                                {
                                    parser_reject_code = FixConstants::FIX_ERROR_CODE_INVALID_TAG_NUMBER;
                                }
 
                                FixStringView* value = fix_string_view_cache->allocate();
                                value->set_buffer(const_cast<char*>(buffer + current_value_start), static_cast<std::size_t>(current_value_len));
 
                                current_tag_index++;
                                FixSession::validate_header_tags_order(tag, current_tag_index, parser_reject_code);
 
                                // Some tags may be used as a group member but also as an individual tag so we can't fully rely on is_a_repeating_group_tag
                                // and need to detect start and end of a group
                                if(encoded_current_message_type != static_cast<uint32_t>(-1))
                                {
                                    if(llfix_likely(!in_a_repeating_group))
                                    {
                                        if (llfix_unlikely(FixSession::get_repeating_group_specs().is_a_repeating_group_count_tag(encoded_current_message_type, tag)))
                                        {
                                            current_rg_count_tag = tag;
                                            in_a_repeating_group = true;
                                        }
                                    }
                                    else
                                    {
                                        if (llfix_likely(FixSession::get_repeating_group_specs().is_a_repeating_group_tag(encoded_current_message_type, current_rg_count_tag, tag) == false))
                                        {
                                            in_a_repeating_group = false;
                                        }
                                    }
 
                                    #ifdef LLFIX_ENABLE_BINARY_FIELDS
                                    if (llfix_unlikely(FixSession::get_binary_field_specs().is_binary_data_length_tag(encoded_current_message_type, tag)))
                                    {
                                        binary_field_length = Converters::chars_to_int<int>(value->data(), value->length());
                                    }
                                    #endif
                                }
 
                                if (llfix_likely(in_a_repeating_group == false))
                                {
                                    if (llfix_likely(incoming_fix_message->has_tag(tag) == false))
                                    {
                                        incoming_fix_message->set_tag(tag, value);
                                    }
                                    else
                                    {
                                        parser_reject_code = FixConstants::FIX_ERROR_CODE_TAG_APPEARS_MORE_THAN_ONCE;
                                    }
                                }
                                else
                                {
                                    incoming_fix_message->set_repeating_group_tag(tag, value);
                                }
 
                                if (tag == FixConstants::TAG_CHECKSUM)
                                {
                                    current_tag_10_delimiter_index = current_index;
                                    FixSession::validate_tag9_and_tag35((incoming_fix_message), current_tag_start, current_tag_35_tag_start_index, parser_reject_code);
                                    break;
                                }
                                else if (tag == FixConstants::TAG_MSG_TYPE)
                                {
                                    current_tag_35_tag_start_index = current_tag_start;
                                    encoded_current_message_type = FixUtilities::pack_message_type(value->to_string_view());
                                }
                            } // if (llfix_likely(is_current_tag_numeric))
 
                            current_tag_start = current_index + 1;
                            looking_for_equals = true;
                        #ifdef LLFIX_ENABLE_BINARY_FIELDS
                        } // if(llfix_likely(reached_value_end))
                        #endif
                    } // else if (current_char == FixConstants::FIX_DELIMITER) or if (looking_for_equals == false && current_char == FixConstants::FIX_DELIMITER)
 
                    // Apart from the check below (else if (static_cast<int>(buffer_read_index) > final_tag10_delimiter_index)),
                    // we also need to check here if found last tag 10 delimiter is before the found tag8
                    if (current_index >= static_cast<int>(buffer_size) - 1)
                    {
                        this->set_incomplete_buffer(peer_index, buffer + buffer_read_index, buffer_size - buffer_read_index);
                        return;
                    }
 
                    current_index++;
                } // while
 
                auto current_message_length = current_tag_10_delimiter_index + 1 - buffer_read_index;
                if (llfix_unlikely(session == nullptr))
                {
                    session = accept_session(peer_index, incoming_fix_message, buffer + buffer_read_index, current_message_length, parser_reject_code);
 
                    if (session == nullptr) // Means that it was an invalid logon attempt and we already closed the connection
                    {
                        return;
                    }
                }
                session->lock();
                process_incoming_fix_message(session, peer_index, buffer + buffer_read_index, current_message_length, parser_reject_code);
                session->unlock();
 
                #ifndef LLFIX_UNIT_TEST
                if(llfix_unlikely(session->get_state() == SessionState::DISCONNECTED)) // process_incoming_fix_message may terminate connection
                {
                    return;
                }
                #endif
 
                buffer_read_index = current_tag_10_delimiter_index + 1;
 
                if (current_tag_10_delimiter_index == static_cast<int>(buffer_size) - 1)
                {
                    this->reset_incomplete_buffer(peer_index);
                    return;
                }
                else if (static_cast<int>(buffer_read_index) > final_tag10_delimiter_index)
                {
                    this->set_incomplete_buffer(peer_index, buffer + buffer_read_index, buffer_size - buffer_read_index);
                    return;
                }
            }
        }
 
        void process_invalid_logon(std::size_t peer_index, const std::string& log_message, bool send_logout = false, const std::string& logout_reason_text = "")
        {
            if(!log_message.empty())
            {
                int peer_port{0};
                std::string peer_ip;
                this->get_peer_details(peer_index, peer_ip, peer_port);
 
                LLFIX_LOG_ERROR("Invalid logon attempt from " + peer_ip + " : " + log_message);
            }
 
            process_connection_closure(peer_index, send_logout, logout_reason_text);
        }
 
        void process_connection_closure(std::size_t peer_index, bool send_logout = false, const std::string& logout_reason_text = "")
        {
            if (send_logout)
            {
                auto session = m_fix_connectors.get_session(peer_index);
 
                if (session != nullptr)
                {
                    send_logout_message(session, logout_reason_text);
                }
            }
 
            this->close_connection(peer_index);
        }
 
        void process_incoming_fix_message(FixSession* session, std::size_t peer_index, const char* buffer_message, std::size_t buffer_message_length, uint32_t parser_reject_code)
        {
            session->set_last_received_message_timestamp_nanoseconds(VDSO::nanoseconds_monotonic());
 
            if(session->serialisation_enabled())
                session->get_incoming_message_serialiser()->write(buffer_message, buffer_message_length, true);
 
            if (llfix_unlikely(session->expecting_response_for_outgoing_test_request()))
            {
                // We are permissive , any message not just 35=0 with expected t112, satisfies our outgoing test request
                session->set_expecting_response_for_outgoing_test_request(false);
                LLFIX_LOG_DEBUG("FixServer " + m_name + " session " + session->get_name() + " : other end satisfied the test request");
            }
            if (session->validations_enabled())
            {
                uint32_t reject_message_code = static_cast<uint32_t>(-1);
 
                if (session->validate_fix_message(*session->get_incoming_fix_message(), buffer_message, buffer_message_length, parser_reject_code, reject_message_code) == false)
                {
                    if (reject_message_code != static_cast<uint32_t>(-1))
                    {
                        send_reject_message(session, session->get_incoming_fix_message(), reject_message_code, buffer_message, buffer_message_length, session->get_last_error_tag());
                    }
 
                    return;
                }
            }
            // SEQUENCE NO & MESSAGE PERSIST PLUGIN
            auto sequence_store = session->get_sequence_store();
            sequence_store->increment_incoming_seq_no();
            auto sequence_store_incoming_seq_no = sequence_store->get_incoming_seq_no();
 
            if (m_message_persist_plugin)
                m_message_persist_plugin->persist_incoming_message(session->get_name(), sequence_store_incoming_seq_no, buffer_message, buffer_message_length);
 
            // SEQUENCE NO CHECKS
            auto incoming_seq_no = session->get_incoming_fix_message()->get_tag_value_as<uint32_t>(FixConstants::TAG_MSG_SEQ_NUM);
 
            if (llfix_unlikely(incoming_seq_no > sequence_store_incoming_seq_no))
            {
                if(FixSession::is_a_hard_sequence_reset_message(*session->get_incoming_fix_message()) == false)
                {
                    if (session->get_state() != SessionState::PENDING_LOGON) // on_logon_request handles this case on its own
                    {
                        session->queue_outgoing_resend_request(sequence_store_incoming_seq_no, incoming_seq_no);
                        return;
                    }
                }
            }
            else if (llfix_unlikely(incoming_seq_no < sequence_store_incoming_seq_no))
            {
                if(FixSession::is_a_hard_sequence_reset_message(*session->get_incoming_fix_message()) == false)
                {
                    std::string logout_reason_text;
                    bool send_logout = false;
 
                    if (session->get_state() == SessionState::PENDING_LOGON)
                    {
                        send_logout = true;
                        logout_reason_text = "MsgSeqNum too low, expecting " + std::to_string(sequence_store_incoming_seq_no) + " but received " + std::to_string(incoming_seq_no);
                    }
 
                    sequence_store->set_incoming_seq_no(sequence_store_incoming_seq_no - 1);
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " : terminating session " + session->get_name() + " as the incoming sequence no (" + std::to_string(incoming_seq_no) + ") is lower than expected (" + std::to_string(sequence_store_incoming_seq_no) + ")");
                    this->process_connection_closure(peer_index, send_logout, logout_reason_text);
                    return;
                }
            }
 
            if (llfix_unlikely(session->get_state() == SessionState::IN_RETRANSMISSION_INITIATED_BY_SELF))
            {
                if (incoming_seq_no == session->get_outgoing_resend_request_end_no())
                {
                    LLFIX_LOG_DEBUG("FixServer " + m_name + " : other end satisfied the resend request");
                    session->set_state(llfix::SessionState::LOGGED_ON);
                }
            }
            if(process_incoming_throttling(session, session->get_incoming_fix_message()) == false)
            {
                return;
            }
 
            auto message_type = session->get_incoming_fix_message()->get_tag_value(FixConstants::TAG_MSG_TYPE);
 
            if (llfix_likely(message_type->length() == 1))
            {
                switch (message_type->data()[0])
                {
                    case FixConstants::MSG_TYPE_NEW_ORDER: on_new_order(session, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_ORDER_CANCEL: on_cancel_order(session, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_ORDER_CANCEL_REPLACE: on_replace_order(session, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_HEARTBEAT: on_client_heartbeat(session); break;
                    case FixConstants::MSG_TYPE_TEST_REQUEST: session->process_test_request_message(*session->get_incoming_fix_message()); on_client_test_request(session, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_RESEND_REQUEST: session->process_resend_request(*session->get_incoming_fix_message()); on_client_resend_request(session, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_REJECT: on_session_level_reject(session, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_BUSINESS_REJECT: on_application_level_reject(session, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_LOGON: process_logon_request(session, peer_index, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_LOGOUT: process_logout_request(session, peer_index, session->get_incoming_fix_message()); break;
                    case FixConstants::MSG_TYPE_SEQUENCE_RESET: if (session->process_incoming_sequence_reset_message(*session->get_incoming_fix_message()) == false) { send_reject_message(session, session->get_incoming_fix_message(), FixConstants::FIX_ERROR_CODE_VALUE_INCORRECT_FOR_TAG, buffer_message, buffer_message_length, FixConstants::TAG_NEW_SEQ_NO); }; break;
                        // Anything else
                    default: on_custom_message(session, session->get_incoming_fix_message()); break;
                }
            }
            else
            {
                on_custom_message(session, session->get_incoming_fix_message());
            }
        }
 
        bool validate_logon_request(FixSession* session, std::size_t peer_index, const IncomingFixMessage* message)
        {
            LLFIX_UNUSED(session);
            int peer_port{0};
            std::string peer_ip;
            this->get_peer_details(peer_index, peer_ip, peer_port);
 
            if (message->has_tag(FixConstants::TAG_ENCRYPT_METHOD) == false)
            {
                LLFIX_LOG_ERROR("Incoming logon message for server " + m_name + " from " + peer_ip + " does not have required t98(encryption method), compid : " + message->get_tag_value_as<std::string>(FixConstants::TAG_SENDER_COMP_ID));
                return false;
            }
 
            if (message->has_tag(FixConstants::TAG_HEART_BT_INT) == false)
            {
                LLFIX_LOG_ERROR("Incoming logon message for server " + m_name + " from " + peer_ip + " does not have required t108(heartbeat interval), compid : " + message->get_tag_value_as<std::string>(FixConstants::TAG_SENDER_COMP_ID));
                return false;
            }
 
            auto t108_string = message->get_tag_value_as<std::string>(FixConstants::TAG_HEART_BT_INT);
 
            if (t108_string.find('-') != std::string::npos)
            {
                LLFIX_LOG_ERROR("Incoming logon message for server " + m_name + " from " + peer_ip + " has invalid(negative) t108(heartbeat interval) value, compid : " + message->get_tag_value_as<std::string>(FixConstants::TAG_SENDER_COMP_ID));
                return false;
            }
 
            if (t108_string.find('.') != std::string::npos)
            {
                LLFIX_LOG_ERROR("Incoming logon message for server " + m_name + " from " + peer_ip + " has invalid(decimal points) t108(heartbeat interval) value, compid : " + message->get_tag_value_as<std::string>(FixConstants::TAG_SENDER_COMP_ID));
                return false;
            }
 
            if (message->get_tag_value_as<uint32_t>(FixConstants::TAG_HEART_BT_INT) == 0)
            {
                LLFIX_LOG_ERROR("Incoming logon message for server " + m_name + " from " + peer_ip + " has invalid(zero) t108(heartbeat interval) value, compid : " + message->get_tag_value_as<std::string>(FixConstants::TAG_SENDER_COMP_ID));
                return false;
            }
 
            return true;
        }
 
        void do_post_logon_sequence_number_check(FixSession* session)
        {
            if(session->get_state() == SessionState::LOGGED_ON)
            {
                // IF LOGON WAS WITH A SEQ NO HIGHER THAN EXPECTED , WE NEED TO SEND A RESEND REQUEST
                auto sequence_store_incoming_seq_no = session->get_sequence_store()->get_incoming_seq_no();
                auto incoming_seq_no = session->get_incoming_fix_message()->get_tag_value_as<uint32_t>(FixConstants::TAG_MSG_SEQ_NUM);
 
                if (llfix_unlikely(incoming_seq_no > sequence_store_incoming_seq_no))
                {
                    if(FixSession::is_a_hard_sequence_reset_message(*session->get_incoming_fix_message()) == false)
                    {
                        session->queue_outgoing_resend_request(sequence_store_incoming_seq_no, incoming_seq_no);
                    }
                }
 
                // LOW SEQ NOS ARE HANDLED IN process_incoming_fix_message
            }
        }
 
        void process_logout_request(FixSession* session, std::size_t peer_index, const IncomingFixMessage* message)
        {
            send_logout_message(session);
            session->set_state(SessionState::LOGGED_OUT);
            LLFIX_LOG_DEBUG(m_name + ", session logged out : " + session->get_name() + " , peer index : " + std::to_string(peer_index));
            on_logout_request(session, message);
        }
 
        FixServer(const FixServer& other) = delete;
        FixServer& operator= (const FixServer& other) = delete;
        FixServer(FixServer&& other) = delete;
        FixServer& operator=(FixServer&& other) = delete;
};
 
} // namespace