gst/rtpmanager/gstrtpmux.c

   1 /* RTP muxer element for GStreamer
   2  *
   3  * gstrtpmux.c:
   4  *
   5  * Copyright (C) <2007-2010> Nokia Corporation.
   6  *   Contact: Zeeshan Ali <zeeshan.ali@nokia.com>
   7  * Copyright (C) <2007-2010> Collabora Ltd
   8  *   Contact: Olivier Crete <olivier.crete@collabora.co.uk>
   9  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
  10  *               2000,2005 Wim Taymans <wim@fluendo.com>
  11  *
  12  * This library is free software; you can redistribute it and/or
  13  * modify it under the terms of the GNU Library General Public
  14  * License as published by the Free Software Foundation; either
  15  * version 2 of the License, or (at your option) any later version.
  16  *
  17  * This library is distributed in the hope that it will be useful,
  18  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  19  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  20  * Library General Public License for more details.
  21  *
  22  * You should have received a copy of the GNU Library General Public
  23  * License along with this library; if not, write to the
  24  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  25  * Boston, MA 02110-1301, USA.
  26  */
  27
  28 /**
  29  * SECTION:element-rtpmux
  30  * @see_also: rtpdtmfmux
  31  *
  32  * The rtp muxer takes multiple RTP streams having the same clock-rate and
  33  * muxes into a single stream with a single SSRC.
  34  *
  35  * <refsect2>
  36  * <title>Example pipelines</title>
  37  * |[
  38  * gst-launch-1.0 rtpmux name=mux ! udpsink host=127.0.0.1 port=8888        \
  39  *              alsasrc ! alawenc ! rtppcmapay !                        \
  40  *              application/x-rtp, payload=8, rate=8000 ! mux.sink_0    \
  41  *              audiotestsrc is-live=1 !                                \
  42  *              mulawenc ! rtppcmupay !                                 \
  43  *              application/x-rtp, payload=0, rate=8000 ! mux.sink_1
  44  * ]|
  45  * In this example, an audio stream is captured from ALSA and another is
  46  * generated, both are encoded into different payload types and muxed together
  47  * so they can be sent on the same port.
  48  * </refsect2>
  49  */
  50
  51 #ifdef HAVE_CONFIG_H
  52 #include "config.h"
  53 #endif
  54
  55 #include <gst/gst.h>
  56 #include <gst/rtp/gstrtpbuffer.h>
  57 #include <string.h>
  58
  59 #include "gstrtpmux.h"
  60
  61 GST_DEBUG_CATEGORY_STATIC (gst_rtp_mux_debug);
  62 #define GST_CAT_DEFAULT gst_rtp_mux_debug
  63
  64 enum
  65 {
  66   PROP_0,
  67   PROP_TIMESTAMP_OFFSET,
  68   PROP_SEQNUM_OFFSET,
  69   PROP_SEQNUM,
  70   PROP_SSRC
  71 };
  72
  73 #define DEFAULT_TIMESTAMP_OFFSET -1
  74 #define DEFAULT_SEQNUM_OFFSET    -1
  75 #define DEFAULT_SSRC             -1
  76
  77 static GstStaticPadTemplate src_factory = GST_STATIC_PAD_TEMPLATE ("src",
  78     GST_PAD_SRC,
  79     GST_PAD_ALWAYS,
  80     GST_STATIC_CAPS ("application/x-rtp")
  81     );
  82
  83 static GstStaticPadTemplate sink_factory = GST_STATIC_PAD_TEMPLATE ("sink_%u",
  84     GST_PAD_SINK,
  85     GST_PAD_REQUEST,
  86     GST_STATIC_CAPS ("application/x-rtp")
  87     );
  88
  89 static GstPad *gst_rtp_mux_request_new_pad (GstElement * element,
  90     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
  91 static void gst_rtp_mux_release_pad (GstElement * element, GstPad * pad);
  92 static GstFlowReturn gst_rtp_mux_chain (GstPad * pad, GstObject * parent,
  93     GstBuffer * buffer);
  94 static GstFlowReturn gst_rtp_mux_chain_list (GstPad * pad, GstObject * parent,
  95     GstBufferList * bufferlist);
  96 static gboolean gst_rtp_mux_setcaps (GstPad * pad, GstRTPMux * rtp_mux,
  97     GstCaps * caps);
  98 static gboolean gst_rtp_mux_sink_event (GstPad * pad, GstObject * parent,
  99     GstEvent * event);
 100 static gboolean gst_rtp_mux_sink_query (GstPad * pad, GstObject * parent,
 101     GstQuery * query);
 102
 103 static GstStateChangeReturn gst_rtp_mux_change_state (GstElement *
 104     element, GstStateChange transition);
 105
 106 static void gst_rtp_mux_set_property (GObject * object, guint prop_id,
 107     const GValue * value, GParamSpec * pspec);
 108 static void gst_rtp_mux_get_property (GObject * object, guint prop_id,
 109     GValue * value, GParamSpec * pspec);
 110 static void gst_rtp_mux_dispose (GObject * object);
 111
 112 static gboolean gst_rtp_mux_src_event_real (GstRTPMux * rtp_mux,
 113     GstEvent * event);
 114
 115 G_DEFINE_TYPE (GstRTPMux, gst_rtp_mux, GST_TYPE_ELEMENT);
 116
 117
 118 static void
 119 gst_rtp_mux_class_init (GstRTPMuxClass * klass)
 120 {
 121   GObjectClass *gobject_class;
 122   GstElementClass *gstelement_class;
 123
 124   gobject_class = (GObjectClass *) klass;
 125   gstelement_class = (GstElementClass *) klass;
 126
 127
 128   gst_element_class_add_static_pad_template (gstelement_class, &src_factory);
 129   gst_element_class_add_static_pad_template (gstelement_class, &sink_factory);
 130
 131   gst_element_class_set_static_metadata (gstelement_class, "RTP muxer",
 132       "Codec/Muxer",
 133       "multiplex N rtp streams into one", "Zeeshan Ali <first.last@nokia.com>");
 134
 135   gobject_class->get_property = gst_rtp_mux_get_property;
 136   gobject_class->set_property = gst_rtp_mux_set_property;
 137   gobject_class->dispose = gst_rtp_mux_dispose;
 138
 139   klass->src_event = gst_rtp_mux_src_event_real;
 140
 141   g_object_class_install_property (G_OBJECT_CLASS (klass),
 142       PROP_TIMESTAMP_OFFSET, g_param_spec_int ("timestamp-offset",
 143           "Timestamp Offset",
 144           "Offset to add to all outgoing timestamps (-1 = random)", -1,
 145           G_MAXINT, DEFAULT_TIMESTAMP_OFFSET,
 146           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 147   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SEQNUM_OFFSET,
 148       g_param_spec_int ("seqnum-offset", "Sequence number Offset",
 149           "Offset to add to all outgoing seqnum (-1 = random)", -1, G_MAXINT,
 150           DEFAULT_SEQNUM_OFFSET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 151   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SEQNUM,
 152       g_param_spec_uint ("seqnum", "Sequence number",
 153           "The RTP sequence number of the last processed packet",
 154           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 155   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SSRC,
 156       g_param_spec_uint ("ssrc", "SSRC",
 157           "The SSRC of the packets (default == random)",
 158           0, G_MAXUINT, DEFAULT_SSRC,
 159           GST_PARAM_MUTABLE_PLAYING | G_PARAM_READWRITE |
 160           G_PARAM_STATIC_STRINGS));
 161
 162   gstelement_class->request_new_pad =
 163       GST_DEBUG_FUNCPTR (gst_rtp_mux_request_new_pad);
 164   gstelement_class->release_pad = GST_DEBUG_FUNCPTR (gst_rtp_mux_release_pad);
 165   gstelement_class->change_state = GST_DEBUG_FUNCPTR (gst_rtp_mux_change_state);
 166 }
 167
 168 static void
 169 gst_rtp_mux_dispose (GObject * object)
 170 {
 171   GstRTPMux *rtp_mux = GST_RTP_MUX (object);
 172   GList *item;
 173
 174   g_clear_object (&rtp_mux->last_pad);
 175
 176 restart:
 177   for (item = GST_ELEMENT_PADS (object); item; item = g_list_next (item)) {
 178     GstPad *pad = GST_PAD (item->data);
 179     if (GST_PAD_IS_SINK (pad)) {
 180       gst_element_release_request_pad (GST_ELEMENT (object), pad);
 181       goto restart;
 182     }
 183   }
 184
 185   G_OBJECT_CLASS (gst_rtp_mux_parent_class)->dispose (object);
 186 }
 187
 188 static gboolean
 189 gst_rtp_mux_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
 190 {
 191   GstRTPMux *rtp_mux = GST_RTP_MUX (parent);
 192   GstRTPMuxClass *klass;
 193   gboolean ret;
 194
 195   klass = GST_RTP_MUX_GET_CLASS (rtp_mux);
 196
 197   ret = klass->src_event (rtp_mux, event);
 198
 199   return ret;
 200 }
 201
 202 static gboolean
 203 gst_rtp_mux_src_event_real (GstRTPMux * rtp_mux, GstEvent * event)
 204 {
 205   switch (GST_EVENT_TYPE (event)) {
 206     case GST_EVENT_CUSTOM_UPSTREAM:
 207     {
 208       const GstStructure *s = gst_event_get_structure (event);
 209
 210       if (gst_structure_has_name (s, "GstRTPCollision")) {
 211         guint ssrc = 0;
 212
 213         if (!gst_structure_get_uint (s, "ssrc", &ssrc))
 214           ssrc = -1;
 215
 216         GST_DEBUG_OBJECT (rtp_mux, "collided ssrc: %" G_GUINT32_FORMAT, ssrc);
 217
 218         /* choose another ssrc for our stream */
 219         GST_OBJECT_LOCK (rtp_mux);
 220         if (ssrc == rtp_mux->current_ssrc) {
 221           GstCaps *caps;
 222           guint suggested_ssrc = 0;
 223           guint32 new_ssrc;
 224
 225           if (gst_structure_get_uint (s, "suggested-ssrc", &suggested_ssrc))
 226             rtp_mux->current_ssrc = suggested_ssrc;
 227
 228           while (ssrc == rtp_mux->current_ssrc)
 229             rtp_mux->current_ssrc = g_random_int ();
 230
 231           new_ssrc = rtp_mux->current_ssrc;
 232           GST_OBJECT_UNLOCK (rtp_mux);
 233
 234           caps = gst_pad_get_current_caps (rtp_mux->srcpad);
 235           caps = gst_caps_make_writable (caps);
 236           gst_caps_set_simple (caps, "ssrc", G_TYPE_UINT, new_ssrc, NULL);
 237           gst_pad_set_caps (rtp_mux->srcpad, caps);
 238           gst_caps_unref (caps);
 239         } else {
 240           GST_OBJECT_UNLOCK (rtp_mux);
 241         }
 242       }
 243       break;
 244     }
 245     default:
 246       break;
 247   }
 248
 249
 250   return gst_pad_event_default (rtp_mux->srcpad, GST_OBJECT (rtp_mux), event);
 251 }
 252
 253 static void
 254 gst_rtp_mux_init (GstRTPMux * rtp_mux)
 255 {
 256   GstElementClass *klass = GST_ELEMENT_GET_CLASS (rtp_mux);
 257
 258   rtp_mux->srcpad =
 259       gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
 260           "src"), "src");
 261   gst_pad_set_event_function (rtp_mux->srcpad,
 262       GST_DEBUG_FUNCPTR (gst_rtp_mux_src_event));
 263   gst_pad_use_fixed_caps (rtp_mux->srcpad);
 264   gst_element_add_pad (GST_ELEMENT (rtp_mux), rtp_mux->srcpad);
 265
 266   rtp_mux->ssrc = DEFAULT_SSRC;
 267   rtp_mux->current_ssrc = DEFAULT_SSRC;
 268   rtp_mux->ssrc_random = TRUE;
 269   rtp_mux->ts_offset = DEFAULT_TIMESTAMP_OFFSET;
 270   rtp_mux->seqnum_offset = DEFAULT_SEQNUM_OFFSET;
 271
 272   rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
 273 }
 274
 275 static void
 276 gst_rtp_mux_setup_sinkpad (GstRTPMux * rtp_mux, GstPad * sinkpad)
 277 {
 278   GstRTPMuxPadPrivate *padpriv = g_slice_new0 (GstRTPMuxPadPrivate);
 279
 280   /* setup some pad functions */
 281   gst_pad_set_chain_function (sinkpad, GST_DEBUG_FUNCPTR (gst_rtp_mux_chain));
 282   gst_pad_set_chain_list_function (sinkpad,
 283       GST_DEBUG_FUNCPTR (gst_rtp_mux_chain_list));
 284   gst_pad_set_event_function (sinkpad,
 285       GST_DEBUG_FUNCPTR (gst_rtp_mux_sink_event));
 286   gst_pad_set_query_function (sinkpad,
 287       GST_DEBUG_FUNCPTR (gst_rtp_mux_sink_query));
 288
 289
 290   gst_segment_init (&padpriv->segment, GST_FORMAT_UNDEFINED);
 291
 292   gst_pad_set_element_private (sinkpad, padpriv);
 293
 294   gst_pad_set_active (sinkpad, TRUE);
 295   gst_element_add_pad (GST_ELEMENT (rtp_mux), sinkpad);
 296 }
 297
 298 static GstPad *
 299 gst_rtp_mux_request_new_pad (GstElement * element,
 300     GstPadTemplate * templ, const gchar * req_name, const GstCaps * caps)
 301 {
 302   GstRTPMux *rtp_mux;
 303   GstPad *newpad;
 304
 305   g_return_val_if_fail (templ != NULL, NULL);
 306   g_return_val_if_fail (GST_IS_RTP_MUX (element), NULL);
 307
 308   rtp_mux = GST_RTP_MUX (element);
 309
 310   if (templ->direction != GST_PAD_SINK) {
 311     GST_WARNING_OBJECT (rtp_mux, "request pad that is not a SINK pad");
 312     return NULL;
 313   }
 314
 315   newpad = gst_pad_new_from_template (templ, req_name);
 316   if (newpad)
 317     gst_rtp_mux_setup_sinkpad (rtp_mux, newpad);
 318   else
 319     GST_WARNING_OBJECT (rtp_mux, "failed to create request pad");
 320
 321   return newpad;
 322 }
 323
 324 static void
 325 gst_rtp_mux_release_pad (GstElement * element, GstPad * pad)
 326 {
 327   GstRTPMuxPadPrivate *padpriv;
 328
 329   GST_OBJECT_LOCK (element);
 330   padpriv = gst_pad_get_element_private (pad);
 331   gst_pad_set_element_private (pad, NULL);
 332   GST_OBJECT_UNLOCK (element);
 333
 334   gst_element_remove_pad (element, pad);
 335
 336   if (padpriv) {
 337     g_slice_free (GstRTPMuxPadPrivate, padpriv);
 338   }
 339 }
 340
 341 /* Put our own timestamp-offset on the buffer */
 342 static void
 343 gst_rtp_mux_readjust_rtp_timestamp_locked (GstRTPMux * rtp_mux,
 344     GstRTPMuxPadPrivate * padpriv, GstRTPBuffer * rtpbuffer)
 345 {
 346   guint32 ts;
 347   guint32 sink_ts_base = 0;
 348
 349   if (padpriv && padpriv->have_timestamp_offset)
 350     sink_ts_base = padpriv->timestamp_offset;
 351
 352   ts = gst_rtp_buffer_get_timestamp (rtpbuffer) - sink_ts_base +
 353       rtp_mux->ts_base;
 354   GST_LOG_OBJECT (rtp_mux, "Re-adjusting RTP ts %u to %u",
 355       gst_rtp_buffer_get_timestamp (rtpbuffer), ts);
 356   gst_rtp_buffer_set_timestamp (rtpbuffer, ts);
 357 }
 358
 359 static gboolean
 360 process_buffer_locked (GstRTPMux * rtp_mux, GstRTPMuxPadPrivate * padpriv,
 361     GstRTPBuffer * rtpbuffer)
 362 {
 363   GstRTPMuxClass *klass = GST_RTP_MUX_GET_CLASS (rtp_mux);
 364
 365   if (klass->accept_buffer_locked)
 366     if (!klass->accept_buffer_locked (rtp_mux, padpriv, rtpbuffer))
 367       return FALSE;
 368
 369   rtp_mux->seqnum++;
 370   gst_rtp_buffer_set_seq (rtpbuffer, rtp_mux->seqnum);
 371
 372   gst_rtp_buffer_set_ssrc (rtpbuffer, rtp_mux->current_ssrc);
 373   gst_rtp_mux_readjust_rtp_timestamp_locked (rtp_mux, padpriv, rtpbuffer);
 374   GST_LOG_OBJECT (rtp_mux,
 375       "Pushing packet size %" G_GSIZE_FORMAT ", seq=%d, ts=%u",
 376       rtpbuffer->map[0].size, rtp_mux->seqnum,
 377       gst_rtp_buffer_get_timestamp (rtpbuffer));
 378
 379   if (padpriv) {
 380     if (padpriv->segment.format == GST_FORMAT_TIME) {
 381       GST_BUFFER_PTS (rtpbuffer->buffer) =
 382           gst_segment_to_running_time (&padpriv->segment, GST_FORMAT_TIME,
 383           GST_BUFFER_PTS (rtpbuffer->buffer));
 384       GST_BUFFER_DTS (rtpbuffer->buffer) =
 385           gst_segment_to_running_time (&padpriv->segment, GST_FORMAT_TIME,
 386           GST_BUFFER_DTS (rtpbuffer->buffer));
 387     }
 388   }
 389
 390   return TRUE;
 391 }
 392
 393 struct BufferListData
 394 {
 395   GstRTPMux *rtp_mux;
 396   GstRTPMuxPadPrivate *padpriv;
 397   gboolean drop;
 398 };
 399
 400 static gboolean
 401 process_list_item (GstBuffer ** buffer, guint idx, gpointer user_data)
 402 {
 403   struct BufferListData *bd = user_data;
 404   GstRTPBuffer rtpbuffer = GST_RTP_BUFFER_INIT;
 405
 406   *buffer = gst_buffer_make_writable (*buffer);
 407
 408   gst_rtp_buffer_map (*buffer, GST_MAP_READWRITE, &rtpbuffer);
 409
 410   bd->drop = !process_buffer_locked (bd->rtp_mux, bd->padpriv, &rtpbuffer);
 411
 412   gst_rtp_buffer_unmap (&rtpbuffer);
 413
 414   if (bd->drop)
 415     return FALSE;
 416
 417   if (GST_BUFFER_DURATION_IS_VALID (*buffer) &&
 418       GST_BUFFER_PTS_IS_VALID (*buffer))
 419     bd->rtp_mux->last_stop = GST_BUFFER_PTS (*buffer) +
 420         GST_BUFFER_DURATION (*buffer);
 421   else
 422     bd->rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
 423
 424   return TRUE;
 425 }
 426
 427 static gboolean resend_events (GstPad * pad, GstEvent ** event,
 428     gpointer user_data);
 429
 430 static GstFlowReturn
 431 gst_rtp_mux_chain_list (GstPad * pad, GstObject * parent,
 432     GstBufferList * bufferlist)
 433 {
 434   GstRTPMux *rtp_mux;
 435   GstFlowReturn ret;
 436   GstRTPMuxPadPrivate *padpriv;
 437   gboolean changed = FALSE;
 438   struct BufferListData bd;
 439
 440   rtp_mux = GST_RTP_MUX (parent);
 441
 442   if (gst_pad_check_reconfigure (rtp_mux->srcpad)) {
 443     GstCaps *current_caps = gst_pad_get_current_caps (pad);
 444
 445     if (!gst_rtp_mux_setcaps (pad, rtp_mux, current_caps)) {
 446       gst_pad_mark_reconfigure (rtp_mux->srcpad);
 447       if (GST_PAD_IS_FLUSHING (rtp_mux->srcpad))
 448         ret = GST_FLOW_FLUSHING;
 449       else
 450         ret = GST_FLOW_NOT_NEGOTIATED;
 451       gst_buffer_list_unref (bufferlist);
 452       goto out;
 453     }
 454     gst_caps_unref (current_caps);
 455   }
 456
 457   GST_OBJECT_LOCK (rtp_mux);
 458
 459   padpriv = gst_pad_get_element_private (pad);
 460   if (!padpriv) {
 461     GST_OBJECT_UNLOCK (rtp_mux);
 462     ret = GST_FLOW_NOT_LINKED;
 463     gst_buffer_list_unref (bufferlist);
 464     goto out;
 465   }
 466
 467   bd.rtp_mux = rtp_mux;
 468   bd.padpriv = padpriv;
 469   bd.drop = FALSE;
 470
 471   bufferlist = gst_buffer_list_make_writable (bufferlist);
 472   gst_buffer_list_foreach (bufferlist, process_list_item, &bd);
 473
 474   if (!bd.drop && pad != rtp_mux->last_pad) {
 475     changed = TRUE;
 476     g_clear_object (&rtp_mux->last_pad);
 477     rtp_mux->last_pad = g_object_ref (pad);
 478   }
 479
 480   GST_OBJECT_UNLOCK (rtp_mux);
 481
 482   if (changed)
 483     gst_pad_sticky_events_foreach (pad, resend_events, rtp_mux);
 484
 485   if (bd.drop) {
 486     gst_buffer_list_unref (bufferlist);
 487     ret = GST_FLOW_OK;
 488   } else {
 489     ret = gst_pad_push_list (rtp_mux->srcpad, bufferlist);
 490   }
 491
 492 out:
 493
 494   return ret;
 495 }
 496
 497 static gboolean
 498 resend_events (GstPad * pad, GstEvent ** event, gpointer user_data)
 499 {
 500   GstRTPMux *rtp_mux = user_data;
 501
 502   if (GST_EVENT_TYPE (*event) == GST_EVENT_CAPS) {
 503     GstCaps *caps;
 504
 505     gst_event_parse_caps (*event, &caps);
 506     gst_rtp_mux_setcaps (pad, rtp_mux, caps);
 507   } else if (GST_EVENT_TYPE (*event) == GST_EVENT_SEGMENT) {
 508     GstSegment new_segment;
 509     gst_segment_init (&new_segment, GST_FORMAT_TIME);
 510     gst_pad_push_event (rtp_mux->srcpad, gst_event_new_segment (&new_segment));
 511   } else {
 512     gst_pad_push_event (rtp_mux->srcpad, gst_event_ref (*event));
 513   }
 514
 515   return TRUE;
 516 }
 517
 518 static GstFlowReturn
 519 gst_rtp_mux_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 520 {
 521   GstRTPMux *rtp_mux;
 522   GstFlowReturn ret;
 523   GstRTPMuxPadPrivate *padpriv;
 524   gboolean drop;
 525   gboolean changed = FALSE;
 526   GstRTPBuffer rtpbuffer = GST_RTP_BUFFER_INIT;
 527
 528   rtp_mux = GST_RTP_MUX (parent);
 529
 530   if (gst_pad_check_reconfigure (rtp_mux->srcpad)) {
 531     GstCaps *current_caps = gst_pad_get_current_caps (pad);
 532
 533     if (!gst_rtp_mux_setcaps (pad, rtp_mux, current_caps)) {
 534       gst_pad_mark_reconfigure (rtp_mux->srcpad);
 535       if (GST_PAD_IS_FLUSHING (rtp_mux->srcpad))
 536         ret = GST_FLOW_FLUSHING;
 537       else
 538         ret = GST_FLOW_NOT_NEGOTIATED;
 539       gst_buffer_unref (buffer);
 540       goto out;
 541     }
 542     gst_caps_unref (current_caps);
 543   }
 544
 545   GST_OBJECT_LOCK (rtp_mux);
 546   padpriv = gst_pad_get_element_private (pad);
 547
 548   if (!padpriv) {
 549     GST_OBJECT_UNLOCK (rtp_mux);
 550     gst_buffer_unref (buffer);
 551     return GST_FLOW_NOT_LINKED;
 552   }
 553
 554   buffer = gst_buffer_make_writable (buffer);
 555
 556   if (!gst_rtp_buffer_map (buffer, GST_MAP_READWRITE, &rtpbuffer)) {
 557     GST_OBJECT_UNLOCK (rtp_mux);
 558     gst_buffer_unref (buffer);
 559     GST_ERROR_OBJECT (rtp_mux, "Invalid RTP buffer");
 560     return GST_FLOW_ERROR;
 561   }
 562
 563   drop = !process_buffer_locked (rtp_mux, padpriv, &rtpbuffer);
 564
 565   gst_rtp_buffer_unmap (&rtpbuffer);
 566
 567   if (!drop) {
 568     if (pad != rtp_mux->last_pad) {
 569       changed = TRUE;
 570       g_clear_object (&rtp_mux->last_pad);
 571       rtp_mux->last_pad = g_object_ref (pad);
 572     }
 573
 574     if (GST_BUFFER_DURATION_IS_VALID (buffer) &&
 575         GST_BUFFER_PTS_IS_VALID (buffer))
 576       rtp_mux->last_stop = GST_BUFFER_PTS (buffer) +
 577           GST_BUFFER_DURATION (buffer);
 578     else
 579       rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
 580   }
 581
 582   GST_OBJECT_UNLOCK (rtp_mux);
 583
 584   if (changed)
 585     gst_pad_sticky_events_foreach (pad, resend_events, rtp_mux);
 586
 587   if (drop) {
 588     gst_buffer_unref (buffer);
 589     ret = GST_FLOW_OK;
 590   } else {
 591     ret = gst_pad_push (rtp_mux->srcpad, buffer);
 592   }
 593
 594 out:
 595   return ret;
 596 }
 597
 598 static gboolean
 599 gst_rtp_mux_setcaps (GstPad * pad, GstRTPMux * rtp_mux, GstCaps * caps)
 600 {
 601   GstStructure *structure;
 602   gboolean ret = FALSE;
 603   GstRTPMuxPadPrivate *padpriv;
 604   GstCaps *peercaps;
 605
 606   if (!gst_caps_is_fixed (caps))
 607     return FALSE;
 608
 609   peercaps = gst_pad_peer_query_caps (rtp_mux->srcpad, NULL);
 610   if (peercaps) {
 611     GstCaps *tcaps, *othercaps;;
 612     tcaps = gst_pad_get_pad_template_caps (pad);
 613     othercaps = gst_caps_intersect_full (peercaps, tcaps,
 614         GST_CAPS_INTERSECT_FIRST);
 615
 616     if (gst_caps_get_size (othercaps) > 0) {
 617       structure = gst_caps_get_structure (othercaps, 0);
 618       GST_OBJECT_LOCK (rtp_mux);
 619       if (gst_structure_get_uint (structure, "ssrc", &rtp_mux->current_ssrc)) {
 620         GST_DEBUG_OBJECT (pad, "Use downstream ssrc: %x",
 621             rtp_mux->current_ssrc);
 622         rtp_mux->have_ssrc = TRUE;
 623       }
 624       GST_OBJECT_UNLOCK (rtp_mux);
 625     }
 626
 627     gst_caps_unref (othercaps);
 628
 629     gst_caps_unref (peercaps);
 630     gst_caps_unref (tcaps);
 631   }
 632
 633   structure = gst_caps_get_structure (caps, 0);
 634
 635   if (!structure)
 636     return FALSE;
 637
 638   GST_OBJECT_LOCK (rtp_mux);
 639   padpriv = gst_pad_get_element_private (pad);
 640   if (padpriv &&
 641       gst_structure_get_uint (structure, "timestamp-offset",
 642           &padpriv->timestamp_offset)) {
 643     padpriv->have_timestamp_offset = TRUE;
 644   }
 645
 646   caps = gst_caps_copy (caps);
 647
 648   /* if we don't have a specified ssrc, first try to take one from the caps,
 649      and if that fails, generate one */
 650   if (!rtp_mux->have_ssrc) {
 651     if (rtp_mux->ssrc_random) {
 652       if (!gst_structure_get_uint (structure, "ssrc", &rtp_mux->current_ssrc))
 653         rtp_mux->current_ssrc = g_random_int ();
 654       rtp_mux->have_ssrc = TRUE;
 655     }
 656   }
 657
 658   gst_caps_set_simple (caps,
 659       "timestamp-offset", G_TYPE_UINT, rtp_mux->ts_base,
 660       "seqnum-offset", G_TYPE_UINT, rtp_mux->seqnum_base,
 661       "ssrc", G_TYPE_UINT, rtp_mux->current_ssrc, NULL);
 662
 663   GST_OBJECT_UNLOCK (rtp_mux);
 664
 665   if (rtp_mux->send_stream_start) {
 666     gchar s_id[32];
 667
 668     /* stream-start (FIXME: create id based on input ids) */
 669     g_snprintf (s_id, sizeof (s_id), "interleave-%08x", g_random_int ());
 670     gst_pad_push_event (rtp_mux->srcpad, gst_event_new_stream_start (s_id));
 671
 672     rtp_mux->send_stream_start = FALSE;
 673   }
 674
 675   GST_DEBUG_OBJECT (rtp_mux,
 676       "setting caps %" GST_PTR_FORMAT " on src pad..", caps);
 677   ret = gst_pad_set_caps (rtp_mux->srcpad, caps);
 678
 679
 680   gst_caps_unref (caps);
 681
 682   return ret;
 683 }
 684
 685 static void
 686 clear_caps (GstCaps * caps, gboolean only_clock_rate)
 687 {
 688   gint i, j;
 689
 690   /* Lets only match on the clock-rate */
 691   for (i = 0; i < gst_caps_get_size (caps); i++) {
 692     GstStructure *s = gst_caps_get_structure (caps, i);
 693
 694     for (j = 0; j < gst_structure_n_fields (s); j++) {
 695       const gchar *name = gst_structure_nth_field_name (s, j);
 696
 697       if (strcmp (name, "clock-rate") && (only_clock_rate ||
 698               (strcmp (name, "ssrc")))) {
 699         gst_structure_remove_field (s, name);
 700         j--;
 701       }
 702     }
 703   }
 704 }
 705
 706 static gboolean
 707 same_clock_rate_fold (const GValue * item, GValue * ret, gpointer user_data)
 708 {
 709   GstPad *mypad = user_data;
 710   GstPad *pad = g_value_get_object (item);
 711   GstCaps *peercaps;
 712   GstCaps *accumcaps;
 713
 714   if (pad == mypad)
 715     return TRUE;
 716
 717   accumcaps = g_value_get_boxed (ret);
 718   peercaps = gst_pad_peer_query_caps (pad, accumcaps);
 719   if (!peercaps) {
 720     g_warning ("no peercaps");
 721     return TRUE;
 722   }
 723   peercaps = gst_caps_make_writable (peercaps);
 724   clear_caps (peercaps, TRUE);
 725
 726   g_value_take_boxed (ret, peercaps);
 727
 728   return !gst_caps_is_empty (peercaps);
 729 }
 730
 731 static GstCaps *
 732 gst_rtp_mux_getcaps (GstPad * pad, GstRTPMux * mux, GstCaps * filter)
 733 {
 734   GstCaps *caps = NULL;
 735   GstIterator *iter = NULL;
 736   GValue v = { 0 };
 737   GstIteratorResult res;
 738   GstCaps *peercaps;
 739   GstCaps *othercaps;
 740   GstCaps *tcaps;
 741
 742   peercaps = gst_pad_peer_query_caps (mux->srcpad, NULL);
 743
 744   if (peercaps) {
 745     tcaps = gst_pad_get_pad_template_caps (pad);
 746     othercaps = gst_caps_intersect_full (peercaps, tcaps,
 747         GST_CAPS_INTERSECT_FIRST);
 748     gst_caps_unref (peercaps);
 749   } else {
 750     tcaps = gst_pad_get_pad_template_caps (mux->srcpad);
 751     if (filter)
 752       othercaps = gst_caps_intersect_full (filter, tcaps,
 753           GST_CAPS_INTERSECT_FIRST);
 754     else
 755       othercaps = gst_caps_copy (tcaps);
 756   }
 757   gst_caps_unref (tcaps);
 758
 759   GST_LOG_OBJECT (pad, "Intersected srcpad-peercaps and template caps: %"
 760       GST_PTR_FORMAT, othercaps);
 761
 762   clear_caps (othercaps, TRUE);
 763
 764   g_value_init (&v, GST_TYPE_CAPS);
 765
 766   iter = gst_element_iterate_sink_pads (GST_ELEMENT (mux));
 767   do {
 768     gst_value_set_caps (&v, othercaps);
 769     res = gst_iterator_fold (iter, same_clock_rate_fold, &v, pad);
 770     gst_iterator_resync (iter);
 771   } while (res == GST_ITERATOR_RESYNC);
 772   gst_iterator_free (iter);
 773
 774   caps = gst_caps_intersect ((GstCaps *) gst_value_get_caps (&v), othercaps);
 775
 776   g_value_unset (&v);
 777   gst_caps_unref (othercaps);
 778
 779   if (res == GST_ITERATOR_ERROR) {
 780     gst_caps_unref (caps);
 781     caps = gst_caps_new_empty ();
 782   }
 783
 784
 785   return caps;
 786 }
 787
 788 static gboolean
 789 gst_rtp_mux_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
 790 {
 791   GstRTPMux *mux = GST_RTP_MUX (parent);
 792   gboolean res = FALSE;
 793
 794   switch (GST_QUERY_TYPE (query)) {
 795     case GST_QUERY_CAPS:
 796     {
 797       GstCaps *filter, *caps;
 798
 799       gst_query_parse_caps (query, &filter);
 800       GST_LOG_OBJECT (pad, "Received caps-query with filter-caps: %"
 801           GST_PTR_FORMAT, filter);
 802       caps = gst_rtp_mux_getcaps (pad, mux, filter);
 803       gst_query_set_caps_result (query, caps);
 804       GST_LOG_OBJECT (mux, "Answering caps-query with caps: %"
 805           GST_PTR_FORMAT, caps);
 806       gst_caps_unref (caps);
 807       res = TRUE;
 808       break;
 809     }
 810     default:
 811       res = gst_pad_query_default (pad, parent, query);
 812       break;
 813   }
 814
 815   return res;
 816 }
 817
 818 static void
 819 gst_rtp_mux_get_property (GObject * object,
 820     guint prop_id, GValue * value, GParamSpec * pspec)
 821 {
 822   GstRTPMux *rtp_mux;
 823
 824   rtp_mux = GST_RTP_MUX (object);
 825
 826   GST_OBJECT_LOCK (rtp_mux);
 827   switch (prop_id) {
 828     case PROP_TIMESTAMP_OFFSET:
 829       g_value_set_int (value, rtp_mux->ts_offset);
 830       break;
 831     case PROP_SEQNUM_OFFSET:
 832       g_value_set_int (value, rtp_mux->seqnum_offset);
 833       break;
 834     case PROP_SEQNUM:
 835       g_value_set_uint (value, rtp_mux->seqnum);
 836       break;
 837     case PROP_SSRC:
 838       g_value_set_uint (value, rtp_mux->ssrc);
 839       break;
 840     default:
 841       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 842       break;
 843   }
 844   GST_OBJECT_UNLOCK (rtp_mux);
 845 }
 846
 847 static void
 848 gst_rtp_mux_set_property (GObject * object,
 849     guint prop_id, const GValue * value, GParamSpec * pspec)
 850 {
 851   GstRTPMux *rtp_mux;
 852
 853   rtp_mux = GST_RTP_MUX (object);
 854
 855   switch (prop_id) {
 856     case PROP_TIMESTAMP_OFFSET:
 857       rtp_mux->ts_offset = g_value_get_int (value);
 858       break;
 859     case PROP_SEQNUM_OFFSET:
 860       rtp_mux->seqnum_offset = g_value_get_int (value);
 861       break;
 862     case PROP_SSRC:
 863       GST_OBJECT_LOCK (rtp_mux);
 864       rtp_mux->ssrc = g_value_get_uint (value);
 865       rtp_mux->current_ssrc = rtp_mux->ssrc;
 866       rtp_mux->have_ssrc = TRUE;
 867       rtp_mux->ssrc_random = FALSE;
 868       GST_OBJECT_UNLOCK (rtp_mux);
 869       break;
 870     default:
 871       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 872       break;
 873   }
 874 }
 875
 876 static gboolean
 877 gst_rtp_mux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
 878 {
 879   GstRTPMux *mux = GST_RTP_MUX (parent);
 880   gboolean is_pad;
 881   gboolean ret;
 882
 883   GST_OBJECT_LOCK (mux);
 884   is_pad = (pad == mux->last_pad);
 885   GST_OBJECT_UNLOCK (mux);
 886
 887   switch (GST_EVENT_TYPE (event)) {
 888     case GST_EVENT_CAPS:
 889     {
 890       GstCaps *caps;
 891
 892       gst_event_parse_caps (event, &caps);
 893       GST_LOG_OBJECT (pad, "Received caps-event with caps: %"
 894           GST_PTR_FORMAT, caps);
 895       ret = gst_rtp_mux_setcaps (pad, mux, caps);
 896       gst_event_unref (event);
 897       return ret;
 898     }
 899     case GST_EVENT_FLUSH_STOP:
 900     {
 901       GST_OBJECT_LOCK (mux);
 902       mux->last_stop = GST_CLOCK_TIME_NONE;
 903       GST_OBJECT_UNLOCK (mux);
 904       break;
 905     }
 906     case GST_EVENT_SEGMENT:
 907     {
 908       GstRTPMuxPadPrivate *padpriv;
 909
 910       GST_OBJECT_LOCK (mux);
 911       padpriv = gst_pad_get_element_private (pad);
 912
 913       if (padpriv) {
 914         gst_event_copy_segment (event, &padpriv->segment);
 915       }
 916       GST_OBJECT_UNLOCK (mux);
 917
 918       if (is_pad) {
 919         GstSegment new_segment;
 920         gst_segment_init (&new_segment, GST_FORMAT_TIME);
 921         gst_event_unref (event);
 922         event = gst_event_new_segment (&new_segment);
 923       }
 924       break;
 925     }
 926     default:
 927       break;
 928   }
 929
 930   if (is_pad) {
 931     return gst_pad_push_event (mux->srcpad, event);
 932   } else {
 933     gst_event_unref (event);
 934     return TRUE;
 935   }
 936 }
 937
 938 static void
 939 gst_rtp_mux_ready_to_paused (GstRTPMux * rtp_mux)
 940 {
 941
 942   GST_OBJECT_LOCK (rtp_mux);
 943
 944   g_clear_object (&rtp_mux->last_pad);
 945   rtp_mux->send_stream_start = TRUE;
 946
 947   if (rtp_mux->seqnum_offset == -1)
 948     rtp_mux->seqnum_base = g_random_int_range (0, G_MAXUINT16);
 949   else
 950     rtp_mux->seqnum_base = rtp_mux->seqnum_offset;
 951   rtp_mux->seqnum = rtp_mux->seqnum_base;
 952
 953   if (rtp_mux->ts_offset == -1)
 954     rtp_mux->ts_base = g_random_int ();
 955   else
 956     rtp_mux->ts_base = rtp_mux->ts_offset;
 957
 958   rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
 959
 960   if (rtp_mux->ssrc_random) {
 961     rtp_mux->have_ssrc = FALSE;
 962   } else {
 963     rtp_mux->current_ssrc = rtp_mux->ssrc;
 964     rtp_mux->have_ssrc = TRUE;
 965   }
 966
 967   GST_DEBUG_OBJECT (rtp_mux, "set timestamp-offset to %u", rtp_mux->ts_base);
 968
 969   GST_OBJECT_UNLOCK (rtp_mux);
 970 }
 971
 972 static GstStateChangeReturn
 973 gst_rtp_mux_change_state (GstElement * element, GstStateChange transition)
 974 {
 975   GstRTPMux *rtp_mux;
 976   GstStateChangeReturn ret;
 977
 978   rtp_mux = GST_RTP_MUX (element);
 979
 980   switch (transition) {
 981     case GST_STATE_CHANGE_READY_TO_PAUSED:
 982       gst_rtp_mux_ready_to_paused (rtp_mux);
 983       break;
 984     default:
 985       break;
 986   }
 987
 988   ret = GST_ELEMENT_CLASS (gst_rtp_mux_parent_class)->change_state (element,
 989       transition);
 990
 991   switch (transition) {
 992     case GST_STATE_CHANGE_PAUSED_TO_READY:
 993       g_clear_object (&rtp_mux->last_pad);
 994       break;
 995     default:
 996       break;
 997   }
 998
 999   return ret;
1000 }
1001
1002 gboolean
1003 gst_rtp_mux_plugin_init (GstPlugin * plugin)
1004 {
1005   GST_DEBUG_CATEGORY_INIT (gst_rtp_mux_debug, "rtpmux", 0, "rtp muxer");
1006
1007   return gst_element_register (plugin, "rtpmux", GST_RANK_NONE,
1008       GST_TYPE_RTP_MUX);
1009 }