session.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <chrono>
#include <string>
#include <thread>
#include <atomic>
#include <functional>
#include <memory>
#include <map>
 
#include "fix/fix_codec.hpp"
#include "fix/application.hpp"
#include "base/concurrentqueue.h"
#include "base/timing_wheel.hpp"
 
namespace fix40 {
 
class Connection;
class Session;
class IStore;  // 前向声明
 
class IStateHandler {
public:
    virtual ~IStateHandler() = default;
 
    virtual void onMessageReceived(Session& context, const FixMessage& msg) = 0;
 
    virtual void onTimerCheck(Session& context) = 0;
 
    virtual void onSessionStart(Session& context) = 0;
 
    virtual void onLogoutRequest(Session& context, const std::string& reason) = 0;
 
    virtual const char* getStateName() const = 0;
};
 
 
    class Session : public std::enable_shared_from_this<Session> {
    public:
        using ShutdownCallback = std::function<void()>;
 
        using EstablishedCallback = std::function<void(std::shared_ptr<Session>)>;
 
    Session(const std::string& sender,
            const std::string& target,
            int hb,
            ShutdownCallback shutdown_cb,
            IStore* store = nullptr);
 
    ~Session();
 
    void set_connection(std::weak_ptr<Connection> conn);
 
    void set_application(Application* app);
 
    Application* get_application() const;
 
        SessionID get_session_id() const;
 
        void set_established_callback(EstablishedCallback cb);
 
        void notify_established();
 
        void set_target_comp_id(const std::string& target);
 
    void send_app_message(FixMessage& msg);
 
    void start();
 
    void stop();
 
    bool is_running() const { return running_; }
 
    void send(FixMessage& msg);
 
    void send_buffered_data();
 
    void handle_write_ready();
 
    void enqueue_raw_for_send(std::string&& raw_msg);
 
    bool is_outbound_queue_empty() const;
 
    void on_message_received(const FixMessage& msg);
 
    void on_timer_check();
 
    void on_io_error(const std::string& reason);
 
    void on_shutdown(const std::string& reason);
 
    void initiate_logout(const std::string& reason);
 
    void schedule_timer_tasks(TimingWheel* wheel);
 
        void changeState(std::unique_ptr<IStateHandler> newState);
 
        const std::string senderCompID;  
        std::string targetCompID;  
        FixCodec codec_;                 
 
        // --- 公共辅助函数（供状态类调用）---
 
    void send_logout(const std::string& reason);
 
    void send_heartbeat(const std::string& test_req_id = "");
 
    void send_test_request(const std::string& id);
 
    void perform_shutdown(const std::string& reason);
 
    void update_last_recv_time();
 
    void update_last_send_time();
 
    std::chrono::steady_clock::time_point get_last_recv_time() const;
 
    std::chrono::steady_clock::time_point get_last_send_time() const;
 
    int get_heart_bt_int() const;
 
    void set_heart_bt_int(int new_hb);
 
    int get_min_heart_bt_int() const;
 
    int get_max_heart_bt_int() const;
 
    // --- 序列号管理 ---
 
    int get_send_seq_num() { return sendSeqNum; }
 
    int get_recv_seq_num() { return recvSeqNum; }
 
    void increment_send_seq_num() { sendSeqNum++; }
 
    void increment_recv_seq_num() { recvSeqNum++; }
 
    void set_recv_seq_num(int seq) { recvSeqNum = seq; }
 
    void set_send_seq_num(int seq) { sendSeqNum = seq; }
 
    // --- 断线恢复相关 ---
 
    void send_resend_request(int begin_seq_no, int end_seq_no);
 
    void send_sequence_reset_gap_fill(int seq_num, int new_seq_no);
 
    void save_session_state();
 
    bool restore_session_state();
 
    IStore* get_store() const { return store_; }
 
    std::weak_ptr<Connection> get_connection() const { return connection_; }
 
    bool is_processing_resend() const { return processingResend_; }
 
    void set_processing_resend(bool processing) { processingResend_ = processing; }
 
    // --- 客户端身份标识 ---
 
    const std::string& get_client_comp_id() const { return clientCompID_; }
 
    void set_client_comp_id(const std::string& clientId) { clientCompID_ = clientId; }
 
private:
    std::atomic<bool> shutting_down_{false};   
    std::recursive_mutex state_mutex_;          
    std::unique_ptr<IStateHandler> currentState_; 
 
    void internal_send(const std::string& raw_msg);
 
    void drain_pending_inbound_locked();
 
    int heartBtInt;                  
    const int minHeartBtInt_;        
    const int maxHeartBtInt_;        
    ShutdownCallback shutdown_callback_; 
    std::weak_ptr<Connection> connection_; 
 
    moodycamel::ConcurrentQueue<std::string> outbound_q_; 
 
    std::atomic<bool> running_{false}; 
 
    int sendSeqNum = 1;  
    int recvSeqNum = 1;  
    std::chrono::steady_clock::time_point lastRecv; 
    std::chrono::steady_clock::time_point lastSend; 
 
    std::map<int, FixMessage> pending_inbound_;
    int last_resend_request_end_ = 0;
 
    TimingWheel* timing_wheel_ = nullptr;  
    TimerTaskId timer_task_id_ = INVALID_TIMER_ID; 
 
        Application* application_ = nullptr;  
        IStore* store_ = nullptr;             
        bool processingResend_ = false;       
        std::string clientCompID_;            
        EstablishedCallback established_callback_; 
        std::atomic<bool> established_notified_{false}; 
    };
 
    } // namespace fix40