duplication: add meta_duplication_service (part 3) (#246)

src/dist/replication/common/duplication_common.cpp

@@ -122,9 +122,7 @@ class duplication_group_registry : public utils::singleton<duplication_group_reg
 
     doc.AddMember("dupid", dup.dupid, alloc);
     doc.AddMember("status", rapidjson::StringRef(duplication_status_to_string(dup.status)), alloc);
-    doc.AddMember("remote",
-                  rapidjson::StringRef(dup.remote_address.data(), dup.remote_address.length()),
-                  alloc);
+    doc.AddMember("remote", rapidjson::StringRef(dup.remote.data(), dup.remote.length()), alloc);
     doc.AddMember("create_ts", dup.create_ts, alloc);
 
     doc.AddMember("progress", rapidjson::Value(), alloc);

src/dist/replication/ddl_lib/replication_ddl_client.cpp

@@ -1418,11 +1418,11 @@ dsn::error_code replication_ddl_client::query_restore(int32_t restore_app_id, bo
 }
 
 error_with<duplication_add_response>
-replication_ddl_client::add_dup(std::string app_name, std::string remote_address, bool freezed)
+replication_ddl_client::add_dup(std::string app_name, std::string remote_cluster_name, bool freezed)
 {
     auto req = make_unique<duplication_add_request>();
     req->app_name = std::move(app_name);
-    req->remote_cluster_address = std::move(remote_address);
+    req->remote_cluster_name = std::move(remote_cluster_name);
     req->freezed = freezed;
     return call_rpc_sync(duplication_add_rpc(std::move(req), RPC_CM_ADD_DUPLICATION));
 }

src/dist/replication/meta_server/duplication/duplication_info.cpp

@@ -24,7 +24,8 @@
  * THE SOFTWARE.
  */
 
-#include "dist/replication/meta_server/duplication/duplication_info.h"
+#include "duplication_info.h"
+#include "dist/replication/meta_server/meta_data.h"
 
 #include <rapidjson/prettywriter.h>
 #include <dsn/dist/fmt_logging.h>
@@ -77,9 +78,9 @@ error_code duplication_info::do_alter_status(duplication_status::type to_status)
         return ERR_INVALID_PARAMETERS;
     }
 
-    if (status != to_status) {
+    if (_status != to_status) {
         _is_altering = true;
-        next_status = to_status;
+        _next_status = to_status;
     }
 
     return ERR_OK;
@@ -135,22 +136,22 @@ void duplication_info::persist_status()
 
     dassert_dup(_is_altering, this, "");
     _is_altering = false;
-    status = next_status;
-    next_status = duplication_status::DS_INIT;
+    _status = _next_status;
+    _next_status = duplication_status::DS_INIT;
 }
 
 std::string duplication_info::to_string() const
 {
     return duplication_entry_to_string(to_duplication_entry());
 }
 
-blob duplication_info::to_json_blob_in_status(duplication_status::type to_status) const
+blob duplication_info::to_json_blob() const
 {
-    duplication_info copy;
-    const_cast<uint64_t &>(copy.create_timestamp_ms) = create_timestamp_ms;
-    const_cast<std::string &>(copy.remote) = remote;
-    copy.status = to_status;
-    return json::json_forwarder<duplication_info>::encode(copy);
+    json_helper copy;
+    copy.create_timestamp_ms = create_timestamp_ms;
+    copy.remote = remote;
+    copy.status = _next_status;
+    return json::json_forwarder<json_helper>::encode(copy);
 }
 
 void duplication_info::report_progress_if_time_up()
@@ -162,5 +163,21 @@ void duplication_info::report_progress_if_time_up()
     }
 }
 
+duplication_info_s_ptr duplication_info::decode_from_blob(dupid_t dup_id,
+                                                          int32_t app_id,
+                                                          int32_t partition_count,
+                                                          std::string store_path,
+                                                          const blob &json)
+{
+    json_helper info;
+    if (!json::json_forwarder<json_helper>::decode(json, info)) {
+        return nullptr;
+    }
+    auto dup = std::make_shared<duplication_info>(
+        dup_id, app_id, partition_count, std::move(info.remote), std::move(store_path));
+    dup->_status = info.status;
+    return dup;
+}
+
 } // namespace replication
 } // namespace dsn