1 /*
2 * Copyright (c) 2021 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "single_ver_sync_engine.h"
17 #include "db_common.h"
18 #include "log_print.h"
19 #include "single_ver_kv_sync_task_context.h"
20 #include "single_ver_relational_sync_task_context.h"
21 #include "single_ver_sync_task_context.h"
22
23 namespace DistributedDB {
CreateSyncTaskContext(const ISyncInterface & syncInterface)24 ISyncTaskContext *SingleVerSyncEngine::CreateSyncTaskContext(const ISyncInterface &syncInterface)
25 {
26 SingleVerSyncTaskContext *context = nullptr;
27 switch (syncInterface.GetInterfaceType()) {
28 case ISyncInterface::SYNC_SVD:
29 context = new (std::nothrow) SingleVerKvSyncTaskContext();
30 break;
31 #ifdef RELATIONAL_STORE
32 case ISyncInterface::SYNC_RELATION:
33 context = new (std::nothrow) SingleVerRelationalSyncTaskContext();
34 break;
35 #endif
36 default:
37 break;
38 }
39
40 if (context == nullptr) {
41 LOGE("[SingleVerSyncEngine][CreateSyncTaskContext] create failed, may be out of memory");
42 return nullptr;
43 }
44 context->SetSyncRetry(GetSyncRetry());
45 context->EnableClearRemoteStaleData(needClearRemoteStaleData_);
46 context->SetSubscribeManager(subManager_);
47 return context;
48 }
49
EnableClearRemoteStaleData(bool enable)50 void SingleVerSyncEngine::EnableClearRemoteStaleData(bool enable)
51 {
52 LOGI("[SingleVerSyncEngine][EnableClearRemoteStaleData] enabled %d", enable);
53 needClearRemoteStaleData_ = enable;
54 std::unique_lock<std::mutex> lock(contextMapLock_);
55 for (auto &iter : syncTaskContextMap_) {
56 auto context = static_cast<SingleVerSyncTaskContext *>(iter.second);
57 if (context != nullptr) { // LCOV_EXCL_BR_LINE
58 context->EnableClearRemoteStaleData(enable);
59 }
60 }
61 }
62
StartAutoSubscribeTimer(const ISyncInterface & syncInterface)63 int SingleVerSyncEngine::StartAutoSubscribeTimer(const ISyncInterface &syncInterface)
64 {
65 if (syncInterface.IsSupportSubscribe() == -E_NOT_SUPPORT) {
66 LOGI("[StartAutoSubscribeTimer] no need to start subscribe timer");
67 return E_OK;
68 }
69
70 std::lock_guard<std::mutex> lockGuard(timerLock_);
71 if (subscribeTimerId_ > 0) {
72 LOGI("[SingleSyncEngine] subscribeTimerId is already set");
73 return -E_INTERNAL_ERROR;
74 }
75 TimerId timerId = 0;
76 TimerAction timeOutCallback = [this](TimerId id) { return SubscribeTimeOut(id); };
77 int errCode = RuntimeContext::GetInstance()->SetTimer(SUBSCRIBE_TRIGGER_TIME_OUT, timeOutCallback, nullptr,
78 timerId);
79 if (errCode != E_OK) {
80 return errCode;
81 }
82 subscribeTimerId_ = timerId;
83 LOGI("[SingleSyncEngine] start auto subscribe timerId=%" PRIu64 " finished", timerId);
84 return errCode;
85 }
86
StopAutoSubscribeTimer()87 void SingleVerSyncEngine::StopAutoSubscribeTimer()
88 {
89 TimerId subscribeTimerId = 0u;
90 {
91 std::lock_guard<std::mutex> lockGuard(timerLock_);
92 if (subscribeTimerId_ == 0) {
93 return;
94 }
95 subscribeTimerId = subscribeTimerId_;
96 subscribeTimerId_ = 0;
97 }
98 RuntimeContext::GetInstance()->RemoveTimer(subscribeTimerId, true);
99 LOGI("[SingleSyncEngine] stop auto subscribe timerId=%" PRIu64 " finished", subscribeTimerId);
100 }
101
SubscribeTimeOut(TimerId id)102 int SingleVerSyncEngine::SubscribeTimeOut(TimerId id)
103 {
104 if (!queryAutoSyncCallback_) { // LCOV_EXCL_BR_LINE
105 return E_OK;
106 }
107 std::lock_guard<std::mutex> lockGuard(timerLock_);
108 std::map<std::string, std::vector<QuerySyncObject>> allSyncQueries;
109 GetAllUnFinishSubQueries(allSyncQueries);
110 if (allSyncQueries.empty()) { // LCOV_EXCL_BR_LINE
111 return E_OK;
112 }
113 for (const auto &item : allSyncQueries) {
114 for (const auto &query : item.second) {
115 InternalSyncParma param;
116 GetSubscribeSyncParam(item.first, query, param);
117 queryAutoSyncCallback_(param);
118 }
119 }
120 return E_OK;
121 }
122 DEFINE_OBJECT_TAG_FACILITIES(SingleVerSyncEngine);
123 } // namespace DistributedDB