1 /*
2 * Copyright (c) 2023 Shenzhen Kaihong Digital Industry Development 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 "event_manager.h"
17 #include "common/common_macro.h"
18 #include "common/sharing_log.h"
19 #include "event_base.h"
20 #include "magic_enum.hpp"
21
22 namespace OHOS {
23 namespace Sharing {
24 constexpr uint32_t MAX_SHARING_EVENT_NUM = 5000;
25
EventManager()26 EventManager::EventManager()
27 {
28 SHARING_LOGD("trace.");
29 }
30
~EventManager()31 EventManager::~EventManager()
32 {
33 SHARING_LOGD("trace.");
34 }
35
Init()36 int32_t EventManager::Init()
37 {
38 SHARING_LOGD("trace.");
39 SetMaxTaskNum(30); // 30: thread numbers
40 SetTimeoutInterval(2000); // 2000: timeout
41 return 0;
42 }
43
StartEventLoop()44 int32_t EventManager::StartEventLoop()
45 {
46 SHARING_LOGD("trace.");
47 int32_t ret = Start(GetMaxTaskNum());
48 if (ret != 0) {
49 return ret;
50 }
51
52 eventThread_ = std::make_unique<std::thread>(&EventManager::ProcessEvent, this);
53 RETURN_INVALID_IF_NULL(eventThread_);
54 std::string name = "eventmgr";
55 pthread_setname_np(eventThread_->native_handle(), name.c_str());
56 return 0;
57 }
58
StopEventLoop()59 void EventManager::StopEventLoop()
60 {
61 SHARING_LOGD("trace.");
62 Stop();
63 eventThread_->join();
64 eventThread_.reset();
65 eventThread_ = nullptr;
66 }
67
AddListener(std::shared_ptr<EventListener> listener)68 int32_t EventManager::AddListener(std::shared_ptr<EventListener> listener)
69 {
70 SHARING_LOGD("trace.");
71 RETURN_INVALID_IF_NULL(listener);
72 SHARING_LOGD("classtype %{public}d.", listener->GetListenerClassType());
73 std::unique_lock<std::mutex> locker(mutex_);
74 auto it = listeners_.find(listener->GetListenerClassType());
75 if (it != listeners_.end()) {
76 it->second.emplace_back(listener);
77 } else {
78 auto newClassList = std::pair<ClassType, std::list<std::shared_ptr<EventListener>>>();
79 newClassList.first = listener->GetListenerClassType();
80 newClassList.second.emplace_back(listener);
81 listeners_.emplace(newClassList);
82 }
83
84 SHARING_LOGD("listeners type count %{public}zu.", listeners_.size());
85 return 0;
86 }
87
DelListener(std::shared_ptr<EventListener> listener)88 int32_t EventManager::DelListener(std::shared_ptr<EventListener> listener)
89 {
90 SHARING_LOGD("trace.");
91 (void)listener;
92 return 0;
93 }
94
DrainAllListeners()95 int32_t EventManager::DrainAllListeners()
96 {
97 SHARING_LOGD("trace.");
98 std::unique_lock<std::mutex> locker(mutex_);
99 listeners_.clear();
100 return 0;
101 }
102
PushEvent(const SharingEvent & event)103 int32_t EventManager::PushEvent(const SharingEvent &event)
104 {
105 SHARING_LOGD("trace.");
106 RETURN_INVALID_IF_NULL(event.eventMsg);
107 SHARING_LOGI("push a async event, type: %{public}u %{public}s.", event.eventMsg->type,
108 std::string(magic_enum::enum_name(event.eventMsg->type)).c_str());
109 std::unique_lock<std::mutex> locker(mutex_);
110 if (events_.size() >= MAX_SHARING_EVENT_NUM) {
111 SHARING_LOGE("events size excced the limit");
112 return -1;
113 }
114 events_.emplace(event);
115 hasEvent_.notify_one();
116 return 0;
117 }
118
PushSyncEvent(const SharingEvent & event)119 int32_t EventManager::PushSyncEvent(const SharingEvent &event)
120 {
121 SHARING_LOGD("trace.");
122 RETURN_INVALID_IF_NULL(event.eventMsg);
123 SHARING_LOGI("push a sync event, type: %{public}u %{public}s.", event.eventMsg->type,
124 std::string(magic_enum::enum_name(event.eventMsg->type)).c_str());
125
126 for (auto listenMap : listeners_) {
127 if (listenMap.first == event.listenerType) {
128 SHARING_LOGD("find Listener type %{public}d %{public}s.", event.eventMsg->type,
129 std::string(magic_enum::enum_name(event.eventMsg->type)).c_str());
130 auto listener = listenMap.second.front();
131 std::packaged_task<BindedTask> task(std::bind(&EventListener::OnEvent, listener, event));
132 auto future = task.get_future();
133 PushTask(task);
134 if (future.wait_for(timeoutInterval_) == std::future_status::ready) {
135 SHARING_LOGD("task dispatched success %{public}s.",
136 std::string(magic_enum::enum_name(event.eventMsg->type)).c_str());
137 return future.get();
138 } else {
139 SHARING_LOGW("task timeout %{public}s.",
140 std::string(magic_enum::enum_name(event.eventMsg->type)).c_str());
141 return -1;
142 }
143 }
144 }
145
146 return 0;
147 }
148
ProcessEvent()149 void EventManager::ProcessEvent()
150 {
151 SHARING_LOGD("trace.");
152 std::unique_lock<std::mutex> locker(mutex_);
153 while (isRunning_) {
154 if (events_.empty()) {
155 hasEvent_.wait(locker);
156 } else {
157 auto event = events_.front();
158 events_.pop();
159 for (auto &listenMap : listeners_) {
160 if (listenMap.first == event.listenerType) {
161 for (auto &listener : listenMap.second) {
162 SHARING_LOGD("task dispatched success.");
163 std::packaged_task<BindedTask> task(std::bind(&EventListener::OnEvent, listener, event));
164 PushTask(task);
165 }
166 break;
167 }
168 }
169 }
170 }
171 }
172
173 } // namespace Sharing
174 } // namespace OHOS