feat(dup): implement pegasus_mutation_duplicator

src/base/pegasus_key_schema.h

@@ -150,4 +150,10 @@ inline uint64_t pegasus_key_hash(const ::dsn::blob &key)
     }
 }
 
+/// Calculate hash value from hash key.
+inline uint64_t pegasus_hash_key_hash(const ::dsn::blob &hash_key)
+{
+    return dsn::utils::crc64_calc(hash_key.data(), hash_key.length(), 0);
+}
+
 } // namespace pegasus

src/base/pegasus_rpc_types.h

@@ -22,6 +22,8 @@ using incr_rpc = dsn::rpc_holder<dsn::apps::incr_request, dsn::apps::incr_respon
 using check_and_set_rpc =
     dsn::rpc_holder<dsn::apps::check_and_set_request, dsn::apps::check_and_set_response>;
 
+using duplicate_rpc = dsn::apps::duplicate_rpc;
+
 using check_and_mutate_rpc =
     dsn::rpc_holder<dsn::apps::check_and_mutate_request, dsn::apps::check_and_mutate_response>;
 

src/client_lib/pegasus_client_impl.cpp

@@ -1244,6 +1244,13 @@ int pegasus_client_impl::get_unordered_scanners(int max_split_count,
     return ret;
 }
 
+void pegasus_client_impl::async_duplicate(dsn::apps::duplicate_rpc rpc,
+                                          std::function<void(dsn::error_code)> &&callback,
+                                          dsn::task_tracker *tracker)
+{
+    _client->duplicate(rpc, std::move(callback), tracker);
+}
+
 const char *pegasus_client_impl::get_error_string(int error_code) const
 {
     auto it = _client_error_to_string.find(error_code);

src/client_lib/pegasus_client_impl.h

@@ -221,6 +221,13 @@ class pegasus_client_impl : public pegasus_client
                                  const scan_options &options,
                                  async_get_unordered_scanners_callback_t &&callback) override;
 
+    /// \internal
+    /// This is an internal function for duplication.
+    /// \see pegasus::server::pegasus_mutation_duplicator
+    void async_duplicate(dsn::apps::duplicate_rpc rpc,
+                         std::function<void(dsn::error_code)> &&callback,
+                         dsn::task_tracker *tracker);
+
     virtual const char *get_error_string(int error_code) const override;
 
     static void init_error();
@@ -279,6 +286,9 @@ class pegasus_client_impl : public pegasus_client
         static const ::dsn::blob _max;
     };
 
+    static int get_client_error(int server_error);
+    static int get_rocksdb_server_error(int rocskdb_error);
+
 private:
     class pegasus_scanner_impl_wrapper : public abstract_pegasus_scanner
     {
@@ -298,9 +308,6 @@ class pegasus_client_impl : public pegasus_client
         }
     };
 
-    static int get_client_error(int server_error);
-    static int get_rocksdb_server_error(int rocskdb_error);
-
 private:
     std::string _cluster_name;
     std::string _app_name;

src/idl/dsn.thrift

@@ -4,3 +4,7 @@ namespace cpp dsn
 struct blob
 {
 }
+
+struct task_code
+{
+}

src/idl/rrdb.thrift

@@ -251,6 +251,26 @@ struct scan_response
     6:string        server;
 }
 
+struct duplicate_request
+{
+    // The timestamp of this write.
+    1: optional i64 timestamp
+
+    // The code to identify this write.
+    2: optional dsn.task_code task_code
+
+    // The binary form of the write.
+    3: optional dsn.blob raw_message
+
+    // ID of the cluster where this write comes from.
+    4: optional byte cluster_id
+}
+
+struct duplicate_response
+{
+    1: optional i32 error;
+}
+
 service rrdb
 {
     update_response put(1:update_request update);