fix_client.h

// DISCLAIMER_PLACEHOLDER
#pragma once
 
#include "common.h"
 
#include <atomic>
#include <cassert>
#include <cstdint>
#include <cstddef>
#include <string>
#include <string_view>
#include <memory>
#include <vector>
#include <thread>
#include <chrono>
#include <new>
 
#include "core/compiler/hints_hot_code.h"
#include "core/compiler/hints_branch_predictor.h"
#include "core/compiler/unused.h"
 
#include "core/cpu/alignment_constants.h"
 
#include "core/os/vdso.h"
#include "core/os/thread_utilities.h"
#include "core/os/process_utilities.h"
 
#include "core/utilities/converters.h"
#include "core/utilities/logger.h"
#include "core/utilities/std_string_utilities.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 "incoming_fix_message.h"
#include "outgoing_fix_message.h"
#include "fix_session.h"
#include "fix_session_settings.h"
#include "fix_client_settings.h"
#include "fix_string.h"
#include "fix_string_view.h"
#include "fix_utilities.h"
#include "fix_parser_error_codes.h"
 
#include "core/utilities/tcp_connector_options.h"
 
#ifdef LLFIX_UNIT_TEST // VOLTRON_EXCLUDE
#include <cstdlib>
#endif // VOLTRON_EXCLUDE
 
namespace llfix
{
 
template<typename Transport>
class FixClient : public Transport, public ManagedInstance
{
    public:
 
        FixClient() = default;
 
        virtual ~FixClient()
        {
            if (is_threaded())
            {
                m_is_exiting.store(true);
                m_thread->join();
            }
        }
 
        [[nodiscard]] bool create(const std::string& client_name, const FixClientSettings& settings, const std::string& session_name, const FixSessionSettings& session_settings)
        {
            m_settings = settings;
            return create(client_name, session_name, session_settings);
        }
 
        [[nodiscard]] bool create(const std::string& client_config_file_path, const std::string& client_name, const std::string& session_config_file_path, const std::string& session_name)
        {
            if (m_settings.load_from_config_file(client_config_file_path, client_name) == false)
            {
                LLFIX_LOG_ERROR("Loading settings for client " + client_name + " failed : " + m_settings.config_load_error);
                return false;
            }
 
            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 create(client_name, session_name, session_settings);
        }
 
        std::string get_name() const override
        {
            return m_name;
        }
 
        bool is_instance_ha_primary() const override
        {
            return m_is_ha_primary;
        }
 
        FixSession* get_session(const std::string& session_name="") override
        {
            LLFIX_UNUSED(session_name);
            return &m_session;
        }
 
        void get_session_names(std::vector<std::string>& target) override
        {
            target.push_back(m_session.get_name());
        }
 
        void initialise_thread()
        {
            if (m_settings.cpu_core_id >= 0)
            {
                if (ThreadUtilities::pin_calling_thread_to_cpu_core(m_settings.cpu_core_id) == 0)
                {
                    LLFIX_LOG_INFO(m_name + " thread pinned to CPU core " + std::to_string(m_settings.cpu_core_id));
                }
                else
                {
                    LLFIX_LOG_ERROR(m_name + " thread pinning to CPU core " + std::to_string(m_settings.cpu_core_id) + " failed");
                }
            }
        }
 
        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
 
        [[nodiscard]] bool start()
        {
            try
            {
                m_thread.reset(new std::thread(&FixClient::thread_function, this));
                return true;
            }
            catch(...)
            {}
 
            LLFIX_LOG_ERROR(m_name + " creation failed during thread creation");
            return false;
        }
 
        [[nodiscard]] bool connect()
        {
            if(m_session.get_state() == SessionState::DISABLED)
            {
                LLFIX_LOG_DEBUG("Cannot allow connection attempt as session " + m_session.get_name() + " is disabled.");
                return false;
            }
 
            if (m_session.is_now_valid_session_datetime() == false)
            {
                LLFIX_LOG_DEBUG("Cannot allow connection attempt for session " + m_session.get_name() + " due to schedule settings.");
                return false;
            }
 
            m_session.set_state(SessionState::PENDING_CONNECTION);
 
            m_session.reset_flags();
 
            TCPConnectorOptions options;
 
            options.m_rx_buffer_capacity = m_settings.rx_buffer_capacity;
            options.m_receive_size = m_settings.receive_size;
            options.m_send_try_count = m_settings.send_try_count;
 
            options.m_async_io_timeout_nanoseconds = m_settings.async_io_timeout_nanoseconds;
            options.m_busy_poll_microseconds = m_settings.busy_poll_microseconds;
            options.m_spin_count = m_settings.spin_count;
 
            options.m_nic_interface_name = m_settings.nic_name.c_str();
            options.m_nic_interface_ip = m_settings.nic_address.c_str();
 
            options.m_socket_tx_size = m_settings.socket_tx_size;
            options.m_socket_rx_size = m_settings.socket_rx_size;
            options.m_nic_ringbuffer_rx_size = m_settings.nic_ringbuffer_rx_size;
            options.m_nic_ringbuffer_tx_size = m_settings.nic_ringbuffer_tx_size;
 
            options.m_disable_nagle = m_settings.disable_nagle;
            options.m_enable_quick_ack = m_settings.quick_ack;
 
            options.m_stack = m_settings.stack;
            options.m_connect_timeout_seconds = m_settings.connect_timeout_seconds;
 
            #ifdef LLFIX_ENABLE_OPENSSL
            options.m_use_ssl = m_settings.use_ssl;
            options.m_ssl_verify_peer = m_settings.ssl_verify_peer;
            options.m_ssl_ca_pem_file = m_settings.ssl_ca_pem_file;
            options.m_ssl_cert_pem_file = m_settings.ssl_certificate_pem_file;
            options.m_ssl_private_key_pem_file = m_settings.ssl_private_key_pem_file;
            options.m_ssl_private_key_password = m_settings.ssl_private_key_password;
            options.m_ssl_version = m_settings.ssl_version;
            options.m_ssl_cipher_suite = m_settings.ssl_cipher_suite;
            #endif
 
            Transport::set_params(options);
 
            bool success = Transport::connect(m_settings.primary_address.c_str(), m_settings.primary_port);
            m_connected_to_primary = success;
 
            if (success == false)
            {
                // TRY SECONDARY CONNNECTION IF SET
                if (m_settings.secondary_address.length() > 0 && m_settings.secondary_port > 0)
                {
                    LLFIX_LOG_DEBUG(m_name + " : Primary connection failed. Trying secondary connection...");
                    success = Transport::connect(m_settings.secondary_address.c_str(), m_settings.secondary_port);
                    m_connected_to_secondary = success;
                }
            }
 
            if (success == true)
            {
                m_session.set_last_received_message_timestamp_nanoseconds(VDSO::nanoseconds_monotonic()); // This is to avoid sending unnecessary test request
 
                LLFIX_LOG_INFO(m_name + " : TCP connection established on " + (m_connected_to_primary?"primary":"secondary") );
                m_session.set_state(SessionState::PENDING_LOGON);
                m_last_logon_attempt_timestamp = VDSO::nanoseconds_monotonic();
 
                on_connection();
 
                if (send_logon_request() == false)
                {
                    success = false;
                }
            }
            else
            {
                m_session.set_state(SessionState::DISCONNECTED);
            }
 
            return success;
        }
 
        void process()
        {
            auto session_state = m_session.get_state();
 
            if(session_state > SessionState::DISCONNECTED) // In case of BSD sockets, calls on select may hang on some Linux distros if not connected
                this->process_incoming_messages();
 
            process_admin_commands();
 
            if (is_state_live(session_state))
            {
                process_outgoing_resend_request_if_necessary(VDSO::nanoseconds_monotonic());
                process_outgoing_test_request_if_necessary(VDSO::nanoseconds_monotonic());
 
                respond_to_resend_request_if_necessary();
                respond_to_test_request_if_necessary();
 
                send_client_heartbeat_if_necessary(VDSO::nanoseconds_monotonic());
 
                process_schedule_validator();
            }
            else if(session_state == SessionState::PENDING_LOGON)
            {
                process_outgoing_logon_request(VDSO::nanoseconds_monotonic());
            }
        }
 
        void push_admin_command(const std::string& session_name, ModifyingAdminCommandType type, uint32_t arg = 0) override
        {
            LLFIX_UNUSED(session_name);
            auto admin_command = new (std::nothrow) ModifyingAdminCommand;
 
            if (llfix_likely(admin_command))
            {
                admin_command->type = type;
                admin_command->arg = arg;
 
                m_session.get_admin_commands()->push(admin_command);
            }
            else
            {
                LLFIX_LOG_ERROR("Failed to process setter admin command for FixClient " + m_name);
            }
        }
 
        void shutdown(bool graceful_shutdown = true)
        {
            if(is_threaded())
            {
                m_thread_grace_full_exit.store(graceful_shutdown);
                m_is_exiting.store(true);
                return;
            }
 
            do_shutdown(graceful_shutdown);
        }
 
        OutgoingFixMessage* outgoing_message_instance()
        {
            return m_session.get_outgoing_fix_message();
        }
 
        virtual bool send_outgoing_message(OutgoingFixMessage* message)
        {
            std::size_t encoded_length = 0;
            auto int_sequence_no = m_session.get_sequence_store()->get_outgoing_seq_no() + 1;
 
            message->encode(m_session.get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, int_sequence_no, encoded_length);
 
            return send_bytes<true>(m_session.get_tx_encode_buffer(), encoded_length);
        }
 
        std::string get_settings_as_string(const std::string& delimiter) override
        {
            return m_settings.to_string(delimiter);
        }
 
        FixClientSettings& get_settings() { return m_settings; }
 
        bool connected_to_primary() const { return m_connected_to_primary; }
 
        bool connected_to_secondary() const { return m_connected_to_secondary; }
 
        // Run method in case this class is maintaining the client thread
        virtual void run() {};
 
        // TCP connection
        virtual void on_connection() {};
 
        virtual void on_disconnection() {};
 
        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) override
        {
            LLFIX_UNUSED(error_code);
        }
        // Logon messages
        virtual void on_logon_response(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        virtual void on_logout_response(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        virtual void on_logon_reject(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        // Session/admin messages
        virtual void on_server_heartbeat() {};
        virtual void on_server_test_request(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        virtual void on_server_resend_request(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        // Application messages
        virtual void on_execution_report(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        virtual void on_order_cancel_replace_reject(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        virtual void on_session_level_reject(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        virtual void on_application_level_reject(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
        virtual void on_custom_message_type(const IncomingFixMessage* message) {LLFIX_UNUSED(message);};
 
    public:
 
        void on_connection_lost() override
        {
            m_session.set_state(SessionState::DISCONNECTED);
            m_connected_to_primary = false;
            m_connected_to_secondary = false;
            on_disconnection();
        }
 
        void on_data_ready() override
        {
            std::size_t read = this->receive();
 
            if (read > 0 && read <= m_settings.rx_buffer_capacity)
            {
                process_rx_buffer(this->get_rx_buffer(), this->get_rx_buffer_size());
            }
 
            this->receive_done();
        }
 
        void set_message_persist_plugin(MessagePersistPlugin* plugin)
        {
            assert(plugin);
            m_message_persist_plugin = plugin;
        }
 
        #ifdef LLFIX_AUTOMATION
        // Used for testing outgoing msg resends  and gap fills
        bool send_fake_outgoing_message(OutgoingFixMessage* message)
        {
            auto int_sequence_no = m_session.get_sequence_store()->get_outgoing_seq_no() + 1;
            std::size_t encoded_length = 0;
 
            message->encode(m_session.get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, int_sequence_no, encoded_length);
 
            auto sequence_store = m_session.get_sequence_store();
            sequence_store->increment_outgoing_seq_no();
 
            if(m_session.serialisation_enabled())
                m_session.get_outgoing_message_serialiser()->write(reinterpret_cast<const void*>(m_session.get_tx_encode_buffer()), encoded_length, true, sequence_store->get_outgoing_seq_no());
 
            m_session.set_last_sent_message_timestamp_nanoseconds(VDSO::nanoseconds_monotonic());
 
            return true;
        }
        #endif
 
    protected:
        // ADMIN MESSAGES TO SERVER
        virtual bool send_logon_request()
        {
            auto logon_request = outgoing_message_instance();
            logon_request->set_msg_type(FixConstants::MSG_TYPE_LOGON);
 
            // TAG 98 ENCRYPTION METHOD
            logon_request->set_tag(FixConstants::TAG_ENCRYPT_METHOD, 0);
 
            // TAG 108 HEARTBEAT INTERVAL
            logon_request->set_tag(FixConstants::TAG_HEART_BT_INT, static_cast<uint32_t>(m_session.get_heartbeart_interval_in_nanoseconds() / 1'000'000'000));
 
            // TAG 1137 DEFAULT APP VER ID
            auto default_app_ver_id = m_session.get_default_app_ver_id();
 
            if (default_app_ver_id.length() > 0)
            {
                logon_request->set_tag(FixConstants::TAG_DEFAULT_APPL_VER_ID, default_app_ver_id);
            }
 
            // TAG 141 RESET SEQ NOS
            if(m_session.logon_reset_sequence_numbers_flag())
            {
                m_session.get_sequence_store()->reset_numbers();
                logon_request->set_tag(FixConstants::TAG_RESET_SEQ_NUM_FLAG, FixConstants::FIX_BOOLEAN_TRUE);
            }
 
            // TAG 789 NEXT EXPECTED SEQ NO
            if(m_session.logon_include_next_expected_seq_no())
            {
                const auto next_expected_seq_no = m_session.get_sequence_store()->get_incoming_seq_no()+1;
                logon_request->set_tag(FixConstants::TAG_NEXT_EXPECTED_SEQ_NUM, next_expected_seq_no);
            }
 
            // TAG 553 SESSION USER NAME
            auto session_username = m_session.get_username();
 
            if (session_username.length() > 0)
            {
                logon_request->set_tag(FixConstants::TAG_USERNAME, session_username);
            }
 
            // TAG 554 SESSION PASSWORD
            auto session_password = m_session.get_password();
 
            if (session_password.length() > 0)
            {
                logon_request->set_tag(FixConstants::TAG_PASSWORD, session_password);
            }
 
            // TAG 925 NEW PASSWORD
            if (m_session.logon_message_new_password().length() > 0)
            {
                logon_request->set_tag(FixConstants::TAG_NEW_PASSWORD, m_session.logon_message_new_password());
            }
 
            return send_outgoing_message(logon_request);
        }
 
        virtual bool send_logout_request()
        {
            m_session.build_logout_message(outgoing_message_instance());
            return send_outgoing_message(outgoing_message_instance());
        }
 
        virtual bool send_client_heartbeat(FixString* test_request_id)
        {
            m_session.build_heartbeat_message(outgoing_message_instance(), test_request_id);
            return send_outgoing_message(outgoing_message_instance());
        }
 
        virtual bool send_test_request()
        {
            m_session.build_test_request_message(outgoing_message_instance());
            return send_outgoing_message(outgoing_message_instance());
        }
 
        virtual bool send_resend_request()
        {
            m_session.build_resend_request_message(outgoing_message_instance(), "0");
            return send_outgoing_message(outgoing_message_instance());
        }
 
        virtual bool send_sequence_reset_message(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();
            m_session.build_sequence_reset_message(message, desired_sequence_no);
 
            // ENCODE , NO NEED TO INCREMENT OUTGOING SEQ NO AS WE ARE RESETTING WITH THIS MESSAGE
            std::size_t encoded_length = 0;
            message->encode(m_session.get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, desired_sequence_no, encoded_length);
 
            // SEND
            return send_bytes<false>(m_session.get_tx_encode_buffer(), encoded_length); // false is for not incrementing seq no for this message
        }
 
        virtual bool send_gap_fill_message()
        {
            /*
                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();
            m_session.build_gap_fill_message(message);
 
            // ENCODE , NO NEED TO INCREMENT OUTGOING SEQ NO AS WE ARE RESETTING WITH THIS MESSAGE
            std::size_t encoded_length = 0;
            message->encode(m_session.get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, m_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>(m_session.get_tx_encode_buffer(), encoded_length); // false is for not incrementing seq no for this message
        }
 
        virtual void resend_messages_to_server(uint32_t begin_seq_no, uint32_t end_seq_no)
        {
            for (uint32_t i = begin_seq_no; i <= end_seq_no; i++)
            {
                std::size_t message_length = 0;
                m_session.get_outgoing_message_serialiser()->read_message(i, m_session.get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, message_length);
 
                auto message = outgoing_message_instance();
                message->load_from_buffer(m_session.get_tx_encode_buffer(), message_length); // t122(orig sending time) will be handled by load_from_buffer
 
                message->template set_tag<FixMessageComponent::HEADER, bool>(FixConstants::TAG_POSS_DUP_FLAG, FixConstants::FIX_BOOLEAN_TRUE);
 
                if (m_session.include_t97_during_resends())
                {
                    message->template set_tag<FixMessageComponent::HEADER, bool>(FixConstants::TAG_POSS_RESEND, FixConstants::FIX_BOOLEAN_TRUE);
                }
 
                std::size_t encoded_length = 0;
                message->encode(m_session.get_tx_encode_buffer(), m_settings.tx_encode_buffer_capacity, i, encoded_length);
                send_bytes<false>(m_session.get_tx_encode_buffer(), encoded_length); // false is for not incrementing seq no for this message
            }
        }
 
        virtual bool send_reject_message(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("FixClient " + m_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("FixClient " + m_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));
            }
 
            m_session.build_session_level_reject_message(outgoing_message_instance(), reject_reason_code, reject_reason_text, error_tag);
            return send_outgoing_message(outgoing_message_instance());
        }
 
        std::atomic<bool> m_is_exiting = false;
 
    private:
        LLFIX_ALIGN_DATA(AlignmentConstants::CPU_CACHE_LINE_SIZE) FixSession m_session;
 
        std::string m_name;
        bool m_is_ha_primary = true;
 
        static inline constexpr std::size_t MINIMUM_REQUIRED_INITIAL_BUFFER_FOR_PARSING = 4; // 10=<DELIMITER>
 
        uint64_t m_last_logon_attempt_timestamp = 0;
 
        bool m_connected_to_primary = false;
        bool m_connected_to_secondary = false;
 
        FixClientSettings m_settings;
 
        MessagePersistPlugin* m_message_persist_plugin = nullptr;
        std::atomic<bool> m_thread_grace_full_exit = true;
        std::unique_ptr<std::thread> m_thread;
 
        bool is_threaded() const { return m_thread.get() != nullptr;}
 
        void thread_function()
        {
            initialise_thread();
 
            while (true)
            {
                if (m_is_exiting.load() == true)
                {
                    do_shutdown(m_thread_grace_full_exit.load());
                    break;
                }
 
                run();
            }
        }
 
        bool create(const std::string& name, const std::string& session_name, const FixSessionSettings& session_settings)
        {
            m_name = name;
            m_is_ha_primary = m_settings.starts_as_primary_instance;
 
            if (m_settings.validate() == false)
            {
                LLFIX_LOG_ERROR("FixClientSettings for " + m_name + " validation failed : " + m_settings.validation_error);
                return false;
            }
 
            session_settings.is_server = false;
            session_settings.tx_encode_buffer_capacity = m_settings.tx_encode_buffer_capacity; // Propagate
 
            if (m_session.initialise(session_name, session_settings) == false)
            {
                LLFIX_LOG_ERROR(m_name + " creation failed during session initialisation");
                return false;
            }
 
            if (m_settings.starts_as_primary_instance == false)
            {
                disable_session();
            }
 
            LLFIX_LOG_INFO(m_name + " : Loaded client config =>\n" + m_settings.to_string());
 
            LLFIX_LOG_INFO("FixClient " + m_name + " creation success");
 
            return true;
        }
 
        void process_admin_commands()
        {
            ModifyingAdminCommand* admin_command{ nullptr };
 
            if (m_session.get_admin_commands()->try_pop(&admin_command) == true)
            {
                switch (admin_command->type)
                {
                    case ModifyingAdminCommandType::SET_INCOMING_SEQUENCE_NUMBER:
                    {
                        m_session.get_sequence_store()->set_incoming_seq_no(admin_command->arg);
                        LLFIX_LOG_DEBUG(m_name + " : processed set incoming sequence number admin command , new seq no : " + std::to_string(admin_command->arg));
                        break;
                    }
 
                    case ModifyingAdminCommandType::SET_OUTGOING_SEQUENCE_NUMBER:
                    {
                        m_session.get_sequence_store()->set_outgoing_seq_no(admin_command->arg);
                        LLFIX_LOG_DEBUG(m_name + " : processed set outgoing sequence number admin command , new seq no : " + std::to_string(admin_command->arg));
                        break;
                    }
 
                    case ModifyingAdminCommandType::SEND_SEQUENCE_RESET:
                    {
                        if( m_session.get_state() == SessionState::LOGGED_ON )
                        {
                            send_sequence_reset_message(admin_command->arg);
                            LLFIX_LOG_DEBUG(m_name + " : processed send sequence reset admin command , 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");
                        }
 
                        break;
                    }
 
                    case ModifyingAdminCommandType::DISABLE_SESSION:
                    {
                        disable_session();
                        LLFIX_LOG_DEBUG(m_name + " : processed disable session admin command , new session state : Disabled");
                        break;
                    }
 
                    case ModifyingAdminCommandType::ENABLE_SESSION:
                    {
                        if(m_is_ha_primary)
                        {
                            enable_session();
                            LLFIX_LOG_DEBUG(m_name + " : processed enable session admin command");
                        }
                        else
                        {
                            LLFIX_LOG_ERROR(m_name + " ignoring enable session admin command 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();
                            LLFIX_LOG_DEBUG(m_name + " : processed set is ha primary instance 1 admin command");
                        }
                        else
                        {
                            disable_session();
                            LLFIX_LOG_DEBUG(m_name + " : processed set is ha primary instance 0 admin command");
                        }
 
                        break;
                    }
 
                    default: break;
                }
 
                delete admin_command;
            }
        }
 
        void disable_session()
        {
            if(is_state_live(m_session.get_state()))
            {
                do_shutdown(true);
            }
            m_session.set_state(SessionState::DISABLED);
        }
 
        void enable_session()
        {
            if(m_session.get_state() == SessionState::DISABLED)
            {
                m_session.set_state(SessionState::DISCONNECTED);
                if(m_settings.refresh_resend_cache_during_promotion)
                    m_session.reinitialise_outgoing_serialiser();
            }
        }
 
        void process_schedule_validator()
        {
            if (m_session.is_now_valid_session_datetime() == false)
            {
                LLFIX_LOG_DEBUG("FixClient " + m_name + " : terminating session due to schedule settings");
                do_shutdown(true);
            }
        }
 
        void do_shutdown(bool graceful_shutdown = true)
        {
            auto call_time_state = m_session.get_state();
            LLFIX_LOG_DEBUG(m_name + " : shutdown called. Graceful shutdown param : " + (graceful_shutdown?"true":"false") );
 
            on_connection_lost();
 
            if (graceful_shutdown)
            {
                if (call_time_state == SessionState::LOGGED_ON)
                {
                    m_session.set_received_logout_response(false);
 
                    send_logout_request();
                    m_session.set_state(SessionState::PENDING_LOGOUT);
 
                    auto start = std::chrono::steady_clock::now();
                    const auto timeout = std::chrono::milliseconds(m_session.settings()->logout_timeout_seconds*1000);
 
                    while(true)
                    {
                        this->process_incoming_messages();
 
                        if(m_session.received_logout_response() == true)
                        {
                            break;
                        }
 
                        if (m_session.get_state() == SessionState::DISCONNECTED)
                        {
                            break;
                        }
 
                        auto now = std::chrono::steady_clock::now();
                        if (now - start >= timeout)
                        {
                            LLFIX_LOG_DEBUG("FixClient " + m_name + " shutdown timed out before receiving logout response");
                            break;
                        }
                    }
                }
            }
 
            this->close();
            m_session.set_state(SessionState::DISCONNECTED);
        }
 
        // TO SERVER
        template <bool increment_outgoing_seq_no>
        LLFIX_FORCE_INLINE bool send_bytes(const char* buffer, std::size_t buffer_size)
        {
            auto sequence_store = m_session.get_sequence_store();
 
            if(llfix_likely(m_session.settings()->throttle_limit != 0))
            {
                m_session.throttler()->update();
                m_session.throttler()->wait();
            }
 
            auto ret = this->send(buffer, buffer_size);
 
            if constexpr (increment_outgoing_seq_no == true)
            {
                if(ret == true)
                    sequence_store->increment_outgoing_seq_no();
            }
 
            if(m_session.serialisation_enabled())
                m_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(m_session.get_name(), sequence_store->get_outgoing_seq_no(), buffer, buffer_size, ret);
 
            m_session.set_last_sent_message_timestamp_nanoseconds(VDSO::nanoseconds_monotonic());
 
            return ret;
        }
 
        // ADMIN UTILITY TO SERVER
        void process_outgoing_logon_request(uint64_t current_timestamp_nanoseconds)
        {
            auto delta_nanoseconds = current_timestamp_nanoseconds - m_last_logon_attempt_timestamp;
 
            if (delta_nanoseconds >= (static_cast<uint64_t>(m_session.settings()->logon_timeout_seconds) * 1'000'000'000))
            {
                m_session.set_state(SessionState::DISCONNECTED);
            }
        }
 
        void send_client_heartbeat_if_necessary(uint64_t current_timestamp_nanoseconds)
        {
            auto delta_nanoseconds = current_timestamp_nanoseconds - m_session.last_sent_message_timestamp_nanoseconds();
 
            if (delta_nanoseconds >= (m_session.get_heartbeart_interval_in_nanoseconds() * static_cast<uint64_t>(9) / static_cast<uint64_t>(10))) // 0.9 for safety
            {
                send_client_heartbeat(nullptr);
            }
        }
 
        void process_outgoing_test_request_if_necessary(uint64_t current_timestamp_nanoseconds)
        {
            if (m_session.expecting_response_for_outgoing_test_request() == false)
            {
                // SEND IF NECESSARY
                auto delta_nanoseconds = current_timestamp_nanoseconds - m_session.last_received_message_timestamp_nanoseconds();
 
                if (delta_nanoseconds >= (m_session.get_outgoing_test_request_interval_in_nanoseconds()))
                {
                    LLFIX_LOG_DEBUG("FixClient " + m_name + " : sending test request");
                    send_test_request();
                    m_session.set_outgoing_test_request_timestamp_nanoseconds(current_timestamp_nanoseconds);
                    m_session.set_expecting_response_for_outgoing_test_request(true);
                }
            }
            else
            {
                // TERMINATE SESSION IF NECESSARY
                auto delta_nanoseconds = current_timestamp_nanoseconds - m_session.outgoing_test_request_timestamp_nanoseconds();
 
                if (delta_nanoseconds >= (m_session.get_heartbeart_interval_in_nanoseconds() * static_cast<uint64_t>(2)))
                {
                    m_session.set_expecting_response_for_outgoing_test_request(false);
                    LLFIX_LOG_DEBUG("FixClient " + m_name + " : terminating session as the other end didn't respond to 35=1/test request");
                    do_shutdown(false);
                }
            }
        }
 
        void process_outgoing_resend_request_if_necessary(uint64_t current_timestamp_nanoseconds)
        {
            if (m_session.get_state() == SessionState::IN_RETRANSMISSION_INITIATED_BY_SELF)
            {
                // TERMINATE SESSION IF NECESSARY
                uint64_t delta_nanoseconds = current_timestamp_nanoseconds - m_session.outgoing_resend_request_timestamp_nanoseconds();
                uint64_t required_delta_nanoseconds = static_cast<uint64_t>(m_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("FixClient " + m_name + " : terminating session as the other end didn't respond to 35=2/resend request in pre-configured timeout (outgoing_resend_request_expire_secs) : " + std::to_string(m_session.outgoing_resend_request_expire_secs()));
                    do_shutdown(false);
                }
            }
            else
            {
                // SEND IF NECESSARY
                if (m_session.needs_to_send_resend_request())
                {
                    LLFIX_LOG_DEBUG("FixClient " + m_name + " : sending resend request , begin no : " + std::to_string(m_session.get_outgoing_resend_request_begin_no()));
                    send_resend_request();
                    m_session.set_outgoing_resend_request_timestamp_nanoseconds(current_timestamp_nanoseconds);
                    m_session.set_state(llfix::SessionState::IN_RETRANSMISSION_INITIATED_BY_SELF);
                    m_session.set_needs_to_send_resend_request(false);
                }
            }
        }
 
        void respond_to_test_request_if_necessary()
        {
            if (m_session.needs_responding_to_incoming_test_request())
            {
                if (m_session.get_incoming_test_request_id()->length() > 0)
                {
                    send_client_heartbeat(m_session.get_incoming_test_request_id());
                    m_session.get_incoming_test_request_id()->set_length(0);
                }
                else
                {
                    // We should not hit here , but better than risking disconnection
                    send_client_heartbeat(nullptr);
                }
 
                LLFIX_LOG_DEBUG("FixClient " + m_name + " : responded to incoming test request");
                m_session.set_needs_responding_to_incoming_test_request(false);
            }
        }
 
        void respond_to_resend_request_if_necessary()
        {
            if (m_session.needs_responding_to_incoming_resend_request())
            {
                const auto last_outgoing_seq_no = m_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 = (m_session.serialisation_enabled()) && (m_session.replay_messages_on_incoming_resend_request() && (m_session.get_incoming_resend_request_end_no() == 0
                                   || m_session.get_incoming_resend_request_end_no() == last_outgoing_seq_no) && m_session.get_incoming_resend_request_begin_no()<=last_outgoing_seq_no);
 
                if(last_outgoing_seq_no > m_session.get_incoming_resend_request_begin_no())
                    if(last_outgoing_seq_no-m_session.get_incoming_resend_request_begin_no()+1 > m_session.max_resend_range())
                        will_replay = false;
 
                if(will_replay)
                {
                    for (uint32_t i = m_session.get_incoming_resend_request_begin_no(); i <= last_outgoing_seq_no; i++)
                    {
                        if (m_session.get_outgoing_message_serialiser()->has_message_in_memory(i) == false)
                        {
                            will_replay = false; // We don't have all the messages
                            break;
                        }
                    }
                }
 
                if(will_replay)
                {
                    LLFIX_LOG_DEBUG("FixClient " + m_name + " : replaying messages , begin seq no : " + std::to_string(m_session.get_incoming_resend_request_begin_no()) + " , end seq no : " + std::to_string(last_outgoing_seq_no));
                    resend_messages_to_server(m_session.get_incoming_resend_request_begin_no(), last_outgoing_seq_no);
                }
                else
                {
                    LLFIX_LOG_DEBUG("FixClient " + m_name + " : sending gap fill message");
                    send_gap_fill_message();
                }
 
                m_session.set_needs_responding_to_incoming_resend_request(false);
                m_session.set_state(llfix::SessionState::LOGGED_ON);
            }
        }
 
        // FROM SERVER
        #if defined(LLFIX_BENCHMARK) || defined(LLFIX_UNIT_TEST)
        public:
        #else
        private:
        #endif
        LLFIX_HOT void process_rx_buffer(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(buffer, buffer_size);
                return;
            }
            // 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(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(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;
                }
                m_session.get_incoming_fix_message()->reset();
                m_session.get_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 = m_session.get_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(m_session.get_incoming_fix_message()->has_tag(tag) == false))
                                    {
                                        m_session.get_incoming_fix_message()->set_tag(tag, value);
                                    }
                                    else
                                    {
                                        parser_reject_code = FixConstants::FIX_ERROR_CODE_TAG_APPEARS_MORE_THAN_ONCE;
                                    }
                                }
                                else
                                {
                                    m_session.get_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(m_session.get_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(buffer + buffer_read_index, buffer_size - buffer_read_index);
                        return;
                    }
 
                    current_index++;
                }// while true
                auto current_message_length = current_tag_10_delimiter_index+1-buffer_read_index;
                process_incoming_fix_message(m_session.get_incoming_fix_message(), buffer+ buffer_read_index, current_message_length, parser_reject_code);
 
                #ifndef LLFIX_UNIT_TEST
                if(llfix_unlikely(m_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();
                    return;
                }
                else if (static_cast<int>(buffer_read_index)> final_tag10_delimiter_index)
                {
                    #ifdef LLFIX_UNIT_TEST
                    this->m_incomplete_buffer = static_cast<char*>(malloc(65536));
                    #endif
                    this->set_incomplete_buffer(buffer + buffer_read_index, buffer_size - buffer_read_index);
                    return;
                }
            }
        }
 
        void process_incoming_fix_message(IncomingFixMessage* incoming_message, const char* buffer_message, std::size_t buffer_message_length, uint32_t parser_reject_code)
        {
            m_session.set_last_received_message_timestamp_nanoseconds(VDSO::nanoseconds_monotonic());
 
            if(m_session.serialisation_enabled())
                m_session.get_incoming_message_serialiser()->write(buffer_message, buffer_message_length, true);
 
            if(llfix_unlikely(m_session.expecting_response_for_outgoing_test_request()))
            {
                // We are permissive , any message not just 35=0 with expected t112, satisfies our outgoing test request
                m_session.set_expecting_response_for_outgoing_test_request(false);
                LLFIX_LOG_DEBUG(m_name + " : other end satisfied the test request");
            }
            if(m_session.validations_enabled())
            {
                uint32_t reject_message_code = static_cast<uint32_t>(-1);
 
                if (m_session.validate_fix_message(*m_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(*m_session.get_incoming_fix_message(), reject_message_code, buffer_message, buffer_message_length, m_session.get_last_error_tag());
                    }
 
                    return;
                }
            }
            // SEQUENCE NO & MESSAGE PERSIST PLUGIN
            auto sequence_store = m_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(m_session.get_name(), sequence_store_incoming_seq_no, buffer_message, buffer_message_length);
 
            // SEQUENCE NO CHECKS
            auto incoming_seq_no = incoming_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(*incoming_message) == false)
                {
                    m_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(*incoming_message) == false)
                {
                    sequence_store->set_incoming_seq_no(sequence_store_incoming_seq_no-1);
                    LLFIX_LOG_DEBUG("FixClient " + m_name + " : terminating session as the incoming sequence no (" + std::to_string(incoming_seq_no) + ") is lower than expected ("+ std::to_string(sequence_store_incoming_seq_no) + ")");
                    do_shutdown(false);
                    return;
                }
            }
 
            if(llfix_unlikely(m_session.get_state() == SessionState::IN_RETRANSMISSION_INITIATED_BY_SELF))
            {
                if(incoming_seq_no == m_session.get_outgoing_resend_request_end_no())
                {
                    m_session.set_state(llfix::SessionState::LOGGED_ON);
                    LLFIX_LOG_DEBUG(m_name + " : other end satisfied the resend request");
                }
            }
            auto message_type = incoming_message->get_tag_value(FixConstants::TAG_MSG_TYPE);
 
            if (llfix_likely(message_type->length() == 1))
            {
                switch (message_type->data()[0])
                {
                    case FixConstants::MSG_TYPE_EXECUTION_REPORT: on_execution_report(incoming_message); break;
                    case FixConstants::MSG_TYPE_HEARTBEAT: on_server_heartbeat(); break;
                    case FixConstants::MSG_TYPE_TEST_REQUEST: m_session.process_test_request_message(*incoming_message); on_server_test_request(incoming_message); break;
                    case FixConstants::MSG_TYPE_RESEND_REQUEST: m_session.process_resend_request(*incoming_message); on_server_resend_request(incoming_message); break;
                    case FixConstants::MSG_TYPE_ORDER_CANCEL_REJECT: on_order_cancel_replace_reject(incoming_message); break;
                    case FixConstants::MSG_TYPE_REJECT: process_session_level_reject(incoming_message); break;
                    case FixConstants::MSG_TYPE_BUSINESS_REJECT: process_application_level_reject(incoming_message); break;
                    case FixConstants::MSG_TYPE_LOGON: process_logon_response(incoming_message); break;
                    case FixConstants::MSG_TYPE_LOGOUT: process_logout_response(incoming_message); break;
                    case FixConstants::MSG_TYPE_SEQUENCE_RESET: if (m_session.process_incoming_sequence_reset_message(*incoming_message) == false) { send_reject_message(*incoming_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_type(incoming_message); break;
                }
            }
            else
            {
                on_custom_message_type(incoming_message);
            }
        }
 
        void process_session_level_reject(const IncomingFixMessage* message)
        {
            if (m_session.get_state() == SessionState::PENDING_LOGON)
            {
                process_logon_reject(message);
            }
            else
            {
                on_session_level_reject(message);
            }
        }
 
        void process_application_level_reject(const IncomingFixMessage* message)
        {
            if (m_session.get_state() == SessionState::PENDING_LOGON)
            {
                process_logon_reject(message);
            }
            else
            {
                on_application_level_reject(message);
            }
        }
 
        void process_logon_response(const IncomingFixMessage* logon_response)
        {
            m_session.set_state(SessionState::LOGGED_ON);
            this->on_logon_response(logon_response);
        }
 
        void process_logout_response(const IncomingFixMessage* logout_response)
        {
            auto state = m_session.get_state();
 
            if (state == SessionState::PENDING_LOGON)
            {
                process_logon_reject(logout_response);
            }
            else
            {
                m_session.set_received_logout_response(true);
                m_session.set_state(SessionState::LOGGED_OUT);
                this->on_logout_response(logout_response);
            }
        }
 
        void process_logon_reject(const IncomingFixMessage* logon_reject)
        {
            m_session.set_state(SessionState::LOGON_REJECTED);
 
            if (logon_reject->has_tag(FixConstants::TAG_TEXT))
            {
                [[maybe_unused]] auto reject_message = logon_reject->get_tag_value(FixConstants::TAG_TEXT);
                LLFIX_LOG_DEBUG("FixClient " + m_name + " logon message rejected : " + reject_message->to_string());
            }
            else
            {
                LLFIX_LOG_DEBUG("FixClient " + m_name + " logon message rejected with no reason text/tag58.");
            }
 
            on_logon_reject(logon_reject);
        }
 
        FixClient(const FixClient& other) = delete;
        FixClient& operator= (const FixClient& other) = delete;
        FixClient(FixClient&& other) = delete;
        FixClient& operator=(FixClient&& other) = delete;
};
 
} // namespace