subprojects/gst-plugins-good/tests/examples/rtp/client-rtpaux.c

   1 /* GStreamer
   2  * Copyright (C) 2013 Collabora Ltd.
   3  *   @author Torrie Fischer <torrie.fischer@collabora.co.uk>
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  */
  20 #include <gst/gst.h>
  21 #include <gst/rtp/rtp.h>
  22 #include <stdlib.h>
  23
  24 /*
  25  * RTP receiver with RFC4588 retransmission handling enabled
  26  *
  27  *  In this example we have two RTP sessions, one for video and one for audio.
  28  *  Video is received on port 5000, with its RTCP stream received on port 5001
  29  *  and sent on port 5005. Audio is received on port 5005, with its RTCP stream
  30  *  received on port 5006 and sent on port 5011.
  31  *
  32  *  In both sessions, we set "rtprtxreceive" as the session's "aux" element
  33  *  in rtpbin, which enables RFC4588 retransmission handling for that session.
  34  *
  35  *             .-------.      .----------.        .-----------.   .---------.   .-------------.
  36  *  RTP        |udpsrc |      | rtpbin   |        |theoradepay|   |theoradec|   |autovideosink|
  37  *  port=5000  |      src->recv_rtp_0 recv_rtp_0->sink       src->sink     src->sink          |
  38  *             '-------'      |          |        '-----------'   '---------'   '-------------'
  39  *                            |          |
  40  *                            |          |     .-------.
  41  *                            |          |     |udpsink|  RTCP
  42  *                            |  send_rtcp_0->sink     | port=5005
  43  *             .-------.      |          |     '-------' sync=false
  44  *  RTCP       |udpsrc |      |          |               async=false
  45  *  port=5001  |     src->recv_rtcp_0    |
  46  *             '-------'      |          |
  47  *                            |          |
  48  *             .-------.      |          |        .---------.   .-------.   .-------------.
  49  *  RTP        |udpsrc |      |          |        |pcmadepay|   |alawdec|   |autoaudiosink|
  50  *  port=5006  |      src->recv_rtp_1 recv_rtp_1->sink     src->sink   src->sink          |
  51  *             '-------'      |          |        '---------'   '-------'   '-------------'
  52  *                            |          |
  53  *                            |          |     .-------.
  54  *                            |          |     |udpsink|  RTCP
  55  *                            |  send_rtcp_1->sink     | port=5011
  56  *             .-------.      |          |     '-------' sync=false
  57  *  RTCP       |udpsrc |      |          |               async=false
  58  *  port=5007  |     src->recv_rtcp_1    |
  59  *             '-------'      '----------'
  60  *
  61  */
  62
  63 GMainLoop *loop = NULL;
  64
  65 typedef struct _SessionData
  66 {
  67   int ref;
  68   GstElement *rtpbin;
  69   guint sessionNum;
  70   guint pt, rtxPt;
  71   GstCaps *caps;
  72   GstElement *output;
  73 } SessionData;
  74
  75 static SessionData *
  76 session_ref (SessionData * data)
  77 {
  78   g_atomic_int_inc (&data->ref);
  79   return data;
  80 }
  81
  82 static void
  83 session_unref (gpointer data)
  84 {
  85   SessionData *session = (SessionData *) data;
  86   if (g_atomic_int_dec_and_test (&session->ref)) {
  87     g_object_unref (session->rtpbin);
  88     gst_caps_unref (session->caps);
  89     g_free (session);
  90   }
  91 }
  92
  93 static SessionData *
  94 session_new (guint sessionNum)
  95 {
  96   SessionData *ret = g_new0 (SessionData, 1);
  97   ret->sessionNum = sessionNum;
  98   return session_ref (ret);
  99 }
 100
 101 static void
 102 setup_ghost_sink (GstElement * sink, GstBin * bin)
 103 {
 104   GstPad *sinkPad = gst_element_get_static_pad (sink, "sink");
 105   GstPad *binPad = gst_ghost_pad_new ("sink", sinkPad);
 106   gst_element_add_pad (GST_ELEMENT (bin), binPad);
 107 }
 108
 109 static SessionData *
 110 make_audio_session (guint sessionNum)
 111 {
 112   SessionData *ret = session_new (sessionNum);
 113   GstBin *bin = GST_BIN (gst_bin_new ("audio"));
 114   GstElement *queue = gst_element_factory_make ("queue", NULL);
 115   GstElement *sink = gst_element_factory_make ("autoaudiosink", NULL);
 116   GstElement *audioconvert = gst_element_factory_make ("audioconvert", NULL);
 117   GstElement *audioresample = gst_element_factory_make ("audioresample", NULL);
 118   GstElement *depayloader = gst_element_factory_make ("rtppcmadepay", NULL);
 119   GstElement *decoder = gst_element_factory_make ("alawdec", NULL);
 120
 121   gst_bin_add_many (bin, queue, depayloader, decoder, audioconvert,
 122       audioresample, sink, NULL);
 123   gst_element_link_many (queue, depayloader, decoder, audioconvert,
 124       audioresample, sink, NULL);
 125
 126   setup_ghost_sink (queue, bin);
 127
 128   ret->output = GST_ELEMENT (bin);
 129   ret->caps = gst_caps_new_simple ("application/x-rtp",
 130       "media", G_TYPE_STRING, "audio",
 131       "clock-rate", G_TYPE_INT, 8000,
 132       "encoding-name", G_TYPE_STRING, "PCMA", NULL);
 133   ret->pt = 8;
 134   ret->rtxPt = 98;
 135
 136   return ret;
 137 }
 138
 139 static SessionData *
 140 make_video_session (guint sessionNum)
 141 {
 142   SessionData *ret = session_new (sessionNum);
 143   GstBin *bin = GST_BIN (gst_bin_new ("video"));
 144   GstElement *queue = gst_element_factory_make ("queue", NULL);
 145   GstElement *depayloader = gst_element_factory_make ("rtptheoradepay", NULL);
 146   GstElement *decoder = gst_element_factory_make ("theoradec", NULL);
 147   GstElement *converter = gst_element_factory_make ("videoconvert", NULL);
 148   GstElement *sink = gst_element_factory_make ("autovideosink", NULL);
 149
 150   gst_bin_add_many (bin, depayloader, decoder, converter, queue, sink, NULL);
 151   gst_element_link_many (queue, depayloader, decoder, converter, sink, NULL);
 152
 153   setup_ghost_sink (queue, bin);
 154
 155   ret->output = GST_ELEMENT (bin);
 156   ret->caps = gst_caps_new_simple ("application/x-rtp",
 157       "media", G_TYPE_STRING, "video",
 158       "clock-rate", G_TYPE_INT, 90000,
 159       "encoding-name", G_TYPE_STRING, "THEORA", NULL);
 160   ret->pt = 96;
 161   ret->rtxPt = 99;
 162
 163   return ret;
 164 }
 165
 166 static GstCaps *
 167 request_pt_map (GstElement * rtpbin, guint session, guint pt,
 168     gpointer user_data)
 169 {
 170   SessionData *data = (SessionData *) user_data;
 171   gchar *caps_str;
 172   g_print ("Looking for caps for pt %u in session %u, have %u\n", pt, session,
 173       data->sessionNum);
 174   if (session == data->sessionNum) {
 175     GstCaps *caps;
 176
 177     if (pt == data->pt) {
 178       caps = gst_caps_ref (data->caps);
 179     } else if (pt == data->rtxPt) {
 180       caps = gst_caps_copy (data->caps);
 181       gst_caps_set_simple (caps, "encoding-name", G_TYPE_STRING, "rtx", NULL);
 182     } else {
 183       return NULL;
 184     }
 185
 186     caps_str = gst_caps_to_string (caps);
 187     g_print ("Returning %s\n", caps_str);
 188     g_free (caps_str);
 189
 190     return caps;
 191   }
 192   return NULL;
 193 }
 194
 195 static void
 196 cb_eos (GstBus * bus, GstMessage * message, gpointer data)
 197 {
 198   g_print ("Got EOS\n");
 199   g_main_loop_quit (loop);
 200 }
 201
 202 static void
 203 cb_state (GstBus * bus, GstMessage * message, gpointer data)
 204 {
 205   GstObject *pipe = GST_OBJECT (data);
 206   GstState old, new, pending;
 207   gst_message_parse_state_changed (message, &old, &new, &pending);
 208   if (message->src == pipe) {
 209     g_print ("Pipeline %s changed state from %s to %s\n",
 210         GST_OBJECT_NAME (message->src),
 211         gst_element_state_get_name (old), gst_element_state_get_name (new));
 212   }
 213 }
 214
 215 static void
 216 cb_warning (GstBus * bus, GstMessage * message, gpointer data)
 217 {
 218   GError *error = NULL;
 219   gst_message_parse_warning (message, &error, NULL);
 220   g_printerr ("Got warning from %s: %s\n", GST_OBJECT_NAME (message->src),
 221       error->message);
 222   g_error_free (error);
 223 }
 224
 225 static void
 226 cb_error (GstBus * bus, GstMessage * message, gpointer data)
 227 {
 228   GError *error = NULL;
 229   gst_message_parse_error (message, &error, NULL);
 230   g_printerr ("Got error from %s: %s\n", GST_OBJECT_NAME (message->src),
 231       error->message);
 232   g_error_free (error);
 233   g_main_loop_quit (loop);
 234 }
 235
 236 static void
 237 handle_new_stream (GstElement * element, GstPad * newPad, gpointer data)
 238 {
 239   SessionData *session = (SessionData *) data;
 240   gchar *padName;
 241   gchar *myPrefix;
 242
 243   padName = gst_pad_get_name (newPad);
 244   myPrefix = g_strdup_printf ("recv_rtp_src_%u", session->sessionNum);
 245
 246   g_print ("New pad: %s, looking for %s_*\n", padName, myPrefix);
 247
 248   if (g_str_has_prefix (padName, myPrefix)) {
 249     GstPad *outputSinkPad;
 250     GstElement *parent;
 251
 252     parent = GST_ELEMENT (gst_element_get_parent (session->rtpbin));
 253     gst_bin_add (GST_BIN (parent), session->output);
 254     gst_element_sync_state_with_parent (session->output);
 255     gst_object_unref (parent);
 256
 257     outputSinkPad = gst_element_get_static_pad (session->output, "sink");
 258     g_assert_cmpint (gst_pad_link (newPad, outputSinkPad), ==, GST_PAD_LINK_OK);
 259     gst_object_unref (outputSinkPad);
 260
 261     g_print ("Linked!\n");
 262   }
 263   g_free (myPrefix);
 264   g_free (padName);
 265 }
 266
 267 static GstElement *
 268 request_aux_receiver (GstElement * rtpbin, guint sessid, SessionData * session)
 269 {
 270   GstElement *rtx, *bin;
 271   GstPad *pad;
 272   gchar *name;
 273   GstStructure *pt_map;
 274   gchar *media_pt;
 275
 276   if (sessid != session->sessionNum)
 277     return NULL;
 278
 279   GST_INFO ("creating AUX receiver for session %u", sessid);
 280   bin = gst_bin_new (NULL);
 281   rtx = gst_element_factory_make ("rtprtxreceive", NULL);
 282
 283   media_pt = g_strdup_printf ("%d", session->pt);
 284   pt_map = gst_structure_new ("application/x-rtp-pt-map",
 285       media_pt, G_TYPE_UINT, session->rtxPt, NULL);
 286   g_object_set (rtx, "payload-type-map", pt_map, NULL);
 287   gst_structure_free (pt_map);
 288   g_free (media_pt);
 289
 290   gst_bin_add (GST_BIN (bin), rtx);
 291
 292   pad = gst_element_get_static_pad (rtx, "src");
 293   name = g_strdup_printf ("src_%u", sessid);
 294   gst_element_add_pad (bin, gst_ghost_pad_new (name, pad));
 295   g_free (name);
 296   gst_object_unref (pad);
 297
 298   pad = gst_element_get_static_pad (rtx, "sink");
 299   name = g_strdup_printf ("sink_%u", sessid);
 300   gst_element_add_pad (bin, gst_ghost_pad_new (name, pad));
 301   g_free (name);
 302   gst_object_unref (pad);
 303
 304   return bin;
 305 }
 306
 307 static void
 308 join_session (GstElement * pipeline, GstElement * rtpBin, SessionData * session)
 309 {
 310   GstElement *rtpSrc;
 311   GstElement *rtcpSrc;
 312   GstElement *rtcpSink;
 313   gchar *padName;
 314   guint basePort;
 315
 316   g_print ("Joining session %p\n", session);
 317
 318   session->rtpbin = g_object_ref (rtpBin);
 319
 320   basePort = 5000 + (session->sessionNum * 6);
 321
 322   rtpSrc = gst_element_factory_make ("udpsrc", NULL);
 323   rtcpSrc = gst_element_factory_make ("udpsrc", NULL);
 324   rtcpSink = gst_element_factory_make ("udpsink", NULL);
 325   g_object_set (rtpSrc, "port", basePort, "caps", session->caps, NULL);
 326   g_object_set (rtcpSink, "port", basePort + 5, "host", "127.0.0.1", "sync",
 327       FALSE, "async", FALSE, NULL);
 328   g_object_set (rtcpSrc, "port", basePort + 1, NULL);
 329
 330   g_print ("Connecting to %i/%i/%i\n", basePort, basePort + 1, basePort + 5);
 331
 332   /* enable RFC4588 retransmission handling by setting rtprtxreceive
 333    * as the "aux" element of rtpbin */
 334   g_signal_connect (rtpBin, "request-aux-receiver",
 335       (GCallback) request_aux_receiver, session);
 336
 337   gst_bin_add_many (GST_BIN (pipeline), rtpSrc, rtcpSrc, rtcpSink, NULL);
 338
 339   g_signal_connect_data (rtpBin, "pad-added", G_CALLBACK (handle_new_stream),
 340       session_ref (session), (GClosureNotify) session_unref, 0);
 341
 342   g_signal_connect_data (rtpBin, "request-pt-map", G_CALLBACK (request_pt_map),
 343       session_ref (session), (GClosureNotify) session_unref, 0);
 344
 345   padName = g_strdup_printf ("recv_rtp_sink_%u", session->sessionNum);
 346   gst_element_link_pads (rtpSrc, "src", rtpBin, padName);
 347   g_free (padName);
 348
 349   padName = g_strdup_printf ("recv_rtcp_sink_%u", session->sessionNum);
 350   gst_element_link_pads (rtcpSrc, "src", rtpBin, padName);
 351   g_free (padName);
 352
 353   padName = g_strdup_printf ("send_rtcp_src_%u", session->sessionNum);
 354   gst_element_link_pads (rtpBin, padName, rtcpSink, "sink");
 355   g_free (padName);
 356
 357   session_unref (session);
 358 }
 359
 360 int
 361 main (int argc, char **argv)
 362 {
 363   GstPipeline *pipe;
 364   SessionData *videoSession;
 365   SessionData *audioSession;
 366   GstElement *rtpBin;
 367   GstBus *bus;
 368
 369   gst_init (&argc, &argv);
 370
 371   loop = g_main_loop_new (NULL, FALSE);
 372   pipe = GST_PIPELINE (gst_pipeline_new (NULL));
 373
 374   bus = gst_element_get_bus (GST_ELEMENT (pipe));
 375   g_signal_connect (bus, "message::error", G_CALLBACK (cb_error), pipe);
 376   g_signal_connect (bus, "message::warning", G_CALLBACK (cb_warning), pipe);
 377   g_signal_connect (bus, "message::state-changed", G_CALLBACK (cb_state), pipe);
 378   g_signal_connect (bus, "message::eos", G_CALLBACK (cb_eos), NULL);
 379   gst_bus_add_signal_watch (bus);
 380   gst_object_unref (bus);
 381
 382   rtpBin = gst_element_factory_make ("rtpbin", NULL);
 383   gst_bin_add (GST_BIN (pipe), rtpBin);
 384   g_object_set (rtpBin, "latency", 200, "do-retransmission", TRUE,
 385       "rtp-profile", GST_RTP_PROFILE_AVPF, NULL);
 386
 387   videoSession = make_video_session (0);
 388   audioSession = make_audio_session (1);
 389
 390   join_session (GST_ELEMENT (pipe), rtpBin, videoSession);
 391   join_session (GST_ELEMENT (pipe), rtpBin, audioSession);
 392
 393   g_print ("starting client pipeline\n");
 394   gst_element_set_state (GST_ELEMENT (pipe), GST_STATE_PLAYING);
 395
 396   g_main_loop_run (loop);
 397
 398   g_print ("stopping client pipeline\n");
 399   gst_element_set_state (GST_ELEMENT (pipe), GST_STATE_NULL);
 400
 401   gst_object_unref (pipe);
 402   g_main_loop_unref (loop);
 403
 404   return 0;
 405 }