Line data Source code
1 : /*
2 : * Copyright (C) 2023 Sky UK
3 : *
4 : * This library is free software; you can redistribute it and/or
5 : * modify it under the terms of the GNU Lesser General Public
6 : * License as published by the Free Software Foundation;
7 : * version 2.1 of the License.
8 : *
9 : * This library is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 : * Lesser General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU Lesser General Public
15 : * License along with this library; if not, write to the Free Software
16 : * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17 : */
18 :
19 : #include "MessageQueue.h"
20 : #include "GstreamerCatLog.h"
21 : #define GST_CAT_DEFAULT rialtoGStreamerCat
22 2094 : CallInEventLoopMessage::CallInEventLoopMessage(const std::function<void()> &func) : m_func(func), m_done{false} {}
23 :
24 2091 : void CallInEventLoopMessage::handle()
25 : {
26 2091 : std::unique_lock<std::mutex> lock(m_callInEventLoopMutex);
27 2091 : m_func();
28 2091 : m_done = true;
29 2091 : m_callInEventLoopCondVar.notify_all();
30 : }
31 :
32 1 : void CallInEventLoopMessage::skip()
33 : {
34 1 : std::unique_lock<std::mutex> lock(m_callInEventLoopMutex);
35 1 : m_done = true;
36 1 : m_callInEventLoopCondVar.notify_all();
37 : }
38 :
39 2092 : void CallInEventLoopMessage::wait()
40 : {
41 2092 : std::unique_lock<std::mutex> lock(m_callInEventLoopMutex);
42 6275 : m_callInEventLoopCondVar.wait(lock, [this]() { return m_done; });
43 2091 : }
44 :
45 1 : ScheduleInEventLoopMessage::ScheduleInEventLoopMessage(const std::function<void()> &func) : m_func(func) {}
46 :
47 1 : void ScheduleInEventLoopMessage::handle()
48 : {
49 1 : m_func();
50 : }
51 :
52 171 : std::shared_ptr<IMessageQueueFactory> IMessageQueueFactory::createFactory()
53 : {
54 171 : return std::make_shared<MessageQueueFactory>();
55 : }
56 :
57 322 : std::unique_ptr<IMessageQueue> MessageQueueFactory::createMessageQueue() const
58 : {
59 322 : return std::make_unique<rialto::MessageQueue>();
60 : }
61 :
62 : namespace rialto
63 : {
64 379 : MessageQueue::MessageQueue() : m_running(false), m_acceptingMessages{false} {}
65 :
66 748 : MessageQueue::~MessageQueue()
67 : {
68 379 : doStop();
69 748 : }
70 :
71 378 : void MessageQueue::start()
72 : {
73 378 : if (m_running)
74 : {
75 : // queue is running
76 1 : return;
77 : }
78 377 : m_running = true;
79 377 : m_acceptingMessages = true;
80 377 : std::thread startThread(&MessageQueue::processMessages, this);
81 377 : m_workerThread.swap(startThread);
82 : }
83 :
84 222 : void MessageQueue::stop()
85 : {
86 222 : doStop();
87 : }
88 :
89 1 : void MessageQueue::clear()
90 : {
91 1 : doClear();
92 : }
93 :
94 2150 : std::shared_ptr<Message> MessageQueue::waitForMessage()
95 : {
96 2150 : std::unique_lock<std::mutex> lock(m_mutex);
97 4113 : while (m_queue.empty())
98 : {
99 1963 : m_condVar.wait(lock);
100 : }
101 2150 : std::shared_ptr<Message> message = m_queue.front();
102 2150 : m_queue.pop_front();
103 4300 : return message;
104 2150 : }
105 :
106 1779 : bool MessageQueue::postMessage(const std::shared_ptr<Message> &msg)
107 : {
108 1779 : const std::lock_guard<std::mutex> lock(m_mutex);
109 1779 : if (!m_running || !m_acceptingMessages)
110 : {
111 3 : GST_ERROR("Message queue is not running or not accepting messages");
112 3 : return false;
113 : }
114 1776 : m_queue.push_back(msg);
115 1776 : m_condVar.notify_all();
116 :
117 1776 : return true;
118 1779 : }
119 :
120 377 : void MessageQueue::processMessages()
121 : {
122 : do
123 : {
124 2150 : std::shared_ptr<Message> message = waitForMessage();
125 2150 : message->handle();
126 2150 : } while (m_running);
127 377 : }
128 :
129 1 : bool MessageQueue::scheduleInEventLoop(const std::function<void()> &func)
130 : {
131 1 : auto message = std::make_shared<ScheduleInEventLoopMessage>(func);
132 1 : if (!postMessage(message))
133 : {
134 0 : return false;
135 : }
136 :
137 1 : return true;
138 : }
139 :
140 1773 : bool MessageQueue::callInEventLoop(const std::function<void()> &func)
141 : {
142 1773 : return callInEventLoopInternal(func);
143 : }
144 :
145 1773 : bool MessageQueue::callInEventLoopInternal(const std::function<void()> &func)
146 : {
147 1773 : if (std::this_thread::get_id() != m_workerThread.get_id())
148 : {
149 1719 : auto message = std::make_shared<CallInEventLoopMessage>(func);
150 1719 : if (!postMessage(message))
151 : {
152 2 : return false;
153 : }
154 1717 : message->wait();
155 1719 : }
156 : else
157 : {
158 54 : func();
159 : }
160 :
161 1771 : return true;
162 : }
163 :
164 601 : void MessageQueue::doStop()
165 : {
166 601 : if (!m_running)
167 : {
168 : // queue is not running
169 224 : return;
170 : }
171 377 : if (std::this_thread::get_id() == m_workerThread.get_id())
172 : {
173 2 : m_acceptingMessages = false;
174 2 : m_running = false;
175 2 : m_workerThread.detach();
176 : }
177 : else
178 : {
179 750 : auto message = std::make_shared<CallInEventLoopMessage>([this]() { m_running = false; });
180 : {
181 375 : const std::lock_guard<std::mutex> lock(m_mutex);
182 375 : m_acceptingMessages = false;
183 375 : m_queue.push_back(message);
184 375 : m_condVar.notify_all();
185 : }
186 375 : message->wait();
187 :
188 375 : if (m_workerThread.joinable())
189 375 : m_workerThread.join();
190 : }
191 :
192 377 : doClear();
193 : }
194 :
195 378 : void MessageQueue::doClear()
196 : {
197 378 : std::unique_lock<std::mutex> lock(m_mutex);
198 379 : while (!m_queue.empty())
199 : {
200 1 : m_queue.front()->skip();
201 1 : m_queue.pop_front();
202 : }
203 378 : }
204 : } // namespace rialto
|
