Line data Source code
1 : /*
2 : * Copyright (C) 2022 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 "GStreamerMSEMediaPlayerClient.h"
20 : #include "Constants.h"
21 : #include "GstreamerCatLog.h"
22 : #include "RialtoGStreamerMSEAudioSink.h"
23 : #include "RialtoGStreamerMSEBaseSink.h"
24 : #include "RialtoGStreamerMSEBaseSinkPrivate.h"
25 : #include "RialtoGStreamerMSEVideoSink.h"
26 :
27 : #include <algorithm>
28 : #include <chrono>
29 : #include <thread>
30 :
31 : namespace
32 : {
33 : // The start time of segment might differ from the first sample which is injected.
34 : // That difference should not be bigger than 1 video / audio frame.
35 : // 1 second is probably erring on the side of caution, but should not have side effect.
36 : const int64_t segmentStartMaximumDiff = 1000000000;
37 : const int32_t UNKNOWN_STREAMS_NUMBER = -1;
38 :
39 1 : const char *toString(const firebolt::rialto::PlaybackError &error)
40 : {
41 1 : switch (error)
42 : {
43 0 : case firebolt::rialto::PlaybackError::DECRYPTION:
44 0 : return "DECRYPTION";
45 1 : case firebolt::rialto::PlaybackError::OUTPUT_PROTECTION:
46 1 : return "OUTPUT_PROTECTION";
47 0 : case firebolt::rialto::PlaybackError::UNKNOWN:
48 0 : return "UNKNOWN";
49 : }
50 0 : return "UNKNOWN";
51 : }
52 0 : const char *toString(const firebolt::rialto::MediaSourceType &src)
53 : {
54 0 : switch (src)
55 : {
56 0 : case firebolt::rialto::MediaSourceType::AUDIO:
57 0 : return "AUDIO";
58 0 : case firebolt::rialto::MediaSourceType::VIDEO:
59 0 : return "VIDEO";
60 0 : case firebolt::rialto::MediaSourceType::SUBTITLE:
61 0 : return "SUBTITLE";
62 0 : case firebolt::rialto::MediaSourceType::UNKNOWN:
63 0 : return "UNKNOWN";
64 : }
65 0 : return "UNKNOWN";
66 : }
67 : } // namespace
68 : #define GST_CAT_DEFAULT rialtoGStreamerCat
69 306 : GStreamerMSEMediaPlayerClient::GStreamerMSEMediaPlayerClient(
70 : const std::shared_ptr<IMessageQueueFactory> &messageQueueFactory,
71 : const std::shared_ptr<firebolt::rialto::client::MediaPlayerClientBackendInterface> &MediaPlayerClientBackend,
72 306 : const uint32_t maxVideoWidth, const uint32_t maxVideoHeight, bool isLive)
73 306 : : m_backendQueue{messageQueueFactory->createMessageQueue()}, m_messageQueueFactory{messageQueueFactory},
74 306 : m_clientBackend(MediaPlayerClientBackend), m_position(0), m_duration(0), m_audioStreams{UNKNOWN_STREAMS_NUMBER},
75 306 : m_videoStreams{UNKNOWN_STREAMS_NUMBER}, m_subtitleStreams{UNKNOWN_STREAMS_NUMBER},
76 306 : m_videoRectangle{0, 0, 1920, 1080}, m_streamingStopped(false),
77 306 : m_maxWidth(maxVideoWidth == 0 ? DEFAULT_MAX_VIDEO_WIDTH : maxVideoWidth),
78 918 : m_maxHeight(maxVideoHeight == 0 ? DEFAULT_MAX_VIDEO_HEIGHT : maxVideoHeight), m_isLive{isLive}
79 : {
80 306 : m_backendQueue->start();
81 : }
82 :
83 306 : GStreamerMSEMediaPlayerClient::~GStreamerMSEMediaPlayerClient()
84 : {
85 306 : stopStreaming();
86 : }
87 :
88 473 : void GStreamerMSEMediaPlayerClient::stopStreaming()
89 : {
90 473 : if (!m_streamingStopped)
91 : {
92 306 : m_backendQueue->stop();
93 :
94 380 : for (auto &source : m_attachedSources)
95 : {
96 74 : source.second.m_bufferPuller->stop();
97 : }
98 :
99 306 : m_streamingStopped = true;
100 : }
101 473 : }
102 :
103 : // Deletes client backend -> this deletes mediapipeline object
104 172 : void GStreamerMSEMediaPlayerClient::destroyClientBackend()
105 : {
106 172 : m_clientBackend.reset();
107 : }
108 :
109 1 : void GStreamerMSEMediaPlayerClient::notifyDuration(int64_t duration)
110 : {
111 1 : m_backendQueue->postMessage(std::make_shared<SetDurationMessage>(duration, m_duration));
112 : }
113 :
114 1 : void GStreamerMSEMediaPlayerClient::notifyPosition(int64_t position)
115 : {
116 1 : m_backendQueue->postMessage(std::make_shared<SetPositionMessage>(position, m_attachedSources));
117 : }
118 :
119 : void GStreamerMSEMediaPlayerClient::notifyNativeSize(uint32_t width, uint32_t height, double aspect) {}
120 :
121 : void GStreamerMSEMediaPlayerClient::notifyNetworkState(firebolt::rialto::NetworkState state) {}
122 :
123 60 : void GStreamerMSEMediaPlayerClient::notifyPlaybackState(firebolt::rialto::PlaybackState state)
124 : {
125 60 : m_backendQueue->postMessage(std::make_shared<PlaybackStateMessage>(state, this));
126 : }
127 :
128 : void GStreamerMSEMediaPlayerClient::notifyVideoData(bool hasData) {}
129 :
130 : void GStreamerMSEMediaPlayerClient::notifyAudioData(bool hasData) {}
131 :
132 11 : void GStreamerMSEMediaPlayerClient::notifyNeedMediaData(
133 : int32_t sourceId, size_t frameCount, uint32_t needDataRequestId,
134 : const std::shared_ptr<firebolt::rialto::MediaPlayerShmInfo> & /*shmInfo*/)
135 : {
136 11 : m_backendQueue->postMessage(std::make_shared<NeedDataMessage>(sourceId, frameCount, needDataRequestId, this));
137 :
138 11 : return;
139 : }
140 :
141 : void GStreamerMSEMediaPlayerClient::notifyCancelNeedMediaData(int sourceId) {}
142 :
143 5 : void GStreamerMSEMediaPlayerClient::notifyQos(int32_t sourceId, const firebolt::rialto::QosInfo &qosInfo)
144 : {
145 5 : m_backendQueue->postMessage(std::make_shared<QosMessage>(sourceId, qosInfo, this));
146 : }
147 :
148 2 : void GStreamerMSEMediaPlayerClient::notifyBufferUnderflow(int32_t sourceId)
149 : {
150 2 : m_backendQueue->postMessage(std::make_shared<BufferUnderflowMessage>(sourceId, this));
151 : }
152 :
153 3 : void GStreamerMSEMediaPlayerClient::notifyFirstFrameReceived(int32_t sourceId)
154 : {
155 3 : m_backendQueue->postMessage(std::make_shared<FirstFrameReceivedMessage>(sourceId, this));
156 : }
157 :
158 7 : void GStreamerMSEMediaPlayerClient::notifyPlaybackError(int32_t sourceId, firebolt::rialto::PlaybackError error)
159 : {
160 7 : m_backendQueue->postMessage(std::make_shared<PlaybackErrorMessage>(sourceId, error, this));
161 : }
162 :
163 9 : void GStreamerMSEMediaPlayerClient::notifySourceFlushed(int32_t sourceId)
164 : {
165 9 : m_backendQueue->postMessage(std::make_shared<SourceFlushedMessage>(sourceId, this));
166 : }
167 :
168 8 : void GStreamerMSEMediaPlayerClient::notifyPlaybackInfo(const firebolt::rialto::PlaybackInfo &playbackInfo)
169 : {
170 8 : if (m_flushAndDataSynchronizer.isAnySourceFlushing())
171 : {
172 1 : GST_WARNING("Not updating playback info, because flush is ongoing");
173 1 : return;
174 : }
175 7 : std::unique_lock lock{m_playbackInfoMutex};
176 7 : m_playbackInfo = playbackInfo;
177 : }
178 :
179 8 : int64_t GStreamerMSEMediaPlayerClient::getPosition(int32_t sourceId)
180 : {
181 8 : std::unique_lock lock{m_playbackInfoMutex};
182 8 : return m_playbackInfo.currentPosition;
183 : }
184 :
185 6 : bool GStreamerMSEMediaPlayerClient::getDuration(int64_t &duration)
186 : {
187 6 : if (!m_clientBackend)
188 : {
189 1 : return false;
190 : }
191 :
192 5 : bool status{false};
193 10 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->getDuration(duration); });
194 5 : return status;
195 : }
196 :
197 5 : bool GStreamerMSEMediaPlayerClient::setImmediateOutput(int32_t sourceId, bool immediateOutput)
198 : {
199 5 : if (!m_clientBackend)
200 : {
201 1 : return false;
202 : }
203 :
204 4 : bool status{false};
205 8 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->setImmediateOutput(sourceId, immediateOutput); });
206 4 : return status;
207 : }
208 :
209 5 : bool GStreamerMSEMediaPlayerClient::setReportDecodeErrors(int32_t sourceId, bool reportDecodeErrors)
210 : {
211 5 : if (!m_clientBackend)
212 : {
213 1 : return false;
214 : }
215 :
216 4 : bool status{false};
217 4 : m_backendQueue->callInEventLoop([&]()
218 4 : { status = m_clientBackend->setReportDecodeErrors(sourceId, reportDecodeErrors); });
219 4 : return status;
220 : }
221 :
222 4 : bool GStreamerMSEMediaPlayerClient::getImmediateOutput(int32_t sourceId, bool &immediateOutput)
223 : {
224 4 : if (!m_clientBackend)
225 : {
226 1 : return false;
227 : }
228 :
229 3 : bool status{false};
230 6 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->getImmediateOutput(sourceId, immediateOutput); });
231 3 : return status;
232 : }
233 :
234 4 : bool GStreamerMSEMediaPlayerClient::getQueuedFrames(int32_t sourceId, uint32_t &queuedFrames)
235 : {
236 4 : if (!m_clientBackend)
237 : {
238 1 : return false;
239 : }
240 :
241 3 : bool status{false};
242 6 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->getQueuedFrames(sourceId, queuedFrames); });
243 3 : return status;
244 : }
245 :
246 1 : bool GStreamerMSEMediaPlayerClient::getStats(int32_t sourceId, uint64_t &renderedFrames, uint64_t &droppedFrames)
247 : {
248 1 : if (!m_clientBackend)
249 : {
250 0 : return false;
251 : }
252 :
253 1 : bool status{false};
254 1 : m_backendQueue->callInEventLoop([&]()
255 1 : { status = m_clientBackend->getStats(sourceId, renderedFrames, droppedFrames); });
256 1 : return status;
257 : }
258 :
259 175 : bool GStreamerMSEMediaPlayerClient::createBackend()
260 : {
261 175 : bool result = false;
262 350 : m_backendQueue->callInEventLoop(
263 175 : [&]()
264 : {
265 175 : if (!m_clientBackend)
266 : {
267 1 : GST_ERROR("Client backend is NULL");
268 1 : result = false;
269 1 : return;
270 : }
271 174 : m_clientBackend->createMediaPlayerBackend(shared_from_this(), m_maxWidth, m_maxHeight);
272 :
273 174 : if (m_clientBackend->isMediaPlayerBackendCreated())
274 : {
275 169 : std::string utf8url = "mse://1";
276 169 : firebolt::rialto::MediaType mediaType = firebolt::rialto::MediaType::MSE;
277 507 : if (!m_clientBackend->load(mediaType, "", utf8url, m_isLive))
278 : {
279 1 : GST_ERROR("Could not load RialtoClient");
280 1 : return;
281 : }
282 168 : result = true;
283 169 : }
284 : else
285 : {
286 5 : GST_ERROR("Media player backend could not be created");
287 : }
288 : });
289 :
290 175 : return result;
291 : }
292 :
293 23 : StateChangeResult GStreamerMSEMediaPlayerClient::play(int32_t sourceId)
294 : {
295 23 : StateChangeResult result = StateChangeResult::NOT_ATTACHED;
296 46 : m_backendQueue->callInEventLoop(
297 23 : [&]()
298 : {
299 23 : auto sourceIt = m_attachedSources.find(sourceId);
300 23 : if (sourceIt == m_attachedSources.end())
301 : {
302 1 : GST_ERROR("Cannot play - there's no attached source with id %d", sourceId);
303 1 : result = StateChangeResult::NOT_ATTACHED;
304 4 : return;
305 : }
306 :
307 22 : if (m_serverPlaybackState == firebolt::rialto::PlaybackState::PLAYING ||
308 20 : (m_serverPlaybackState == firebolt::rialto::PlaybackState::END_OF_STREAM && wasPlayingBeforeEos))
309 : {
310 2 : GST_INFO("Server is already playing");
311 2 : sourceIt->second.m_state = ClientState::PLAYING;
312 :
313 6 : if (checkIfAllAttachedSourcesInStates({ClientState::PLAYING}))
314 : {
315 1 : m_clientState = ClientState::PLAYING;
316 : }
317 :
318 2 : result = StateChangeResult::SUCCESS_SYNC;
319 2 : return;
320 : }
321 :
322 20 : sourceIt->second.m_state = ClientState::AWAITING_PLAYING;
323 :
324 20 : if (m_clientState == ClientState::PAUSED)
325 : {
326 : // If one source is AWAITING_PLAYING, the other source can still be PLAYING.
327 : // This happends when we are switching out audio.
328 48 : if (checkIfAllAttachedSourcesInStates({ClientState::AWAITING_PLAYING, ClientState::PLAYING}))
329 : {
330 11 : GST_INFO("Sending play command");
331 11 : bool async{true};
332 11 : m_clientBackend->play(async);
333 11 : m_clientState = ClientState::AWAITING_PLAYING;
334 11 : if (!async)
335 : {
336 : // Synchronous playing state change. Finish procedure for other sources and return SUCCESS_SYNC
337 1 : result = StateChangeResult::SUCCESS_SYNC;
338 2 : m_backendQueue->postMessage(
339 2 : std::make_shared<PlaybackStateMessage>(firebolt::rialto::PlaybackState::PLAYING, this));
340 1 : return;
341 : }
342 : }
343 : else
344 : {
345 5 : GST_DEBUG("Not all sources are ready to play");
346 : }
347 : }
348 : else
349 : {
350 4 : GST_WARNING("Not in PAUSED state in client state %u state; server playback state: %u",
351 : static_cast<uint32_t>(m_clientState), static_cast<uint32_t>(m_serverPlaybackState));
352 : }
353 :
354 19 : result = StateChangeResult::SUCCESS_ASYNC;
355 19 : sourceIt->second.m_delegate->postAsyncStart();
356 : });
357 :
358 23 : return result;
359 : }
360 :
361 343 : StateChangeResult GStreamerMSEMediaPlayerClient::pause(int32_t sourceId)
362 : {
363 343 : StateChangeResult result = StateChangeResult::NOT_ATTACHED;
364 686 : m_backendQueue->callInEventLoop(
365 343 : [&]()
366 : {
367 343 : auto sourceIt = m_attachedSources.find(sourceId);
368 343 : if (sourceIt == m_attachedSources.end())
369 : {
370 161 : GST_WARNING("Cannot pause - there's no attached source with id %d", sourceId);
371 :
372 161 : result = StateChangeResult::NOT_ATTACHED;
373 161 : return;
374 : }
375 :
376 182 : if (m_serverPlaybackState == firebolt::rialto::PlaybackState::PAUSED &&
377 4 : m_clientState != ClientState::AWAITING_PLAYING && m_clientState != ClientState::AWAITING_PAUSED)
378 : {
379 : // if the server is already paused and we are not in async, we don't need to send pause command
380 2 : GST_INFO("Server is already paused");
381 2 : sourceIt->second.m_state = ClientState::PAUSED;
382 :
383 6 : if (checkIfAllAttachedSourcesInStates({ClientState::PAUSED}))
384 : {
385 1 : m_clientState = ClientState::PAUSED;
386 : }
387 :
388 2 : result = StateChangeResult::SUCCESS_SYNC;
389 : }
390 : else
391 : {
392 180 : sourceIt->second.m_state = ClientState::AWAITING_PAUSED;
393 :
394 180 : bool shouldPause = false;
395 180 : if (m_clientState == ClientState::READY)
396 : {
397 504 : if (checkIfAllAttachedSourcesInStates({ClientState::AWAITING_PAUSED}))
398 : {
399 159 : shouldPause = true;
400 : }
401 : else
402 : {
403 9 : GST_DEBUG("Not all attached sources are ready to pause");
404 : }
405 : }
406 12 : else if (m_clientState == ClientState::AWAITING_PLAYING || m_clientState == ClientState::PLAYING)
407 : {
408 9 : shouldPause = true;
409 : }
410 : else
411 : {
412 3 : GST_DEBUG("Cannot pause in %u state", static_cast<uint32_t>(m_clientState));
413 : }
414 :
415 180 : if (shouldPause)
416 : {
417 168 : GST_INFO("Sending pause command in %u state", static_cast<uint32_t>(m_clientState));
418 168 : m_clientBackend->pause();
419 168 : m_clientState = ClientState::AWAITING_PAUSED;
420 : }
421 :
422 180 : result = StateChangeResult::SUCCESS_ASYNC;
423 180 : sourceIt->second.m_delegate->postAsyncStart();
424 : }
425 : });
426 :
427 343 : return result;
428 : }
429 :
430 168 : void GStreamerMSEMediaPlayerClient::stop()
431 : {
432 336 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->stop(); });
433 168 : }
434 :
435 4 : void GStreamerMSEMediaPlayerClient::setPlaybackRate(double rate)
436 : {
437 8 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->setPlaybackRate(rate); });
438 4 : }
439 :
440 12 : void GStreamerMSEMediaPlayerClient::flush(int32_t sourceId, bool resetTime)
441 : {
442 12 : m_flushAndDataSynchronizer.notifyFlushStarted(sourceId);
443 24 : m_backendQueue->callInEventLoop(
444 12 : [&]()
445 : {
446 12 : wasPlayingBeforeEos = false;
447 12 : bool async{true};
448 12 : auto sourceIt = m_attachedSources.find(sourceId);
449 12 : if (sourceIt == m_attachedSources.end())
450 : {
451 2 : GST_ERROR("Cannot flush - there's no attached source with id %d", sourceId);
452 3 : return;
453 : }
454 10 : if (!m_clientBackend->flush(sourceId, resetTime, async))
455 : {
456 1 : GST_ERROR("Flush operation failed for source with id %d", sourceId);
457 1 : return;
458 : }
459 9 : sourceIt->second.m_isFlushing = true;
460 :
461 9 : if (async)
462 : {
463 9 : GST_INFO("Flush request sent for async source %d. Sink will lose state now", sourceId);
464 9 : sourceIt->second.m_delegate->lostState();
465 :
466 9 : sourceIt->second.m_state = ClientState::AWAITING_PAUSED;
467 9 : if (m_clientState == ClientState::PLAYING)
468 : {
469 0 : m_clientState = ClientState::AWAITING_PLAYING;
470 : }
471 9 : else if (m_clientState == ClientState::PAUSED)
472 : {
473 4 : m_clientState = ClientState::AWAITING_PAUSED;
474 : }
475 : }
476 : });
477 12 : }
478 :
479 9 : void GStreamerMSEMediaPlayerClient::setSourcePosition(int32_t sourceId, int64_t position, bool resetTime,
480 : double appliedRate, uint64_t stopPosition)
481 : {
482 18 : m_backendQueue->callInEventLoop(
483 9 : [&]()
484 : {
485 9 : auto sourceIt = m_attachedSources.find(sourceId);
486 9 : if (sourceIt == m_attachedSources.end())
487 : {
488 1 : GST_ERROR("Cannot Set Source Position - there's no attached source with id %d", sourceId);
489 2 : return;
490 : }
491 8 : if (!m_clientBackend->setSourcePosition(sourceId, position, resetTime, appliedRate, stopPosition))
492 : {
493 1 : GST_ERROR("Set Source Position operation failed for source with id %d", sourceId);
494 1 : return;
495 : }
496 7 : sourceIt->second.m_position = position;
497 : });
498 9 : }
499 :
500 1 : void GStreamerMSEMediaPlayerClient::setSubtitleOffset(int32_t sourceId, int64_t position)
501 : {
502 2 : m_backendQueue->callInEventLoop(
503 1 : [&]()
504 : {
505 1 : auto sourceIt = m_attachedSources.find(sourceId);
506 1 : if (sourceIt == m_attachedSources.end())
507 : {
508 0 : GST_ERROR("Cannot Set Subtitle Offset - there's no attached source with id %d", sourceId);
509 0 : return;
510 : }
511 1 : if (!m_clientBackend->setSubtitleOffset(sourceId, position))
512 : {
513 0 : GST_ERROR("Set Subtitle Offset operation failed for source with id %d", sourceId);
514 0 : return;
515 : }
516 : });
517 1 : }
518 :
519 4 : void GStreamerMSEMediaPlayerClient::processAudioGap(int64_t position, uint32_t duration, int64_t discontinuityGap,
520 : bool audioAac)
521 : {
522 8 : m_backendQueue->callInEventLoop(
523 4 : [&]()
524 : {
525 4 : if (!m_clientBackend->processAudioGap(position, duration, discontinuityGap, audioAac))
526 : {
527 1 : GST_ERROR("Process Audio Gap operation failed");
528 1 : return;
529 : }
530 : });
531 4 : }
532 :
533 227 : bool GStreamerMSEMediaPlayerClient::attachSource(std::unique_ptr<firebolt::rialto::IMediaPipeline::MediaSource> &source,
534 : RialtoMSEBaseSink *rialtoSink,
535 : const std::shared_ptr<IPullModePlaybackDelegate> &delegate)
536 : {
537 227 : if (source->getType() != firebolt::rialto::MediaSourceType::AUDIO &&
538 239 : source->getType() != firebolt::rialto::MediaSourceType::VIDEO &&
539 12 : source->getType() != firebolt::rialto::MediaSourceType::SUBTITLE)
540 : {
541 1 : GST_WARNING_OBJECT(rialtoSink, "Invalid source type %u", static_cast<uint32_t>(source->getType()));
542 1 : return false;
543 : }
544 :
545 226 : bool result = false;
546 452 : m_backendQueue->callInEventLoop(
547 226 : [&]()
548 : {
549 226 : result = m_clientBackend->attachSource(source);
550 :
551 226 : if (result)
552 : {
553 225 : std::shared_ptr<BufferParser> bufferParser;
554 225 : if (source->getType() == firebolt::rialto::MediaSourceType::AUDIO)
555 : {
556 164 : bufferParser = std::make_shared<AudioBufferParser>();
557 : }
558 61 : else if (source->getType() == firebolt::rialto::MediaSourceType::VIDEO)
559 : {
560 50 : bufferParser = std::make_shared<VideoBufferParser>();
561 : }
562 11 : else if (source->getType() == firebolt::rialto::MediaSourceType::SUBTITLE)
563 : {
564 11 : bufferParser = std::make_shared<SubtitleBufferParser>();
565 : }
566 :
567 225 : std::shared_ptr<BufferPuller> bufferPuller = std::make_shared<BufferPuller>(m_messageQueueFactory,
568 0 : GST_ELEMENT_CAST(rialtoSink),
569 225 : bufferParser, delegate);
570 :
571 225 : if (m_attachedSources.find(source->getId()) == m_attachedSources.end())
572 : {
573 225 : m_attachedSources.emplace(source->getId(),
574 450 : AttachedSource(rialtoSink, bufferPuller, delegate, source->getType()));
575 225 : delegate->setSourceId(source->getId());
576 225 : m_flushAndDataSynchronizer.addSource(source->getId());
577 225 : bufferPuller->start();
578 : }
579 : }
580 :
581 226 : sendAllSourcesAttachedIfPossibleInternal();
582 226 : });
583 :
584 226 : return result;
585 : }
586 :
587 4 : void GStreamerMSEMediaPlayerClient::sendAllSourcesAttachedIfPossible()
588 : {
589 8 : m_backendQueue->callInEventLoop([&]() { sendAllSourcesAttachedIfPossibleInternal(); });
590 4 : }
591 :
592 323 : void GStreamerMSEMediaPlayerClient::setStopping(bool stopping)
593 : {
594 323 : m_backendQueue->callInEventLoop([this, stopping]() { m_isStopping = stopping; });
595 : }
596 :
597 230 : void GStreamerMSEMediaPlayerClient::sendAllSourcesAttachedIfPossibleInternal()
598 : {
599 230 : if (m_isStopping)
600 : {
601 1 : GST_INFO("Skip sending allSourcesAttached, because a stop was already requested");
602 1 : return;
603 : }
604 :
605 229 : if (!m_wasAllSourcesAttachedSent && areAllStreamsAttached())
606 : {
607 : // RialtoServer doesn't support dynamic source attachment.
608 : // It means that when we notify that all sources were attached, we cannot add any more sources in the current session
609 166 : GST_INFO("All sources attached");
610 166 : m_clientBackend->allSourcesAttached();
611 166 : m_wasAllSourcesAttachedSent = true;
612 166 : m_clientState = ClientState::READY;
613 :
614 : // In playbin3 streams, confirmation about number of available sources comes after attaching the source,
615 : // so we need to check if all sources are ready to pause
616 498 : if (checkIfAllAttachedSourcesInStates({ClientState::AWAITING_PAUSED}))
617 : {
618 1 : GST_INFO("Sending pause command, because all attached sources are ready to pause");
619 1 : m_clientBackend->pause();
620 1 : m_clientState = ClientState::AWAITING_PAUSED;
621 : }
622 : }
623 : }
624 :
625 162 : void GStreamerMSEMediaPlayerClient::removeSource(int32_t sourceId)
626 : {
627 324 : m_backendQueue->callInEventLoop(
628 162 : [&]()
629 : {
630 162 : if (!m_clientBackend->removeSource(sourceId))
631 : {
632 1 : GST_WARNING("Remove source %d failed", sourceId);
633 : }
634 162 : m_attachedSources.erase(sourceId);
635 162 : m_flushAndDataSynchronizer.removeSource(sourceId);
636 162 : });
637 : }
638 :
639 61 : void GStreamerMSEMediaPlayerClient::handlePlaybackStateChange(firebolt::rialto::PlaybackState state)
640 : {
641 61 : GST_DEBUG("Received state change to state %u", static_cast<uint32_t>(state));
642 122 : m_backendQueue->callInEventLoop(
643 61 : [&]()
644 : {
645 61 : const auto kPreviousState{m_serverPlaybackState};
646 61 : m_serverPlaybackState = state;
647 61 : switch (state)
648 : {
649 54 : case firebolt::rialto::PlaybackState::PAUSED:
650 : case firebolt::rialto::PlaybackState::PLAYING:
651 : {
652 54 : wasPlayingBeforeEos = false;
653 54 : if (state == firebolt::rialto::PlaybackState::PAUSED && m_clientState == ClientState::AWAITING_PAUSED)
654 : {
655 39 : m_clientState = ClientState::PAUSED;
656 : }
657 15 : else if (state == firebolt::rialto::PlaybackState::PLAYING &&
658 13 : m_clientState == ClientState::AWAITING_PLAYING)
659 : {
660 9 : m_clientState = ClientState::PLAYING;
661 : }
662 6 : else if (state == firebolt::rialto::PlaybackState::PLAYING &&
663 4 : m_clientState == ClientState::AWAITING_PAUSED)
664 : {
665 1 : GST_WARNING("Outdated Playback State change to PLAYING received. Discarding...");
666 1 : break;
667 : }
668 :
669 119 : for (auto &source : m_attachedSources)
670 : {
671 66 : if (state == firebolt::rialto::PlaybackState::PAUSED &&
672 49 : source.second.m_state == ClientState::AWAITING_PAUSED)
673 : {
674 46 : source.second.m_state = ClientState::PAUSED;
675 : }
676 20 : else if (state == firebolt::rialto::PlaybackState::PLAYING &&
677 17 : source.second.m_state == ClientState::AWAITING_PLAYING)
678 : {
679 12 : source.second.m_state = ClientState::PLAYING;
680 : }
681 66 : source.second.m_delegate->handleStateChanged(state);
682 : }
683 :
684 53 : break;
685 : }
686 4 : case firebolt::rialto::PlaybackState::END_OF_STREAM:
687 : {
688 4 : if (!wasPlayingBeforeEos && firebolt::rialto::PlaybackState::PLAYING == kPreviousState)
689 : {
690 2 : wasPlayingBeforeEos = true;
691 : }
692 8 : for (const auto &source : m_attachedSources)
693 : {
694 4 : source.second.m_delegate->handleEos();
695 : }
696 : }
697 4 : break;
698 1 : case firebolt::rialto::PlaybackState::SEEK_DONE:
699 : {
700 1 : GST_WARNING("firebolt::rialto::PlaybackState::SEEK_DONE notification not supported");
701 1 : break;
702 : }
703 1 : case firebolt::rialto::PlaybackState::FAILURE:
704 : {
705 1 : wasPlayingBeforeEos = false;
706 2 : for (const auto &source : m_attachedSources)
707 : {
708 3 : source.second.m_delegate->handleError("Rialto server playback failed");
709 : }
710 2 : for (auto &source : m_attachedSources)
711 : {
712 1 : source.second.m_position = 0;
713 : }
714 : {
715 1 : std::unique_lock lock{m_playbackInfoMutex};
716 1 : m_playbackInfo.currentPosition = 0;
717 : }
718 :
719 1 : break;
720 : }
721 : break;
722 1 : default:
723 1 : break;
724 : }
725 61 : });
726 : }
727 :
728 9 : void GStreamerMSEMediaPlayerClient::handleSourceFlushed(int32_t sourceId)
729 : {
730 18 : m_backendQueue->callInEventLoop(
731 9 : [&]()
732 : {
733 9 : auto sourceIt = m_attachedSources.find(sourceId);
734 9 : if (sourceIt == m_attachedSources.end())
735 : {
736 1 : GST_ERROR("Cannot finish flush - there's no attached source with id %d", sourceId);
737 2 : return;
738 : }
739 8 : if (!sourceIt->second.m_isFlushing)
740 : {
741 1 : GST_ERROR("Cannot finish flush - source with id %d is not flushing!", sourceId);
742 1 : return;
743 : }
744 7 : sourceIt->second.m_isFlushing = false;
745 7 : sourceIt->second.m_delegate->handleFlushCompleted();
746 7 : m_flushAndDataSynchronizer.notifyFlushCompleted(sourceId);
747 : });
748 9 : }
749 :
750 7 : void GStreamerMSEMediaPlayerClient::setVideoRectangle(const std::string &rectangleString)
751 : {
752 14 : m_backendQueue->callInEventLoop(
753 7 : [&]()
754 : {
755 7 : if (!m_clientBackend || !m_clientBackend->isMediaPlayerBackendCreated())
756 : {
757 1 : GST_WARNING("Missing RialtoClient backend - can't set video window now");
758 3 : return;
759 : }
760 :
761 6 : if (rectangleString.empty())
762 : {
763 1 : GST_WARNING("Empty video rectangle string");
764 1 : return;
765 : }
766 :
767 5 : Rectangle rect = {0, 0, 0, 0};
768 5 : if (sscanf(rectangleString.c_str(), "%u,%u,%u,%u", &rect.x, &rect.y, &rect.width, &rect.height) != 4)
769 : {
770 1 : GST_WARNING("Invalid video rectangle values");
771 1 : return;
772 : }
773 :
774 4 : m_clientBackend->setVideoWindow(rect.x, rect.y, rect.width, rect.height);
775 4 : m_videoRectangle = rect;
776 : });
777 7 : }
778 :
779 4 : std::string GStreamerMSEMediaPlayerClient::getVideoRectangle()
780 : {
781 4 : char rectangle[64] = {0};
782 8 : m_backendQueue->callInEventLoop(
783 8 : [&]()
784 : {
785 4 : sprintf(rectangle, "%u,%u,%u,%u", m_videoRectangle.x, m_videoRectangle.y, m_videoRectangle.width,
786 : m_videoRectangle.height);
787 4 : });
788 :
789 8 : return std::string(rectangle);
790 : }
791 :
792 4 : bool GStreamerMSEMediaPlayerClient::renderFrame(int32_t sourceId)
793 : {
794 4 : bool result = false;
795 8 : m_backendQueue->callInEventLoop(
796 4 : [&]()
797 : {
798 4 : result = m_clientBackend->renderFrame();
799 4 : if (result)
800 : {
801 : // RialtoServer's video sink should drop PAUSED state due to skipping prerolled buffer in PAUSED state
802 3 : auto sourceIt = m_attachedSources.find(sourceId);
803 3 : if (sourceIt != m_attachedSources.end())
804 : {
805 3 : sourceIt->second.m_delegate->lostState();
806 : }
807 : }
808 4 : });
809 4 : return result;
810 : }
811 :
812 6 : void GStreamerMSEMediaPlayerClient::setVolume(double targetVolume, uint32_t volumeDuration,
813 : firebolt::rialto::EaseType easeType)
814 : {
815 12 : m_backendQueue->callInEventLoop(
816 6 : [&]()
817 : {
818 6 : m_clientBackend->setVolume(targetVolume, volumeDuration, easeType);
819 6 : std::unique_lock lock{m_playbackInfoMutex};
820 6 : m_playbackInfo.volume = targetVolume;
821 6 : });
822 : }
823 :
824 2 : bool GStreamerMSEMediaPlayerClient::getVolume(double &volume)
825 : {
826 2 : bool success{false};
827 4 : m_backendQueue->callInEventLoop([&]() { success = m_clientBackend->getVolume(volume); });
828 2 : return success;
829 : }
830 :
831 2 : bool GStreamerMSEMediaPlayerClient::getCachedVolume(double &volume)
832 : {
833 2 : std::unique_lock lock{m_playbackInfoMutex};
834 2 : volume = m_playbackInfo.volume;
835 2 : return true;
836 : }
837 :
838 7 : void GStreamerMSEMediaPlayerClient::setMute(bool mute, int32_t sourceId)
839 : {
840 14 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->setMute(mute, sourceId); });
841 7 : }
842 :
843 3 : bool GStreamerMSEMediaPlayerClient::getMute(int sourceId)
844 : {
845 3 : bool mute{false};
846 6 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->getMute(mute, sourceId); });
847 :
848 3 : return mute;
849 : }
850 :
851 3 : void GStreamerMSEMediaPlayerClient::setTextTrackIdentifier(const std::string &textTrackIdentifier)
852 : {
853 6 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->setTextTrackIdentifier(textTrackIdentifier); });
854 3 : }
855 :
856 2 : std::string GStreamerMSEMediaPlayerClient::getTextTrackIdentifier()
857 : {
858 2 : std::string getTextTrackIdentifier;
859 4 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->getTextTrackIdentifier(getTextTrackIdentifier); });
860 2 : return getTextTrackIdentifier;
861 : }
862 :
863 6 : bool GStreamerMSEMediaPlayerClient::setLowLatency(bool lowLatency)
864 : {
865 6 : if (!m_clientBackend)
866 : {
867 1 : return false;
868 : }
869 :
870 5 : bool status{false};
871 10 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->setLowLatency(lowLatency); });
872 5 : return status;
873 : }
874 :
875 6 : bool GStreamerMSEMediaPlayerClient::setSync(bool sync)
876 : {
877 6 : if (!m_clientBackend)
878 : {
879 1 : return false;
880 : }
881 :
882 5 : bool status{false};
883 10 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->setSync(sync); });
884 5 : return status;
885 : }
886 :
887 4 : bool GStreamerMSEMediaPlayerClient::getSync(bool &sync)
888 : {
889 4 : if (!m_clientBackend)
890 : {
891 1 : return false;
892 : }
893 :
894 3 : bool status{false};
895 6 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->getSync(sync); });
896 3 : return status;
897 : }
898 :
899 6 : bool GStreamerMSEMediaPlayerClient::setSyncOff(bool syncOff)
900 : {
901 6 : if (!m_clientBackend)
902 : {
903 1 : return false;
904 : }
905 :
906 5 : bool status{false};
907 10 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->setSyncOff(syncOff); });
908 5 : return status;
909 : }
910 :
911 10 : bool GStreamerMSEMediaPlayerClient::setStreamSyncMode(int32_t sourceId, int32_t streamSyncMode)
912 : {
913 10 : if (!m_clientBackend)
914 : {
915 1 : return false;
916 : }
917 :
918 9 : bool status{false};
919 18 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->setStreamSyncMode(sourceId, streamSyncMode); });
920 9 : return status;
921 : }
922 :
923 4 : bool GStreamerMSEMediaPlayerClient::getStreamSyncMode(int32_t &streamSyncMode)
924 : {
925 4 : if (!m_clientBackend)
926 : {
927 1 : return false;
928 : }
929 :
930 3 : bool status{false};
931 6 : m_backendQueue->callInEventLoop([&]() { status = m_clientBackend->getStreamSyncMode(streamSyncMode); });
932 3 : return status;
933 : }
934 :
935 46 : ClientState GStreamerMSEMediaPlayerClient::getClientState()
936 : {
937 46 : ClientState state{ClientState::IDLE};
938 46 : m_backendQueue->callInEventLoop([&]() { state = m_clientState; });
939 46 : return state;
940 : }
941 :
942 184 : void GStreamerMSEMediaPlayerClient::handleStreamCollection(int32_t audioStreams, int32_t videoStreams,
943 : int32_t subtitleStreams)
944 : {
945 368 : m_backendQueue->callInEventLoop(
946 184 : [&]()
947 : {
948 184 : if (m_audioStreams == UNKNOWN_STREAMS_NUMBER)
949 181 : m_audioStreams = audioStreams;
950 184 : if (m_videoStreams == UNKNOWN_STREAMS_NUMBER)
951 180 : m_videoStreams = videoStreams;
952 184 : if (m_subtitleStreams == UNKNOWN_STREAMS_NUMBER)
953 180 : m_subtitleStreams = subtitleStreams;
954 :
955 184 : GST_INFO("Updated number of streams. New streams' numbers; video=%d, audio=%d, text=%d", m_videoStreams,
956 : m_audioStreams, m_subtitleStreams);
957 184 : });
958 : }
959 :
960 3 : void GStreamerMSEMediaPlayerClient::setBufferingLimit(uint32_t limitBufferingMs)
961 : {
962 3 : if (!m_clientBackend)
963 : {
964 0 : return;
965 : }
966 6 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->setBufferingLimit(limitBufferingMs); });
967 : }
968 :
969 2 : uint32_t GStreamerMSEMediaPlayerClient::getBufferingLimit()
970 : {
971 2 : if (!m_clientBackend)
972 : {
973 0 : return kDefaultBufferingLimit;
974 : }
975 :
976 2 : uint32_t result{kDefaultBufferingLimit};
977 4 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->getBufferingLimit(result); });
978 2 : return result;
979 : }
980 :
981 3 : void GStreamerMSEMediaPlayerClient::setUseBuffering(bool useBuffering)
982 : {
983 3 : if (!m_clientBackend)
984 : {
985 0 : return;
986 : }
987 6 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->setUseBuffering(useBuffering); });
988 : }
989 :
990 2 : bool GStreamerMSEMediaPlayerClient::getUseBuffering()
991 : {
992 2 : if (!m_clientBackend)
993 : {
994 0 : return kDefaultUseBuffering;
995 : }
996 :
997 2 : bool result{kDefaultUseBuffering};
998 4 : m_backendQueue->callInEventLoop([&]() { m_clientBackend->getUseBuffering(result); });
999 2 : return result;
1000 : }
1001 :
1002 3 : bool GStreamerMSEMediaPlayerClient::switchSource(const std::unique_ptr<firebolt::rialto::IMediaPipeline::MediaSource> &source)
1003 : {
1004 3 : bool result = false;
1005 6 : m_backendQueue->callInEventLoop([&]() { result = m_clientBackend->switchSource(source); });
1006 :
1007 3 : return result;
1008 : }
1009 :
1010 36 : IFlushAndDataSynchronizer &GStreamerMSEMediaPlayerClient::getFlushAndDataSynchronizer()
1011 : {
1012 36 : return m_flushAndDataSynchronizer;
1013 : }
1014 :
1015 354 : bool GStreamerMSEMediaPlayerClient::checkIfAllAttachedSourcesInStates(const std::vector<ClientState> &states)
1016 : {
1017 708 : return std::all_of(m_attachedSources.begin(), m_attachedSources.end(), [states](const auto &source)
1018 1077 : { return std::find(states.begin(), states.end(), source.second.m_state) != states.end(); });
1019 : }
1020 :
1021 228 : bool GStreamerMSEMediaPlayerClient::areAllStreamsAttached()
1022 : {
1023 228 : int32_t attachedVideoSources = 0;
1024 228 : int32_t attachedAudioSources = 0;
1025 228 : int32_t attachedSubtitleSources = 0;
1026 467 : for (auto &source : m_attachedSources)
1027 : {
1028 239 : if (source.second.getType() == firebolt::rialto::MediaSourceType::VIDEO)
1029 : {
1030 50 : attachedVideoSources++;
1031 : }
1032 189 : else if (source.second.getType() == firebolt::rialto::MediaSourceType::AUDIO)
1033 : {
1034 178 : attachedAudioSources++;
1035 : }
1036 11 : else if (source.second.getType() == firebolt::rialto::MediaSourceType::SUBTITLE)
1037 : {
1038 11 : attachedSubtitleSources++;
1039 : }
1040 : }
1041 :
1042 394 : return attachedVideoSources == m_videoStreams && attachedAudioSources == m_audioStreams &&
1043 394 : attachedSubtitleSources == m_subtitleStreams;
1044 : }
1045 :
1046 11 : bool GStreamerMSEMediaPlayerClient::requestPullBuffer(int streamId, size_t frameCount, unsigned int needDataRequestId)
1047 : {
1048 11 : bool result = false;
1049 22 : m_backendQueue->callInEventLoop(
1050 11 : [&]()
1051 : {
1052 11 : auto sourceIt = m_attachedSources.find(streamId);
1053 11 : if (sourceIt == m_attachedSources.end())
1054 : {
1055 1 : GST_ERROR("There's no attached source with id %d", streamId);
1056 :
1057 1 : result = false;
1058 1 : return;
1059 : }
1060 10 : result = sourceIt->second.m_bufferPuller->requestPullBuffer(streamId, frameCount, needDataRequestId, this);
1061 : });
1062 :
1063 11 : return result;
1064 : }
1065 :
1066 5 : bool GStreamerMSEMediaPlayerClient::handleQos(int sourceId, firebolt::rialto::QosInfo qosInfo)
1067 : {
1068 5 : bool result = false;
1069 10 : m_backendQueue->callInEventLoop(
1070 5 : [&]()
1071 : {
1072 5 : auto sourceIt = m_attachedSources.find(sourceId);
1073 5 : if (sourceIt == m_attachedSources.end())
1074 : {
1075 1 : result = false;
1076 1 : return;
1077 : }
1078 4 : sourceIt->second.m_delegate->handleQos(qosInfo.processed, qosInfo.dropped);
1079 4 : result = true;
1080 : });
1081 :
1082 5 : return result;
1083 : }
1084 :
1085 2 : bool GStreamerMSEMediaPlayerClient::handleBufferUnderflow(int sourceId)
1086 : {
1087 2 : bool result = false;
1088 4 : m_backendQueue->callInEventLoop(
1089 2 : [&]()
1090 : {
1091 2 : auto sourceIt = m_attachedSources.find(sourceId);
1092 2 : if (sourceIt == m_attachedSources.end())
1093 : {
1094 1 : result = false;
1095 1 : return;
1096 : }
1097 :
1098 1 : rialto_mse_base_handle_rialto_server_sent_buffer_underflow(sourceIt->second.m_rialtoSink);
1099 :
1100 1 : result = true;
1101 : });
1102 :
1103 2 : return result;
1104 : }
1105 :
1106 3 : bool GStreamerMSEMediaPlayerClient::handleFirstFrameReceived(int sourceId)
1107 : {
1108 3 : bool result = false;
1109 6 : m_backendQueue->callInEventLoop(
1110 3 : [&]()
1111 : {
1112 3 : auto sourceIt = m_attachedSources.find(sourceId);
1113 3 : if (sourceIt == m_attachedSources.end())
1114 : {
1115 1 : result = false;
1116 1 : return;
1117 : }
1118 2 : if (sourceIt->second.getType() == firebolt::rialto::MediaSourceType::VIDEO)
1119 : {
1120 1 : rialto_mse_video_handle_rialto_server_sent_first_video_frame_received(
1121 1 : RIALTO_MSE_VIDEO_SINK(sourceIt->second.m_rialtoSink));
1122 1 : result = true;
1123 : }
1124 1 : else if (sourceIt->second.getType() == firebolt::rialto::MediaSourceType::AUDIO)
1125 : {
1126 1 : rialto_mse_audio_handle_rialto_server_sent_first_audio_frame_received(
1127 1 : RIALTO_MSE_AUDIO_SINK(sourceIt->second.m_rialtoSink));
1128 1 : result = true;
1129 : }
1130 : });
1131 :
1132 3 : return result;
1133 : }
1134 :
1135 7 : bool GStreamerMSEMediaPlayerClient::handlePlaybackError(int sourceId, firebolt::rialto::PlaybackError error)
1136 : {
1137 7 : bool result = false;
1138 14 : m_backendQueue->callInEventLoop(
1139 7 : [&]()
1140 : {
1141 7 : auto sourceIt = m_attachedSources.find(sourceId);
1142 7 : if (sourceIt == m_attachedSources.end())
1143 : {
1144 2 : result = false;
1145 2 : return;
1146 : }
1147 :
1148 : // OUTPUT_PROTECTION is handled separately by posting an application message (not a pipeline error)
1149 5 : if (firebolt::rialto::PlaybackError::OUTPUT_PROTECTION == error)
1150 : {
1151 1 : GST_WARNING("HDCP output protection failure, posting HDCPProtectionFailure application message");
1152 1 : GstStructure *hdcpMsg = gst_structure_new("HDCPProtectionFailure", "message", G_TYPE_STRING,
1153 : "HDCP Output Protection Error", "error", G_TYPE_STRING,
1154 : toString(error), nullptr);
1155 : GstMessage *message =
1156 1 : gst_message_new_application(GST_OBJECT_CAST(sourceIt->second.m_rialtoSink), hdcpMsg);
1157 1 : result = gst_element_post_message(GST_ELEMENT_CAST(sourceIt->second.m_rialtoSink), message);
1158 1 : if (!result)
1159 : {
1160 0 : GST_WARNING("Failed to post HDCPProtectionFailure application message");
1161 : }
1162 : }
1163 : else
1164 : {
1165 : // For other playback errors, fail the pipeline from rialto-gstreamer
1166 4 : GST_ERROR("Received Playback error '%s', posting error on %s sink", toString(error),
1167 : toString(sourceIt->second.getType()));
1168 4 : if (firebolt::rialto::PlaybackError::DECRYPTION == error)
1169 : {
1170 6 : sourceIt->second.m_delegate->handleError("Rialto dropped a frame that failed to decrypt",
1171 : GST_STREAM_ERROR_DECRYPT);
1172 : }
1173 : else
1174 : {
1175 6 : sourceIt->second.m_delegate->handleError("Rialto server playback failed");
1176 : }
1177 :
1178 4 : result = true;
1179 : }
1180 : });
1181 :
1182 7 : return result;
1183 : }
1184 :
1185 6 : firebolt::rialto::AddSegmentStatus GStreamerMSEMediaPlayerClient::addSegment(
1186 : unsigned int needDataRequestId, const std::unique_ptr<firebolt::rialto::IMediaPipeline::MediaSegment> &mediaSegment)
1187 : {
1188 : // rialto client's addSegment call is MT safe, so it's ok to call it from the Puller's thread
1189 6 : return m_clientBackend->addSegment(needDataRequestId, mediaSegment);
1190 : }
1191 :
1192 225 : BufferPuller::BufferPuller(const std::shared_ptr<IMessageQueueFactory> &messageQueueFactory, GstElement *rialtoSink,
1193 : const std::shared_ptr<BufferParser> &bufferParser,
1194 225 : const std::shared_ptr<IPullModePlaybackDelegate> &delegate)
1195 225 : : m_queue{messageQueueFactory->createMessageQueue()}, m_rialtoSink(rialtoSink), m_bufferParser(bufferParser),
1196 225 : m_delegate{delegate}
1197 : {
1198 : }
1199 :
1200 225 : void BufferPuller::start()
1201 : {
1202 225 : m_queue->start();
1203 : }
1204 :
1205 74 : void BufferPuller::stop()
1206 : {
1207 74 : m_queue->stop();
1208 : }
1209 :
1210 10 : bool BufferPuller::requestPullBuffer(int sourceId, size_t frameCount, unsigned int needDataRequestId,
1211 : GStreamerMSEMediaPlayerClient *player)
1212 : {
1213 30 : return m_queue->postMessage(std::make_shared<PullBufferMessage>(sourceId, frameCount, needDataRequestId, m_rialtoSink,
1214 30 : m_bufferParser, *m_queue, player, m_delegate));
1215 : }
1216 :
1217 11 : HaveDataMessage::HaveDataMessage(firebolt::rialto::MediaSourceStatus status, int sourceId,
1218 11 : unsigned int needDataRequestId, GStreamerMSEMediaPlayerClient *player)
1219 11 : : m_status(status), m_sourceId(sourceId), m_needDataRequestId(needDataRequestId), m_player(player)
1220 : {
1221 : }
1222 :
1223 11 : void HaveDataMessage::handle()
1224 : {
1225 11 : if (m_player->m_attachedSources.find(m_sourceId) == m_player->m_attachedSources.end())
1226 : {
1227 1 : GST_WARNING("Source id %d is invalid", m_sourceId);
1228 1 : return;
1229 : }
1230 :
1231 10 : m_player->m_clientBackend->haveData(m_status, m_needDataRequestId);
1232 : }
1233 :
1234 10 : PullBufferMessage::PullBufferMessage(int sourceId, size_t frameCount, unsigned int needDataRequestId,
1235 : GstElement *rialtoSink, const std::shared_ptr<BufferParser> &bufferParser,
1236 : IMessageQueue &pullerQueue, GStreamerMSEMediaPlayerClient *player,
1237 10 : const std::shared_ptr<IPullModePlaybackDelegate> &delegate)
1238 10 : : m_sourceId(sourceId), m_frameCount(frameCount), m_needDataRequestId(needDataRequestId), m_rialtoSink(rialtoSink),
1239 10 : m_bufferParser(bufferParser), m_pullerQueue(pullerQueue), m_player(player), m_delegate{delegate}
1240 : {
1241 : }
1242 :
1243 9 : void PullBufferMessage::handle()
1244 : {
1245 9 : bool isEos = false;
1246 9 : bool isNoSpace{false};
1247 9 : unsigned int addedSegments = 0;
1248 :
1249 12 : for (unsigned int frame = 0; frame < m_frameCount; ++frame)
1250 : {
1251 10 : if (!m_delegate->isReadyToSendData())
1252 : {
1253 2 : GST_INFO_OBJECT(m_rialtoSink, "Not ready to send data - segment or eos not received yet");
1254 7 : break;
1255 : }
1256 8 : GstRefSample sample = m_delegate->getFrontSample();
1257 8 : if (!sample)
1258 : {
1259 3 : if (m_delegate->isEos())
1260 : {
1261 1 : isEos = true;
1262 : }
1263 : else
1264 : {
1265 : // it's not a critical issue. It might be caused by receiving too many need data requests.
1266 2 : GST_INFO_OBJECT(m_rialtoSink, "Could not get a sample");
1267 : }
1268 3 : break;
1269 : }
1270 :
1271 : // we pass GstMapInfo's pointers on data buffers to RialtoClient
1272 : // so we need to hold it until RialtoClient copies them to shm
1273 5 : GstBuffer *buffer = sample.getBuffer();
1274 : GstMapInfo map;
1275 5 : if (!gst_buffer_map(buffer, &map, GST_MAP_READ))
1276 : {
1277 0 : GST_ERROR_OBJECT(m_rialtoSink, "Could not map buffer");
1278 0 : m_delegate->popSample();
1279 0 : continue;
1280 : }
1281 :
1282 : std::unique_ptr<firebolt::rialto::IMediaPipeline::MediaSegment> mseData =
1283 5 : m_bufferParser->parseBuffer(sample, buffer, map, m_sourceId);
1284 5 : if (!mseData)
1285 : {
1286 0 : GST_ERROR_OBJECT(m_rialtoSink, "No data returned from the parser");
1287 0 : gst_buffer_unmap(buffer, &map);
1288 0 : m_delegate->popSample();
1289 0 : continue;
1290 : }
1291 :
1292 5 : firebolt::rialto::AddSegmentStatus addSegmentStatus = m_player->addSegment(m_needDataRequestId, mseData);
1293 5 : if (addSegmentStatus == firebolt::rialto::AddSegmentStatus::NO_SPACE)
1294 : {
1295 2 : gst_buffer_unmap(buffer, &map);
1296 2 : GST_INFO_OBJECT(m_rialtoSink, "There's no space to add sample");
1297 2 : isNoSpace = true;
1298 2 : break;
1299 : }
1300 :
1301 3 : gst_buffer_unmap(buffer, &map);
1302 3 : m_delegate->popSample();
1303 3 : addedSegments++;
1304 10 : }
1305 :
1306 9 : firebolt::rialto::MediaSourceStatus status = firebolt::rialto::MediaSourceStatus::OK;
1307 9 : if (isEos)
1308 : {
1309 1 : status = firebolt::rialto::MediaSourceStatus::EOS;
1310 : }
1311 8 : else if (addedSegments == 0)
1312 : {
1313 5 : status = firebolt::rialto::MediaSourceStatus::NO_AVAILABLE_SAMPLES;
1314 : }
1315 3 : else if (addedSegments != m_frameCount && isNoSpace)
1316 : {
1317 1 : status = firebolt::rialto::MediaSourceStatus::NO_SPACE_FOR_SAMPLES;
1318 : }
1319 9 : if (firebolt::rialto::MediaSourceStatus::OK == status || firebolt::rialto::MediaSourceStatus::EOS == status)
1320 : {
1321 3 : m_player->getFlushAndDataSynchronizer().notifyDataPushed(m_sourceId);
1322 : }
1323 :
1324 18 : m_player->m_backendQueue->postMessage(
1325 18 : std::make_shared<HaveDataMessage>(status, m_sourceId, m_needDataRequestId, m_player));
1326 9 : }
1327 :
1328 11 : NeedDataMessage::NeedDataMessage(int sourceId, size_t frameCount, unsigned int needDataRequestId,
1329 11 : GStreamerMSEMediaPlayerClient *player)
1330 11 : : m_sourceId(sourceId), m_frameCount(frameCount), m_needDataRequestId(needDataRequestId), m_player(player)
1331 : {
1332 : }
1333 :
1334 11 : void NeedDataMessage::handle()
1335 : {
1336 11 : if (!m_player->requestPullBuffer(m_sourceId, m_frameCount, m_needDataRequestId))
1337 : {
1338 2 : GST_ERROR("Failed to pull buffer for sourceId=%d and NeedDataRequestId %u", m_sourceId, m_needDataRequestId);
1339 4 : m_player->m_backendQueue->postMessage(
1340 2 : std::make_shared<HaveDataMessage>(firebolt::rialto::MediaSourceStatus::ERROR, m_sourceId,
1341 2 : m_needDataRequestId, m_player));
1342 : }
1343 11 : }
1344 :
1345 61 : PlaybackStateMessage::PlaybackStateMessage(firebolt::rialto::PlaybackState state, GStreamerMSEMediaPlayerClient *player)
1346 61 : : m_state(state), m_player(player)
1347 : {
1348 : }
1349 :
1350 61 : void PlaybackStateMessage::handle()
1351 : {
1352 61 : m_player->handlePlaybackStateChange(m_state);
1353 : }
1354 :
1355 5 : QosMessage::QosMessage(int sourceId, firebolt::rialto::QosInfo qosInfo, GStreamerMSEMediaPlayerClient *player)
1356 5 : : m_sourceId(sourceId), m_qosInfo(qosInfo), m_player(player)
1357 : {
1358 : }
1359 :
1360 5 : void QosMessage::handle()
1361 : {
1362 5 : if (!m_player->handleQos(m_sourceId, m_qosInfo))
1363 : {
1364 1 : GST_ERROR("Failed to handle qos for sourceId=%d", m_sourceId);
1365 : }
1366 5 : }
1367 :
1368 2 : BufferUnderflowMessage::BufferUnderflowMessage(int sourceId, GStreamerMSEMediaPlayerClient *player)
1369 2 : : m_sourceId(sourceId), m_player(player)
1370 : {
1371 : }
1372 :
1373 2 : void BufferUnderflowMessage::handle()
1374 : {
1375 2 : if (!m_player->handleBufferUnderflow(m_sourceId))
1376 : {
1377 1 : GST_ERROR("Failed to handle buffer underflow for sourceId=%d", m_sourceId);
1378 : }
1379 2 : }
1380 :
1381 3 : FirstFrameReceivedMessage::FirstFrameReceivedMessage(int sourceId, GStreamerMSEMediaPlayerClient *player)
1382 3 : : m_sourceId(sourceId), m_player(player)
1383 : {
1384 : }
1385 :
1386 3 : void FirstFrameReceivedMessage::handle()
1387 : {
1388 3 : if (!m_player->handleFirstFrameReceived(m_sourceId))
1389 : {
1390 1 : GST_ERROR("Failed to handle first frame received for sourceId=%d", m_sourceId);
1391 : }
1392 3 : }
1393 :
1394 7 : PlaybackErrorMessage::PlaybackErrorMessage(int sourceId, firebolt::rialto::PlaybackError error,
1395 7 : GStreamerMSEMediaPlayerClient *player)
1396 7 : : m_sourceId(sourceId), m_error(error), m_player(player)
1397 : {
1398 : }
1399 :
1400 7 : void PlaybackErrorMessage::handle()
1401 : {
1402 7 : if (!m_player->handlePlaybackError(m_sourceId, m_error))
1403 : {
1404 2 : GST_ERROR("Failed to handle playback error for sourceId=%d, error %s", m_sourceId, toString(m_error));
1405 : }
1406 7 : }
1407 :
1408 1 : SetPositionMessage::SetPositionMessage(int64_t newPosition, std::unordered_map<int32_t, AttachedSource> &attachedSources)
1409 1 : : m_newPosition(newPosition), m_attachedSources(attachedSources)
1410 : {
1411 : }
1412 :
1413 1 : void SetPositionMessage::handle()
1414 : {
1415 2 : for (auto &source : m_attachedSources)
1416 : {
1417 1 : source.second.setPosition(m_newPosition);
1418 : }
1419 : }
1420 :
1421 1 : SetDurationMessage::SetDurationMessage(int64_t newDuration, int64_t &targetDuration)
1422 1 : : m_newDuration(newDuration), m_targetDuration(targetDuration)
1423 : {
1424 : }
1425 :
1426 1 : void SetDurationMessage::handle()
1427 : {
1428 1 : m_targetDuration = m_newDuration;
1429 : }
1430 :
1431 9 : SourceFlushedMessage::SourceFlushedMessage(int32_t sourceId, GStreamerMSEMediaPlayerClient *player)
1432 9 : : m_sourceId{sourceId}, m_player{player}
1433 : {
1434 : }
1435 :
1436 9 : void SourceFlushedMessage::handle()
1437 : {
1438 9 : m_player->handleSourceFlushed(m_sourceId);
1439 : }
|
