*
* gstrtpmux.c:
*
- * Copyright (C) <2007> Nokia Corporation.
+ * Copyright (C) <2007-2010> Nokia Corporation.
* Contact: Zeeshan Ali <zeeshan.ali@nokia.com>
+ * Copyright (C) <2007-2010> Collabora Ltd
+ * Contact: Olivier Crete <olivier.crete@collabora.co.uk>
* Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
* 2000,2005 Wim Taymans <wim@fluendo.com>
*
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
- * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
- * Boston, MA 02111-1307, USA.
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
*/
/**
* SECTION:element-rtpmux
+ * @title: rtpmux
+ * @see_also: rtpdtmfmux
*
* The rtp muxer takes multiple RTP streams having the same clock-rate and
* muxes into a single stream with a single SSRC.
+ *
+ * ## Example pipelines
+ * |[
+ * gst-launch-1.0 rtpmux name=mux ! udpsink host=127.0.0.1 port=8888 \
+ * alsasrc ! alawenc ! rtppcmapay ! \
+ * application/x-rtp, payload=8, rate=8000 ! mux.sink_0 \
+ * audiotestsrc is-live=1 ! \
+ * mulawenc ! rtppcmupay ! \
+ * application/x-rtp, payload=0, rate=8000 ! mux.sink_1
+ * ]|
+ * In this example, an audio stream is captured from ALSA and another is
+ * generated, both are encoded into different payload types and muxed together
+ * so they can be sent on the same port.
+ *
*/
#ifdef HAVE_CONFIG_H
GST_DEBUG_CATEGORY_STATIC (gst_rtp_mux_debug);
#define GST_CAT_DEFAULT gst_rtp_mux_debug
-/* elementfactory information */
-static const GstElementDetails gst_rtp_mux_details =
-GST_ELEMENT_DETAILS ("RTP muxer",
- "Codec/Muxer",
- "multiplex N rtp streams into one",
- "Zeeshan Ali <first.last@nokia.com>");
-
enum
{
- ARG_0,
+ PROP_0,
PROP_TIMESTAMP_OFFSET,
PROP_SEQNUM_OFFSET,
PROP_SEQNUM,
#define DEFAULT_SEQNUM_OFFSET -1
#define DEFAULT_SSRC -1
-typedef struct
-{
- gboolean have_clock_base;
- guint clock_base;
-
- GstCaps *out_caps;
-} GstRTPMuxPadPrivate;
-
static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
GST_PAD_SRC,
GST_PAD_ALWAYS,
GST_STATIC_CAPS ("application/x-rtp")
);
-static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink_%d",
+static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink_%u",
GST_PAD_SINK,
GST_PAD_REQUEST,
GST_STATIC_CAPS ("application/x-rtp")
);
-static void gst_rtp_mux_finalize (GObject * object);
-
static GstPad *gst_rtp_mux_request_new_pad (GstElement * element,
- GstPadTemplate * templ, const gchar * name);
-static GstFlowReturn gst_rtp_mux_chain (GstPad * pad, GstBuffer * buffer);
-static gboolean gst_rtp_mux_setcaps (GstPad * pad, GstCaps * caps);
-static GstCaps *gst_rtp_mux_getcaps (GstPad * pad);
+ GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
+static void gst_rtp_mux_release_pad (GstElement * element, GstPad * pad);
+static GstFlowReturn gst_rtp_mux_chain (GstPad * pad, GstObject * parent,
+ GstBuffer * buffer);
+static GstFlowReturn gst_rtp_mux_chain_list (GstPad * pad, GstObject * parent,
+ GstBufferList * bufferlist);
+static gboolean gst_rtp_mux_setcaps (GstPad * pad, GstRTPMux * rtp_mux,
+ GstCaps * caps);
+static gboolean gst_rtp_mux_sink_event (GstPad * pad, GstObject * parent,
+ GstEvent * event);
+static gboolean gst_rtp_mux_sink_query (GstPad * pad, GstObject * parent,
+ GstQuery * query);
static GstStateChangeReturn gst_rtp_mux_change_state (GstElement *
element, GstStateChange transition);
const GValue * value, GParamSpec * pspec);
static void gst_rtp_mux_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
+static void gst_rtp_mux_dispose (GObject * object);
+static gboolean gst_rtp_mux_src_event_real (GstRTPMux * rtp_mux,
+ GstEvent * event);
-GST_BOILERPLATE (GstRTPMux, gst_rtp_mux, GstElement, GST_TYPE_ELEMENT);
-
-static void
-gst_rtp_mux_base_init (gpointer g_class)
-{
- GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
+G_DEFINE_TYPE (GstRTPMux, gst_rtp_mux, GST_TYPE_ELEMENT);
- gst_element_class_add_pad_template (element_class,
- gst_static_pad_template_get (&src_factory));
- gst_element_class_add_pad_template (element_class,
- gst_static_pad_template_get (&sink_factory));
-
- gst_element_class_set_details (element_class, &gst_rtp_mux_details);
-}
static void
gst_rtp_mux_class_init (GstRTPMuxClass * klass)
gobject_class = (GObjectClass *) klass;
gstelement_class = (GstElementClass *) klass;
- gobject_class->finalize = gst_rtp_mux_finalize;
+
+ gst_element_class_add_static_pad_template (gstelement_class, &src_factory);
+ gst_element_class_add_static_pad_template (gstelement_class, &sink_factory);
+
+ gst_element_class_set_static_metadata (gstelement_class, "RTP muxer",
+ "Codec/Muxer",
+ "multiplex N rtp streams into one", "Zeeshan Ali <first.last@nokia.com>");
+
gobject_class->get_property = gst_rtp_mux_get_property;
gobject_class->set_property = gst_rtp_mux_set_property;
+ gobject_class->dispose = gst_rtp_mux_dispose;
+
+ klass->src_event = gst_rtp_mux_src_event_real;
g_object_class_install_property (G_OBJECT_CLASS (klass),
PROP_TIMESTAMP_OFFSET, g_param_spec_int ("timestamp-offset",
"Timestamp Offset",
"Offset to add to all outgoing timestamps (-1 = random)", -1,
- G_MAXINT, DEFAULT_TIMESTAMP_OFFSET, G_PARAM_READWRITE));
+ G_MAXINT, DEFAULT_TIMESTAMP_OFFSET,
+ G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SEQNUM_OFFSET,
g_param_spec_int ("seqnum-offset", "Sequence number Offset",
"Offset to add to all outgoing seqnum (-1 = random)", -1, G_MAXINT,
- DEFAULT_SEQNUM_OFFSET, G_PARAM_READWRITE));
+ DEFAULT_SEQNUM_OFFSET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SEQNUM,
g_param_spec_uint ("seqnum", "Sequence number",
"The RTP sequence number of the last processed packet",
- 0, G_MAXUINT, 0, G_PARAM_READABLE));
+ 0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SSRC,
g_param_spec_uint ("ssrc", "SSRC",
- "The SSRC of the packets (-1 == random)",
- 0, G_MAXUINT, DEFAULT_SSRC, G_PARAM_READWRITE));
+ "The SSRC of the packets (default == random)",
+ 0, G_MAXUINT, DEFAULT_SSRC,
+ GST_PARAM_MUTABLE_PLAYING | G_PARAM_READWRITE |
+ G_PARAM_STATIC_STRINGS));
gstelement_class->request_new_pad =
GST_DEBUG_FUNCPTR (gst_rtp_mux_request_new_pad);
+ gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_rtp_mux_release_pad);
gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_rtp_mux_change_state);
-
- klass->chain_func = gst_rtp_mux_chain;
}
-static gboolean
-gst_rtp_mux_src_event (GstPad * pad, GstEvent * event)
+static void
+gst_rtp_mux_dispose (GObject * object)
{
- GstElement *rtp_mux;
- GstIterator *iter;
- GstPad *sinkpad;
- gboolean result = FALSE;
- gboolean done = FALSE;
-
- rtp_mux = gst_pad_get_parent_element (pad);
- g_return_val_if_fail (rtp_mux != NULL, FALSE);
-
- iter = gst_element_iterate_sink_pads (rtp_mux);
-
- while (!done) {
- switch (gst_iterator_next (iter, (gpointer) & sinkpad)) {
- case GST_ITERATOR_OK:
- gst_event_ref (event);
- result |= gst_pad_push_event (sinkpad, event);
- gst_object_unref (sinkpad);
- break;
- case GST_ITERATOR_RESYNC:
- gst_iterator_resync (iter);
- result = FALSE;
- break;
- case GST_ITERATOR_ERROR:
- GST_WARNING_OBJECT (rtp_mux, "Error iterating sinkpads");
- case GST_ITERATOR_DONE:
- done = TRUE;
- break;
+ GstRTPMux *rtp_mux = GST_RTP_MUX (object);
+ GList *item;
+
+ g_clear_object (&rtp_mux->last_pad);
+
+restart:
+ for (item = GST_ELEMENT_PADS (object); item; item = g_list_next (item)) {
+ GstPad *pad = GST_PAD (item->data);
+ if (GST_PAD_IS_SINK (pad)) {
+ gst_element_release_request_pad (GST_ELEMENT (object), pad);
+ goto restart;
}
}
- gst_iterator_free (iter);
- gst_object_unref (rtp_mux);
- gst_event_unref (event);
- return result;
+ G_OBJECT_CLASS (gst_rtp_mux_parent_class)->dispose (object);
}
-static void
-gst_rtp_mux_init (GstRTPMux * object, GstRTPMuxClass * g_class)
+static gboolean
+gst_rtp_mux_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
- GstElementClass *klass = GST_ELEMENT_GET_CLASS (object);
-
- object->srcpad =
- gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
- "src"), "src");
- gst_pad_set_event_function (object->srcpad,
- GST_DEBUG_FUNCPTR (gst_rtp_mux_src_event));
- gst_element_add_pad (GST_ELEMENT (object), object->srcpad);
-
- object->ssrc = DEFAULT_SSRC;
- object->ts_offset = DEFAULT_TIMESTAMP_OFFSET;
- object->seqnum_offset = DEFAULT_SEQNUM_OFFSET;
-}
+ GstRTPMux *rtp_mux = GST_RTP_MUX (parent);
+ GstRTPMuxClass *klass;
+ gboolean ret;
-static void
-gst_rtp_mux_finalize (GObject * object)
-{
- GstRTPMux *rtp_mux;
+ klass = GST_RTP_MUX_GET_CLASS (rtp_mux);
- rtp_mux = GST_RTP_MUX (object);
+ ret = klass->src_event (rtp_mux, event);
- G_OBJECT_CLASS (parent_class)->finalize (object);
+ return ret;
}
-static GstPad *
-gst_rtp_mux_create_sinkpad (GstRTPMux * rtp_mux, GstPadTemplate * templ)
+static gboolean
+gst_rtp_mux_src_event_real (GstRTPMux * rtp_mux, GstEvent * event)
{
- GstPad *newpad = NULL;
- GstPadTemplate *class_templ;
-
- class_templ =
- gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS (rtp_mux),
- "sink_%d");
-
- if (templ == class_templ) {
- gchar *name;
-
- /* create new pad with the name */
- name = g_strdup_printf ("sink_%02d", rtp_mux->numpads);
- newpad = gst_pad_new_from_template (templ, name);
- g_free (name);
-
- rtp_mux->numpads++;
- } else {
- GST_WARNING_OBJECT (rtp_mux, "this is not our template!\n");
+ switch (GST_EVENT_TYPE (event)) {
+ case GST_EVENT_CUSTOM_UPSTREAM:
+ {
+ const GstStructure *s = gst_event_get_structure (event);
+
+ if (gst_structure_has_name (s, "GstRTPCollision")) {
+ guint ssrc = 0;
+
+ if (!gst_structure_get_uint (s, "ssrc", &ssrc))
+ ssrc = -1;
+
+ GST_DEBUG_OBJECT (rtp_mux, "collided ssrc: %x", ssrc);
+
+ /* choose another ssrc for our stream */
+ GST_OBJECT_LOCK (rtp_mux);
+ if (ssrc == rtp_mux->current_ssrc) {
+ GstCaps *caps;
+ guint suggested_ssrc = 0;
+ guint32 new_ssrc;
+
+ if (gst_structure_get_uint (s, "suggested-ssrc", &suggested_ssrc))
+ rtp_mux->current_ssrc = suggested_ssrc;
+
+ while (ssrc == rtp_mux->current_ssrc)
+ rtp_mux->current_ssrc = g_random_int ();
+
+ new_ssrc = rtp_mux->current_ssrc;
+ GST_INFO_OBJECT (rtp_mux, "New ssrc after collision %x (was: %x)",
+ new_ssrc, ssrc);
+ GST_OBJECT_UNLOCK (rtp_mux);
+
+ caps = gst_pad_get_current_caps (rtp_mux->srcpad);
+ caps = gst_caps_make_writable (caps);
+ gst_caps_set_simple (caps, "ssrc", G_TYPE_UINT, new_ssrc, NULL);
+ gst_pad_set_caps (rtp_mux->srcpad, caps);
+ gst_caps_unref (caps);
+ } else {
+ GST_OBJECT_UNLOCK (rtp_mux);
+ }
+ }
+ break;
+ }
+ default:
+ break;
}
- return newpad;
+
+ return gst_pad_event_default (rtp_mux->srcpad, GST_OBJECT (rtp_mux), event);
}
static void
-free_pad_private (gpointer data, GObject * where_the_object_was)
+gst_rtp_mux_init (GstRTPMux * rtp_mux)
{
- GstRTPMuxPadPrivate *padpriv = data;
+ GstElementClass *klass = GST_ELEMENT_GET_CLASS (rtp_mux);
+
+ rtp_mux->srcpad =
+ gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
+ "src"), "src");
+ gst_pad_set_event_function (rtp_mux->srcpad,
+ GST_DEBUG_FUNCPTR (gst_rtp_mux_src_event));
+ gst_pad_use_fixed_caps (rtp_mux->srcpad);
+ gst_element_add_pad (GST_ELEMENT (rtp_mux), rtp_mux->srcpad);
+
+ rtp_mux->ssrc = DEFAULT_SSRC;
+ rtp_mux->current_ssrc = DEFAULT_SSRC;
+ rtp_mux->ts_offset = DEFAULT_TIMESTAMP_OFFSET;
+ rtp_mux->seqnum_offset = DEFAULT_SEQNUM_OFFSET;
- gst_caps_replace (&padpriv->out_caps, NULL);
- g_slice_free (GstRTPMuxPadPrivate, padpriv);
+ rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
}
static void
gst_rtp_mux_setup_sinkpad (GstRTPMux * rtp_mux, GstPad * sinkpad)
{
- GstRTPMuxClass *klass;
GstRTPMuxPadPrivate *padpriv = g_slice_new0 (GstRTPMuxPadPrivate);
- klass = GST_RTP_MUX_GET_CLASS (rtp_mux);
-
/* setup some pad functions */
- gst_pad_set_setcaps_function (sinkpad, gst_rtp_mux_setcaps);
- gst_pad_set_getcaps_function (sinkpad, gst_rtp_mux_getcaps);
- if (klass->chain_func)
- gst_pad_set_chain_function (sinkpad, klass->chain_func);
- if (klass->sink_event_func)
- gst_pad_set_event_function (sinkpad, klass->sink_event_func);
-
- /* This could break with gstreamer 0.10.9 */
- gst_pad_set_active (sinkpad, TRUE);
+ gst_pad_set_chain_function (sinkpad, GST_DEBUG_FUNCPTR (gst_rtp_mux_chain));
+ gst_pad_set_chain_list_function (sinkpad,
+ GST_DEBUG_FUNCPTR (gst_rtp_mux_chain_list));
+ gst_pad_set_event_function (sinkpad,
+ GST_DEBUG_FUNCPTR (gst_rtp_mux_sink_event));
+ gst_pad_set_query_function (sinkpad,
+ GST_DEBUG_FUNCPTR (gst_rtp_mux_sink_query));
+
+
+ gst_segment_init (&padpriv->segment, GST_FORMAT_UNDEFINED);
gst_pad_set_element_private (sinkpad, padpriv);
- g_object_weak_ref (G_OBJECT (sinkpad), free_pad_private, padpriv);
- /* dd the pad to the element */
+ gst_pad_set_active (sinkpad, TRUE);
gst_element_add_pad (GST_ELEMENT (rtp_mux), sinkpad);
}
static GstPad *
gst_rtp_mux_request_new_pad (GstElement * element,
- GstPadTemplate * templ, const gchar * req_name)
+ GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
{
GstRTPMux *rtp_mux;
GstPad *newpad;
return NULL;
}
- newpad = gst_rtp_mux_create_sinkpad (rtp_mux, templ);
+ newpad = gst_pad_new_from_template (templ, req_name);
if (newpad)
gst_rtp_mux_setup_sinkpad (rtp_mux, newpad);
else
return newpad;
}
-/* Put our own clock-base on the buffer */
static void
-gst_rtp_mux_readjust_rtp_timestamp (GstRTPMux * rtp_mux, GstPad * pad,
- GstBuffer * buffer)
+gst_rtp_mux_release_pad (GstElement * element, GstPad * pad)
+{
+ GstRTPMuxPadPrivate *padpriv;
+
+ GST_OBJECT_LOCK (element);
+ padpriv = gst_pad_get_element_private (pad);
+ gst_pad_set_element_private (pad, NULL);
+ GST_OBJECT_UNLOCK (element);
+
+ gst_element_remove_pad (element, pad);
+
+ if (padpriv) {
+ g_slice_free (GstRTPMuxPadPrivate, padpriv);
+ }
+}
+
+/* Put our own timestamp-offset on the buffer */
+static void
+gst_rtp_mux_readjust_rtp_timestamp_locked (GstRTPMux * rtp_mux,
+ GstRTPMuxPadPrivate * padpriv, GstRTPBuffer * rtpbuffer)
{
guint32 ts;
guint32 sink_ts_base = 0;
- GstRTPMuxPadPrivate *padpriv = gst_pad_get_element_private (pad);
- if (padpriv->have_clock_base)
- sink_ts_base = padpriv->clock_base;
+ if (padpriv && padpriv->have_timestamp_offset)
+ sink_ts_base = padpriv->timestamp_offset;
- ts = gst_rtp_buffer_get_timestamp (buffer) - sink_ts_base + rtp_mux->ts_base;
+ ts = gst_rtp_buffer_get_timestamp (rtpbuffer) - sink_ts_base +
+ rtp_mux->ts_base;
GST_LOG_OBJECT (rtp_mux, "Re-adjusting RTP ts %u to %u",
- gst_rtp_buffer_get_timestamp (buffer), ts);
- gst_rtp_buffer_set_timestamp (buffer, ts);
+ gst_rtp_buffer_get_timestamp (rtpbuffer), ts);
+ gst_rtp_buffer_set_timestamp (rtpbuffer, ts);
+}
+
+static gboolean
+process_buffer_locked (GstRTPMux * rtp_mux, GstRTPMuxPadPrivate * padpriv,
+ GstRTPBuffer * rtpbuffer)
+{
+ GstRTPMuxClass *klass = GST_RTP_MUX_GET_CLASS (rtp_mux);
+
+ if (klass->accept_buffer_locked)
+ if (!klass->accept_buffer_locked (rtp_mux, padpriv, rtpbuffer))
+ return FALSE;
+
+ rtp_mux->seqnum++;
+ gst_rtp_buffer_set_seq (rtpbuffer, rtp_mux->seqnum);
+
+ gst_rtp_buffer_set_ssrc (rtpbuffer, rtp_mux->current_ssrc);
+ gst_rtp_mux_readjust_rtp_timestamp_locked (rtp_mux, padpriv, rtpbuffer);
+ GST_LOG_OBJECT (rtp_mux,
+ "Pushing packet size %" G_GSIZE_FORMAT ", seq=%d, ts=%u, ssrc=%x",
+ rtpbuffer->map[0].size, rtp_mux->seqnum,
+ gst_rtp_buffer_get_timestamp (rtpbuffer), rtp_mux->current_ssrc);
+
+ if (padpriv) {
+ if (padpriv->segment.format == GST_FORMAT_TIME) {
+ GST_BUFFER_PTS (rtpbuffer->buffer) =
+ gst_segment_to_running_time (&padpriv->segment, GST_FORMAT_TIME,
+ GST_BUFFER_PTS (rtpbuffer->buffer));
+ GST_BUFFER_DTS (rtpbuffer->buffer) =
+ gst_segment_to_running_time (&padpriv->segment, GST_FORMAT_TIME,
+ GST_BUFFER_DTS (rtpbuffer->buffer));
+ }
+ }
+
+ return TRUE;
+}
+
+struct BufferListData
+{
+ GstRTPMux *rtp_mux;
+ GstRTPMuxPadPrivate *padpriv;
+ gboolean drop;
+};
+
+static gboolean
+process_list_item (GstBuffer ** buffer, guint idx, gpointer user_data)
+{
+ struct BufferListData *bd = user_data;
+ GstRTPBuffer rtpbuffer = GST_RTP_BUFFER_INIT;
+
+ *buffer = gst_buffer_make_writable (*buffer);
+
+ gst_rtp_buffer_map (*buffer, GST_MAP_READWRITE, &rtpbuffer);
+
+ bd->drop = !process_buffer_locked (bd->rtp_mux, bd->padpriv, &rtpbuffer);
+
+ gst_rtp_buffer_unmap (&rtpbuffer);
+
+ if (bd->drop)
+ return FALSE;
+
+ if (GST_BUFFER_DURATION_IS_VALID (*buffer) &&
+ GST_BUFFER_PTS_IS_VALID (*buffer))
+ bd->rtp_mux->last_stop = GST_BUFFER_PTS (*buffer) +
+ GST_BUFFER_DURATION (*buffer);
+ else
+ bd->rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
+
+ return TRUE;
}
+static gboolean resend_events (GstPad * pad, GstEvent ** event,
+ gpointer user_data);
+
static GstFlowReturn
-gst_rtp_mux_chain (GstPad * pad, GstBuffer * buffer)
+gst_rtp_mux_chain_list (GstPad * pad, GstObject * parent,
+ GstBufferList * bufferlist)
{
GstRTPMux *rtp_mux;
GstFlowReturn ret;
- GstRTPMuxPadPrivate *padpriv = gst_pad_get_element_private (pad);
+ GstRTPMuxPadPrivate *padpriv;
+ gboolean changed = FALSE;
+ struct BufferListData bd;
+
+ rtp_mux = GST_RTP_MUX (parent);
+
+ if (gst_pad_check_reconfigure (rtp_mux->srcpad)) {
+ GstCaps *current_caps = gst_pad_get_current_caps (pad);
+
+ if (!gst_rtp_mux_setcaps (pad, rtp_mux, current_caps)) {
+ gst_pad_mark_reconfigure (rtp_mux->srcpad);
+ if (GST_PAD_IS_FLUSHING (rtp_mux->srcpad))
+ ret = GST_FLOW_FLUSHING;
+ else
+ ret = GST_FLOW_NOT_NEGOTIATED;
+ gst_buffer_list_unref (bufferlist);
+ goto out;
+ }
+ gst_caps_unref (current_caps);
+ }
+
+ GST_OBJECT_LOCK (rtp_mux);
+
+ padpriv = gst_pad_get_element_private (pad);
+ if (!padpriv) {
+ GST_OBJECT_UNLOCK (rtp_mux);
+ ret = GST_FLOW_NOT_LINKED;
+ gst_buffer_list_unref (bufferlist);
+ goto out;
+ }
+
+ bd.rtp_mux = rtp_mux;
+ bd.padpriv = padpriv;
+ bd.drop = FALSE;
+
+ bufferlist = gst_buffer_list_make_writable (bufferlist);
+ gst_buffer_list_foreach (bufferlist, process_list_item, &bd);
+
+ if (!bd.drop && pad != rtp_mux->last_pad) {
+ changed = TRUE;
+ g_clear_object (&rtp_mux->last_pad);
+ rtp_mux->last_pad = g_object_ref (pad);
+ }
+
+ GST_OBJECT_UNLOCK (rtp_mux);
+
+ if (changed)
+ gst_pad_sticky_events_foreach (pad, resend_events, rtp_mux);
- rtp_mux = GST_RTP_MUX (gst_pad_get_parent (pad));
+ if (bd.drop) {
+ gst_buffer_list_unref (bufferlist);
+ ret = GST_FLOW_OK;
+ } else {
+ ret = gst_pad_push_list (rtp_mux->srcpad, bufferlist);
+ }
+
+out:
+
+ return ret;
+}
+
+static gboolean
+resend_events (GstPad * pad, GstEvent ** event, gpointer user_data)
+{
+ GstRTPMux *rtp_mux = user_data;
+
+ if (GST_EVENT_TYPE (*event) == GST_EVENT_CAPS) {
+ GstCaps *caps;
+
+ gst_event_parse_caps (*event, &caps);
+ gst_rtp_mux_setcaps (pad, rtp_mux, caps);
+ } else if (GST_EVENT_TYPE (*event) == GST_EVENT_SEGMENT) {
+ GstSegment new_segment;
+ gst_segment_init (&new_segment, GST_FORMAT_TIME);
+ gst_pad_push_event (rtp_mux->srcpad, gst_event_new_segment (&new_segment));
+ } else {
+ gst_pad_push_event (rtp_mux->srcpad, gst_event_ref (*event));
+ }
+
+ return TRUE;
+}
- if (!gst_rtp_buffer_validate (buffer)) {
+static GstFlowReturn
+gst_rtp_mux_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
+{
+ GstRTPMux *rtp_mux;
+ GstFlowReturn ret;
+ GstRTPMuxPadPrivate *padpriv;
+ gboolean drop;
+ gboolean changed = FALSE;
+ GstRTPBuffer rtpbuffer = GST_RTP_BUFFER_INIT;
+
+ rtp_mux = GST_RTP_MUX (parent);
+
+ if (gst_pad_check_reconfigure (rtp_mux->srcpad)) {
+ GstCaps *current_caps = gst_pad_get_current_caps (pad);
+
+ if (!gst_rtp_mux_setcaps (pad, rtp_mux, current_caps)) {
+ gst_pad_mark_reconfigure (rtp_mux->srcpad);
+ if (GST_PAD_IS_FLUSHING (rtp_mux->srcpad))
+ ret = GST_FLOW_FLUSHING;
+ else
+ ret = GST_FLOW_NOT_NEGOTIATED;
+ gst_buffer_unref (buffer);
+ goto out;
+ }
+ gst_caps_unref (current_caps);
+ }
+
+ GST_OBJECT_LOCK (rtp_mux);
+ padpriv = gst_pad_get_element_private (pad);
+
+ if (!padpriv) {
+ GST_OBJECT_UNLOCK (rtp_mux);
+ gst_buffer_unref (buffer);
+ return GST_FLOW_NOT_LINKED;
+ }
+
+ buffer = gst_buffer_make_writable (buffer);
+
+ if (!gst_rtp_buffer_map (buffer, GST_MAP_READWRITE, &rtpbuffer)) {
+ GST_OBJECT_UNLOCK (rtp_mux);
+ gst_buffer_unref (buffer);
GST_ERROR_OBJECT (rtp_mux, "Invalid RTP buffer");
- gst_object_unref (rtp_mux);
return GST_FLOW_ERROR;
}
- buffer = gst_buffer_make_writable (buffer);
+ drop = !process_buffer_locked (rtp_mux, padpriv, &rtpbuffer);
+
+ gst_rtp_buffer_unmap (&rtpbuffer);
+
+ if (!drop) {
+ if (pad != rtp_mux->last_pad) {
+ changed = TRUE;
+ g_clear_object (&rtp_mux->last_pad);
+ rtp_mux->last_pad = g_object_ref (pad);
+ }
+
+ if (GST_BUFFER_DURATION_IS_VALID (buffer) &&
+ GST_BUFFER_PTS_IS_VALID (buffer))
+ rtp_mux->last_stop = GST_BUFFER_PTS (buffer) +
+ GST_BUFFER_DURATION (buffer);
+ else
+ rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
+ }
- GST_OBJECT_LOCK (rtp_mux);
- rtp_mux->seqnum++;
- gst_rtp_buffer_set_seq (buffer, rtp_mux->seqnum);
GST_OBJECT_UNLOCK (rtp_mux);
- gst_rtp_buffer_set_ssrc (buffer, rtp_mux->current_ssrc);
- gst_rtp_mux_readjust_rtp_timestamp (rtp_mux, pad, buffer);
- GST_LOG_OBJECT (rtp_mux, "Pushing packet size %d, seq=%d, ts=%u",
- GST_BUFFER_SIZE (buffer), rtp_mux->seqnum,
- gst_rtp_buffer_get_timestamp (buffer));
- gst_buffer_set_caps (buffer, padpriv->out_caps);
+ if (changed)
+ gst_pad_sticky_events_foreach (pad, resend_events, rtp_mux);
- ret = gst_pad_push (rtp_mux->srcpad, buffer);
+ if (drop) {
+ gst_buffer_unref (buffer);
+ ret = GST_FLOW_OK;
+ } else {
+ ret = gst_pad_push (rtp_mux->srcpad, buffer);
+ }
- gst_object_unref (rtp_mux);
+out:
return ret;
}
static gboolean
-gst_rtp_mux_setcaps (GstPad * pad, GstCaps * caps)
+gst_rtp_mux_setcaps (GstPad * pad, GstRTPMux * rtp_mux, GstCaps * caps)
{
- GstRTPMux *rtp_mux;
GstStructure *structure;
- gboolean ret = TRUE;
- GstRTPMuxPadPrivate *padpriv = gst_pad_get_element_private (pad);
+ gboolean ret = FALSE;
+ GstRTPMuxPadPrivate *padpriv;
+ GstCaps *peercaps;
+
+ if (caps == NULL)
+ return FALSE;
+
+ if (!gst_caps_is_fixed (caps))
+ return FALSE;
+
+ peercaps = gst_pad_peer_query_caps (rtp_mux->srcpad, NULL);
+ if (peercaps) {
+ GstCaps *tcaps, *othercaps;;
+ tcaps = gst_pad_get_pad_template_caps (pad);
+ othercaps = gst_caps_intersect_full (peercaps, tcaps,
+ GST_CAPS_INTERSECT_FIRST);
+
+ if (gst_caps_get_size (othercaps) > 0) {
+ structure = gst_caps_get_structure (othercaps, 0);
+ GST_OBJECT_LOCK (rtp_mux);
+ if (gst_structure_get_uint (structure, "ssrc", &rtp_mux->current_ssrc)) {
+ GST_INFO_OBJECT (pad, "Use downstream ssrc: %x", rtp_mux->current_ssrc);
+ rtp_mux->have_ssrc = TRUE;
+ }
+ if (gst_structure_get_uint (structure,
+ "timestamp-offset", &rtp_mux->ts_base)) {
+ GST_INFO_OBJECT (pad, "Use downstream timestamp-offset: %u",
+ rtp_mux->ts_base);
+ }
+ GST_OBJECT_UNLOCK (rtp_mux);
+ }
- rtp_mux = GST_RTP_MUX (gst_pad_get_parent (pad));
+ gst_caps_unref (othercaps);
+
+ gst_caps_unref (peercaps);
+ gst_caps_unref (tcaps);
+ }
structure = gst_caps_get_structure (caps, 0);
- if (!ret)
- goto out;
+ if (!structure)
+ return FALSE;
- if (gst_structure_get_uint (structure, "clock-base", &padpriv->clock_base)) {
- padpriv->have_clock_base = TRUE;
+ GST_OBJECT_LOCK (rtp_mux);
+ padpriv = gst_pad_get_element_private (pad);
+ if (padpriv &&
+ gst_structure_get_uint (structure, "timestamp-offset",
+ &padpriv->timestamp_offset)) {
+ padpriv->have_timestamp_offset = TRUE;
}
caps = gst_caps_copy (caps);
+ /* if we don't have a specified ssrc, first try to take one from the caps,
+ and if that fails, generate one */
+ if (rtp_mux->ssrc == DEFAULT_SSRC) {
+ if (rtp_mux->current_ssrc == DEFAULT_SSRC) {
+ if (!gst_structure_get_uint (structure, "ssrc", &rtp_mux->current_ssrc)) {
+ rtp_mux->current_ssrc = g_random_int ();
+ GST_INFO_OBJECT (rtp_mux, "Set random ssrc %x", rtp_mux->current_ssrc);
+ }
+ }
+ } else {
+ rtp_mux->current_ssrc = rtp_mux->ssrc;
+ GST_INFO_OBJECT (rtp_mux, "Set ssrc %x", rtp_mux->current_ssrc);
+ }
+
gst_caps_set_simple (caps,
- "clock-base", G_TYPE_UINT, rtp_mux->ts_base,
- "seqnum-base", G_TYPE_UINT, rtp_mux->seqnum_base, NULL);
+ "timestamp-offset", G_TYPE_UINT, rtp_mux->ts_base,
+ "seqnum-offset", G_TYPE_UINT, rtp_mux->seqnum_base,
+ "ssrc", G_TYPE_UINT, rtp_mux->current_ssrc, NULL);
+
+ GST_OBJECT_UNLOCK (rtp_mux);
+
+ if (rtp_mux->send_stream_start) {
+ gchar s_id[32];
+
+ /* stream-start (FIXME: create id based on input ids) */
+ g_snprintf (s_id, sizeof (s_id), "interleave-%08x", g_random_int ());
+ gst_pad_push_event (rtp_mux->srcpad, gst_event_new_stream_start (s_id));
+
+ rtp_mux->send_stream_start = FALSE;
+ }
GST_DEBUG_OBJECT (rtp_mux,
"setting caps %" GST_PTR_FORMAT " on src pad..", caps);
ret = gst_pad_set_caps (rtp_mux->srcpad, caps);
- if (ret)
- gst_caps_replace (&padpriv->out_caps, caps);
- else
- gst_caps_unref (caps);
-out:
- gst_object_unref (rtp_mux);
+ gst_caps_unref (caps);
return ret;
}
}
static gboolean
-same_clock_rate_fold (gpointer item, GValue * ret, gpointer user_data)
+same_clock_rate_fold (const GValue * item, GValue * ret, gpointer user_data)
{
GstPad *mypad = user_data;
- GstPad *pad = item;
+ GstPad *pad = g_value_get_object (item);
GstCaps *peercaps;
- GstCaps *othercaps;
- const GstCaps *accumcaps;
- GstCaps *intersect;
+ GstCaps *accumcaps;
if (pad == mypad)
return TRUE;
- peercaps = gst_pad_peer_get_caps (pad);
- if (!peercaps)
+ accumcaps = g_value_get_boxed (ret);
+ peercaps = gst_pad_peer_query_caps (pad, accumcaps);
+ if (!peercaps) {
+ g_warning ("no peercaps");
return TRUE;
+ }
+ peercaps = gst_caps_make_writable (peercaps);
+ clear_caps (peercaps, TRUE);
- othercaps = gst_caps_intersect (peercaps,
- gst_pad_get_pad_template_caps (pad));
- gst_caps_unref (peercaps);
-
- accumcaps = gst_value_get_caps (ret);
-
- clear_caps (othercaps, TRUE);
-
- intersect = gst_caps_intersect (accumcaps, othercaps);
-
- g_value_take_boxed (ret, intersect);
-
- gst_caps_unref (othercaps);
+ g_value_take_boxed (ret, peercaps);
- return !gst_caps_is_empty (intersect);
+ return !gst_caps_is_empty (peercaps);
}
static GstCaps *
-gst_rtp_mux_getcaps (GstPad * pad)
+gst_rtp_mux_getcaps (GstPad * pad, GstRTPMux * mux, GstCaps * filter)
{
- GstRTPMux *mux = GST_RTP_MUX (gst_pad_get_parent (pad));
GstCaps *caps = NULL;
GstIterator *iter = NULL;
GValue v = { 0 };
GstIteratorResult res;
- GstCaps *peercaps = gst_pad_peer_get_caps (mux->srcpad);
- GstCaps *othercaps = NULL;
+ GstCaps *peercaps;
+ GstCaps *othercaps;
+ GstCaps *tcaps;
+ const GstStructure *structure;
+
+ peercaps = gst_pad_peer_query_caps (mux->srcpad, NULL);
if (peercaps) {
- othercaps = gst_caps_intersect (peercaps,
- gst_pad_get_pad_template_caps (pad));
+ tcaps = gst_pad_get_pad_template_caps (pad);
+ othercaps = gst_caps_intersect_full (peercaps, tcaps,
+ GST_CAPS_INTERSECT_FIRST);
gst_caps_unref (peercaps);
} else {
- othercaps = gst_caps_copy (gst_pad_get_pad_template_caps (mux->srcpad));
+ tcaps = gst_pad_get_pad_template_caps (mux->srcpad);
+ if (filter)
+ othercaps = gst_caps_intersect_full (filter, tcaps,
+ GST_CAPS_INTERSECT_FIRST);
+ else
+ othercaps = gst_caps_copy (tcaps);
+ }
+ gst_caps_unref (tcaps);
+
+ GST_LOG_OBJECT (pad, "Intersected srcpad-peercaps and template caps: %"
+ GST_PTR_FORMAT, othercaps);
+
+ structure = gst_caps_get_structure (othercaps, 0);
+ if (mux->ssrc == DEFAULT_SSRC) {
+ if (gst_structure_get_uint (structure, "ssrc", &mux->current_ssrc))
+ GST_DEBUG_OBJECT (pad, "Use downstream ssrc: %x", mux->current_ssrc);
}
- clear_caps (othercaps, FALSE);
+ clear_caps (othercaps, TRUE);
g_value_init (&v, GST_TYPE_CAPS);
do {
gst_value_set_caps (&v, othercaps);
res = gst_iterator_fold (iter, same_clock_rate_fold, &v, pad);
+ gst_iterator_resync (iter);
} while (res == GST_ITERATOR_RESYNC);
gst_iterator_free (iter);
- caps = (GstCaps *) gst_value_get_caps (&v);
+ caps = gst_caps_intersect ((GstCaps *) gst_value_get_caps (&v), othercaps);
+
+ g_value_unset (&v);
+ gst_caps_unref (othercaps);
if (res == GST_ITERATOR_ERROR) {
gst_caps_unref (caps);
caps = gst_caps_new_empty ();
}
- if (othercaps)
- gst_caps_unref (othercaps);
- gst_object_unref (mux);
return caps;
}
+static gboolean
+gst_rtp_mux_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
+{
+ GstRTPMux *mux = GST_RTP_MUX (parent);
+ gboolean res = FALSE;
+
+ switch (GST_QUERY_TYPE (query)) {
+ case GST_QUERY_CAPS:
+ {
+ GstCaps *filter, *caps;
+
+ gst_query_parse_caps (query, &filter);
+ GST_LOG_OBJECT (pad, "Received caps-query with filter-caps: %"
+ GST_PTR_FORMAT, filter);
+ caps = gst_rtp_mux_getcaps (pad, mux, filter);
+ gst_query_set_caps_result (query, caps);
+ GST_LOG_OBJECT (mux, "Answering caps-query with caps: %"
+ GST_PTR_FORMAT, caps);
+ gst_caps_unref (caps);
+ res = TRUE;
+ break;
+ }
+ default:
+ res = gst_pad_query_default (pad, parent, query);
+ break;
+ }
+
+ return res;
+}
static void
gst_rtp_mux_get_property (GObject * object,
rtp_mux = GST_RTP_MUX (object);
+ GST_OBJECT_LOCK (rtp_mux);
switch (prop_id) {
case PROP_TIMESTAMP_OFFSET:
g_value_set_int (value, rtp_mux->ts_offset);
g_value_set_int (value, rtp_mux->seqnum_offset);
break;
case PROP_SEQNUM:
- GST_OBJECT_LOCK (rtp_mux);
g_value_set_uint (value, rtp_mux->seqnum);
- GST_OBJECT_UNLOCK (rtp_mux);
break;
case PROP_SSRC:
g_value_set_uint (value, rtp_mux->ssrc);
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
+ GST_OBJECT_UNLOCK (rtp_mux);
}
static void
rtp_mux->seqnum_offset = g_value_get_int (value);
break;
case PROP_SSRC:
+ GST_OBJECT_LOCK (rtp_mux);
rtp_mux->ssrc = g_value_get_uint (value);
+ rtp_mux->current_ssrc = rtp_mux->ssrc;
+ rtp_mux->have_ssrc = TRUE;
+ GST_DEBUG_OBJECT (rtp_mux, "ssrc prop set to %x", rtp_mux->ssrc);
+ GST_OBJECT_UNLOCK (rtp_mux);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
}
}
+static gboolean
+gst_rtp_mux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
+{
+ GstRTPMux *mux = GST_RTP_MUX (parent);
+ gboolean is_pad;
+ gboolean ret;
+
+ GST_OBJECT_LOCK (mux);
+ is_pad = (pad == mux->last_pad);
+ GST_OBJECT_UNLOCK (mux);
+
+ switch (GST_EVENT_TYPE (event)) {
+ case GST_EVENT_CAPS:
+ {
+ GstCaps *caps;
+
+ gst_event_parse_caps (event, &caps);
+ GST_LOG_OBJECT (pad, "Received caps-event with caps: %"
+ GST_PTR_FORMAT, caps);
+ ret = gst_rtp_mux_setcaps (pad, mux, caps);
+ gst_event_unref (event);
+ return ret;
+ }
+ case GST_EVENT_FLUSH_STOP:
+ {
+ GST_OBJECT_LOCK (mux);
+ mux->last_stop = GST_CLOCK_TIME_NONE;
+ GST_OBJECT_UNLOCK (mux);
+ break;
+ }
+ case GST_EVENT_SEGMENT:
+ {
+ GstRTPMuxPadPrivate *padpriv;
+
+ GST_OBJECT_LOCK (mux);
+ padpriv = gst_pad_get_element_private (pad);
+
+ if (padpriv) {
+ gst_event_copy_segment (event, &padpriv->segment);
+ }
+ GST_OBJECT_UNLOCK (mux);
+
+ if (is_pad) {
+ GstSegment new_segment;
+ gst_segment_init (&new_segment, GST_FORMAT_TIME);
+ gst_event_unref (event);
+ event = gst_event_new_segment (&new_segment);
+ }
+ break;
+ }
+ default:
+ break;
+ }
+
+ if (is_pad) {
+ return gst_pad_push_event (mux->srcpad, event);
+ } else {
+ gst_event_unref (event);
+ return TRUE;
+ }
+}
+
static void
gst_rtp_mux_ready_to_paused (GstRTPMux * rtp_mux)
{
+
GST_OBJECT_LOCK (rtp_mux);
- if (rtp_mux->ssrc == -1)
- rtp_mux->current_ssrc = g_random_int ();
- else
- rtp_mux->current_ssrc = rtp_mux->ssrc;
+ g_clear_object (&rtp_mux->last_pad);
+ rtp_mux->send_stream_start = TRUE;
if (rtp_mux->seqnum_offset == -1)
rtp_mux->seqnum_base = g_random_int_range (0, G_MAXUINT16);
rtp_mux->ts_base = g_random_int ();
else
rtp_mux->ts_base = rtp_mux->ts_offset;
- GST_DEBUG_OBJECT (rtp_mux, "set clock-base to %u", rtp_mux->ts_base);
+
+ rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
+
+ if (rtp_mux->have_ssrc)
+ rtp_mux->current_ssrc = rtp_mux->ssrc;
+
+ GST_DEBUG_OBJECT (rtp_mux, "set timestamp-offset to %u", rtp_mux->ts_base);
GST_OBJECT_UNLOCK (rtp_mux);
}
gst_rtp_mux_change_state (GstElement * element, GstStateChange transition)
{
GstRTPMux *rtp_mux;
+ GstStateChangeReturn ret;
rtp_mux = GST_RTP_MUX (element);
switch (transition) {
- case GST_STATE_CHANGE_NULL_TO_READY:
- break;
case GST_STATE_CHANGE_READY_TO_PAUSED:
gst_rtp_mux_ready_to_paused (rtp_mux);
break;
+ default:
+ break;
+ }
+
+ ret = GST_ELEMENT_CLASS (gst_rtp_mux_parent_class)->change_state (element,
+ transition);
+
+ switch (transition) {
case GST_STATE_CHANGE_PAUSED_TO_READY:
+ g_clear_object (&rtp_mux->last_pad);
break;
default:
break;
}
- return GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
+ return ret;
}
gboolean
projects / platform / upstream / gst-plugins-good.git / blobdiff