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 "device_manager.h"
17
18 #include <algorithm>
19
20 #include "message_transform.h"
21 #include "parcel.h"
22 #include "db_errno.h"
23 #include "message.h"
24 #include "log_print.h"
25 #include "performance_analysis.h"
26 #include "sync_types.h"
27
28 namespace DistributedDB {
DeviceManager()29 DeviceManager::DeviceManager() : communicator_(nullptr)
30 {
31 }
32
~DeviceManager()33 DeviceManager::~DeviceManager()
34 {
35 if (communicator_ != nullptr) {
36 RefObject::DecObjRef(communicator_);
37 communicator_ = nullptr;
38 }
39 }
40
CalculateLen()41 uint32_t DeviceManager::CalculateLen()
42 {
43 return Parcel::GetUInt64Len();
44 }
45
RegisterTransformFunc()46 int DeviceManager::RegisterTransformFunc()
47 {
48 TransformFunc func;
49 func.computeFunc = [](const Message *msg) { return DeviceManager::CalculateLen(); };
50 // LocalDataChanged has no dataPct
51 func.serializeFunc = [](uint8_t *buffer, uint32_t length, const Message *inMsg) { return E_OK; };
52 func.deserializeFunc = [](const uint8_t *buffer, uint32_t length, Message *inMsg) { return E_OK; };
53 return MessageTransform::RegTransformFunction(LOCAL_DATA_CHANGED, func);
54 }
55
56 // Initialize the DeviceManager
Initialize(ICommunicator * communicator,const std::function<void (std::string)> & onlineCallback,const std::function<void (std::string)> & offlineCallback)57 int DeviceManager::Initialize(ICommunicator *communicator, const std::function<void(std::string)> &onlineCallback,
58 const std::function<void(std::string)> &offlineCallback)
59 {
60 if (communicator == nullptr) {
61 return -E_INVALID_ARGS;
62 }
63 RefObject::IncObjRef(communicator);
64 communicator_ = communicator;
65 RegDeviceOnLineCallBack(onlineCallback);
66 RegDeviceOffLineCallBack(offlineCallback);
67 return E_OK;
68 }
69
RegDeviceOnLineCallBack(const std::function<void (std::string)> & callback)70 void DeviceManager::RegDeviceOnLineCallBack(const std::function<void(std::string)> &callback)
71 {
72 onlineCallback_ = callback;
73 }
74
RegDeviceOffLineCallBack(const std::function<void (std::string)> & callback)75 void DeviceManager::RegDeviceOffLineCallBack(const std::function<void(std::string)> &callback)
76 {
77 offlineCallback_ = callback;
78 }
79
OnDeviceConnectCallback(const std::string & targetDev,bool isConnect)80 void DeviceManager::OnDeviceConnectCallback(const std::string &targetDev, bool isConnect)
81 {
82 LOGD("[DeviceManager] DeviceConnectCallback dev = %s{private}, status = %d", targetDev.c_str(), isConnect);
83 if (targetDev.empty()) {
84 LOGE("[DeviceManager] DeviceConnectCallback invalid device!");
85 }
86 if (isConnect) {
87 {
88 std::lock_guard<std::mutex> lockOnline(devicesLock_);
89 devices_.insert(targetDev);
90 }
91 if (onlineCallback_) {
92 onlineCallback_(targetDev);
93 LOGD("[DeviceManager] DeviceConnectCallback call online callback");
94 }
95 } else {
96 {
97 std::lock_guard<std::mutex> lockOffline(devicesLock_);
98 devices_.erase(targetDev);
99 }
100 if (offlineCallback_) {
101 offlineCallback_(targetDev);
102 LOGD("[DeviceManager] DeviceConnectCallback call offline callback");
103 }
104 }
105 }
106
GetOnlineDevices(std::vector<std::string> & devices) const107 void DeviceManager::GetOnlineDevices(std::vector<std::string> &devices) const
108 {
109 std::lock_guard<std::mutex> lock(devicesLock_);
110 devices.assign(devices_.begin(), devices_.end());
111 }
112
SendBroadCast(uint32_t msgId)113 int DeviceManager::SendBroadCast(uint32_t msgId)
114 {
115 if (msgId == LOCAL_DATA_CHANGED) {
116 return SendLocalDataChanged();
117 }
118 LOGE("[DeviceManager] invalid BroadCast msgId:%u", msgId);
119 return -E_INVALID_ARGS;
120 }
121
SendLocalDataChanged()122 int DeviceManager::SendLocalDataChanged()
123 {
124 PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
125 if (performance != nullptr) {
126 performance->StepTimeRecordStart(MV_TEST_RECORDS::RECORD_SEND_LOCAL_DATA_CHANGED_TO_COMMIT_REQUEST_RECV);
127 }
128 std::vector<std::string> copyDevices;
129 GetOnlineDevices(copyDevices);
130 if (copyDevices.empty()) {
131 LOGI("[DeviceManager] no device online to SendLocalDataChanged!");
132 }
133 for (const auto &deviceId : copyDevices) {
134 Message *msg = new (std::nothrow) Message();
135 if (msg == nullptr) {
136 LOGE("[DeviceManager] Message alloc failed when SendBroadCast!");
137 return -E_OUT_OF_MEMORY;
138 }
139 msg->SetMessageId(LOCAL_DATA_CHANGED);
140 msg->SetTarget(deviceId);
141 SendConfig conf = {false, false, SEND_TIME_OUT, {}};
142 int errCode = communicator_->SendMessage(deviceId, msg, conf);
143 if (errCode != E_OK) {
144 LOGE("[DeviceManager] SendLocalDataChanged to dev %s{private} failed. err %d",
145 deviceId.c_str(), errCode);
146 delete msg;
147 msg = nullptr;
148 }
149 }
150 return E_OK;
151 }
152
IsDeviceOnline(const std::string & deviceId) const153 bool DeviceManager::IsDeviceOnline(const std::string &deviceId) const
154 {
155 std::lock_guard<std::mutex> lock(devicesLock_);
156 auto iter = std::find(devices_.begin(), devices_.end(), deviceId);
157 return (iter != devices_.end());
158 }
159 } // namespace DistributedDB