/*
        Copyright (c) 2025, MOEX. All rights reserved.
        
        Plaza-2 Client Gate API Usage Sample.
        
        All the software and documentation included in this and any
        other MOEX CGate Release is copyrighted by MOEX.
        
        Redistribution and use in source and binary forms, with or without
        modification, are permitted only by the terms of a valid
        software license agreement with MOEX.
        
        THIS SOFTWARE IS PROVIDED "AS IS" AND MOEX DISCLAIMS ALL WARRANTIES
        EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION, ANY IMPLIED WARRANTIES OF
        NON-INFRINGEMENT, MERCHANTABILITY OR FITNESS FOR A PARTICULAR
        PURPOSE.  MOEX DOES NOT WARRANT THAT USE OF THE SOFTWARE WILL BE
        UNINTERRUPTED OR ERROR-FREE.  MOEX SHALL NOT, UNDER ANY CIRCUMSTANCES, BE
        LIABLE TO LICENSEE FOR LOST PROFITS, CONSEQUENTIAL, INCIDENTAL, SPECIAL OR
        INDIRECT DAMAGES ARISING OUT OF OR RELATED TO THIS AGREEMENT OR THE
        TRANSACTIONS CONTEMPLATED HEREUNDER, EVEN IF MOEX HAS BEEN APPRISED OF
        THE LIKELIHOOD OF SUCH DAMAGES.
*/

/*
    Настоящий пример демонстрирует алгоритм расчета статусов инструментов по потоку FORTS_SECURITYGROUPSTATE_REPL.

    This example demonstrates the algorithm for calculating the status of instruments using the FORTS_SECURITYGROUPSTATE_REPL stream.
*/

#include "instruments_state_other.h"
#include "instruments_state_schemes.h"

class DisconnectException : public std::exception {};

// Типы обрабатываемых синхрособытий

// Types of processed synchronization events
enum SysEvents
{
    DataReady = 1,
    ClearingStarted = 5,
};

// Состояния группы инструментов

// Instrument Group States
enum SecurityGroupState
{
    TRADING_MODE_ALL_ALLOWED = 0, // разрешено все (everything is allowed)
    TRADING_MODE_ACCEPT_ORDERS = 8, // сбор заявок в АО (collection of orders for the opening auction)
    TRADING_MODE_REJECT_INCREASE_POSITION = 16, // запрещено увеличение позиции (position increase prohibited)
    TRADING_MODE_REJECT_EVERYTHING_EXCEPT_CANCEL_ORDER = 2, // запрещено все кроме снятия заявки (everything is prohibited except for cancel order)
    TRADING_MODE_REJECT_EVERYTHING = 4 // запрещено всё (everything is prohibited)
};

// Тип хальта в группе инструментов

// Halt type in the tool group
enum HaltType
{
    EmptyHalts = 0, // Empty all halts on client
    Session = 1, // Session
    Section = 2, // Section
    SectionAndTradeMode = 3, // SectionAndTradeMode
    FutBaseContract = 4, // FutBaseContract
    InstrumentsFromTradeMode = 5, // InstrumentsFromTradeMode
    InstrumentsFromTradeModeAndFutBaseContract = 6, // InstrumentsFromTradeModeAndFutBaseContract
    GroupInstruments = 7, // GroupInstruments
    GroupInstrumentsAndFutBaseContract = 8, // GroupInstrumentsAndFutBaseContract
};

// Инструмент, его состояние и принадлежность к базовому контракту, секции, режиму торгов и группе инструментов
// заполняется по справочникам из FORTS_REFDATA_REPL. Статус считается по FORTS_SECURITYGROUPSTATE_REPL.

// Instrument, its status and belonging to the underlying contract, section, trading mode and instrument group
// is filled in according to the reference books from FORTS_REFDATA_REPL. The status is calculated according to FORTS_SECURITYGROUPSTATE_REPL.
struct InstrumentState {
    int32_t isin_id;
    int32_t state;
    
    int32_t section_id;
    int32_t base_contract_id;
    int32_t trade_mode_id;
    int64_t group_mask;
    std::unordered_map<int64_t, SecurityGroupState> security_group_state;
};


// основной объект - приложение Instruments state
// реализует логику обработки сообщений
//
// main instruments state application
// implements message callback and other message handling functions
class InstrumentsStateApp
{
    cg_conn_t* _connection;
    cg_listener_t *_listener_refdata;
    cg_listener_t *_listener_scrgrpst;
    // файл куда будут выводиться статусы инструментов
    // 
    // file where the statuses of the tools will be output
    std::string _outfile;

    int64_t _last_revision;

    // данные получаемые по реплике FORTS_REFDATA_REPL
    // 
    // data received from the FORTS_REFDATA_REPL replica
    std::unordered_map<int64_t, FutSessContentsRecord> _fut_sess_contents;
    std::unordered_map<int64_t, OptSessContentsRecord> _opt_sess_contents;
    std::unordered_map<int64_t, FutVcbRecord> _fut_vcb;
    std::unordered_map<int64_t, OptVcbRecord> _opt_vcb;
    std::unordered_map<int64_t, SysEventsRecord> _sys_events;

    // данные получаемые по реплике FORTS_SECURITYGROUPSTATE_REPL

    // data received from the FORTS_SECURITYGROUPSTATE_REPL replica
    std::unordered_map<int64_t, SysEventsRecord> _scty_events;
    std::unordered_map<int64_t, SecurityGroupStateRecord> _scty_grps;

    // список инструментов и их статусов

    // list of tools and their statuses
    std::list<InstrumentState> _instruments_state;
    // связка базовый контракт строка - числовой код базовый контракт  + числокой код секция базового контракта 
    
    // bundle base contract string - numeric code base contract + numeric code section of base contract
    using VcbValueType = std::pair<int32_t, int32_t>;
    std::unordered_map<std::string, VcbValueType> _vcb_map;

    // признак получен ли по потоку FORTS_REFDATA_REPL синхрособытие session_data_ready

    // flag whether the FORTS_REFDATA_REPL synchronization event session_data_ready was received via the FORTS_REFDATA_REPL stream
    bool _is_refdata_ready;
    // признак вышел ли поток FORTS_REFDATA_REPL в online
    // 
    // flag whether the FORTS_REFDATA_REPL thread has gone online
    bool _is_refdata_online;

    // признак получен ли по потоку FORTS_SECURITYGROUPSTATE_REPL синхрособытие session_data_ready
    // flag whether the FORTS_SECURITYGROUPSTATE_REPL synchronization event session_data_ready was received via the stream
    bool _is_security_group_state_ready;
    // признак вышел ли поток FORTS_SECURITYGROUPSTATE_REPL в online
    // flag whether the FORTS_SECURITYGROUPSTATE_REPL thread has gone online
    bool _is_security_group_state_online;

    std::unique_ptr<Scheme> _refdata_scheme;
    std::unique_ptr<Scheme> _security_group_state_scheme;

    // функция логирования
    // logging function
    void logMessage(const cg_msg_t* message, cg_scheme_desc_t* scheme = nullptr) {
        static const size_t MaxMessageSize = 2048;
        static char message_str[MaxMessageSize];
        size_t buf_size = MaxMessageSize;
        if (cg_msg_dump(message, scheme, message_str, &buf_size) == CG_ERR_OK) {
            log_info(message_str);
        }
    }

    // заполнение значений отступов полей в таблицах
    // вызывается из обработчика CG_MSG_OPEN
    // при изменении серверной схемы следует поменять значения имен полей в SchemeInfo
    //
    // filling table fields offset values
    // called from CG_MSG_OPEN handler
    // in case of server scheme change it is necessary to update field names values in SchemeInfo
    void initializeSchemeRefData(cg_listener_t* listener) {
        cg_scheme_desc_t* scheme_desc = 0;
        WARN_FAIL(cg_lsn_getscheme(listener, &scheme_desc));
        assert(scheme_desc);
        _refdata_scheme = std::make_unique<Scheme>(scheme_desc);
    }

    // заполнение значений отступов полей в таблицах
    // вызывается из обработчика CG_MSG_OPEN
    // при изменении серверной схемы следует поменять значения имен полей в SchemeInfo
    //
    // filling table fields offset values
    // called from CG_MSG_OPEN handler
    // in case of server scheme change it is necessary to update field names values in SchemeInfo
    void initializeSchemeSecurityGroupState(cg_listener_t* listener) {
        cg_scheme_desc_t* scheme_desc = nullptr;
        WARN_FAIL(cg_lsn_getscheme(listener, &scheme_desc));
        assert(scheme_desc);
        _security_group_state_scheme = std::make_unique<Scheme>(scheme_desc);
    }

    // обработка репилики FORTS_REFDATA_REPL таблицы fut_sess_contents. Накапливаем данные в контейнере, для последующей обработки, при получении session_data_ready
    // processing the FORTS_REFDATA_REPL replica of the fut_sess_contents table. Accumulate data in a container for subsequent processing upon receiving session_data_ready
    void processRefDataFutSessContents(const cg_msg_streamdata_t* stream_msg) {
        FutSessContentsRecord record(stream_msg->data, _refdata_scheme->message(stream_msg->msg_index));
        process_record(record, _fut_sess_contents);
    }

    // обработка репилики FORTS_REFDATA_REPL таблицы opt_sess_contents. Накапливаем данные в контейнере, для последующей обработки, при получении session_data_ready
    //processing the FORTS_REFDATA_REPL replica of the opt_sess_contents table. We accumulate data in a container for subsequent processing, upon receiving session_data_ready
    void processRefDataOptSessContents(const cg_msg_streamdata_t* stream_msg) {
        OptSessContentsRecord record(stream_msg->data, _refdata_scheme->message(stream_msg->msg_index));
        process_record(record, _opt_sess_contents);
    }

    // обработка репилики FORTS_REFDATA_REPL таблицы fut_vcb. Накапливаем данные в контейнере, для последующей обработки, при получении session_data_ready
    // processing the FORTS_REFDATA_REPL replica of the fut_vcb table. Accumulate data in a container for subsequent processing upon receiving session_data_ready
    void processRefDataFutVcb(const cg_msg_streamdata_t* stream_msg) {
        FutVcbRecord record(stream_msg->data, _refdata_scheme->message(stream_msg->msg_index));
        process_record(record, _fut_vcb);
    }

    // обработка репилики FORTS_REFDATA_REPL таблицы opt_vcb. Накапливаем данные в контейнере, для последующей обработки, при получении session_data_ready
    // processing the FORTS_REFDATA_REPL replica of the opt_vcb table. Accumulate data in a container for subsequent processing upon receiving session_data_ready
    void processRefDataOptVcb(const cg_msg_streamdata_t* stream_msg) {
        OptVcbRecord record(stream_msg->data, _refdata_scheme->message(stream_msg->msg_index));
        process_record(record, _opt_vcb);
    }

    // обработка репилики FORTS_REFDATA_REPL таблицы sys_events. При получении session_data_ready выставляется 
    // признак _is_refdata_ready, при получении clearing_started сбрасывается признак _is_refdata_ready. 
    // session_data_ready сообщается о полноте и ссылочной связанности полученных данных.

    // processing the FORTS_REFDATA_REPL replica of the sys_events table. When receiving session_data_ready, the
    // _is_refdata_ready flag is set, when receiving clearing_started, the _is_refdata_ready flag is reset.
    // session_data_ready reports the completeness and referential coherence of the received data.
    void processRefDataSysEvents(const cg_msg_streamdata_t* stream_msg) {
        SysEventsRecord record(stream_msg->data, _refdata_scheme->message(stream_msg->msg_index));
        process_record(record, _sys_events);

        // Добавляем запись, если replAct равен нулю
        // Add a record if replAct is zero
        if (record.repl_act == 0) {
            if (record.event_type == SysEvents::DataReady) {
                _is_refdata_ready = true;
            }
            else if (record.event_type == SysEvents::ClearingStarted) {
                _is_refdata_ready = false;
            }
            if (_is_refdata_ready && _is_refdata_online) {
                // Получены все инструменты, данные консистенты, мы в online.
                // Делаем пересчет по всем данным, так как при получении snapshot мы не считали статусы.
                // Нужно recalcAllInstrumentsState вызвать только в начале в момент выхода в online, но так как синхрособытий мало, 
                // чтобы не усложнять код, можно и по кажому синхрособытию пересчитать.
                // 
                // All instruments received, data consistency, we are online.
                // We recalculate for all data, since we did not calculate statuses when receiving a snapshot.
                // We need to call recalcAllInstrumentsState only at the beginning at the moment of going online, but since there are few synchronization events,
                // in order not to complicate the code, we can recalculate for each synchronization event.
                recalcAllInstrumentsState();
            }
        }
    }

    // обработка репилики FORTS_SECURITYGROUPSTATE_REPL таблицы security_group_state. 
    // До момента сихронизации справочников и потока с состояними накапливаем данные в контейнере. После
    // синхронизации потоков справочников и  FORTS_SECURITYGROUPSTATE_REPL, пересчитываем статусы online по группе, для которой пришло изменение статуса.
    // processing the FORTS_SECURITYGROUPSTATE_REPL replica of the security_group_state table.
    // Until the moment of synchronization of the directories and the state stream, we accumulate data in the container. After
    // synchronization of the directory streams and FORTS_SECURITYGROUPSTATE_REPL, we recalculate the online statuses for the group for which the status change has arrived.

    void processSecGrpStatSecurityGroupState(const cg_msg_streamdata_t* stream_msg) {
        SecurityGroupStateRecord record(stream_msg->data, _security_group_state_scheme->message(stream_msg->msg_index));
        process_record(record, _scty_grps);

        // Добавляем запись, если replAct равен нулю
        // Add a record if replAct is zero
        if (record.repl_act == 0) {
            if (_is_security_group_state_online && _is_security_group_state_ready) {
                // пересчитать статусы инстурментов, которые входят в группу, у которой поменялся статус
                // recalculate the statuses of the instruments that are part of the group whose status changed
                calcInstrumentsState(record);
            }
        }
    }

    // обработка репилики FORTS_SECURITYGROUPSTATE_REPL таблицы sys_events. 
    // При получении session_data_ready выставляется 
    // признак _is_security_group_state_ready, при получении clearing_started сбрасывается признак _is_security_group_state_ready. 
    // session_data_ready сообщается о полноте и ссылочной связанности полученных данных.
    //
    // processing the FORTS_SECURITYGROUPSTATE_REPL replica of the sys_events table.
    // When receiving session_data_ready, the
    // _is_security_group_state_ready flag is set, when receiving clearing_started, the _is_security_group_state_ready flag is reset.
    // session_data_ready reports the completeness and referential coherence of the received data.
    void processSecGrpStateSysEvents(const cg_msg_streamdata_t* stream_msg) {
        SysEventsRecord record(stream_msg->data, _security_group_state_scheme->message(stream_msg->msg_index));
        process_record(record, _scty_events);

        // Добавляем запись, если replAct равен нулю
        // Add a record if replAct is zero
        if (record.repl_act == 0) {
            if (record.event_type == SysEvents::DataReady) {
                _is_security_group_state_ready = true;
            }
            else if (record.event_type == SysEvents::ClearingStarted) {
                _is_security_group_state_ready = false;
            }
            if (_is_security_group_state_online && _is_security_group_state_ready) {
                recalcAllInstrumentsState();
            }
        }
    }

    // возвращает для статуса группы приоритет по значимости статуса.
    // returns the priority of the group status by the significance of the status.
    int32_t getSecurityGroupStatePriority(SecurityGroupState state) {
        static const std::map<SecurityGroupState, int32_t> priority = {
            {TRADING_MODE_ALL_ALLOWED, 0},
            {TRADING_MODE_ACCEPT_ORDERS, 1},
            {TRADING_MODE_REJECT_INCREASE_POSITION, 2},
            {TRADING_MODE_REJECT_EVERYTHING_EXCEPT_CANCEL_ORDER, 3},
            {TRADING_MODE_REJECT_EVERYTHING, 4}
        };
        
        const auto it = priority.find(state);
        if (it != priority.end()) {
            return it->second;
        }
        else {
            return -1; // Если состояние не найдено (if the state is not found)
        }
    }

    // возвращает худший статус из всех имеющихся статусов для данной группы.
    // returns the worst status of all available statuses for the given group.
    SecurityGroupState findWorstState(const std::unordered_map<int64_t, SecurityGroupState>& security_group_state) {
        std::pair<SecurityGroupState, int32_t> worstState = { TRADING_MODE_ALL_ALLOWED, 0};
        for (const auto& state: security_group_state) {
            const SecurityGroupState& current_state = state.second;
            const int32_t current_priority = getSecurityGroupStatePriority(current_state);
            if (current_priority  > worstState.second) {
                worstState.first = current_state;
                worstState.second = current_priority;
            }
        }
        return worstState.first;
    }

    // пересчитывает статусы по всем инструментам
    // recalculates statuses for all instruments
    void recalcAllInstrumentsState() {
        for (const auto& kvp : _scty_grps) {
            auto& record = kvp.second;
            calcInstrumentsState(record);
        }
    }

    // рассчитывает статусы инструментов по изменению статуса группы 
    // calculates the status of instruments based on the change in group status
    void calcInstrumentsState(const SecurityGroupStateRecord& record) {
        for (auto& instr_state : _instruments_state) {
            auto& security_group_state = instr_state.security_group_state;
            if (record.type == HaltType::Session) {
                // all instruments
            }
            else if (record.type == HaltType::Section) {
                if (instr_state.section_id != record.section_id)
                    continue;
            }
            else if (record.type == HaltType::SectionAndTradeMode) {
                if (
                        (instr_state.section_id != record.section_id) || 
                        (record.trade_mode_mask & (1 << (instr_state.trade_mode_id - 1))) == 0
                )
                    continue;
            }
            else if (record.type == HaltType::FutBaseContract) {
                if (instr_state.base_contract_id != record.base_contract_id)
                    continue;
            }
            else if (record.type == HaltType::InstrumentsFromTradeMode) {
                if (
                    (record.trade_mode_mask & (1 << (instr_state.trade_mode_id - 1))) == 0
                )
                    continue;
            }
            else if (record.type == HaltType::InstrumentsFromTradeModeAndFutBaseContract) {
                if (
                    (instr_state.base_contract_id != record.base_contract_id) ||
                    (record.trade_mode_mask & (1 << (instr_state.trade_mode_id - 1))) == 0
                )
                    continue;
            }
            else if (record.type == HaltType::GroupInstruments) {
                if (
                        (record.group_mask & instr_state.group_mask) == 0
                    )
                    continue;
            }
            else if (record.type == HaltType::GroupInstrumentsAndFutBaseContract) {
                if (
                    (instr_state.base_contract_id != record.base_contract_id) ||
                    (record.group_mask & instr_state.group_mask) == 0
                    )
                    continue;
            }
            if (security_group_state.find(record.security_group_id) == security_group_state.end())
                security_group_state.emplace(record.security_group_id, static_cast<SecurityGroupState>(record.state));
            else
                security_group_state[record.security_group_id] = static_cast<SecurityGroupState>(record.state);
            instr_state.state = findWorstState(security_group_state);
        }
    }

    // инициализирует список со статусами инструментов, вызывается после выхода в online потока справочников и наличиия события session_data_ready
    // initializes the list with the statuses of the instruments, called after the directory stream goes online and the session_data_ready event is present
    void fillInitialInstrumentState() {
        _instruments_state.clear();
        _vcb_map.clear();

        for (auto& kvp : _fut_vcb) {
            auto& record = kvp.second;
            _vcb_map.emplace(record.base_contract_code, VcbValueType(record.base_contract_id, record.section_id));
        }
        for (auto& kvp : _opt_vcb) {
            auto& record = kvp.second;
            _vcb_map.emplace(record.base_contract_code, VcbValueType(record.base_contract_id, record.section_id));
        }
        for (auto& kvp : _fut_sess_contents) {
            auto& record = kvp.second;
            InstrumentState instrument_state;
            
            if (_vcb_map.find(record.base_contract_code) == _vcb_map.end())
                continue;
            const auto& vcb_data = _vcb_map[record.base_contract_code];
            instrument_state.base_contract_id = vcb_data.first;
            instrument_state.section_id = vcb_data.second;
            instrument_state.isin_id = record.isin_id;
            instrument_state.group_mask = record.group_mask;
            instrument_state.trade_mode_id = record.trade_mode_id;
            instrument_state.state = 0;
            _instruments_state.push_back(instrument_state);
        }
        for (auto& kvp : _opt_sess_contents) {
            auto& record = kvp.second;
            InstrumentState instrument_state;

            if (_vcb_map.find(record.base_contract_code) == _vcb_map.end())
                continue;
            const auto& vcb_data = _vcb_map[record.base_contract_code];
            instrument_state.base_contract_id = vcb_data.first;
            instrument_state.section_id = vcb_data.second;
            instrument_state.isin_id = record.isin_id;
            instrument_state.group_mask = record.group_mask;
            instrument_state.trade_mode_id = record.trade_mode_id;
            instrument_state.state = 0;
            _instruments_state.push_back(instrument_state);
        }

    }

    // обобщенная обработка ClearDeleted
    // generic ClearDeleted handling
    template <typename Container>
    void do_clear(int64_t rev, Container& container) {
        for (auto iter = container.begin(); iter != container.end(); ) {

            // Проверяем, соответствует ли replRev записи текущему max_repl_rev
            // Check if the entry's replRev matches the current max_repl_rev
            if ((*iter).second.repl_rev < rev) {
                iter = container.erase(iter);
            } else {
                ++iter;
            }
        }
    }

    // обобщенная обработка записи в контейнере
    // generalized handling of a record in a container
    template <typename Record, typename Container>
    void process_record(const Record& record, Container& container) {
        // Добавляем/Обновляем запись, если replAct равен нулю
        // Add/Update record if replAct is null
        if (record.repl_act == 0) {
            if (container.find(record.repl_id) == container.end())
                container.try_emplace(record.repl_id, record);
            else
                container[record.repl_id] = record;
        } // Удаляем запись, если replAct ненулевой (Delete the record if replAct is non-zero)
        else {
            container.erase(record.repl_id);
        }
    }

    // основной обработчик сообщений FORTS_REFDATA_REPL
    //
    // main message callback function
    static CG_RESULT messageCallbackRefData(cg_conn_t* conn, cg_listener_t* listener, cg_msg_t* message, void* data) {
        InstrumentsStateApp* _this = static_cast<InstrumentsStateApp*>(data);
        if (_this == nullptr)
            return CG_ERR_INVALIDARGUMENT;

        // открытие нового соединения с сервером
        // инициализировать схему
        //
        // opening new server connection
        // initialize scheme description
        switch (message->type) {
            case CG_MSG_OPEN: {
                _this->logMessage(message);
                _this->cleanupRefData(); // in case of reconnect
                _this->initializeSchemeRefData(listener);
                break;
            }

            case CG_MSG_STREAM_DATA: {
                const Scheme& scheme = *_this->_refdata_scheme;
                cg_msg_streamdata_t* stream_msg = reinterpret_cast<cg_msg_streamdata_t*>(message);
                const std::string name = stream_msg->msg_name;
                _this->logMessage(message, scheme.cgscheme());

                // Обработка для fut_sess_contents
                // Processing for fut_sess_contents
                if (name == "fut_sess_contents") {
                    _this->processRefDataFutSessContents(stream_msg);
                }
                // Обработка для opt_sess_contents
                // Processing for opt_sess_contents
                else if (name == "opt_sess_contents") {
                    _this->processRefDataOptSessContents(stream_msg);
                }
                // Обработка для fut_vcb
                // Processing for fut_vcb
                else if (name == "fut_vcb") {
                    _this->processRefDataFutVcb(stream_msg);
                }
                // Обработка для opt_vcb
                // Processing for opt_vcb
                else if (name == "opt_vcb") {
                    _this->processRefDataOptVcb(stream_msg);
                }
                // Обработка для sys_events
                // Processing for sys_events
                else if (name == "sys_events") {
                    _this->processRefDataSysEvents(stream_msg);
                }
                break;
            }
        
            // данное сообщение говорит о получении полной реплики базы данных
            // с этого момента мы будем получать обновления
            //
            // this message tells that we have completed database replica
            // since this moment we shall receive updates only
            case CG_MSG_P2REPL_ONLINE: {
                log_info("REFDATA: ONLINE");
                _this->logMessage(message);
                _this->_is_refdata_online = true;
                if (_this->_is_refdata_ready && _this->_is_refdata_online) {
                    _this->fillInitialInstrumentState();
                    _this->recalcAllInstrumentsState();
                }

                break;
            }

            // очистить все данные при смене номера жизни
            //
            // clear all data when life number is changed
            case CG_MSG_P2REPL_LIFENUM: {
                _this->logMessage(message);
                log_info("Life number is changed");
                _this->cleanupRefData();
                break;
            }

            // удалить все заявки с ревизией, старшей, чем указанная
            //
            // clear older orders
            case CG_MSG_P2REPL_CLEARDELETED: {
                const cg_data_cleardeleted_t* rev_info = reinterpret_cast<cg_data_cleardeleted_t*>(message->data);
                _this->logMessage(message);
                const std::string name = _this->_refdata_scheme->name(rev_info->table_idx);
                // Обработка для fut_sess_contents
                // Processing for fut_sess_contents
                if (name == "fut_sess_contents") {
                    _this->do_clear(rev_info->table_rev, _this->_fut_sess_contents);
                }
                // Обработка для opt_sess_contents
                // Processing for opt_sess_contents
                else if (name == "opt_sess_contents") {
                    _this->do_clear(rev_info->table_rev, _this->_opt_sess_contents);
                }
                // Обработка для fut_vcb
                // Processing for fut_vcb
                else if (name == "fut_vcb") {
                    _this->do_clear(rev_info->table_rev, _this->_fut_vcb);
                }
                // Обработка для opt_vcb
                // Processing for opt_vcb
                else if (name == "opt_vcb") {
                    _this->do_clear(rev_info->table_rev, _this->_opt_vcb);
                }
                // Обработка для sys_events
                // Processing for sys_events
                else if (name == "sys_events") {
                    _this->do_clear(rev_info->table_rev, _this->_sys_events);
                }
                break;
            }

            // закрытие соединения 
            // записать содержимое стакана в файл
            // произвести очистку всех данных
            //
            // closing current connection
            // write order book contents into file
            // perform complete cleanup
            case CG_MSG_CLOSE: {
                _this->logMessage(message);
                _this->print();
                _this->cleanupRefData();
                break;
            }

            default:
                break;
        };

        return CG_ERR_OK;
    }

    // основной обработчик сообщений FORTS_SECURITYGROUPSTATE_REPL
    // main message handler FORTS_SECURITYGROUPSTATE_REPL
    static CG_RESULT messageCallbackSecurityGroupState(cg_conn_t* conn, cg_listener_t* listener, cg_msg_t* message, void* data) {
        InstrumentsStateApp* _this = static_cast<InstrumentsStateApp*>(data);
        if (_this == nullptr)
            return CG_ERR_INVALIDARGUMENT;

        // открытие нового соединения с сервером
        // инициализировать схему
        //
        // opening new server connection
        // initialize scheme description
        switch (message->type) {
            case CG_MSG_OPEN: {
                _this->logMessage(message);
                _this->cleanupSecurityGroupState(); // in case of reconnect
                _this->initializeSchemeSecurityGroupState(listener);
                break;
            }

            case CG_MSG_STREAM_DATA: {
                const Scheme& scheme = *_this->_security_group_state_scheme;
                cg_msg_streamdata_t* stream_msg = reinterpret_cast<cg_msg_streamdata_t*>(message);
                const std::string name = stream_msg->msg_name;
                _this->logMessage(message, scheme.cgscheme());

                if (name == "security_group_state") {
                    _this->processSecGrpStatSecurityGroupState(stream_msg);
                }
                else if (name == "sys_events") {
                    _this->processSecGrpStateSysEvents(stream_msg);
                }
                break;
            }

            // данное сообщение говорит о получении полной реплики базы данных
            // с этого момента мы будем получать обновления
            //
            // this message tells that we have completed database replica
            // since this moment we shall receive updates only
            case CG_MSG_P2REPL_ONLINE: {
                log_info("SGS: ONLINE");
                _this->logMessage(message);
                _this->_is_security_group_state_online = true;
                if (_this->_is_security_group_state_ready && _this->_is_security_group_state_online) {
                    _this->recalcAllInstrumentsState();
                }
                break;
            }

            // очистить все данные при смене номера жизни
            //
            // clear all data when life number is changed
            case CG_MSG_P2REPL_LIFENUM: {
                _this->logMessage(message);
                log_info("SGS: Life number is changed");
                _this->cleanupSecurityGroupState();
                break;
            }

            // удалить все заявки с ревизией, старшей, чем указанная
            //
            // clear older orders
            case CG_MSG_P2REPL_CLEARDELETED: {
                const cg_data_cleardeleted_t* rev_info = reinterpret_cast<cg_data_cleardeleted_t*>(message->data);
                _this->logMessage(message);
                const std::string name = _this->_security_group_state_scheme->name(rev_info->table_idx);
                // Обработка для fut_sess_contents
                // Handling for fut_sess_contents
                if (name == "security_group_state") {
                    _this->do_clear(rev_info->table_rev, _this->_scty_grps);
                }
                // Обработка для sys_events
                // Handling for sys_events
                else if (name == "sys_events") {
                    _this->do_clear(rev_info->table_rev, _this->_scty_events);
                }
                break;
            }

            // закрытие соединения 
            // записать содержимое стакана в файл
            // произвести очистку всех данных
            //
            // closing current connection
            // write order book contents into file
            // perform complete cleanup
            case CG_MSG_CLOSE: {
                _this->logMessage(message);
                _this->print();
                _this->cleanupSecurityGroupState();
                break;
            }
            default:
                break;
        };

        return CG_ERR_OK;
    }

public:
    InstrumentsStateApp(cg_conn_t* conn, const char* refdata_lsn_url, const char* secgrpst_lsn_url, const std::string& outfile)
        : _connection(conn)
        , _listener_refdata(nullptr)
        , _listener_scrgrpst(nullptr)
        , _outfile(outfile)
        , _last_revision(0)
        , _is_refdata_ready(false)
        , _is_refdata_online(false)
        ,_is_security_group_state_ready(false)
        ,_is_security_group_state_online(false) {
        if (cg_lsn_new(_connection, refdata_lsn_url, &messageCallbackRefData, this, &_listener_refdata) != CG_ERR_OK)
            throw std::exception();

        if (cg_lsn_new(_connection, secgrpst_lsn_url, &messageCallbackSecurityGroupState, this, &_listener_scrgrpst) != CG_ERR_OK)
            throw std::exception();

        log_info("Instruments state created, refdata = %s, secure group state = %s", refdata_lsn_url, secgrpst_lsn_url);
    }

    ~InstrumentsStateApp() {
        log_info("Closing Instrument state application");
        if (_listener_refdata)
            cg_lsn_destroy(_listener_refdata);
        
        if (_listener_scrgrpst)
            cg_lsn_destroy(_listener_scrgrpst);
                
        cleanupRefData();
    }
    
    void cleanupRefData() {
        log_info("Performing cleanup ref data");
        _fut_sess_contents.clear();
        _opt_sess_contents.clear();
        _fut_vcb.clear();
        _opt_vcb.clear();
        _sys_events.clear();
        _is_refdata_ready = false;
        _is_refdata_online = false;
    }

    void cleanupSecurityGroupState() {
        log_trace("Performing cleanup security group state");
        _is_security_group_state_ready = false;
        _is_security_group_state_online = false;
    }

    void print() {
        FILE* file = fopen(_outfile.c_str(), "w");
        if (file == nullptr) {
            EXIT_WITH_ERROR("Unable to create output file");
        }
        for (auto& instr_state : _instruments_state) {
            fprintf(file, "%d : %d\n", instr_state.isin_id, instr_state.state);
        }
        fclose(file);
        return;
    }

    // основной цикл приложения
    //
    // main application loop body
    void run()
    {
        uint32_t state{};
        WARN_FAIL(cg_conn_getstate(_connection, &state));

        if (state == CG_STATE_ERROR)
        {
            WARN_FAIL(cg_conn_close(_connection));
        }
        else
        if (state == CG_STATE_CLOSED)
        {
            WARN_FAIL(cg_conn_open(_connection, 0));
        }
        else
        if (state == CG_STATE_ACTIVE)
        {
            uint32_t result = cg_conn_process(_connection, 1, 0);
            if (result != CG_ERR_OK && result != CG_ERR_TIMEOUT) 
            {
                log_info("Warning: connection state request failed: %X", result);
            }
        
            WARN_FAIL(cg_lsn_getstate(_listener_refdata, &state));

            switch (state)
            {
            case CG_STATE_CLOSED:
                    WARN_FAIL(cg_lsn_open(_listener_refdata, 0));
                    break;
            case CG_STATE_ERROR:
                    WARN_FAIL(cg_lsn_close(_listener_refdata));
                    throw DisconnectException();
                    break;
            }

            WARN_FAIL(cg_lsn_getstate(_listener_scrgrpst, &state));

            switch (state)
            {
            case CG_STATE_CLOSED:
                    WARN_FAIL(cg_lsn_open(_listener_scrgrpst, 0));
                    break;
            case CG_STATE_ERROR:
                    WARN_FAIL(cg_lsn_close(_listener_scrgrpst));
                    throw DisconnectException();
                    break;
            }
        }
    }
};

// флаг выхода из программы
//
// helper variable used as a notification flag to exit
#ifdef _WIN32
static volatile int bExit = 0;
#elif __unix__
volatile sig_atomic_t bExit = 0;
#endif

#ifdef _WIN32
BOOL WINAPI InterruptHandler(DWORD reason)
{
    bExit = 1;
    return 1;
}
#elif __unix__
void InterruptHandler(int signal)
{
    bExit = 1;
}
#endif

int main(int argc, char* argv[])
{
    // connection pointer
    conn_t* conn = 0;

    // connection init string
    const char* conn_str = "p2tcp://127.0.0.1:4001;app_name=instrs_state";

#ifdef _WIN32
    SetConsoleCtrlHandler(InterruptHandler, 1); 
#elif __unix__
    signal(SIGINT, InterruptHandler);
#endif

    if (argc < 2) 
    {
        printf("Usage:\n\tinstruments_state outfile\n\n"
            "\toutfile is output filename with path");
        return 2;
    }
    
    std::string outfile(argv[1]);

    // Environment is initialized using provided INI-file
    CHECK_FAIL(env_open("ini=instruments_state.ini;key=11111111;minloglevel=trace"));
    // Connection is initialized using connection init string
    CHECK_FAIL(conn_new(conn_str, &conn));

    int code = 0;
    try
    {
        InstrumentsStateApp app(conn, "p2repl://FORTS_REFDATA_REPL", "p2repl://FORTS_SECURITYGROUPSTATE_REPL", outfile);
        while (!bExit) // check global interrupt flag
        {
            try
            {
                app.run();
            }
            catch(DisconnectException&)
            {
                log_info("Disconnection");
            }

            if(_kbhit())
            {
                char ch = getchar();
                if(ch == 0x0a)
                {
                    app.print();
                }
            }
        }
        log_info("----BREAK----");
    }
    catch (std::exception& e)
    {
        fprintf(stderr, "Exception: %s", e.what());
        code = 1;
    }

    // destroy connection
    if (conn != nullptr)
        conn_destroy(conn);

    // close environment
    CHECK_FAIL(env_close());
    return code;
}

