1 /*
2 * Copyright (C) 2018 Centricular Ltd.
3 * Author: Sebastian Dröge <sebastian@centricular.com>
4 * Author: Nirbheek Chauhan <nirbheek@centricular.com>
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Library General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Library General Public License for more details.
15 *
16 * You should have received a copy of the GNU Library General Public
17 * License along with this library; if not, write to the
18 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
19 * Boston, MA 02110-1301, USA.
20 */
21
22 /**
23 * SECTION:element-proxysrc
24 * @title: proxysrc
25 *
26 * Proxysrc is a source element that proxies events, queries, and buffers from
27 * another pipeline that contains a matching proxysink element. The purpose is
28 * to allow two decoupled pipelines to function as though they are one without
29 * having to manually shuttle buffers, events, queries, etc between the two.
30 *
31 * The element queues buffers from the matching proxysink to an internal queue,
32 * so everything downstream is properly decoupled from the upstream pipeline.
33 * However, the queue may get filled up if the downstream pipeline does not
34 * accept buffers quickly enough; perhaps because it is not yet PLAYING.
35 *
36 * ## Usage
37 *
38 * |[<!-- language="C" -->
39 * GstElement *pipe1, *pipe2, *psink, *psrc;
40 * GstClock *clock;
41 *
42 * pipe1 = gst_parse_launch ("audiotestsrc ! proxysink name=psink", NULL);
43 * psink = gst_bin_get_by_name (GST_BIN (pipe1), "psink");
44 *
45 * pipe2 = gst_parse_launch ("proxysrc name=psrc ! autoaudiosink", NULL);
46 * psrc = gst_bin_get_by_name (GST_BIN (pipe2), "psrc");
47 *
48 * // Connect the two pipelines
49 * g_object_set (psrc, "proxysink", psink, NULL);
50 *
51 * // Both pipelines must agree on the timing information or we'll get glitches
52 * // or overruns/underruns. Ideally, we should tell pipe1 to use the same clock
53 * // as pipe2, but since that will be set asynchronously to the audio clock, it
54 * // is simpler and likely accurate enough to use the system clock for both
55 * // pipelines. If no element in either pipeline will provide a clock, this
56 * // is not needed.
57 * clock = gst_system_clock_obtain ();
58 * gst_pipeline_use_clock (GST_PIPELINE (pipe1), clock);
59 * gst_pipeline_use_clock (GST_PIPELINE (pipe2), clock);
60 * g_object_unref (clock);
61 *
62 * // This is not really needed in this case since the pipelines are created and
63 * // started at the same time. However, an application that dynamically
64 * // generates pipelines must ensure that all the pipelines that will be
65 * // connected together share the same base time.
66 * gst_element_set_base_time (pipe1, 0);
67 * gst_element_set_base_time (pipe2, 0);
68 *
69 * gst_element_set_state (pipe1, GST_STATE_PLAYING);
70 * gst_element_set_state (pipe2, GST_STATE_PLAYING);
71 * ]|
72 *
73 */
74
75 #ifdef HAVE_CONFIG_H
76 #include "config.h"
77 #endif
78 #include "gstproxysrc.h"
79 #include "gstproxysink.h"
80 #include "gstproxy-priv.h"
81
82 #define GST_CAT_DEFAULT gst_proxy_src_debug
83 GST_DEBUG_CATEGORY_STATIC (GST_CAT_DEFAULT);
84
85 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
86 GST_PAD_SRC,
87 GST_PAD_ALWAYS,
88 GST_STATIC_CAPS_ANY);
89
90 enum
91 {
92 PROP_0,
93 PROP_PROXYSINK,
94 };
95
96 /* We're not subclassing from basesrc because we don't want any of the special
97 * handling it has for events/queries/etc. We just pass-through everything. */
98
99 /* Our parent type is a GstBin instead of GstElement because we contain a queue
100 * element */
101 #define parent_class gst_proxy_src_parent_class
102 G_DEFINE_TYPE (GstProxySrc, gst_proxy_src, GST_TYPE_BIN);
103
104 static gboolean gst_proxy_src_internal_src_query (GstPad * pad,
105 GstObject * parent, GstQuery * query);
106 static gboolean gst_proxy_src_internal_src_event (GstPad * pad,
107 GstObject * parent, GstEvent * event);
108
109 static GstStateChangeReturn gst_proxy_src_change_state (GstElement * element,
110 GstStateChange transition);
111 static gboolean gst_proxy_src_send_event (GstElement * element,
112 GstEvent * event);
113 static gboolean gst_proxy_src_query (GstElement * element, GstQuery * query);
114 static void gst_proxy_src_dispose (GObject * object);
115
116 static void
gst_proxy_src_get_property(GObject * object,guint prop_id,GValue * value,GParamSpec * spec)117 gst_proxy_src_get_property (GObject * object, guint prop_id, GValue * value,
118 GParamSpec * spec)
119 {
120 GstProxySrc *self = GST_PROXY_SRC (object);
121
122 switch (prop_id) {
123 case PROP_PROXYSINK:
124 g_value_take_object (value, g_weak_ref_get (&self->proxysink));
125 break;
126 default:
127 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, spec);
128 break;
129 }
130 }
131
132 static void
gst_proxy_src_set_property(GObject * object,guint prop_id,const GValue * value,GParamSpec * spec)133 gst_proxy_src_set_property (GObject * object, guint prop_id,
134 const GValue * value, GParamSpec * spec)
135 {
136 GstProxySrc *self = GST_PROXY_SRC (object);
137 GstProxySink *sink;
138
139 switch (prop_id) {
140 case PROP_PROXYSINK:
141 sink = g_value_dup_object (value);
142 if (sink == NULL) {
143 /* Unset proxysrc property on the existing proxysink to break the
144 * connection in that direction */
145 GstProxySink *old_sink = g_weak_ref_get (&self->proxysink);
146 if (old_sink) {
147 gst_proxy_sink_set_proxysrc (old_sink, NULL);
148 g_object_unref (old_sink);
149 }
150 g_weak_ref_set (&self->proxysink, NULL);
151 } else {
152 /* Set proxysrc property on the new proxysink to point to us */
153 gst_proxy_sink_set_proxysrc (sink, self);
154 g_weak_ref_set (&self->proxysink, sink);
155 g_object_unref (sink);
156 }
157 break;
158 default:
159 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, spec);
160 }
161 }
162
163 static void
gst_proxy_src_class_init(GstProxySrcClass * klass)164 gst_proxy_src_class_init (GstProxySrcClass * klass)
165 {
166 GObjectClass *gobject_class = (GObjectClass *) klass;
167 GstElementClass *gstelement_class = (GstElementClass *) klass;
168
169 GST_DEBUG_CATEGORY_INIT (gst_proxy_src_debug, "proxysrc", 0, "proxy sink");
170
171 gobject_class->dispose = gst_proxy_src_dispose;
172
173 gobject_class->get_property = gst_proxy_src_get_property;
174 gobject_class->set_property = gst_proxy_src_set_property;
175
176 g_object_class_install_property (gobject_class, PROP_PROXYSINK,
177 g_param_spec_object ("proxysink", "Proxysink", "Matching proxysink",
178 GST_TYPE_PROXY_SINK, G_PARAM_READWRITE));
179
180 gstelement_class->change_state = gst_proxy_src_change_state;
181 gstelement_class->send_event = gst_proxy_src_send_event;
182 gstelement_class->query = gst_proxy_src_query;
183 gst_element_class_add_pad_template (gstelement_class,
184 gst_static_pad_template_get (&src_template));
185
186 gst_element_class_set_static_metadata (gstelement_class, "Proxy source",
187 "Source", "Proxy source for internal process communication",
188 "Sebastian Dröge <sebastian@centricular.com>");
189 }
190
191 static void
gst_proxy_src_init(GstProxySrc * self)192 gst_proxy_src_init (GstProxySrc * self)
193 {
194 GstPad *srcpad, *sinkpad;
195 GstPadTemplate *templ;
196
197 GST_OBJECT_FLAG_SET (self, GST_ELEMENT_FLAG_SOURCE);
198
199 /* We feed incoming buffers into a queue to decouple the downstream pipeline
200 * from the upstream pipeline */
201 self->queue = gst_element_factory_make ("queue", NULL);
202 gst_bin_add (GST_BIN (self), self->queue);
203
204 srcpad = gst_element_get_static_pad (self->queue, "src");
205 templ = gst_static_pad_template_get (&src_template);
206 self->srcpad = gst_ghost_pad_new_from_template ("src", srcpad, templ);
207 gst_object_unref (templ);
208 gst_object_unref (srcpad);
209
210 gst_element_add_pad (GST_ELEMENT (self), self->srcpad);
211
212 /* A dummy sinkpad that's not actually used anywhere
213 * Explanation for why this is needed is below */
214 self->dummy_sinkpad = gst_pad_new ("dummy_sinkpad", GST_PAD_SINK);
215 gst_object_set_parent (GST_OBJECT (self->dummy_sinkpad), GST_OBJECT (self));
216
217 self->internal_srcpad = gst_pad_new ("internal_src", GST_PAD_SRC);
218 gst_object_set_parent (GST_OBJECT (self->internal_srcpad),
219 GST_OBJECT (self->dummy_sinkpad));
220 gst_pad_set_event_function (self->internal_srcpad,
221 gst_proxy_src_internal_src_event);
222 gst_pad_set_query_function (self->internal_srcpad,
223 gst_proxy_src_internal_src_query);
224
225 /* We need to link internal_srcpad from proxysink to the sinkpad of our
226 * queue. However, two pads can only be linked if they share a common parent.
227 * Above, we set the parent of the dummy_sinkpad as proxysrc, and then we set
228 * the parent of internal_srcpad as dummy_sinkpad. This causes both these pads
229 * to share a parent allowing us to link them.
230 * Yes, this is a hack/workaround. */
231 sinkpad = gst_element_get_static_pad (self->queue, "sink");
232 gst_pad_link (self->internal_srcpad, sinkpad);
233 gst_object_unref (sinkpad);
234
235 gst_bin_set_suppressed_flags (GST_BIN (self),
236 GST_ELEMENT_FLAG_SOURCE | GST_ELEMENT_FLAG_SINK);
237 GST_OBJECT_FLAG_SET (self, GST_ELEMENT_FLAG_SOURCE);
238 }
239
240 static void
gst_proxy_src_dispose(GObject * object)241 gst_proxy_src_dispose (GObject * object)
242 {
243 GstProxySrc *self = GST_PROXY_SRC (object);
244
245 gst_object_unparent (GST_OBJECT (self->dummy_sinkpad));
246 self->dummy_sinkpad = NULL;
247
248 gst_object_unparent (GST_OBJECT (self->internal_srcpad));
249 self->internal_srcpad = NULL;
250
251 g_weak_ref_set (&self->proxysink, NULL);
252
253 G_OBJECT_CLASS (gst_proxy_src_parent_class)->dispose (object);
254 }
255
256 static GstStateChangeReturn
gst_proxy_src_change_state(GstElement * element,GstStateChange transition)257 gst_proxy_src_change_state (GstElement * element, GstStateChange transition)
258 {
259 GstElementClass *gstelement_class =
260 GST_ELEMENT_CLASS (gst_proxy_src_parent_class);
261 GstProxySrc *self = GST_PROXY_SRC (element);
262 GstStateChangeReturn ret;
263
264 ret = gstelement_class->change_state (element, transition);
265 if (ret == GST_STATE_CHANGE_FAILURE)
266 return ret;
267
268 switch (transition) {
269 case GST_STATE_CHANGE_READY_TO_PAUSED:
270 ret = GST_STATE_CHANGE_NO_PREROLL;
271 gst_pad_set_active (self->internal_srcpad, TRUE);
272 break;
273 case GST_STATE_CHANGE_PAUSED_TO_READY:
274 gst_pad_set_active (self->internal_srcpad, FALSE);
275 break;
276 default:
277 break;
278 }
279
280 return ret;
281 }
282
283 static gboolean
gst_proxy_src_send_event(GstElement * element,GstEvent * event)284 gst_proxy_src_send_event (GstElement * element, GstEvent * event)
285 {
286 GstProxySrc *self = GST_PROXY_SRC (element);
287
288 if (GST_EVENT_IS_DOWNSTREAM (event)) {
289 GstPad *sinkpad = gst_element_get_static_pad (self->queue, "sink");
290 gboolean ret;
291
292 ret = gst_pad_send_event (sinkpad, event);
293 gst_object_unref (sinkpad);
294 return ret;
295 } else {
296 gst_event_unref (event);
297 return FALSE;
298 }
299 }
300
301 static gboolean
gst_proxy_src_query(GstElement * element,GstQuery * query)302 gst_proxy_src_query (GstElement * element, GstQuery * query)
303 {
304 GstProxySrc *self = GST_PROXY_SRC (element);
305
306 if (GST_QUERY_IS_DOWNSTREAM (query)) {
307 GstPad *sinkpad = gst_element_get_static_pad (self->queue, "sink");
308 gboolean ret;
309
310 ret = gst_pad_query (sinkpad, query);
311 gst_object_unref (sinkpad);
312 return ret;
313 } else {
314 return FALSE;
315 }
316 }
317
318 static gboolean
gst_proxy_src_internal_src_query(GstPad * pad,GstObject * parent,GstQuery * query)319 gst_proxy_src_internal_src_query (GstPad * pad, GstObject * parent,
320 GstQuery * query)
321 {
322 GstProxySrc *self = GST_PROXY_SRC (gst_object_get_parent (parent));
323 GstProxySink *sink;
324 gboolean ret = FALSE;
325
326 if (!self)
327 return ret;
328
329 GST_LOG_OBJECT (pad, "Handling query of type '%s'",
330 gst_query_type_get_name (GST_QUERY_TYPE (query)));
331
332 sink = g_weak_ref_get (&self->proxysink);
333 if (sink) {
334 GstPad *sinkpad;
335 sinkpad = gst_proxy_sink_get_internal_sinkpad (sink);
336
337 ret = gst_pad_peer_query (sinkpad, query);
338 gst_object_unref (sinkpad);
339 gst_object_unref (sink);
340 }
341
342 gst_object_unref (self);
343
344 return ret;
345 }
346
347 static gboolean
gst_proxy_src_internal_src_event(GstPad * pad,GstObject * parent,GstEvent * event)348 gst_proxy_src_internal_src_event (GstPad * pad, GstObject * parent,
349 GstEvent * event)
350 {
351 GstProxySrc *self = GST_PROXY_SRC (gst_object_get_parent (parent));
352 GstProxySink *sink;
353 gboolean ret = FALSE;
354
355 if (!self)
356 return ret;
357
358 GST_LOG_OBJECT (pad, "Got %s event", GST_EVENT_TYPE_NAME (event));
359
360 sink = g_weak_ref_get (&self->proxysink);
361 if (sink) {
362 GstPad *sinkpad;
363 sinkpad = gst_proxy_sink_get_internal_sinkpad (sink);
364
365 ret = gst_pad_push_event (sinkpad, event);
366 gst_object_unref (sinkpad);
367 gst_object_unref (sink);
368 } else
369 gst_event_unref (event);
370
371
372 gst_object_unref (self);
373
374 return ret;
375 }
376
377 /* Wrapper function for accessing private member */
378 GstPad *
gst_proxy_src_get_internal_srcpad(GstProxySrc * self)379 gst_proxy_src_get_internal_srcpad (GstProxySrc * self)
380 {
381 return gst_object_ref (self->internal_srcpad);
382 }
383