1// Copyright (C) 2016 The Qt Company Ltd.
2// SPDX-License-Identifier: LicenseRef-Qt-Commercial OR LGPL-3.0-only OR GPL-2.0-only OR GPL-3.0-only
3
4#include <QtCore/qcoreapplication.h>
5#include <QtCore/qloggingcategory.h>
6
7#include "qgstpipeline_p.h"
8#include "qgst_bus_observer_p.h"
9
10#include <thread>
11
12QT_BEGIN_NAMESPACE
13
14static Q_LOGGING_CATEGORY(qLcGstPipeline, "qt.multimedia.gstpipeline");
15
16struct QGstPipelinePrivate
17{
18 explicit QGstPipelinePrivate(QGstBusHandle);
19 ~QGstPipelinePrivate();
20
21 std::chrono::nanoseconds m_position{};
22 double m_rate = 1.;
23 std::unique_ptr<QGstBusObserver> m_busObserver;
24};
25
26QGstPipelinePrivate::QGstPipelinePrivate(QGstBusHandle bus)
27 : m_busObserver{
28 std::make_unique<QGstBusObserver>(args: std::move(bus)),
29 }
30{
31 Q_ASSERT(QThread::isMainThread());
32}
33
34QGstPipelinePrivate::~QGstPipelinePrivate()
35{
36 m_busObserver->close();
37
38 if (m_busObserver->currentThreadIsNotifierThread())
39 return;
40
41 // The QGstPipelinePrivate is owned the the GstPipeline and can be destroyed from a gstreamer
42 // thread. In this case we cannot destroy the object immediately, but need to marshall it
43 // through the event loop of the main thread
44 QMetaObject::invokeMethod(qApp, function: [bus = std::move(m_busObserver)] {
45 // nothing to do, we just extend the lifetime of the bus
46 });
47}
48
49// QGstPipeline
50
51QGstPipeline QGstPipeline::create(const char *name)
52{
53 GstPipeline *pipeline = qGstCheckedCast<GstPipeline>(arg: gst_pipeline_new(name));
54 return adopt(pipeline);
55}
56
57QGstPipeline QGstPipeline::createFromFactory(const char *factory, const char *name)
58{
59 QGstElement playbin3 = QGstElement::createFromFactory(factory, name);
60 GstPipeline *pipeline = qGstCheckedCast<GstPipeline>(arg: playbin3.element());
61
62 return QGstPipeline::adopt(pipeline);
63}
64
65QGstPipeline QGstPipeline::adopt(GstPipeline *pipeline)
66{
67 QGstPipeline wrappedObject{
68 pipeline,
69 QGstPipeline::NeedsRef,
70 };
71
72 QGstBusHandle bus{
73 gst_pipeline_get_bus(pipeline),
74 QGstBusHandle::HasRef,
75 };
76
77 auto d = std::make_unique<QGstPipelinePrivate>(args: std::move(bus));
78 wrappedObject.set(property: "pipeline-private", object: std::move(d));
79
80 return wrappedObject;
81}
82
83QGstPipeline::QGstPipeline(GstPipeline *p, RefMode mode) : QGstBin(qGstCheckedCast<GstBin>(arg: p), mode)
84{
85}
86
87QGstPipeline::~QGstPipeline() = default;
88
89void QGstPipeline::installMessageFilter(QGstreamerBusMessageFilter *filter)
90{
91 QGstPipelinePrivate *d = getPrivate();
92 d->m_busObserver->installMessageFilter(filter);
93}
94
95void QGstPipeline::removeMessageFilter(QGstreamerBusMessageFilter *filter)
96{
97 QGstPipelinePrivate *d = getPrivate();
98 d->m_busObserver->removeMessageFilter(filter);
99}
100
101GstStateChangeReturn QGstPipeline::setState(GstState state)
102{
103 return gst_element_set_state(element: element(), state);
104}
105
106bool QGstPipeline::processNextPendingMessage(GstMessageType types, std::chrono::nanoseconds timeout)
107{
108 QGstPipelinePrivate *d = getPrivate();
109 return d->m_busObserver->processNextPendingMessage(type: types, timeout);
110}
111
112bool QGstPipeline::processNextPendingMessage(std::chrono::nanoseconds timeout)
113{
114 return processNextPendingMessage(types: GST_MESSAGE_ANY, timeout);
115}
116
117void QGstPipeline::flush()
118{
119 seek(pos: position(), /*flush=*/true);
120}
121
122void QGstPipeline::seek(std::chrono::nanoseconds pos, double rate, bool flush)
123{
124 using namespace std::chrono_literals;
125
126 QGstPipelinePrivate *d = getPrivate();
127 // always adjust the rate, so it can be set before playback starts
128 // setting position needs a loaded media file that's seekable
129
130 qCDebug(qLcGstPipeline) << "QGstPipeline::seek to" << pos << "rate:" << rate
131 << (flush ? "flushing" : "not flushing");
132
133 GstSeekFlags seekFlags = flush ? GST_SEEK_FLAG_FLUSH : GST_SEEK_FLAG_NONE;
134 seekFlags = GstSeekFlags(seekFlags | GST_SEEK_FLAG_SEGMENT | GST_SEEK_FLAG_ACCURATE);
135
136 bool success = (rate > 0)
137 ? gst_element_seek(element: element(), rate, format: GST_FORMAT_TIME, flags: seekFlags, start_type: GST_SEEK_TYPE_SET,
138 start: pos.count(), stop_type: GST_SEEK_TYPE_END, stop: 0)
139 : gst_element_seek(element: element(), rate, format: GST_FORMAT_TIME, flags: seekFlags, start_type: GST_SEEK_TYPE_SET, start: 0,
140 stop_type: GST_SEEK_TYPE_SET, stop: pos.count());
141
142 if (!success) {
143 qDebug() << "seek: gst_element_seek failed" << pos;
144 dumpGraph(fileNamePrefix: "seekSeekFailed");
145 return;
146 }
147
148 d->m_position = pos;
149}
150
151void QGstPipeline::seek(std::chrono::nanoseconds pos, bool flush)
152{
153 qCDebug(qLcGstPipeline) << "QGstPipeline::seek to" << pos;
154 seek(pos, rate: getPrivate()->m_rate, flush);
155}
156
157void QGstPipeline::setPlaybackRate(double rate, bool forceFlushingSeek)
158{
159 QGstPipelinePrivate *d = getPrivate();
160 if (rate == d->m_rate)
161 return;
162
163 d->m_rate = rate;
164
165 qCDebug(qLcGstPipeline) << "QGstPipeline::setPlaybackRate to" << rate;
166
167 applyPlaybackRate(forceFlushingSeek);
168}
169
170double QGstPipeline::playbackRate() const
171{
172 QGstPipelinePrivate *d = getPrivate();
173 return d->m_rate;
174}
175
176void QGstPipeline::applyPlaybackRate(bool forceFlushingSeek)
177{
178 QGstPipelinePrivate *d = getPrivate();
179
180 // do not GST_SEEK_FLAG_FLUSH with GST_SEEK_TYPE_NONE
181 // https://gitlab.freedesktop.org/gstreamer/gstreamer/-/issues/3604
182 if (!forceFlushingSeek) {
183 bool asyncChangeSuccess = waitForAsyncStateChangeComplete();
184 if (!asyncChangeSuccess) {
185 qWarning()
186 << "QGstPipeline::seek: async pipeline change in progress. Seeking impossible";
187 return;
188 }
189
190 qCDebug(qLcGstPipeline) << "QGstPipeline::applyPlaybackRate instantly";
191 bool success = gst_element_seek(
192 element: element(), rate: d->m_rate, format: GST_FORMAT_UNDEFINED, flags: GST_SEEK_FLAG_INSTANT_RATE_CHANGE,
193 start_type: GST_SEEK_TYPE_NONE, GST_CLOCK_TIME_NONE, stop_type: GST_SEEK_TYPE_NONE, GST_CLOCK_TIME_NONE);
194 if (!success) {
195 qDebug() << "setPlaybackRate: gst_element_seek failed";
196 dumpGraph(fileNamePrefix: "applyPlaybackRateSeekFailed");
197 }
198 } else {
199 seek(pos: position(), flush: d->m_rate);
200 }
201}
202
203void QGstPipeline::setPosition(std::chrono::nanoseconds pos, bool flush)
204{
205 seek(pos, flush);
206}
207
208std::chrono::nanoseconds QGstPipeline::position() const
209{
210 QGstPipelinePrivate *d = getPrivate();
211 std::optional<std::chrono::nanoseconds> pos = QGstElement::position();
212 if (pos) {
213 d->m_position = *pos;
214 qCDebug(qLcGstPipeline) << "QGstPipeline::position:"
215 << std::chrono::round<std::chrono::milliseconds>(d: *pos);
216 } else {
217 qDebug() << "QGstPipeline: failed to query position, using previous position";
218 dumpGraph(fileNamePrefix: "positionQueryFailed");
219 }
220
221 return d->m_position;
222}
223
224std::chrono::milliseconds QGstPipeline::positionInMs() const
225{
226 using namespace std::chrono;
227 return round<milliseconds>(d: position());
228}
229
230void QGstPipeline::setPositionAndRate(std::chrono::nanoseconds pos, double rate)
231{
232 QGstPipelinePrivate *d = getPrivate();
233 d->m_rate = rate;
234 seek(pos, flush: rate);
235}
236
237std::optional<std::chrono::nanoseconds>
238QGstPipeline::queryPosition(std::chrono::nanoseconds timeout) const
239{
240 using namespace std::chrono_literals;
241 using namespace std::chrono;
242
243 std::chrono::nanoseconds totalSleepTime{};
244
245 for (;;) {
246 std::optional<nanoseconds> dur = QGstElement::duration();
247 if (dur)
248 return dur;
249
250 if (totalSleepTime >= timeout)
251 return std::nullopt;
252 std::this_thread::sleep_for(rtime: 20ms);
253 totalSleepTime += 20ms;
254 }
255}
256
257std::optional<std::chrono::nanoseconds>
258QGstPipeline::queryDuration(std::chrono::nanoseconds timeout) const
259{
260 using namespace std::chrono_literals;
261 using namespace std::chrono;
262
263 std::chrono::nanoseconds totalSleepTime{};
264
265 for (;;) {
266 std::optional<nanoseconds> dur = QGstElement::duration();
267 if (dur)
268 return dur;
269
270 if (totalSleepTime >= timeout)
271 return std::nullopt;
272
273 std::this_thread::sleep_for(rtime: 20ms);
274 totalSleepTime += 20ms;
275 }
276}
277
278std::optional<std::pair<std::chrono::nanoseconds, std::chrono::nanoseconds>>
279QGstPipeline::queryPositionAndDuration(std::chrono::nanoseconds timeout) const
280{
281 using namespace std::chrono_literals;
282 using namespace std::chrono;
283
284 std::chrono::nanoseconds totalSleepTime{};
285
286 std::optional<nanoseconds> dur;
287 std::optional<nanoseconds> pos;
288
289 for (;;) {
290 if (!dur)
291 dur = QGstElement::duration();
292 if (!pos)
293 pos = QGstElement::position();
294
295 if (dur && pos)
296 return std::pair{ *dur, *pos };
297
298 if (totalSleepTime >= timeout)
299 return std::nullopt;
300
301 std::this_thread::sleep_for(rtime: 20ms);
302 totalSleepTime += 20ms;
303 }
304}
305
306void QGstPipeline::seekToEndWithEOS()
307{
308 QGstPipelinePrivate *d = getPrivate();
309
310 gst_element_seek(element: element(), rate: d->m_rate, format: GST_FORMAT_TIME, flags: GST_SEEK_FLAG_NONE, start_type: GST_SEEK_TYPE_END,
311 start: 0, stop_type: GST_SEEK_TYPE_END, stop: 0);
312}
313
314QGstPipelinePrivate *QGstPipeline::getPrivate() const
315{
316 QGstPipelinePrivate *ret = getObject<QGstPipelinePrivate>(property: "pipeline-private");
317 Q_ASSERT(ret);
318 return ret;
319}
320
321QT_END_NAMESPACE
322

source code of qtmultimedia/src/plugins/multimedia/gstreamer/common/qgstpipeline.cpp