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   }
 385
 386   return TRUE;
 387 }
 388
 389 struct BufferListData
 390 {
 391   GstRTPMux *rtp_mux;
 392   GstRTPMuxPadPrivate *padpriv;
 393   gboolean drop;
 394 };
 395
 396 static gboolean
 397 process_list_item (GstBuffer ** buffer, guint idx, gpointer user_data)
 398 {
 399   struct BufferListData *bd = user_data;
 400   GstRTPBuffer rtpbuffer = GST_RTP_BUFFER_INIT;
 401
 402   *buffer = gst_buffer_make_writable (*buffer);
 403
 404   gst_rtp_buffer_map (*buffer, GST_MAP_READWRITE, &rtpbuffer);
 405
 406   bd->drop = !process_buffer_locked (bd->rtp_mux, bd->padpriv, &rtpbuffer);
 407
 408   gst_rtp_buffer_unmap (&rtpbuffer);
 409
 410   if (bd->drop)
 411     return FALSE;
 412
 413   if (GST_BUFFER_DURATION_IS_VALID (*buffer) &&
 414       GST_BUFFER_PTS_IS_VALID (*buffer))
 415     bd->rtp_mux->last_stop = GST_BUFFER_PTS (*buffer) +
 416         GST_BUFFER_DURATION (*buffer);
 417   else
 418     bd->rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
 419
 420   return TRUE;
 421 }
 422
 423 static gboolean resend_events (GstPad * pad, GstEvent ** event,
 424     gpointer user_data);
 425
 426 static GstFlowReturn
 427 gst_rtp_mux_chain_list (GstPad * pad, GstObject * parent,
 428     GstBufferList * bufferlist)
 429 {
 430   GstRTPMux *rtp_mux;
 431   GstFlowReturn ret;
 432   GstRTPMuxPadPrivate *padpriv;
 433   gboolean changed = FALSE;
 434   struct BufferListData bd;
 435
 436   rtp_mux = GST_RTP_MUX (parent);
 437
 438   if (gst_pad_check_reconfigure (rtp_mux->srcpad)) {
 439     GstCaps *current_caps = gst_pad_get_current_caps (pad);
 440
 441     if (!gst_rtp_mux_setcaps (pad, rtp_mux, current_caps)) {
 442       ret = GST_FLOW_NOT_NEGOTIATED;
 443       gst_buffer_list_unref (bufferlist);
 444       goto out;
 445     }
 446     gst_caps_unref (current_caps);
 447   }
 448
 449   GST_OBJECT_LOCK (rtp_mux);
 450
 451   padpriv = gst_pad_get_element_private (pad);
 452   if (!padpriv) {
 453     GST_OBJECT_UNLOCK (rtp_mux);
 454     ret = GST_FLOW_NOT_LINKED;
 455     gst_buffer_list_unref (bufferlist);
 456     goto out;
 457   }
 458
 459   bd.rtp_mux = rtp_mux;
 460   bd.padpriv = padpriv;
 461   bd.drop = FALSE;
 462
 463   bufferlist = gst_buffer_list_make_writable (bufferlist);
 464   gst_buffer_list_foreach (bufferlist, process_list_item, &bd);
 465
 466   if (!bd.drop && pad != rtp_mux->last_pad) {
 467     changed = TRUE;
 468     g_clear_object (&rtp_mux->last_pad);
 469     rtp_mux->last_pad = g_object_ref (pad);
 470   }
 471
 472   GST_OBJECT_UNLOCK (rtp_mux);
 473
 474   if (changed)
 475     gst_pad_sticky_events_foreach (pad, resend_events, rtp_mux);
 476
 477   if (bd.drop) {
 478     gst_buffer_list_unref (bufferlist);
 479     ret = GST_FLOW_OK;
 480   } else {
 481     ret = gst_pad_push_list (rtp_mux->srcpad, bufferlist);
 482   }
 483
 484 out:
 485
 486   return ret;
 487 }
 488
 489 static gboolean
 490 resend_events (GstPad * pad, GstEvent ** event, gpointer user_data)
 491 {
 492   GstRTPMux *rtp_mux = user_data;
 493
 494   if (GST_EVENT_TYPE (*event) == GST_EVENT_CAPS) {
 495     GstCaps *caps;
 496
 497     gst_event_parse_caps (*event, &caps);
 498     gst_rtp_mux_setcaps (pad, rtp_mux, caps);
 499   } else {
 500     gst_pad_push_event (rtp_mux->srcpad, gst_event_ref (*event));
 501   }
 502
 503   return TRUE;
 504 }
 505
 506 static GstFlowReturn
 507 gst_rtp_mux_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
 508 {
 509   GstRTPMux *rtp_mux;
 510   GstFlowReturn ret;
 511   GstRTPMuxPadPrivate *padpriv;
 512   gboolean drop;
 513   gboolean changed = FALSE;
 514   GstRTPBuffer rtpbuffer = GST_RTP_BUFFER_INIT;
 515
 516   rtp_mux = GST_RTP_MUX (parent);
 517
 518   if (gst_pad_check_reconfigure (rtp_mux->srcpad)) {
 519     GstCaps *current_caps = gst_pad_get_current_caps (pad);
 520
 521     if (!gst_rtp_mux_setcaps (pad, rtp_mux, current_caps)) {
 522       ret = GST_FLOW_NOT_NEGOTIATED;
 523       gst_buffer_unref (buffer);
 524       goto out;
 525     }
 526     gst_caps_unref (current_caps);
 527   }
 528
 529   GST_OBJECT_LOCK (rtp_mux);
 530   padpriv = gst_pad_get_element_private (pad);
 531
 532   if (!padpriv) {
 533     GST_OBJECT_UNLOCK (rtp_mux);
 534     gst_buffer_unref (buffer);
 535     return GST_FLOW_NOT_LINKED;
 536   }
 537
 538   buffer = gst_buffer_make_writable (buffer);
 539
 540   if (!gst_rtp_buffer_map (buffer, GST_MAP_READWRITE, &rtpbuffer)) {
 541     GST_OBJECT_UNLOCK (rtp_mux);
 542     gst_buffer_unref (buffer);
 543     GST_ERROR_OBJECT (rtp_mux, "Invalid RTP buffer");
 544     return GST_FLOW_ERROR;
 545   }
 546
 547   drop = !process_buffer_locked (rtp_mux, padpriv, &rtpbuffer);
 548
 549   gst_rtp_buffer_unmap (&rtpbuffer);
 550
 551   if (!drop) {
 552     if (pad != rtp_mux->last_pad) {
 553       changed = TRUE;
 554       g_clear_object (&rtp_mux->last_pad);
 555       rtp_mux->last_pad = g_object_ref (pad);
 556     }
 557
 558     if (GST_BUFFER_DURATION_IS_VALID (buffer) &&
 559         GST_BUFFER_PTS_IS_VALID (buffer))
 560       rtp_mux->last_stop = GST_BUFFER_PTS (buffer) +
 561           GST_BUFFER_DURATION (buffer);
 562     else
 563       rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
 564   }
 565
 566   GST_OBJECT_UNLOCK (rtp_mux);
 567
 568   if (changed)
 569     gst_pad_sticky_events_foreach (pad, resend_events, rtp_mux);
 570
 571   if (drop) {
 572     gst_buffer_unref (buffer);
 573     ret = GST_FLOW_OK;
 574   } else {
 575     ret = gst_pad_push (rtp_mux->srcpad, buffer);
 576   }
 577
 578 out:
 579   return ret;
 580 }
 581
 582 static gboolean
 583 gst_rtp_mux_setcaps (GstPad * pad, GstRTPMux * rtp_mux, GstCaps * caps)
 584 {
 585   GstStructure *structure;
 586   gboolean ret = FALSE;
 587   GstRTPMuxPadPrivate *padpriv;
 588   GstCaps *peercaps;
 589
 590   if (!gst_caps_is_fixed (caps))
 591     return FALSE;
 592
 593   peercaps = gst_pad_peer_query_caps (rtp_mux->srcpad, NULL);
 594   if (peercaps) {
 595     GstCaps *tcaps, *othercaps;;
 596     tcaps = gst_pad_get_pad_template_caps (pad);
 597     othercaps = gst_caps_intersect_full (peercaps, tcaps,
 598         GST_CAPS_INTERSECT_FIRST);
 599
 600     if (gst_caps_get_size (othercaps) > 0) {
 601       structure = gst_caps_get_structure (othercaps, 0);
 602       GST_OBJECT_LOCK (rtp_mux);
 603       if (gst_structure_get_uint (structure, "ssrc", &rtp_mux->current_ssrc)) {
 604         GST_DEBUG_OBJECT (pad, "Use downstream ssrc: %x",
 605             rtp_mux->current_ssrc);
 606         rtp_mux->have_ssrc = TRUE;
 607       }
 608       GST_OBJECT_UNLOCK (rtp_mux);
 609     }
 610
 611     gst_caps_unref (othercaps);
 612
 613     gst_caps_unref (peercaps);
 614     gst_caps_unref (tcaps);
 615   }
 616
 617   structure = gst_caps_get_structure (caps, 0);
 618
 619   if (!structure)
 620     return FALSE;
 621
 622   GST_OBJECT_LOCK (rtp_mux);
 623   padpriv = gst_pad_get_element_private (pad);
 624   if (padpriv &&
 625       gst_structure_get_uint (structure, "timestamp-offset",
 626           &padpriv->timestamp_offset)) {
 627     padpriv->have_timestamp_offset = TRUE;
 628   }
 629
 630   caps = gst_caps_copy (caps);
 631
 632   /* if we don't have a specified ssrc, first try to take one from the caps,
 633      and if that fails, generate one */
 634   if (!rtp_mux->have_ssrc) {
 635     if (rtp_mux->ssrc_random) {
 636       if (!gst_structure_get_uint (structure, "ssrc", &rtp_mux->current_ssrc))
 637         rtp_mux->current_ssrc = g_random_int ();
 638       rtp_mux->have_ssrc = TRUE;
 639     }
 640   }
 641
 642   gst_caps_set_simple (caps,
 643       "timestamp-offset", G_TYPE_UINT, rtp_mux->ts_base,
 644       "seqnum-offset", G_TYPE_UINT, rtp_mux->seqnum_base,
 645       "ssrc", G_TYPE_UINT, rtp_mux->current_ssrc, NULL);
 646
 647   GST_OBJECT_UNLOCK (rtp_mux);
 648
 649   if (rtp_mux->send_stream_start) {
 650     gchar s_id[32];
 651
 652     /* stream-start (FIXME: create id based on input ids) */
 653     g_snprintf (s_id, sizeof (s_id), "interleave-%08x", g_random_int ());
 654     gst_pad_push_event (rtp_mux->srcpad, gst_event_new_stream_start (s_id));
 655
 656     rtp_mux->send_stream_start = FALSE;
 657   }
 658
 659   GST_DEBUG_OBJECT (rtp_mux,
 660       "setting caps %" GST_PTR_FORMAT " on src pad..", caps);
 661   ret = gst_pad_set_caps (rtp_mux->srcpad, caps);
 662
 663
 664   gst_caps_unref (caps);
 665
 666   return ret;
 667 }
 668
 669 static void
 670 clear_caps (GstCaps * caps, gboolean only_clock_rate)
 671 {
 672   gint i, j;
 673
 674   /* Lets only match on the clock-rate */
 675   for (i = 0; i < gst_caps_get_size (caps); i++) {
 676     GstStructure *s = gst_caps_get_structure (caps, i);
 677
 678     for (j = 0; j < gst_structure_n_fields (s); j++) {
 679       const gchar *name = gst_structure_nth_field_name (s, j);
 680
 681       if (strcmp (name, "clock-rate") && (only_clock_rate ||
 682               (strcmp (name, "ssrc")))) {
 683         gst_structure_remove_field (s, name);
 684         j--;
 685       }
 686     }
 687   }
 688 }
 689
 690 static gboolean
 691 same_clock_rate_fold (const GValue * item, GValue * ret, gpointer user_data)
 692 {
 693   GstPad *mypad = user_data;
 694   GstPad *pad = g_value_get_object (item);
 695   GstCaps *peercaps;
 696   GstCaps *accumcaps;
 697
 698   if (pad == mypad)
 699     return TRUE;
 700
 701   accumcaps = g_value_get_boxed (ret);
 702   peercaps = gst_pad_peer_query_caps (pad, accumcaps);
 703   if (!peercaps) {
 704     g_warning ("no peercaps");
 705     return TRUE;
 706   }
 707   peercaps = gst_caps_make_writable (peercaps);
 708   clear_caps (peercaps, TRUE);
 709
 710   g_value_take_boxed (ret, peercaps);
 711
 712   return !gst_caps_is_empty (peercaps);
 713 }
 714
 715 static GstCaps *
 716 gst_rtp_mux_getcaps (GstPad * pad, GstRTPMux * mux, GstCaps * filter)
 717 {
 718   GstCaps *caps = NULL;
 719   GstIterator *iter = NULL;
 720   GValue v = { 0 };
 721   GstIteratorResult res;
 722   GstCaps *peercaps;
 723   GstCaps *othercaps;
 724   GstCaps *tcaps;
 725
 726   peercaps = gst_pad_peer_query_caps (mux->srcpad, NULL);
 727
 728   if (peercaps) {
 729     tcaps = gst_pad_get_pad_template_caps (pad);
 730     othercaps = gst_caps_intersect_full (peercaps, tcaps,
 731         GST_CAPS_INTERSECT_FIRST);
 732     gst_caps_unref (peercaps);
 733   } else {
 734     tcaps = gst_pad_get_pad_template_caps (mux->srcpad);
 735     if (filter)
 736       othercaps = gst_caps_intersect_full (filter, tcaps,
 737           GST_CAPS_INTERSECT_FIRST);
 738     else
 739       othercaps = gst_caps_copy (tcaps);
 740   }
 741   gst_caps_unref (tcaps);
 742
 743   GST_LOG_OBJECT (pad, "Intersected srcpad-peercaps and template caps: %"
 744       GST_PTR_FORMAT, othercaps);
 745
 746   clear_caps (othercaps, TRUE);
 747
 748   g_value_init (&v, GST_TYPE_CAPS);
 749
 750   iter = gst_element_iterate_sink_pads (GST_ELEMENT (mux));
 751   do {
 752     gst_value_set_caps (&v, othercaps);
 753     res = gst_iterator_fold (iter, same_clock_rate_fold, &v, pad);
 754     gst_iterator_resync (iter);
 755   } while (res == GST_ITERATOR_RESYNC);
 756   gst_iterator_free (iter);
 757
 758   caps = gst_caps_intersect ((GstCaps *) gst_value_get_caps (&v), othercaps);
 759
 760   g_value_unset (&v);
 761   gst_caps_unref (othercaps);
 762
 763   if (res == GST_ITERATOR_ERROR) {
 764     gst_caps_unref (caps);
 765     caps = gst_caps_new_empty ();
 766   }
 767
 768
 769   return caps;
 770 }
 771
 772 static gboolean
 773 gst_rtp_mux_sink_query (GstPad * pad, GstObject * parent, GstQuery * query)
 774 {
 775   GstRTPMux *mux = GST_RTP_MUX (parent);
 776   gboolean res = FALSE;
 777
 778   switch (GST_QUERY_TYPE (query)) {
 779     case GST_QUERY_CAPS:
 780     {
 781       GstCaps *filter, *caps;
 782
 783       gst_query_parse_caps (query, &filter);
 784       GST_LOG_OBJECT (pad, "Received caps-query with filter-caps: %"
 785           GST_PTR_FORMAT, filter);
 786       caps = gst_rtp_mux_getcaps (pad, mux, filter);
 787       gst_query_set_caps_result (query, caps);
 788       GST_LOG_OBJECT (mux, "Answering caps-query with caps: %"
 789           GST_PTR_FORMAT, caps);
 790       gst_caps_unref (caps);
 791       res = TRUE;
 792       break;
 793     }
 794     default:
 795       res = gst_pad_query_default (pad, parent, query);
 796       break;
 797   }
 798
 799   return res;
 800 }
 801
 802 static void
 803 gst_rtp_mux_get_property (GObject * object,
 804     guint prop_id, GValue * value, GParamSpec * pspec)
 805 {
 806   GstRTPMux *rtp_mux;
 807
 808   rtp_mux = GST_RTP_MUX (object);
 809
 810   GST_OBJECT_LOCK (rtp_mux);
 811   switch (prop_id) {
 812     case PROP_TIMESTAMP_OFFSET:
 813       g_value_set_int (value, rtp_mux->ts_offset);
 814       break;
 815     case PROP_SEQNUM_OFFSET:
 816       g_value_set_int (value, rtp_mux->seqnum_offset);
 817       break;
 818     case PROP_SEQNUM:
 819       g_value_set_uint (value, rtp_mux->seqnum);
 820       break;
 821     case PROP_SSRC:
 822       g_value_set_uint (value, rtp_mux->ssrc);
 823       break;
 824     default:
 825       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 826       break;
 827   }
 828   GST_OBJECT_UNLOCK (rtp_mux);
 829 }
 830
 831 static void
 832 gst_rtp_mux_set_property (GObject * object,
 833     guint prop_id, const GValue * value, GParamSpec * pspec)
 834 {
 835   GstRTPMux *rtp_mux;
 836
 837   rtp_mux = GST_RTP_MUX (object);
 838
 839   switch (prop_id) {
 840     case PROP_TIMESTAMP_OFFSET:
 841       rtp_mux->ts_offset = g_value_get_int (value);
 842       break;
 843     case PROP_SEQNUM_OFFSET:
 844       rtp_mux->seqnum_offset = g_value_get_int (value);
 845       break;
 846     case PROP_SSRC:
 847       GST_OBJECT_LOCK (rtp_mux);
 848       rtp_mux->ssrc = g_value_get_uint (value);
 849       rtp_mux->current_ssrc = rtp_mux->ssrc;
 850       rtp_mux->have_ssrc = TRUE;
 851       rtp_mux->ssrc_random = FALSE;
 852       GST_OBJECT_UNLOCK (rtp_mux);
 853       break;
 854     default:
 855       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 856       break;
 857   }
 858 }
 859
 860 static gboolean
 861 gst_rtp_mux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
 862 {
 863   GstRTPMux *mux = GST_RTP_MUX (parent);
 864   gboolean is_pad;
 865   gboolean ret;
 866
 867   switch (GST_EVENT_TYPE (event)) {
 868     case GST_EVENT_CAPS:
 869     {
 870       GstCaps *caps;
 871
 872       gst_event_parse_caps (event, &caps);
 873       GST_LOG_OBJECT (pad, "Received caps-event with caps: %"
 874           GST_PTR_FORMAT, caps);
 875       ret = gst_rtp_mux_setcaps (pad, mux, caps);
 876       gst_event_unref (event);
 877       return ret;
 878     }
 879     case GST_EVENT_FLUSH_STOP:
 880     {
 881       GST_OBJECT_LOCK (mux);
 882       mux->last_stop = GST_CLOCK_TIME_NONE;
 883       GST_OBJECT_UNLOCK (mux);
 884       break;
 885     }
 886     case GST_EVENT_SEGMENT:
 887     {
 888       GstRTPMuxPadPrivate *padpriv;
 889
 890       GST_OBJECT_LOCK (mux);
 891       padpriv = gst_pad_get_element_private (pad);
 892
 893       if (padpriv) {
 894         gst_event_copy_segment (event, &padpriv->segment);
 895       }
 896       GST_OBJECT_UNLOCK (mux);
 897       break;
 898     }
 899     default:
 900       break;
 901   }
 902
 903   GST_OBJECT_LOCK (mux);
 904   is_pad = (pad == mux->last_pad);
 905   GST_OBJECT_UNLOCK (mux);
 906
 907   if (is_pad) {
 908     return gst_pad_push_event (mux->srcpad, event);
 909   } else {
 910     gst_event_unref (event);
 911     return TRUE;
 912   }
 913 }
 914
 915 static void
 916 gst_rtp_mux_ready_to_paused (GstRTPMux * rtp_mux)
 917 {
 918
 919   GST_OBJECT_LOCK (rtp_mux);
 920
 921   g_clear_object (&rtp_mux->last_pad);
 922   rtp_mux->send_stream_start = TRUE;
 923
 924   if (rtp_mux->seqnum_offset == -1)
 925     rtp_mux->seqnum_base = g_random_int_range (0, G_MAXUINT16);
 926   else
 927     rtp_mux->seqnum_base = rtp_mux->seqnum_offset;
 928   rtp_mux->seqnum = rtp_mux->seqnum_base;
 929
 930   if (rtp_mux->ts_offset == -1)
 931     rtp_mux->ts_base = g_random_int ();
 932   else
 933     rtp_mux->ts_base = rtp_mux->ts_offset;
 934
 935   rtp_mux->last_stop = GST_CLOCK_TIME_NONE;
 936
 937   if (rtp_mux->ssrc_random) {
 938     rtp_mux->have_ssrc = FALSE;
 939   } else {
 940     rtp_mux->current_ssrc = rtp_mux->ssrc;
 941     rtp_mux->have_ssrc = TRUE;
 942   }
 943
 944   GST_DEBUG_OBJECT (rtp_mux, "set timestamp-offset to %u", rtp_mux->ts_base);
 945
 946   GST_OBJECT_UNLOCK (rtp_mux);
 947 }
 948
 949 static GstStateChangeReturn
 950 gst_rtp_mux_change_state (GstElement * element, GstStateChange transition)
 951 {
 952   GstRTPMux *rtp_mux;
 953   GstStateChangeReturn ret;
 954
 955   rtp_mux = GST_RTP_MUX (element);
 956
 957   switch (transition) {
 958     case GST_STATE_CHANGE_READY_TO_PAUSED:
 959       gst_rtp_mux_ready_to_paused (rtp_mux);
 960       break;
 961     default:
 962       break;
 963   }
 964
 965   ret = GST_ELEMENT_CLASS (gst_rtp_mux_parent_class)->change_state (element,
 966       transition);
 967
 968   switch (transition) {
 969     case GST_STATE_CHANGE_PAUSED_TO_READY:
 970       g_clear_object (&rtp_mux->last_pad);
 971       break;
 972     default:
 973       break;
 974   }
 975
 976   return ret;
 977 }
 978
 979 gboolean
 980 gst_rtp_mux_plugin_init (GstPlugin * plugin)
 981 {
 982   GST_DEBUG_CATEGORY_INIT (gst_rtp_mux_debug, "rtpmux", 0, "rtp muxer");
 983
 984   return gst_element_register (plugin, "rtpmux", GST_RANK_NONE,
 985       GST_TYPE_RTP_MUX);
 986 }