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