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
  22 #include <gst/app/gstappsrc.h>
  23 #include <gst/app/gstappsink.h>
  24
  25 #include "rtsp-media.h"
  26
  27 #define DEFAULT_SHARED         FALSE
  28 #define DEFAULT_REUSABLE       FALSE
  29
  30 enum
  31 {
  32   PROP_0,
  33   PROP_SHARED,
  34   PROP_REUSABLE,
  35   PROP_LAST
  36 };
  37
  38 enum
  39 {
  40   SIGNAL_UNPREPARED,
  41   SIGNAL_LAST
  42 };
  43
  44 static GQuark ssrc_stream_map_key;
  45
  46 static void gst_rtsp_media_get_property (GObject *object, guint propid,
  47     GValue *value, GParamSpec *pspec);
  48 static void gst_rtsp_media_set_property (GObject *object, guint propid,
  49     const GValue *value, GParamSpec *pspec);
  50 static void gst_rtsp_media_finalize (GObject * obj);
  51
  52 static gpointer do_loop (GstRTSPMediaClass *klass);
  53 static gboolean default_handle_message (GstRTSPMedia *media, GstMessage *message);
  54 static void unlock_streams (GstRTSPMedia *media);
  55
  56 static guint gst_rtsp_media_signals[SIGNAL_LAST] = { 0 };
  57
  58 G_DEFINE_TYPE (GstRTSPMedia, gst_rtsp_media, G_TYPE_OBJECT);
  59
  60 static void
  61 gst_rtsp_media_class_init (GstRTSPMediaClass * klass)
  62 {
  63   GObjectClass *gobject_class;
  64   GError *error = NULL;
  65
  66   gobject_class = G_OBJECT_CLASS (klass);
  67
  68   gobject_class->get_property = gst_rtsp_media_get_property;
  69   gobject_class->set_property = gst_rtsp_media_set_property;
  70   gobject_class->finalize = gst_rtsp_media_finalize;
  71
  72   g_object_class_install_property (gobject_class, PROP_SHARED,
  73       g_param_spec_boolean ("shared", "Shared", "If this media pipeline can be shared",
  74           DEFAULT_SHARED, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  75
  76   g_object_class_install_property (gobject_class, PROP_REUSABLE,
  77       g_param_spec_boolean ("reusable", "Reusable",
  78           "If this media pipeline can be reused after an unprepare",
  79           DEFAULT_REUSABLE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  80
  81   gst_rtsp_media_signals[SIGNAL_UNPREPARED] =
  82       g_signal_new ("unprepared", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_LAST,
  83       G_STRUCT_OFFSET (GstRTSPMediaClass, unprepared), NULL, NULL,
  84       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0, G_TYPE_NONE);
  85
  86   klass->context = g_main_context_new ();
  87   klass->loop = g_main_loop_new (klass->context, TRUE);
  88
  89   klass->thread = g_thread_create ((GThreadFunc) do_loop, klass, TRUE, &error);
  90   if (error != NULL) {
  91     g_critical ("could not start bus thread: %s", error->message);
  92   }
  93   klass->handle_message = default_handle_message;
  94
  95   ssrc_stream_map_key = g_quark_from_static_string ("GstRTSPServer.stream");
  96 }
  97
  98 static void
  99 gst_rtsp_media_init (GstRTSPMedia * media)
 100 {
 101   media->streams = g_array_new (FALSE, TRUE, sizeof (GstRTSPMediaStream *));
 102   media->is_live = FALSE;
 103   media->buffering = FALSE;
 104 }
 105
 106 static void
 107 gst_rtsp_media_stream_free (GstRTSPMediaStream *stream)
 108 {
 109   if (stream->session)
 110     g_object_unref (stream->session);
 111
 112   if (stream->caps)
 113     gst_caps_unref (stream->caps);
 114
 115   g_list_free (stream->transports);
 116
 117   g_free (stream);
 118 }
 119
 120 static void
 121 gst_rtsp_media_finalize (GObject * obj)
 122 {
 123   GstRTSPMedia *media;
 124   guint i;
 125
 126   media = GST_RTSP_MEDIA (obj);
 127
 128   g_message ("finalize media %p", media);
 129
 130   if (media->pipeline) {
 131     unlock_streams (media);
 132     gst_element_set_state (media->pipeline, GST_STATE_NULL);
 133     gst_object_unref (media->pipeline);
 134   }
 135
 136   for (i = 0; i < media->streams->len; i++) {
 137     GstRTSPMediaStream *stream;
 138
 139     stream = g_array_index (media->streams, GstRTSPMediaStream *, i);
 140
 141     gst_rtsp_media_stream_free (stream);
 142   }
 143   g_array_free (media->streams, TRUE);
 144
 145   g_list_foreach (media->dynamic, (GFunc) gst_object_unref, NULL);
 146   g_list_free (media->dynamic);
 147
 148   if (media->source) {
 149     g_source_destroy (media->source);
 150     g_source_unref (media->source);
 151   }
 152
 153   G_OBJECT_CLASS (gst_rtsp_media_parent_class)->finalize (obj);
 154 }
 155
 156 static void
 157 gst_rtsp_media_get_property (GObject *object, guint propid,
 158     GValue *value, GParamSpec *pspec)
 159 {
 160   GstRTSPMedia *media = GST_RTSP_MEDIA (object);
 161
 162   switch (propid) {
 163     case PROP_SHARED:
 164       g_value_set_boolean (value, gst_rtsp_media_is_shared (media));
 165       break;
 166     case PROP_REUSABLE:
 167       g_value_set_boolean (value, gst_rtsp_media_is_reusable (media));
 168       break;
 169     default:
 170       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 171   }
 172 }
 173
 174 static void
 175 gst_rtsp_media_set_property (GObject *object, guint propid,
 176     const GValue *value, GParamSpec *pspec)
 177 {
 178   GstRTSPMedia *media = GST_RTSP_MEDIA (object);
 179
 180   switch (propid) {
 181     case PROP_SHARED:
 182       gst_rtsp_media_set_shared (media, g_value_get_boolean (value));
 183       break;
 184     case PROP_REUSABLE:
 185       gst_rtsp_media_set_reusable (media, g_value_get_boolean (value));
 186       break;
 187     default:
 188       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 189   }
 190 }
 191
 192 static gpointer
 193 do_loop (GstRTSPMediaClass *klass)
 194 {
 195   g_message ("enter mainloop");
 196   g_main_loop_run (klass->loop);
 197   g_message ("exit mainloop");
 198
 199   return NULL;
 200 }
 201
 202 static void
 203 collect_media_stats (GstRTSPMedia *media)
 204 {
 205   GstFormat format;
 206   gint64 position, duration;
 207
 208   media->range.unit = GST_RTSP_RANGE_NPT;
 209
 210   if (media->is_live) {
 211     media->range.min.type = GST_RTSP_TIME_NOW;
 212     media->range.min.seconds = -1;
 213     media->range.max.type = GST_RTSP_TIME_END;
 214     media->range.max.seconds = -1;
 215   }
 216   else {
 217     /* get the position */
 218     format = GST_FORMAT_TIME;
 219     if (!gst_element_query_position (media->pipeline, &format, &position)) {
 220       g_message ("position query failed");
 221       position = 0;
 222     }
 223
 224     /* get the duration */
 225     format = GST_FORMAT_TIME;
 226     if (!gst_element_query_duration (media->pipeline, &format, &duration)) {
 227       g_message ("duration query failed");
 228       duration = -1;
 229     }
 230
 231     g_message ("stats: position %"GST_TIME_FORMAT", duration %"GST_TIME_FORMAT,
 232         GST_TIME_ARGS (position), GST_TIME_ARGS (duration));
 233
 234     if (position == -1) {
 235       media->range.min.type = GST_RTSP_TIME_NOW;
 236       media->range.min.seconds = -1;
 237     }
 238     else {
 239       media->range.min.type = GST_RTSP_TIME_SECONDS;
 240       media->range.min.seconds = ((gdouble)position) / GST_SECOND;
 241     }
 242     if (duration == -1) {
 243       media->range.max.type = GST_RTSP_TIME_END;
 244       media->range.max.seconds = -1;
 245     }
 246     else {
 247       media->range.max.type = GST_RTSP_TIME_SECONDS;
 248       media->range.max.seconds = ((gdouble)duration) / GST_SECOND;
 249     }
 250   }
 251 }
 252
 253 /**
 254  * gst_rtsp_media_new:
 255  *
 256  * Create a new #GstRTSPMedia instance. The #GstRTSPMedia object contains the
 257  * element to produde RTP data for one or more related (audio/video/..)
 258  * streams.
 259  *
 260  * Returns: a new #GstRTSPMedia object.
 261  */
 262 GstRTSPMedia *
 263 gst_rtsp_media_new (void)
 264 {
 265   GstRTSPMedia *result;
 266
 267   result = g_object_new (GST_TYPE_RTSP_MEDIA, NULL);
 268
 269   return result;
 270 }
 271
 272 /**
 273  * gst_rtsp_media_set_shared:
 274  * @media: a #GstRTSPMedia
 275  * @shared: the new value
 276  *
 277  * Set or unset if the pipeline for @media can be shared will multiple clients.
 278  * When @shared is %TRUE, client requests for this media will share the media
 279  * pipeline.
 280  */
 281 void
 282 gst_rtsp_media_set_shared (GstRTSPMedia *media, gboolean shared)
 283 {
 284   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 285
 286   media->shared = shared;
 287 }
 288
 289 /**
 290  * gst_rtsp_media_is_shared:
 291  * @media: a #GstRTSPMedia
 292  *
 293  * Check if the pipeline for @media can be shared between multiple clients.
 294  *
 295  * Returns: %TRUE if the media can be shared between clients.
 296  */
 297 gboolean
 298 gst_rtsp_media_is_shared (GstRTSPMedia *media)
 299 {
 300   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 301
 302   return media->shared;
 303 }
 304
 305 /**
 306  * gst_rtsp_media_set_reusable:
 307  * @media: a #GstRTSPMedia
 308  * @reusable: the new value
 309  *
 310  * Set or unset if the pipeline for @media can be reused after the pipeline has
 311  * been unprepared.
 312  */
 313 void
 314 gst_rtsp_media_set_reusable (GstRTSPMedia *media, gboolean reusable)
 315 {
 316   g_return_if_fail (GST_IS_RTSP_MEDIA (media));
 317
 318   media->reusable = reusable;
 319 }
 320
 321 /**
 322  * gst_rtsp_media_is_reusable:
 323  * @media: a #GstRTSPMedia
 324  *
 325  * Check if the pipeline for @media can be reused after an unprepare.
 326  *
 327  * Returns: %TRUE if the media can be reused
 328  */
 329 gboolean
 330 gst_rtsp_media_is_reusable (GstRTSPMedia *media)
 331 {
 332   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 333
 334   return media->reusable;
 335 }
 336
 337 /**
 338  * gst_rtsp_media_n_streams:
 339  * @media: a #GstRTSPMedia
 340  *
 341  * Get the number of streams in this media.
 342  *
 343  * Returns: The number of streams.
 344  */
 345 guint
 346 gst_rtsp_media_n_streams (GstRTSPMedia *media)
 347 {
 348   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), 0);
 349
 350   return media->streams->len;
 351 }
 352
 353 /**
 354  * gst_rtsp_media_get_stream:
 355  * @media: a #GstRTSPMedia
 356  * @idx: the stream index
 357  *
 358  * Retrieve the stream with index @idx from @media.
 359  *
 360  * Returns: the #GstRTSPMediaStream at index @idx or %NULL when a stream with
 361  * that index did not exist.
 362  */
 363 GstRTSPMediaStream *
 364 gst_rtsp_media_get_stream (GstRTSPMedia *media, guint idx)
 365 {
 366   GstRTSPMediaStream *res;
 367
 368   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), NULL);
 369
 370   if (idx < media->streams->len)
 371     res = g_array_index (media->streams, GstRTSPMediaStream *, idx);
 372   else
 373     res = NULL;
 374
 375   return res;
 376 }
 377
 378 /**
 379  * gst_rtsp_media_seek:
 380  * @stream: a #GstRTSPMediaStream
 381  * @range: a #GstRTSPTimeRange
 382  *
 383  * Seek the pipeline to @range.
 384  *
 385  * Returns: %TRUE on success.
 386  */
 387 gboolean
 388 gst_rtsp_media_seek (GstRTSPMedia *media, GstRTSPTimeRange *range)
 389 {
 390   GstSeekFlags flags;
 391   gboolean res;
 392   gint64 start, stop;
 393   GstSeekType start_type, stop_type;
 394
 395   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
 396   g_return_val_if_fail (range != NULL, FALSE);
 397
 398   if (range->unit != GST_RTSP_RANGE_NPT)
 399     goto not_supported;
 400
 401   /* depends on the current playing state of the pipeline. We might need to
 402    * queue this until we get EOS. */
 403   flags = GST_SEEK_FLAG_FLUSH | GST_SEEK_FLAG_ACCURATE | GST_SEEK_FLAG_KEY_UNIT;
 404
 405   start_type = stop_type = GST_SEEK_TYPE_NONE;
 406
 407   switch (range->min.type) {
 408     case GST_RTSP_TIME_NOW:
 409       start = -1;
 410       break;
 411     case GST_RTSP_TIME_SECONDS:
 412       /* only seek when something changed */
 413       if (media->range.min.seconds == range->min.seconds) {
 414         start = -1;
 415       } else {
 416         start = range->min.seconds * GST_SECOND;
 417         start_type = GST_SEEK_TYPE_SET;
 418       }
 419       break;
 420     case GST_RTSP_TIME_END:
 421     default:
 422       goto weird_type;
 423   }
 424   switch (range->max.type) {
 425     case GST_RTSP_TIME_SECONDS:
 426       /* only seek when something changed */
 427       if (media->range.max.seconds == range->max.seconds) {
 428         stop = -1;
 429       } else {
 430         stop = range->max.seconds * GST_SECOND;
 431         stop_type = GST_SEEK_TYPE_SET;
 432       }
 433       break;
 434     case GST_RTSP_TIME_END:
 435       stop = -1;
 436       stop_type = GST_SEEK_TYPE_SET;
 437       break;
 438     case GST_RTSP_TIME_NOW:
 439     default:
 440       goto weird_type;
 441   }
 442
 443   if (start != -1 || stop != -1) {
 444     g_message ("seeking to %"GST_TIME_FORMAT" - %"GST_TIME_FORMAT,
 445                 GST_TIME_ARGS (start), GST_TIME_ARGS (stop));
 446
 447     res = gst_element_seek (media->pipeline, 1.0, GST_FORMAT_TIME,
 448         flags, start_type, start, stop_type, stop);
 449
 450     /* and block for the seek to complete */
 451     g_message ("done seeking %d", res);
 452     gst_element_get_state (media->pipeline, NULL, NULL, -1);
 453     g_message ("prerolled again");
 454
 455     collect_media_stats (media);
 456   }
 457   else {
 458     g_message ("no seek needed");
 459     res = TRUE;
 460   }
 461
 462   return res;
 463
 464   /* ERRORS */
 465 not_supported:
 466   {
 467     g_warning ("seek unit %d not supported", range->unit);
 468     return FALSE;
 469   }
 470 weird_type:
 471   {
 472     g_warning ("weird range type %d not supported", range->min.type);
 473     return FALSE;
 474   }
 475 }
 476
 477 /**
 478  * gst_rtsp_media_stream_rtp:
 479  * @stream: a #GstRTSPMediaStream
 480  * @buffer: a #GstBuffer
 481  *
 482  * Handle an RTP buffer for the stream. This method is usually called when a
 483  * message has been received from a client using the TCP transport.
 484  *
 485  * This function takes ownership of @buffer.
 486  *
 487  * Returns: a GstFlowReturn.
 488  */
 489 GstFlowReturn
 490 gst_rtsp_media_stream_rtp (GstRTSPMediaStream *stream, GstBuffer *buffer)
 491 {
 492   GstFlowReturn ret;
 493
 494   ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (stream->appsrc[0]), buffer);
 495
 496   return ret;
 497 }
 498
 499 /**
 500  * gst_rtsp_media_stream_rtcp:
 501  * @stream: a #GstRTSPMediaStream
 502  * @buffer: a #GstBuffer
 503  *
 504  * Handle an RTCP buffer for the stream. This method is usually called when a
 505  * message has been received from a client using the TCP transport.
 506  *
 507  * This function takes ownership of @buffer.
 508  *
 509  * Returns: a GstFlowReturn.
 510  */
 511 GstFlowReturn
 512 gst_rtsp_media_stream_rtcp (GstRTSPMediaStream *stream, GstBuffer *buffer)
 513 {
 514   GstFlowReturn ret;
 515
 516   ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (stream->appsrc[1]), buffer);
 517
 518   return ret;
 519 }
 520
 521 /* Allocate the udp ports and sockets */
 522 static gboolean
 523 alloc_udp_ports (GstRTSPMediaStream * stream)
 524 {
 525   GstStateChangeReturn ret;
 526   GstElement *udpsrc0, *udpsrc1;
 527   GstElement *udpsink0, *udpsink1;
 528   gint tmp_rtp, tmp_rtcp;
 529   guint count;
 530   gint rtpport, rtcpport, sockfd;
 531
 532   udpsrc0 = NULL;
 533   udpsrc1 = NULL;
 534   udpsink0 = NULL;
 535   udpsink1 = NULL;
 536   count = 0;
 537
 538   /* Start with random port */
 539   tmp_rtp = 0;
 540
 541   /* try to allocate 2 UDP ports, the RTP port should be an even
 542    * number and the RTCP port should be the next (uneven) port */
 543 again:
 544   udpsrc0 = gst_element_make_from_uri (GST_URI_SRC, "udp://0.0.0.0", NULL);
 545   if (udpsrc0 == NULL)
 546     goto no_udp_protocol;
 547   g_object_set (G_OBJECT (udpsrc0), "port", tmp_rtp, NULL);
 548
 549   ret = gst_element_set_state (udpsrc0, GST_STATE_PAUSED);
 550   if (ret == GST_STATE_CHANGE_FAILURE) {
 551     if (tmp_rtp != 0) {
 552       tmp_rtp += 2;
 553       if (++count > 20)
 554         goto no_ports;
 555
 556       gst_element_set_state (udpsrc0, GST_STATE_NULL);
 557       gst_object_unref (udpsrc0);
 558
 559       goto again;
 560     }
 561     goto no_udp_protocol;
 562   }
 563
 564   g_object_get (G_OBJECT (udpsrc0), "port", &tmp_rtp, NULL);
 565
 566   /* check if port is even */
 567   if ((tmp_rtp & 1) != 0) {
 568     /* port not even, close and allocate another */
 569     if (++count > 20)
 570       goto no_ports;
 571
 572     gst_element_set_state (udpsrc0, GST_STATE_NULL);
 573     gst_object_unref (udpsrc0);
 574
 575     tmp_rtp++;
 576     goto again;
 577   }
 578
 579   /* allocate port+1 for RTCP now */
 580   udpsrc1 = gst_element_make_from_uri (GST_URI_SRC, "udp://0.0.0.0", NULL);
 581   if (udpsrc1 == NULL)
 582     goto no_udp_rtcp_protocol;
 583
 584   /* set port */
 585   tmp_rtcp = tmp_rtp + 1;
 586   g_object_set (G_OBJECT (udpsrc1), "port", tmp_rtcp, NULL);
 587
 588   ret = gst_element_set_state (udpsrc1, GST_STATE_PAUSED);
 589   /* tmp_rtcp port is busy already : retry to make rtp/rtcp pair */
 590   if (ret == GST_STATE_CHANGE_FAILURE) {
 591
 592     if (++count > 20)
 593       goto no_ports;
 594
 595     gst_element_set_state (udpsrc0, GST_STATE_NULL);
 596     gst_object_unref (udpsrc0);
 597
 598     gst_element_set_state (udpsrc1, GST_STATE_NULL);
 599     gst_object_unref (udpsrc1);
 600
 601     tmp_rtp += 2;
 602     goto again;
 603   }
 604
 605   /* all fine, do port check */
 606   g_object_get (G_OBJECT (udpsrc0), "port", &rtpport, NULL);
 607   g_object_get (G_OBJECT (udpsrc1), "port", &rtcpport, NULL);
 608
 609   /* this should not happen... */
 610   if (rtpport != tmp_rtp || rtcpport != tmp_rtcp)
 611     goto port_error;
 612
 613   udpsink0 = gst_element_factory_make ("multiudpsink", NULL);
 614   if (!udpsink0)
 615     goto no_udp_protocol;
 616
 617   g_object_get (G_OBJECT (udpsrc0), "sock", &sockfd, NULL);
 618   g_object_set (G_OBJECT (udpsink0), "sockfd", sockfd, NULL);
 619   g_object_set (G_OBJECT (udpsink0), "closefd", FALSE, NULL);
 620
 621   udpsink1 = gst_element_factory_make ("multiudpsink", NULL);
 622   if (!udpsink1)
 623     goto no_udp_protocol;
 624
 625   g_object_get (G_OBJECT (udpsrc1), "sock", &sockfd, NULL);
 626   g_object_set (G_OBJECT (udpsink1), "sockfd", sockfd, NULL);
 627   g_object_set (G_OBJECT (udpsink1), "closefd", FALSE, NULL);
 628   g_object_set (G_OBJECT (udpsink1), "sync", FALSE, NULL);
 629   g_object_set (G_OBJECT (udpsink1), "async", FALSE, NULL);
 630
 631   /* we keep these elements, we configure all in configure_transport when the
 632    * server told us to really use the UDP ports. */
 633   stream->udpsrc[0] = udpsrc0;
 634   stream->udpsrc[1] = udpsrc1;
 635   stream->udpsink[0] = udpsink0;
 636   stream->udpsink[1] = udpsink1;
 637   stream->server_port.min = rtpport;
 638   stream->server_port.max = rtcpport;
 639
 640   return TRUE;
 641
 642   /* ERRORS */
 643 no_udp_protocol:
 644   {
 645     goto cleanup;
 646   }
 647 no_ports:
 648   {
 649     goto cleanup;
 650   }
 651 no_udp_rtcp_protocol:
 652   {
 653     goto cleanup;
 654   }
 655 port_error:
 656   {
 657     goto cleanup;
 658   }
 659 cleanup:
 660   {
 661     if (udpsrc0) {
 662       gst_element_set_state (udpsrc0, GST_STATE_NULL);
 663       gst_object_unref (udpsrc0);
 664     }
 665     if (udpsrc1) {
 666       gst_element_set_state (udpsrc1, GST_STATE_NULL);
 667       gst_object_unref (udpsrc1);
 668     }
 669     if (udpsink0) {
 670       gst_element_set_state (udpsink0, GST_STATE_NULL);
 671       gst_object_unref (udpsink0);
 672     }
 673     if (udpsink1) {
 674       gst_element_set_state (udpsink1, GST_STATE_NULL);
 675       gst_object_unref (udpsink1);
 676     }
 677     return FALSE;
 678   }
 679 }
 680
 681 static void
 682 caps_notify (GstPad * pad, GParamSpec * unused, GstRTSPMediaStream * stream)
 683 {
 684   gchar *capsstr;
 685   GstCaps *newcaps, *oldcaps;
 686
 687   if ((newcaps = GST_PAD_CAPS (pad)))
 688     gst_caps_ref (newcaps);
 689
 690   oldcaps = stream->caps;
 691   stream->caps = newcaps;
 692
 693   if (oldcaps)
 694     gst_caps_unref (oldcaps);
 695
 696   capsstr = gst_caps_to_string (newcaps);
 697   g_message ("stream %p received caps %s", stream, capsstr);
 698   g_free (capsstr);
 699 }
 700
 701 static void
 702 dump_structure (const GstStructure *s)
 703 {
 704   gchar *sstr;
 705
 706   sstr = gst_structure_to_string (s);
 707   g_message ("structure: %s", sstr);
 708   g_free (sstr);
 709 }
 710
 711 static GstRTSPMediaTrans *
 712 find_transport (GstRTSPMediaStream *stream, const gchar *rtcp_from)
 713 {
 714   GList *walk;
 715   GstRTSPMediaTrans *result = NULL;
 716   const gchar *tmp;
 717   gchar *dest;
 718   guint port;
 719
 720   if (rtcp_from == NULL)
 721     return NULL;
 722
 723   tmp = g_strrstr (rtcp_from, ":");
 724   if (tmp == NULL)
 725     return NULL;
 726
 727   port = atoi (tmp + 1);
 728   dest = g_strndup (rtcp_from, tmp - rtcp_from);
 729
 730   g_message ("finding %s:%d", dest, port);
 731
 732   for (walk = stream->transports; walk; walk = g_list_next (walk)) {
 733     GstRTSPMediaTrans *trans = walk->data;
 734     gint min, max;
 735
 736     min = trans->transport->client_port.min;
 737     max = trans->transport->client_port.max;
 738
 739     if ((strcmp (trans->transport->destination, dest) == 0) && (min == port || max == port)) {
 740       result = trans;
 741       break;
 742     }
 743   }
 744   g_free (dest);
 745
 746   return result;
 747 }
 748
 749 static void
 750 on_new_ssrc (GObject *session, GObject *source, GstRTSPMediaStream *stream)
 751 {
 752   GstStructure *stats;
 753   GstRTSPMediaTrans *trans;
 754
 755   g_message ("%p: new source %p", stream, source);
 756
 757   /* see if we have a stream to match with the origin of the RTCP packet */
 758   trans = g_object_get_qdata (source, ssrc_stream_map_key);
 759   if (trans == NULL) {
 760     g_object_get (source, "stats", &stats, NULL);
 761     if (stats) {
 762       const gchar *rtcp_from;
 763
 764       dump_structure (stats);
 765
 766       rtcp_from = gst_structure_get_string (stats, "rtcp-from");
 767       if ((trans = find_transport (stream, rtcp_from))) {
 768         g_message ("%p: found transport %p for source  %p", stream, trans, source);
 769
 770         /* keep ref to the source */
 771         trans->rtpsource = source;
 772
 773         g_object_set_qdata (source, ssrc_stream_map_key, trans);
 774       }
 775       gst_structure_free (stats);
 776     }
 777   } else {
 778     g_message ("%p: source %p for transport %p", stream, source, trans);
 779   }
 780 }
 781
 782 static void
 783 on_ssrc_sdes (GObject *session, GObject *source, GstRTSPMediaStream *stream)
 784 {
 785   g_message ("%p: new SDES %p", stream, source);
 786 }
 787
 788 static void
 789 on_ssrc_active (GObject *session, GObject *source, GstRTSPMediaStream *stream)
 790 {
 791   GstRTSPMediaTrans *trans;
 792
 793   trans = g_object_get_qdata (source, ssrc_stream_map_key);
 794
 795   g_message ("%p: source %p in transport %p is active", stream, trans, source);
 796
 797   if (trans && trans->keep_alive)
 798     trans->keep_alive (trans->ka_user_data);
 799 }
 800
 801 static void
 802 on_bye_ssrc (GObject *session, GObject *source, GstRTSPMediaStream *stream)
 803 {
 804   g_message ("%p: source %p bye", stream, source);
 805 }
 806
 807 static void
 808 on_bye_timeout (GObject *session, GObject *source, GstRTSPMediaStream *stream)
 809 {
 810   GstRTSPMediaTrans *trans;
 811
 812   g_message ("%p: source %p bye timeout", stream, source);
 813
 814   if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
 815     trans->rtpsource = NULL;
 816     trans->timeout = TRUE;
 817   }
 818 }
 819
 820 static void
 821 on_timeout (GObject *session, GObject *source, GstRTSPMediaStream *stream)
 822 {
 823   GstRTSPMediaTrans *trans;
 824
 825   g_message ("%p: source %p timeout", stream, source);
 826
 827   if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
 828     trans->rtpsource = NULL;
 829     trans->timeout = TRUE;
 830   }
 831 }
 832
 833 static GstFlowReturn
 834 handle_new_buffer (GstAppSink *sink, gpointer user_data)
 835 {
 836   GList *walk;
 837   GstBuffer *buffer;
 838   GstRTSPMediaStream *stream;
 839
 840   buffer = gst_app_sink_pull_buffer (sink);
 841   if (!buffer)
 842     return GST_FLOW_OK;
 843
 844   stream = (GstRTSPMediaStream *) user_data;
 845
 846   for (walk = stream->transports; walk; walk = g_list_next (walk)) {
 847     GstRTSPMediaTrans *tr = (GstRTSPMediaTrans *) walk->data;
 848
 849     if (GST_ELEMENT_CAST (sink) == stream->appsink[0]) {
 850       if (tr->send_rtp)
 851         tr->send_rtp (buffer, tr->transport->interleaved.min, tr->user_data);
 852     }
 853     else {
 854       if (tr->send_rtcp)
 855         tr->send_rtcp (buffer, tr->transport->interleaved.max, tr->user_data);
 856     }
 857   }
 858   gst_buffer_unref (buffer);
 859
 860   return GST_FLOW_OK;
 861 }
 862
 863 static GstAppSinkCallbacks sink_cb = {
 864   NULL,  /* not interested in EOS */
 865   NULL,  /* not interested in preroll buffers */
 866   handle_new_buffer
 867 };
 868
 869 /* prepare the pipeline objects to handle @stream in @media */
 870 static gboolean
 871 setup_stream (GstRTSPMediaStream *stream, guint idx, GstRTSPMedia *media)
 872 {
 873   gchar *name;
 874   GstPad *pad, *teepad, *selpad;
 875   GstPadLinkReturn ret;
 876   gint i;
 877
 878   /* allocate udp ports, we will have 4 of them, 2 for receiving RTP/RTCP and 2
 879    * for sending RTP/RTCP. The sender and receiver ports are shared between the
 880    * elements */
 881   if (!alloc_udp_ports (stream))
 882     return FALSE;
 883
 884   /* add the ports to the pipeline */
 885   for (i = 0; i < 2; i++) {
 886     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->udpsink[i]);
 887     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->udpsrc[i]);
 888   }
 889
 890   /* create elements for the TCP transfer */
 891   for (i = 0; i < 2; i++) {
 892     stream->appsrc[i] = gst_element_factory_make ("appsrc", NULL);
 893     stream->appsink[i] = gst_element_factory_make ("appsink", NULL);
 894     g_object_set (stream->appsink[i], "async", FALSE, "sync", FALSE, NULL);
 895     g_object_set (stream->appsink[i], "emit-signals", FALSE, NULL);
 896     g_object_set (stream->appsink[i], "preroll-queue-len", 1, NULL);
 897     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->appsink[i]);
 898     gst_bin_add (GST_BIN_CAST (media->pipeline), stream->appsrc[i]);
 899     gst_app_sink_set_callbacks (GST_APP_SINK_CAST (stream->appsink[i]),
 900                   &sink_cb, stream, NULL);
 901   }
 902
 903   /* hook up the stream to the RTP session elements. */
 904   name = g_strdup_printf ("send_rtp_sink_%d", idx);
 905   stream->send_rtp_sink = gst_element_get_request_pad (media->rtpbin, name);
 906   g_free (name);
 907   name = g_strdup_printf ("send_rtp_src_%d", idx);
 908   stream->send_rtp_src = gst_element_get_static_pad (media->rtpbin, name);
 909   g_free (name);
 910   name = g_strdup_printf ("send_rtcp_src_%d", idx);
 911   stream->send_rtcp_src = gst_element_get_request_pad (media->rtpbin, name);
 912   g_free (name);
 913   name = g_strdup_printf ("recv_rtcp_sink_%d", idx);
 914   stream->recv_rtcp_sink = gst_element_get_request_pad (media->rtpbin, name);
 915   g_free (name);
 916   name = g_strdup_printf ("recv_rtp_sink_%d", idx);
 917   stream->recv_rtp_sink = gst_element_get_request_pad (media->rtpbin, name);
 918   g_free (name);
 919
 920   /* get the session */
 921   g_signal_emit_by_name (media->rtpbin, "get-internal-session", idx,
 922                   &stream->session);
 923
 924   g_signal_connect (stream->session, "on-new-ssrc", (GCallback) on_new_ssrc,
 925       stream);
 926   g_signal_connect (stream->session, "on-ssrc-sdes", (GCallback) on_ssrc_sdes,
 927       stream);
 928   g_signal_connect (stream->session, "on-ssrc-active", (GCallback) on_ssrc_active,
 929       stream);
 930   g_signal_connect (stream->session, "on-bye-ssrc", (GCallback) on_bye_ssrc,
 931       stream);
 932   g_signal_connect (stream->session, "on-bye-timeout", (GCallback) on_bye_timeout,
 933       stream);
 934   g_signal_connect (stream->session, "on-timeout", (GCallback) on_timeout,
 935       stream);
 936
 937   /* link the RTP pad to the session manager */
 938   ret = gst_pad_link (stream->srcpad, stream->send_rtp_sink);
 939   if (ret != GST_PAD_LINK_OK)
 940     goto link_failed;
 941
 942   /* make tee for RTP and link to stream */
 943   stream->tee[0] = gst_element_factory_make ("tee", NULL);
 944   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->tee[0]);
 945
 946   pad = gst_element_get_static_pad (stream->tee[0], "sink");
 947   gst_pad_link (stream->send_rtp_src, pad);
 948   gst_object_unref (pad);
 949
 950   /* link RTP sink, we're pretty sure this will work. */
 951   teepad = gst_element_get_request_pad (stream->tee[0], "src%d");
 952   pad = gst_element_get_static_pad (stream->udpsink[0], "sink");
 953   gst_pad_link (teepad, pad);
 954   gst_object_unref (pad);
 955   gst_object_unref (teepad);
 956
 957   teepad = gst_element_get_request_pad (stream->tee[0], "src%d");
 958   pad = gst_element_get_static_pad (stream->appsink[0], "sink");
 959   gst_pad_link (teepad, pad);
 960   gst_object_unref (pad);
 961   gst_object_unref (teepad);
 962
 963   /* make tee for RTCP */
 964   stream->tee[1] = gst_element_factory_make ("tee", NULL);
 965   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->tee[1]);
 966
 967   pad = gst_element_get_static_pad (stream->tee[1], "sink");
 968   gst_pad_link (stream->send_rtcp_src, pad);
 969   gst_object_unref (pad);
 970
 971   /* link RTCP elements */
 972   teepad = gst_element_get_request_pad (stream->tee[1], "src%d");
 973   pad = gst_element_get_static_pad (stream->udpsink[1], "sink");
 974   gst_pad_link (teepad, pad);
 975   gst_object_unref (pad);
 976   gst_object_unref (teepad);
 977
 978   teepad = gst_element_get_request_pad (stream->tee[1], "src%d");
 979   pad = gst_element_get_static_pad (stream->appsink[1], "sink");
 980   gst_pad_link (teepad, pad);
 981   gst_object_unref (pad);
 982   gst_object_unref (teepad);
 983
 984   /* make selector for the RTP receivers */
 985   stream->selector[0] = gst_element_factory_make ("input-selector", NULL);
 986   g_object_set (stream->selector[0], "select-all", TRUE, NULL);
 987   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->selector[0]);
 988
 989   pad = gst_element_get_static_pad (stream->selector[0], "src");
 990   gst_pad_link (pad, stream->recv_rtp_sink);
 991   gst_object_unref (pad);
 992
 993   selpad = gst_element_get_request_pad (stream->selector[0], "sink%d");
 994   pad = gst_element_get_static_pad (stream->udpsrc[0], "src");
 995   gst_pad_link (pad, selpad);
 996   gst_object_unref (pad);
 997   gst_object_unref (selpad);
 998
 999   selpad = gst_element_get_request_pad (stream->selector[0], "sink%d");
1000   pad = gst_element_get_static_pad (stream->appsrc[0], "src");
1001   gst_pad_link (pad, selpad);
1002   gst_object_unref (pad);
1003   gst_object_unref (selpad);
1004
1005   /* make selector for the RTCP receivers */
1006   stream->selector[1] = gst_element_factory_make ("input-selector", NULL);
1007   g_object_set (stream->selector[1], "select-all", TRUE, NULL);
1008   gst_bin_add (GST_BIN_CAST (media->pipeline), stream->selector[1]);
1009
1010   pad = gst_element_get_static_pad (stream->selector[1], "src");
1011   gst_pad_link (pad, stream->recv_rtcp_sink);
1012   gst_object_unref (pad);
1013
1014   selpad = gst_element_get_request_pad (stream->selector[1], "sink%d");
1015   pad = gst_element_get_static_pad (stream->udpsrc[1], "src");
1016   gst_pad_link (pad, selpad);
1017   gst_object_unref (pad);
1018   gst_object_unref (selpad);
1019
1020   selpad = gst_element_get_request_pad (stream->selector[1], "sink%d");
1021   pad = gst_element_get_static_pad (stream->appsrc[1], "src");
1022   gst_pad_link (pad, selpad);
1023   gst_object_unref (pad);
1024   gst_object_unref (selpad);
1025
1026   /* we set and keep these to playing so that they don't cause NO_PREROLL return
1027    * values */
1028   gst_element_set_state (stream->udpsrc[0], GST_STATE_PLAYING);
1029   gst_element_set_state (stream->udpsrc[1], GST_STATE_PLAYING);
1030   gst_element_set_locked_state (stream->udpsrc[0], TRUE);
1031   gst_element_set_locked_state (stream->udpsrc[1], TRUE);
1032
1033   /* be notified of caps changes */
1034   stream->caps_sig = g_signal_connect (stream->send_rtp_sink, "notify::caps",
1035                   (GCallback) caps_notify, stream);
1036
1037   stream->prepared = TRUE;
1038
1039   return TRUE;
1040
1041   /* ERRORS */
1042 link_failed:
1043   {
1044     g_warning ("failed to link stream %d", idx);
1045     return FALSE;
1046   }
1047 }
1048
1049 static void
1050 unlock_streams (GstRTSPMedia *media)
1051 {
1052   guint i, n_streams;
1053
1054   /* unlock the udp src elements */
1055   n_streams = gst_rtsp_media_n_streams (media);
1056   for (i = 0; i < n_streams; i++) {
1057     GstRTSPMediaStream *stream;
1058
1059     stream = gst_rtsp_media_get_stream (media, i);
1060
1061     gst_element_set_locked_state (stream->udpsrc[0], FALSE);
1062     gst_element_set_locked_state (stream->udpsrc[1], FALSE);
1063   }
1064 }
1065
1066 static gboolean
1067 default_handle_message (GstRTSPMedia *media, GstMessage *message)
1068 {
1069   GstMessageType type;
1070
1071   type = GST_MESSAGE_TYPE (message);
1072
1073   switch (type) {
1074     case GST_MESSAGE_STATE_CHANGED:
1075       break;
1076     case GST_MESSAGE_BUFFERING:
1077     {
1078       gint percent;
1079
1080       gst_message_parse_buffering (message, &percent);
1081
1082       /* no state management needed for live pipelines */
1083       if (media->is_live)
1084         break;
1085
1086       if (percent == 100) {
1087         /* a 100% message means buffering is done */
1088         media->buffering = FALSE;
1089         /* if the desired state is playing, go back */
1090         if (media->target_state == GST_STATE_PLAYING) {
1091           g_message ("Buffering done, setting pipeline to PLAYING");
1092           gst_element_set_state (media->pipeline, GST_STATE_PLAYING);
1093         }
1094         else {
1095           g_message ("Buffering done");
1096         }
1097       } else {
1098         /* buffering busy */
1099         if (media->buffering == FALSE) {
1100           if (media->target_state == GST_STATE_PLAYING) {
1101             /* we were not buffering but PLAYING, PAUSE  the pipeline. */
1102             g_message ("Buffering, setting pipeline to PAUSED ...");
1103             gst_element_set_state (media->pipeline, GST_STATE_PAUSED);
1104           }
1105           else {
1106             g_message ("Buffering ...");
1107           }
1108         }
1109         media->buffering = TRUE;
1110       }
1111       break;
1112     }
1113     case GST_MESSAGE_LATENCY:
1114     {
1115       gst_bin_recalculate_latency (GST_BIN_CAST (media->pipeline));
1116       break;
1117     }
1118     case GST_MESSAGE_ERROR:
1119     {
1120       GError *gerror;
1121       gchar *debug;
1122
1123       gst_message_parse_error (message, &gerror, &debug);
1124       g_warning ("%p: got error %s (%s)", media, gerror->message, debug);
1125       g_error_free (gerror);
1126       g_free (debug);
1127       break;
1128     }
1129     case GST_MESSAGE_WARNING:
1130     {
1131       GError *gerror;
1132       gchar *debug;
1133
1134       gst_message_parse_warning (message, &gerror, &debug);
1135       g_warning ("%p: got warning %s (%s)", media, gerror->message, debug);
1136       g_error_free (gerror);
1137       g_free (debug);
1138       break;
1139     }
1140     case GST_MESSAGE_ELEMENT:
1141     {
1142       break;
1143     }
1144     default:
1145       g_message ("%p: got message type %s", media, gst_message_type_get_name (type));
1146       break;
1147   }
1148   return TRUE;
1149 }
1150
1151 static gboolean
1152 bus_message (GstBus *bus, GstMessage *message, GstRTSPMedia *media)
1153 {
1154   GstRTSPMediaClass *klass;
1155   gboolean ret;
1156
1157   klass = GST_RTSP_MEDIA_GET_CLASS (media);
1158
1159   if (klass->handle_message)
1160     ret = klass->handle_message (media, message);
1161   else
1162     ret = FALSE;
1163
1164   return ret;
1165 }
1166
1167 static void
1168 pad_added_cb (GstElement *element, GstPad *pad, GstRTSPMedia *media)
1169 {
1170   GstRTSPMediaStream *stream;
1171   gchar *name;
1172   gint i;
1173
1174   i = media->streams->len + 1;
1175
1176   g_message ("pad added %s:%s, stream %d", GST_DEBUG_PAD_NAME (pad), i);
1177
1178   stream = g_new0 (GstRTSPMediaStream, 1);
1179   stream->payloader = element;
1180
1181   name = g_strdup_printf ("dynpay%d", i);
1182
1183   /* ghost the pad of the payloader to the element */
1184   stream->srcpad = gst_ghost_pad_new (name, pad);
1185   gst_pad_set_active (stream->srcpad, TRUE);
1186   gst_element_add_pad (media->element, stream->srcpad);
1187   g_free (name);
1188
1189   /* add stream now */
1190   g_array_append_val (media->streams, stream);
1191
1192   setup_stream (stream, i, media);
1193
1194   for (i = 0; i < 2; i++) {
1195     gst_element_set_state (stream->udpsink[i], GST_STATE_PAUSED);
1196     gst_element_set_state (stream->appsink[i], GST_STATE_PAUSED);
1197     gst_element_set_state (stream->tee[i], GST_STATE_PAUSED);
1198     gst_element_set_state (stream->selector[i], GST_STATE_PAUSED);
1199     gst_element_set_state (stream->appsrc[i], GST_STATE_PAUSED);
1200   }
1201 }
1202
1203 static void
1204 no_more_pads_cb (GstElement *element, GstRTSPMedia *media)
1205 {
1206   g_message ("no more pads");
1207   if (media->fakesink) {
1208     gst_object_ref (media->fakesink);
1209     gst_bin_remove (GST_BIN (media->pipeline), media->fakesink);
1210     gst_element_set_state (media->fakesink, GST_STATE_NULL);
1211     gst_object_unref (media->fakesink);
1212     media->fakesink = NULL;
1213     g_message ("removed fakesink");
1214   }
1215 }
1216
1217 /**
1218  * gst_rtsp_media_prepare:
1219  * @obj: a #GstRTSPMedia
1220  *
1221  * Prepare @media for streaming. This function will create the pipeline and
1222  * other objects to manage the streaming.
1223  *
1224  * It will preroll the pipeline and collect vital information about the streams
1225  * such as the duration.
1226  *
1227  * Returns: %TRUE on success.
1228  */
1229 gboolean
1230 gst_rtsp_media_prepare (GstRTSPMedia *media)
1231 {
1232   GstStateChangeReturn ret;
1233   guint i, n_streams;
1234   GstRTSPMediaClass *klass;
1235   GstBus *bus;
1236   GList *walk;
1237
1238   if (media->prepared)
1239     goto was_prepared;
1240
1241   if (!media->reusable && media->reused)
1242     goto is_reused;
1243
1244   g_message ("preparing media %p", media);
1245
1246   media->pipeline = gst_pipeline_new ("media-pipeline");
1247   bus = gst_pipeline_get_bus (GST_PIPELINE_CAST (media->pipeline));
1248
1249   /* add the pipeline bus to our custom mainloop */
1250   media->source = gst_bus_create_watch (bus);
1251   gst_object_unref (bus);
1252
1253   g_source_set_callback (media->source, (GSourceFunc) bus_message, media, NULL);
1254
1255   klass = GST_RTSP_MEDIA_GET_CLASS (media);
1256   media->id = g_source_attach (media->source, klass->context);
1257
1258   gst_bin_add (GST_BIN_CAST (media->pipeline), media->element);
1259
1260   media->rtpbin = gst_element_factory_make ("gstrtpbin", "rtpbin");
1261
1262   /* add stuff to the bin */
1263   gst_bin_add (GST_BIN (media->pipeline), media->rtpbin);
1264
1265   /* link streams we already have, other streams might appear when we have
1266    * dynamic elements */
1267   n_streams = gst_rtsp_media_n_streams (media);
1268   for (i = 0; i < n_streams; i++) {
1269     GstRTSPMediaStream *stream;
1270
1271     stream = gst_rtsp_media_get_stream (media, i);
1272
1273     setup_stream (stream, i, media);
1274   }
1275
1276   for (walk = media->dynamic; walk; walk = g_list_next (walk)) {
1277     GstElement *elem = walk->data;
1278
1279     g_signal_connect (elem, "pad-added", (GCallback) pad_added_cb, media);
1280     g_signal_connect (elem, "no-more-pads", (GCallback) no_more_pads_cb, media);
1281
1282     media->fakesink = gst_element_factory_make ("fakesink", "fakesink");
1283     gst_bin_add (GST_BIN (media->pipeline), media->fakesink);
1284   }
1285
1286   /* first go to PAUSED */
1287   ret = gst_element_set_state (media->pipeline, GST_STATE_PAUSED);
1288   media->target_state = GST_STATE_PAUSED;
1289
1290   switch (ret) {
1291     case GST_STATE_CHANGE_SUCCESS:
1292       break;
1293     case GST_STATE_CHANGE_ASYNC:
1294       break;
1295     case GST_STATE_CHANGE_NO_PREROLL:
1296       /* we need to go to PLAYING */
1297       g_message ("live media %p", media);
1298       media->is_live = TRUE;
1299       ret = gst_element_set_state (media->pipeline, GST_STATE_PLAYING);
1300       if (ret == GST_STATE_CHANGE_FAILURE)
1301         goto state_failed;
1302       break;
1303     case GST_STATE_CHANGE_FAILURE:
1304       goto state_failed;
1305   }
1306
1307   /* now wait for all pads to be prerolled */
1308   ret = gst_element_get_state (media->pipeline, NULL, NULL, -1);
1309   if (ret == GST_STATE_CHANGE_FAILURE)
1310     goto state_failed;
1311
1312   /* collect stats about the media */
1313   collect_media_stats (media);
1314
1315   g_message ("object %p is prerolled", media);
1316
1317   media->prepared = TRUE;
1318
1319   return TRUE;
1320
1321   /* OK */
1322 was_prepared:
1323   {
1324     return TRUE;
1325   }
1326   /* ERRORS */
1327 state_failed:
1328   {
1329     g_warning ("failed to preroll pipeline");
1330     unlock_streams (media);
1331     gst_element_set_state (media->pipeline, GST_STATE_NULL);
1332     return FALSE;
1333   }
1334 is_reused:
1335   {
1336     g_warning ("can not reuse media %p", media);
1337     return FALSE;
1338   }
1339 }
1340
1341 /**
1342  * gst_rtsp_media_unprepare:
1343  * @obj: a #GstRTSPMedia
1344  *
1345  * Unprepare @media. After this call, the media should be prepared again before
1346  * it can be used again. If the media is set to be non-reusable, a new instance
1347  * must be created.
1348  *
1349  * Returns: %TRUE on success.
1350  */
1351 gboolean
1352 gst_rtsp_media_unprepare (GstRTSPMedia *media)
1353 {
1354   if (!media->prepared)
1355     return TRUE;
1356
1357   g_message ("unprepare media %p", media);
1358   media->target_state = GST_STATE_NULL;
1359   gst_element_set_state (media->pipeline, GST_STATE_NULL);
1360
1361   media->prepared = FALSE;
1362   media->reused = TRUE;
1363
1364   /* when the media is not reusable, this will effectively unref the media and
1365    * recreate it */
1366   g_signal_emit (media, gst_rtsp_media_signals[SIGNAL_UNPREPARED], 0, NULL);
1367
1368   return TRUE;
1369 }
1370
1371 /**
1372  * gst_rtsp_media_set_state:
1373  * @media: a #GstRTSPMedia
1374  * @state: the target state of the media
1375  * @transports: a GArray of #GstRTSPMediaTrans pointers
1376  *
1377  * Set the state of @media to @state and for the transports in @transports.
1378  *
1379  * Returns: %TRUE on success.
1380  */
1381 gboolean
1382 gst_rtsp_media_set_state (GstRTSPMedia *media, GstState state, GArray *transports)
1383 {
1384   gint i;
1385   GstStateChangeReturn ret;
1386   gboolean add, remove, do_state;
1387   gint old_active;
1388
1389   g_return_val_if_fail (GST_IS_RTSP_MEDIA (media), FALSE);
1390   g_return_val_if_fail (transports != NULL, FALSE);
1391
1392   /* NULL and READY are the same */
1393   if (state == GST_STATE_READY)
1394     state = GST_STATE_NULL;
1395
1396   add = remove = FALSE;
1397
1398   g_message ("going to state %s media %p", gst_element_state_get_name (state), media);
1399
1400   switch (state) {
1401     case GST_STATE_NULL:
1402       /* unlock the streams so that they follow the state changes from now on */
1403       unlock_streams (media);
1404       /* fallthrough */
1405     case GST_STATE_PAUSED:
1406       /* we're going from PLAYING to PAUSED, READY or NULL, remove */
1407       if (media->target_state == GST_STATE_PLAYING)
1408         remove = TRUE;
1409       break;
1410     case GST_STATE_PLAYING:
1411       /* we're going to PLAYING, add */
1412       add = TRUE;
1413       break;
1414     default:
1415       break;
1416   }
1417   old_active = media->active;
1418
1419   for (i = 0; i < transports->len; i++) {
1420     GstRTSPMediaTrans *tr;
1421     GstRTSPMediaStream *stream;
1422     GstRTSPTransport *trans;
1423
1424     /* we need a non-NULL entry in the array */
1425     tr = g_array_index (transports, GstRTSPMediaTrans *, i);
1426     if (tr == NULL)
1427       continue;
1428
1429     /* we need a transport */
1430     if (!(trans = tr->transport))
1431       continue;
1432
1433     /* get the stream and add the destinations */
1434     stream = gst_rtsp_media_get_stream (media, tr->idx);
1435     switch (trans->lower_transport) {
1436       case GST_RTSP_LOWER_TRANS_UDP:
1437       case GST_RTSP_LOWER_TRANS_UDP_MCAST:
1438       {
1439         gchar *dest;
1440         gint min, max;
1441
1442         dest = trans->destination;
1443         min = trans->client_port.min;
1444         max = trans->client_port.max;
1445
1446         if (add && !tr->active) {
1447           g_message ("adding %s:%d-%d", dest, min, max);
1448           g_signal_emit_by_name (stream->udpsink[0], "add", dest, min, NULL);
1449           g_signal_emit_by_name (stream->udpsink[1], "add", dest, max, NULL);
1450           stream->transports = g_list_prepend (stream->transports, tr);
1451           tr->active = TRUE;
1452           media->active++;
1453         } else if (remove && tr->active) {
1454           g_message ("removing %s:%d-%d", dest, min, max);
1455           g_signal_emit_by_name (stream->udpsink[0], "remove", dest, min, NULL);
1456           g_signal_emit_by_name (stream->udpsink[1], "remove", dest, max, NULL);
1457           stream->transports = g_list_remove (stream->transports, tr);
1458           tr->active = FALSE;
1459           media->active--;
1460         }
1461         break;
1462       }
1463       case GST_RTSP_LOWER_TRANS_TCP:
1464         if (add && !tr->active) {
1465           g_message ("adding TCP %s", trans->destination);
1466           stream->transports = g_list_prepend (stream->transports, tr);
1467           tr->active = TRUE;
1468           media->active++;
1469         } else if (remove && tr->active) {
1470           g_message ("removing TCP %s", trans->destination);
1471           stream->transports = g_list_remove (stream->transports, tr);
1472           tr->active = FALSE;
1473           media->active--;
1474         }
1475         break;
1476       default:
1477         g_message ("Unknown transport %d", trans->lower_transport);
1478         break;
1479     }
1480   }
1481
1482   /* we just added the first media, do the playing state change */
1483   if (old_active == 0 && add)
1484     do_state = TRUE;
1485   /* if we have no more active media, do the downward state changes */
1486   else if (media->active == 0)
1487     do_state = TRUE;
1488   else
1489     do_state = FALSE;
1490
1491   g_message ("active %d media %p", media->active, media);
1492
1493   if (do_state && media->target_state != state) {
1494     if (state == GST_STATE_NULL) {
1495       gst_rtsp_media_unprepare (media);
1496     } else {
1497       g_message ("state %s media %p", gst_element_state_get_name (state), media);
1498       media->target_state = state;
1499       ret = gst_element_set_state (media->pipeline, state);
1500     }
1501   }
1502
1503   /* remember where we are */
1504   if (state == GST_STATE_PAUSED)
1505     collect_media_stats (media);
1506
1507   return TRUE;
1508 }
1509