gst/sdp/gstsdpdemux.c

   1 /* GStreamer
   2  * Copyright (C) <2007> Wim Taymans <wim dot taymans at gmail dot com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19 /**
  20  * SECTION:element-sdpdemux
  21  *
  22  * sdpdemux currently understands SDP as the input format of the session description.
  23  * For each stream listed in the SDP a new stream_%u pad will be created
  24  * with caps derived from the SDP media description. This is a caps of mime type
  25  * "application/x-rtp" that can be connected to any available RTP depayloader
  26  * element.
  27  *
  28  * sdpdemux will internally instantiate an RTP session manager element
  29  * that will handle the RTCP messages to and from the server, jitter removal,
  30  * packet reordering along with providing a clock for the pipeline.
  31  *
  32  * sdpdemux acts like a live element and will therefore only generate data in the
  33  * PLAYING state.
  34  *
  35  * <refsect2>
  36  * <title>Example launch line</title>
  37  * |[
  38  * gst-launch gnomevfssrc location=http://some.server/session.sdp ! sdpdemux ! fakesink
  39  * ]| Establish a connection to an HTTP server that contains an SDP session description
  40  * that gets parsed by sdpdemux and send the raw RTP packets to a fakesink.
  41  * </refsect2>
  42  */
  43
  44 #ifdef HAVE_CONFIG_H
  45 #include "config.h"
  46 #endif
  47
  48 #include "gstsdpdemux.h"
  49
  50 #include <gst/rtp/gstrtppayloads.h>
  51 #include <gst/sdp/gstsdpmessage.h>
  52
  53 #include <stdio.h>
  54 #include <stdlib.h>
  55 #include <string.h>
  56
  57 GST_DEBUG_CATEGORY_STATIC (sdpdemux_debug);
  58 #define GST_CAT_DEFAULT (sdpdemux_debug)
  59
  60 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  61     GST_PAD_SINK,
  62     GST_PAD_ALWAYS,
  63     GST_STATIC_CAPS ("application/sdp"));
  64
  65 static GstStaticPadTemplate rtptemplate = GST_STATIC_PAD_TEMPLATE ("stream_%u",
  66     GST_PAD_SRC,
  67     GST_PAD_SOMETIMES,
  68     GST_STATIC_CAPS ("application/x-rtp"));
  69
  70 enum
  71 {
  72   /* FILL ME */
  73   LAST_SIGNAL
  74 };
  75
  76 #define DEFAULT_DEBUG            FALSE
  77 #define DEFAULT_TIMEOUT          10000000
  78 #define DEFAULT_LATENCY_MS       200
  79 #define DEFAULT_REDIRECT         TRUE
  80
  81 enum
  82 {
  83   PROP_0,
  84   PROP_DEBUG,
  85   PROP_TIMEOUT,
  86   PROP_LATENCY,
  87   PROP_REDIRECT,
  88   PROP_LAST
  89 };
  90
  91 static void gst_sdp_demux_finalize (GObject * object);
  92
  93 static void gst_sdp_demux_set_property (GObject * object, guint prop_id,
  94     const GValue * value, GParamSpec * pspec);
  95 static void gst_sdp_demux_get_property (GObject * object, guint prop_id,
  96     GValue * value, GParamSpec * pspec);
  97
  98 static GstCaps *gst_sdp_demux_media_to_caps (gint pt,
  99     const GstSDPMedia * media);
 100
 101 static GstStateChangeReturn gst_sdp_demux_change_state (GstElement * element,
 102     GstStateChange transition);
 103 static void gst_sdp_demux_handle_message (GstBin * bin, GstMessage * message);
 104
 105 static void gst_sdp_demux_stream_push_event (GstSDPDemux * demux,
 106     GstSDPStream * stream, GstEvent * event);
 107
 108 static gboolean gst_sdp_demux_sink_event (GstPad * pad, GstObject * parent,
 109     GstEvent * event);
 110 static GstFlowReturn gst_sdp_demux_sink_chain (GstPad * pad, GstObject * parent,
 111     GstBuffer * buffer);
 112
 113 /*static guint gst_sdp_demux_signals[LAST_SIGNAL] = { 0 }; */
 114
 115 #define gst_sdp_demux_parent_class parent_class
 116 G_DEFINE_TYPE (GstSDPDemux, gst_sdp_demux, GST_TYPE_BIN);
 117
 118 static void
 119 gst_sdp_demux_class_init (GstSDPDemuxClass * klass)
 120 {
 121   GObjectClass *gobject_class;
 122   GstElementClass *gstelement_class;
 123   GstBinClass *gstbin_class;
 124
 125   gobject_class = (GObjectClass *) klass;
 126   gstelement_class = (GstElementClass *) klass;
 127   gstbin_class = (GstBinClass *) klass;
 128
 129   gobject_class->set_property = gst_sdp_demux_set_property;
 130   gobject_class->get_property = gst_sdp_demux_get_property;
 131
 132   gobject_class->finalize = gst_sdp_demux_finalize;
 133
 134   g_object_class_install_property (gobject_class, PROP_DEBUG,
 135       g_param_spec_boolean ("debug", "Debug",
 136           "Dump request and response messages to stdout",
 137           DEFAULT_DEBUG,
 138           G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
 139
 140   g_object_class_install_property (gobject_class, PROP_TIMEOUT,
 141       g_param_spec_uint64 ("timeout", "Timeout",
 142           "Fail transport after UDP timeout microseconds (0 = disabled)",
 143           0, G_MAXUINT64, DEFAULT_TIMEOUT,
 144           G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
 145
 146   g_object_class_install_property (gobject_class, PROP_LATENCY,
 147       g_param_spec_uint ("latency", "Buffer latency in ms",
 148           "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS,
 149           G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
 150
 151   g_object_class_install_property (gobject_class, PROP_REDIRECT,
 152       g_param_spec_boolean ("redirect", "Redirect",
 153           "Sends a redirection message instead of using a custom session element",
 154           DEFAULT_REDIRECT,
 155           G_PARAM_READWRITE | G_PARAM_CONSTRUCT | G_PARAM_STATIC_STRINGS));
 156
 157   gst_element_class_add_pad_template (gstelement_class,
 158       gst_static_pad_template_get (&sinktemplate));
 159   gst_element_class_add_pad_template (gstelement_class,
 160       gst_static_pad_template_get (&rtptemplate));
 161
 162   gst_element_class_set_static_metadata (gstelement_class, "SDP session setup",
 163       "Codec/Demuxer/Network/RTP",
 164       "Receive data over the network via SDP",
 165       "Wim Taymans <wim.taymans@gmail.com>");
 166
 167   gstelement_class->change_state = gst_sdp_demux_change_state;
 168
 169   gstbin_class->handle_message = gst_sdp_demux_handle_message;
 170
 171   GST_DEBUG_CATEGORY_INIT (sdpdemux_debug, "sdpdemux", 0, "SDP demux");
 172 }
 173
 174 static void
 175 gst_sdp_demux_init (GstSDPDemux * demux)
 176 {
 177   demux->sinkpad = gst_pad_new_from_static_template (&sinktemplate, "sink");
 178   gst_pad_set_event_function (demux->sinkpad,
 179       GST_DEBUG_FUNCPTR (gst_sdp_demux_sink_event));
 180   gst_pad_set_chain_function (demux->sinkpad,
 181       GST_DEBUG_FUNCPTR (gst_sdp_demux_sink_chain));
 182   gst_element_add_pad (GST_ELEMENT (demux), demux->sinkpad);
 183
 184   /* protects the streaming thread in interleaved mode or the polling
 185    * thread in UDP mode. */
 186   g_rec_mutex_init (&demux->stream_rec_lock);
 187
 188   demux->adapter = gst_adapter_new ();
 189 }
 190
 191 static void
 192 gst_sdp_demux_finalize (GObject * object)
 193 {
 194   GstSDPDemux *demux;
 195
 196   demux = GST_SDP_DEMUX (object);
 197
 198   /* free locks */
 199   g_rec_mutex_clear (&demux->stream_rec_lock);
 200
 201   g_object_unref (demux->adapter);
 202
 203   G_OBJECT_CLASS (parent_class)->finalize (object);
 204 }
 205
 206 static void
 207 gst_sdp_demux_set_property (GObject * object, guint prop_id,
 208     const GValue * value, GParamSpec * pspec)
 209 {
 210   GstSDPDemux *demux;
 211
 212   demux = GST_SDP_DEMUX (object);
 213
 214   switch (prop_id) {
 215     case PROP_DEBUG:
 216       demux->debug = g_value_get_boolean (value);
 217       break;
 218     case PROP_TIMEOUT:
 219       demux->udp_timeout = g_value_get_uint64 (value);
 220       break;
 221     case PROP_LATENCY:
 222       demux->latency = g_value_get_uint (value);
 223       break;
 224     case PROP_REDIRECT:
 225       demux->redirect = g_value_get_boolean (value);
 226       break;
 227     default:
 228       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 229       break;
 230   }
 231 }
 232
 233 static void
 234 gst_sdp_demux_get_property (GObject * object, guint prop_id, GValue * value,
 235     GParamSpec * pspec)
 236 {
 237   GstSDPDemux *demux;
 238
 239   demux = GST_SDP_DEMUX (object);
 240
 241   switch (prop_id) {
 242     case PROP_DEBUG:
 243       g_value_set_boolean (value, demux->debug);
 244       break;
 245     case PROP_TIMEOUT:
 246       g_value_set_uint64 (value, demux->udp_timeout);
 247       break;
 248     case PROP_LATENCY:
 249       g_value_set_uint (value, demux->latency);
 250       break;
 251     case PROP_REDIRECT:
 252       g_value_set_boolean (value, demux->redirect);
 253       break;
 254     default:
 255       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 256       break;
 257   }
 258 }
 259
 260 static gint
 261 find_stream_by_id (GstSDPStream * stream, gconstpointer a)
 262 {
 263   gint id = GPOINTER_TO_INT (a);
 264
 265   if (stream->id == id)
 266     return 0;
 267
 268   return -1;
 269 }
 270
 271 static gint
 272 find_stream_by_pt (GstSDPStream * stream, gconstpointer a)
 273 {
 274   gint pt = GPOINTER_TO_INT (a);
 275
 276   if (stream->pt == pt)
 277     return 0;
 278
 279   return -1;
 280 }
 281
 282 static gint
 283 find_stream_by_udpsrc (GstSDPStream * stream, gconstpointer a)
 284 {
 285   GstElement *src = (GstElement *) a;
 286
 287   if (stream->udpsrc[0] == src)
 288     return 0;
 289   if (stream->udpsrc[1] == src)
 290     return 0;
 291
 292   return -1;
 293 }
 294
 295 static GstSDPStream *
 296 find_stream (GstSDPDemux * demux, gconstpointer data, gconstpointer func)
 297 {
 298   GList *lstream;
 299
 300   /* find and get stream */
 301   if ((lstream =
 302           g_list_find_custom (demux->streams, data, (GCompareFunc) func)))
 303     return (GstSDPStream *) lstream->data;
 304
 305   return NULL;
 306 }
 307
 308 static void
 309 gst_sdp_demux_stream_free (GstSDPDemux * demux, GstSDPStream * stream)
 310 {
 311   gint i;
 312
 313   GST_DEBUG_OBJECT (demux, "free stream %p", stream);
 314
 315   if (stream->caps)
 316     gst_caps_unref (stream->caps);
 317
 318   for (i = 0; i < 2; i++) {
 319     GstElement *udpsrc = stream->udpsrc[i];
 320
 321     if (udpsrc) {
 322       gst_element_set_state (udpsrc, GST_STATE_NULL);
 323       gst_bin_remove (GST_BIN_CAST (demux), udpsrc);
 324       stream->udpsrc[i] = NULL;
 325     }
 326   }
 327   if (stream->udpsink) {
 328     gst_element_set_state (stream->udpsink, GST_STATE_NULL);
 329     gst_bin_remove (GST_BIN_CAST (demux), stream->udpsink);
 330     stream->udpsink = NULL;
 331   }
 332   if (stream->srcpad) {
 333     gst_pad_set_active (stream->srcpad, FALSE);
 334     if (stream->added) {
 335       gst_element_remove_pad (GST_ELEMENT_CAST (demux), stream->srcpad);
 336       stream->added = FALSE;
 337     }
 338     stream->srcpad = NULL;
 339   }
 340   g_free (stream);
 341 }
 342
 343 static gboolean
 344 is_multicast_address (const gchar * host_name)
 345 {
 346   GInetAddress *addr;
 347   GResolver *resolver = NULL;
 348   gboolean ret = FALSE;
 349
 350   addr = g_inet_address_new_from_string (host_name);
 351   if (!addr) {
 352     GList *results;
 353
 354     resolver = g_resolver_get_default ();
 355     results = g_resolver_lookup_by_name (resolver, host_name, NULL, NULL);
 356     if (!results)
 357       goto out;
 358     addr = G_INET_ADDRESS (g_object_ref (results->data));
 359
 360     g_resolver_free_addresses (results);
 361   }
 362   g_assert (addr != NULL);
 363
 364   ret = g_inet_address_get_is_multicast (addr);
 365
 366 out:
 367   if (resolver)
 368     g_object_unref (resolver);
 369   if (addr)
 370     g_object_unref (addr);
 371   return ret;
 372 }
 373
 374 static GstSDPStream *
 375 gst_sdp_demux_create_stream (GstSDPDemux * demux, GstSDPMessage * sdp, gint idx)
 376 {
 377   GstSDPStream *stream;
 378   const gchar *payload;
 379   const GstSDPMedia *media;
 380   const GstSDPConnection *conn;
 381
 382   /* get media, should not return NULL */
 383   media = gst_sdp_message_get_media (sdp, idx);
 384   if (media == NULL)
 385     return NULL;
 386
 387   stream = g_new0 (GstSDPStream, 1);
 388   stream->parent = demux;
 389   /* we mark the pad as not linked, we will mark it as OK when we add the pad to
 390    * the element. */
 391   stream->last_ret = GST_FLOW_OK;
 392   stream->added = FALSE;
 393   stream->disabled = FALSE;
 394   stream->id = demux->numstreams++;
 395   stream->eos = FALSE;
 396
 397   /* we must have a payload. No payload means we cannot create caps */
 398   /* FIXME, handle multiple formats. */
 399   if ((payload = gst_sdp_media_get_format (media, 0))) {
 400     stream->pt = atoi (payload);
 401     /* convert caps */
 402     stream->caps = gst_sdp_demux_media_to_caps (stream->pt, media);
 403
 404     if (stream->pt >= 96) {
 405       /* If we have a dynamic payload type, see if we have a stream with the
 406        * same payload number. If there is one, they are part of the same
 407        * container and we only need to add one pad. */
 408       if (find_stream (demux, GINT_TO_POINTER (stream->pt),
 409               (gpointer) find_stream_by_pt)) {
 410         stream->container = TRUE;
 411       }
 412     }
 413   }
 414   if (!(conn = gst_sdp_media_get_connection (media, 0))) {
 415     if (!(conn = gst_sdp_message_get_connection (sdp)))
 416       goto no_connection;
 417   }
 418
 419   if (!conn->address)
 420     goto no_connection;
 421
 422   stream->destination = conn->address;
 423   stream->ttl = conn->ttl;
 424   stream->multicast = is_multicast_address (stream->destination);
 425
 426   stream->rtp_port = gst_sdp_media_get_port (media);
 427   if (gst_sdp_media_get_attribute_val (media, "rtcp")) {
 428     /* FIXME, RFC 3605 */
 429     stream->rtcp_port = stream->rtp_port + 1;
 430   } else {
 431     stream->rtcp_port = stream->rtp_port + 1;
 432   }
 433
 434   GST_DEBUG_OBJECT (demux, "stream %d, (%p)", stream->id, stream);
 435   GST_DEBUG_OBJECT (demux, " pt: %d", stream->pt);
 436   GST_DEBUG_OBJECT (demux, " container: %d", stream->container);
 437   GST_DEBUG_OBJECT (demux, " caps: %" GST_PTR_FORMAT, stream->caps);
 438
 439   /* we keep track of all streams */
 440   demux->streams = g_list_append (demux->streams, stream);
 441
 442   return stream;
 443
 444   /* ERRORS */
 445 no_connection:
 446   {
 447     gst_sdp_demux_stream_free (demux, stream);
 448     return NULL;
 449   }
 450 }
 451
 452 static void
 453 gst_sdp_demux_cleanup (GstSDPDemux * demux)
 454 {
 455   GList *walk;
 456
 457   GST_DEBUG_OBJECT (demux, "cleanup");
 458
 459   for (walk = demux->streams; walk; walk = g_list_next (walk)) {
 460     GstSDPStream *stream = (GstSDPStream *) walk->data;
 461
 462     gst_sdp_demux_stream_free (demux, stream);
 463   }
 464   g_list_free (demux->streams);
 465   demux->streams = NULL;
 466   if (demux->session) {
 467     if (demux->session_sig_id) {
 468       g_signal_handler_disconnect (demux->session, demux->session_sig_id);
 469       demux->session_sig_id = 0;
 470     }
 471     if (demux->session_nmp_id) {
 472       g_signal_handler_disconnect (demux->session, demux->session_nmp_id);
 473       demux->session_nmp_id = 0;
 474     }
 475     if (demux->session_ptmap_id) {
 476       g_signal_handler_disconnect (demux->session, demux->session_ptmap_id);
 477       demux->session_ptmap_id = 0;
 478     }
 479     gst_element_set_state (demux->session, GST_STATE_NULL);
 480     gst_bin_remove (GST_BIN_CAST (demux), demux->session);
 481     demux->session = NULL;
 482   }
 483   demux->numstreams = 0;
 484 }
 485
 486 #define PARSE_INT(p, del, res)          \
 487 G_STMT_START {                          \
 488   gchar *t = p;                         \
 489   p = strstr (p, del);                  \
 490   if (p == NULL)                        \
 491     res = -1;                           \
 492   else {                                \
 493     *p = '\0';                          \
 494     p++;                                \
 495     res = atoi (t);                     \
 496   }                                     \
 497 } G_STMT_END
 498
 499 #define PARSE_STRING(p, del, res)       \
 500 G_STMT_START {                          \
 501   gchar *t = p;                         \
 502   p = strstr (p, del);                  \
 503   if (p == NULL) {                      \
 504     res = NULL;                         \
 505     p = t;                              \
 506   }                                     \
 507   else {                                \
 508     *p = '\0';                          \
 509     p++;                                \
 510     res = t;                            \
 511   }                                     \
 512 } G_STMT_END
 513
 514 #define SKIP_SPACES(p)                  \
 515   while (*p && g_ascii_isspace (*p))    \
 516     p++;
 517
 518 /* rtpmap contains:
 519  *
 520  *  <payload> <encoding_name>/<clock_rate>[/<encoding_params>]
 521  */
 522 static gboolean
 523 gst_sdp_demux_parse_rtpmap (const gchar * rtpmap, gint * payload, gchar ** name,
 524     gint * rate, gchar ** params)
 525 {
 526   gchar *p, *t;
 527
 528   t = p = (gchar *) rtpmap;
 529
 530   PARSE_INT (p, " ", *payload);
 531   if (*payload == -1)
 532     return FALSE;
 533
 534   SKIP_SPACES (p);
 535   if (*p == '\0')
 536     return FALSE;
 537
 538   PARSE_STRING (p, "/", *name);
 539   if (*name == NULL) {
 540     GST_DEBUG ("no rate, name %s", p);
 541     /* no rate, assume -1 then */
 542     *name = p;
 543     *rate = -1;
 544     return TRUE;
 545   }
 546
 547   t = p;
 548   p = strstr (p, "/");
 549   if (p == NULL) {
 550     *rate = atoi (t);
 551     return TRUE;
 552   }
 553   *p = '\0';
 554   p++;
 555   *rate = atoi (t);
 556
 557   t = p;
 558   if (*p == '\0')
 559     return TRUE;
 560   *params = t;
 561
 562   return TRUE;
 563 }
 564
 565 /*
 566  *  Mapping of caps to and from SDP fields:
 567  *
 568  *   m=<media> <UDP port> RTP/AVP <payload>
 569  *   a=rtpmap:<payload> <encoding_name>/<clock_rate>[/<encoding_params>]
 570  *   a=fmtp:<payload> <param>[=<value>];...
 571  */
 572 static GstCaps *
 573 gst_sdp_demux_media_to_caps (gint pt, const GstSDPMedia * media)
 574 {
 575   GstCaps *caps;
 576   const gchar *rtpmap;
 577   const gchar *fmtp;
 578   gchar *name = NULL;
 579   gint rate = -1;
 580   gchar *params = NULL;
 581   gchar *tmp;
 582   GstStructure *s;
 583   gint payload = 0;
 584   gboolean ret;
 585
 586   /* get and parse rtpmap */
 587   if ((rtpmap = gst_sdp_media_get_attribute_val (media, "rtpmap"))) {
 588     ret = gst_sdp_demux_parse_rtpmap (rtpmap, &payload, &name, &rate, &params);
 589     if (ret) {
 590       if (payload != pt) {
 591         /* we ignore the rtpmap if the payload type is different. */
 592         g_warning ("rtpmap of wrong payload type, ignoring");
 593         name = NULL;
 594         rate = -1;
 595         params = NULL;
 596       }
 597     } else {
 598       /* if we failed to parse the rtpmap for a dynamic payload type, we have an
 599        * error */
 600       if (pt >= 96)
 601         goto no_rtpmap;
 602       /* else we can ignore */
 603       g_warning ("error parsing rtpmap, ignoring");
 604     }
 605   } else {
 606     /* dynamic payloads need rtpmap or we fail */
 607     if (pt >= 96)
 608       goto no_rtpmap;
 609   }
 610   /* check if we have a rate, if not, we need to look up the rate from the
 611    * default rates based on the payload types. */
 612   if (rate == -1) {
 613     const GstRTPPayloadInfo *info;
 614
 615     if (GST_RTP_PAYLOAD_IS_DYNAMIC (pt)) {
 616       /* dynamic types, use media and encoding_name */
 617       tmp = g_ascii_strdown (media->media, -1);
 618       info = gst_rtp_payload_info_for_name (tmp, name);
 619       g_free (tmp);
 620     } else {
 621       /* static types, use payload type */
 622       info = gst_rtp_payload_info_for_pt (pt);
 623     }
 624
 625     if (info) {
 626       if ((rate = info->clock_rate) == 0)
 627         rate = -1;
 628     }
 629     /* we fail if we cannot find one */
 630     if (rate == -1)
 631       goto no_rate;
 632   }
 633
 634   tmp = g_ascii_strdown (media->media, -1);
 635   caps = gst_caps_new_simple ("application/x-rtp",
 636       "media", G_TYPE_STRING, tmp, "payload", G_TYPE_INT, pt, NULL);
 637   g_free (tmp);
 638   s = gst_caps_get_structure (caps, 0);
 639
 640   gst_structure_set (s, "clock-rate", G_TYPE_INT, rate, NULL);
 641
 642   /* encoding name must be upper case */
 643   if (name != NULL) {
 644     tmp = g_ascii_strup (name, -1);
 645     gst_structure_set (s, "encoding-name", G_TYPE_STRING, tmp, NULL);
 646     g_free (tmp);
 647   }
 648
 649   /* params must be lower case */
 650   if (params != NULL) {
 651     tmp = g_ascii_strdown (params, -1);
 652     gst_structure_set (s, "encoding-params", G_TYPE_STRING, tmp, NULL);
 653     g_free (tmp);
 654   }
 655
 656   /* parse optional fmtp: field */
 657   if ((fmtp = gst_sdp_media_get_attribute_val (media, "fmtp"))) {
 658     gchar *p;
 659     gint payload = 0;
 660
 661     p = (gchar *) fmtp;
 662
 663     /* p is now of the format <payload> <param>[=<value>];... */
 664     PARSE_INT (p, " ", payload);
 665     if (payload != -1 && payload == pt) {
 666       gchar **pairs;
 667       gint i;
 668
 669       /* <param>[=<value>] are separated with ';' */
 670       pairs = g_strsplit (p, ";", 0);
 671       for (i = 0; pairs[i]; i++) {
 672         gchar *valpos, *key;
 673         const gchar *val;
 674
 675         /* the key may not have a '=', the value can have other '='s */
 676         valpos = strstr (pairs[i], "=");
 677         if (valpos) {
 678           /* we have a '=' and thus a value, remove the '=' with \0 */
 679           *valpos = '\0';
 680           /* value is everything between '=' and ';'. FIXME, strip? */
 681           val = g_strstrip (valpos + 1);
 682         } else {
 683           /* simple <param>;.. is translated into <param>=1;... */
 684           val = "1";
 685         }
 686         /* strip the key of spaces, convert key to lowercase but not the value. */
 687         key = g_strstrip (pairs[i]);
 688         if (strlen (key) > 1) {
 689           tmp = g_ascii_strdown (key, -1);
 690           gst_structure_set (s, tmp, G_TYPE_STRING, val, NULL);
 691           g_free (tmp);
 692         }
 693       }
 694       g_strfreev (pairs);
 695     }
 696   }
 697   return caps;
 698
 699   /* ERRORS */
 700 no_rtpmap:
 701   {
 702     g_warning ("rtpmap type not given for dynamic payload %d", pt);
 703     return NULL;
 704   }
 705 no_rate:
 706   {
 707     g_warning ("rate unknown for payload type %d", pt);
 708     return NULL;
 709   }
 710 }
 711
 712 /* this callback is called when the session manager generated a new src pad with
 713  * payloaded RTP packets. We simply ghost the pad here. */
 714 static void
 715 new_session_pad (GstElement * session, GstPad * pad, GstSDPDemux * demux)
 716 {
 717   gchar *name;
 718   GstPadTemplate *template;
 719   gint id, ssrc, pt;
 720   GList *lstream;
 721   GstSDPStream *stream;
 722   gboolean all_added;
 723
 724   GST_DEBUG_OBJECT (demux, "got new session pad %" GST_PTR_FORMAT, pad);
 725
 726   GST_SDP_STREAM_LOCK (demux);
 727   /* find stream */
 728   name = gst_object_get_name (GST_OBJECT_CAST (pad));
 729   if (sscanf (name, "recv_rtp_src_%u_%u_%u", &id, &ssrc, &pt) != 3)
 730     goto unknown_stream;
 731
 732   GST_DEBUG_OBJECT (demux, "stream: %u, SSRC %d, PT %d", id, ssrc, pt);
 733
 734   stream =
 735       find_stream (demux, GINT_TO_POINTER (id), (gpointer) find_stream_by_id);
 736   if (stream == NULL)
 737     goto unknown_stream;
 738
 739   /* no need for a timeout anymore now */
 740   g_object_set (G_OBJECT (stream->udpsrc[0]), "timeout", (guint64) 0, NULL);
 741
 742   /* create a new pad we will use to stream to */
 743   template = gst_static_pad_template_get (&rtptemplate);
 744   stream->srcpad = gst_ghost_pad_new_from_template (name, pad, template);
 745   gst_object_unref (template);
 746   g_free (name);
 747
 748   stream->added = TRUE;
 749   gst_pad_set_active (stream->srcpad, TRUE);
 750   gst_element_add_pad (GST_ELEMENT_CAST (demux), stream->srcpad);
 751
 752   /* check if we added all streams */
 753   all_added = TRUE;
 754   for (lstream = demux->streams; lstream; lstream = g_list_next (lstream)) {
 755     stream = (GstSDPStream *) lstream->data;
 756     /* a container stream only needs one pad added. Also disabled streams don't
 757      * count */
 758     if (!stream->container && !stream->disabled && !stream->added) {
 759       all_added = FALSE;
 760       break;
 761     }
 762   }
 763   GST_SDP_STREAM_UNLOCK (demux);
 764
 765   if (all_added) {
 766     GST_DEBUG_OBJECT (demux, "We added all streams");
 767     /* when we get here, all stream are added and we can fire the no-more-pads
 768      * signal. */
 769     gst_element_no_more_pads (GST_ELEMENT_CAST (demux));
 770   }
 771
 772   return;
 773
 774   /* ERRORS */
 775 unknown_stream:
 776   {
 777     GST_DEBUG_OBJECT (demux, "ignoring unknown stream");
 778     GST_SDP_STREAM_UNLOCK (demux);
 779     g_free (name);
 780     return;
 781   }
 782 }
 783
 784 static void
 785 rtsp_session_pad_added (GstElement * session, GstPad * pad, GstSDPDemux * demux)
 786 {
 787   GstPad *srcpad = NULL;
 788   gchar *name;
 789
 790   GST_DEBUG_OBJECT (demux, "got new session pad %" GST_PTR_FORMAT, pad);
 791
 792   name = gst_pad_get_name (pad);
 793   srcpad = gst_ghost_pad_new (name, pad);
 794   g_free (name);
 795
 796   gst_pad_set_active (srcpad, TRUE);
 797   gst_element_add_pad (GST_ELEMENT_CAST (demux), srcpad);
 798 }
 799
 800 static void
 801 rtsp_session_no_more_pads (GstElement * session, GstSDPDemux * demux)
 802 {
 803   GST_DEBUG_OBJECT (demux, "got no-more-pads");
 804   gst_element_no_more_pads (GST_ELEMENT_CAST (demux));
 805 }
 806
 807 static GstCaps *
 808 request_pt_map (GstElement * sess, guint session, guint pt, GstSDPDemux * demux)
 809 {
 810   GstSDPStream *stream;
 811   GstCaps *caps;
 812
 813   GST_DEBUG_OBJECT (demux, "getting pt map for pt %d in session %d", pt,
 814       session);
 815
 816   GST_SDP_STREAM_LOCK (demux);
 817   stream =
 818       find_stream (demux, GINT_TO_POINTER (session),
 819       (gpointer) find_stream_by_id);
 820   if (!stream)
 821     goto unknown_stream;
 822
 823   caps = stream->caps;
 824   if (caps)
 825     gst_caps_ref (caps);
 826   GST_SDP_STREAM_UNLOCK (demux);
 827
 828   return caps;
 829
 830 unknown_stream:
 831   {
 832     GST_DEBUG_OBJECT (demux, "unknown stream %d", session);
 833     GST_SDP_STREAM_UNLOCK (demux);
 834     return NULL;
 835   }
 836 }
 837
 838 static void
 839 gst_sdp_demux_do_stream_eos (GstSDPDemux * demux, guint session)
 840 {
 841   GstSDPStream *stream;
 842
 843   GST_DEBUG_OBJECT (demux, "setting stream for session %u to EOS", session);
 844
 845   /* get stream for session */
 846   stream =
 847       find_stream (demux, GINT_TO_POINTER (session),
 848       (gpointer) find_stream_by_id);
 849   if (!stream)
 850     goto unknown_stream;
 851
 852   if (stream->eos)
 853     goto was_eos;
 854
 855   stream->eos = TRUE;
 856   gst_sdp_demux_stream_push_event (demux, stream, gst_event_new_eos ());
 857   return;
 858
 859   /* ERRORS */
 860 unknown_stream:
 861   {
 862     GST_DEBUG_OBJECT (demux, "unknown stream for session %u", session);
 863     return;
 864   }
 865 was_eos:
 866   {
 867     GST_DEBUG_OBJECT (demux, "stream for session %u was already EOS", session);
 868     return;
 869   }
 870 }
 871
 872 static void
 873 on_bye_ssrc (GstElement * manager, guint session, guint32 ssrc,
 874     GstSDPDemux * demux)
 875 {
 876   GST_DEBUG_OBJECT (demux, "SSRC %08x in session %u received BYE", ssrc,
 877       session);
 878
 879   gst_sdp_demux_do_stream_eos (demux, session);
 880 }
 881
 882 static void
 883 on_timeout (GstElement * manager, guint session, guint32 ssrc,
 884     GstSDPDemux * demux)
 885 {
 886   GST_DEBUG_OBJECT (demux, "SSRC %08x in session %u timed out", ssrc, session);
 887
 888   gst_sdp_demux_do_stream_eos (demux, session);
 889 }
 890
 891 /* try to get and configure a manager */
 892 static gboolean
 893 gst_sdp_demux_configure_manager (GstSDPDemux * demux, char *rtsp_sdp)
 894 {
 895   /* configure the session manager */
 896   if (rtsp_sdp != NULL) {
 897     if (!(demux->session = gst_element_factory_make ("rtspsrc", NULL)))
 898       goto rtspsrc_failed;
 899
 900     g_object_set (demux->session, "location", rtsp_sdp, NULL);
 901
 902     GST_DEBUG_OBJECT (demux, "connect to signals on rtspsrc");
 903     demux->session_sig_id =
 904         g_signal_connect (demux->session, "pad-added",
 905         (GCallback) rtsp_session_pad_added, demux);
 906     demux->session_nmp_id =
 907         g_signal_connect (demux->session, "no-more-pads",
 908         (GCallback) rtsp_session_no_more_pads, demux);
 909   } else {
 910     if (!(demux->session = gst_element_factory_make ("rtpbin", NULL)))
 911       goto manager_failed;
 912
 913     /* connect to signals if we did not already do so */
 914     GST_DEBUG_OBJECT (demux, "connect to signals on session manager");
 915     demux->session_sig_id =
 916         g_signal_connect (demux->session, "pad-added",
 917         (GCallback) new_session_pad, demux);
 918     demux->session_ptmap_id =
 919         g_signal_connect (demux->session, "request-pt-map",
 920         (GCallback) request_pt_map, demux);
 921     g_signal_connect (demux->session, "on-bye-ssrc", (GCallback) on_bye_ssrc,
 922         demux);
 923     g_signal_connect (demux->session, "on-bye-timeout", (GCallback) on_timeout,
 924         demux);
 925     g_signal_connect (demux->session, "on-timeout", (GCallback) on_timeout,
 926         demux);
 927   }
 928
 929   g_object_set (demux->session, "latency", demux->latency, NULL);
 930
 931   /* we manage this element */
 932   gst_bin_add (GST_BIN_CAST (demux), demux->session);
 933
 934   return TRUE;
 935
 936   /* ERRORS */
 937 manager_failed:
 938   {
 939     GST_DEBUG_OBJECT (demux, "no session manager element gstrtpbin found");
 940     return FALSE;
 941   }
 942 rtspsrc_failed:
 943   {
 944     GST_DEBUG_OBJECT (demux, "no manager element rtspsrc found");
 945     return FALSE;
 946   }
 947 }
 948
 949 static gboolean
 950 gst_sdp_demux_stream_configure_udp (GstSDPDemux * demux, GstSDPStream * stream)
 951 {
 952   gchar *uri, *name;
 953   const gchar *destination;
 954   GstPad *pad;
 955
 956   GST_DEBUG_OBJECT (demux, "creating UDP sources for multicast");
 957
 958   /* if the destination is not a multicast address, we just want to listen on
 959    * our local ports */
 960   if (!stream->multicast)
 961     destination = "0.0.0.0";
 962   else
 963     destination = stream->destination;
 964
 965   /* creating UDP source */
 966   if (stream->rtp_port != -1) {
 967     GST_DEBUG_OBJECT (demux, "receiving RTP from %s:%d", destination,
 968         stream->rtp_port);
 969
 970     uri = g_strdup_printf ("udp://%s:%d", destination, stream->rtp_port);
 971     stream->udpsrc[0] =
 972         gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
 973     g_free (uri);
 974     if (stream->udpsrc[0] == NULL)
 975       goto no_element;
 976
 977     /* take ownership */
 978     gst_bin_add (GST_BIN_CAST (demux), stream->udpsrc[0]);
 979
 980     GST_DEBUG_OBJECT (demux,
 981         "setting up UDP source with timeout %" G_GINT64_FORMAT,
 982         demux->udp_timeout);
 983
 984     /* configure a timeout on the UDP port. When the timeout message is
 985      * posted, we assume UDP transport is not possible. */
 986     g_object_set (G_OBJECT (stream->udpsrc[0]), "timeout",
 987         demux->udp_timeout * 1000, NULL);
 988
 989     /* get output pad of the UDP source. */
 990     pad = gst_element_get_static_pad (stream->udpsrc[0], "src");
 991
 992     name = g_strdup_printf ("recv_rtp_sink_%u", stream->id);
 993     stream->channelpad[0] = gst_element_get_request_pad (demux->session, name);
 994     g_free (name);
 995
 996     GST_DEBUG_OBJECT (demux, "connecting RTP source 0 to manager");
 997     /* configure for UDP delivery, we need to connect the UDP pads to
 998      * the session plugin. */
 999     gst_pad_link (pad, stream->channelpad[0]);
1000     gst_object_unref (pad);
1001
1002     /* change state */
1003     gst_element_set_state (stream->udpsrc[0], GST_STATE_PAUSED);
1004   }
1005
1006   /* creating another UDP source */
1007   if (stream->rtcp_port != -1) {
1008     GST_DEBUG_OBJECT (demux, "receiving RTCP from %s:%d", destination,
1009         stream->rtcp_port);
1010     uri = g_strdup_printf ("udp://%s:%d", destination, stream->rtcp_port);
1011     stream->udpsrc[1] =
1012         gst_element_make_from_uri (GST_URI_SRC, uri, NULL, NULL);
1013     g_free (uri);
1014     if (stream->udpsrc[1] == NULL)
1015       goto no_element;
1016
1017     /* take ownership */
1018     gst_bin_add (GST_BIN_CAST (demux), stream->udpsrc[1]);
1019
1020     GST_DEBUG_OBJECT (demux, "connecting RTCP source to manager");
1021
1022     name = g_strdup_printf ("recv_rtcp_sink_%u", stream->id);
1023     stream->channelpad[1] = gst_element_get_request_pad (demux->session, name);
1024     g_free (name);
1025
1026     pad = gst_element_get_static_pad (stream->udpsrc[1], "src");
1027     gst_pad_link (pad, stream->channelpad[1]);
1028     gst_object_unref (pad);
1029
1030     gst_element_set_state (stream->udpsrc[1], GST_STATE_PAUSED);
1031   }
1032   return TRUE;
1033
1034   /* ERRORS */
1035 no_element:
1036   {
1037     GST_DEBUG_OBJECT (demux, "no UDP source element found");
1038     return FALSE;
1039   }
1040 }
1041
1042 /* configure the UDP sink back to the server for status reports */
1043 static gboolean
1044 gst_sdp_demux_stream_configure_udp_sink (GstSDPDemux * demux,
1045     GstSDPStream * stream)
1046 {
1047   GstPad *pad, *sinkpad;
1048   gint port;
1049   GSocket *socket;
1050   gchar *destination, *uri, *name;
1051
1052   /* get destination and port */
1053   port = stream->rtcp_port;
1054   destination = stream->destination;
1055
1056   GST_DEBUG_OBJECT (demux, "configure UDP sink for %s:%d", destination, port);
1057
1058   uri = g_strdup_printf ("udp://%s:%d", destination, port);
1059   stream->udpsink = gst_element_make_from_uri (GST_URI_SINK, uri, NULL, NULL);
1060   g_free (uri);
1061   if (stream->udpsink == NULL)
1062     goto no_sink_element;
1063
1064   /* we clear all destinations because we don't really know where to send the
1065    * RTCP to and we want to avoid sending it to our own ports.
1066    * FIXME when we get an RTCP packet from the sender, we could look at its
1067    * source port and address and try to send RTCP there. */
1068   if (!stream->multicast)
1069     g_signal_emit_by_name (stream->udpsink, "clear");
1070
1071   g_object_set (G_OBJECT (stream->udpsink), "auto-multicast", FALSE, NULL);
1072   g_object_set (G_OBJECT (stream->udpsink), "loop", FALSE, NULL);
1073   /* no sync needed */
1074   g_object_set (G_OBJECT (stream->udpsink), "sync", FALSE, NULL);
1075   /* no async state changes needed */
1076   g_object_set (G_OBJECT (stream->udpsink), "async", FALSE, NULL);
1077
1078   if (stream->udpsrc[1]) {
1079     /* configure socket, we give it the same UDP socket as the udpsrc for RTCP
1080      * because some servers check the port number of where it sends RTCP to identify
1081      * the RTCP packets it receives */
1082     g_object_get (G_OBJECT (stream->udpsrc[1]), "used_socket", &socket, NULL);
1083     GST_DEBUG_OBJECT (demux, "UDP src has socket %p", socket);
1084     /* configure socket and make sure udpsink does not close it when shutting
1085      * down, it belongs to udpsrc after all. */
1086     g_object_set (G_OBJECT (stream->udpsink), "socket", socket, NULL);
1087     g_object_set (G_OBJECT (stream->udpsink), "close-socket", FALSE, NULL);
1088     g_object_unref (socket);
1089   }
1090
1091   /* we keep this playing always */
1092   gst_element_set_locked_state (stream->udpsink, TRUE);
1093   gst_element_set_state (stream->udpsink, GST_STATE_PLAYING);
1094
1095   gst_bin_add (GST_BIN_CAST (demux), stream->udpsink);
1096
1097   /* get session RTCP pad */
1098   name = g_strdup_printf ("send_rtcp_src_%u", stream->id);
1099   pad = gst_element_get_request_pad (demux->session, name);
1100   g_free (name);
1101
1102   /* and link */
1103   if (pad) {
1104     sinkpad = gst_element_get_static_pad (stream->udpsink, "sink");
1105     gst_pad_link (pad, sinkpad);
1106     gst_object_unref (sinkpad);
1107   } else {
1108     /* not very fatal, we just won't be able to send RTCP */
1109     GST_WARNING_OBJECT (demux, "could not get session RTCP pad");
1110   }
1111
1112
1113   return TRUE;
1114
1115   /* ERRORS */
1116 no_sink_element:
1117   {
1118     GST_DEBUG_OBJECT (demux, "no UDP sink element found");
1119     return FALSE;
1120   }
1121 }
1122
1123 static GstFlowReturn
1124 gst_sdp_demux_combine_flows (GstSDPDemux * demux, GstSDPStream * stream,
1125     GstFlowReturn ret)
1126 {
1127   GList *streams;
1128
1129   /* store the value */
1130   stream->last_ret = ret;
1131
1132   /* if it's success we can return the value right away */
1133   if (ret == GST_FLOW_OK)
1134     goto done;
1135
1136   /* any other error that is not-linked can be returned right
1137    * away */
1138   if (ret != GST_FLOW_NOT_LINKED)
1139     goto done;
1140
1141   /* only return NOT_LINKED if all other pads returned NOT_LINKED */
1142   for (streams = demux->streams; streams; streams = g_list_next (streams)) {
1143     GstSDPStream *ostream = (GstSDPStream *) streams->data;
1144
1145     ret = ostream->last_ret;
1146     /* some other return value (must be SUCCESS but we can return
1147      * other values as well) */
1148     if (ret != GST_FLOW_NOT_LINKED)
1149       goto done;
1150   }
1151   /* if we get here, all other pads were unlinked and we return
1152    * NOT_LINKED then */
1153 done:
1154   return ret;
1155 }
1156
1157 static void
1158 gst_sdp_demux_stream_push_event (GstSDPDemux * demux, GstSDPStream * stream,
1159     GstEvent * event)
1160 {
1161   /* only streams that have a connection to the outside world */
1162   if (stream->srcpad == NULL)
1163     goto done;
1164
1165   if (stream->channelpad[0]) {
1166     gst_event_ref (event);
1167     gst_pad_send_event (stream->channelpad[0], event);
1168   }
1169
1170   if (stream->channelpad[1]) {
1171     gst_event_ref (event);
1172     gst_pad_send_event (stream->channelpad[1], event);
1173   }
1174
1175 done:
1176   gst_event_unref (event);
1177 }
1178
1179 static void
1180 gst_sdp_demux_handle_message (GstBin * bin, GstMessage * message)
1181 {
1182   GstSDPDemux *demux;
1183
1184   demux = GST_SDP_DEMUX (bin);
1185
1186   switch (GST_MESSAGE_TYPE (message)) {
1187     case GST_MESSAGE_ELEMENT:
1188     {
1189       const GstStructure *s = gst_message_get_structure (message);
1190
1191       if (gst_structure_has_name (s, "GstUDPSrcTimeout")) {
1192         gboolean ignore_timeout;
1193
1194         GST_DEBUG_OBJECT (bin, "timeout on UDP port");
1195
1196         GST_OBJECT_LOCK (demux);
1197         ignore_timeout = demux->ignore_timeout;
1198         demux->ignore_timeout = TRUE;
1199         GST_OBJECT_UNLOCK (demux);
1200
1201         /* we only act on the first udp timeout message, others are irrelevant
1202          * and can be ignored. */
1203         if (ignore_timeout)
1204           gst_message_unref (message);
1205         else {
1206           GST_ELEMENT_ERROR (demux, RESOURCE, READ, (NULL),
1207               ("Could not receive any UDP packets for %.4f seconds, maybe your "
1208                   "firewall is blocking it.",
1209                   gst_guint64_to_gdouble (demux->udp_timeout / 1000000.0)));
1210         }
1211         return;
1212       }
1213       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
1214       break;
1215     }
1216     case GST_MESSAGE_ERROR:
1217     {
1218       GstObject *udpsrc;
1219       GstSDPStream *stream;
1220       GstFlowReturn ret;
1221
1222       udpsrc = GST_MESSAGE_SRC (message);
1223
1224       GST_DEBUG_OBJECT (demux, "got error from %s", GST_ELEMENT_NAME (udpsrc));
1225
1226       stream = find_stream (demux, udpsrc, (gpointer) find_stream_by_udpsrc);
1227       /* fatal but not our message, forward */
1228       if (!stream)
1229         goto forward;
1230
1231       /* we ignore the RTCP udpsrc */
1232       if (stream->udpsrc[1] == GST_ELEMENT_CAST (udpsrc))
1233         goto done;
1234
1235       /* if we get error messages from the udp sources, that's not a problem as
1236        * long as not all of them error out. We also don't really know what the
1237        * problem is, the message does not give enough detail... */
1238       ret = gst_sdp_demux_combine_flows (demux, stream, GST_FLOW_NOT_LINKED);
1239       GST_DEBUG_OBJECT (demux, "combined flows: %s", gst_flow_get_name (ret));
1240       if (ret != GST_FLOW_OK)
1241         goto forward;
1242
1243     done:
1244       gst_message_unref (message);
1245       break;
1246
1247     forward:
1248       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
1249       break;
1250     }
1251     default:
1252     {
1253       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
1254       break;
1255     }
1256   }
1257 }
1258
1259 static gboolean
1260 gst_sdp_demux_start (GstSDPDemux * demux)
1261 {
1262   guint8 *data;
1263   guint size;
1264   gint i, n_streams;
1265   GstSDPMessage sdp = { 0 };
1266   GstSDPStream *stream = NULL;
1267   GList *walk;
1268   gchar *uri = NULL;
1269   GstStateChangeReturn ret;
1270
1271   /* grab the lock so that no state change can interfere */
1272   GST_SDP_STREAM_LOCK (demux);
1273
1274   GST_DEBUG_OBJECT (demux, "parse SDP...");
1275
1276   size = gst_adapter_available (demux->adapter);
1277   data = gst_adapter_take (demux->adapter, size);
1278
1279   gst_sdp_message_init (&sdp);
1280   if (gst_sdp_message_parse_buffer (data, size, &sdp) != GST_SDP_OK)
1281     goto could_not_parse;
1282
1283   if (demux->debug)
1284     gst_sdp_message_dump (&sdp);
1285
1286   /* maybe this is plain RTSP DESCRIBE rtsp and we should redirect */
1287   /* look for rtsp control url */
1288   {
1289     const gchar *control;
1290
1291     for (i = 0;; i++) {
1292       control = gst_sdp_message_get_attribute_val_n (&sdp, "control", i);
1293       if (control == NULL)
1294         break;
1295
1296       /* only take fully qualified urls */
1297       if (g_str_has_prefix (control, "rtsp://"))
1298         break;
1299     }
1300     if (!control) {
1301       gint idx;
1302
1303       /* try to find non-aggragate control */
1304       n_streams = gst_sdp_message_medias_len (&sdp);
1305
1306       for (idx = 0; idx < n_streams; idx++) {
1307         const GstSDPMedia *media;
1308
1309         /* get media, should not return NULL */
1310         media = gst_sdp_message_get_media (&sdp, idx);
1311         if (media == NULL)
1312           break;
1313
1314         for (i = 0;; i++) {
1315           control = gst_sdp_media_get_attribute_val_n (media, "control", i);
1316           if (control == NULL)
1317             break;
1318
1319           /* only take fully qualified urls */
1320           if (g_str_has_prefix (control, "rtsp://"))
1321             break;
1322         }
1323         /* this media has no control, exit */
1324         if (!control)
1325           break;
1326       }
1327     }
1328
1329     if (control) {
1330       /* we have RTSP now */
1331       uri = gst_sdp_message_as_uri ("rtsp-sdp", &sdp);
1332
1333       if (demux->redirect) {
1334         GST_INFO_OBJECT (demux, "redirect to %s", uri);
1335
1336         gst_element_post_message (GST_ELEMENT_CAST (demux),
1337             gst_message_new_element (GST_OBJECT_CAST (demux),
1338                 gst_structure_new ("redirect",
1339                     "new-location", G_TYPE_STRING, uri, NULL)));
1340         goto sent_redirect;
1341       }
1342     }
1343   }
1344
1345   /* we get here when we didn't do a redirect */
1346
1347   /* try to get and configure a manager */
1348   if (!gst_sdp_demux_configure_manager (demux, uri))
1349     goto no_manager;
1350   if (!uri) {
1351     /* create streams with UDP sources and sinks */
1352     n_streams = gst_sdp_message_medias_len (&sdp);
1353     for (i = 0; i < n_streams; i++) {
1354       stream = gst_sdp_demux_create_stream (demux, &sdp, i);
1355
1356       if (!stream)
1357         continue;
1358
1359       GST_DEBUG_OBJECT (demux, "configuring transport for stream %p", stream);
1360
1361       if (!gst_sdp_demux_stream_configure_udp (demux, stream))
1362         goto transport_failed;
1363       if (!gst_sdp_demux_stream_configure_udp_sink (demux, stream))
1364         goto transport_failed;
1365     }
1366
1367     if (!demux->streams)
1368       goto no_streams;
1369   }
1370
1371   /* set target state on session manager */
1372   /* setting rtspsrc to PLAYING may cause it to loose it that target state
1373    * along the way due to no-preroll udpsrc elements, so ...
1374    * do it in two stages here (similar to other elements) */
1375   if (demux->target > GST_STATE_PAUSED) {
1376     ret = gst_element_set_state (demux->session, GST_STATE_PAUSED);
1377     if (ret == GST_STATE_CHANGE_FAILURE)
1378       goto start_session_failure;
1379   }
1380   ret = gst_element_set_state (demux->session, demux->target);
1381   if (ret == GST_STATE_CHANGE_FAILURE)
1382     goto start_session_failure;
1383
1384   if (!uri) {
1385     /* activate all streams */
1386     for (walk = demux->streams; walk; walk = g_list_next (walk)) {
1387       stream = (GstSDPStream *) walk->data;
1388
1389       /* configure target state on udp sources */
1390       gst_element_set_state (stream->udpsrc[0], demux->target);
1391       gst_element_set_state (stream->udpsrc[1], demux->target);
1392     }
1393   }
1394   GST_SDP_STREAM_UNLOCK (demux);
1395   gst_sdp_message_uninit (&sdp);
1396   g_free (data);
1397
1398   return TRUE;
1399
1400   /* ERRORS */
1401 done:
1402   {
1403     GST_SDP_STREAM_UNLOCK (demux);
1404     gst_sdp_message_uninit (&sdp);
1405     g_free (data);
1406     return FALSE;
1407   }
1408 transport_failed:
1409   {
1410     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL),
1411         ("Could not create RTP stream transport."));
1412     goto done;
1413   }
1414 no_manager:
1415   {
1416     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL),
1417         ("Could not create RTP session manager."));
1418     goto done;
1419   }
1420 could_not_parse:
1421   {
1422     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL),
1423         ("Could not parse SDP message."));
1424     goto done;
1425   }
1426 no_streams:
1427   {
1428     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL),
1429         ("No streams in SDP message."));
1430     goto done;
1431   }
1432 sent_redirect:
1433   {
1434     /* avoid hanging if redirect not handled */
1435     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL),
1436         ("Sent RTSP redirect."));
1437     goto done;
1438   }
1439 start_session_failure:
1440   {
1441     GST_ELEMENT_ERROR (demux, STREAM, TYPE_NOT_FOUND, (NULL),
1442         ("Could not start RTP session manager."));
1443     gst_element_set_state (demux->session, GST_STATE_NULL);
1444     gst_bin_remove (GST_BIN_CAST (demux), demux->session);
1445     demux->session = NULL;
1446     goto done;
1447   }
1448 }
1449
1450 static gboolean
1451 gst_sdp_demux_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
1452 {
1453   GstSDPDemux *demux;
1454   gboolean res = TRUE;
1455
1456   demux = GST_SDP_DEMUX (parent);
1457
1458   switch (GST_EVENT_TYPE (event)) {
1459     case GST_EVENT_EOS:
1460       /* when we get EOS, start parsing the SDP */
1461       res = gst_sdp_demux_start (demux);
1462       gst_event_unref (event);
1463       break;
1464     default:
1465       gst_event_unref (event);
1466       break;
1467   }
1468
1469   return res;
1470 }
1471
1472 static GstFlowReturn
1473 gst_sdp_demux_sink_chain (GstPad * pad, GstObject * parent, GstBuffer * buffer)
1474 {
1475   GstSDPDemux *demux;
1476
1477   demux = GST_SDP_DEMUX (parent);
1478
1479   /* push the SDP message in an adapter, we start doing something with it when
1480    * we receive EOS */
1481   gst_adapter_push (demux->adapter, buffer);
1482
1483   return GST_FLOW_OK;
1484 }
1485
1486 static GstStateChangeReturn
1487 gst_sdp_demux_change_state (GstElement * element, GstStateChange transition)
1488 {
1489   GstSDPDemux *demux;
1490   GstStateChangeReturn ret;
1491
1492   demux = GST_SDP_DEMUX (element);
1493
1494   GST_SDP_STREAM_LOCK (demux);
1495
1496   switch (transition) {
1497     case GST_STATE_CHANGE_NULL_TO_READY:
1498       break;
1499     case GST_STATE_CHANGE_READY_TO_PAUSED:
1500       /* first attempt, don't ignore timeouts */
1501       gst_adapter_clear (demux->adapter);
1502       demux->ignore_timeout = FALSE;
1503       demux->target = GST_STATE_PAUSED;
1504       break;
1505     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
1506       demux->target = GST_STATE_PLAYING;
1507       break;
1508     default:
1509       break;
1510   }
1511
1512   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
1513   if (ret == GST_STATE_CHANGE_FAILURE)
1514     goto done;
1515
1516   switch (transition) {
1517     case GST_STATE_CHANGE_READY_TO_PAUSED:
1518       ret = GST_STATE_CHANGE_NO_PREROLL;
1519       break;
1520     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
1521       ret = GST_STATE_CHANGE_NO_PREROLL;
1522       demux->target = GST_STATE_PAUSED;
1523       break;
1524     case GST_STATE_CHANGE_PAUSED_TO_READY:
1525       gst_sdp_demux_cleanup (demux);
1526       break;
1527     case GST_STATE_CHANGE_READY_TO_NULL:
1528       break;
1529     default:
1530       break;
1531   }
1532
1533 done:
1534   GST_SDP_STREAM_UNLOCK (demux);
1535
1536   return ret;
1537 }