HA.h

Go to the documentation of this file.

#ifndef MEDUSA_HA_H
#define MEDUSA_HA_H

#include "BasicComponent.h"
#include "Recovery.h" 

MEDUSA_NAMESPACE_BEGIN


class HA : public BasicComponent {

 public:

    // RPC: #include "Schema.h"
    // RPC: #include "StreamDef.h"
    // RPC: #include "Query.h"
    // RPC: #include "Name.h"

    class NodeType {
    public:
        static const NodeType PRIMARY;
        static const NodeType REPLICA;
        static const NodeType PROVIDER;
        static const NodeType NONE;
        bool operator==(const NodeType& t) const {
            return (t.m_type == m_type);
        }
        string as_string() const { return m_type; }
        // Ideally the constructor should be private to
        // ensure that no one creates new instances of PrimaryType
        NodeType(string type) : m_type(type) {}
    private:
        string m_type;
    };

    class MonitoredNode {
    public:
        MonitoredNode(MedusaID id, NodeType t) 
            : m_node_id(id), m_type(t), m_started(false), m_recovered(false), m_failed_pings(0), 
            m_pending_ping(false) {}

        MedusaID m_node_id;

        NodeType m_type;

        bool m_started;

        bool m_recovered;

        int m_failed_pings;
        // After a few failed pings we consider monitored node to have failed    
        static const int MAX_FAILED_PINGS = 3; 

        bool m_pending_ping;

        // The stuff below is only used for PROVIDER type of monitored node so 
        // we should probably decouple this
        // List of streams monitored because of that node and alternates if available
        typedef vector< ptr<StreamDef> > AltStreams; // Includes the StreamDef of main stream
        typedef map< StreamID, AltStreams > MonitoredStreams;
        MonitoredStreams m_streams;
        
        // The stuff below is only used for PRIMARY type of monitored node so
        // we should probably decouple this
        vector<Query> m_queries;        
        vector<Subscription> m_subs;

        string as_string() const {
            return "Monitored {id=" + to_string(m_node_id) + ",type=" + m_type + "}";
        }
    };

    class Secondary {
    public:
        Secondary(MedusaID secondary) 
            : m_secondary_id(secondary) {}

        MedusaID m_secondary_id;
        string as_string() const {
            return "Secondary {id=" + to_string(m_secondary_id) + "}";
        }
    };

    typedef map<MedusaID, ptr<MonitoredNode> > MonitoredNodes;
    
    typedef map<MedusaID, ptr<Secondary> > Secondaries;
    
    HA(string id, string configFileName=DEFAULT_CONFIG_FILE, 
       RecoveryMethod recovery_method = UPSTREAM_BACKUP);
    
    ~HA() {}

    AsyncRPC<void> replicate_query(MedusaID primary, Query query);

    AsyncRPC<void> replicate_subs(MedusaID primary, vector<Subscription> subs);                            

    RPC<void> keep_alive();

    RPC<void> failure_notification(MedusaID failed_node, Time time_failure_detected);

    RPC<void> start();

    RPC<void> stop();

    
    NMSTL_RPC_OBJECT(HA);
    
    inline RecoveryMethod get_recovery_method() {return m_recovery_method;};
    
    inline bool is_primary() {return m_is_primary;};

    RPC< vector<MedusaID> > get_secondary_ids() { return m_secondary_ids; }

 protected:
    // Starts periodic primary monitoring
    void in_thread_init();

 private:

    void init_qp();

    void init_config();
    void init_global(const DOMElement* ha_config);
    void init_process_pairs(const DOMElement* ha_config);
    void init_replicas(const DOMElement* ha_config);

    void register_primaries();
    void register_primaries_2(MedusaID primary, RPC<void> result);

    void replicate_response(MedusaID secondary, RPC<void> result);
    
    void monitor_nodes();

    void monitor_nodes_2(MedusaID primary,RPC<void> result);

    void failure_notification_2(MedusaID failed_node, Time time_failure_detected, RPC<MedusaID> result);
    void failure_notification_3(MedusaID failed_node, MedusaID secondary_node, RPC<void> result);

    void primary_failure(MedusaID failed_node);

    void primary_failure_2(RPC<void> result);


    void monitor_admin();

    void admin_stream_notification(RPC<InputStreamInfo> changed_input);
    void local_admin_notification(StreamDef changed_input, MonitoredNode::AltStreams& alt_streams);

    void admin_query_notification(RPC<Query> new_query);

    void admin_sub_notification(RPC< vector<Subscription> > new_subs);

    void provider_failure(MedusaID failed_node, Time failure_time);

    void update_stream_resp(RPC<void> result);

    void generic_qp_resp(RPC<void> result);

    ptr<MonitoredNode> get_monitored_node(MedusaID node);

    MonitoredNodes m_monitored_nodes;

    vector<MedusaID> m_replica_ids; 

    Secondaries m_secondaries;
    vector<MedusaID> m_secondary_ids;
   
    // File with load management and HA config information
    string m_configFileName;
    int m_ping_interval;
    int m_ping_start;

    bool m_enabled;

    RecoveryMethod m_recovery_method;

    bool m_is_primary;
    
};

NMSTL_TO_STRING(HA::NodeType);
NMSTL_TO_STRING(HA::MonitoredNode);
NMSTL_TO_STRING(HA::Secondary);
MEDUSA_NAMESPACE_END

#endif

---