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) {
50 (void) msg;
51 return DeviceManager::CalculateLen();
52 };
53 // LocalDataChanged has no dataPct
54 func.serializeFunc = [](uint8_t *buffer, uint32_t length, const Message *inMsg) {
55 (void) buffer;
56 (void) length;
57 (void) inMsg;
58 return E_OK;
59 };
60 func.deserializeFunc = [](const uint8_t *buffer, uint32_t length, Message *inMsg) {
61 (void) buffer;
62 (void) length;
63 (void) inMsg;
64 return E_OK;
65 };
66 return MessageTransform::RegTransformFunction(LOCAL_DATA_CHANGED, func);
67 }
68
69 // Initialize the DeviceManager
Initialize(ICommunicator * communicator,const std::function<void (std::string)> & onlineCallback,const std::function<void (std::string)> & offlineCallback)70 int DeviceManager::Initialize(ICommunicator *communicator, const std::function<void(std::string)> &onlineCallback,
71 const std::function<void(std::string)> &offlineCallback)
72 {
73 if (communicator == nullptr) {
74 return -E_INVALID_ARGS;
75 }
76 RefObject::IncObjRef(communicator);
77 communicator_ = communicator;
78 RegDeviceOnLineCallBack(onlineCallback);
79 RegDeviceOffLineCallBack(offlineCallback);
80 return E_OK;
81 }
82
RegDeviceOnLineCallBack(const std::function<void (std::string)> & callback)83 void DeviceManager::RegDeviceOnLineCallBack(const std::function<void(std::string)> &callback)
84 {
85 onlineCallback_ = callback;
86 }
87
RegDeviceOffLineCallBack(const std::function<void (std::string)> & callback)88 void DeviceManager::RegDeviceOffLineCallBack(const std::function<void(std::string)> &callback)
89 {
90 offlineCallback_ = callback;
91 }
92
OnDeviceConnectCallback(const std::string & targetDev,bool isConnect)93 void DeviceManager::OnDeviceConnectCallback(const std::string &targetDev, bool isConnect)
94 {
95 LOGD("[DeviceManager] DeviceConnectCallback dev = %s{private}, status = %d", targetDev.c_str(), isConnect);
96 if (targetDev.empty()) {
97 LOGE("[DeviceManager] DeviceConnectCallback invalid device!");
98 }
99 if (isConnect) {
100 {
101 std::lock_guard<std::mutex> lockOnline(devicesLock_);
102 devices_.insert(targetDev);
103 }
104 if (onlineCallback_) {
105 onlineCallback_(targetDev);
106 LOGD("[DeviceManager] DeviceConnectCallback call online callback");
107 }
108 } else {
109 {
110 std::lock_guard<std::mutex> lockOffline(devicesLock_);
111 devices_.erase(targetDev);
112 }
113 if (offlineCallback_) {
114 offlineCallback_(targetDev);
115 LOGD("[DeviceManager] DeviceConnectCallback call offline callback");
116 }
117 }
118 }
119
GetOnlineDevices(std::vector<std::string> & devices) const120 void DeviceManager::GetOnlineDevices(std::vector<std::string> &devices) const
121 {
122 std::lock_guard<std::mutex> lock(devicesLock_);
123 devices.assign(devices_.begin(), devices_.end());
124 }
125
SendBroadCast(uint32_t msgId)126 int DeviceManager::SendBroadCast(uint32_t msgId)
127 {
128 if (msgId == LOCAL_DATA_CHANGED) {
129 return SendLocalDataChanged();
130 }
131 LOGE("[DeviceManager] invalid BroadCast msgId:%u", msgId);
132 return -E_INVALID_ARGS;
133 }
134
SendLocalDataChanged()135 int DeviceManager::SendLocalDataChanged()
136 {
137 PerformanceAnalysis *performance = PerformanceAnalysis::GetInstance();
138 if (performance != nullptr) {
139 performance->StepTimeRecordStart(MV_TEST_RECORDS::RECORD_SEND_LOCAL_DATA_CHANGED_TO_COMMIT_REQUEST_RECV);
140 }
141 std::vector<std::string> copyDevices;
142 GetOnlineDevices(copyDevices);
143 if (copyDevices.empty()) {
144 LOGI("[DeviceManager] no device online to SendLocalDataChanged!");
145 }
146 for (const auto &deviceId : copyDevices) {
147 Message *msg = new (std::nothrow) Message();
148 if (msg == nullptr) {
149 LOGE("[DeviceManager] Message alloc failed when SendBroadCast!");
150 return -E_OUT_OF_MEMORY;
151 }
152 msg->SetMessageId(LOCAL_DATA_CHANGED);
153 msg->SetTarget(deviceId);
154 SendConfig conf = {false, false, SEND_TIME_OUT, {}};
155 int errCode = communicator_->SendMessage(deviceId, msg, conf);
156 if (errCode != E_OK) {
157 LOGE("[DeviceManager] SendLocalDataChanged to dev %s{private} failed. err %d",
158 deviceId.c_str(), errCode);
159 delete msg;
160 msg = nullptr;
161 }
162 }
163 return E_OK;
164 }
165
IsDeviceOnline(const std::string & deviceId) const166 bool DeviceManager::IsDeviceOnline(const std::string &deviceId) const
167 {
168 std::lock_guard<std::mutex> lock(devicesLock_);
169 auto iter = std::find(devices_.begin(), devices_.end(), deviceId);
170 return (iter != devices_.end());
171 }
172 } // namespace DistributedDB