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