1 /*
2 * Copyright (c) 2021-2024 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 #define LOG_TAG "KVDBNotifierClient"
17
18 #include "kvdb_notifier_client.h"
19 #include <atomic>
20 #include <cinttypes>
21 #include <functional>
22 #include "dds_trace.h"
23 #include "dev_manager.h"
24 #include "log_print.h"
25 #include "store_util.h"
26
27 namespace OHOS {
28 namespace DistributedKv {
29 using namespace OHOS::DistributedDataDfx;
~KVDBNotifierClient()30 KVDBNotifierClient::~KVDBNotifierClient()
31 {
32 syncCallbackInfo_.Clear();
33 }
34
SyncCompleted(const std::map<std::string,Status> & results,uint64_t sequenceId)35 void KVDBNotifierClient::SyncCompleted(const std::map<std::string, Status> &results, uint64_t sequenceId)
36 {
37 DdsTrace trace(std::string(LOG_TAG "::") + std::string(__FUNCTION__), TraceSwitch::BYTRACE_ON);
38 auto finded = syncCallbackInfo_.Find(sequenceId);
39 if (finded.first) {
40 finded.second->SyncCompleted(results);
41 finded.second->SyncCompleted(results, sequenceId);
42 DeleteSyncCallback(sequenceId);
43 }
44 }
45
SyncCompleted(uint64_t seqNum,ProgressDetail && detail)46 void KVDBNotifierClient::SyncCompleted(uint64_t seqNum, ProgressDetail &&detail)
47 {
48 DdsTrace trace(std::string(LOG_TAG "::") + std::string(__FUNCTION__), TraceSwitch::BYTRACE_ON);
49 cloudSyncCallbacks_.ComputeIfPresent(seqNum, [&detail](const auto &key, const AsyncDetail &callback) {
50 auto finished = (detail.progress == SYNC_FINISH);
51 ZLOGD("Sync complete, seqNum%{public}" PRIu64, key);
52 if (callback != nullptr) {
53 callback(std::move(detail));
54 }
55 return !finished;
56 });
57 }
58
OnRemoteChange(const std::map<std::string,bool> & mask,int32_t dataType)59 void KVDBNotifierClient::OnRemoteChange(const std::map<std::string, bool> &mask, int32_t dataType)
60 {
61 ZLOGD("remote changed mask:%{public}zu dataType:%{public}d", mask.size(), dataType);
62 DataType type = static_cast<DataType>(dataType);
63 for (const auto &[device, changed] : mask) {
64 auto clientUuid = DevManager::GetInstance().ToUUID(device);
65 if (clientUuid.empty()) {
66 continue;
67 }
68 if (!remotes_.Contains(clientUuid)) {
69 remotes_.InsertOrAssign(clientUuid, std::make_pair<bool, bool>(true, true));
70 }
71 remotes_.Compute(clientUuid, [isChange = changed, type](const auto &key, auto &value) -> bool {
72 switch (type) {
73 case DataType::TYPE_STATICS:
74 value.first = isChange;
75 break;
76 case DataType::TYPE_DYNAMICAL:
77 value.second = isChange;
78 break;
79 default:
80 break;
81 }
82 return true;
83 });
84 }
85 }
86
OnSwitchChange(const SwitchNotification & notification)87 void KVDBNotifierClient::OnSwitchChange(const SwitchNotification ¬ification)
88 {
89 switchObservers_.ForEachCopies(
90 [¬ification](auto &, std::shared_ptr<KvStoreObserver> &observer) {
91 observer->OnSwitchChange(notification);
92 return false;
93 });
94 }
95
IsChanged(const std::string & deviceId,DataType dataType)96 bool KVDBNotifierClient::IsChanged(const std::string &deviceId, DataType dataType)
97 {
98 auto [exist, value] = remotes_.Find(deviceId);
99 ZLOGD("exist:%{public}d, statics:%{public}d dynamic:%{public}d",
100 exist, value.first, value.second);
101 if (!exist) {
102 return true;
103 }
104 switch (dataType) {
105 case DataType::TYPE_STATICS:
106 return value.first;
107 case DataType::TYPE_DYNAMICAL:
108 return value.second;
109 default:
110 break;
111 }
112 return true;
113 }
114
AddSyncCallback(const std::shared_ptr<KvStoreSyncCallback> callback,uint64_t sequenceId)115 void KVDBNotifierClient::AddSyncCallback(
116 const std::shared_ptr<KvStoreSyncCallback> callback, uint64_t sequenceId)
117 {
118 if (callback == nullptr) {
119 ZLOGE("callback is nullptr");
120 return;
121 }
122 auto inserted = syncCallbackInfo_.Insert(sequenceId, callback);
123 if (!inserted) {
124 ZLOGE("The sequeuceId %{public}" PRIu64 "is repeat!", sequenceId);
125 }
126 }
127
DeleteSyncCallback(uint64_t sequenceId)128 void KVDBNotifierClient::DeleteSyncCallback(uint64_t sequenceId)
129 {
130 syncCallbackInfo_.Erase(sequenceId);
131 }
132
AddCloudSyncCallback(uint64_t sequenceId,const AsyncDetail & async)133 void KVDBNotifierClient::AddCloudSyncCallback(uint64_t sequenceId, const AsyncDetail &async)
134 {
135 if (async == nullptr) {
136 ZLOGE("Cloud sync callback is nullptr");
137 return;
138 }
139 cloudSyncCallbacks_.Insert(sequenceId, async);
140 }
141
DeleteCloudSyncCallback(uint64_t sequenceId)142 void KVDBNotifierClient::DeleteCloudSyncCallback(uint64_t sequenceId)
143 {
144 cloudSyncCallbacks_.Erase(sequenceId);
145 }
146
AddSwitchCallback(const std::string & appId,std::shared_ptr<KvStoreObserver> observer)147 void KVDBNotifierClient::AddSwitchCallback(const std::string &appId, std::shared_ptr<KvStoreObserver> observer)
148 {
149 if (observer == nullptr) {
150 return;
151 }
152 if (switchObservers_.Contains(uintptr_t(observer.get()))) {
153 ZLOGI("duplicate observer");
154 return;
155 }
156 switchObservers_.Insert(uintptr_t(observer.get()), observer);
157 }
158
DeleteSwitchCallback(const std::string & appId,std::shared_ptr<KvStoreObserver> observer)159 void KVDBNotifierClient::DeleteSwitchCallback(const std::string &appId, std::shared_ptr<KvStoreObserver> observer)
160 {
161 if (observer == nullptr) {
162 return;
163 }
164 switchObservers_.Erase(uintptr_t(observer.get()));
165 }
166 } // namespace DistributedKv
167 } // namespace OHOS
168