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 #include "capture_message.h"
16
17 namespace OHOS::Camera {
ICaptureMessage(int32_t streamId,int32_t captureId,uint64_t time,uint32_t count)18 ICaptureMessage::ICaptureMessage(int32_t streamId, int32_t captureId, uint64_t time, uint32_t count)
19 {
20 peerMessageCount_ = count;
21 timestamp_ = time;
22 streamId_ = streamId;
23 captureId_ = captureId;
24 }
25
GetPeerMessageCount() const26 uint32_t ICaptureMessage::GetPeerMessageCount() const
27 {
28 return peerMessageCount_;
29 }
30
GetTimestamp() const31 uint64_t ICaptureMessage::GetTimestamp() const
32 {
33 return timestamp_;
34 }
35
GetStreamId() const36 int32_t ICaptureMessage::GetStreamId() const
37 {
38 return streamId_;
39 }
40
GetCaptureId() const41 int32_t ICaptureMessage::GetCaptureId() const
42 {
43 return captureId_;
44 }
45
GetMessageType() const46 CaptureMessageType ICaptureMessage::GetMessageType() const
47 {
48 return CAPTURE_MESSAGE_TYPE_INVALID;
49 }
50
CaptureMessageOperator(MessageOperatorFunc f)51 CaptureMessageOperator::CaptureMessageOperator(MessageOperatorFunc f)
52 {
53 messageOperator_ = f;
54 }
55
~CaptureMessageOperator()56 CaptureMessageOperator::~CaptureMessageOperator()
57 {
58 running_ = false;
59 cv_.notify_one();
60 if (messageHandler_ != nullptr) {
61 messageHandler_->join();
62 }
63 messageBox_.clear();
64 }
65
SendMessage(std::shared_ptr<ICaptureMessage> & message)66 void CaptureMessageOperator::SendMessage(std::shared_ptr<ICaptureMessage>& message)
67 {
68 if (message == nullptr) {
69 return;
70 }
71
72 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
73 std::unique_lock<std::mutex> l(lock_);
74 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
75 auto it = messageBox_.find(message->GetMessageType());
76 if (it == messageBox_.end()) {
77 messageBox_[message->GetMessageType()] = {{message}};
78 if (message->GetPeerMessageCount() == 1) {
79 wakeup_ = true;
80 cv_.notify_one();
81 }
82 } else {
83 bool isPeerMessage = false;
84 for (auto& mit : it->second) {
85 if (mit.empty()) {
86 continue;
87 }
88 if (mit[0] == nullptr) {
89 CAMERA_LOGE("unknown error, message is null");
90 continue;
91 }
92 if (message->GetTimestamp() == mit[0]->GetTimestamp() &&
93 message->GetMessageType() == mit[0]->GetMessageType()) {
94 mit.emplace_back(message);
95 if (mit.size() == mit[0]->GetPeerMessageCount()) {
96 wakeup_ = true;
97 cv_.notify_one();
98 }
99 isPeerMessage = true;
100 break;
101 }
102 }
103 if (!isPeerMessage) {
104 MessageGroup mg = {message};
105 it->second.emplace_back(mg);
106 if (message->GetPeerMessageCount() == 1) {
107 wakeup_ = true;
108 cv_.notify_one();
109 }
110 }
111 }
112 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
113
114 return;
115 }
116
StartProcess()117 RetCode CaptureMessageOperator::StartProcess()
118 {
119 running_ = true;
120 messageHandler_ = std::make_unique<std::thread>([this]() {
121 prctl(PR_SET_NAME, "MessageOperator");
122 while (running_) {
123 HandleMessage();
124 }
125 });
126 if (messageHandler_ == nullptr) {
127 return RC_ERROR;
128 }
129 return RC_OK;
130 }
131
HandleMessage()132 void CaptureMessageOperator::HandleMessage()
133 {
134 {
135 std::unique_lock<std::mutex> l(lock_);
136 cv_.wait(l, [this] { return !running_ || wakeup_; });
137 wakeup_ = false;
138 }
139
140 if (!running_) {
141 return;
142 }
143
144 std::list<MessageGroup> messages = {};
145 {
146 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
147 std::unique_lock<std::mutex> l(lock_);
148 for (auto& lit : messageBox_) {
149 for (auto& vit : lit.second) {
150 if (vit.empty()) {
151 continue;
152 }
153
154 if (vit.size() == vit[0]->GetPeerMessageCount()) {
155 messages.emplace_back(vit);
156 vit.clear();
157 }
158 }
159 for (auto it = lit.second.begin(); it != lit.second.end();) {
160 if (it->empty()) {
161 it = lit.second.erase(it);
162 continue;
163 }
164 it++;
165 }
166 }
167 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
168 }
169 for (auto it = messages.begin(); it != messages.end();) {
170 messageOperator_(*it);
171 it = messages.erase(it);
172 }
173 return;
174 }
175 }
176 // namespace OHOS::Camera
177