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 "single_ver_sync_task_context.h"
19 #include "single_ver_kv_sync_task_context.h"
20 #include "single_ver_relational_sync_task_context.h"
21 #include "log_print.h"
22
23 namespace DistributedDB {
CreateSyncTaskContext()24 ISyncTaskContext *SingleVerSyncEngine::CreateSyncTaskContext()
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) {
58 context->EnableClearRemoteStaleData(enable);
59 }
60 }
61 }
62
StartAutoSubscribeTimer()63 int SingleVerSyncEngine::StartAutoSubscribeTimer()
64 {
65 std::lock_guard<std::mutex> lockGuard(timerLock_);
66 if (subscribeTimerId_ > 0) {
67 LOGI("[SingleSyncEngine] subscribeTimerId is already set");
68 return -E_INTERNAL_ERROR;
69 }
70 TimerId timerId = 0;
71 TimerAction timeOutCallback = std::bind(&SingleVerSyncEngine::SubscribeTimeOut, this, std::placeholders::_1);
72 int errCode = RuntimeContext::GetInstance()->SetTimer(SUBSCRIBE_TRIGGER_TIME_OUT, timeOutCallback, nullptr,
73 timerId);
74 if (errCode != E_OK) {
75 return errCode;
76 }
77 subscribeTimerId_ = timerId;
78 LOGI("[SingleSyncEngine] start auto subscribe timerId=%" PRIu64 " finished", timerId);
79 return errCode;
80 }
81
StopAutoSubscribeTimer()82 void SingleVerSyncEngine::StopAutoSubscribeTimer()
83 {
84 std::lock_guard<std::mutex> lockGuard(timerLock_);
85 if (subscribeTimerId_ == 0) {
86 return;
87 }
88 LOGI("[SingleSyncEngine] stop auto subscribe timerId=%" PRIu64 " finished", subscribeTimerId_);
89 RuntimeContext::GetInstance()->RemoveTimer(subscribeTimerId_);
90 subscribeTimerId_ = 0;
91 }
92
SubscribeTimeOut(TimerId id)93 int SingleVerSyncEngine::SubscribeTimeOut(TimerId id)
94 {
95 if (!queryAutoSyncCallback_) {
96 return E_OK;
97 }
98 std::lock_guard<std::mutex> lockGuard(timerLock_);
99 std::map<std::string, std::vector<QuerySyncObject>> allSyncQueries;
100 GetAllUnFinishSubQueries(allSyncQueries);
101 LOGI("[SingleVerSyncEngine] SubscribeTimeOut,size=%zu", allSyncQueries.size());
102 if (allSyncQueries.size() == 0) {
103 LOGI("no need to trigger auto subscribe");
104 return E_OK;
105 }
106 for (const auto &item : allSyncQueries) {
107 for (const auto &query : item.second) {
108 InternalSyncParma param;
109 GetSubscribeSyncParam(item.first, query, param);
110 queryAutoSyncCallback_(param);
111 }
112 }
113 return E_OK;
114 }
115
SetIsNeedResetAbilitySync(const std::string & deviceId,bool isNeedReset)116 void SingleVerSyncEngine::SetIsNeedResetAbilitySync(const std::string &deviceId, bool isNeedReset)
117 {
118 ISyncTaskContext *context = GetSyncTaskContextAndInc(deviceId);
119 if (context != nullptr) {
120 context->SetIsNeedResetAbilitySync(isNeedReset);
121 RefObject::DecObjRef(context);
122 }
123 }
124 DEFINE_OBJECT_TAG_FACILITIES(SingleVerSyncEngine);
125 } // namespace DistributedDB