Line data Source code
1 : /*
2 : * Copyright (C) 2025 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 "PullModePlaybackDelegate.h"
20 : #include "ControlBackend.h"
21 : #include "GstreamerCatLog.h"
22 : #include "RialtoGStreamerMSEBaseSink.h"
23 : #include "RialtoGStreamerMSEBaseSinkPrivate.h"
24 :
25 : #define GST_CAT_DEFAULT rialtoGStreamerCat
26 :
27 : namespace
28 : {
29 328 : GstObject *getOldestGstBinParent(GstElement *element)
30 : {
31 328 : GstObject *parent = gst_object_get_parent(GST_OBJECT_CAST(element));
32 328 : GstObject *result = GST_OBJECT_CAST(element);
33 328 : if (parent)
34 : {
35 164 : if (GST_IS_BIN(parent))
36 : {
37 164 : result = getOldestGstBinParent(GST_ELEMENT_CAST(parent));
38 : }
39 164 : gst_object_unref(parent);
40 : }
41 :
42 328 : return result;
43 : }
44 :
45 6 : unsigned getGstPlayFlag(const char *nick)
46 : {
47 6 : GFlagsClass *flagsClass = static_cast<GFlagsClass *>(g_type_class_ref(g_type_from_name("GstPlayFlags")));
48 6 : GFlagsValue *flag = g_flags_get_value_by_nick(flagsClass, nick);
49 6 : const unsigned kResult{flag ? flag->value : 0};
50 6 : g_type_class_unref(flagsClass);
51 6 : return kResult;
52 : }
53 :
54 157 : bool getNStreamsFromParent(GstObject *parentObject, gint &n_video, gint &n_audio, gint &n_text)
55 : {
56 157 : if (g_object_class_find_property(G_OBJECT_GET_CLASS(parentObject), "n-video") &&
57 159 : g_object_class_find_property(G_OBJECT_GET_CLASS(parentObject), "n-audio") &&
58 2 : g_object_class_find_property(G_OBJECT_GET_CLASS(parentObject), "n-text"))
59 : {
60 2 : g_object_get(parentObject, "n-video", &n_video, "n-audio", &n_audio, "n-text", &n_text, nullptr);
61 :
62 2 : if (g_object_class_find_property(G_OBJECT_GET_CLASS(parentObject), "flags"))
63 : {
64 2 : guint flags = 0;
65 2 : g_object_get(parentObject, "flags", &flags, nullptr);
66 2 : n_video = (flags & getGstPlayFlag("video")) ? n_video : 0;
67 2 : n_audio = (flags & getGstPlayFlag("audio")) ? n_audio : 0;
68 2 : n_text = (flags & getGstPlayFlag("text")) ? n_text : 0;
69 : }
70 :
71 2 : return true;
72 : }
73 :
74 155 : return false;
75 : }
76 : } // namespace
77 :
78 306 : PullModePlaybackDelegate::PullModePlaybackDelegate(GstElement *sink) : m_sink{sink}
79 : {
80 306 : m_sinkPad = RIALTO_MSE_BASE_SINK(sink)->priv->m_sinkPad;
81 306 : gst_segment_init(&m_lastSegment, GST_FORMAT_TIME);
82 : }
83 :
84 306 : void PullModePlaybackDelegate::createControlBackend()
85 : {
86 306 : if (!m_rialtoControlClient)
87 : {
88 306 : m_rialtoControlClient = std::make_unique<firebolt::rialto::client::ControlBackend>(shared_from_this());
89 : }
90 : }
91 :
92 306 : PullModePlaybackDelegate::~PullModePlaybackDelegate()
93 : {
94 306 : if (m_caps)
95 155 : gst_caps_unref(m_caps);
96 306 : clearBuffersUnlocked();
97 : }
98 :
99 483 : void PullModePlaybackDelegate::clearBuffersUnlocked()
100 : {
101 483 : m_isSinkFlushOngoing = true;
102 483 : m_needDataCondVariable.notify_all();
103 514 : while (!m_samples.empty())
104 : {
105 31 : GstSample *sample = m_samples.front();
106 31 : m_samples.pop();
107 31 : gst_sample_unref(sample);
108 : }
109 483 : setLastBuffer(nullptr);
110 : }
111 :
112 151 : void PullModePlaybackDelegate::setSourceId(int32_t sourceId)
113 : {
114 151 : m_sourceId = sourceId;
115 : }
116 :
117 3 : void PullModePlaybackDelegate::handleEos()
118 : {
119 3 : GstState currentState = GST_STATE(m_sink);
120 3 : if ((currentState != GST_STATE_PAUSED) && (currentState != GST_STATE_PLAYING))
121 : {
122 1 : GST_ERROR_OBJECT(m_sink, "Sink cannot post a EOS message in state '%s', posting an error instead",
123 : gst_element_state_get_name(currentState));
124 :
125 1 : const char *errMessage = "Rialto sinks received EOS in non-playing state";
126 1 : GError *gError{g_error_new_literal(GST_STREAM_ERROR, 0, errMessage)};
127 1 : gst_element_post_message(m_sink, gst_message_new_error(GST_OBJECT_CAST(m_sink), gError, errMessage));
128 1 : g_error_free(gError);
129 : }
130 : else
131 : {
132 2 : std::unique_lock lock{m_sinkMutex};
133 2 : if (!m_isSinkFlushOngoing && !m_isServerFlushOngoing)
134 : {
135 1 : gst_element_post_message(m_sink, gst_message_new_eos(GST_OBJECT_CAST(m_sink)));
136 : }
137 : else
138 : {
139 1 : GST_WARNING_OBJECT(m_sink, "Skip sending eos message - flush is ongoing...");
140 : }
141 2 : }
142 3 : }
143 :
144 5 : void PullModePlaybackDelegate::handleFlushCompleted()
145 : {
146 5 : GST_INFO_OBJECT(m_sink, "Flush completed");
147 5 : std::unique_lock<std::mutex> lock(m_sinkMutex);
148 5 : m_isServerFlushOngoing = false;
149 5 : m_isTimeResetOngoing = false;
150 : }
151 :
152 40 : void PullModePlaybackDelegate::handleStateChanged(firebolt::rialto::PlaybackState state)
153 : {
154 40 : GstState current = GST_STATE(m_sink);
155 40 : GstState next = GST_STATE_NEXT(m_sink);
156 40 : GstState pending = GST_STATE_PENDING(m_sink);
157 40 : GstState postNext = next == pending ? GST_STATE_VOID_PENDING : pending;
158 :
159 40 : GST_DEBUG_OBJECT(m_sink,
160 : "Received server's state change to %u. Sink's states are: current state: %s next state: %s "
161 : "pending state: %s, last return state %s",
162 : static_cast<uint32_t>(state), gst_element_state_get_name(current),
163 : gst_element_state_get_name(next), gst_element_state_get_name(pending),
164 : gst_element_state_change_return_get_name(GST_STATE_RETURN(m_sink)));
165 :
166 40 : if (m_isStateCommitNeeded)
167 : {
168 40 : if ((state == firebolt::rialto::PlaybackState::PAUSED && next == GST_STATE_PAUSED) ||
169 7 : (state == firebolt::rialto::PlaybackState::PLAYING && next == GST_STATE_PLAYING))
170 : {
171 39 : GST_STATE(m_sink) = next;
172 39 : GST_STATE_NEXT(m_sink) = postNext;
173 39 : GST_STATE_PENDING(m_sink) = GST_STATE_VOID_PENDING;
174 39 : GST_STATE_RETURN(m_sink) = GST_STATE_CHANGE_SUCCESS;
175 :
176 39 : GST_INFO_OBJECT(m_sink, "Async state transition to state %s done", gst_element_state_get_name(next));
177 :
178 39 : gst_element_post_message(m_sink,
179 39 : gst_message_new_state_changed(GST_OBJECT_CAST(m_sink), current, next, pending));
180 39 : postAsyncDone();
181 : }
182 : /* Immediately transition to PLAYING when prerolled and PLAY is requested */
183 1 : else if (state == firebolt::rialto::PlaybackState::PAUSED && current == GST_STATE_PAUSED &&
184 : next == GST_STATE_PLAYING)
185 : {
186 1 : GST_INFO_OBJECT(m_sink, "Async state transition to PAUSED done. Transitioning to PLAYING");
187 1 : changeState(GST_STATE_CHANGE_PAUSED_TO_PLAYING);
188 : }
189 : }
190 40 : }
191 :
192 950 : GstStateChangeReturn PullModePlaybackDelegate::changeState(GstStateChange transition)
193 : {
194 950 : GstState current_state = GST_STATE_TRANSITION_CURRENT(transition);
195 950 : GstState next_state = GST_STATE_TRANSITION_NEXT(transition);
196 950 : GST_INFO_OBJECT(m_sink, "State change: (%s) -> (%s)", gst_element_state_get_name(current_state),
197 : gst_element_state_get_name(next_state));
198 :
199 950 : GstStateChangeReturn status = GST_STATE_CHANGE_SUCCESS;
200 950 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
201 :
202 950 : switch (transition)
203 : {
204 306 : case GST_STATE_CHANGE_NULL_TO_READY:
205 306 : if (!m_sinkPad)
206 : {
207 0 : GST_ERROR_OBJECT(m_sink, "Cannot start, because there's no sink pad");
208 0 : return GST_STATE_CHANGE_FAILURE;
209 : }
210 306 : if (!m_rialtoControlClient->waitForRunning())
211 : {
212 0 : GST_ERROR_OBJECT(m_sink, "Control: Rialto client cannot reach running state");
213 0 : return GST_STATE_CHANGE_FAILURE;
214 : }
215 306 : GST_INFO_OBJECT(m_sink, "Control: Rialto client reached running state");
216 306 : break;
217 160 : case GST_STATE_CHANGE_READY_TO_PAUSED:
218 : {
219 160 : if (!client)
220 : {
221 0 : GST_ERROR_OBJECT(m_sink, "Cannot get the media player client object");
222 0 : return GST_STATE_CHANGE_FAILURE;
223 : }
224 :
225 160 : client->setStopping(false);
226 :
227 160 : m_isSinkFlushOngoing = false;
228 :
229 160 : StateChangeResult result = client->pause(m_sourceId);
230 160 : if (result == StateChangeResult::SUCCESS_ASYNC || result == StateChangeResult::NOT_ATTACHED)
231 : {
232 : // NOT_ATTACHED is not a problem here, because source will be attached later when GST_EVENT_CAPS is received
233 160 : if (result == StateChangeResult::NOT_ATTACHED)
234 : {
235 160 : postAsyncStart();
236 : }
237 160 : status = GST_STATE_CHANGE_ASYNC;
238 : }
239 :
240 160 : break;
241 : }
242 8 : case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
243 : {
244 8 : if (!client)
245 : {
246 0 : GST_ERROR_OBJECT(m_sink, "Cannot get the media player client object");
247 0 : return GST_STATE_CHANGE_FAILURE;
248 : }
249 :
250 8 : StateChangeResult result = client->play(m_sourceId);
251 8 : if (result == StateChangeResult::SUCCESS_ASYNC)
252 : {
253 8 : status = GST_STATE_CHANGE_ASYNC;
254 : }
255 0 : else if (result == StateChangeResult::NOT_ATTACHED)
256 : {
257 0 : GST_ERROR_OBJECT(m_sink, "Failed to change state to playing");
258 0 : return GST_STATE_CHANGE_FAILURE;
259 : }
260 :
261 8 : break;
262 : }
263 7 : case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
264 : {
265 7 : if (!client)
266 : {
267 0 : GST_ERROR_OBJECT(m_sink, "Cannot get the media player client object");
268 0 : return GST_STATE_CHANGE_FAILURE;
269 : }
270 :
271 7 : StateChangeResult result = client->pause(m_sourceId);
272 7 : if (result == StateChangeResult::SUCCESS_ASYNC)
273 : {
274 7 : status = GST_STATE_CHANGE_ASYNC;
275 : }
276 0 : else if (result == StateChangeResult::NOT_ATTACHED)
277 : {
278 0 : GST_ERROR_OBJECT(m_sink, "Failed to change state to paused");
279 0 : return GST_STATE_CHANGE_FAILURE;
280 : }
281 :
282 7 : break;
283 : }
284 160 : case GST_STATE_CHANGE_PAUSED_TO_READY:
285 160 : if (!client)
286 : {
287 0 : GST_ERROR_OBJECT(m_sink, "Cannot get the media player client object");
288 0 : return GST_STATE_CHANGE_FAILURE;
289 : }
290 :
291 160 : client->setStopping(true);
292 :
293 160 : if (m_isStateCommitNeeded)
294 : {
295 139 : GST_DEBUG_OBJECT(m_sink, "Sending async_done in PAUSED->READY transition");
296 139 : postAsyncDone();
297 : }
298 :
299 160 : client->removeSource(m_sourceId);
300 : {
301 160 : std::lock_guard<std::mutex> lock(m_sinkMutex);
302 160 : clearBuffersUnlocked();
303 160 : m_sourceAttached = false;
304 : }
305 160 : break;
306 305 : case GST_STATE_CHANGE_READY_TO_NULL:
307 : // Playback will be stopped once all sources are finished and ref count
308 : // of the media pipeline object reaches 0
309 305 : m_mediaPlayerManager.releaseMediaPlayerClient();
310 305 : m_rialtoControlClient->removeControlBackend();
311 305 : break;
312 4 : default:
313 4 : break;
314 : }
315 :
316 950 : return status;
317 : }
318 :
319 3 : void PullModePlaybackDelegate::handleError(const std::string &message, gint code)
320 : {
321 3 : GError *gError{g_error_new_literal(GST_STREAM_ERROR, code, message.c_str())};
322 6 : gst_element_post_message(GST_ELEMENT_CAST(m_sink),
323 3 : gst_message_new_error(GST_OBJECT_CAST(m_sink), gError, message.c_str()));
324 3 : g_error_free(gError);
325 : }
326 :
327 2 : void PullModePlaybackDelegate::notifyApplicationState(firebolt::rialto::ApplicationState state)
328 : {
329 2 : if (state == firebolt::rialto::ApplicationState::UNKNOWN)
330 : {
331 1 : GST_WARNING_OBJECT(m_sink, "Rialto control sent unknown application state");
332 3 : handleError("Rialto client reached unknown application state");
333 : }
334 2 : }
335 :
336 326 : void PullModePlaybackDelegate::postAsyncStart()
337 : {
338 326 : m_isStateCommitNeeded = true;
339 326 : gst_element_post_message(GST_ELEMENT_CAST(m_sink), gst_message_new_async_start(GST_OBJECT(m_sink)));
340 : }
341 :
342 178 : void PullModePlaybackDelegate::postAsyncDone()
343 : {
344 178 : m_isStateCommitNeeded = false;
345 178 : gst_element_post_message(m_sink, gst_message_new_async_done(GST_OBJECT_CAST(m_sink), GST_CLOCK_TIME_NONE));
346 : }
347 :
348 344 : void PullModePlaybackDelegate::setProperty(const Property &type, const GValue *value)
349 : {
350 344 : switch (type)
351 : {
352 171 : case Property::IsSinglePathStream:
353 : {
354 171 : std::lock_guard<std::mutex> lock(m_sinkMutex);
355 171 : m_isSinglePathStream = g_value_get_boolean(value) != FALSE;
356 171 : break;
357 : }
358 171 : case Property::NumberOfStreams:
359 : {
360 171 : std::lock_guard<std::mutex> lock(m_sinkMutex);
361 171 : m_numOfStreams = g_value_get_int(value);
362 171 : break;
363 : }
364 1 : case Property::HasDrm:
365 : {
366 1 : std::lock_guard<std::mutex> lock(m_sinkMutex);
367 1 : m_hasDrm = g_value_get_boolean(value) != FALSE;
368 1 : break;
369 : }
370 1 : case Property::EnableLastSample:
371 : {
372 1 : std::lock_guard<std::mutex> lock(m_sinkMutex);
373 1 : m_enableLastSample = g_value_get_boolean(value) != FALSE;
374 1 : if (!m_enableLastSample)
375 : {
376 0 : if (m_lastBuffer)
377 : {
378 0 : gst_buffer_unref(m_lastBuffer);
379 0 : m_lastBuffer = nullptr;
380 : }
381 : }
382 1 : break;
383 : }
384 0 : default:
385 : {
386 0 : break;
387 : }
388 : }
389 344 : }
390 :
391 11 : void PullModePlaybackDelegate::getProperty(const Property &type, GValue *value)
392 : {
393 11 : switch (type)
394 : {
395 1 : case Property::IsSinglePathStream:
396 : {
397 1 : std::lock_guard<std::mutex> lock(m_sinkMutex);
398 1 : g_value_set_boolean(value, m_isSinglePathStream ? TRUE : FALSE);
399 1 : break;
400 : }
401 1 : case Property::NumberOfStreams:
402 : {
403 1 : std::lock_guard<std::mutex> lock(m_sinkMutex);
404 1 : g_value_set_int(value, m_numOfStreams);
405 1 : break;
406 : }
407 1 : case Property::HasDrm:
408 : {
409 1 : std::lock_guard<std::mutex> lock(m_sinkMutex);
410 1 : g_value_set_boolean(value, m_hasDrm ? TRUE : FALSE);
411 1 : break;
412 : }
413 2 : case Property::Stats:
414 : {
415 2 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
416 2 : if (!client)
417 : {
418 1 : GST_ERROR_OBJECT(m_sink, "Could not get the media player client");
419 1 : break;
420 : }
421 :
422 1 : guint64 totalVideoFrames{0};
423 1 : guint64 droppedVideoFrames{0};
424 1 : if (client->getStats(m_sourceId, totalVideoFrames, droppedVideoFrames))
425 : {
426 1 : GstStructure *stats{gst_structure_new("stats", "rendered", G_TYPE_UINT64, totalVideoFrames, "dropped",
427 : G_TYPE_UINT64, droppedVideoFrames, nullptr)};
428 1 : g_value_set_pointer(value, stats);
429 : }
430 : else
431 : {
432 0 : GST_ERROR_OBJECT(m_sink, "No stats returned from client");
433 : }
434 1 : break;
435 2 : }
436 2 : case Property::EnableLastSample:
437 : {
438 2 : std::lock_guard<std::mutex> lock(m_sinkMutex);
439 2 : g_value_set_boolean(value, m_enableLastSample ? TRUE : FALSE);
440 2 : break;
441 : }
442 4 : case Property::LastSample:
443 : {
444 : // Mutex inside getLastSample function
445 4 : gst_value_take_sample(value, getLastSample());
446 4 : break;
447 : }
448 0 : default:
449 : {
450 0 : break;
451 : }
452 : }
453 11 : }
454 :
455 36 : std::optional<gboolean> PullModePlaybackDelegate::handleQuery(GstQuery *query) const
456 : {
457 36 : GST_DEBUG_OBJECT(m_sink, "handling query '%s'", GST_QUERY_TYPE_NAME(query));
458 36 : switch (GST_QUERY_TYPE(query))
459 : {
460 5 : case GST_QUERY_SEEKING:
461 : {
462 5 : if (m_sinkPad && gst_pad_peer_query(m_sinkPad, query))
463 : {
464 2 : return TRUE;
465 : }
466 : GstFormat fmt;
467 3 : gst_query_parse_seeking(query, &fmt, NULL, NULL, NULL);
468 3 : gst_query_set_seeking(query, fmt, FALSE, 0, -1);
469 3 : return TRUE;
470 : }
471 6 : case GST_QUERY_POSITION:
472 : {
473 6 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
474 6 : if (!client)
475 : {
476 1 : return FALSE;
477 : }
478 : {
479 5 : std::unique_lock<std::mutex> lock(m_sinkMutex);
480 5 : if (m_isServerFlushOngoing && m_isTimeResetOngoing)
481 : {
482 1 : GST_WARNING_OBJECT(m_sink, "Position query during server flush and time reset, returning FALSE");
483 1 : return FALSE;
484 : }
485 5 : }
486 :
487 : GstFormat fmt;
488 4 : gst_query_parse_position(query, &fmt, NULL);
489 4 : switch (fmt)
490 : {
491 3 : case GST_FORMAT_TIME:
492 : {
493 3 : gint64 position = client->getPosition(m_sourceId);
494 3 : GST_DEBUG_OBJECT(m_sink, "Queried position is %" GST_TIME_FORMAT, GST_TIME_ARGS(position));
495 3 : if (position < 0)
496 : {
497 2 : return FALSE;
498 : }
499 :
500 1 : gst_query_set_position(query, fmt, position);
501 1 : break;
502 : }
503 1 : default:
504 1 : break;
505 : }
506 2 : return TRUE;
507 6 : }
508 2 : case GST_QUERY_SEGMENT:
509 : {
510 2 : std::lock_guard<std::mutex> lock(m_sinkMutex);
511 2 : GstFormat format{m_lastSegment.format};
512 2 : gint64 start{static_cast<gint64>(gst_segment_to_stream_time(&m_lastSegment, format, m_lastSegment.start))};
513 2 : gint64 stop{0};
514 2 : if (m_lastSegment.stop == GST_CLOCK_TIME_NONE)
515 : {
516 2 : stop = m_lastSegment.duration;
517 : }
518 : else
519 : {
520 0 : stop = gst_segment_to_stream_time(&m_lastSegment, format, m_lastSegment.stop);
521 : }
522 2 : gst_query_set_segment(query, m_lastSegment.rate, format, start, stop);
523 2 : return TRUE;
524 : }
525 9 : case GST_QUERY_DURATION:
526 : {
527 : GstFormat fmt;
528 9 : gst_query_parse_duration(query, &fmt, NULL);
529 9 : if (GST_FORMAT_TIME != fmt)
530 : {
531 1 : return FALSE;
532 : }
533 :
534 8 : if (m_sinkPad && gst_pad_peer_query(m_sinkPad, query))
535 : {
536 4 : gint64 upstreamDuration{-1};
537 4 : gst_query_parse_duration(query, nullptr, &upstreamDuration);
538 4 : if (upstreamDuration > 0)
539 : {
540 2 : GST_DEBUG_OBJECT(m_sink, "Duration from upstream is %" GST_TIME_FORMAT, GST_TIME_ARGS(upstreamDuration));
541 2 : return TRUE;
542 : }
543 : }
544 :
545 6 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
546 6 : if (!client)
547 : {
548 2 : return FALSE;
549 : }
550 4 : int64_t duration{-1};
551 4 : if (!client->getDuration(duration))
552 : {
553 1 : return FALSE;
554 : }
555 3 : gst_query_set_duration(query, fmt, duration);
556 3 : return TRUE;
557 6 : }
558 14 : default:
559 14 : break;
560 : }
561 14 : return std::nullopt;
562 : }
563 :
564 29 : gboolean PullModePlaybackDelegate::handleSendEvent(GstEvent *event)
565 : {
566 29 : GST_DEBUG_OBJECT(m_sink, "handling event '%s'", GST_EVENT_TYPE_NAME(event));
567 29 : bool shouldForwardUpstream = GST_EVENT_IS_UPSTREAM(event);
568 :
569 29 : switch (GST_EVENT_TYPE(event))
570 : {
571 13 : case GST_EVENT_SEEK:
572 : {
573 13 : gdouble rate{1.0};
574 13 : GstFormat seekFormat{GST_FORMAT_UNDEFINED};
575 13 : GstSeekFlags flags{GST_SEEK_FLAG_NONE};
576 13 : GstSeekType startType{GST_SEEK_TYPE_NONE}, stopType{GST_SEEK_TYPE_NONE};
577 13 : gint64 start{0}, stop{0};
578 13 : gst_event_parse_seek(event, &rate, &seekFormat, &flags, &startType, &start, &stopType, &stop);
579 :
580 13 : if (flags & GST_SEEK_FLAG_FLUSH)
581 : {
582 9 : if (seekFormat == GST_FORMAT_TIME && startType == GST_SEEK_TYPE_END)
583 : {
584 1 : GST_ERROR_OBJECT(m_sink, "GST_SEEK_TYPE_END seek is not supported");
585 1 : gst_event_unref(event);
586 5 : return FALSE;
587 : }
588 : // Update last segment
589 8 : if (seekFormat == GST_FORMAT_TIME)
590 : {
591 7 : gboolean update{FALSE};
592 7 : std::lock_guard<std::mutex> lock(m_sinkMutex);
593 7 : gst_segment_do_seek(&m_lastSegment, rate, seekFormat, flags, startType, start, stopType, stop, &update);
594 : }
595 : }
596 : #if GST_CHECK_VERSION(1, 18, 0)
597 4 : else if (flags & GST_SEEK_FLAG_INSTANT_RATE_CHANGE)
598 : {
599 2 : gdouble rateMultiplier = rate / m_lastSegment.rate;
600 2 : GstEvent *rateChangeEvent = gst_event_new_instant_rate_change(rateMultiplier, (GstSegmentFlags)flags);
601 2 : gst_event_set_seqnum(rateChangeEvent, gst_event_get_seqnum(event));
602 2 : gst_event_unref(event);
603 2 : if (gst_pad_send_event(m_sinkPad, rateChangeEvent) != TRUE)
604 : {
605 1 : GST_ERROR_OBJECT(m_sink, "Sending instant rate change failed.");
606 1 : return FALSE;
607 : }
608 1 : return TRUE;
609 : }
610 : #endif
611 : else
612 : {
613 2 : GST_WARNING_OBJECT(m_sink, "Seek with flags 0x%X is not supported", flags);
614 2 : gst_event_unref(event);
615 2 : return FALSE;
616 : }
617 8 : break;
618 : }
619 : #if GST_CHECK_VERSION(1, 18, 0)
620 2 : case GST_EVENT_INSTANT_RATE_SYNC_TIME:
621 : {
622 2 : double rate{0.0};
623 2 : GstClockTime runningTime{GST_CLOCK_TIME_NONE}, upstreamRunningTime{GST_CLOCK_TIME_NONE};
624 2 : guint32 seqnum = gst_event_get_seqnum(event);
625 2 : gst_event_parse_instant_rate_sync_time(event, &rate, &runningTime, &upstreamRunningTime);
626 :
627 2 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
628 2 : if ((client) && (m_mediaPlayerManager.hasControl()))
629 : {
630 2 : GST_DEBUG_OBJECT(m_sink, "Instant playback rate change: %.2f", rate);
631 2 : m_currentInstantRateChangeSeqnum = seqnum;
632 2 : client->setPlaybackRate(rate);
633 : }
634 2 : break;
635 : }
636 : #endif
637 14 : default:
638 14 : break;
639 : }
640 :
641 24 : if (shouldForwardUpstream)
642 : {
643 24 : bool result = gst_pad_push_event(m_sinkPad, event);
644 24 : if (!result)
645 : {
646 10 : GST_DEBUG_OBJECT(m_sink, "forwarding upstream event '%s' failed", GST_EVENT_TYPE_NAME(event));
647 : }
648 :
649 24 : return result;
650 : }
651 :
652 0 : gst_event_unref(event);
653 0 : return TRUE;
654 : }
655 :
656 213 : gboolean PullModePlaybackDelegate::handleEvent(GstPad *pad, GstObject *parent, GstEvent *event)
657 : {
658 213 : GST_DEBUG_OBJECT(m_sink, "handling event %" GST_PTR_FORMAT, event);
659 213 : switch (GST_EVENT_TYPE(event))
660 : {
661 7 : case GST_EVENT_SEGMENT:
662 : {
663 7 : copySegment(event);
664 7 : setSegment();
665 7 : break;
666 : }
667 3 : case GST_EVENT_EOS:
668 : {
669 3 : std::lock_guard<std::mutex> lock(m_sinkMutex);
670 3 : m_isEos = true;
671 3 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
672 3 : if (client)
673 : {
674 0 : client->getFlushAndDataSynchronizer().notifyDataReceived(m_sourceId);
675 : }
676 3 : break;
677 : }
678 159 : case GST_EVENT_CAPS:
679 : {
680 : GstCaps *caps;
681 159 : gst_event_parse_caps(event, &caps);
682 : {
683 159 : std::lock_guard<std::mutex> lock(m_sinkMutex);
684 159 : if (m_caps)
685 : {
686 4 : if (!gst_caps_is_equal(caps, m_caps))
687 : {
688 1 : gst_caps_unref(m_caps);
689 1 : m_caps = gst_caps_copy(caps);
690 : }
691 : }
692 : else
693 : {
694 155 : m_caps = gst_caps_copy(caps);
695 : }
696 159 : }
697 159 : break;
698 : }
699 1 : case GST_EVENT_SINK_MESSAGE:
700 : {
701 1 : GstMessage *message = nullptr;
702 1 : gst_event_parse_sink_message(event, &message);
703 :
704 1 : if (message)
705 : {
706 1 : gst_element_post_message(m_sink, message);
707 : }
708 :
709 1 : break;
710 : }
711 8 : case GST_EVENT_CUSTOM_DOWNSTREAM:
712 : case GST_EVENT_CUSTOM_DOWNSTREAM_OOB:
713 : {
714 8 : if (gst_event_has_name(event, "custom-instant-rate-change"))
715 : {
716 2 : GST_DEBUG_OBJECT(m_sink, "Change rate event received");
717 2 : changePlaybackRate(event);
718 : }
719 8 : break;
720 : }
721 17 : case GST_EVENT_FLUSH_START:
722 : {
723 17 : startFlushing();
724 17 : break;
725 : }
726 8 : case GST_EVENT_FLUSH_STOP:
727 : {
728 8 : gboolean resetTime{FALSE};
729 8 : gst_event_parse_flush_stop(event, &resetTime);
730 :
731 8 : stopFlushing(resetTime);
732 8 : break;
733 : }
734 3 : case GST_EVENT_STREAM_COLLECTION:
735 : {
736 3 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
737 3 : if (!client)
738 : {
739 1 : gst_event_unref(event);
740 1 : return FALSE;
741 : }
742 2 : int32_t videoStreams{0}, audioStreams{0}, textStreams{0};
743 2 : GstStreamCollection *streamCollection{nullptr};
744 2 : gst_event_parse_stream_collection(event, &streamCollection);
745 2 : guint streamsSize = gst_stream_collection_get_size(streamCollection);
746 6 : for (guint i = 0; i < streamsSize; ++i)
747 : {
748 4 : auto *stream = gst_stream_collection_get_stream(streamCollection, i);
749 4 : auto type = gst_stream_get_stream_type(stream);
750 4 : if (type & GST_STREAM_TYPE_AUDIO)
751 : {
752 2 : ++audioStreams;
753 : }
754 2 : else if (type & GST_STREAM_TYPE_VIDEO)
755 : {
756 1 : ++videoStreams;
757 : }
758 1 : else if (type & GST_STREAM_TYPE_TEXT)
759 : {
760 1 : ++textStreams;
761 : }
762 : }
763 2 : gst_object_unref(streamCollection);
764 2 : client->handleStreamCollection(audioStreams, videoStreams, textStreams);
765 2 : client->sendAllSourcesAttachedIfPossible();
766 2 : break;
767 3 : }
768 : #if GST_CHECK_VERSION(1, 18, 0)
769 4 : case GST_EVENT_INSTANT_RATE_CHANGE:
770 : {
771 4 : guint32 seqnum = gst_event_get_seqnum(event);
772 7 : if (m_lastInstantRateChangeSeqnum == seqnum || m_currentInstantRateChangeSeqnum.load() == seqnum)
773 : {
774 : /* Ignore if we already received the instant-rate-sync-time event from the pipeline */
775 2 : GST_DEBUG_OBJECT(m_sink, "Instant rate change event with seqnum %u already handled. Ignoring...", seqnum);
776 2 : break;
777 : }
778 :
779 2 : m_lastInstantRateChangeSeqnum = seqnum;
780 2 : gdouble rate{0.0};
781 2 : GstSegmentFlags flags{GST_SEGMENT_FLAG_NONE};
782 2 : gst_event_parse_instant_rate_change(event, &rate, &flags);
783 2 : GstMessage *msg = gst_message_new_instant_rate_request(GST_OBJECT_CAST(m_sink), rate);
784 2 : gst_message_set_seqnum(msg, seqnum);
785 2 : gst_element_post_message(m_sink, msg);
786 2 : break;
787 : }
788 : #endif
789 3 : default:
790 3 : break;
791 : }
792 :
793 212 : gst_event_unref(event);
794 :
795 212 : return TRUE;
796 : }
797 :
798 7 : void PullModePlaybackDelegate::copySegment(GstEvent *event)
799 : {
800 7 : std::lock_guard<std::mutex> lock(m_sinkMutex);
801 7 : gst_event_copy_segment(event, &m_lastSegment);
802 : }
803 :
804 7 : void PullModePlaybackDelegate::setSegment()
805 : {
806 7 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
807 7 : if (!client)
808 : {
809 1 : GST_ERROR_OBJECT(m_sink, "Could not get the media player client");
810 1 : return;
811 : }
812 6 : const bool kResetTime{m_lastSegment.flags == GST_SEGMENT_FLAG_RESET};
813 6 : int64_t position = static_cast<int64_t>(m_lastSegment.start);
814 6 : client->setSourcePosition(m_sourceId, position, kResetTime, m_lastSegment.applied_rate, m_lastSegment.stop);
815 6 : m_segmentSet = true;
816 7 : }
817 :
818 2 : void PullModePlaybackDelegate::changePlaybackRate(GstEvent *event)
819 : {
820 2 : const GstStructure *structure{gst_event_get_structure(event)};
821 2 : gdouble playbackRate{1.0};
822 2 : if (gst_structure_get_double(structure, "rate", &playbackRate) == TRUE)
823 : {
824 2 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
825 2 : if (client && m_mediaPlayerManager.hasControl())
826 : {
827 1 : GST_DEBUG_OBJECT(m_sink, "Instant playback rate change: %.2f", playbackRate);
828 1 : client->setPlaybackRate(playbackRate);
829 : }
830 2 : }
831 : }
832 :
833 17 : void PullModePlaybackDelegate::startFlushing()
834 : {
835 17 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
836 17 : if (client)
837 : {
838 5 : client->getFlushAndDataSynchronizer().waitIfRequired(m_sourceId);
839 : }
840 17 : std::lock_guard<std::mutex> lock(m_sinkMutex);
841 17 : if (!m_isSinkFlushOngoing)
842 : {
843 17 : GST_INFO_OBJECT(m_sink, "Starting flushing");
844 17 : if (m_isEos)
845 : {
846 2 : GST_DEBUG_OBJECT(m_sink, "Flush will clear EOS state.");
847 2 : m_isEos = false;
848 : }
849 17 : m_isSinkFlushOngoing = true;
850 17 : m_segmentSet = false;
851 17 : clearBuffersUnlocked();
852 : }
853 : }
854 :
855 8 : void PullModePlaybackDelegate::stopFlushing(bool resetTime)
856 : {
857 8 : GST_INFO_OBJECT(m_sink, "Stopping flushing");
858 8 : flushServer(resetTime);
859 8 : std::lock_guard<std::mutex> lock(m_sinkMutex);
860 8 : m_isSinkFlushOngoing = false;
861 :
862 8 : if (resetTime)
863 : {
864 8 : GST_DEBUG_OBJECT(m_sink, "sending reset_time message");
865 8 : gst_element_post_message(m_sink, gst_message_new_reset_time(GST_OBJECT_CAST(m_sink), 0));
866 : }
867 : }
868 :
869 8 : void PullModePlaybackDelegate::flushServer(bool resetTime)
870 : {
871 8 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
872 8 : if (!client)
873 : {
874 1 : GST_ERROR_OBJECT(m_sink, "Could not get the media player client");
875 1 : return;
876 : }
877 :
878 : {
879 7 : std::unique_lock<std::mutex> lock(m_sinkMutex);
880 7 : m_isServerFlushOngoing = true;
881 7 : m_isTimeResetOngoing = true;
882 : }
883 7 : client->flush(m_sourceId, resetTime);
884 8 : }
885 :
886 35 : GstFlowReturn PullModePlaybackDelegate::handleBuffer(GstBuffer *buffer)
887 : {
888 35 : constexpr size_t kMaxInternalBuffersQueueSize = 24;
889 35 : GST_LOG_OBJECT(m_sink, "Handling buffer %p with PTS %" GST_TIME_FORMAT, buffer,
890 : GST_TIME_ARGS(GST_BUFFER_PTS(buffer)));
891 :
892 35 : std::unique_lock<std::mutex> lock(m_sinkMutex);
893 :
894 35 : const auto maxSize = kMaxInternalBuffersQueueSize;
895 35 : if (m_samples.size() >= maxSize)
896 : {
897 1 : GST_DEBUG_OBJECT(m_sink, "Waiting for more space in buffers queue");
898 3 : m_needDataCondVariable.wait(lock, [this, maxSize]() { return m_samples.size() < maxSize; });
899 : }
900 :
901 35 : if (m_isSinkFlushOngoing)
902 : {
903 3 : GST_DEBUG_OBJECT(m_sink, "Discarding buffer which was received during flushing");
904 3 : gst_buffer_unref(buffer);
905 3 : return GST_FLOW_FLUSHING;
906 : }
907 :
908 32 : GstSample *sample = gst_sample_new(buffer, m_caps, &m_lastSegment, nullptr);
909 32 : if (sample)
910 32 : m_samples.push(sample);
911 : else
912 0 : GST_ERROR_OBJECT(m_sink, "Failed to create a sample");
913 :
914 32 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
915 32 : if (client)
916 : {
917 28 : client->getFlushAndDataSynchronizer().notifyDataReceived(m_sourceId);
918 : }
919 :
920 32 : setLastBuffer(buffer);
921 :
922 32 : gst_buffer_unref(buffer);
923 :
924 32 : return GST_FLOW_OK;
925 35 : }
926 :
927 1 : GstRefSample PullModePlaybackDelegate::getFrontSample() const
928 : {
929 1 : std::lock_guard<std::mutex> lock(m_sinkMutex);
930 1 : if (m_isServerFlushOngoing)
931 : {
932 0 : GST_WARNING_OBJECT(m_sink, "Skip pulling buffer - flush is ongoing on server side...");
933 0 : return GstRefSample{};
934 : }
935 1 : if (!m_samples.empty())
936 : {
937 1 : GstSample *sample = m_samples.front();
938 1 : GstBuffer *buffer = gst_sample_get_buffer(sample);
939 1 : GST_LOG_OBJECT(m_sink, "Pulling buffer %p with PTS %" GST_TIME_FORMAT, buffer,
940 : GST_TIME_ARGS(GST_BUFFER_PTS(buffer)));
941 :
942 1 : return GstRefSample{sample};
943 : }
944 :
945 0 : return GstRefSample{};
946 1 : }
947 :
948 1 : void PullModePlaybackDelegate::popSample()
949 : {
950 1 : std::lock_guard<std::mutex> lock(m_sinkMutex);
951 1 : if (!m_samples.empty())
952 : {
953 1 : gst_sample_unref(m_samples.front());
954 1 : m_samples.pop();
955 : }
956 1 : m_needDataCondVariable.notify_all();
957 : }
958 :
959 0 : bool PullModePlaybackDelegate::isEos() const
960 : {
961 0 : std::lock_guard<std::mutex> lock(m_sinkMutex);
962 0 : return m_samples.empty() && m_isEos;
963 : }
964 :
965 2 : bool PullModePlaybackDelegate::isReadyToSendData() const
966 : {
967 2 : std::lock_guard<std::mutex> lock(m_sinkMutex);
968 4 : return m_isEos || m_segmentSet;
969 2 : }
970 :
971 8 : void PullModePlaybackDelegate::lostState()
972 : {
973 8 : m_isStateCommitNeeded = true;
974 8 : gst_element_lost_state(m_sink);
975 : }
976 :
977 164 : bool PullModePlaybackDelegate::attachToMediaClientAndSetStreamsNumber(const uint32_t maxVideoWidth,
978 : const uint32_t maxVideoHeight)
979 : {
980 164 : GstObject *parentObject = getOldestGstBinParent(m_sink);
981 164 : if (!m_mediaPlayerManager.attachMediaPlayerClient(parentObject, maxVideoWidth, maxVideoHeight, isLiveLatencyEnabled()))
982 : {
983 3 : GST_ERROR_OBJECT(m_sink, "Cannot attach the MediaPlayerClient");
984 3 : return false;
985 : }
986 :
987 161 : gchar *parentObjectName = gst_object_get_name(parentObject);
988 161 : GST_INFO_OBJECT(m_sink, "Attached media player client with parent %s(%p)", parentObjectName, parentObject);
989 161 : g_free(parentObjectName);
990 :
991 161 : return setStreamsNumber(parentObject);
992 : }
993 :
994 161 : bool PullModePlaybackDelegate::setStreamsNumber(GstObject *parentObject)
995 : {
996 161 : int32_t videoStreams{-1}, audioStreams{-1}, subtitleStreams{-1};
997 :
998 161 : GstContext *context = gst_element_get_context(m_sink, "streams-info");
999 161 : if (context)
1000 : {
1001 4 : GST_DEBUG_OBJECT(m_sink, "Getting number of streams from \"streams-info\" context");
1002 :
1003 4 : guint n_video{0}, n_audio{0}, n_text{0};
1004 :
1005 4 : const GstStructure *streamsInfoStructure = gst_context_get_structure(context);
1006 4 : gst_structure_get_uint(streamsInfoStructure, "video-streams", &n_video);
1007 4 : gst_structure_get_uint(streamsInfoStructure, "audio-streams", &n_audio);
1008 4 : gst_structure_get_uint(streamsInfoStructure, "text-streams", &n_text);
1009 :
1010 7 : if (n_video > std::numeric_limits<int32_t>::max() || n_audio > std::numeric_limits<int32_t>::max() ||
1011 3 : n_text > std::numeric_limits<int32_t>::max())
1012 : {
1013 1 : GST_ERROR_OBJECT(m_sink, "Number of streams is too big, video=%u, audio=%u, text=%u", n_video, n_audio,
1014 : n_text);
1015 1 : gst_context_unref(context);
1016 1 : return false;
1017 : }
1018 :
1019 3 : videoStreams = n_video;
1020 3 : audioStreams = n_audio;
1021 3 : subtitleStreams = n_text;
1022 :
1023 3 : gst_context_unref(context);
1024 : }
1025 157 : else if (getNStreamsFromParent(parentObject, videoStreams, audioStreams, subtitleStreams))
1026 : {
1027 2 : GST_DEBUG_OBJECT(m_sink, "Got number of streams from playbin2 properties");
1028 : }
1029 : else
1030 : {
1031 : // The default value of streams is V:1, A:1, S:0
1032 : // Changing the default setting via properties is considered as DEPRECATED
1033 155 : subtitleStreams = 0;
1034 155 : std::lock_guard<std::mutex> lock{m_sinkMutex};
1035 155 : if (m_mediaSourceType == firebolt::rialto::MediaSourceType::VIDEO)
1036 : {
1037 33 : videoStreams = m_numOfStreams;
1038 33 : if (m_isSinglePathStream)
1039 : {
1040 32 : audioStreams = 0;
1041 32 : subtitleStreams = 0;
1042 : }
1043 : }
1044 122 : else if (m_mediaSourceType == firebolt::rialto::MediaSourceType::AUDIO)
1045 : {
1046 112 : audioStreams = m_numOfStreams;
1047 112 : if (m_isSinglePathStream)
1048 : {
1049 111 : videoStreams = 0;
1050 111 : subtitleStreams = 0;
1051 : }
1052 : }
1053 10 : else if (m_mediaSourceType == firebolt::rialto::MediaSourceType::SUBTITLE)
1054 : {
1055 10 : subtitleStreams = m_numOfStreams;
1056 10 : if (m_isSinglePathStream)
1057 : {
1058 10 : videoStreams = 0;
1059 10 : audioStreams = 0;
1060 : }
1061 : }
1062 155 : }
1063 :
1064 160 : std::shared_ptr<GStreamerMSEMediaPlayerClient> client = m_mediaPlayerManager.getMediaPlayerClient();
1065 160 : if (!client)
1066 : {
1067 0 : GST_ERROR_OBJECT(m_sink, "MediaPlayerClient is nullptr");
1068 0 : return false;
1069 : }
1070 :
1071 160 : client->handleStreamCollection(audioStreams, videoStreams, subtitleStreams);
1072 :
1073 160 : return true;
1074 : }
1075 :
1076 164 : bool PullModePlaybackDelegate::isLiveLatencyEnabled() const
1077 : {
1078 164 : GstContext *context = gst_element_get_context(m_sink, "streams-info");
1079 164 : if (context)
1080 : {
1081 4 : GST_DEBUG_OBJECT(m_sink, "Checking if live latency is enabled from \"streams-info\" context");
1082 :
1083 4 : gboolean isEnabled{FALSE};
1084 :
1085 4 : const GstStructure *streamsInfoStructure = gst_context_get_structure(context);
1086 4 : gst_structure_get_boolean(streamsInfoStructure, "enable-live-latency", &isEnabled);
1087 :
1088 4 : gst_context_unref(context);
1089 4 : return isEnabled != FALSE;
1090 : }
1091 160 : return false;
1092 : }
1093 :
1094 4 : GstSample *PullModePlaybackDelegate::getLastSample() const
1095 : {
1096 4 : std::lock_guard<std::mutex> lock(m_sinkMutex);
1097 4 : if (m_enableLastSample && m_lastBuffer)
1098 : {
1099 2 : return gst_sample_new(m_lastBuffer, m_caps, &m_lastSegment, nullptr);
1100 : }
1101 2 : return nullptr;
1102 4 : }
1103 :
1104 515 : void PullModePlaybackDelegate::setLastBuffer(GstBuffer *buffer)
1105 : {
1106 515 : if (m_enableLastSample)
1107 : {
1108 3 : if (m_lastBuffer)
1109 : {
1110 2 : gst_buffer_unref(m_lastBuffer);
1111 : }
1112 3 : if (buffer)
1113 : {
1114 2 : m_lastBuffer = gst_buffer_ref(buffer);
1115 : }
1116 : else
1117 : {
1118 1 : m_lastBuffer = nullptr;
1119 : }
1120 : }
1121 515 : }
|
