rtpbin: receive bundle support

A new signal named on-bundled-ssrc is provided and can be
used by the application to redirect a stream to a different
GstRtpSession or to keep the RTX stream grouped within the
GstRtpSession of the same media type.

https://bugzilla.gnome.org/show_bug.cgi?id=772740

8 files changed, 852 insertions, 1 deletions

diff --git a/tests/check/Makefile.am b/tests/check/Makefile.am
index aa40c705e..9c855ae5c 100644
--- a/tests/check/Makefile.am
+++ b/tests/check/Makefile.am
@@ -233,6 +233,7 @@ check_rtpmanager = \
 	elements/rtpaux \
 	elements/rtpbin \
 	elements/rtpbin_buffer_list \
+	elements/rtpbundle \
 	elements/rtpcollision \
 	elements/rtpjitterbuffer \
 	elements/rtpmux \
@@ -576,6 +577,9 @@ elements_rtpcollision_LDADD = $(GST_PLUGINS_BASE_LIBS) $(GST_NET_LIBS) -lgstrtp-
 elements_rtpaux_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(CFLAGS) $(AM_CFLAGS)
 elements_rtpaux_LDADD = $(GST_PLUGINS_BASE_LIBS) -lgstrtp-$(GST_API_VERSION) $(LDADD)
 
+elements_rtpbundle_CFLAGS = $(GST_PLUGINS_BASE_CFLAGS) $(CFLAGS) $(AM_CFLAGS)
+elements_rtpbundle_LDADD = $(GST_PLUGINS_BASE_LIBS) -lgstrtp-$(GST_API_VERSION) $(LDADD)
+
 # FIXME: configure should check for gdk-pixbuf not gtk
 # only need video.h header, not the lib
 elements_gdkpixbufsink_CFLAGS = \
diff --git a/tests/check/elements/.gitignore b/tests/check/elements/.gitignore
index ac1624ba0..de196b799 100644
--- a/tests/check/elements/.gitignore
+++ b/tests/check/elements/.gitignore
@@ -54,6 +54,7 @@ rtp-payloading
 rtpaux
 rtpbin
 rtpbin_buffer_list
+rtpbundle
 rtpcollision
 rtph261
 rtph263
diff --git a/tests/check/elements/rtpbundle.c b/tests/check/elements/rtpbundle.c
new file mode 100644
index 000000000..9b477e1a0
--- /dev/null
+++ b/tests/check/elements/rtpbundle.c
@@ -0,0 +1,390 @@
+/* GStreamer
+ *
+ * Copyright (C) 2016 Igalia S.L.
+ *   @author Philippe Normand <philn@igalia.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include <gst/check/gstcheck.h>
+#include <gst/check/gstconsistencychecker.h>
+#include <gst/check/gsttestclock.h>
+#include <gst/rtp/gstrtpbuffer.h>
+
+static GMainLoop *main_loop;
+
+static void
+message_received (GstBus * bus, GstMessage * message, GstPipeline * bin)
+{
+  GST_INFO ("bus message from \"%" GST_PTR_FORMAT "\": %" GST_PTR_FORMAT,
+      GST_MESSAGE_SRC (message), message);
+
+  switch (message->type) {
+    case GST_MESSAGE_EOS:
+      g_main_loop_quit (main_loop);
+      break;
+    case GST_MESSAGE_WARNING:{
+      GError *gerror;
+      gchar *debug;
+
+      gst_message_parse_warning (message, &gerror, &debug);
+      gst_object_default_error (GST_MESSAGE_SRC (message), gerror, debug);
+      g_error_free (gerror);
+      g_free (debug);
+      break;
+    }
+    case GST_MESSAGE_ERROR:{
+      GError *gerror;
+      gchar *debug;
+
+      gst_message_parse_error (message, &gerror, &debug);
+      gst_object_default_error (GST_MESSAGE_SRC (message), gerror, debug);
+      g_error_free (gerror);
+      g_free (debug);
+      fail ("Error!");
+      break;
+    }
+    default:
+      break;
+  }
+}
+
+static void
+on_rtpbinreceive_pad_added (GstElement * element, GstPad * new_pad,
+    gpointer data)
+{
+  GstElement *pipeline = GST_ELEMENT (data);
+  gchar *pad_name = gst_pad_get_name (new_pad);
+
+  if (g_str_has_prefix (pad_name, "recv_rtp_src_")) {
+    GstCaps *caps = gst_pad_get_current_caps (new_pad);
+    GstStructure *s = gst_caps_get_structure (caps, 0);
+    const gchar *media_type = gst_structure_get_string (s, "media");
+    gchar *depayloader_name = g_strdup_printf ("%s_rtpdepayloader", media_type);
+    GstElement *rtpdepayloader =
+        gst_bin_get_by_name (GST_BIN (pipeline), depayloader_name);
+    GstPad *sinkpad;
+
+    g_free (depayloader_name);
+    fail_unless (rtpdepayloader != NULL, NULL);
+
+    sinkpad = gst_element_get_static_pad (rtpdepayloader, "sink");
+    gst_pad_link (new_pad, sinkpad);
+    gst_object_unref (sinkpad);
+    gst_object_unref (rtpdepayloader);
+
+    gst_caps_unref (caps);
+  }
+  g_free (pad_name);
+}
+
+static guint
+on_bundled_ssrc (GstElement * rtpbin, guint ssrc, gpointer user_data)
+{
+  static gboolean create_session = FALSE;
+  guint session_id = 0;
+
+  if (create_session) {
+    session_id = 1;
+  } else {
+    create_session = TRUE;
+    /* use existing session 0, a new session will be created for the next discovered bundled SSRC */
+  }
+  return session_id;
+}
+
+static GstCaps *
+on_request_pt_map (GstElement * rtpbin, guint session_id, guint pt,
+    gpointer user_data)
+{
+  GstCaps *caps = NULL;
+  if (pt == 96) {
+    caps =
+        gst_caps_from_string
+        ("application/x-rtp,media=(string)audio,encoding-name=(string)PCMA,clock-rate=(int)8000");
+  } else if (pt == 100) {
+    caps =
+        gst_caps_from_string
+        ("application/x-rtp,media=(string)video,encoding-name=(string)RAW,clock-rate=(int)90000,sampling=(string)\"YCbCr-4:2:0\",depth=(string)8,width=(string)320,height=(string)240");
+  }
+  return caps;
+}
+
+
+static GstElement *
+create_pipeline (gboolean send)
+{
+  GstElement *pipeline, *rtpbin, *audiosrc, *audio_encoder,
+      *audio_rtppayloader, *sendrtp_udpsink, *recv_rtp_udpsrc,
+      *send_rtcp_udpsink, *recv_rtcp_udpsrc, *sendrtcp_funnel, *sendrtp_funnel;
+  GstElement *audio_rtpdepayloader, *audio_decoder, *audio_sink;
+  GstElement *videosrc, *video_rtppayloader, *video_rtpdepayloader, *video_sink;
+  gboolean res;
+  GstPad *funnel_pad, *rtp_src_pad;
+  GstCaps *rtpcaps;
+  gint rtp_udp_port = 5001;
+  gint rtcp_udp_port = 5002;
+
+  pipeline = gst_pipeline_new (send ? "pipeline_send" : "pipeline_receive");
+
+  rtpbin =
+      gst_element_factory_make ("rtpbin",
+      send ? "rtpbin_send" : "rtpbin_receive");
+  g_object_set (rtpbin, "latency", 200, NULL);
+
+  if (!send) {
+    g_signal_connect (rtpbin, "on-bundled-ssrc",
+        G_CALLBACK (on_bundled_ssrc), NULL);
+    g_signal_connect (rtpbin, "request-pt-map",
+        G_CALLBACK (on_request_pt_map), NULL);
+  }
+
+  g_signal_connect (rtpbin, "pad-added",
+      G_CALLBACK (on_rtpbinreceive_pad_added), pipeline);
+
+  gst_bin_add (GST_BIN (pipeline), rtpbin);
+
+  if (send) {
+    audiosrc = gst_element_factory_make ("audiotestsrc", NULL);
+    audio_encoder = gst_element_factory_make ("alawenc", NULL);
+    audio_rtppayloader = gst_element_factory_make ("rtppcmapay", NULL);
+    g_object_set (audio_rtppayloader, "pt", 96, NULL);
+    g_object_set (audio_rtppayloader, "seqnum-offset", 1, NULL);
+
+    videosrc = gst_element_factory_make ("videotestsrc", NULL);
+    video_rtppayloader = gst_element_factory_make ("rtpvrawpay", NULL);
+    g_object_set (video_rtppayloader, "pt", 100, "seqnum-offset", 1, NULL);
+
+    g_object_set (audiosrc, "num-buffers", 5, NULL);
+    g_object_set (videosrc, "num-buffers", 5, NULL);
+
+    /* muxed rtcp */
+    sendrtcp_funnel = gst_element_factory_make ("funnel", "send_rtcp_funnel");
+    send_rtcp_udpsink = gst_element_factory_make ("udpsink", NULL);
+    g_object_set (send_rtcp_udpsink, "host", "127.0.0.1", NULL);
+    g_object_set (send_rtcp_udpsink, "port", rtcp_udp_port, NULL);
+    g_object_set (send_rtcp_udpsink, "sync", FALSE, NULL);
+    g_object_set (send_rtcp_udpsink, "async", FALSE, NULL);
+
+    /* outgoing bundled stream */
+    sendrtp_funnel = gst_element_factory_make ("funnel", "send_rtp_funnel");
+    sendrtp_udpsink = gst_element_factory_make ("udpsink", NULL);
+    g_object_set (sendrtp_udpsink, "host", "127.0.0.1", NULL);
+    g_object_set (sendrtp_udpsink, "port", rtp_udp_port, NULL);
+
+    gst_bin_add_many (GST_BIN (pipeline), audiosrc, audio_encoder,
+        audio_rtppayloader, sendrtp_udpsink, send_rtcp_udpsink,
+        sendrtp_funnel, sendrtcp_funnel, videosrc, video_rtppayloader, NULL);
+
+    res = gst_element_link (audiosrc, audio_encoder);
+    fail_unless (res == TRUE, NULL);
+    res = gst_element_link (audio_encoder, audio_rtppayloader);
+    fail_unless (res == TRUE, NULL);
+    res =
+        gst_element_link_pads_full (audio_rtppayloader, "src", rtpbin,
+        "send_rtp_sink_0", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+
+    res = gst_element_link (videosrc, video_rtppayloader);
+    fail_unless (res == TRUE, NULL);
+    res =
+        gst_element_link_pads_full (video_rtppayloader, "src", rtpbin,
+        "send_rtp_sink_1", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+
+    res =
+        gst_element_link_pads_full (sendrtp_funnel, "src", sendrtp_udpsink,
+        "sink", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+
+    funnel_pad = gst_element_get_request_pad (sendrtp_funnel, "sink_%u");
+    rtp_src_pad = gst_element_get_static_pad (rtpbin, "send_rtp_src_0");
+    res = gst_pad_link (rtp_src_pad, funnel_pad);
+    gst_object_unref (funnel_pad);
+    gst_object_unref (rtp_src_pad);
+
+    funnel_pad = gst_element_get_request_pad (sendrtp_funnel, "sink_%u");
+    rtp_src_pad = gst_element_get_static_pad (rtpbin, "send_rtp_src_1");
+    res = gst_pad_link (rtp_src_pad, funnel_pad);
+    gst_object_unref (funnel_pad);
+    gst_object_unref (rtp_src_pad);
+
+    res =
+        gst_element_link_pads_full (sendrtcp_funnel, "src", send_rtcp_udpsink,
+        "sink", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+
+    funnel_pad = gst_element_get_request_pad (sendrtcp_funnel, "sink_%u");
+    rtp_src_pad = gst_element_get_request_pad (rtpbin, "send_rtcp_src_0");
+    res =
+        gst_pad_link_full (rtp_src_pad, funnel_pad, GST_PAD_LINK_CHECK_NOTHING);
+    gst_object_unref (funnel_pad);
+    gst_object_unref (rtp_src_pad);
+
+    funnel_pad = gst_element_get_request_pad (sendrtcp_funnel, "sink_%u");
+    rtp_src_pad = gst_element_get_request_pad (rtpbin, "send_rtcp_src_1");
+    res =
+        gst_pad_link_full (rtp_src_pad, funnel_pad, GST_PAD_LINK_CHECK_NOTHING);
+    gst_object_unref (funnel_pad);
+    gst_object_unref (rtp_src_pad);
+
+  } else {
+    recv_rtp_udpsrc = gst_element_factory_make ("udpsrc", NULL);
+    g_object_set (recv_rtp_udpsrc, "port", rtp_udp_port, NULL);
+    rtpcaps = gst_caps_from_string ("application/x-rtp");
+    g_object_set (recv_rtp_udpsrc, "caps", rtpcaps, NULL);
+    gst_caps_unref (rtpcaps);
+
+    recv_rtcp_udpsrc = gst_element_factory_make ("udpsrc", NULL);
+    g_object_set (recv_rtcp_udpsrc, "port", rtcp_udp_port, NULL);
+
+    audio_rtpdepayloader =
+        gst_element_factory_make ("rtppcmadepay", "audio_rtpdepayloader");
+    audio_decoder = gst_element_factory_make ("alawdec", NULL);
+    audio_sink = gst_element_factory_make ("fakesink", NULL);
+    g_object_set (audio_sink, "sync", TRUE, NULL);
+
+    video_rtpdepayloader =
+        gst_element_factory_make ("rtpvrawdepay", "video_rtpdepayloader");
+    video_sink = gst_element_factory_make ("fakesink", NULL);
+    g_object_set (video_sink, "sync", TRUE, NULL);
+
+    gst_bin_add_many (GST_BIN (pipeline), recv_rtp_udpsrc, recv_rtcp_udpsrc,
+        audio_rtpdepayloader, audio_decoder, audio_sink, video_rtpdepayloader,
+        video_sink, NULL);
+
+    res =
+        gst_element_link_pads_full (audio_rtpdepayloader, "src", audio_decoder,
+        "sink", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+    res = gst_element_link (audio_decoder, audio_sink);
+    fail_unless (res == TRUE, NULL);
+
+    res =
+        gst_element_link_pads_full (video_rtpdepayloader, "src", video_sink,
+        "sink", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+
+    /* request a single receiving RTP session. */
+    res =
+        gst_element_link_pads_full (recv_rtcp_udpsrc, "src", rtpbin,
+        "recv_rtcp_sink_0", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+    res =
+        gst_element_link_pads_full (recv_rtp_udpsrc, "src", rtpbin,
+        "recv_rtp_sink_0", GST_PAD_LINK_CHECK_NOTHING);
+    fail_unless (res == TRUE, NULL);
+  }
+
+  return pipeline;
+}
+
+GST_START_TEST (test_simple_rtpbin_bundle)
+{
+  GstElement *send_pipeline, *recv_pipeline;
+  GstBus *send_bus, *recv_bus;
+  GstStateChangeReturn state_res = GST_STATE_CHANGE_FAILURE;
+  GstElement *rtpbin_receive;
+  GObject *rtp_session;
+
+  main_loop = g_main_loop_new (NULL, FALSE);
+
+  send_pipeline = create_pipeline (TRUE);
+  recv_pipeline = create_pipeline (FALSE);
+
+  send_bus = gst_element_get_bus (send_pipeline);
+  gst_bus_add_signal_watch_full (send_bus, G_PRIORITY_HIGH);
+
+  g_signal_connect (send_bus, "message::error", (GCallback) message_received,
+      send_pipeline);
+  g_signal_connect (send_bus, "message::warning", (GCallback) message_received,
+      send_pipeline);
+  g_signal_connect (send_bus, "message::eos", (GCallback) message_received,
+      send_pipeline);
+
+  recv_bus = gst_element_get_bus (recv_pipeline);
+  gst_bus_add_signal_watch_full (recv_bus, G_PRIORITY_HIGH);
+
+  g_signal_connect (recv_bus, "message::error", (GCallback) message_received,
+      recv_pipeline);
+  g_signal_connect (recv_bus, "message::warning", (GCallback) message_received,
+      recv_pipeline);
+  g_signal_connect (recv_bus, "message::eos", (GCallback) message_received,
+      recv_pipeline);
+
+  state_res = gst_element_set_state (recv_pipeline, GST_STATE_PLAYING);
+  ck_assert_int_ne (state_res, GST_STATE_CHANGE_FAILURE);
+
+  state_res = gst_element_set_state (send_pipeline, GST_STATE_PLAYING);
+  ck_assert_int_ne (state_res, GST_STATE_CHANGE_FAILURE);
+
+  GST_INFO ("enter mainloop");
+  g_main_loop_run (main_loop);
+  GST_INFO ("exit mainloop");
+
+  rtpbin_receive =
+      gst_bin_get_by_name (GST_BIN (recv_pipeline), "rtpbin_receive");
+  fail_if (rtpbin_receive == NULL, NULL);
+
+  /* Check that 2 RTP sessions where created while only one was explicitely requested. */
+  g_signal_emit_by_name (rtpbin_receive, "get-internal-session", 0,
+      &rtp_session);
+  fail_if (rtp_session == NULL, NULL);
+  g_object_unref (rtp_session);
+  g_signal_emit_by_name (rtpbin_receive, "get-internal-session", 1,
+      &rtp_session);
+  fail_if (rtp_session == NULL, NULL);
+  g_object_unref (rtp_session);
+
+  gst_object_unref (rtpbin_receive);
+
+  state_res = gst_element_set_state (send_pipeline, GST_STATE_NULL);
+  ck_assert_int_ne (state_res, GST_STATE_CHANGE_FAILURE);
+
+  state_res = gst_element_set_state (recv_pipeline, GST_STATE_NULL);
+  ck_assert_int_ne (state_res, GST_STATE_CHANGE_FAILURE);
+
+  /* cleanup */
+  g_main_loop_unref (main_loop);
+
+  gst_bus_remove_signal_watch (send_bus);
+  gst_object_unref (send_bus);
+  gst_object_unref (send_pipeline);
+
+  gst_bus_remove_signal_watch (recv_bus);
+  gst_object_unref (recv_bus);
+  gst_object_unref (recv_pipeline);
+
+}
+
+GST_END_TEST;
+
+static Suite *
+rtpbundle_suite (void)
+{
+  Suite *s = suite_create ("rtpbundle");
+  TCase *tc_chain = tcase_create ("general");
+
+  tcase_set_timeout (tc_chain, 10000);
+
+  suite_add_tcase (s, tc_chain);
+
+  tcase_add_test (tc_chain, test_simple_rtpbin_bundle);
+
+  return s;
+}
+
+GST_CHECK_MAIN (rtpbundle);
diff --git a/tests/check/meson.build b/tests/check/meson.build
index 4559e66c8..0f413777b 100644
--- a/tests/check/meson.build
+++ b/tests/check/meson.build
@@ -72,6 +72,7 @@ good_tests = [
   [ 'elements/rtpaux' ],
   [ 'elements/rtpbin' ],
   [ 'elements/rtpbin_buffer_list' ],
+  [ 'elements/rtpbundle' ],
   [ 'elements/rtpcollision' ],
   [ 'elements/rtpjitterbuffer' ],
   [ 'elements/rtpmux' ],
diff --git a/tests/examples/rtp/.gitignore b/tests/examples/rtp/.gitignore
index 02f551f12..8c3c6d3ff 100644
--- a/tests/examples/rtp/.gitignore
+++ b/tests/examples/rtp/.gitignore
@@ -2,3 +2,5 @@ client-PCMA
 server-alsasrc-PCMA
 client-rtpaux
 server-rtpaux
+client-rtpbundle
+server-rtpbundle
diff --git a/tests/examples/rtp/Makefile.am b/tests/examples/rtp/Makefile.am
index 40ab2c55e..4d7f7c1f8 100644
--- a/tests/examples/rtp/Makefile.am
+++ b/tests/examples/rtp/Makefile.am
@@ -1,5 +1,5 @@
 noinst_PROGRAMS = server-alsasrc-PCMA client-PCMA \
-	client-rtpaux server-rtpaux
+	client-rtpaux server-rtpaux client-rtpbundle server-rtpbundle
 
 # FIXME 0.11: ignore GValueArray warnings for now until this is sorted
 ERROR_CFLAGS=
@@ -12,6 +12,14 @@ client_rtpaux_SOURCES = client-rtpaux.c
 client_rtpaux_CFLAGS = $(GST_CFLAGS) $(GST_PLUGINS_BASE_CFLAGS)
 client_rtpaux_LDADD = $(GST_LIBS)
 
+server_rtpbundle_SOURCES = server-rtpbundle.c
+server_rtpbundle_CFLAGS = $(GST_CFLAGS) $(GST_PLUGINS_BASE_CFLAGS)
+server_rtpbundle_LDADD = $(GST_LIBS)
+
+client_rtpbundle_SOURCES = client-rtpbundle.c
+client_rtpbundle_CFLAGS = $(GST_CFLAGS) $(GST_PLUGINS_BASE_CFLAGS)
+client_rtpbundle_LDADD = $(GST_LIBS)
+
 server_alsasrc_PCMA_SOURCES = server-alsasrc-PCMA.c
 server_alsasrc_PCMA_CFLAGS = $(GST_CFLAGS)
 server_alsasrc_PCMA_LDADD = $(GST_LIBS) $(LIBM)
diff --git a/tests/examples/rtp/client-rtpbundle.c b/tests/examples/rtp/client-rtpbundle.c
new file mode 100644
index 000000000..cd3a737f8
--- /dev/null
+++ b/tests/examples/rtp/client-rtpbundle.c
@@ -0,0 +1,266 @@
+/* GStreamer
+ * Copyright (C) 2016 Igalia S.L
+ *   @author Philippe Normand <philn@igalia.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+
+#include <gst/gst.h>
+
+/*
+ * RTP bundle receiver
+ *
+ * In this example we initially create one RTP session but the incoming RTP
+ * and RTCP streams actually bundle 2 different media type, one audio stream
+ * and one video stream. We are notified of the discovery of the streams by
+ * the on-bundled-ssrc rtpbin signal. In the handler we decide to assign the
+ * first SSRC to the (existing) audio session and the second SSRC to a new
+ * session (id: 1).
+ *
+ *             .-------.      .----------.        .-----------.    .-------.    .-------------.
+ *  RTP        |udpsrc |      | rtpbin   |        | pcmadepay |    |alawdec|    |autoaudiosink|
+ *  port=5001  |      src->recv_rtp_0 recv_rtp_0->sink       src->sink    src->sink           |
+ *             '-------'      |          |        '-----------'    '-------'    '-------------'
+ *                            |          |
+ *                            |          |     .-------.
+ *                            |          |     |udpsink|  RTCP
+ *                            |  send_rtcp_0->sink     | port=5003
+ *             .-------.      |          |     '-------' sync=false
+ *  RTCP       |udpsrc |      |          |               async=false
+ *  port=5002  |     src->recv_rtcp_0    |
+ *             '-------'      |          |
+ *                            |          |
+ *                            |          |        .---------.    .-------------.
+ *                            |          |        |vrawdepay|    |autovideosink|
+ *                            |       recv_rtp_1->sink     src->sink           |
+ *                            |          |        '---------'    '-------------'
+ *                            |          |
+ *                            |          |     .-------.
+ *                            |          |     |udpsink|  RTCP
+ *                            |  send_rtcp_1->sink     | port=5004
+ *                            |          |     '-------' sync=false
+ *                            |          |               async=false
+ *                            |          |
+ *                            '----------'
+ *
+ */
+
+static gboolean
+plug_video_rtcp_sender (gpointer user_data)
+{
+  gint send_video_rtcp_port = 5004;
+  GstElement *rtpbin = GST_ELEMENT_CAST (user_data);
+  GstElement *send_video_rtcp_udpsink;
+  GstElement *pipeline =
+      GST_ELEMENT_CAST (gst_object_get_parent (GST_OBJECT (rtpbin)));
+
+  send_video_rtcp_udpsink = gst_element_factory_make ("udpsink", NULL);
+  g_object_set (send_video_rtcp_udpsink, "host", "127.0.0.1", NULL);
+  g_object_set (send_video_rtcp_udpsink, "port", send_video_rtcp_port, NULL);
+  g_object_set (send_video_rtcp_udpsink, "sync", FALSE, NULL);
+  g_object_set (send_video_rtcp_udpsink, "async", FALSE, NULL);
+  gst_bin_add (GST_BIN (pipeline), send_video_rtcp_udpsink);
+  gst_element_link_pads (rtpbin, "send_rtcp_src_1", send_video_rtcp_udpsink,
+      "sink");
+  gst_element_sync_state_with_parent (send_video_rtcp_udpsink);
+
+  gst_object_unref (pipeline);
+  gst_object_unref (rtpbin);
+  return G_SOURCE_REMOVE;
+}
+
+static void
+on_rtpbinreceive_pad_added (GstElement * rtpbin, GstPad * new_pad,
+    gpointer data)
+{
+  GstElement *pipeline = GST_ELEMENT (data);
+  gchar *pad_name = gst_pad_get_name (new_pad);
+
+  if (g_str_has_prefix (pad_name, "recv_rtp_src_")) {
+    GstCaps *caps = gst_pad_get_current_caps (new_pad);
+    GstStructure *s = gst_caps_get_structure (caps, 0);
+    const gchar *media_type = gst_structure_get_string (s, "media");
+    gchar *depayloader_name = g_strdup_printf ("%s_rtpdepayloader", media_type);
+    GstElement *rtpdepayloader =
+        gst_bin_get_by_name (GST_BIN (pipeline), depayloader_name);
+    GstPad *sinkpad;
+
+    g_free (depayloader_name);
+
+    sinkpad = gst_element_get_static_pad (rtpdepayloader, "sink");
+    gst_pad_link (new_pad, sinkpad);
+    gst_object_unref (sinkpad);
+    gst_object_unref (rtpdepayloader);
+
+    gst_caps_unref (caps);
+
+    if (g_str_has_prefix (pad_name, "recv_rtp_src_1")) {
+      g_timeout_add (0, plug_video_rtcp_sender, gst_object_ref (rtpbin));
+    }
+  }
+  g_free (pad_name);
+}
+
+static guint
+on_bundled_ssrc (GstElement * rtpbin, guint ssrc, gpointer user_data)
+{
+  static gboolean create_session = FALSE;
+  guint session_id = 0;
+
+  if (create_session) {
+    session_id = 1;
+  } else {
+    create_session = TRUE;
+    /* use existing session 0, a new session will be created for the next discovered bundled SSRC */
+  }
+  return session_id;
+}
+
+static GstCaps *
+on_request_pt_map (GstElement * rtpbin, guint session_id, guint pt,
+    gpointer user_data)
+{
+  GstCaps *caps = NULL;
+  if (pt == 96) {
+    caps =
+        gst_caps_from_string
+        ("application/x-rtp,media=(string)audio,encoding-name=(string)PCMA,clock-rate=(int)8000");
+  } else if (pt == 100) {
+    caps =
+        gst_caps_from_string
+        ("application/x-rtp,media=(string)video,encoding-name=(string)RAW,clock-rate=(int)90000,sampling=(string)\"YCbCr-4:2:0\",depth=(string)8,width=(string)320,height=(string)240");
+  }
+  return caps;
+}
+
+static GstElement *
+create_pipeline (void)
+{
+  GstElement *pipeline, *rtpbin, *recv_rtp_udpsrc, *recv_rtcp_udpsrc,
+      *audio_rtpdepayloader, *audio_decoder, *audio_sink, *video_rtpdepayloader,
+      *video_sink, *send_audio_rtcp_udpsink;
+  GstCaps *rtpcaps;
+  gint rtp_udp_port = 5001;
+  gint rtcp_udp_port = 5002;
+  gint send_audio_rtcp_port = 5003;
+
+  pipeline = gst_pipeline_new (NULL);
+
+  rtpbin = gst_element_factory_make ("rtpbin", NULL);
+  g_object_set (rtpbin, "latency", 200, NULL);
+
+  g_signal_connect (rtpbin, "on-bundled-ssrc",
+      G_CALLBACK (on_bundled_ssrc), NULL);
+  g_signal_connect (rtpbin, "request-pt-map",
+      G_CALLBACK (on_request_pt_map), NULL);
+
+  g_signal_connect (rtpbin, "pad-added",
+      G_CALLBACK (on_rtpbinreceive_pad_added), pipeline);
+
+  gst_bin_add (GST_BIN (pipeline), rtpbin);
+
+  recv_rtp_udpsrc = gst_element_factory_make ("udpsrc", NULL);
+  g_object_set (recv_rtp_udpsrc, "port", rtp_udp_port, NULL);
+  rtpcaps = gst_caps_from_string ("application/x-rtp");
+  g_object_set (recv_rtp_udpsrc, "caps", rtpcaps, NULL);
+  gst_caps_unref (rtpcaps);
+
+  recv_rtcp_udpsrc = gst_element_factory_make ("udpsrc", NULL);
+  g_object_set (recv_rtcp_udpsrc, "port", rtcp_udp_port, NULL);
+
+  audio_rtpdepayloader =
+      gst_element_factory_make ("rtppcmadepay", "audio_rtpdepayloader");
+  audio_decoder = gst_element_factory_make ("alawdec", NULL);
+  audio_sink = gst_element_factory_make ("autoaudiosink", NULL);
+
+  video_rtpdepayloader =
+      gst_element_factory_make ("rtpvrawdepay", "video_rtpdepayloader");
+  video_sink = gst_element_factory_make ("autovideosink", NULL);
+
+  gst_bin_add_many (GST_BIN (pipeline), recv_rtp_udpsrc, recv_rtcp_udpsrc,
+      audio_rtpdepayloader, audio_decoder, audio_sink, video_rtpdepayloader,
+      video_sink, NULL);
+
+  gst_element_link_pads (audio_rtpdepayloader, "src", audio_decoder, "sink");
+  gst_element_link (audio_decoder, audio_sink);
+
+  gst_element_link_pads (video_rtpdepayloader, "src", video_sink, "sink");
+
+  /* request a single receiving RTP session. */
+  gst_element_link_pads (recv_rtcp_udpsrc, "src", rtpbin, "recv_rtcp_sink_0");
+  gst_element_link_pads (recv_rtp_udpsrc, "src", rtpbin, "recv_rtp_sink_0");
+
+  send_audio_rtcp_udpsink = gst_element_factory_make ("udpsink", NULL);
+  g_object_set (send_audio_rtcp_udpsink, "host", "127.0.0.1", NULL);
+  g_object_set (send_audio_rtcp_udpsink, "port", send_audio_rtcp_port, NULL);
+  g_object_set (send_audio_rtcp_udpsink, "sync", FALSE, NULL);
+  g_object_set (send_audio_rtcp_udpsink, "async", FALSE, NULL);
+  gst_bin_add (GST_BIN (pipeline), send_audio_rtcp_udpsink);
+  gst_element_link_pads (rtpbin, "send_rtcp_src_0", send_audio_rtcp_udpsink,
+      "sink");
+
+  return pipeline;
+}
+
+/*
+ * Used to generate informative messages during pipeline startup
+ */
+static void
+cb_state (GstBus * bus, GstMessage * message, gpointer data)
+{
+  GstObject *pipe = GST_OBJECT (data);
+  GstState old, new, pending;
+  gst_message_parse_state_changed (message, &old, &new, &pending);
+  if (message->src == pipe) {
+    g_print ("Pipeline %s changed state from %s to %s\n",
+        GST_OBJECT_NAME (message->src),
+        gst_element_state_get_name (old), gst_element_state_get_name (new));
+    if (old == GST_STATE_PAUSED && new == GST_STATE_PLAYING)
+      GST_DEBUG_BIN_TO_DOT_FILE (GST_BIN (pipe), GST_DEBUG_GRAPH_SHOW_ALL,
+          GST_OBJECT_NAME (message->src));
+  }
+}
+
+int
+main (int argc, char **argv)
+{
+  GstElement *pipe;
+  GstBus *bus;
+  GMainLoop *loop;
+
+  gst_init (&argc, &argv);
+
+  loop = g_main_loop_new (NULL, FALSE);
+
+  pipe = create_pipeline ();
+  bus = gst_element_get_bus (pipe);
+  g_signal_connect (bus, "message::state-changed", G_CALLBACK (cb_state), pipe);
+  gst_bus_add_signal_watch (bus);
+  gst_object_unref (bus);
+
+  g_print ("starting server pipeline\n");
+  gst_element_set_state (pipe, GST_STATE_PLAYING);
+
+  g_main_loop_run (loop);
+
+  g_print ("stopping server pipeline\n");
+  gst_element_set_state (pipe, GST_STATE_NULL);
+
+  gst_object_unref (pipe);
+  g_main_loop_unref (loop);
+
+  return 0;
+}
diff --git a/tests/examples/rtp/server-rtpbundle.c b/tests/examples/rtp/server-rtpbundle.c
new file mode 100644
index 000000000..1f6d01bef
--- /dev/null
+++ b/tests/examples/rtp/server-rtpbundle.c
@@ -0,0 +1,179 @@
+/* GStreamer
+ * Copyright (C) 2016 Igalia S.L
+ *   @author Philippe Normand <philn@igalia.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
+ * Boston, MA 02110-1301, USA.
+ */
+#include <gst/gst.h>
+
+/*
+ * An bundling RTP server
+ *  creates two sessions and streams audio on one, video on the other, with RTCP
+ *  on both sessions. The destination is 127.0.0.1.
+ *
+ *  The RTP streams are bundled to a single outgoing connection. Same for the RTCP streams.
+ *
+ *  .-------.    .-------.    .-------.      .------------.         .------.
+ *  |audiots|    |alawenc|    |pcmapay|      | rtpbin     |         |funnel|
+ *  |      src->sink    src->sink    src->send_rtp_0 send_rtp_0--->sink_0  |    .-------.
+ *  '-------'    '-------'    '-------'      |            |         |      |    |udpsink|
+ *                                           |            |         |     src->sink     |
+ *  .-------.               .---------.      |            |         |      |    '-------'
+ *  |videots|               | vrawpay |      |            |         |      |
+ *  |      src------------>sink      src->send_rtp_1 send_rtp_1--->sink_1  |
+ *  '-------'               '---------'      |            |         '------'
+ *                                           |            |
+ *                               .------.    |            |
+ *                               |udpsrc|    |            |         .------.
+ *                               |     src->recv_rtcp_0   |         |funnel|
+ *                               '------'    |       send_rtcp_0-->sink_0  |   .-------.
+ *                                           |            |         |      |   |udpsink|
+ *                               .------.    |            |         |    src->sink     |
+ *                               |udpsrc|    |            |         |      |   '-------'
+ *                               |     src->recv_rtcp_1   |         |      |
+ *                               '------'    |       send_rtcp_1-->sink_1  |
+ *                                           '------------'         '------'
+ *
+ */
+
+static GstElement *
+create_pipeline (void)
+{
+  GstElement *pipeline, *rtpbin, *audiosrc, *audio_encoder,
+      *audio_rtppayloader, *sendrtp_udpsink,
+      *send_rtcp_udpsink, *sendrtcp_funnel, *sendrtp_funnel;
+  GstElement *videosrc, *video_rtppayloader, *time_overlay;
+  gint rtp_udp_port = 5001;
+  gint rtcp_udp_port = 5002;
+  gint recv_audio_rtcp_port = 5003;
+  gint recv_video_rtcp_port = 5004;
+  GstElement *audio_rtcp_udpsrc, *video_rtcp_udpsrc;
+
+  pipeline = gst_pipeline_new (NULL);
+
+  rtpbin = gst_element_factory_make ("rtpbin", NULL);
+
+  audiosrc = gst_element_factory_make ("audiotestsrc", NULL);
+  g_object_set (audiosrc, "is-live", TRUE, NULL);
+  audio_encoder = gst_element_factory_make ("alawenc", NULL);
+  audio_rtppayloader = gst_element_factory_make ("rtppcmapay", NULL);
+  g_object_set (audio_rtppayloader, "pt", 96, NULL);
+
+  videosrc = gst_element_factory_make ("videotestsrc", NULL);
+  g_object_set (videosrc, "is-live", TRUE, NULL);
+  time_overlay = gst_element_factory_make ("timeoverlay", NULL);
+  video_rtppayloader = gst_element_factory_make ("rtpvrawpay", NULL);
+  g_object_set (video_rtppayloader, "pt", 100, NULL);
+
+  /* muxed rtcp */
+  sendrtcp_funnel = gst_element_factory_make ("funnel", "send_rtcp_funnel");
+  send_rtcp_udpsink = gst_element_factory_make ("udpsink", NULL);
+  g_object_set (send_rtcp_udpsink, "host", "127.0.0.1", NULL);
+  g_object_set (send_rtcp_udpsink, "port", rtcp_udp_port, NULL);
+  g_object_set (send_rtcp_udpsink, "sync", FALSE, NULL);
+  g_object_set (send_rtcp_udpsink, "async", FALSE, NULL);
+
+  /* outgoing bundled stream */
+  sendrtp_funnel = gst_element_factory_make ("funnel", "send_rtp_funnel");
+  sendrtp_udpsink = gst_element_factory_make ("udpsink", NULL);
+  g_object_set (sendrtp_udpsink, "host", "127.0.0.1", NULL);
+  g_object_set (sendrtp_udpsink, "port", rtp_udp_port, NULL);
+  g_object_set (sendrtp_udpsink, "sync", FALSE, NULL);
+  g_object_set (sendrtp_udpsink, "async", FALSE, NULL);
+
+  gst_bin_add_many (GST_BIN (pipeline), rtpbin, audiosrc, audio_encoder,
+      audio_rtppayloader, sendrtp_udpsink, send_rtcp_udpsink,
+      sendrtp_funnel, sendrtcp_funnel, videosrc, video_rtppayloader, NULL);
+
+  if (time_overlay)
+    gst_bin_add (GST_BIN (pipeline), time_overlay);
+
+  gst_element_link_many (audiosrc, audio_encoder, audio_rtppayloader, NULL);
+  gst_element_link_pads (audio_rtppayloader, "src", rtpbin, "send_rtp_sink_0");
+
+  if (time_overlay) {
+    gst_element_link_many (videosrc, time_overlay, video_rtppayloader, NULL);
+  } else {
+    gst_element_link (videosrc, video_rtppayloader);
+  }
+
+  gst_element_link_pads (video_rtppayloader, "src", rtpbin, "send_rtp_sink_1");
+
+  gst_element_link_pads (sendrtp_funnel, "src", sendrtp_udpsink, "sink");
+  gst_element_link_pads (rtpbin, "send_rtp_src_0", sendrtp_funnel, "sink_%u");
+  gst_element_link_pads (rtpbin, "send_rtp_src_1", sendrtp_funnel, "sink_%u");
+  gst_element_link_pads (sendrtcp_funnel, "src", send_rtcp_udpsink, "sink");
+  gst_element_link_pads (rtpbin, "send_rtcp_src_0", sendrtcp_funnel, "sink_%u");
+  gst_element_link_pads (rtpbin, "send_rtcp_src_1", sendrtcp_funnel, "sink_%u");
+
+  audio_rtcp_udpsrc = gst_element_factory_make ("udpsrc", NULL);
+  g_object_set (audio_rtcp_udpsrc, "port", recv_audio_rtcp_port, NULL);
+  video_rtcp_udpsrc = gst_element_factory_make ("udpsrc", NULL);
+  g_object_set (video_rtcp_udpsrc, "port", recv_video_rtcp_port, NULL);
+  gst_bin_add_many (GST_BIN (pipeline), audio_rtcp_udpsrc, video_rtcp_udpsrc,
+      NULL);
+  gst_element_link_pads (audio_rtcp_udpsrc, "src", rtpbin, "recv_rtcp_sink_0");
+  gst_element_link_pads (video_rtcp_udpsrc, "src", rtpbin, "recv_rtcp_sink_1");
+
+  return pipeline;
+}
+
+/*
+ * Used to generate informative messages during pipeline startup
+ */
+static void
+cb_state (GstBus * bus, GstMessage * message, gpointer data)
+{
+  GstObject *pipe = GST_OBJECT (data);
+  GstState old, new, pending;
+  gst_message_parse_state_changed (message, &old, &new, &pending);
+  if (message->src == pipe) {
+    g_print ("Pipeline %s changed state from %s to %s\n",
+        GST_OBJECT_NAME (message->src),
+        gst_element_state_get_name (old), gst_element_state_get_name (new));
+  }
+}
+
+int
+main (int argc, char **argv)
+{
+  GstElement *pipe;
+  GstBus *bus;
+  GMainLoop *loop;
+
+  gst_init (&argc, &argv);
+
+  loop = g_main_loop_new (NULL, FALSE);
+
+  pipe = create_pipeline ();
+  bus = gst_element_get_bus (pipe);
+  g_signal_connect (bus, "message::state-changed", G_CALLBACK (cb_state), pipe);
+  gst_bus_add_signal_watch (bus);
+  gst_object_unref (bus);
+
+  g_print ("starting server pipeline\n");
+  gst_element_set_state (pipe, GST_STATE_PLAYING);
+
+  g_main_loop_run (loop);
+
+  g_print ("stopping server pipeline\n");
+  gst_element_set_state (pipe, GST_STATE_NULL);
+
+  gst_object_unref (pipe);
+  g_main_loop_unref (loop);
+
+  return 0;
+}