1 /*
2 * Copyright (c) 2022 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 "app_pipe_mgr.h"
17
18 namespace OHOS {
19 namespace ObjectStore {
20 static const int MAX_TRANSFER_SIZE = 1024 * 1024 * 5;
StartWatchDataChange(const AppDataChangeListener * observer,const PipeInfo & pipeInfo)21 Status AppPipeMgr::StartWatchDataChange(const AppDataChangeListener *observer, const PipeInfo &pipeInfo)
22 {
23 LOG_INFO("begin");
24 if (observer == nullptr || pipeInfo.pipeId.empty()) {
25 LOG_ERROR("argument invalid");
26 return Status::INVALID_ARGUMENT;
27 }
28 std::lock_guard<std::mutex> lock(dataBusMapMutex_);
29 auto it = dataBusMap_.find(pipeInfo.pipeId);
30 if (it == dataBusMap_.end()) {
31 LOG_ERROR("pipeid not found");
32 return Status::ERROR;
33 }
34 LOG_INFO("end");
35 return it->second->StartWatchDataChange(observer, pipeInfo);
36 }
37
38 // stop DataChangeListener to watch data change;
StopWatchDataChange(const AppDataChangeListener * observer,const PipeInfo & pipeInfo)39 Status AppPipeMgr::StopWatchDataChange(const AppDataChangeListener *observer, const PipeInfo &pipeInfo)
40 {
41 LOG_INFO("begin");
42 if (observer == nullptr || pipeInfo.pipeId.empty()) {
43 LOG_ERROR("argument invalid");
44 return Status::INVALID_ARGUMENT;
45 }
46 std::lock_guard<std::mutex> lock(dataBusMapMutex_);
47 auto it = dataBusMap_.find(pipeInfo.pipeId);
48 if (it == dataBusMap_.end()) {
49 LOG_ERROR("pipeid not found");
50 return Status::ERROR;
51 }
52 LOG_INFO("end");
53 return it->second->StopWatchDataChange(observer, pipeInfo);
54 }
55
56 // Send data to other device, function will be called back after sent to notify send result.
SendData(const PipeInfo & pipeInfo,const DeviceId & deviceId,const uint8_t * ptr,int size,const MessageInfo & info)57 Status AppPipeMgr::SendData(
58 const PipeInfo &pipeInfo, const DeviceId &deviceId, const uint8_t *ptr, int size, const MessageInfo &info)
59 {
60 if (size > MAX_TRANSFER_SIZE || size <= 0 || ptr == nullptr || pipeInfo.pipeId.empty()
61 || deviceId.deviceId.empty()) {
62 LOG_WARN("Input is invalid, maxSize:%{public}d, current size:%{public}d", MAX_TRANSFER_SIZE, size);
63 return Status::ERROR;
64 }
65 LOG_DEBUG("pipeInfo:%{public}s ,size:%{public}d", pipeInfo.pipeId.c_str(), size);
66 std::shared_ptr<AppPipeHandler> appPipeHandler;
67 {
68 std::lock_guard<std::mutex> lock(dataBusMapMutex_);
69 auto it = dataBusMap_.find(pipeInfo.pipeId);
70 if (it == dataBusMap_.end()) {
71 LOG_WARN("pipeInfo:%{public}s not found", pipeInfo.pipeId.c_str());
72 return Status::KEY_NOT_FOUND;
73 }
74 appPipeHandler = it->second;
75 }
76 return appPipeHandler->SendData(pipeInfo, deviceId, ptr, size, info);
77 }
78
79 // start server
Start(const PipeInfo & pipeInfo)80 Status AppPipeMgr::Start(const PipeInfo &pipeInfo)
81 {
82 if (pipeInfo.pipeId.empty()) {
83 LOG_WARN("Start Failed, pipeInfo is empty.");
84 return Status::INVALID_ARGUMENT;
85 }
86 std::lock_guard<std::mutex> lock(dataBusMapMutex_);
87 auto it = dataBusMap_.find(pipeInfo.pipeId);
88 if (it != dataBusMap_.end()) {
89 LOG_WARN("repeated start, pipeInfo:%{public}s.", pipeInfo.pipeId.c_str());
90 return Status::REPEATED_REGISTER;
91 }
92 LOG_DEBUG("Start pipeInfo:%{public}s ", pipeInfo.pipeId.c_str());
93 auto handler = std::make_shared<AppPipeHandler>(pipeInfo);
94 if (handler == nullptr) {
95 LOG_WARN("pipeInfo:%{public}s. new failed", pipeInfo.pipeId.c_str());
96 return Status::ILLEGAL_STATE;
97 }
98 int ret = handler->CreateSessionServer(pipeInfo.pipeId);
99 if (ret != 0) {
100 LOG_WARN("Start pipeInfo:%{public}s, failed ret:%{public}d.", pipeInfo.pipeId.c_str(), ret);
101 return Status::ILLEGAL_STATE;
102 }
103
104 dataBusMap_.insert(std::pair<std::string, std::shared_ptr<AppPipeHandler>>(pipeInfo.pipeId, handler));
105 return Status::SUCCESS;
106 }
107
108 // stop server
Stop(const PipeInfo & pipeInfo)109 Status AppPipeMgr::Stop(const PipeInfo &pipeInfo)
110 {
111 std::shared_ptr<AppPipeHandler> appPipeHandler;
112 {
113 std::lock_guard<std::mutex> lock(dataBusMapMutex_);
114 auto it = dataBusMap_.find(pipeInfo.pipeId);
115 if (it == dataBusMap_.end()) {
116 LOG_WARN("pipeInfo:%{public}s not found", pipeInfo.pipeId.c_str());
117 return Status::KEY_NOT_FOUND;
118 }
119 appPipeHandler = it->second;
120 int ret = appPipeHandler->RemoveSessionServer(pipeInfo.pipeId);
121 if (ret != 0) {
122 LOG_WARN("Stop pipeInfo:%{public}s ret:%{public}d.", pipeInfo.pipeId.c_str(), ret);
123 return Status::ERROR;
124 }
125 dataBusMap_.erase(pipeInfo.pipeId);
126 return Status::SUCCESS;
127 }
128 return Status::KEY_NOT_FOUND;
129 }
130
IsSameStartedOnPeer(const struct PipeInfo & pipeInfo,const struct DeviceId & peer)131 bool AppPipeMgr::IsSameStartedOnPeer(const struct PipeInfo &pipeInfo, const struct DeviceId &peer)
132 {
133 LOG_INFO("start");
134 if (pipeInfo.pipeId.empty() || peer.deviceId.empty()) {
135 LOG_ERROR("pipeId or deviceId is empty. Return false.");
136 return false;
137 }
138 LOG_INFO("pipeInfo == [%{public}s]", pipeInfo.pipeId.c_str());
139 std::shared_ptr<AppPipeHandler> appPipeHandler;
140 {
141 std::lock_guard<std::mutex> lock(dataBusMapMutex_);
142 auto it = dataBusMap_.find(pipeInfo.pipeId);
143 if (it == dataBusMap_.end()) {
144 LOG_ERROR("pipeInfo:%{public}s not found. Return false.", pipeInfo.pipeId.c_str());
145 return false;
146 }
147 appPipeHandler = it->second;
148 }
149 return appPipeHandler->IsSameStartedOnPeer(pipeInfo, peer);
150 }
151 } // namespace ObjectStore
152 } // namespace OHOS
153