Line data Source code
1 : /*
2 : * If not stated otherwise in this file or this component's LICENSE file the
3 : * following copyright and licenses apply:
4 : *
5 : * Copyright 2023 Sky UK
6 : *
7 : * Licensed under the Apache License, Version 2.0 (the "License");
8 : * you may not use this file except in compliance with the License.
9 : * You may obtain a copy of the License at
10 : *
11 : * http://www.apache.org/licenses/LICENSE-2.0
12 : *
13 : * Unless required by applicable law or agreed to in writing, software
14 : * distributed under the License is distributed on an "AS IS" BASIS,
15 : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 : * See the License for the specific language governing permissions and
17 : * limitations under the License.
18 : */
19 :
20 : #include "GstWebAudioPlayer.h"
21 : #include "GstDispatcherThread.h"
22 : #include "RialtoServerLogging.h"
23 : #include "WorkerThread.h"
24 : #include "tasks/webAudio/WebAudioPlayerTaskFactory.h"
25 : #include <stdexcept>
26 :
27 : namespace
28 : {
29 : constexpr uint32_t kMaxWriteBufferTimeoutMs{2000};
30 : } // namespace
31 :
32 : namespace firebolt::rialto::server
33 : {
34 : std::weak_ptr<IGstWebAudioPlayerFactory> GstWebAudioPlayerFactory::m_factory;
35 :
36 25 : std::shared_ptr<IGstWebAudioPlayerFactory> IGstWebAudioPlayerFactory::getFactory()
37 : {
38 25 : std::shared_ptr<IGstWebAudioPlayerFactory> factory = GstWebAudioPlayerFactory::m_factory.lock();
39 :
40 25 : if (!factory)
41 : {
42 : try
43 : {
44 25 : factory = std::make_shared<GstWebAudioPlayerFactory>();
45 : }
46 0 : catch (const std::exception &e)
47 : {
48 0 : RIALTO_SERVER_LOG_ERROR("Failed to create the gstreamer player factory, reason: %s", e.what());
49 : }
50 :
51 25 : GstWebAudioPlayerFactory::m_factory = factory;
52 : }
53 :
54 25 : return factory;
55 : }
56 :
57 1 : std::unique_ptr<IGstWebAudioPlayer> GstWebAudioPlayerFactory::createGstWebAudioPlayer(IGstWebAudioPlayerClient *client,
58 : const uint32_t priority)
59 : {
60 1 : std::unique_ptr<IGstWebAudioPlayer> gstPlayer;
61 :
62 : try
63 : {
64 1 : auto gstWrapperFactory = firebolt::rialto::wrappers::IGstWrapperFactory::getFactory();
65 1 : auto glibWrapperFactory = firebolt::rialto::wrappers::IGlibWrapperFactory::getFactory();
66 1 : std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> gstWrapper;
67 1 : std::shared_ptr<firebolt::rialto::wrappers::IGlibWrapper> glibWrapper;
68 1 : if ((!gstWrapperFactory) || (!(gstWrapper = gstWrapperFactory->getGstWrapper())))
69 : {
70 0 : throw std::runtime_error("Cannot create GstWrapper");
71 : }
72 1 : if ((!glibWrapperFactory) || (!(glibWrapper = glibWrapperFactory->getGlibWrapper())))
73 : {
74 0 : throw std::runtime_error("Cannot create GlibWrapper");
75 : }
76 2 : gstPlayer = std::make_unique<GstWebAudioPlayer>(client, priority, gstWrapper, glibWrapper,
77 2 : IGstInitialiser::instance(), IGstSrcFactory::getFactory(),
78 2 : std::make_unique<WebAudioPlayerTaskFactory>(client, gstWrapper,
79 : glibWrapper),
80 2 : std::make_unique<WorkerThreadFactory>(),
81 3 : std::make_unique<GstDispatcherThreadFactory>());
82 1 : }
83 0 : catch (const std::exception &e)
84 : {
85 0 : RIALTO_SERVER_LOG_ERROR("Failed to create the gstreamer player, reason: %s", e.what());
86 : }
87 :
88 1 : return gstPlayer;
89 : }
90 :
91 33 : GstWebAudioPlayer::GstWebAudioPlayer(IGstWebAudioPlayerClient *client, const uint32_t priority,
92 : const std::shared_ptr<firebolt::rialto::wrappers::IGstWrapper> &gstWrapper,
93 : const std::shared_ptr<firebolt::rialto::wrappers::IGlibWrapper> &glibWrapper,
94 : const IGstInitialiser &gstInitialiser,
95 : const std::shared_ptr<IGstSrcFactory> &gstSrcFactory,
96 : std::unique_ptr<IWebAudioPlayerTaskFactory> taskFactory,
97 : std::unique_ptr<IWorkerThreadFactory> workerThreadFactory,
98 33 : std::unique_ptr<IGstDispatcherThreadFactory> gstDispatcherThreadFactory)
99 33 : : m_gstPlayerClient(client), m_gstWrapper{gstWrapper}, m_glibWrapper{glibWrapper},
100 66 : m_taskFactory{std::move(taskFactory)}
101 : {
102 33 : RIALTO_SERVER_LOG_DEBUG("GstWebAudioPlayer is constructed.");
103 :
104 33 : gstInitialiser.waitForInitialisation();
105 :
106 33 : if ((!gstSrcFactory) || (!(m_context.gstSrc = gstSrcFactory->getGstSrc())))
107 : {
108 1 : throw std::runtime_error("Cannot create GstSrc");
109 : }
110 :
111 : // Ensure that rialtosrc has been initalised
112 32 : m_context.gstSrc->initSrc();
113 :
114 : // Start task thread
115 32 : if ((!workerThreadFactory) || (!(m_workerThread = workerThreadFactory->createWorkerThread())))
116 : {
117 1 : throw std::runtime_error("Failed to create the worker thread");
118 : }
119 :
120 31 : if (!initWebAudioPipeline(priority))
121 : {
122 12 : termWebAudioPipeline();
123 12 : resetWorkerThread();
124 12 : throw std::runtime_error("Failed to initalise the pipeline");
125 : }
126 :
127 19 : if (!gstDispatcherThreadFactory)
128 : {
129 0 : termWebAudioPipeline();
130 0 : resetWorkerThread();
131 0 : throw std::runtime_error("Failed to create the dispatcher thread");
132 : }
133 :
134 57 : m_gstDispatcherThread = gstDispatcherThreadFactory->createGstDispatcherThread(*this, m_context.pipeline, nullptr,
135 38 : m_gstWrapper, m_glibWrapper);
136 19 : if (!m_gstDispatcherThread)
137 : {
138 1 : termWebAudioPipeline();
139 1 : resetWorkerThread();
140 1 : throw std::runtime_error("Failed to create the dispatcher thread");
141 : }
142 153 : }
143 :
144 36 : GstWebAudioPlayer::~GstWebAudioPlayer()
145 : {
146 18 : RIALTO_SERVER_LOG_DEBUG("GstWebAudioPlayer is destructed.");
147 :
148 18 : m_gstDispatcherThread.reset();
149 :
150 18 : resetWorkerThread();
151 :
152 18 : termWebAudioPipeline();
153 36 : }
154 :
155 31 : bool GstWebAudioPlayer::initWebAudioPipeline(const uint32_t priority)
156 : {
157 31 : m_context.pipeline = m_gstWrapper->gstPipelineNew(std::string("webaudiopipeline" + std::to_string(priority)).c_str());
158 31 : if (!m_context.pipeline)
159 : {
160 1 : RIALTO_SERVER_LOG_ERROR("Failed to create the webaudiopipeline");
161 1 : return false;
162 : }
163 :
164 30 : RIALTO_SERVER_LOG_MIL("RialtoServer's webaudio pipeline constructed");
165 :
166 : // Create and initalise appsrc
167 30 : m_context.source = m_gstWrapper->gstElementFactoryMake("appsrc", "audsrc");
168 30 : if (!m_context.source)
169 : {
170 1 : RIALTO_SERVER_LOG_ERROR("Failed to create the appsrc");
171 1 : return false;
172 : }
173 29 : m_gstWrapper->gstAppSrcSetMaxBytes(GST_APP_SRC(m_context.source), kMaxWebAudioBytes);
174 29 : m_glibWrapper->gObjectSet(m_context.source, "format", GST_FORMAT_TIME, nullptr);
175 :
176 : // Perform sink specific initalisation
177 29 : GstPluginFeature *feature = nullptr;
178 29 : GstRegistry *reg = m_gstWrapper->gstRegistryGet();
179 29 : if (!reg)
180 : {
181 1 : RIALTO_SERVER_LOG_ERROR("Failed get the gst registry");
182 1 : return false;
183 : }
184 :
185 28 : GstElement *sink = nullptr;
186 28 : if (nullptr != (feature = m_gstWrapper->gstRegistryLookupFeature(reg, "amlhalasink")))
187 : {
188 : // LLama
189 3 : RIALTO_SERVER_LOG_INFO("Use amlhalasink");
190 3 : sink = createAmlhalaSink();
191 3 : m_gstWrapper->gstObjectUnref(feature);
192 : }
193 25 : else if (nullptr != (feature = m_gstWrapper->gstRegistryLookupFeature(reg, "rtkaudiosink")))
194 : {
195 : // XiOne
196 5 : RIALTO_SERVER_LOG_INFO("Use rtkaudiosink");
197 5 : sink = createRtkAudioSink();
198 5 : m_gstWrapper->gstObjectUnref(feature);
199 : }
200 : else
201 : {
202 20 : RIALTO_SERVER_LOG_INFO("Use autoaudiosink");
203 20 : sink = createAutoSink();
204 : }
205 :
206 28 : if (sink)
207 : {
208 25 : return linkElementsToSrc(sink);
209 : }
210 : else
211 : {
212 3 : m_gstWrapper->gstObjectUnref(m_context.source);
213 3 : m_context.source = nullptr;
214 : }
215 3 : return false;
216 : }
217 :
218 3 : GstElement *GstWebAudioPlayer::createAmlhalaSink()
219 : {
220 3 : GstElement *sink = m_gstWrapper->gstElementFactoryMake("amlhalasink", "webaudiosink");
221 3 : if (!sink)
222 : {
223 1 : RIALTO_SERVER_LOG_ERROR("Failed create the amlhalasink");
224 1 : return nullptr;
225 : }
226 2 : m_glibWrapper->gObjectSet(G_OBJECT(sink), "direct-mode", FALSE, NULL);
227 :
228 2 : return sink;
229 : }
230 :
231 5 : GstElement *GstWebAudioPlayer::createRtkAudioSink()
232 : {
233 5 : GstElement *sink = m_gstWrapper->gstElementFactoryMake("rtkaudiosink", "webaudiosink");
234 5 : if (!sink)
235 : {
236 1 : RIALTO_SERVER_LOG_ERROR("Failed create the rtkaudiosink");
237 1 : return nullptr;
238 : }
239 4 : m_glibWrapper->gObjectSet(G_OBJECT(sink), "media-tunnel", FALSE, NULL);
240 4 : m_glibWrapper->gObjectSet(G_OBJECT(sink), "audio-service", TRUE, NULL);
241 :
242 4 : return sink;
243 : }
244 :
245 20 : GstElement *GstWebAudioPlayer::createAutoSink()
246 : {
247 20 : GstElement *sink = m_gstWrapper->gstElementFactoryMake("autoaudiosink", "webaudiosink");
248 20 : if (!sink)
249 : {
250 1 : RIALTO_SERVER_LOG_ERROR("Failed create the autoaudiosink");
251 1 : return nullptr;
252 : }
253 :
254 19 : return sink;
255 : }
256 :
257 : // NOTE:-
258 : // This method hands the responsibility for the destruction of both { "sink", "m_context.source" }
259 : // over to the pipeline (or if handover wasn't possible, it will unref)
260 25 : bool GstWebAudioPlayer::linkElementsToSrc(GstElement *sink)
261 : {
262 25 : bool status{true};
263 :
264 25 : GstElement *convert{nullptr};
265 25 : GstElement *resample{nullptr};
266 25 : GstElement *volume{nullptr};
267 25 : GstElement *queue{nullptr};
268 :
269 25 : convert = m_gstWrapper->gstElementFactoryMake("audioconvert", NULL);
270 25 : if (!convert)
271 : {
272 1 : RIALTO_SERVER_LOG_ERROR("Failed create the audioconvert");
273 1 : status = false;
274 : }
275 :
276 25 : if (status)
277 : {
278 24 : resample = m_gstWrapper->gstElementFactoryMake("audioresample", NULL);
279 24 : if (!resample)
280 : {
281 1 : RIALTO_SERVER_LOG_ERROR("Failed create the audioresample");
282 1 : status = false;
283 : }
284 : }
285 :
286 25 : if (status)
287 : {
288 23 : volume = m_gstWrapper->gstElementFactoryMake("volume", NULL);
289 23 : if (!volume)
290 : {
291 1 : RIALTO_SERVER_LOG_ERROR("Failed create the volume");
292 1 : status = false;
293 : }
294 : }
295 25 : if (status)
296 : {
297 22 : queue = m_gstWrapper->gstElementFactoryMake("queue", NULL);
298 22 : if (!queue)
299 : {
300 1 : RIALTO_SERVER_LOG_ERROR("Failed create the queue");
301 1 : status = false;
302 : }
303 : else
304 : {
305 21 : constexpr guint kWebAudioQueueSize{8192};
306 21 : m_glibWrapper->gObjectSet(queue, "max-size-bytes", kWebAudioQueueSize, nullptr);
307 : }
308 : }
309 :
310 25 : std::queue<GstElement *> elementsToAdd;
311 25 : elementsToAdd.push(m_context.source);
312 25 : if (convert)
313 24 : elementsToAdd.push(convert);
314 25 : if (resample)
315 23 : elementsToAdd.push(resample);
316 25 : if (volume)
317 22 : elementsToAdd.push(volume);
318 25 : if (queue)
319 21 : elementsToAdd.push(queue);
320 25 : elementsToAdd.push(sink);
321 :
322 25 : if (status)
323 : {
324 : // Add elements to the pipeline
325 21 : GstBin *pipelineBin = GST_BIN(m_context.pipeline);
326 146 : while (!elementsToAdd.empty())
327 : {
328 126 : if (m_gstWrapper->gstBinAdd(pipelineBin, elementsToAdd.front()))
329 125 : elementsToAdd.pop();
330 : else
331 : {
332 1 : RIALTO_SERVER_LOG_ERROR("Failed to add element to the bin");
333 1 : status = false;
334 1 : break;
335 : }
336 : }
337 : }
338 :
339 25 : if (status)
340 : {
341 20 : if ((!m_gstWrapper->gstElementLink(m_context.source, convert)) ||
342 19 : (!m_gstWrapper->gstElementLink(convert, resample)) || (!m_gstWrapper->gstElementLink(resample, volume)) ||
343 39 : (!m_gstWrapper->gstElementLink(volume, queue)) || (!m_gstWrapper->gstElementLink(queue, sink)))
344 : {
345 1 : RIALTO_SERVER_LOG_ERROR("Failed to link elements");
346 1 : status = false;
347 : }
348 20 : m_context.gstVolumeElement = GST_STREAM_VOLUME(volume);
349 : }
350 :
351 25 : if (!status)
352 : {
353 : // Unref anything that wasn't added to the pipeline
354 21 : while (!elementsToAdd.empty())
355 : {
356 15 : m_gstWrapper->gstObjectUnref(elementsToAdd.front());
357 15 : elementsToAdd.pop();
358 : }
359 6 : m_context.source = nullptr;
360 : }
361 :
362 25 : return status;
363 : }
364 :
365 31 : void GstWebAudioPlayer::termWebAudioPipeline()
366 : {
367 31 : if (m_context.pipeline)
368 : {
369 30 : m_taskFactory->createStop(*this)->execute();
370 30 : GstBus *bus = m_gstWrapper->gstPipelineGetBus(GST_PIPELINE(m_context.pipeline));
371 30 : if (bus)
372 : {
373 29 : m_gstWrapper->gstBusSetSyncHandler(bus, nullptr, nullptr, nullptr);
374 29 : m_gstWrapper->gstObjectUnref(bus);
375 : }
376 :
377 30 : m_gstWrapper->gstObjectUnref(m_context.pipeline);
378 : }
379 31 : RIALTO_SERVER_LOG_MIL("RialtoServer's webaudio pipeline terminated.");
380 : }
381 :
382 31 : void GstWebAudioPlayer::resetWorkerThread()
383 : {
384 31 : m_workerThread->enqueueTask(m_taskFactory->createShutdown(*this));
385 31 : m_workerThread->join();
386 31 : m_workerThread.reset();
387 : }
388 :
389 1 : void GstWebAudioPlayer::setCaps(const std::string &audioMimeType, std::weak_ptr<const WebAudioConfig> config)
390 : {
391 1 : m_workerThread->enqueueTask(m_taskFactory->createSetCaps(m_context, audioMimeType, config));
392 : }
393 :
394 1 : void GstWebAudioPlayer::play()
395 : {
396 1 : m_workerThread->enqueueTask(m_taskFactory->createPlay(*this));
397 : }
398 :
399 1 : void GstWebAudioPlayer::pause()
400 : {
401 1 : m_workerThread->enqueueTask(m_taskFactory->createPause(*this));
402 : }
403 :
404 1 : void GstWebAudioPlayer::setVolume(double volume)
405 : {
406 1 : m_workerThread->enqueueTask(m_taskFactory->createSetVolume(m_context, volume));
407 : }
408 :
409 1 : bool GstWebAudioPlayer::getVolume(double &volume)
410 : {
411 : // Must be called on the main thread, otherwise the pipeline can be destroyed during the query.
412 1 : volume = m_gstWrapper->gstStreamVolumeGetVolume(m_context.gstVolumeElement, GST_STREAM_VOLUME_FORMAT_LINEAR);
413 1 : return true;
414 : }
415 :
416 2 : uint32_t GstWebAudioPlayer::writeBuffer(uint8_t *mainPtr, uint32_t mainLength, uint8_t *wrapPtr, uint32_t wrapLength)
417 : {
418 : // Must block and wait for the data to be written from the shared buffer.
419 2 : std::unique_lock<std::mutex> lock(m_context.writeBufferMutex);
420 2 : const uint32_t initialCompletionCounter = m_context.writeCompletionCounter;
421 2 : m_workerThread->enqueueTask(m_taskFactory->createWriteBuffer(m_context, mainPtr, mainLength, wrapPtr, wrapLength));
422 : bool success =
423 2 : m_context.writeBufferCond.wait_for(lock, std::chrono::milliseconds(kMaxWriteBufferTimeoutMs),
424 4 : [this, initialCompletionCounter]()
425 4 : { return m_context.writeCompletionCounter != initialCompletionCounter; });
426 2 : if (!success)
427 : {
428 1 : RIALTO_SERVER_LOG_ERROR("Timed out writing to the gstreamer buffers");
429 1 : return 0;
430 : }
431 : else
432 : {
433 1 : return m_context.lastBytesWritten;
434 : }
435 2 : }
436 :
437 1 : void GstWebAudioPlayer::setEos()
438 : {
439 1 : m_workerThread->enqueueTask(m_taskFactory->createEos(m_context));
440 : }
441 :
442 1 : uint64_t GstWebAudioPlayer::getQueuedBytes()
443 : {
444 : // Must be called on the main thread, otherwise the pipeline can be destroyed during the query.
445 1 : return m_gstWrapper->gstAppSrcGetCurrentLevelBytes(GST_APP_SRC(m_context.source));
446 : }
447 :
448 3 : bool GstWebAudioPlayer::changePipelineState(GstState newState)
449 : {
450 3 : if (m_gstWrapper->gstElementSetState(m_context.pipeline, newState) == GST_STATE_CHANGE_FAILURE)
451 : {
452 1 : RIALTO_SERVER_LOG_ERROR("Change state failed - Gstreamer returned an error");
453 1 : if (m_gstPlayerClient)
454 1 : m_gstPlayerClient->notifyState(WebAudioPlayerState::FAILURE);
455 1 : return false;
456 : }
457 2 : return true;
458 : }
459 :
460 2 : void GstWebAudioPlayer::stopWorkerThread()
461 : {
462 2 : if (m_workerThread)
463 : {
464 2 : m_workerThread->stop();
465 : }
466 : }
467 :
468 1 : void GstWebAudioPlayer::handleBusMessage(GstMessage *message)
469 : {
470 1 : if (m_workerThread)
471 : {
472 1 : m_workerThread->enqueueTask(m_taskFactory->createHandleBusMessage(m_context, *this, message));
473 : }
474 : }
475 :
476 1 : void GstWebAudioPlayer::ping(std::unique_ptr<IHeartbeatHandler> &&heartbeatHandler)
477 : {
478 1 : if (m_workerThread)
479 : {
480 1 : m_workerThread->enqueueTask(m_taskFactory->createPing(std::move(heartbeatHandler)));
481 : }
482 : }
483 :
484 : } // namespace firebolt::rialto::server
|
