add add_data_rpc_service to repl_dev

Author: JacksonYao287

conanfile.py

@@ -9,7 +9,7 @@
 
 class HomestoreConan(ConanFile):
     name = "homestore"
-    version = "7.1.4"
+    version = "7.2.1"
 
     homepage = "https://github.com/eBay/Homestore"
     description = "HomeStore Storage Engine"

src/include/homestore/replication/repl_dev.h

@@ -17,6 +17,7 @@
 #include <homestore/replication/repl_decls.h>
 #include <homestore/blkdata_service.hpp>
 #include <libnuraft/snapshot.hxx>
+#include <nuraft_mesg/mesg_state_mgr.hpp>
 
 namespace nuraft {
 template < typename T >
@@ -438,10 +439,16 @@ class ReplDevListener {
     /// @brief when restart, after all the logs are replayed and before joining raft group, notify the upper layer
     virtual void on_log_replay_done(const group_id_t& group_id) {};
 
+    virtual void on_become_leader(const group_id_t& group_id) {};
+
+    virtual void on_become_follower(const group_id_t& group_id) {};
+
 private:
     std::weak_ptr< ReplDev > m_repl_dev;
 };
 
+using data_service_request_handler_t = std::function< void(boost::intrusive_ptr< sisl::GenericRpcData >& rpc_data) >;
+
 class ReplDev {
 public:
     ReplDev() = default;
@@ -624,6 +631,13 @@ class ReplDev {
     // create a snapshot manually and try to compact logs upto compact_lsn
     virtual void trigger_snapshot_creation(repl_lsn_t compact_lsn, bool wait_for_commit) = 0;
 
+    virtual bool add_data_rpc_service(std::string const& request_name,
+                                      data_service_request_handler_t const& request_handler) {
+        return true;
+    }
+
+    virtual nuraft_mesg::repl_service_ctx* get_group_msg_service() { return nullptr; }
+
 protected:
     shared< ReplDevListener > m_listener;
 

src/lib/replication/repl_dev/raft_repl_dev.cpp

@@ -138,6 +138,13 @@ bool RaftReplDev::join_group() {
     return true;
 }
 
+bool RaftReplDev::add_data_rpc_service(std::string const& request_name,
+                                       data_service_request_handler_t const& request_handler) {
+    return m_msg_mgr.bind_data_service_request(request_name, m_group_id, request_handler);
+}
+
+nuraft_mesg::repl_service_ctx* RaftReplDev::get_group_msg_service() { return group_msg_service(); }
+
 // All the steps in the implementation should be idempotent and retryable.
 AsyncReplResult<> RaftReplDev::start_replace_member(std::string& task_id, const replica_member_info& member_out,
                                                     const replica_member_info& member_in, uint32_t commit_quorum,
@@ -2820,6 +2827,7 @@ void RaftReplDev::become_leader_cb() {
     // becoming a leader.
 
     RD_LOGD(NO_TRACE_ID, "become_leader_cb: setting traffic_ready_lsn from {} to {}", current_gate, new_gate);
+    m_listener->on_become_leader(m_group_id);
 }
 
 bool RaftReplDev::is_ready_for_traffic() const {

src/lib/replication/repl_dev/raft_repl_dev.h

@@ -322,6 +322,9 @@ class RaftReplDev : public ReplDev,
     void pause_state_machine(size_t timeout) override;
     void resume_state_machine() override;
     bool is_state_machine_paused() override;
+    bool add_data_rpc_service(std::string const& request_name,
+                              data_service_request_handler_t const& request_handler) override;
+    virtual nuraft_mesg::repl_service_ctx* get_group_msg_service() override;
 
     std::shared_ptr< snapshot_context > deserialize_snapshot_context(sisl::io_blob_safe& snp_ctx) override {
         return std::make_shared< nuraft_snapshot_context >(snp_ctx);
@@ -355,6 +358,7 @@ class RaftReplDev : public ReplDev,
     void become_follower_cb() {
         m_traffic_ready_lsn.store(0);
         RD_LOGD(NO_TRACE_ID, "become_follower_cb called!");
+        m_listener->on_become_follower(m_group_id);
     }
 
     /// @brief This method is called when the data journal is compacted