gst/rtsp-server/rtsp-media.c

   1 /* GStreamer
   2  * Copyright (C) 2008 Wim Taymans <wim.taymans at gmail.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., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, USA.
  18  */
  19
  20 #include <string.h>
  21 #include <stdlib.h>
  22
  23 #include <gst/app/gstappsrc.h>
  24 #include <gst/app/gstappsink.h>
  25
  26 #include "rtsp-funnel.h"
  27 #include "rtsp-media.h"
  28
  29 #define DEFAULT_SHARED          FALSE
  30 #define DEFAULT_REUSABLE        FALSE
  31 #define DEFAULT_PROTOCOLS       GST_RTSP_LOWER_TRANS_UDP | GST_RTSP_LOWER_TRANS_TCP
  32 //#define DEFAULT_PROTOCOLS      GST_RTSP_LOWER_TRANS_UDP_MCAST
  33 #define DEFAULT_EOS_SHUTDOWN    FALSE
  34 #define DEFAULT_BUFFER_SIZE     0x80000
  35 #define DEFAULT_MULTICAST_GROUP "224.2.0.1"
  36
  37 /* define to dump received RTCP packets */
  38 #undef DUMP_STATS
  39
  40 enum
  41 {
  42   PROP_0,
  43   PROP_SHARED,
  44   PROP_REUSABLE,
  45   PROP_PROTOCOLS,
  46   PROP_EOS_SHUTDOWN,
  47   PROP_BUFFER_SIZE,
  48   PROP_MULTICAST_GROUP,
  49   PROP_LAST
  50 };
  51
  52 enum
  53 {
  54   SIGNAL_PREPARED,
  55   SIGNAL_UNPREPARED,
  56   SIGNAL_NEW_STATE,
  57   SIGNAL_LAST
  58 };
  59
  60 GST_DEBUG_CATEGORY_STATIC (rtsp_media_debug);
  61 #define GST_CAT_DEFAULT rtsp_media_debug
  62
  63 static GQuark ssrc_stream_map_key;
  64
  65 static void gst_rtsp_media_get_property (GObject * object, guint propid,
  66     GValue * value, GParamSpec * pspec);
  67 static void gst_rtsp_media_set_property (GObject * object, guint propid,
  68     const GValue * value, GParamSpec * pspec);
  69 static void gst_rtsp_media_finalize (GObject * obj);
  70
  71 static gpointer do_loop (GstRTSPMediaClass * klass);
  72 static gboolean default_handle_message (GstRTSPMedia * media,
  73     GstMessage * message);
  74 static gboolean default_unprepare (GstRTSPMedia * media);
  75 static void unlock_streams (GstRTSPMedia * media);
  76
  77 static guint gst_rtsp_media_signals[SIGNAL_LAST] = { 0 };
  78
  79 G_DEFINE_TYPE (GstRTSPMedia, gst_rtsp_media, G_TYPE_OBJECT);
  80
  81 static void
  82 gst_rtsp_media_class_init (GstRTSPMediaClass * klass)
  83 {
  84   GObjectClass *gobject_class;
  85   GError *error = NULL;
  86
  87   gobject_class = G_OBJECT_CLASS (klass);
  88
  89   gobject_class->get_property = gst_rtsp_media_get_property;
  90   gobject_class->set_property = gst_rtsp_media_set_property;
  91   gobject_class->finalize = gst_rtsp_media_finalize;
  92
  93   g_object_class_install_property (gobject_class, PROP_SHARED,
  94       g_param_spec_boolean ("shared", "Shared",
  95           "If this media pipeline can be shared", DEFAULT_SHARED,
  96           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  97
  98   g_object_class_install_property (gobject_class, PROP_REUSABLE,
  99       g_param_spec_boolean ("reusable", "Reusable",
 100           "If this media pipeline can be reused after an unprepare",
 101           DEFAULT_REUSABLE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 102
 103   g_object_class_install_property (gobject_class, PROP_PROTOCOLS,
 104       g_param_spec_flags ("protocols", "Protocols",
 105           "Allowed lower transport protocols", GST_TYPE_RTSP_LOWER_TRANS,
 106           DEFAULT_PROTOCOLS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 107
 108   g_object_class_install_property (gobject_class, PROP_EOS_SHUTDOWN,
 109       g_param_spec_boolean ("eos-shutdown", "EOS Shutdown",
 110           "Send an EOS event to the pipeline before unpreparing",
 111           DEFAULT_EOS_SHUTDOWN, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 112
 113   g_object_class_install_property (gobject_class, PROP_BUFFER_SIZE,
 114       g_param_spec_uint ("buffer-size", "Buffer Size",
 115           "The kernel UDP buffer size to use", 0, G_MAXUINT,
 116           DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 117
 118   g_object_class_install_property (gobject_class, PROP_MULTICAST_GROUP,
 119       g_param_spec_string ("multicast-group", "Multicast Group",
 120           "The Multicast group to send media to",
 121           DEFAULT_MULTICAST_GROUP, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 122
 123   gst_rtsp_media_signals[SIGNAL_PREPARED] =
 124       g_signal_new ("prepared", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
 125       G_STRUCT_OFFSET (GstRTSPMediaClass, prepared), NULL, NULL,
 126       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
 127
 128   gst_rtsp_media_signals[SIGNAL_UNPREPARED] =
 129       g_signal_new ("unprepared", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
 130       G_STRUCT_OFFSET (GstRTSPMediaClass, unprepared), NULL, NULL,
 131       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
 132
 133   gst_rtsp_media_signals[SIGNAL_NEW_STATE] =
 134       g_signal_new ("new-state", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
 135       G_STRUCT_OFFSET (GstRTSPMediaClass, new_state), NULL, NULL,
 136       g_cclosure_marshal_VOID__INT, G_TYPE_NONE, 0, G_TYPE_INT);
 137
 138   klass->context = g_main_context_new ();
 139   klass->loop = g_main_loop_new (klass->context, TRUE);
 140
 141   GST_DEBUG_CATEGORY_INIT (rtsp_media_debug, "rtspmedia", 0, "GstRTSPMedia");
 142
 143   klass->thread = g_thread_create ((GThreadFunc) do_loop, klass, TRUE, &error);
 144   if (error != NULL) {
 145     g_critical ("could not start bus thread: %s", error->message);
 146   }
 147   klass->handle_message = default_handle_message;
 148   klass->unprepare = default_unprepare;
 149
 150   ssrc_stream_map_key = g_quark_from_static_string ("GstRTSPServer.stream");
 151
 152   gst_element_register (NULL, "rtspfunnel", GST_RANK_NONE, RTSP_TYPE_FUNNEL);
 153
 154 }
 155
 156 static void
 157 gst_rtsp_media_init (GstRTSPMedia * media)
 158 {
 159   media->streams = g_array_new (FALSE, TRUE, sizeof (GstRTSPMediaStream *));
 160   media->lock = g_mutex_new ();
 161   media->cond = g_cond_new ();
 162
 163   media->shared = DEFAULT_SHARED;
 164   media->reusable = DEFAULT_REUSABLE;
 165   media->protocols = DEFAULT_PROTOCOLS;
 166   media->eos_shutdown = DEFAULT_EOS_SHUTDOWN;
 167   media->buffer_size = DEFAULT_BUFFER_SIZE;
 168   media->multicast_group = g_strdup (DEFAULT_MULTICAST_GROUP);
 169 }
 170
 171 void
 172 gst_rtsp_media_trans_cleanup (GstRTSPMediaTrans * trans)
 173 {
 174   if (trans->transport) {
 175     gst_rtsp_transport_free (trans->transport);
 176     trans->transport = NULL;
 177   }
 178   if (trans->rtpsource) {
 179     g_object_set_qdata (trans->rtpsource, ssrc_stream_map_key, NULL);
 180     trans->rtpsource = NULL;
 181   }
 182 }
 183
 184 static void
 185 gst_rtsp_media_stream_free (GstRTSPMediaStream * stream)
 186 {
 187   if (stream->session)
 188     g_object_unref (stream->session);
 189
 190   if (stream->caps)
 191     gst_caps_unref (stream->caps);
 192
 193   if (stream->send_rtp_sink)
 194     gst_object_unref (stream->send_rtp_sink);
 195   if (stream->send_rtp_src)
 196     gst_object_unref (stream->send_rtp_src);
 197   if (stream->send_rtcp_src)
 198     gst_object_unref (stream->send_rtcp_src);
 199   if (stream->recv_rtcp_sink)
 200     gst_object_unref (stream->recv_rtcp_sink);
 201   if (stream->recv_rtp_sink)
 202     gst_object_unref (stream->recv_rtp_sink);
 203
 204   g_list_free (stream->transports);
 205
 206   g_free (stream);
 207 }
 208
 209 static void
 210 gst_rtsp_media_finalize (GObject * obj)
 211 {
 212   GstRTSPMedia *media;
 213   guint i;
 214
 215   media = GST_RTSP_MEDIA (obj);
 216
 217   GST_INFO ("finalize media %p", media);
 218
 219   if (media->pipeline) {
 220     unlock_streams (media);
 221     gst_element_set_state (media->pipeline, GST_STATE_NULL);
 222     gst_object_unref (media->pipeline);
 223   }
 224
 225   for (i = 0; i < media->streams->len; i++) {
 226     GstRTSPMediaStream *stream;
 227
 228     stream = g_array_index (media->streams, GstRTSPMediaStream *, i);
 229
 230     gst_rtsp_media_stream_free (stream);
 231   }
 232   g_array_free (media->streams, TRUE);
 233
 234   g_list_foreach (media->dynamic, (GFunc) gst_object_unref, NULL);
 235   g_list_free (media->dynamic);
 236
 237   if (media->source) {
 238     g_source_destroy (media->source);
 239     g_source_unref (media->source);
 240   }
 241   g_free (media->multicast_group);
 242   g_mutex_free (media->lock);
 243   g_cond_free (media->cond);
 244
 245   G_OBJECT_CLASS (gst_rtsp_media_parent_class)->finalize (obj);
 246 }
 247
 248 static void
 249 gst_rtsp_media_get_property (GObject * object, guint propid,
 250     GValue * value, GParamSpec * pspec)
 251 {
 252   GstRTSPMedia *media = GST_RTSP_MEDIA (object);
 253
 254   switch (propid) {
 255     case PROP_SHARED:
 256       g_value_set_boolean (value, gst_rtsp_media_is_shared (media));
 257       break;
 258     case PROP_REUSABLE:
 259       g_value_set_boolean (value, gst_rtsp_media_is_reusable (media));
 260       break;
 261     case PROP_PROTOCOLS:
 262       g_value_set_flags (value, gst_rtsp_media_get_protocols (media));
 263       break;
 264     case PROP_EOS_SHUTDOWN:
 265       g_value_set_boolean (value, gst_rtsp_media_is_eos_shutdown (media));
 266       break;
 267     case PROP_BUFFER_SIZE:
 268       g_value_set_uint (value, gst_rtsp_media_get_buffer_size (media));
 269       break;
 270     case PROP_MULTICAST_GROUP:
 271       g_value_take_string (value, gst_rtsp_media_get_multicast_group (media));
 272       break;
 273     default:
 274       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 275   }
 276 }
 277
 278 static void
 279 gst_rtsp_media_set_property (GObject * object, guint propid,
 280     const GValue * value, GParamSpec * pspec)
 281 {
 282   GstRTSPMedia *media = GST_RTSP_MEDIA (object);
 283
 284   switch (propid) {
 285     case PROP_SHARED:
 286       gst_rtsp_media_set_shared (media, g_value_get_boolean (value));
 287       break;
 288     case PROP_REUSABLE:
 289       gst_rtsp_media_set_reusable (media, g_value_get_boolean (value));
 290       break;
 291     case PROP_PROTOCOLS:
 292       gst_rtsp_media_set_protocols (media, g_value_get_flags (value));
 293       break;
 294     case PROP_EOS_SHUTDOWN:
 295       gst_rtsp_media_set_eos_shutdown (media, g_value_get_boolean (value));
 296       break;
 297     case PROP_BUFFER_SIZE:
 298       gst_rtsp_media_set_buffer_size (media, g_value_get_uint (value));
 299       break;
 300     case PROP_MULTICAST_GROUP:
 301       gst_rtsp_media_set_multicast_group (media, g_value_get_string (value));
 302       break;
 303     default:
 304       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 305   }
 306 }
 307
 308 static gpointer
 309 do_loop (GstRTSPMediaClass * klass)
 310 {
 311   GST_INFO ("enter mainloop");
 312   g_main_loop_run (klass->loop);
 313   GST_INFO ("exit mainloop");
 314
 315   return NULL;
 316 }
 317
 318 static void
 319 collect_media_stats (GstRTSPMedia * media)
 320 {
 321   GstFormat format;
 322   gint64 position, duration;
 323
 324   media->range.unit = GST_RTSP_RANGE_NPT;
 325
 326   if (media->is_live) {
 327     media->range.min.type = GST_RTSP_TIME_NOW;
 328     media->range.min.seconds = -1;
 329     media->range.max.type = GST_RTSP_TIME_END;
 330     media->range.max.seconds = -1;
 331   } else {
 332     /* get the position */
 333     format = GST_FORMAT_TIME;
 334     if (!gst_element_query_position (media->pipeline, &format, &position)) {
 335       GST_INFO ("position query failed");
 336       position = 0;
 337     }
 338
 339     /* get the duration */
 340     format = GST_FORMAT_TIME;
 341     if (!gst_element_query_duration (media->pipeline, &format, &duration)) {
 342       GST_INFO ("duration query failed");
 343       duration = -1;
 344     }
 345
 346     GST_INFO ("stats: position %" GST_TIME_FORMAT ", duration %"
 347         GST_TIME_FORMAT, GST_TIME_ARGS (position), GST_TIME_ARGS (duration));
 348
 349     if (position == -1) {
 350       media->range.min.type = GST_RTSP_TIME_NOW;
 351       media->range.min.seconds = -1;
 352     } else {
 353       media->range.min.type = GST_RTSP_TIME_SECONDS;
 354       media->range.min.seconds = ((gdouble) position) / GST_SECOND;
 355     }
 356     if (duration == -1) {
 357       media->range.max.type = GST_RTSP_TIME_END;
 358       media->range.max.seconds = -1;
 359     } else {
 360       media->range.max.type = GST_RTSP_TIME_SECONDS;
 361       media->range.max.seconds = ((gdouble) duration) / GST_SECOND;
 362     }
 363   }
 364 }
 365
 366 /**
 367  * gst_rtsp_media_new:
 368  *
 369  * Create a new #GstRTSPMedia instance. The #GstRTSPMedia object contains the
 370  * element to produde RTP data for one or more related (audio/video/..)
 371  * streams.
 372  *
 373  * Returns: a new #GstRTSPMedia object.
 374  */
 375 GstRTSPMedia *
 376 gst_rtsp_media_new (void)
 377 {
 378   GstRTSPMedia *result;
 379
 380   result = g_object_new (GST_TYPE_RTSP_MEDIA, NULL);
 381
 382   return result;
 383 }
 384
 385 /**
 386  * gst_rtsp_media_set_shared:
 387  * @media: a #GstRTSPMedia
 388  * @shared: the new value
 389  *
 390  * Set or unset if the pipeline for @media can be shared will multiple clients.
 391  * When @shared is %TRUE, client requests for this media will share the media
 392  * pipeline.
 393  */
 394 void
 395 gst_rtsp_media_set_shared (GstRTSPMedia * media, gboolean shared)
 396 {
 397   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 398
 399   media->shared = shared;
 400 }
 401
 402 /**
 403  * gst_rtsp_media_is_shared:
 404  * @media: a #GstRTSPMedia
 405  *
 406  * Check if the pipeline for @media can be shared between multiple clients.
 407  *
 408  * Returns: %TRUE if the media can be shared between clients.
 409  */
 410 gboolean
 411 gst_rtsp_media_is_shared (GstRTSPMedia * media)
 412 {
 413   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 414
 415   return media->shared;
 416 }
 417
 418 /**
 419  * gst_rtsp_media_set_reusable:
 420  * @media: a #GstRTSPMedia
 421  * @reusable: the new value
 422  *
 423  * Set or unset if the pipeline for @media can be reused after the pipeline has
 424  * been unprepared.
 425  */
 426 void
 427 gst_rtsp_media_set_reusable (GstRTSPMedia * media, gboolean reusable)
 428 {
 429   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 430
 431   media->reusable = reusable;
 432 }
 433
 434 /**
 435  * gst_rtsp_media_is_reusable:
 436  * @media: a #GstRTSPMedia
 437  *
 438  * Check if the pipeline for @media can be reused after an unprepare.
 439  *
 440  * Returns: %TRUE if the media can be reused
 441  */
 442 gboolean
 443 gst_rtsp_media_is_reusable (GstRTSPMedia * media)
 444 {
 445   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 446
 447   return media->reusable;
 448 }
 449
 450 /**
 451  * gst_rtsp_media_set_protocols:
 452  * @media: a #GstRTSPMedia
 453  * @protocols: the new flags
 454  *
 455  * Configure the allowed lower transport for @media.
 456  */
 457 void
 458 gst_rtsp_media_set_protocols (GstRTSPMedia * media, GstRTSPLowerTrans protocols)
 459 {
 460   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 461
 462   media->protocols = protocols;
 463 }
 464
 465 /**
 466  * gst_rtsp_media_get_protocols:
 467  * @media: a #GstRTSPMedia
 468  *
 469  * Get the allowed protocols of @media.
 470  *
 471  * Returns: a #GstRTSPLowerTrans
 472  */
 473 GstRTSPLowerTrans
 474 gst_rtsp_media_get_protocols (GstRTSPMedia * media)
 475 {
 476   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media),
 477       GST_RTSP_LOWER_TRANS_UNKNOWN);
 478
 479   return media->protocols;
 480 }
 481
 482 /**
 483  * gst_rtsp_media_set_eos_shutdown:
 484  * @media: a #GstRTSPMedia
 485  * @eos_shutdown: the new value
 486  *
 487  * Set or unset if an EOS event will be sent to the pipeline for @media before
 488  * it is unprepared.
 489  */
 490 void
 491 gst_rtsp_media_set_eos_shutdown (GstRTSPMedia * media, gboolean eos_shutdown)
 492 {
 493   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 494
 495   media->eos_shutdown = eos_shutdown;
 496 }
 497
 498 /**
 499  * gst_rtsp_media_is_eos_shutdown:
 500  * @media: a #GstRTSPMedia
 501  *
 502  * Check if the pipeline for @media will send an EOS down the pipeline before
 503  * unpreparing.
 504  *
 505  * Returns: %TRUE if the media will send EOS before unpreparing.
 506  */
 507 gboolean
 508 gst_rtsp_media_is_eos_shutdown (GstRTSPMedia * media)
 509 {
 510   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 511
 512   return media->eos_shutdown;
 513 }
 514
 515 /**
 516  * gst_rtsp_media_set_buffer_size:
 517  * @media: a #GstRTSPMedia
 518  * @size: the new value
 519  *
 520  * Set the kernel UDP buffer size.
 521  */
 522 void
 523 gst_rtsp_media_set_buffer_size (GstRTSPMedia * media, guint size)
 524 {
 525   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 526
 527   media->buffer_size = size;
 528 }
 529
 530 /**
 531  * gst_rtsp_media_get_buffer_size:
 532  * @media: a #GstRTSPMedia
 533  *
 534  * Get the kernel UDP buffer size.
 535  *
 536  * Returns: the kernel UDP buffer size.
 537  */
 538 guint
 539 gst_rtsp_media_get_buffer_size (GstRTSPMedia * media)
 540 {
 541   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 542
 543   return media->buffer_size;
 544 }
 545
 546 /**
 547  * gst_rtsp_media_set_multicast_group:
 548  * @media: a #GstRTSPMedia
 549  * @mc: the new multicast group
 550  *
 551  * Set the multicast group that media from @media will be streamed to.
 552  */
 553 void
 554 gst_rtsp_media_set_multicast_group (GstRTSPMedia * media, const gchar * mc)
 555 {
 556   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 557
 558   g_mutex_lock (media->lock);
 559   g_free (media->multicast_group);
 560   media->multicast_group = g_strdup (mc);
 561   g_mutex_unlock (media->lock);
 562 }
 563
 564 /**
 565  * gst_rtsp_media_get_multicast_group:
 566  * @media: a #GstRTSPMedia
 567  *
 568  * Get the multicast group that media from @media will be streamed to.
 569  *
 570  * Returns: the multicast group
 571  */
 572 gchar *
 573 gst_rtsp_media_get_multicast_group (GstRTSPMedia * media)
 574 {
 575   gchar *result;
 576
 577   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), NULL);
 578
 579   g_mutex_lock (media->lock);
 580   result = g_strdup (media->multicast_group);
 581   g_mutex_unlock (media->lock);
 582
 583   return result;
 584 }
 585
 586 /**
 587  * gst_rtsp_media_set_auth:
 588  * @media: a #GstRTSPMedia
 589  * @auth: a #GstRTSPAuth
 590  *
 591  * configure @auth to be used as the authentication manager of @media.
 592  */
 593 void
 594 gst_rtsp_media_set_auth (GstRTSPMedia * media, GstRTSPAuth * auth)
 595 {
 596   GstRTSPAuth *old;
 597
 598   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 599
 600   old = media->auth;
 601
 602   if (old != auth) {
 603     if (auth)
 604       g_object_ref (auth);
 605     media->auth = auth;
 606     if (old)
 607       g_object_unref (old);
 608   }
 609 }
 610
 611 /**
 612  * gst_rtsp_media_get_auth:
 613  * @media: a #GstRTSPMedia
 614  *
 615  * Get the #GstRTSPAuth used as the authentication manager of @media.
 616  *
 617  * Returns: the #GstRTSPAuth of @media. g_object_unref() after
 618  * usage.
 619  */
 620 GstRTSPAuth *
 621 gst_rtsp_media_get_auth (GstRTSPMedia * media)
 622 {
 623   GstRTSPAuth *result;
 624
 625   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), NULL);
 626
 627   if ((result = media->auth))
 628     g_object_ref (result);
 629
 630   return result;
 631 }
 632
 633
 634 /**
 635  * gst_rtsp_media_n_streams:
 636  * @media: a #GstRTSPMedia
 637  *
 638  * Get the number of streams in this media.
 639  *
 640  * Returns: The number of streams.
 641  */
 642 guint
 643 gst_rtsp_media_n_streams (GstRTSPMedia * media)
 644 {
 645   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), 0);
 646
 647   return media->streams->len;
 648 }
 649
 650 /**
 651  * gst_rtsp_media_get_stream:
 652  * @media: a #GstRTSPMedia
 653  * @idx: the stream index
 654  *
 655  * Retrieve the stream with index @idx from @media.
 656  *
 657  * Returns: the #GstRTSPMediaStream at index @idx or %NULL when a stream with
 658  * that index did not exist.
 659  */
 660 GstRTSPMediaStream *
 661 gst_rtsp_media_get_stream (GstRTSPMedia * media, guint idx)
 662 {
 663   GstRTSPMediaStream *res;
 664
 665   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), NULL);
 666
 667   if (idx < media->streams->len)
 668     res = g_array_index (media->streams, GstRTSPMediaStream *, idx);
 669   else
 670     res = NULL;
 671
 672   return res;
 673 }
 674
 675 /**
 676  * gst_rtsp_media_get_range_string:
 677  * @media: a #GstRTSPMedia
 678  * @play: for the PLAY request
 679  *
 680  * Get the current range as a string.
 681  *
 682  * Returns: The range as a string, g_free() after usage.
 683  */
 684 gchar *
 685 gst_rtsp_media_get_range_string (GstRTSPMedia * media, gboolean play)
 686 {
 687   gchar *result;
 688   GstRTSPTimeRange range;
 689
 690   /* make copy */
 691   range = media->range;
 692
 693   if (!play && media->active > 0) {
 694     range.min.type = GST_RTSP_TIME_NOW;
 695     range.min.seconds = -1;
 696   }
 697
 698   result = gst_rtsp_range_to_string (&range);
 699
 700   return result;
 701 }
 702
 703 /**
 704  * gst_rtsp_media_seek:
 705  * @media: a #GstRTSPMedia
 706  * @range: a #GstRTSPTimeRange
 707  *
 708  * Seek the pipeline to @range.
 709  *
 710  * Returns: %TRUE on success.
 711  */
 712 gboolean
 713 gst_rtsp_media_seek (GstRTSPMedia * media, GstRTSPTimeRange * range)
 714 {
 715   GstSeekFlags flags;
 716   gboolean res;
 717   gint64 start, stop;
 718   GstSeekType start_type, stop_type;
 719
 720   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 721   g_return_val_if_fail (range != NULL, FALSE);
 722
 723   if (range->unit != GST_RTSP_RANGE_NPT)
 724     goto not_supported;
 725
 726   /* depends on the current playing state of the pipeline. We might need to
 727    * queue this until we get EOS. */
 728   flags = GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE | GST_SEEK_FLAG_KEY_UNIT;
 729
 730   start_type = stop_type = GST_SEEK_TYPE_NONE;
 731
 732   switch (range->min.type) {
 733     case GST_RTSP_TIME_NOW:
 734       start = -1;
 735       break;
 736     case GST_RTSP_TIME_SECONDS:
 737       /* only seek when something changed */
 738       if (media->range.min.seconds == range->min.seconds) {
 739         start = -1;
 740       } else {
 741         start = range->min.seconds * GST_SECOND;
 742         start_type = GST_SEEK_TYPE_SET;
 743       }
 744       break;
 745     case GST_RTSP_TIME_END:
 746     default:
 747       goto weird_type;
 748   }
 749   switch (range->max.type) {
 750     case GST_RTSP_TIME_SECONDS:
 751       /* only seek when something changed */
 752       if (media->range.max.seconds == range->max.seconds) {
 753         stop = -1;
 754       } else {
 755         stop = range->max.seconds * GST_SECOND;
 756         stop_type = GST_SEEK_TYPE_SET;
 757       }
 758       break;
 759     case GST_RTSP_TIME_END:
 760       stop = -1;
 761       stop_type = GST_SEEK_TYPE_SET;
 762       break;
 763     case GST_RTSP_TIME_NOW:
 764     default:
 765       goto weird_type;
 766   }
 767
 768   if (start != -1 || stop != -1) {
 769     GST_INFO ("seeking to %" GST_TIME_FORMAT " - %" GST_TIME_FORMAT,
 770         GST_TIME_ARGS (start), GST_TIME_ARGS (stop));
 771
 772     res = gst_element_seek (media->pipeline, 1.0, GST_FORMAT_TIME,
 773         flags, start_type, start, stop_type, stop);
 774
 775     /* and block for the seek to complete */
 776     GST_INFO ("done seeking %d", res);
 777     gst_element_get_state (media->pipeline, NULL, NULL, -1);
 778     GST_INFO ("prerolled again");
 779
 780     collect_media_stats (media);
 781   } else {
 782     GST_INFO ("no seek needed");
 783     res = TRUE;
 784   }
 785
 786   return res;
 787
 788   /* ERRORS */
 789 not_supported:
 790   {
 791     GST_WARNING ("seek unit %d not supported", range->unit);
 792     return FALSE;
 793   }
 794 weird_type:
 795   {
 796     GST_WARNING ("weird range type %d not supported", range->min.type);
 797     return FALSE;
 798   }
 799 }
 800
 801 /**
 802  * gst_rtsp_media_stream_rtp:
 803  * @stream: a #GstRTSPMediaStream
 804  * @buffer: a #GstBuffer
 805  *
 806  * Handle an RTP buffer for the stream. This method is usually called when a
 807  * message has been received from a client using the TCP transport.
 808  *
 809  * This function takes ownership of @buffer.
 810  *
 811  * Returns: a GstFlowReturn.
 812  */
 813 GstFlowReturn
 814 gst_rtsp_media_stream_rtp (GstRTSPMediaStream * stream, GstBuffer * buffer)
 815 {
 816   GstFlowReturn ret;
 817
 818   ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (stream->appsrc[0]), buffer);
 819
 820   return ret;
 821 }
 822
 823 /**
 824  * gst_rtsp_media_stream_rtcp:
 825  * @stream: a #GstRTSPMediaStream
 826  * @buffer: a #GstBuffer
 827  *
 828  * Handle an RTCP buffer for the stream. This method is usually called when a
 829  * message has been received from a client using the TCP transport.
 830  *
 831  * This function takes ownership of @buffer.
 832  *
 833  * Returns: a GstFlowReturn.
 834  */
 835 GstFlowReturn
 836 gst_rtsp_media_stream_rtcp (GstRTSPMediaStream * stream, GstBuffer * buffer)
 837 {
 838   GstFlowReturn ret;
 839
 840   ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (stream->appsrc[1]), buffer);
 841
 842   return ret;
 843 }
 844
 845 /* Allocate the udp ports and sockets */
 846 static gboolean
 847 alloc_udp_ports (GstRTSPMedia * media, GstRTSPMediaStream * stream)
 848 {
 849   GstStateChangeReturn ret;
 850   GstElement *udpsrc0, *udpsrc1;
 851   GstElement *udpsink0, *udpsink1;
 852   gint tmp_rtp, tmp_rtcp;
 853   guint count;
 854   gint rtpport, rtcpport, sockfd;
 855   const gchar *host;
 856
 857   udpsrc0 = NULL;
 858   udpsrc1 = NULL;
 859   udpsink0 = NULL;
 860   udpsink1 = NULL;
 861   count = 0;
 862
 863   /* Start with random port */
 864   tmp_rtp = 0;
 865
 866   if (media->is_ipv6)
 867     host = "udp://[::0]";
 868   else
 869     host = "udp://0.0.0.0";
 870
 871   /* try to allocate 2 UDP ports, the RTP port should be an even
 872    * number and the RTCP port should be the next (uneven) port */
 873 again:
 874   udpsrc0 = gst_element_make_from_uri (GST_URI_SRC, host, NULL);
 875   if (udpsrc0 == NULL)
 876     goto no_udp_protocol;
 877   g_object_set (G_OBJECT (udpsrc0), "port", tmp_rtp, NULL);
 878
 879   ret = gst_element_set_state (udpsrc0, GST_STATE_PAUSED);
 880   if (ret == GST_STATE_CHANGE_FAILURE) {
 881     if (tmp_rtp != 0) {
 882       tmp_rtp += 2;
 883       if (++count > 20)
 884         goto no_ports;
 885
 886       gst_element_set_state (udpsrc0, GST_STATE_NULL);
 887       gst_object_unref (udpsrc0);
 888
 889       goto again;
 890     }
 891     goto no_udp_protocol;
 892   }
 893
 894   g_object_get (G_OBJECT (udpsrc0), "port", &tmp_rtp, NULL);
 895
 896   /* check if port is even */
 897   if ((tmp_rtp & 1) != 0) {
 898     /* port not even, close and allocate another */
 899     if (++count > 20)
 900       goto no_ports;
 901
 902     gst_element_set_state (udpsrc0, GST_STATE_NULL);
 903     gst_object_unref (udpsrc0);
 904
 905     tmp_rtp++;
 906     goto again;
 907   }
 908
 909   /* allocate port+1 for RTCP now */
 910   udpsrc1 = gst_element_make_from_uri (GST_URI_SRC, host, NULL);
 911   if (udpsrc1 == NULL)
 912     goto no_udp_rtcp_protocol;
 913
 914   /* set port */
 915   tmp_rtcp = tmp_rtp + 1;
 916   g_object_set (G_OBJECT (udpsrc1), "port", tmp_rtcp, NULL);
 917
 918   ret = gst_element_set_state (udpsrc1, GST_STATE_PAUSED);
 919   /* tmp_rtcp port is busy already : retry to make rtp/rtcp pair */
 920   if (ret == GST_STATE_CHANGE_FAILURE) {
 921
 922     if (++count > 20)
 923       goto no_ports;
 924
 925     gst_element_set_state (udpsrc0, GST_STATE_NULL);
 926     gst_object_unref (udpsrc0);
 927
 928     gst_element_set_state (udpsrc1, GST_STATE_NULL);
 929     gst_object_unref (udpsrc1);
 930
 931     tmp_rtp += 2;
 932     goto again;
 933   }
 934
 935   /* all fine, do port check */
 936   g_object_get (G_OBJECT (udpsrc0), "port", &rtpport, NULL);
 937   g_object_get (G_OBJECT (udpsrc1), "port", &rtcpport, NULL);
 938
 939   /* this should not happen... */
 940   if (rtpport != tmp_rtp || rtcpport != tmp_rtcp)
 941     goto port_error;
 942
 943   udpsink0 = gst_element_factory_make ("multiudpsink", NULL);
 944   if (!udpsink0)
 945     goto no_udp_protocol;
 946
 947   g_object_get (G_OBJECT (udpsrc0), "sock", &sockfd, NULL);
 948   g_object_set (G_OBJECT (udpsink0), "sockfd", sockfd, NULL);
 949   g_object_set (G_OBJECT (udpsink0), "closefd", FALSE, NULL);
 950
 951   udpsink1 = gst_element_factory_make ("multiudpsink", NULL);
 952   if (!udpsink1)
 953     goto no_udp_protocol;
 954
 955   if (g_object_class_find_property (G_OBJECT_GET_CLASS (udpsink0),
 956           "send-duplicates")) {
 957     g_object_set (G_OBJECT (udpsink0), "send-duplicates", FALSE, NULL);
 958     g_object_set (G_OBJECT (udpsink1), "send-duplicates", FALSE, NULL);
 959   } else {
 960     g_warning
 961         ("old multiudpsink version found without send-duplicates property");
 962   }
 963
 964   if (g_object_class_find_property (G_OBJECT_GET_CLASS (udpsink0),
 965           "buffer-size")) {
 966     g_object_set (G_OBJECT (udpsink0), "buffer-size", media->buffer_size, NULL);
 967   } else {
 968     GST_WARNING ("multiudpsink version found without buffer-size property");
 969   }
 970
 971   g_object_get (G_OBJECT (udpsrc1), "sock", &sockfd, NULL);
 972   g_object_set (G_OBJECT (udpsink1), "sockfd", sockfd, NULL);
 973   g_object_set (G_OBJECT (udpsink1), "closefd", FALSE, NULL);
 974   g_object_set (G_OBJECT (udpsink1), "sync", FALSE, NULL);
 975   g_object_set (G_OBJECT (udpsink1), "async", FALSE, NULL);
 976
 977   g_object_set (G_OBJECT (udpsink0), "auto-multicast", FALSE, NULL);
 978   g_object_set (G_OBJECT (udpsink0), "loop", FALSE, NULL);
 979   g_object_set (G_OBJECT (udpsink1), "auto-multicast", FALSE, NULL);
 980   g_object_set (G_OBJECT (udpsink1), "loop", FALSE, NULL);
 981
 982   /* we keep these elements, we configure all in configure_transport when the
 983    * server told us to really use the UDP ports. */
 984   stream->udpsrc[0] = udpsrc0;
 985   stream->udpsrc[1] = udpsrc1;
 986   stream->udpsink[0] = udpsink0;
 987   stream->udpsink[1] = udpsink1;
 988   stream->server_port.min = rtpport;
 989   stream->server_port.max = rtcpport;
 990
 991   return TRUE;
 992
 993   /* ERRORS */
 994 no_udp_protocol:
 995   {
 996     goto cleanup;
 997   }
 998 no_ports:
 999   {
1000     goto cleanup;
1001   }
1002 no_udp_rtcp_protocol:
1003   {
1004     goto cleanup;
1005   }
1006 port_error:
1007   {
1008     goto cleanup;
1009   }
1010 cleanup:
1011   {
1012     if (udpsrc0) {
1013       gst_element_set_state (udpsrc0, GST_STATE_NULL);
1014       gst_object_unref (udpsrc0);
1015     }
1016     if (udpsrc1) {
1017       gst_element_set_state (udpsrc1, GST_STATE_NULL);
1018       gst_object_unref (udpsrc1);
1019     }
1020     if (udpsink0) {
1021       gst_element_set_state (udpsink0, GST_STATE_NULL);
1022       gst_object_unref (udpsink0);
1023     }
1024     if (udpsink1) {
1025       gst_element_set_state (udpsink1, GST_STATE_NULL);
1026       gst_object_unref (udpsink1);
1027     }
1028     return FALSE;
1029   }
1030 }
1031
1032 /* executed from streaming thread */
1033 static void
1034 caps_notify (GstPad * pad, GParamSpec * unused, GstRTSPMediaStream * stream)
1035 {
1036   gchar *capsstr;
1037   GstCaps *newcaps, *oldcaps;
1038
1039   if ((newcaps = GST_PAD_CAPS (pad)))
1040     gst_caps_ref (newcaps);
1041
1042   oldcaps = stream->caps;
1043   stream->caps = newcaps;
1044
1045   if (oldcaps)
1046     gst_caps_unref (oldcaps);
1047
1048   capsstr = gst_caps_to_string (newcaps);
1049   GST_INFO ("stream %p received caps %p, %s", stream, newcaps, capsstr);
1050   g_free (capsstr);
1051 }
1052
1053 static void
1054 dump_structure (const GstStructure * s)
1055 {
1056   gchar *sstr;
1057
1058   sstr = gst_structure_to_string (s);
1059   GST_INFO ("structure: %s", sstr);
1060   g_free (sstr);
1061 }
1062
1063 static GstRTSPMediaTrans *
1064 find_transport (GstRTSPMediaStream * stream, const gchar * rtcp_from)
1065 {
1066   GList *walk;
1067   GstRTSPMediaTrans *result = NULL;
1068   const gchar *tmp;
1069   gchar *dest;
1070   guint port;
1071
1072   if (rtcp_from == NULL)
1073     return NULL;
1074
1075   tmp = g_strrstr (rtcp_from, ":");
1076   if (tmp == NULL)
1077     return NULL;
1078
1079   port = atoi (tmp + 1);
1080   dest = g_strndup (rtcp_from, tmp - rtcp_from);
1081
1082   GST_INFO ("finding %s:%d", dest, port);
1083
1084   for (walk = stream->transports; walk; walk = g_list_next (walk)) {
1085     GstRTSPMediaTrans *trans = walk->data;
1086     gint min, max;
1087
1088     min = trans->transport->client_port.min;
1089     max = trans->transport->client_port.max;
1090
1091     if ((strcmp (trans->transport->destination, dest) == 0) && (min == port
1092             || max == port)) {
1093       result = trans;
1094       break;
1095     }
1096   }
1097   g_free (dest);
1098
1099   return result;
1100 }
1101
1102 static void
1103 on_new_ssrc (GObject * session, GObject * source, GstRTSPMediaStream * stream)
1104 {
1105   GstStructure *stats;
1106   GstRTSPMediaTrans *trans;
1107
1108   GST_INFO ("%p: new source %p", stream, source);
1109
1110   /* see if we have a stream to match with the origin of the RTCP packet */
1111   trans = g_object_get_qdata (source, ssrc_stream_map_key);
1112   if (trans == NULL) {
1113     g_object_get (source, "stats", &stats, NULL);
1114     if (stats) {
1115       const gchar *rtcp_from;
1116
1117       dump_structure (stats);
1118
1119       rtcp_from = gst_structure_get_string (stats, "rtcp-from");
1120       if ((trans = find_transport (stream, rtcp_from))) {
1121         GST_INFO ("%p: found transport %p for source  %p", stream, trans,
1122             source);
1123
1124         /* keep ref to the source */
1125         trans->rtpsource = source;
1126
1127         g_object_set_qdata (source, ssrc_stream_map_key, trans);
1128       }
1129       gst_structure_free (stats);
1130     }
1131   } else {
1132     GST_INFO ("%p: source %p for transport %p", stream, source, trans);
1133   }
1134 }
1135
1136 static void
1137 on_ssrc_sdes (GObject * session, GObject * source, GstRTSPMediaStream * stream)
1138 {
1139   GST_INFO ("%p: new SDES %p", stream, source);
1140 }
1141
1142 static void
1143 on_ssrc_active (GObject * session, GObject * source,
1144     GstRTSPMediaStream * stream)
1145 {
1146   GstRTSPMediaTrans *trans;
1147
1148   trans = g_object_get_qdata (source, ssrc_stream_map_key);
1149
1150   GST_INFO ("%p: source %p in transport %p is active", stream, source, trans);
1151
1152   if (trans && trans->keep_alive)
1153     trans->keep_alive (trans->ka_user_data);
1154
1155 #ifdef DUMP_STATS
1156   {
1157     GstStructure *stats;
1158     g_object_get (source, "stats", &stats, NULL);
1159     if (stats) {
1160       dump_structure (stats);
1161       gst_structure_free (stats);
1162     }
1163   }
1164 #endif
1165 }
1166
1167 static void
1168 on_bye_ssrc (GObject * session, GObject * source, GstRTSPMediaStream * stream)
1169 {
1170   GST_INFO ("%p: source %p bye", stream, source);
1171 }
1172
1173 static void
1174 on_bye_timeout (GObject * session, GObject * source,
1175     GstRTSPMediaStream * stream)
1176 {
1177   GstRTSPMediaTrans *trans;
1178
1179   GST_INFO ("%p: source %p bye timeout", stream, source);
1180
1181   if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
1182     trans->rtpsource = NULL;
1183     trans->timeout = TRUE;
1184   }
1185 }
1186
1187 static void
1188 on_timeout (GObject * session, GObject * source, GstRTSPMediaStream * stream)
1189 {
1190   GstRTSPMediaTrans *trans;
1191
1192   GST_INFO ("%p: source %p timeout", stream, source);
1193
1194   if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
1195     trans->rtpsource = NULL;
1196     trans->timeout = TRUE;
1197   }
1198 }
1199
1200 static GstFlowReturn
1201 handle_new_buffer (GstAppSink * sink, gpointer user_data)
1202 {
1203   GList *walk;
1204   GstBuffer *buffer;
1205   GstRTSPMediaStream *stream;
1206
1207   buffer = gst_app_sink_pull_buffer (sink);
1208   if (!buffer)
1209     return GST_FLOW_OK;
1210
1211   stream = (GstRTSPMediaStream *) user_data;
1212
1213   for (walk = stream->transports; walk; walk = g_list_next (walk)) {
1214     GstRTSPMediaTrans *tr = (GstRTSPMediaTrans *) walk->data;
1215
1216     if (GST_ELEMENT_CAST (sink) == stream->appsink[0]) {
1217       if (tr->send_rtp)
1218         tr->send_rtp (buffer, tr->transport->interleaved.min, tr->user_data);
1219     } else {
1220       if (tr->send_rtcp)
1221         tr->send_rtcp (buffer, tr->transport->interleaved.max, tr->user_data);
1222     }
1223   }
1224   gst_buffer_unref (buffer);
1225
1226   return GST_FLOW_OK;
1227 }
1228
1229 static GstFlowReturn
1230 handle_new_buffer_list (GstAppSink * sink, gpointer user_data)
1231 {
1232   GList *walk;
1233   GstBufferList *blist;
1234   GstRTSPMediaStream *stream;
1235
1236   blist = gst_app_sink_pull_buffer_list (sink);
1237   if (!blist)
1238     return GST_FLOW_OK;
1239
1240   stream = (GstRTSPMediaStream *) user_data;
1241
1242   for (walk = stream->transports; walk; walk = g_list_next (walk)) {
1243     GstRTSPMediaTrans *tr = (GstRTSPMediaTrans *) walk->data;
1244
1245     if (GST_ELEMENT_CAST (sink) == stream->appsink[0]) {
1246       if (tr->send_rtp_list)
1247         tr->send_rtp_list (blist, tr->transport->interleaved.min,
1248             tr->user_data);
1249     } else {
1250       if (tr->send_rtcp_list)
1251         tr->send_rtcp_list (blist, tr->transport->interleaved.max,
1252             tr->user_data);
1253     }
1254   }
1255   gst_buffer_list_unref (blist);
1256
1257   return GST_FLOW_OK;
1258 }
1259
1260 static GstAppSinkCallbacks sink_cb = {
1261   NULL,                         /* not interested in EOS */
1262   NULL,                         /* not interested in preroll buffers */
1263   handle_new_buffer,
1264   handle_new_buffer_list
1265 };
1266
1267 /* prepare the pipeline objects to handle @stream in @media */
1268 static gboolean
1269 setup_stream (GstRTSPMediaStream * stream, guint idx, GstRTSPMedia * media)
1270 {
1271   gchar *name;
1272   GstPad *pad, *teepad, *selpad;
1273   GstPadLinkReturn ret;
1274   gint i;
1275
1276   /* allocate udp ports, we will have 4 of them, 2 for receiving RTP/RTCP and 2
1277    * for sending RTP/RTCP. The sender and receiver ports are shared between the
1278    * elements */
1279   if (!alloc_udp_ports (media, stream))
1280     return FALSE;
1281
1282   /* add the ports to the pipeline */
1283   for (i = 0; i < 2; i++) {
1284     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->udpsink[i]);
1285     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->udpsrc[i]);
1286   }
1287
1288   /* create elements for the TCP transfer */
1289   for (i = 0; i < 2; i++) {
1290     stream->appsrc[i] = gst_element_factory_make ("appsrc", NULL);
1291     stream->appsink[i] = gst_element_factory_make ("appsink", NULL);
1292     g_object_set (stream->appsink[i], "async", FALSE, "sync", FALSE, NULL);
1293     g_object_set (stream->appsink[i], "emit-signals", FALSE, NULL);
1294     g_object_set (stream->appsink[i], "preroll-queue-len", 1, NULL);
1295     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->appsink[i]);
1296     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->appsrc[i]);
1297     gst_app_sink_set_callbacks (GST_APP_SINK_CAST (stream->appsink[i]),
1298         &sink_cb, stream, NULL);
1299   }
1300
1301   /* hook up the stream to the RTP session elements. */
1302   name = g_strdup_printf ("send_rtp_sink_%d", idx);
1303   stream->send_rtp_sink = gst_element_get_request_pad (media->rtpbin, name);
1304   g_free (name);
1305   name = g_strdup_printf ("send_rtp_src_%d", idx);
1306   stream->send_rtp_src = gst_element_get_static_pad (media->rtpbin, name);
1307   g_free (name);
1308   name = g_strdup_printf ("send_rtcp_src_%d", idx);
1309   stream->send_rtcp_src = gst_element_get_request_pad (media->rtpbin, name);
1310   g_free (name);
1311   name = g_strdup_printf ("recv_rtcp_sink_%d", idx);
1312   stream->recv_rtcp_sink = gst_element_get_request_pad (media->rtpbin, name);
1313   g_free (name);
1314   name = g_strdup_printf ("recv_rtp_sink_%d", idx);
1315   stream->recv_rtp_sink = gst_element_get_request_pad (media->rtpbin, name);
1316   g_free (name);
1317
1318   /* get the session */
1319   g_signal_emit_by_name (media->rtpbin, "get-internal-session", idx,
1320       &stream->session);
1321
1322   g_signal_connect (stream->session, "on-new-ssrc", (GCallback) on_new_ssrc,
1323       stream);
1324   g_signal_connect (stream->session, "on-ssrc-sdes", (GCallback) on_ssrc_sdes,
1325       stream);
1326   g_signal_connect (stream->session, "on-ssrc-active",
1327       (GCallback) on_ssrc_active, stream);
1328   g_signal_connect (stream->session, "on-bye-ssrc", (GCallback) on_bye_ssrc,
1329       stream);
1330   g_signal_connect (stream->session, "on-bye-timeout",
1331       (GCallback) on_bye_timeout, stream);
1332   g_signal_connect (stream->session, "on-timeout", (GCallback) on_timeout,
1333       stream);
1334
1335   /* link the RTP pad to the session manager */
1336   ret = gst_pad_link (stream->srcpad, stream->send_rtp_sink);
1337   if (ret != GST_PAD_LINK_OK)
1338     goto link_failed;
1339
1340   /* make tee for RTP and link to stream */
1341   stream->tee[0] = gst_element_factory_make ("tee", NULL);
1342   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->tee[0]);
1343
1344   pad = gst_element_get_static_pad (stream->tee[0], "sink");
1345   gst_pad_link (stream->send_rtp_src, pad);
1346   gst_object_unref (pad);
1347
1348   /* link RTP sink, we're pretty sure this will work. */
1349   teepad = gst_element_get_request_pad (stream->tee[0], "src%d");
1350   pad = gst_element_get_static_pad (stream->udpsink[0], "sink");
1351   gst_pad_link (teepad, pad);
1352   gst_object_unref (pad);
1353   gst_object_unref (teepad);
1354
1355   teepad = gst_element_get_request_pad (stream->tee[0], "src%d");
1356   pad = gst_element_get_static_pad (stream->appsink[0], "sink");
1357   gst_pad_link (teepad, pad);
1358   gst_object_unref (pad);
1359   gst_object_unref (teepad);
1360
1361   /* make tee for RTCP */
1362   stream->tee[1] = gst_element_factory_make ("tee", NULL);
1363   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->tee[1]);
1364
1365   pad = gst_element_get_static_pad (stream->tee[1], "sink");
1366   gst_pad_link (stream->send_rtcp_src, pad);
1367   gst_object_unref (pad);
1368
1369   /* link RTCP elements */
1370   teepad = gst_element_get_request_pad (stream->tee[1], "src%d");
1371   pad = gst_element_get_static_pad (stream->udpsink[1], "sink");
1372   gst_pad_link (teepad, pad);
1373   gst_object_unref (pad);
1374   gst_object_unref (teepad);
1375
1376   teepad = gst_element_get_request_pad (stream->tee[1], "src%d");
1377   pad = gst_element_get_static_pad (stream->appsink[1], "sink");
1378   gst_pad_link (teepad, pad);
1379   gst_object_unref (pad);
1380   gst_object_unref (teepad);
1381
1382   /* make selector for the RTP receivers */
1383   stream->selector[0] = gst_element_factory_make ("rtspfunnel", NULL);
1384   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->selector[0]);
1385
1386   pad = gst_element_get_static_pad (stream->selector[0], "src");
1387   gst_pad_link (pad, stream->recv_rtp_sink);
1388   gst_object_unref (pad);
1389
1390   selpad = gst_element_get_request_pad (stream->selector[0], "sink%d");
1391   pad = gst_element_get_static_pad (stream->udpsrc[0], "src");
1392   gst_pad_link (pad, selpad);
1393   gst_object_unref (pad);
1394   gst_object_unref (selpad);
1395
1396   selpad = gst_element_get_request_pad (stream->selector[0], "sink%d");
1397   pad = gst_element_get_static_pad (stream->appsrc[0], "src");
1398   gst_pad_link (pad, selpad);
1399   gst_object_unref (pad);
1400   gst_object_unref (selpad);
1401
1402   /* make selector for the RTCP receivers */
1403   stream->selector[1] = gst_element_factory_make ("rtspfunnel", NULL);
1404   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->selector[1]);
1405
1406   pad = gst_element_get_static_pad (stream->selector[1], "src");
1407   gst_pad_link (pad, stream->recv_rtcp_sink);
1408   gst_object_unref (pad);
1409
1410   selpad = gst_element_get_request_pad (stream->selector[1], "sink%d");
1411   pad = gst_element_get_static_pad (stream->udpsrc[1], "src");
1412   gst_pad_link (pad, selpad);
1413   gst_object_unref (pad);
1414   gst_object_unref (selpad);
1415
1416   selpad = gst_element_get_request_pad (stream->selector[1], "sink%d");
1417   pad = gst_element_get_static_pad (stream->appsrc[1], "src");
1418   gst_pad_link (pad, selpad);
1419   gst_object_unref (pad);
1420   gst_object_unref (selpad);
1421
1422   /* we set and keep these to playing so that they don't cause NO_PREROLL return
1423    * values */
1424   gst_element_set_state (stream->udpsrc[0], GST_STATE_PLAYING);
1425   gst_element_set_state (stream->udpsrc[1], GST_STATE_PLAYING);
1426   gst_element_set_locked_state (stream->udpsrc[0], TRUE);
1427   gst_element_set_locked_state (stream->udpsrc[1], TRUE);
1428
1429   /* be notified of caps changes */
1430   stream->caps_sig = g_signal_connect (stream->send_rtp_sink, "notify::caps",
1431       (GCallback) caps_notify, stream);
1432
1433   stream->prepared = TRUE;
1434
1435   return TRUE;
1436
1437   /* ERRORS */
1438 link_failed:
1439   {
1440     GST_WARNING ("failed to link stream %d", idx);
1441     return FALSE;
1442   }
1443 }
1444
1445 static void
1446 unlock_streams (GstRTSPMedia * media)
1447 {
1448   guint i, n_streams;
1449
1450   /* unlock the udp src elements */
1451   n_streams = gst_rtsp_media_n_streams (media);
1452   for (i = 0; i < n_streams; i++) {
1453     GstRTSPMediaStream *stream;
1454
1455     stream = gst_rtsp_media_get_stream (media, i);
1456
1457     gst_element_set_locked_state (stream->udpsrc[0], FALSE);
1458     gst_element_set_locked_state (stream->udpsrc[1], FALSE);
1459   }
1460 }
1461
1462 static void
1463 gst_rtsp_media_set_status (GstRTSPMedia * media, GstRTSPMediaStatus status)
1464 {
1465   g_mutex_lock (media->lock);
1466   /* never overwrite the error status */
1467   if (media->status != GST_RTSP_MEDIA_STATUS_ERROR)
1468     media->status = status;
1469   GST_DEBUG ("setting new status to %d", status);
1470   g_cond_broadcast (media->cond);
1471   g_mutex_unlock (media->lock);
1472 }
1473
1474 static GstRTSPMediaStatus
1475 gst_rtsp_media_get_status (GstRTSPMedia * media)
1476 {
1477   GstRTSPMediaStatus result;
1478   GTimeVal timeout;
1479
1480   g_mutex_lock (media->lock);
1481   g_get_current_time (&timeout);
1482   g_time_val_add (&timeout, 20 * G_USEC_PER_SEC);
1483   /* while we are preparing, wait */
1484   while (media->status == GST_RTSP_MEDIA_STATUS_PREPARING) {
1485     GST_DEBUG ("waiting for status change");
1486     if (!g_cond_timed_wait (media->cond, media->lock, &timeout)) {
1487       GST_DEBUG ("timeout, assuming error status");
1488       media->status = GST_RTSP_MEDIA_STATUS_ERROR;
1489     }
1490   }
1491   /* could be success or error */
1492   result = media->status;
1493   GST_DEBUG ("got status %d", result);
1494   g_mutex_unlock (media->lock);
1495
1496   return result;
1497 }
1498
1499 static gboolean
1500 default_handle_message (GstRTSPMedia * media, GstMessage * message)
1501 {
1502   GstMessageType type;
1503
1504   type = GST_MESSAGE_TYPE (message);
1505
1506   switch (type) {
1507     case GST_MESSAGE_STATE_CHANGED:
1508       break;
1509     case GST_MESSAGE_BUFFERING:
1510     {
1511       gint percent;
1512
1513       gst_message_parse_buffering (message, &percent);
1514
1515       /* no state management needed for live pipelines */
1516       if (media->is_live)
1517         break;
1518
1519       if (percent == 100) {
1520         /* a 100% message means buffering is done */
1521         media->buffering = FALSE;
1522         /* if the desired state is playing, go back */
1523         if (media->target_state == GST_STATE_PLAYING) {
1524           GST_INFO ("Buffering done, setting pipeline to PLAYING");
1525           gst_element_set_state (media->pipeline, GST_STATE_PLAYING);
1526         } else {
1527           GST_INFO ("Buffering done");
1528         }
1529       } else {
1530         /* buffering busy */
1531         if (media->buffering == FALSE) {
1532           if (media->target_state == GST_STATE_PLAYING) {
1533             /* we were not buffering but PLAYING, PAUSE  the pipeline. */
1534             GST_INFO ("Buffering, setting pipeline to PAUSED ...");
1535             gst_element_set_state (media->pipeline, GST_STATE_PAUSED);
1536           } else {
1537             GST_INFO ("Buffering ...");
1538           }
1539         }
1540         media->buffering = TRUE;
1541       }
1542       break;
1543     }
1544     case GST_MESSAGE_LATENCY:
1545     {
1546       gst_bin_recalculate_latency (GST_BIN_CAST (media->pipeline));
1547       break;
1548     }
1549     case GST_MESSAGE_ERROR:
1550     {
1551       GError *gerror;
1552       gchar *debug;
1553
1554       gst_message_parse_error (message, &gerror, &debug);
1555       GST_WARNING ("%p: got error %s (%s)", media, gerror->message, debug);
1556       g_error_free (gerror);
1557       g_free (debug);
1558
1559       gst_rtsp_media_set_status (media, GST_RTSP_MEDIA_STATUS_ERROR);
1560       break;
1561     }
1562     case GST_MESSAGE_WARNING:
1563     {
1564       GError *gerror;
1565       gchar *debug;
1566
1567       gst_message_parse_warning (message, &gerror, &debug);
1568       GST_WARNING ("%p: got warning %s (%s)", media, gerror->message, debug);
1569       g_error_free (gerror);
1570       g_free (debug);
1571       break;
1572     }
1573     case GST_MESSAGE_ELEMENT:
1574       break;
1575     case GST_MESSAGE_STREAM_STATUS:
1576       break;
1577     case GST_MESSAGE_ASYNC_DONE:
1578       if (!media->adding) {
1579         /* when we are dynamically adding pads, the addition of the udpsrc will
1580          * temporarily produce ASYNC_DONE messages. We have to ignore them and
1581          * wait for the final ASYNC_DONE after everything prerolled */
1582         GST_INFO ("%p: got ASYNC_DONE", media);
1583         collect_media_stats (media);
1584
1585         gst_rtsp_media_set_status (media, GST_RTSP_MEDIA_STATUS_PREPARED);
1586       } else {
1587         GST_INFO ("%p: ignoring ASYNC_DONE", media);
1588       }
1589       break;
1590     case GST_MESSAGE_EOS:
1591       GST_INFO ("%p: got EOS", media);
1592       if (media->eos_pending) {
1593         GST_DEBUG ("shutting down after EOS");
1594         gst_element_set_state (media->pipeline, GST_STATE_NULL);
1595         media->eos_pending = FALSE;
1596         g_object_unref (media);
1597       }
1598       break;
1599     default:
1600       GST_INFO ("%p: got message type %s", media,
1601           gst_message_type_get_name (type));
1602       break;
1603   }
1604   return TRUE;
1605 }
1606
1607 static gboolean
1608 bus_message (GstBus * bus, GstMessage * message, GstRTSPMedia * media)
1609 {
1610   GstRTSPMediaClass *klass;
1611   gboolean ret;
1612
1613   klass = GST_RTSP_MEDIA_GET_CLASS (media);
1614
1615   if (klass->handle_message)
1616     ret = klass->handle_message (media, message);
1617   else
1618     ret = FALSE;
1619
1620   return ret;
1621 }
1622
1623 /* called from streaming threads */
1624 static void
1625 pad_added_cb (GstElement * element, GstPad * pad, GstRTSPMedia * media)
1626 {
1627   GstRTSPMediaStream *stream;
1628   gchar *name;
1629   gint i;
1630
1631   i = media->streams->len + 1;
1632
1633   GST_INFO ("pad added %s:%s, stream %d", GST_DEBUG_PAD_NAME (pad), i);
1634
1635   stream = g_new0 (GstRTSPMediaStream, 1);
1636   stream->payloader = element;
1637
1638   name = g_strdup_printf ("dynpay%d", i);
1639
1640   media->adding = TRUE;
1641
1642   /* ghost the pad of the payloader to the element */
1643   stream->srcpad = gst_ghost_pad_new (name, pad);
1644   gst_pad_set_active (stream->srcpad, TRUE);
1645   gst_element_add_pad (media->element, stream->srcpad);
1646   g_free (name);
1647
1648   /* add stream now */
1649   g_array_append_val (media->streams, stream);
1650
1651   setup_stream (stream, i, media);
1652
1653   for (i = 0; i < 2; i++) {
1654     gst_element_set_state (stream->udpsink[i], GST_STATE_PAUSED);
1655     gst_element_set_state (stream->appsink[i], GST_STATE_PAUSED);
1656     gst_element_set_state (stream->tee[i], GST_STATE_PAUSED);
1657     gst_element_set_state (stream->selector[i], GST_STATE_PAUSED);
1658     gst_element_set_state (stream->appsrc[i], GST_STATE_PAUSED);
1659   }
1660   media->adding = FALSE;
1661 }
1662
1663 static void
1664 no_more_pads_cb (GstElement * element, GstRTSPMedia * media)
1665 {
1666   GST_INFO ("no more pads");
1667   if (media->fakesink) {
1668     gst_object_ref (media->fakesink);
1669     gst_bin_remove (GST_BIN (media->pipeline), media->fakesink);
1670     gst_element_set_state (media->fakesink, GST_STATE_NULL);
1671     gst_object_unref (media->fakesink);
1672     media->fakesink = NULL;
1673     GST_INFO ("removed fakesink");
1674   }
1675 }
1676
1677 /**
1678  * gst_rtsp_media_prepare:
1679  * @media: a #GstRTSPMedia
1680  *
1681  * Prepare @media for streaming. This function will create the pipeline and
1682  * other objects to manage the streaming.
1683  *
1684  * It will preroll the pipeline and collect vital information about the streams
1685  * such as the duration.
1686  *
1687  * Returns: %TRUE on success.
1688  */
1689 gboolean
1690 gst_rtsp_media_prepare (GstRTSPMedia * media)
1691 {
1692   GstStateChangeReturn ret;
1693   GstRTSPMediaStatus status;
1694   guint i, n_streams;
1695   GstRTSPMediaClass *klass;
1696   GstBus *bus;
1697   GList *walk;
1698
1699   if (media->status == GST_RTSP_MEDIA_STATUS_PREPARED)
1700     goto was_prepared;
1701
1702   if (!media->reusable && media->reused)
1703     goto is_reused;
1704
1705   GST_INFO ("preparing media %p", media);
1706
1707   /* reset some variables */
1708   media->is_live = FALSE;
1709   media->buffering = FALSE;
1710   /* we're preparing now */
1711   media->status = GST_RTSP_MEDIA_STATUS_PREPARING;
1712
1713   bus = gst_pipeline_get_bus (GST_PIPELINE_CAST (media->pipeline));
1714
1715   /* add the pipeline bus to our custom mainloop */
1716   media->source = gst_bus_create_watch (bus);
1717   gst_object_unref (bus);
1718
1719   g_source_set_callback (media->source, (GSourceFunc) bus_message, media, NULL);
1720
1721   klass = GST_RTSP_MEDIA_GET_CLASS (media);
1722   media->id = g_source_attach (media->source, klass->context);
1723
1724   media->rtpbin = gst_element_factory_make ("gstrtpbin", NULL);
1725
1726   /* add stuff to the bin */
1727   gst_bin_add (GST_BIN (media->pipeline), media->rtpbin);
1728
1729   /* link streams we already have, other streams might appear when we have
1730    * dynamic elements */
1731   n_streams = gst_rtsp_media_n_streams (media);
1732   for (i = 0; i < n_streams; i++) {
1733     GstRTSPMediaStream *stream;
1734
1735     stream = gst_rtsp_media_get_stream (media, i);
1736
1737     setup_stream (stream, i, media);
1738   }
1739
1740   for (walk = media->dynamic; walk; walk = g_list_next (walk)) {
1741     GstElement *elem = walk->data;
1742
1743     GST_INFO ("adding callbacks for dynamic element %p", elem);
1744
1745     g_signal_connect (elem, "pad-added", (GCallback) pad_added_cb, media);
1746     g_signal_connect (elem, "no-more-pads", (GCallback) no_more_pads_cb, media);
1747
1748     /* we add a fakesink here in order to make the state change async. We remove
1749      * the fakesink again in the no-more-pads callback. */
1750     media->fakesink = gst_element_factory_make ("fakesink", "fakesink");
1751     gst_bin_add (GST_BIN (media->pipeline), media->fakesink);
1752   }
1753
1754   GST_INFO ("setting pipeline to PAUSED for media %p", media);
1755   /* first go to PAUSED */
1756   ret = gst_element_set_state (media->pipeline, GST_STATE_PAUSED);
1757   media->target_state = GST_STATE_PAUSED;
1758
1759   switch (ret) {
1760     case GST_STATE_CHANGE_SUCCESS:
1761       GST_INFO ("SUCCESS state change for media %p", media);
1762       break;
1763     case GST_STATE_CHANGE_ASYNC:
1764       GST_INFO ("ASYNC state change for media %p", media);
1765       break;
1766     case GST_STATE_CHANGE_NO_PREROLL:
1767       /* we need to go to PLAYING */
1768       GST_INFO ("NO_PREROLL state change: live media %p", media);
1769       media->is_live = TRUE;
1770       ret = gst_element_set_state (media->pipeline, GST_STATE_PLAYING);
1771       if (ret == GST_STATE_CHANGE_FAILURE)
1772         goto state_failed;
1773       break;
1774     case GST_STATE_CHANGE_FAILURE:
1775       goto state_failed;
1776   }
1777
1778   /* now wait for all pads to be prerolled */
1779   status = gst_rtsp_media_get_status (media);
1780   if (status == GST_RTSP_MEDIA_STATUS_ERROR)
1781     goto state_failed;
1782
1783   g_signal_emit (media, gst_rtsp_media_signals[SIGNAL_PREPARED], 0, NULL);
1784
1785   GST_INFO ("object %p is prerolled", media);
1786
1787   return TRUE;
1788
1789   /* OK */
1790 was_prepared:
1791   {
1792     return TRUE;
1793   }
1794   /* ERRORS */
1795 is_reused:
1796   {
1797     GST_WARNING ("can not reuse media %p", media);
1798     return FALSE;
1799   }
1800 state_failed:
1801   {
1802     GST_WARNING ("failed to preroll pipeline");
1803     unlock_streams (media);
1804     gst_element_set_state (media->pipeline, GST_STATE_NULL);
1805     gst_rtsp_media_unprepare (media);
1806     return FALSE;
1807   }
1808 }
1809
1810 /**
1811  * gst_rtsp_media_unprepare:
1812  * @media: a #GstRTSPMedia
1813  *
1814  * Unprepare @media. After this call, the media should be prepared again before
1815  * it can be used again. If the media is set to be non-reusable, a new instance
1816  * must be created.
1817  *
1818  * Returns: %TRUE on success.
1819  */
1820 gboolean
1821 gst_rtsp_media_unprepare (GstRTSPMedia * media)
1822 {
1823   GstRTSPMediaClass *klass;
1824   gboolean success;
1825
1826   if (media->status == GST_RTSP_MEDIA_STATUS_UNPREPARED)
1827     return TRUE;
1828
1829   GST_INFO ("unprepare media %p", media);
1830   media->target_state = GST_STATE_NULL;
1831
1832   klass = GST_RTSP_MEDIA_GET_CLASS (media);
1833   if (klass->unprepare)
1834     success = klass->unprepare (media);
1835   else
1836     success = TRUE;
1837
1838   media->status = GST_RTSP_MEDIA_STATUS_UNPREPARED;
1839   media->reused = TRUE;
1840
1841   /* when the media is not reusable, this will effectively unref the media and
1842    * recreate it */
1843   g_signal_emit (media, gst_rtsp_media_signals[SIGNAL_UNPREPARED], 0, NULL);
1844
1845   return success;
1846 }
1847
1848 static gboolean
1849 default_unprepare (GstRTSPMedia * media)
1850 {
1851   if (media->eos_shutdown) {
1852     GST_DEBUG ("sending EOS for shutdown");
1853     /* ref so that we don't disappear */
1854     g_object_ref (media);
1855     media->eos_pending = TRUE;
1856     gst_element_send_event (media->pipeline, gst_event_new_eos ());
1857     /* we need to go to playing again for the EOS to propagate, normally in this
1858      * state, nothing is receiving data from us anymore so this is ok. */
1859     gst_element_set_state (media->pipeline, GST_STATE_PLAYING);
1860   } else {
1861     GST_DEBUG ("shutting down");
1862     gst_element_set_state (media->pipeline, GST_STATE_NULL);
1863   }
1864   return TRUE;
1865 }
1866
1867 static void
1868 add_udp_destination (GstRTSPMedia * media, GstRTSPMediaStream * stream,
1869     gchar * dest, gint min, gint max)
1870 {
1871   GST_INFO ("adding %s:%d-%d", dest, min, max);
1872   g_signal_emit_by_name (stream->udpsink[0], "add", dest, min, NULL);
1873   g_signal_emit_by_name (stream->udpsink[1], "add", dest, max, NULL);
1874 }
1875
1876 static void
1877 remove_udp_destination (GstRTSPMedia * media, GstRTSPMediaStream * stream,
1878     gchar * dest, gint min, gint max)
1879 {
1880   GST_INFO ("removing %s:%d-%d", dest, min, max);
1881   g_signal_emit_by_name (stream->udpsink[0], "remove", dest, min, NULL);
1882   g_signal_emit_by_name (stream->udpsink[1], "remove", dest, max, NULL);
1883 }
1884
1885 /**
1886  * gst_rtsp_media_set_state:
1887  * @media: a #GstRTSPMedia
1888  * @state: the target state of the media
1889  * @transports: a #GArray of #GstRTSPMediaTrans pointers
1890  *
1891  * Set the state of @media to @state and for the transports in @transports.
1892  *
1893  * Returns: %TRUE on success.
1894  */
1895 gboolean
1896 gst_rtsp_media_set_state (GstRTSPMedia * media, GstState state,
1897     GArray * transports)
1898 {
1899   gint i;
1900   gboolean add, remove, do_state;
1901   gint old_active;
1902
1903   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
1904   g_return_val_if_fail (transports != NULL, FALSE);
1905
1906   /* NULL and READY are the same */
1907   if (state == GST_STATE_READY)
1908     state = GST_STATE_NULL;
1909
1910   add = remove = FALSE;
1911
1912   GST_INFO ("going to state %s media %p", gst_element_state_get_name (state),
1913       media);
1914
1915   switch (state) {
1916     case GST_STATE_NULL:
1917       /* unlock the streams so that they follow the state changes from now on */
1918       unlock_streams (media);
1919       /* fallthrough */
1920     case GST_STATE_PAUSED:
1921       /* we're going from PLAYING to PAUSED, READY or NULL, remove */
1922       if (media->target_state == GST_STATE_PLAYING)
1923         remove = TRUE;
1924       break;
1925     case GST_STATE_PLAYING:
1926       /* we're going to PLAYING, add */
1927       add = TRUE;
1928       break;
1929     default:
1930       break;
1931   }
1932   old_active = media->active;
1933
1934   for (i = 0; i < transports->len; i++) {
1935     GstRTSPMediaTrans *tr;
1936     GstRTSPMediaStream *stream;
1937     GstRTSPTransport *trans;
1938
1939     /* we need a non-NULL entry in the array */
1940     tr = g_array_index (transports, GstRTSPMediaTrans *, i);
1941     if (tr == NULL)
1942       continue;
1943
1944     /* we need a transport */
1945     if (!(trans = tr->transport))
1946       continue;
1947
1948     /* get the stream and add the destinations */
1949     stream = gst_rtsp_media_get_stream (media, tr->idx);
1950     switch (trans->lower_transport) {
1951       case GST_RTSP_LOWER_TRANS_UDP:
1952       case GST_RTSP_LOWER_TRANS_UDP_MCAST:
1953       {
1954         gchar *dest;
1955         gint min, max;
1956
1957         dest = trans->destination;
1958         if (trans->lower_transport == GST_RTSP_LOWER_TRANS_UDP_MCAST) {
1959           min = trans->port.min;
1960           max = trans->port.max;
1961         } else {
1962           min = trans->client_port.min;
1963           max = trans->client_port.max;
1964         }
1965
1966         if (add && !tr->active) {
1967           add_udp_destination (media, stream, dest, min, max);
1968           stream->transports = g_list_prepend (stream->transports, tr);
1969           tr->active = TRUE;
1970           media->active++;
1971         } else if (remove && tr->active) {
1972           remove_udp_destination (media, stream, dest, min, max);
1973           stream->transports = g_list_remove (stream->transports, tr);
1974           tr->active = FALSE;
1975           media->active--;
1976         }
1977         break;
1978       }
1979       case GST_RTSP_LOWER_TRANS_TCP:
1980         if (add && !tr->active) {
1981           GST_INFO ("adding TCP %s", trans->destination);
1982           stream->transports = g_list_prepend (stream->transports, tr);
1983           tr->active = TRUE;
1984           media->active++;
1985         } else if (remove && tr->active) {
1986           GST_INFO ("removing TCP %s", trans->destination);
1987           stream->transports = g_list_remove (stream->transports, tr);
1988           tr->active = FALSE;
1989           media->active--;
1990         }
1991         break;
1992       default:
1993         GST_INFO ("Unknown transport %d", trans->lower_transport);
1994         break;
1995     }
1996   }
1997
1998   /* we just added the first media, do the playing state change */
1999   if (old_active == 0 && add)
2000     do_state = TRUE;
2001   /* if we have no more active media, do the downward state changes */
2002   else if (media->active == 0)
2003     do_state = TRUE;
2004   else
2005     do_state = FALSE;
2006
2007   GST_INFO ("state %d active %d media %p do_state %d", state, media->active,
2008       media, do_state);
2009
2010   if (media->target_state != state) {
2011     if (do_state) {
2012       if (state == GST_STATE_NULL) {
2013         gst_rtsp_media_unprepare (media);
2014       } else {
2015         GST_INFO ("state %s media %p", gst_element_state_get_name (state),
2016             media);
2017         media->target_state = state;
2018         gst_element_set_state (media->pipeline, state);
2019       }
2020     }
2021     g_signal_emit (media, gst_rtsp_media_signals[SIGNAL_NEW_STATE], 0, state,
2022         NULL);
2023   }
2024
2025   /* remember where we are */
2026   if (state == GST_STATE_PAUSED || old_active != media->active)
2027     collect_media_stats (media);
2028
2029   return TRUE;
2030 }
2031
2032 /**
2033  * gst_rtsp_media_remove_elements:
2034  * @media: a #GstRTSPMedia
2035  *
2036  * Remove all elements and the pipeline controlled by @media.
2037  */
2038 void
2039 gst_rtsp_media_remove_elements (GstRTSPMedia * media)
2040 {
2041   gint i, j;
2042
2043   unlock_streams (media);
2044
2045   for (i = 0; i < media->streams->len; i++) {
2046     GstRTSPMediaStream *stream;
2047
2048     GST_INFO ("Removing elements of stream %d from pipeline", i);
2049
2050     stream = g_array_index (media->streams, GstRTSPMediaStream *, i);
2051
2052     gst_pad_unlink (stream->srcpad, stream->send_rtp_sink);
2053
2054     g_signal_handler_disconnect (stream->send_rtp_sink, stream->caps_sig);
2055
2056     for (j = 0; j < 2; j++) {
2057       gst_element_set_state (stream->udpsrc[j], GST_STATE_NULL);
2058       gst_element_set_state (stream->udpsink[j], GST_STATE_NULL);
2059       gst_element_set_state (stream->appsrc[j], GST_STATE_NULL);
2060       gst_element_set_state (stream->appsink[j], GST_STATE_NULL);
2061       gst_element_set_state (stream->tee[j], GST_STATE_NULL);
2062       gst_element_set_state (stream->selector[j], GST_STATE_NULL);
2063
2064       gst_bin_remove (GST_BIN (media->pipeline), stream->udpsrc[j]);
2065       gst_bin_remove (GST_BIN (media->pipeline), stream->udpsink[j]);
2066       gst_bin_remove (GST_BIN (media->pipeline), stream->appsrc[j]);
2067       gst_bin_remove (GST_BIN (media->pipeline), stream->appsink[j]);
2068       gst_bin_remove (GST_BIN (media->pipeline), stream->tee[j]);
2069       gst_bin_remove (GST_BIN (media->pipeline), stream->selector[j]);
2070     }
2071     if (stream->caps)
2072       gst_caps_unref (stream->caps);
2073     stream->caps = NULL;
2074     gst_rtsp_media_stream_free (stream);
2075   }
2076   g_array_remove_range (media->streams, 0, media->streams->len);
2077
2078   gst_element_set_state (media->rtpbin, GST_STATE_NULL);
2079   gst_bin_remove (GST_BIN (media->pipeline), media->rtpbin);
2080
2081   gst_object_unref (media->pipeline);
2082   media->pipeline = NULL;
2083 }