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 CAMERA_LOGE("message is nullptr");
70 return;
71 }
72
73 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
74 std::unique_lock<std::mutex> l(lock_);
75 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
76 auto it = messageBox_.find(message->GetMessageType());
77 if (it == messageBox_.end()) {
78 messageBox_[message->GetMessageType()] = {{message}};
79 if (message->GetPeerMessageCount() == 1) {
80 wakeup_ = true;
81 cv_.notify_one();
82 }
83 } else {
84 bool isPeerMessage = false;
85 for (auto& mit : it->second) {
86 if (mit.empty()) {
87 continue;
88 }
89 if (mit[0] == nullptr) {
90 CAMERA_LOGE("unknown error, message is null");
91 continue;
92 }
93 if (message->GetTimestamp() == mit[0]->GetTimestamp() &&
94 message->GetMessageType() == mit[0]->GetMessageType()) {
95 mit.emplace_back(message);
96 if (mit.size() == mit[0]->GetPeerMessageCount()) {
97 wakeup_ = true;
98 cv_.notify_one();
99 }
100 isPeerMessage = true;
101 break;
102 }
103 }
104 if (!isPeerMessage) {
105 MessageGroup mg = {message};
106 it->second.emplace_back(mg);
107 if (message->GetPeerMessageCount() == 1) {
108 wakeup_ = true;
109 cv_.notify_one();
110 }
111 }
112 }
113 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
114
115 return;
116 }
117
StartProcess()118 RetCode CaptureMessageOperator::StartProcess()
119 {
120 running_ = true;
121 messageHandler_ = std::make_unique<std::thread>([this]() {
122 prctl(PR_SET_NAME, "MessageOperator");
123 while (running_) {
124 HandleMessage();
125 }
126 });
127 if (messageHandler_ == nullptr) {
128 return RC_ERROR;
129 }
130 return RC_OK;
131 }
132
HandleMessage()133 void CaptureMessageOperator::HandleMessage()
134 {
135 {
136 std::unique_lock<std::mutex> l(lock_);
137 cv_.wait(l, [this] { return !running_ || wakeup_; });
138 wakeup_ = false;
139 }
140
141 if (!running_) {
142 return;
143 }
144
145 std::list<MessageGroup> messages = {};
146 {
147 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
148 std::unique_lock<std::mutex> l(lock_);
149 for (auto& lit : messageBox_) {
150 for (auto& vit : lit.second) {
151 if (vit.empty()) {
152 continue;
153 }
154
155 if (vit.size() == vit[0]->GetPeerMessageCount()) {
156 messages.emplace_back(vit);
157 vit.clear();
158 }
159 }
160 for (auto it = lit.second.begin(); it != lit.second.end();) {
161 if (it->empty()) {
162 it = lit.second.erase(it);
163 continue;
164 }
165 it++;
166 }
167 }
168 CAMERA_LOGV("%{public}s, %{public}d, enter", __FUNCTION__, __LINE__);
169 }
170 for (auto it = messages.begin(); it != messages.end();) {
171 messageOperator_(*it);
172 it = messages.erase(it);
173 }
174 return;
175 }
176 }
177 // namespace OHOS::Camera
178