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