gst/rtsp-server/rtsp-stream.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., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19 /**
  20  * SECTION:rtsp-stream
  21  * @short_description: A media stream
  22  * @see_also: #GstRTSPMedia
  23  *
  24  * The #GstRTSPStream object manages the data transport for one stream. It
  25  * is created from a payloader element and a source pad that produce the RTP
  26  * packets for the stream.
  27  *
  28  * With gst_rtsp_stream_join_bin() the streaming elements are added to the bin
  29  * and rtpbin. gst_rtsp_stream_leave_bin() removes the elements again.
  30  *
  31  * The #GstRTSPStream will use the configured addresspool, as set with
  32  * gst_rtsp_stream_set_address_pool(), to allocate multicast addresses for the
  33  * stream. With gst_rtsp_stream_get_multicast_address() you can get the
  34  * configured address.
  35  *
  36  * With gst_rtsp_stream_get_server_port () you can get the port that the server
  37  * will use to receive RTCP. This is the part that the clients will use to send
  38  * RTCP to.
  39  *
  40  * With gst_rtsp_stream_add_transport() destinations can be added where the
  41  * stream should be sent to. Use gst_rtsp_stream_remove_transport() to remove
  42  * the destination again.
  43  *
  44  * Last reviewed on 2013-07-16 (1.0.0)
  45  */
  46
  47 #include <stdlib.h>
  48 #include <stdio.h>
  49 #include <string.h>
  50
  51 #include <gio/gio.h>
  52
  53 #include <gst/app/gstappsrc.h>
  54 #include <gst/app/gstappsink.h>
  55
  56 #include "rtsp-stream.h"
  57
  58 #define GST_RTSP_STREAM_GET_PRIVATE(obj)  \
  59      (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_RTSP_STREAM, GstRTSPStreamPrivate))
  60
  61 struct _GstRTSPStreamPrivate
  62 {
  63   GMutex lock;
  64   guint idx;
  65   GstPad *srcpad;
  66   GstElement *payloader;
  67   guint buffer_size;
  68   gboolean is_joined;
  69   gchar *control;
  70
  71   GstRTSPLowerTrans protocols;
  72
  73   /* pads on the rtpbin */
  74   GstPad *send_rtp_sink;
  75   GstPad *recv_sink[2];
  76   GstPad *send_src[2];
  77
  78   /* the RTPSession object */
  79   GObject *session;
  80
  81   /* sinks used for sending and receiving RTP and RTCP over ipv4, they share
  82    * sockets */
  83   GstElement *udpsrc_v4[2];
  84
  85   /* sinks used for sending and receiving RTP and RTCP over ipv6, they share
  86    * sockets */
  87   GstElement *udpsrc_v6[2];
  88
  89   GstElement *udpsink[2];
  90
  91   /* for TCP transport */
  92   GstElement *appsrc[2];
  93   GstElement *appqueue[2];
  94   GstElement *appsink[2];
  95
  96   GstElement *tee[2];
  97   GstElement *funnel[2];
  98
  99   /* server ports for sending/receiving over ipv4 */
 100   GstRTSPRange server_port_v4;
 101   GstRTSPAddress *server_addr_v4;
 102   gboolean have_ipv4;
 103
 104   /* server ports for sending/receiving over ipv6 */
 105   GstRTSPRange server_port_v6;
 106   GstRTSPAddress *server_addr_v6;
 107   gboolean have_ipv6;
 108
 109   /* multicast addresses */
 110   GstRTSPAddressPool *pool;
 111   GstRTSPAddress *addr_v4;
 112   GstRTSPAddress *addr_v6;
 113
 114   /* the caps of the stream */
 115   gulong caps_sig;
 116   GstCaps *caps;
 117
 118   /* transports we stream to */
 119   guint n_active;
 120   GList *transports;
 121
 122   gint dscp_qos;
 123 };
 124
 125 #define DEFAULT_CONTROL         NULL
 126 #define DEFAULT_PROTOCOLS       GST_RTSP_LOWER_TRANS_UDP | GST_RTSP_LOWER_TRANS_UDP_MCAST | \
 127                                         GST_RTSP_LOWER_TRANS_TCP
 128
 129 enum
 130 {
 131   PROP_0,
 132   PROP_CONTROL,
 133   PROP_PROTOCOLS,
 134   PROP_LAST
 135 };
 136
 137 GST_DEBUG_CATEGORY_STATIC (rtsp_stream_debug);
 138 #define GST_CAT_DEFAULT rtsp_stream_debug
 139
 140 static GQuark ssrc_stream_map_key;
 141
 142 static void gst_rtsp_stream_get_property (GObject * object, guint propid,
 143     GValue * value, GParamSpec * pspec);
 144 static void gst_rtsp_stream_set_property (GObject * object, guint propid,
 145     const GValue * value, GParamSpec * pspec);
 146
 147 static void gst_rtsp_stream_finalize (GObject * obj);
 148
 149 G_DEFINE_TYPE (GstRTSPStream, gst_rtsp_stream, G_TYPE_OBJECT);
 150
 151 static void
 152 gst_rtsp_stream_class_init (GstRTSPStreamClass * klass)
 153 {
 154   GObjectClass *gobject_class;
 155
 156   g_type_class_add_private (klass, sizeof (GstRTSPStreamPrivate));
 157
 158   gobject_class = G_OBJECT_CLASS (klass);
 159
 160   gobject_class->get_property = gst_rtsp_stream_get_property;
 161   gobject_class->set_property = gst_rtsp_stream_set_property;
 162   gobject_class->finalize = gst_rtsp_stream_finalize;
 163
 164   g_object_class_install_property (gobject_class, PROP_CONTROL,
 165       g_param_spec_string ("control", "Control",
 166           "The control string for this stream", DEFAULT_CONTROL,
 167           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 168
 169   g_object_class_install_property (gobject_class, PROP_PROTOCOLS,
 170       g_param_spec_flags ("protocols", "Protocols",
 171           "Allowed lower transport protocols", GST_TYPE_RTSP_LOWER_TRANS,
 172           DEFAULT_PROTOCOLS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 173
 174   GST_DEBUG_CATEGORY_INIT (rtsp_stream_debug, "rtspstream", 0, "GstRTSPStream");
 175
 176   ssrc_stream_map_key = g_quark_from_static_string ("GstRTSPServer.stream");
 177 }
 178
 179 static void
 180 gst_rtsp_stream_init (GstRTSPStream * stream)
 181 {
 182   GstRTSPStreamPrivate *priv = GST_RTSP_STREAM_GET_PRIVATE (stream);
 183
 184   GST_DEBUG ("new stream %p", stream);
 185
 186   stream->priv = priv;
 187
 188   priv->dscp_qos = -1;
 189   priv->control = g_strdup (DEFAULT_CONTROL);
 190   priv->protocols = DEFAULT_PROTOCOLS;
 191
 192   g_mutex_init (&priv->lock);
 193 }
 194
 195 static void
 196 gst_rtsp_stream_finalize (GObject * obj)
 197 {
 198   GstRTSPStream *stream;
 199   GstRTSPStreamPrivate *priv;
 200
 201   stream = GST_RTSP_STREAM (obj);
 202   priv = stream->priv;
 203
 204   GST_DEBUG ("finalize stream %p", stream);
 205
 206   /* we really need to be unjoined now */
 207   g_return_if_fail (!priv->is_joined);
 208
 209   if (priv->addr_v4)
 210     gst_rtsp_address_free (priv->addr_v4);
 211   if (priv->addr_v6)
 212     gst_rtsp_address_free (priv->addr_v6);
 213   if (priv->server_addr_v4)
 214     gst_rtsp_address_free (priv->server_addr_v4);
 215   if (priv->server_addr_v6)
 216     gst_rtsp_address_free (priv->server_addr_v6);
 217   if (priv->pool)
 218     g_object_unref (priv->pool);
 219   gst_object_unref (priv->payloader);
 220   gst_object_unref (priv->srcpad);
 221   g_free (priv->control);
 222   g_mutex_clear (&priv->lock);
 223
 224   G_OBJECT_CLASS (gst_rtsp_stream_parent_class)->finalize (obj);
 225 }
 226
 227 static void
 228 gst_rtsp_stream_get_property (GObject * object, guint propid,
 229     GValue * value, GParamSpec * pspec)
 230 {
 231   GstRTSPStream *stream = GST_RTSP_STREAM (object);
 232
 233   switch (propid) {
 234     case PROP_CONTROL:
 235       g_value_take_string (value, gst_rtsp_stream_get_control (stream));
 236       break;
 237     case PROP_PROTOCOLS:
 238       g_value_set_flags (value, gst_rtsp_stream_get_protocols (stream));
 239       break;
 240     default:
 241       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 242   }
 243 }
 244
 245 static void
 246 gst_rtsp_stream_set_property (GObject * object, guint propid,
 247     const GValue * value, GParamSpec * pspec)
 248 {
 249   GstRTSPStream *stream = GST_RTSP_STREAM (object);
 250
 251   switch (propid) {
 252     case PROP_CONTROL:
 253       gst_rtsp_stream_set_control (stream, g_value_get_string (value));
 254       break;
 255     case PROP_PROTOCOLS:
 256       gst_rtsp_stream_set_protocols (stream, g_value_get_flags (value));
 257       break;
 258     default:
 259       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, propid, pspec);
 260   }
 261 }
 262
 263 /**
 264  * gst_rtsp_stream_new:
 265  * @idx: an index
 266  * @srcpad: a #GstPad
 267  * @payloader: a #GstElement
 268  *
 269  * Create a new media stream with index @idx that handles RTP data on
 270  * @srcpad and has a payloader element @payloader.
 271  *
 272  * Returns: a new #GstRTSPStream
 273  */
 274 GstRTSPStream *
 275 gst_rtsp_stream_new (guint idx, GstElement * payloader, GstPad * srcpad)
 276 {
 277   GstRTSPStreamPrivate *priv;
 278   GstRTSPStream *stream;
 279
 280   g_return_val_if_fail (GST_IS_ELEMENT (payloader), NULL);
 281   g_return_val_if_fail (GST_IS_PAD (srcpad), NULL);
 282   g_return_val_if_fail (GST_PAD_IS_SRC (srcpad), NULL);
 283
 284   stream = g_object_new (GST_TYPE_RTSP_STREAM, NULL);
 285   priv = stream->priv;
 286   priv->idx = idx;
 287   priv->payloader = gst_object_ref (payloader);
 288   priv->srcpad = gst_object_ref (srcpad);
 289
 290   return stream;
 291 }
 292
 293 /**
 294  * gst_rtsp_stream_get_index:
 295  * @stream: a #GstRTSPStream
 296  *
 297  * Get the stream index.
 298  *
 299  * Return: the stream index.
 300  */
 301 guint
 302 gst_rtsp_stream_get_index (GstRTSPStream * stream)
 303 {
 304   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), -1);
 305
 306   return stream->priv->idx;
 307 }
 308
 309 /**
 310  * gst_rtsp_stream_get_srcpad:
 311  * @stream: a #GstRTSPStream
 312  *
 313  * Get the srcpad associated with @stream.
 314  *
 315  * Return: the srcpad. Unref after usage.
 316  */
 317 GstPad *
 318 gst_rtsp_stream_get_srcpad (GstRTSPStream * stream)
 319 {
 320   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
 321
 322   return gst_object_ref (stream->priv->srcpad);
 323 }
 324
 325 /**
 326  * gst_rtsp_stream_get_control:
 327  * @stream: a #GstRTSPStream
 328  *
 329  * Get the control string to identify this stream.
 330  *
 331  * Return: the control string. free after usage.
 332  */
 333 gchar *
 334 gst_rtsp_stream_get_control (GstRTSPStream * stream)
 335 {
 336   GstRTSPStreamPrivate *priv;
 337   gchar *result;
 338
 339   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
 340
 341   priv = stream->priv;
 342
 343   g_mutex_lock (&priv->lock);
 344   if ((result = g_strdup (priv->control)) == NULL)
 345     result = g_strdup_printf ("stream=%u", priv->idx);
 346   g_mutex_unlock (&priv->lock);
 347
 348   return result;
 349 }
 350
 351 /**
 352  * gst_rtsp_stream_set_control:
 353  * @stream: a #GstRTSPStream
 354  * @control: a control string
 355  *
 356  * Set the control string in @stream.
 357  */
 358 void
 359 gst_rtsp_stream_set_control (GstRTSPStream * stream, const gchar * control)
 360 {
 361   GstRTSPStreamPrivate *priv;
 362
 363   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
 364
 365   priv = stream->priv;
 366
 367   g_mutex_lock (&priv->lock);
 368   g_free (priv->control);
 369   priv->control = g_strdup (control);
 370   g_mutex_unlock (&priv->lock);
 371 }
 372
 373 /**
 374  * gst_rtsp_stream_has_control:
 375  * @stream: a #GstRTSPStream
 376  * @control: a control string
 377  *
 378  * Check if @stream has the control string @control.
 379  *
 380  * Returns: %TRUE is @stream has @control as the control string
 381  */
 382 gboolean
 383 gst_rtsp_stream_has_control (GstRTSPStream * stream, const gchar * control)
 384 {
 385   GstRTSPStreamPrivate *priv;
 386   gboolean res;
 387
 388   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
 389
 390   priv = stream->priv;
 391
 392   g_mutex_lock (&priv->lock);
 393   if (priv->control)
 394     res = g_strcmp0 (priv->control, control);
 395   else {
 396     guint streamid;
 397     sscanf (control, "stream=%u", &streamid);
 398     res = (streamid == priv->idx);
 399   }
 400   g_mutex_unlock (&priv->lock);
 401
 402   return res;
 403 }
 404
 405 /**
 406  * gst_rtsp_stream_set_mtu:
 407  * @stream: a #GstRTSPStream
 408  * @mtu: a new MTU
 409  *
 410  * Configure the mtu in the payloader of @stream to @mtu.
 411  */
 412 void
 413 gst_rtsp_stream_set_mtu (GstRTSPStream * stream, guint mtu)
 414 {
 415   GstRTSPStreamPrivate *priv;
 416
 417   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
 418
 419   priv = stream->priv;
 420
 421   GST_LOG_OBJECT (stream, "set MTU %u", mtu);
 422
 423   g_object_set (G_OBJECT (priv->payloader), "mtu", mtu, NULL);
 424 }
 425
 426 /**
 427  * gst_rtsp_stream_get_mtu:
 428  * @stream: a #GstRTSPStream
 429  *
 430  * Get the configured MTU in the payloader of @stream.
 431  *
 432  * Returns: the MTU of the payloader.
 433  */
 434 guint
 435 gst_rtsp_stream_get_mtu (GstRTSPStream * stream)
 436 {
 437   GstRTSPStreamPrivate *priv;
 438   guint mtu;
 439
 440   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), 0);
 441
 442   priv = stream->priv;
 443
 444   g_object_get (G_OBJECT (priv->payloader), "mtu", &mtu, NULL);
 445
 446   return mtu;
 447 }
 448
 449 /* Update the dscp qos property on the udp sinks */
 450 static void
 451 update_dscp_qos (GstRTSPStream * stream)
 452 {
 453   GstRTSPStreamPrivate *priv;
 454
 455   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
 456
 457   priv = stream->priv;
 458
 459   if (priv->udpsink[0]) {
 460     g_object_set (G_OBJECT (priv->udpsink[0]), "qos-dscp", priv->dscp_qos,
 461         NULL);
 462   }
 463
 464   if (priv->udpsink[1]) {
 465     g_object_set (G_OBJECT (priv->udpsink[1]), "qos-dscp", priv->dscp_qos,
 466         NULL);
 467   }
 468 }
 469
 470 /**
 471  * gst_rtsp_stream_set_dscp_qos:
 472  * @stream: a #GstRTSPStream
 473  * @dscp_qos: a new dscp qos value (0-63, or -1 to disable)
 474  *
 475  * Configure the dscp qos of the outgoing sockets to @dscp_qos.
 476  */
 477 void
 478 gst_rtsp_stream_set_dscp_qos (GstRTSPStream * stream, gint dscp_qos)
 479 {
 480   GstRTSPStreamPrivate *priv;
 481
 482   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
 483
 484   priv = stream->priv;
 485
 486   GST_LOG_OBJECT (stream, "set DSCP QoS %d", dscp_qos);
 487
 488   if (dscp_qos < -1 || dscp_qos > 63) {
 489     GST_WARNING_OBJECT (stream, "trying to set illegal dscp qos %d", dscp_qos);
 490     return;
 491   }
 492
 493   priv->dscp_qos = dscp_qos;
 494
 495   update_dscp_qos (stream);
 496 }
 497
 498 /**
 499  * gst_rtsp_stream_get_dscp_qos:
 500  * @stream: a #GstRTSPStream
 501  *
 502  * Get the configured DSCP QoS in of the outgoing sockets.
 503  *
 504  * Returns: the DSCP QoS value of the outgoing sockets, or -1 if disbled.
 505  */
 506 gint
 507 gst_rtsp_stream_get_dscp_qos (GstRTSPStream * stream)
 508 {
 509   GstRTSPStreamPrivate *priv;
 510
 511   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), -1);
 512
 513   priv = stream->priv;
 514
 515   return priv->dscp_qos;
 516 }
 517
 518 /**
 519  * gst_rtsp_stream_set_protocols:
 520  * @stream: a #GstRTSPStream
 521  * @protocols: the new flags
 522  *
 523  * Configure the allowed lower transport for @stream.
 524  */
 525 void
 526 gst_rtsp_stream_set_protocols (GstRTSPStream * stream,
 527     GstRTSPLowerTrans protocols)
 528 {
 529   GstRTSPStreamPrivate *priv;
 530
 531   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
 532
 533   priv = stream->priv;
 534
 535   g_mutex_lock (&priv->lock);
 536   priv->protocols = protocols;
 537   g_mutex_unlock (&priv->lock);
 538 }
 539
 540 /**
 541  * gst_rtsp_stream_get_protocols:
 542  * @stream: a #GstRTSPStream
 543  *
 544  * Get the allowed protocols of @stream.
 545  *
 546  * Returns: a #GstRTSPLowerTrans
 547  */
 548 GstRTSPLowerTrans
 549 gst_rtsp_stream_get_protocols (GstRTSPStream * stream)
 550 {
 551   GstRTSPStreamPrivate *priv;
 552   GstRTSPLowerTrans res;
 553
 554   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream),
 555       GST_RTSP_LOWER_TRANS_UNKNOWN);
 556
 557   priv = stream->priv;
 558
 559   g_mutex_lock (&priv->lock);
 560   res = priv->protocols;
 561   g_mutex_unlock (&priv->lock);
 562
 563   return res;
 564 }
 565
 566 /**
 567  * gst_rtsp_stream_set_address_pool:
 568  * @stream: a #GstRTSPStream
 569  * @pool: a #GstRTSPAddressPool
 570  *
 571  * configure @pool to be used as the address pool of @stream.
 572  */
 573 void
 574 gst_rtsp_stream_set_address_pool (GstRTSPStream * stream,
 575     GstRTSPAddressPool * pool)
 576 {
 577   GstRTSPStreamPrivate *priv;
 578   GstRTSPAddressPool *old;
 579
 580   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
 581
 582   priv = stream->priv;
 583
 584   GST_LOG_OBJECT (stream, "set address pool %p", pool);
 585
 586   g_mutex_lock (&priv->lock);
 587   if ((old = priv->pool) != pool)
 588     priv->pool = pool ? g_object_ref (pool) : NULL;
 589   else
 590     old = NULL;
 591   g_mutex_unlock (&priv->lock);
 592
 593   if (old)
 594     g_object_unref (old);
 595 }
 596
 597 /**
 598  * gst_rtsp_stream_get_address_pool:
 599  * @stream: a #GstRTSPStream
 600  *
 601  * Get the #GstRTSPAddressPool used as the address pool of @stream.
 602  *
 603  * Returns: (transfer full): the #GstRTSPAddressPool of @stream. g_object_unref() after
 604  * usage.
 605  */
 606 GstRTSPAddressPool *
 607 gst_rtsp_stream_get_address_pool (GstRTSPStream * stream)
 608 {
 609   GstRTSPStreamPrivate *priv;
 610   GstRTSPAddressPool *result;
 611
 612   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
 613
 614   priv = stream->priv;
 615
 616   g_mutex_lock (&priv->lock);
 617   if ((result = priv->pool))
 618     g_object_ref (result);
 619   g_mutex_unlock (&priv->lock);
 620
 621   return result;
 622 }
 623
 624 /**
 625  * gst_rtsp_stream_get_multicast_address:
 626  * @stream: a #GstRTSPStream
 627  * @family: the #GSocketFamily
 628  *
 629  * Get the multicast address of @stream for @family.
 630  *
 631  * Returns: the #GstRTSPAddress of @stream or %NULL when no address could be
 632  * allocated. gst_rtsp_address_free() after usage.
 633  */
 634 GstRTSPAddress *
 635 gst_rtsp_stream_get_multicast_address (GstRTSPStream * stream,
 636     GSocketFamily family)
 637 {
 638   GstRTSPStreamPrivate *priv;
 639   GstRTSPAddress *result;
 640   GstRTSPAddress **addrp;
 641   GstRTSPAddressFlags flags;
 642
 643   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
 644
 645   priv = stream->priv;
 646
 647   if (family == G_SOCKET_FAMILY_IPV6) {
 648     flags = GST_RTSP_ADDRESS_FLAG_IPV6;
 649     addrp = &priv->addr_v4;
 650   } else {
 651     flags = GST_RTSP_ADDRESS_FLAG_IPV4;
 652     addrp = &priv->addr_v6;
 653   }
 654
 655   g_mutex_lock (&priv->lock);
 656   if (*addrp == NULL) {
 657     if (priv->pool == NULL)
 658       goto no_pool;
 659
 660     flags |= GST_RTSP_ADDRESS_FLAG_EVEN_PORT | GST_RTSP_ADDRESS_FLAG_MULTICAST;
 661
 662     *addrp = gst_rtsp_address_pool_acquire_address (priv->pool, flags, 2);
 663     if (*addrp == NULL)
 664       goto no_address;
 665   }
 666   result = gst_rtsp_address_copy (*addrp);
 667   g_mutex_unlock (&priv->lock);
 668
 669   return result;
 670
 671   /* ERRORS */
 672 no_pool:
 673   {
 674     GST_ERROR_OBJECT (stream, "no address pool specified");
 675     g_mutex_unlock (&priv->lock);
 676     return NULL;
 677   }
 678 no_address:
 679   {
 680     GST_ERROR_OBJECT (stream, "failed to acquire address from pool");
 681     g_mutex_unlock (&priv->lock);
 682     return NULL;
 683   }
 684 }
 685
 686 /**
 687  * gst_rtsp_stream_reserve_address:
 688  * @stream: a #GstRTSPStream
 689  * @address: an address
 690  * @port: a port
 691  * @n_ports: n_ports
 692  * @ttl: a TTL
 693  *
 694  * Reserve @address and @port as the address and port of @stream.
 695  *
 696  * Returns: the #GstRTSPAddress of @stream or %NULL when the address could be
 697  * reserved. gst_rtsp_address_free() after usage.
 698  */
 699 GstRTSPAddress *
 700 gst_rtsp_stream_reserve_address (GstRTSPStream * stream,
 701     const gchar * address, guint port, guint n_ports, guint ttl)
 702 {
 703   GstRTSPStreamPrivate *priv;
 704   GstRTSPAddress *result;
 705   GInetAddress *addr;
 706   GSocketFamily family;
 707   GstRTSPAddress **addrp;
 708
 709   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
 710   g_return_val_if_fail (address != NULL, NULL);
 711   g_return_val_if_fail (port > 0, NULL);
 712   g_return_val_if_fail (n_ports > 0, NULL);
 713   g_return_val_if_fail (ttl > 0, NULL);
 714
 715   priv = stream->priv;
 716
 717   addr = g_inet_address_new_from_string (address);
 718   if (!addr) {
 719     GST_ERROR ("failed to get inet addr from %s", address);
 720     family = G_SOCKET_FAMILY_IPV4;
 721   } else {
 722     family = g_inet_address_get_family (addr);
 723     g_object_unref (addr);
 724   }
 725
 726   if (family == G_SOCKET_FAMILY_IPV6)
 727     addrp = &priv->addr_v4;
 728   else
 729     addrp = &priv->addr_v6;
 730
 731   g_mutex_lock (&priv->lock);
 732   if (*addrp == NULL) {
 733     if (priv->pool == NULL)
 734       goto no_pool;
 735
 736     *addrp = gst_rtsp_address_pool_reserve_address (priv->pool, address,
 737         port, n_ports, ttl);
 738     if (*addrp == NULL)
 739       goto no_address;
 740   } else {
 741     if (strcmp ((*addrp)->address, address) ||
 742         (*addrp)->port != port || (*addrp)->n_ports != n_ports ||
 743         (*addrp)->ttl != ttl)
 744       goto different_address;
 745   }
 746   result = gst_rtsp_address_copy (*addrp);
 747   g_mutex_unlock (&priv->lock);
 748
 749   return result;
 750
 751   /* ERRORS */
 752 no_pool:
 753   {
 754     GST_ERROR_OBJECT (stream, "no address pool specified");
 755     g_mutex_unlock (&priv->lock);
 756     return NULL;
 757   }
 758 no_address:
 759   {
 760     GST_ERROR_OBJECT (stream, "failed to acquire address %s from pool",
 761         address);
 762     g_mutex_unlock (&priv->lock);
 763     return NULL;
 764   }
 765 different_address:
 766   {
 767     GST_ERROR_OBJECT (stream, "address %s is not the same that was already"
 768         " reserved", address);
 769     g_mutex_unlock (&priv->lock);
 770     return NULL;
 771   }
 772 }
 773
 774 static gboolean
 775 alloc_ports_one_family (GstRTSPAddressPool * pool, gint buffer_size,
 776     GSocketFamily family, GstElement * udpsrc_out[2],
 777     GstElement * udpsink_out[2], GstRTSPRange * server_port_out,
 778     GstRTSPAddress ** server_addr_out)
 779 {
 780   GstStateChangeReturn ret;
 781   GstElement *udpsrc0, *udpsrc1;
 782   GstElement *udpsink0, *udpsink1;
 783   GSocket *rtp_socket = NULL;
 784   GSocket *rtcp_socket;
 785   gint tmp_rtp, tmp_rtcp;
 786   guint count;
 787   gint rtpport, rtcpport;
 788   GList *rejected_addresses = NULL;
 789   GstRTSPAddress *addr = NULL;
 790   GInetAddress *inetaddr = NULL;
 791   GSocketAddress *rtp_sockaddr = NULL;
 792   GSocketAddress *rtcp_sockaddr = NULL;
 793   const gchar *multisink_socket;
 794
 795   if (family == G_SOCKET_FAMILY_IPV6)
 796     multisink_socket = "socket-v6";
 797   else
 798     multisink_socket = "socket";
 799
 800   udpsrc0 = NULL;
 801   udpsrc1 = NULL;
 802   udpsink0 = NULL;
 803   udpsink1 = NULL;
 804   count = 0;
 805
 806   /* Start with random port */
 807   tmp_rtp = 0;
 808
 809   rtcp_socket = g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
 810       G_SOCKET_PROTOCOL_UDP, NULL);
 811   if (!rtcp_socket)
 812     goto no_udp_protocol;
 813
 814   if (*server_addr_out)
 815     gst_rtsp_address_free (*server_addr_out);
 816
 817   /* try to allocate 2 UDP ports, the RTP port should be an even
 818    * number and the RTCP port should be the next (uneven) port */
 819 again:
 820
 821   if (rtp_socket == NULL) {
 822     rtp_socket = g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
 823         G_SOCKET_PROTOCOL_UDP, NULL);
 824     if (!rtp_socket)
 825       goto no_udp_protocol;
 826   }
 827
 828   if (pool && gst_rtsp_address_pool_has_unicast_addresses (pool)) {
 829     GstRTSPAddressFlags flags;
 830
 831     if (addr)
 832       rejected_addresses = g_list_prepend (rejected_addresses, addr);
 833
 834     flags = GST_RTSP_ADDRESS_FLAG_EVEN_PORT | GST_RTSP_ADDRESS_FLAG_UNICAST;
 835     if (family == G_SOCKET_FAMILY_IPV6)
 836       flags |= GST_RTSP_ADDRESS_FLAG_IPV6;
 837     else
 838       flags |= GST_RTSP_ADDRESS_FLAG_IPV4;
 839
 840     addr = gst_rtsp_address_pool_acquire_address (pool, flags, 2);
 841
 842     if (addr == NULL)
 843       goto no_ports;
 844
 845     tmp_rtp = addr->port;
 846
 847     g_clear_object (&inetaddr);
 848     inetaddr = g_inet_address_new_from_string (addr->address);
 849   } else {
 850     if (tmp_rtp != 0) {
 851       tmp_rtp += 2;
 852       if (++count > 20)
 853         goto no_ports;
 854     }
 855
 856     if (inetaddr == NULL)
 857       inetaddr = g_inet_address_new_any (family);
 858   }
 859
 860   rtp_sockaddr = g_inet_socket_address_new (inetaddr, tmp_rtp);
 861   if (!g_socket_bind (rtp_socket, rtp_sockaddr, FALSE, NULL)) {
 862     g_object_unref (rtp_sockaddr);
 863     goto again;
 864   }
 865   g_object_unref (rtp_sockaddr);
 866
 867   rtp_sockaddr = g_socket_get_local_address (rtp_socket, NULL);
 868   if (rtp_sockaddr == NULL || !G_IS_INET_SOCKET_ADDRESS (rtp_sockaddr)) {
 869     g_clear_object (&rtp_sockaddr);
 870     goto socket_error;
 871   }
 872
 873   tmp_rtp =
 874       g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (rtp_sockaddr));
 875   g_object_unref (rtp_sockaddr);
 876
 877   /* check if port is even */
 878   if ((tmp_rtp & 1) != 0) {
 879     /* port not even, close and allocate another */
 880     tmp_rtp++;
 881     g_clear_object (&rtp_socket);
 882     goto again;
 883   }
 884
 885   /* set port */
 886   tmp_rtcp = tmp_rtp + 1;
 887
 888   rtcp_sockaddr = g_inet_socket_address_new (inetaddr, tmp_rtcp);
 889   if (!g_socket_bind (rtcp_socket, rtcp_sockaddr, FALSE, NULL)) {
 890     g_object_unref (rtcp_sockaddr);
 891     g_clear_object (&rtp_socket);
 892     goto again;
 893   }
 894   g_object_unref (rtcp_sockaddr);
 895
 896   g_clear_object (&inetaddr);
 897
 898   udpsrc0 = gst_element_factory_make ("udpsrc", NULL);
 899   udpsrc1 = gst_element_factory_make ("udpsrc", NULL);
 900
 901   if (udpsrc0 == NULL || udpsrc1 == NULL)
 902     goto no_udp_protocol;
 903
 904   g_object_set (G_OBJECT (udpsrc0), "socket", rtp_socket, NULL);
 905   g_object_set (G_OBJECT (udpsrc1), "socket", rtcp_socket, NULL);
 906
 907   ret = gst_element_set_state (udpsrc0, GST_STATE_PAUSED);
 908   if (ret == GST_STATE_CHANGE_FAILURE)
 909     goto element_error;
 910   ret = gst_element_set_state (udpsrc1, GST_STATE_PAUSED);
 911   if (ret == GST_STATE_CHANGE_FAILURE)
 912     goto element_error;
 913
 914   /* all fine, do port check */
 915   g_object_get (G_OBJECT (udpsrc0), "port", &rtpport, NULL);
 916   g_object_get (G_OBJECT (udpsrc1), "port", &rtcpport, NULL);
 917
 918   /* this should not happen... */
 919   if (rtpport != tmp_rtp || rtcpport != tmp_rtcp)
 920     goto port_error;
 921
 922   if (udpsink_out[0])
 923     udpsink0 = udpsink_out[0];
 924   else
 925     udpsink0 = gst_element_factory_make ("multiudpsink", NULL);
 926
 927   if (!udpsink0)
 928     goto no_udp_protocol;
 929
 930   g_object_set (G_OBJECT (udpsink0), "close-socket", FALSE, NULL);
 931   g_object_set (G_OBJECT (udpsink0), multisink_socket, rtp_socket, NULL);
 932
 933   if (udpsink_out[1])
 934     udpsink1 = udpsink_out[1];
 935   else
 936     udpsink1 = gst_element_factory_make ("multiudpsink", NULL);
 937
 938   if (!udpsink1)
 939     goto no_udp_protocol;
 940
 941   g_object_set (G_OBJECT (udpsink0), "send-duplicates", FALSE, NULL);
 942   g_object_set (G_OBJECT (udpsink1), "send-duplicates", FALSE, NULL);
 943   g_object_set (G_OBJECT (udpsink0), "buffer-size", buffer_size, NULL);
 944
 945   g_object_set (G_OBJECT (udpsink1), "close-socket", FALSE, NULL);
 946   g_object_set (G_OBJECT (udpsink1), multisink_socket, rtcp_socket, NULL);
 947   g_object_set (G_OBJECT (udpsink1), "sync", FALSE, NULL);
 948   g_object_set (G_OBJECT (udpsink1), "async", FALSE, NULL);
 949   g_object_set (G_OBJECT (udpsink0), "auto-multicast", FALSE, NULL);
 950   g_object_set (G_OBJECT (udpsink0), "loop", FALSE, NULL);
 951   g_object_set (G_OBJECT (udpsink1), "auto-multicast", FALSE, NULL);
 952   g_object_set (G_OBJECT (udpsink1), "loop", FALSE, NULL);
 953
 954   /* we keep these elements, we will further configure them when the
 955    * client told us to really use the UDP ports. */
 956   udpsrc_out[0] = udpsrc0;
 957   udpsrc_out[1] = udpsrc1;
 958   udpsink_out[0] = udpsink0;
 959   udpsink_out[1] = udpsink1;
 960   server_port_out->min = rtpport;
 961   server_port_out->max = rtcpport;
 962
 963   *server_addr_out = addr;
 964   g_list_free_full (rejected_addresses, (GDestroyNotify) gst_rtsp_address_free);
 965
 966   g_object_unref (rtp_socket);
 967   g_object_unref (rtcp_socket);
 968
 969   return TRUE;
 970
 971   /* ERRORS */
 972 no_udp_protocol:
 973   {
 974     goto cleanup;
 975   }
 976 no_ports:
 977   {
 978     goto cleanup;
 979   }
 980 port_error:
 981   {
 982     goto cleanup;
 983   }
 984 socket_error:
 985   {
 986     goto cleanup;
 987   }
 988 element_error:
 989   {
 990     goto cleanup;
 991   }
 992 cleanup:
 993   {
 994     if (udpsrc0) {
 995       gst_element_set_state (udpsrc0, GST_STATE_NULL);
 996       gst_object_unref (udpsrc0);
 997     }
 998     if (udpsrc1) {
 999       gst_element_set_state (udpsrc1, GST_STATE_NULL);
1000       gst_object_unref (udpsrc1);
1001     }
1002     if (udpsink0) {
1003       gst_element_set_state (udpsink0, GST_STATE_NULL);
1004       gst_object_unref (udpsink0);
1005     }
1006     if (udpsink1) {
1007       gst_element_set_state (udpsink1, GST_STATE_NULL);
1008       gst_object_unref (udpsink1);
1009     }
1010     if (inetaddr)
1011       g_object_unref (inetaddr);
1012     g_list_free_full (rejected_addresses,
1013         (GDestroyNotify) gst_rtsp_address_free);
1014     if (addr)
1015       gst_rtsp_address_free (addr);
1016     if (rtp_socket)
1017       g_object_unref (rtp_socket);
1018     if (rtcp_socket)
1019       g_object_unref (rtcp_socket);
1020     return FALSE;
1021   }
1022 }
1023
1024 /* must be called with lock */
1025 static gboolean
1026 alloc_ports (GstRTSPStream * stream)
1027 {
1028   GstRTSPStreamPrivate *priv = stream->priv;
1029
1030   priv->have_ipv4 = alloc_ports_one_family (priv->pool, priv->buffer_size,
1031       G_SOCKET_FAMILY_IPV4, priv->udpsrc_v4, priv->udpsink,
1032       &priv->server_port_v4, &priv->server_addr_v4);
1033
1034   priv->have_ipv6 = alloc_ports_one_family (priv->pool, priv->buffer_size,
1035       G_SOCKET_FAMILY_IPV6, priv->udpsrc_v6, priv->udpsink,
1036       &priv->server_port_v6, &priv->server_addr_v6);
1037
1038   return priv->have_ipv4 || priv->have_ipv6;
1039 }
1040
1041 /**
1042  * gst_rtsp_stream_get_server_port:
1043  * @stream: a #GstRTSPStream
1044  * @server_port: (out): result server port
1045  * @family: the port family to get
1046  *
1047  * Fill @server_port with the port pair used by the server. This function can
1048  * only be called when @stream has been joined.
1049  */
1050 void
1051 gst_rtsp_stream_get_server_port (GstRTSPStream * stream,
1052     GstRTSPRange * server_port, GSocketFamily family)
1053 {
1054   GstRTSPStreamPrivate *priv;
1055
1056   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
1057   priv = stream->priv;
1058   g_return_if_fail (priv->is_joined);
1059
1060   g_mutex_lock (&priv->lock);
1061   if (family == G_SOCKET_FAMILY_IPV4) {
1062     if (server_port)
1063       *server_port = priv->server_port_v4;
1064   } else {
1065     if (server_port)
1066       *server_port = priv->server_port_v6;
1067   }
1068   g_mutex_unlock (&priv->lock);
1069 }
1070
1071 /**
1072  * gst_rtsp_stream_get_rtpsession:
1073  * @stream: a #GstRTSPStream
1074  *
1075  * Get the RTP session of this stream.
1076  *
1077  * Returns: The RTP session of this stream. Unref after usage.
1078  */
1079 GObject *
1080 gst_rtsp_stream_get_rtpsession (GstRTSPStream * stream)
1081 {
1082   GstRTSPStreamPrivate *priv;
1083   GObject *session;
1084
1085   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
1086
1087   priv = stream->priv;
1088
1089   g_mutex_lock (&priv->lock);
1090   if ((session = priv->session))
1091     g_object_ref (session);
1092   g_mutex_unlock (&priv->lock);
1093
1094   return session;
1095 }
1096
1097 /**
1098  * gst_rtsp_stream_get_ssrc:
1099  * @stream: a #GstRTSPStream
1100  * @ssrc: (out): result ssrc
1101  *
1102  * Get the SSRC used by the RTP session of this stream. This function can only
1103  * be called when @stream has been joined.
1104  */
1105 void
1106 gst_rtsp_stream_get_ssrc (GstRTSPStream * stream, guint * ssrc)
1107 {
1108   GstRTSPStreamPrivate *priv;
1109
1110   g_return_if_fail (GST_IS_RTSP_STREAM (stream));
1111   priv = stream->priv;
1112   g_return_if_fail (priv->is_joined);
1113
1114   g_mutex_lock (&priv->lock);
1115   if (ssrc && priv->session)
1116     g_object_get (priv->session, "internal-ssrc", ssrc, NULL);
1117   g_mutex_unlock (&priv->lock);
1118 }
1119
1120 /* executed from streaming thread */
1121 static void
1122 caps_notify (GstPad * pad, GParamSpec * unused, GstRTSPStream * stream)
1123 {
1124   GstRTSPStreamPrivate *priv = stream->priv;
1125   GstCaps *newcaps, *oldcaps;
1126
1127   newcaps = gst_pad_get_current_caps (pad);
1128
1129   GST_INFO ("stream %p received caps %p, %" GST_PTR_FORMAT, stream, newcaps,
1130       newcaps);
1131
1132   g_mutex_lock (&priv->lock);
1133   oldcaps = priv->caps;
1134   priv->caps = newcaps;
1135   g_mutex_unlock (&priv->lock);
1136
1137   if (oldcaps)
1138     gst_caps_unref (oldcaps);
1139 }
1140
1141 static void
1142 dump_structure (const GstStructure * s)
1143 {
1144   gchar *sstr;
1145
1146   sstr = gst_structure_to_string (s);
1147   GST_INFO ("structure: %s", sstr);
1148   g_free (sstr);
1149 }
1150
1151 static GstRTSPStreamTransport *
1152 find_transport (GstRTSPStream * stream, const gchar * rtcp_from)
1153 {
1154   GstRTSPStreamPrivate *priv = stream->priv;
1155   GList *walk;
1156   GstRTSPStreamTransport *result = NULL;
1157   const gchar *tmp;
1158   gchar *dest;
1159   guint port;
1160
1161   if (rtcp_from == NULL)
1162     return NULL;
1163
1164   tmp = g_strrstr (rtcp_from, ":");
1165   if (tmp == NULL)
1166     return NULL;
1167
1168   port = atoi (tmp + 1);
1169   dest = g_strndup (rtcp_from, tmp - rtcp_from);
1170
1171   g_mutex_lock (&priv->lock);
1172   GST_INFO ("finding %s:%d in %d transports", dest, port,
1173       g_list_length (priv->transports));
1174
1175   for (walk = priv->transports; walk; walk = g_list_next (walk)) {
1176     GstRTSPStreamTransport *trans = walk->data;
1177     const GstRTSPTransport *tr;
1178     gint min, max;
1179
1180     tr = gst_rtsp_stream_transport_get_transport (trans);
1181
1182     min = tr->client_port.min;
1183     max = tr->client_port.max;
1184
1185     if ((strcmp (tr->destination, dest) == 0) && (min == port || max == port)) {
1186       result = trans;
1187       break;
1188     }
1189   }
1190   if (result)
1191     g_object_ref (result);
1192   g_mutex_unlock (&priv->lock);
1193
1194   g_free (dest);
1195
1196   return result;
1197 }
1198
1199 static GstRTSPStreamTransport *
1200 check_transport (GObject * source, GstRTSPStream * stream)
1201 {
1202   GstStructure *stats;
1203   GstRTSPStreamTransport *trans;
1204
1205   /* see if we have a stream to match with the origin of the RTCP packet */
1206   trans = g_object_get_qdata (source, ssrc_stream_map_key);
1207   if (trans == NULL) {
1208     g_object_get (source, "stats", &stats, NULL);
1209     if (stats) {
1210       const gchar *rtcp_from;
1211
1212       dump_structure (stats);
1213
1214       rtcp_from = gst_structure_get_string (stats, "rtcp-from");
1215       if ((trans = find_transport (stream, rtcp_from))) {
1216         GST_INFO ("%p: found transport %p for source  %p", stream, trans,
1217             source);
1218         g_object_set_qdata_full (source, ssrc_stream_map_key, trans,
1219             g_object_unref);
1220       }
1221       gst_structure_free (stats);
1222     }
1223   }
1224   return trans;
1225 }
1226
1227
1228 static void
1229 on_new_ssrc (GObject * session, GObject * source, GstRTSPStream * stream)
1230 {
1231   GstRTSPStreamTransport *trans;
1232
1233   GST_INFO ("%p: new source %p", stream, source);
1234
1235   trans = check_transport (source, stream);
1236
1237   if (trans)
1238     GST_INFO ("%p: source %p for transport %p", stream, source, trans);
1239 }
1240
1241 static void
1242 on_ssrc_sdes (GObject * session, GObject * source, GstRTSPStream * stream)
1243 {
1244   GST_INFO ("%p: new SDES %p", stream, source);
1245 }
1246
1247 static void
1248 on_ssrc_active (GObject * session, GObject * source, GstRTSPStream * stream)
1249 {
1250   GstRTSPStreamTransport *trans;
1251
1252   trans = check_transport (source, stream);
1253
1254   if (trans) {
1255     GST_INFO ("%p: source %p in transport %p is active", stream, source, trans);
1256     gst_rtsp_stream_transport_keep_alive (trans);
1257   }
1258 #ifdef DUMP_STATS
1259   {
1260     GstStructure *stats;
1261     g_object_get (source, "stats", &stats, NULL);
1262     if (stats) {
1263       dump_structure (stats);
1264       gst_structure_free (stats);
1265     }
1266   }
1267 #endif
1268 }
1269
1270 static void
1271 on_bye_ssrc (GObject * session, GObject * source, GstRTSPStream * stream)
1272 {
1273   GST_INFO ("%p: source %p bye", stream, source);
1274 }
1275
1276 static void
1277 on_bye_timeout (GObject * session, GObject * source, GstRTSPStream * stream)
1278 {
1279   GstRTSPStreamTransport *trans;
1280
1281   GST_INFO ("%p: source %p bye timeout", stream, source);
1282
1283   if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
1284     gst_rtsp_stream_transport_set_timed_out (trans, TRUE);
1285     g_object_set_qdata (source, ssrc_stream_map_key, NULL);
1286   }
1287 }
1288
1289 static void
1290 on_timeout (GObject * session, GObject * source, GstRTSPStream * stream)
1291 {
1292   GstRTSPStreamTransport *trans;
1293
1294   GST_INFO ("%p: source %p timeout", stream, source);
1295
1296   if ((trans = g_object_get_qdata (source, ssrc_stream_map_key))) {
1297     gst_rtsp_stream_transport_set_timed_out (trans, TRUE);
1298     g_object_set_qdata (source, ssrc_stream_map_key, NULL);
1299   }
1300 }
1301
1302 static GstFlowReturn
1303 handle_new_sample (GstAppSink * sink, gpointer user_data)
1304 {
1305   GstRTSPStreamPrivate *priv;
1306   GList *walk;
1307   GstSample *sample;
1308   GstBuffer *buffer;
1309   GstRTSPStream *stream;
1310
1311   sample = gst_app_sink_pull_sample (sink);
1312   if (!sample)
1313     return GST_FLOW_OK;
1314
1315   stream = (GstRTSPStream *) user_data;
1316   priv = stream->priv;
1317   buffer = gst_sample_get_buffer (sample);
1318
1319   g_mutex_lock (&priv->lock);
1320   for (walk = priv->transports; walk; walk = g_list_next (walk)) {
1321     GstRTSPStreamTransport *tr = (GstRTSPStreamTransport *) walk->data;
1322
1323     if (GST_ELEMENT_CAST (sink) == priv->appsink[0]) {
1324       gst_rtsp_stream_transport_send_rtp (tr, buffer);
1325     } else {
1326       gst_rtsp_stream_transport_send_rtcp (tr, buffer);
1327     }
1328   }
1329   g_mutex_unlock (&priv->lock);
1330
1331   gst_sample_unref (sample);
1332
1333   return GST_FLOW_OK;
1334 }
1335
1336 static GstAppSinkCallbacks sink_cb = {
1337   NULL,                         /* not interested in EOS */
1338   NULL,                         /* not interested in preroll samples */
1339   handle_new_sample,
1340 };
1341
1342 /**
1343  * gst_rtsp_stream_join_bin:
1344  * @stream: a #GstRTSPStream
1345  * @bin: a #GstBin to join
1346  * @rtpbin: a rtpbin element in @bin
1347  * @state: the target state of the new elements
1348  *
1349  * Join the #Gstbin @bin that contains the element @rtpbin.
1350  *
1351  * @stream will link to @rtpbin, which must be inside @bin. The elements
1352  * added to @bin will be set to the state given in @state.
1353  *
1354  * Returns: %TRUE on success.
1355  */
1356 gboolean
1357 gst_rtsp_stream_join_bin (GstRTSPStream * stream, GstBin * bin,
1358     GstElement * rtpbin, GstState state)
1359 {
1360   GstRTSPStreamPrivate *priv;
1361   gint i;
1362   guint idx;
1363   gchar *name;
1364   GstPad *pad, *sinkpad, *selpad;
1365   GstPadLinkReturn ret;
1366
1367   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1368   g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
1369   g_return_val_if_fail (GST_IS_ELEMENT (rtpbin), FALSE);
1370
1371   priv = stream->priv;
1372
1373   g_mutex_lock (&priv->lock);
1374   if (priv->is_joined)
1375     goto was_joined;
1376
1377   /* create a session with the same index as the stream */
1378   idx = priv->idx;
1379
1380   GST_INFO ("stream %p joining bin as session %u", stream, idx);
1381
1382   if (!alloc_ports (stream))
1383     goto no_ports;
1384
1385   /* update the dscp qos field in the sinks */
1386   update_dscp_qos (stream);
1387
1388   /* get a pad for sending RTP */
1389   name = g_strdup_printf ("send_rtp_sink_%u", idx);
1390   priv->send_rtp_sink = gst_element_get_request_pad (rtpbin, name);
1391   g_free (name);
1392   /* link the RTP pad to the session manager, it should not really fail unless
1393    * this is not really an RTP pad */
1394   ret = gst_pad_link (priv->srcpad, priv->send_rtp_sink);
1395   if (ret != GST_PAD_LINK_OK)
1396     goto link_failed;
1397
1398   /* get pads from the RTP session element for sending and receiving
1399    * RTP/RTCP*/
1400   name = g_strdup_printf ("send_rtp_src_%u", idx);
1401   priv->send_src[0] = gst_element_get_static_pad (rtpbin, name);
1402   g_free (name);
1403   name = g_strdup_printf ("send_rtcp_src_%u", idx);
1404   priv->send_src[1] = gst_element_get_request_pad (rtpbin, name);
1405   g_free (name);
1406   name = g_strdup_printf ("recv_rtp_sink_%u", idx);
1407   priv->recv_sink[0] = gst_element_get_request_pad (rtpbin, name);
1408   g_free (name);
1409   name = g_strdup_printf ("recv_rtcp_sink_%u", idx);
1410   priv->recv_sink[1] = gst_element_get_request_pad (rtpbin, name);
1411   g_free (name);
1412
1413   /* get the session */
1414   g_signal_emit_by_name (rtpbin, "get-internal-session", idx, &priv->session);
1415
1416   g_signal_connect (priv->session, "on-new-ssrc", (GCallback) on_new_ssrc,
1417       stream);
1418   g_signal_connect (priv->session, "on-ssrc-sdes", (GCallback) on_ssrc_sdes,
1419       stream);
1420   g_signal_connect (priv->session, "on-ssrc-active",
1421       (GCallback) on_ssrc_active, stream);
1422   g_signal_connect (priv->session, "on-bye-ssrc", (GCallback) on_bye_ssrc,
1423       stream);
1424   g_signal_connect (priv->session, "on-bye-timeout",
1425       (GCallback) on_bye_timeout, stream);
1426   g_signal_connect (priv->session, "on-timeout", (GCallback) on_timeout,
1427       stream);
1428
1429   for (i = 0; i < 2; i++) {
1430     GstPad *teepad, *queuepad;
1431     /* For the sender we create this bit of pipeline for both
1432      * RTP and RTCP. Sync and preroll are enabled on udpsink so
1433      * we need to add a queue before appsink to make the pipeline
1434      * not block. For the TCP case, we want to pump data to the
1435      * client as fast as possible anyway.
1436      *
1437      * .--------.      .-----.    .---------.
1438      * | rtpbin |      | tee |    | udpsink |
1439      * |       send->sink   src->sink       |
1440      * '--------'      |     |    '---------'
1441      *                 |     |    .---------.    .---------.
1442      *                 |     |    |  queue  |    | appsink |
1443      *                 |    src->sink      src->sink       |
1444      *                 '-----'    '---------'    '---------'
1445      *
1446      * When only UDP is allowed, we skip the tee, queue and appsink and link the
1447      * udpsink directly to the session.
1448      */
1449     /* add udpsink */
1450     gst_bin_add (bin, priv->udpsink[i]);
1451     sinkpad = gst_element_get_static_pad (priv->udpsink[i], "sink");
1452
1453     if (priv->protocols & GST_RTSP_LOWER_TRANS_TCP) {
1454       /* make tee for RTP/RTCP */
1455       priv->tee[i] = gst_element_factory_make ("tee", NULL);
1456       gst_bin_add (bin, priv->tee[i]);
1457
1458       /* and link to rtpbin send pad */
1459       pad = gst_element_get_static_pad (priv->tee[i], "sink");
1460       gst_pad_link (priv->send_src[i], pad);
1461       gst_object_unref (pad);
1462
1463       /* link tee to udpsink */
1464       teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
1465       gst_pad_link (teepad, sinkpad);
1466       gst_object_unref (teepad);
1467
1468       /* make queue */
1469       priv->appqueue[i] = gst_element_factory_make ("queue", NULL);
1470       gst_bin_add (bin, priv->appqueue[i]);
1471       /* and link to tee */
1472       teepad = gst_element_get_request_pad (priv->tee[i], "src_%u");
1473       pad = gst_element_get_static_pad (priv->appqueue[i], "sink");
1474       gst_pad_link (teepad, pad);
1475       gst_object_unref (pad);
1476       gst_object_unref (teepad);
1477
1478       /* make appsink */
1479       priv->appsink[i] = gst_element_factory_make ("appsink", NULL);
1480       g_object_set (priv->appsink[i], "async", FALSE, "sync", FALSE, NULL);
1481       g_object_set (priv->appsink[i], "emit-signals", FALSE, NULL);
1482       gst_bin_add (bin, priv->appsink[i]);
1483       gst_app_sink_set_callbacks (GST_APP_SINK_CAST (priv->appsink[i]),
1484           &sink_cb, stream, NULL);
1485       /* and link to queue */
1486       queuepad = gst_element_get_static_pad (priv->appqueue[i], "src");
1487       pad = gst_element_get_static_pad (priv->appsink[i], "sink");
1488       gst_pad_link (queuepad, pad);
1489       gst_object_unref (pad);
1490       gst_object_unref (queuepad);
1491     } else {
1492       /* else only udpsink needed, link it to the session */
1493       gst_pad_link (priv->send_src[i], sinkpad);
1494     }
1495     gst_object_unref (sinkpad);
1496
1497     /* For the receiver we create this bit of pipeline for both
1498      * RTP and RTCP. We receive RTP/RTCP on appsrc and udpsrc
1499      * and it is all funneled into the rtpbin receive pad.
1500      *
1501      * .--------.     .--------.    .--------.
1502      * | udpsrc |     | funnel |    | rtpbin |
1503      * |       src->sink      src->sink      |
1504      * '--------'     |        |    '--------'
1505      * .--------.     |        |
1506      * | appsrc |     |        |
1507      * |       src->sink       |
1508      * '--------'     '--------'
1509      */
1510     /* make funnel for the RTP/RTCP receivers */
1511     priv->funnel[i] = gst_element_factory_make ("funnel", NULL);
1512     gst_bin_add (bin, priv->funnel[i]);
1513
1514     pad = gst_element_get_static_pad (priv->funnel[i], "src");
1515     gst_pad_link (pad, priv->recv_sink[i]);
1516     gst_object_unref (pad);
1517
1518     if (priv->udpsrc_v4[i]) {
1519       /* we set and keep these to playing so that they don't cause NO_PREROLL return
1520        * values */
1521       gst_element_set_state (priv->udpsrc_v4[i], GST_STATE_PLAYING);
1522       gst_element_set_locked_state (priv->udpsrc_v4[i], TRUE);
1523       /* add udpsrc */
1524       gst_bin_add (bin, priv->udpsrc_v4[i]);
1525
1526       /* and link to the funnel v4 */
1527       selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
1528       pad = gst_element_get_static_pad (priv->udpsrc_v4[i], "src");
1529       gst_pad_link (pad, selpad);
1530       gst_object_unref (pad);
1531       gst_object_unref (selpad);
1532     }
1533
1534     if (priv->udpsrc_v6[i]) {
1535       gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_PLAYING);
1536       gst_element_set_locked_state (priv->udpsrc_v6[i], TRUE);
1537       gst_bin_add (bin, priv->udpsrc_v6[i]);
1538
1539       /* and link to the funnel v6 */
1540       selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
1541       pad = gst_element_get_static_pad (priv->udpsrc_v6[i], "src");
1542       gst_pad_link (pad, selpad);
1543       gst_object_unref (pad);
1544       gst_object_unref (selpad);
1545     }
1546
1547     if (priv->protocols & GST_RTSP_LOWER_TRANS_TCP) {
1548       /* make and add appsrc */
1549       priv->appsrc[i] = gst_element_factory_make ("appsrc", NULL);
1550       gst_bin_add (bin, priv->appsrc[i]);
1551       /* and link to the funnel */
1552       selpad = gst_element_get_request_pad (priv->funnel[i], "sink_%u");
1553       pad = gst_element_get_static_pad (priv->appsrc[i], "src");
1554       gst_pad_link (pad, selpad);
1555       gst_object_unref (pad);
1556       gst_object_unref (selpad);
1557     }
1558
1559     /* check if we need to set to a special state */
1560     if (state != GST_STATE_NULL) {
1561       if (priv->udpsink[i])
1562         gst_element_set_state (priv->udpsink[i], state);
1563       if (priv->appsink[i])
1564         gst_element_set_state (priv->appsink[i], state);
1565       if (priv->appqueue[i])
1566         gst_element_set_state (priv->appqueue[i], state);
1567       if (priv->tee[i])
1568         gst_element_set_state (priv->tee[i], state);
1569       if (priv->funnel[i])
1570         gst_element_set_state (priv->funnel[i], state);
1571       if (priv->appsrc[i])
1572         gst_element_set_state (priv->appsrc[i], state);
1573     }
1574   }
1575
1576   /* be notified of caps changes */
1577   priv->caps_sig = g_signal_connect (priv->send_rtp_sink, "notify::caps",
1578       (GCallback) caps_notify, stream);
1579
1580   priv->is_joined = TRUE;
1581   g_mutex_unlock (&priv->lock);
1582
1583   return TRUE;
1584
1585   /* ERRORS */
1586 was_joined:
1587   {
1588     g_mutex_unlock (&priv->lock);
1589     return TRUE;
1590   }
1591 no_ports:
1592   {
1593     g_mutex_unlock (&priv->lock);
1594     GST_WARNING ("failed to allocate ports %u", idx);
1595     return FALSE;
1596   }
1597 link_failed:
1598   {
1599     GST_WARNING ("failed to link stream %u", idx);
1600     gst_object_unref (priv->send_rtp_sink);
1601     priv->send_rtp_sink = NULL;
1602     g_mutex_unlock (&priv->lock);
1603     return FALSE;
1604   }
1605 }
1606
1607 /**
1608  * gst_rtsp_stream_leave_bin:
1609  * @stream: a #GstRTSPStream
1610  * @bin: a #GstBin
1611  * @rtpbin: a rtpbin #GstElement
1612  *
1613  * Remove the elements of @stream from @bin.
1614  *
1615  * Return: %TRUE on success.
1616  */
1617 gboolean
1618 gst_rtsp_stream_leave_bin (GstRTSPStream * stream, GstBin * bin,
1619     GstElement * rtpbin)
1620 {
1621   GstRTSPStreamPrivate *priv;
1622   gint i;
1623
1624   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1625   g_return_val_if_fail (GST_IS_BIN (bin), FALSE);
1626   g_return_val_if_fail (GST_IS_ELEMENT (rtpbin), FALSE);
1627
1628   priv = stream->priv;
1629
1630   g_mutex_lock (&priv->lock);
1631   if (!priv->is_joined)
1632     goto was_not_joined;
1633
1634   /* all transports must be removed by now */
1635   g_return_val_if_fail (priv->transports == NULL, FALSE);
1636
1637   GST_INFO ("stream %p leaving bin", stream);
1638
1639   gst_pad_unlink (priv->srcpad, priv->send_rtp_sink);
1640   g_signal_handler_disconnect (priv->send_rtp_sink, priv->caps_sig);
1641   gst_element_release_request_pad (rtpbin, priv->send_rtp_sink);
1642   gst_object_unref (priv->send_rtp_sink);
1643   priv->send_rtp_sink = NULL;
1644
1645   for (i = 0; i < 2; i++) {
1646     if (priv->udpsink[i])
1647       gst_element_set_state (priv->udpsink[i], GST_STATE_NULL);
1648     if (priv->appsink[i])
1649       gst_element_set_state (priv->appsink[i], GST_STATE_NULL);
1650     if (priv->appqueue[i])
1651       gst_element_set_state (priv->appqueue[i], GST_STATE_NULL);
1652     if (priv->tee[i])
1653       gst_element_set_state (priv->tee[i], GST_STATE_NULL);
1654     if (priv->funnel[i])
1655       gst_element_set_state (priv->funnel[i], GST_STATE_NULL);
1656     if (priv->appsrc[i])
1657       gst_element_set_state (priv->appsrc[i], GST_STATE_NULL);
1658     if (priv->udpsrc_v4[i]) {
1659       /* and set udpsrc to NULL now before removing */
1660       gst_element_set_locked_state (priv->udpsrc_v4[i], FALSE);
1661       gst_element_set_state (priv->udpsrc_v4[i], GST_STATE_NULL);
1662       /* removing them should also nicely release the request
1663        * pads when they finalize */
1664       gst_bin_remove (bin, priv->udpsrc_v4[i]);
1665     }
1666     if (priv->udpsrc_v6[i]) {
1667       gst_element_set_locked_state (priv->udpsrc_v6[i], FALSE);
1668       gst_element_set_state (priv->udpsrc_v6[i], GST_STATE_NULL);
1669       gst_bin_remove (bin, priv->udpsrc_v6[i]);
1670     }
1671     if (priv->udpsink[i])
1672       gst_bin_remove (bin, priv->udpsink[i]);
1673     if (priv->appsrc[i])
1674       gst_bin_remove (bin, priv->appsrc[i]);
1675     if (priv->appsink[i])
1676       gst_bin_remove (bin, priv->appsink[i]);
1677     if (priv->appqueue[i])
1678       gst_bin_remove (bin, priv->appqueue[i]);
1679     if (priv->tee[i])
1680       gst_bin_remove (bin, priv->tee[i]);
1681     if (priv->funnel[i])
1682       gst_bin_remove (bin, priv->funnel[i]);
1683
1684     gst_element_release_request_pad (rtpbin, priv->recv_sink[i]);
1685     gst_object_unref (priv->recv_sink[i]);
1686     priv->recv_sink[i] = NULL;
1687
1688     priv->udpsrc_v4[i] = NULL;
1689     priv->udpsrc_v6[i] = NULL;
1690     priv->udpsink[i] = NULL;
1691     priv->appsrc[i] = NULL;
1692     priv->appsink[i] = NULL;
1693     priv->appqueue[i] = NULL;
1694     priv->tee[i] = NULL;
1695     priv->funnel[i] = NULL;
1696   }
1697   gst_object_unref (priv->send_src[0]);
1698   priv->send_src[0] = NULL;
1699
1700   gst_element_release_request_pad (rtpbin, priv->send_src[1]);
1701   gst_object_unref (priv->send_src[1]);
1702   priv->send_src[1] = NULL;
1703
1704   g_object_unref (priv->session);
1705   priv->session = NULL;
1706   if (priv->caps)
1707     gst_caps_unref (priv->caps);
1708   priv->caps = NULL;
1709
1710   priv->is_joined = FALSE;
1711   g_mutex_unlock (&priv->lock);
1712
1713   return TRUE;
1714
1715 was_not_joined:
1716   {
1717     return TRUE;
1718   }
1719 }
1720
1721 /**
1722  * gst_rtsp_stream_get_rtpinfo:
1723  * @stream: a #GstRTSPStream
1724  * @rtptime: result RTP timestamp
1725  * @seq: result RTP seqnum
1726  *
1727  * Retrieve the current rtptime and seq. This is used to
1728  * construct a RTPInfo reply header.
1729  *
1730  * Returns: %TRUE when rtptime and seq could be determined.
1731  */
1732 gboolean
1733 gst_rtsp_stream_get_rtpinfo (GstRTSPStream * stream,
1734     guint * rtptime, guint * seq)
1735 {
1736   GstRTSPStreamPrivate *priv;
1737   GObjectClass *payobjclass;
1738
1739   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1740   g_return_val_if_fail (rtptime != NULL, FALSE);
1741   g_return_val_if_fail (seq != NULL, FALSE);
1742
1743   priv = stream->priv;
1744
1745   payobjclass = G_OBJECT_GET_CLASS (priv->payloader);
1746
1747   if (!g_object_class_find_property (payobjclass, "seqnum") ||
1748       !g_object_class_find_property (payobjclass, "timestamp"))
1749     return FALSE;
1750
1751   g_object_get (priv->payloader, "seqnum", seq, "timestamp", rtptime, NULL);
1752
1753   return TRUE;
1754 }
1755
1756 /**
1757  * gst_rtsp_stream_get_caps:
1758  * @stream: a #GstRTSPStream
1759  *
1760  * Retrieve the current caps of @stream.
1761  *
1762  * Returns: (transfer full): the #GstCaps of @stream. use gst_caps_unref()
1763  *    after usage.
1764  */
1765 GstCaps *
1766 gst_rtsp_stream_get_caps (GstRTSPStream * stream)
1767 {
1768   GstRTSPStreamPrivate *priv;
1769   GstCaps *result;
1770
1771   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), NULL);
1772
1773   priv = stream->priv;
1774
1775   g_mutex_lock (&priv->lock);
1776   if ((result = priv->caps))
1777     gst_caps_ref (result);
1778   g_mutex_unlock (&priv->lock);
1779
1780   return result;
1781 }
1782
1783 /**
1784  * gst_rtsp_stream_recv_rtp:
1785  * @stream: a #GstRTSPStream
1786  * @buffer: (transfer full): a #GstBuffer
1787  *
1788  * Handle an RTP buffer for the stream. This method is usually called when a
1789  * message has been received from a client using the TCP transport.
1790  *
1791  * This function takes ownership of @buffer.
1792  *
1793  * Returns: a GstFlowReturn.
1794  */
1795 GstFlowReturn
1796 gst_rtsp_stream_recv_rtp (GstRTSPStream * stream, GstBuffer * buffer)
1797 {
1798   GstRTSPStreamPrivate *priv;
1799   GstFlowReturn ret;
1800   GstElement *element;
1801
1802   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), GST_FLOW_ERROR);
1803   priv = stream->priv;
1804   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1805   g_return_val_if_fail (priv->is_joined, FALSE);
1806
1807   g_mutex_lock (&priv->lock);
1808   if (priv->appsrc[0])
1809     element = gst_object_ref (priv->appsrc[0]);
1810   else
1811     element = NULL;
1812   g_mutex_unlock (&priv->lock);
1813
1814   if (element) {
1815     ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (element), buffer);
1816     gst_object_unref (element);
1817   } else {
1818     ret = GST_FLOW_OK;
1819   }
1820   return ret;
1821 }
1822
1823 /**
1824  * gst_rtsp_stream_recv_rtcp:
1825  * @stream: a #GstRTSPStream
1826  * @buffer: (transfer full): a #GstBuffer
1827  *
1828  * Handle an RTCP buffer for the stream. This method is usually called when a
1829  * message has been received from a client using the TCP transport.
1830  *
1831  * This function takes ownership of @buffer.
1832  *
1833  * Returns: a GstFlowReturn.
1834  */
1835 GstFlowReturn
1836 gst_rtsp_stream_recv_rtcp (GstRTSPStream * stream, GstBuffer * buffer)
1837 {
1838   GstRTSPStreamPrivate *priv;
1839   GstFlowReturn ret;
1840   GstElement *element;
1841
1842   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), GST_FLOW_ERROR);
1843   priv = stream->priv;
1844   g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR);
1845   g_return_val_if_fail (priv->is_joined, FALSE);
1846
1847   g_mutex_lock (&priv->lock);
1848   if (priv->appsrc[1])
1849     element = gst_object_ref (priv->appsrc[1]);
1850   else
1851     element = NULL;
1852   g_mutex_unlock (&priv->lock);
1853
1854   if (element) {
1855     ret = gst_app_src_push_buffer (GST_APP_SRC_CAST (element), buffer);
1856     gst_object_unref (element);
1857   } else {
1858     ret = GST_FLOW_OK;
1859   }
1860   return ret;
1861 }
1862
1863 /* must be called with lock */
1864 static gboolean
1865 update_transport (GstRTSPStream * stream, GstRTSPStreamTransport * trans,
1866     gboolean add)
1867 {
1868   GstRTSPStreamPrivate *priv = stream->priv;
1869   const GstRTSPTransport *tr;
1870
1871   tr = gst_rtsp_stream_transport_get_transport (trans);
1872
1873   switch (tr->lower_transport) {
1874     case GST_RTSP_LOWER_TRANS_UDP:
1875     case GST_RTSP_LOWER_TRANS_UDP_MCAST:
1876     {
1877       gchar *dest;
1878       gint min, max;
1879       guint ttl = 0;
1880
1881       dest = tr->destination;
1882       if (tr->lower_transport == GST_RTSP_LOWER_TRANS_UDP_MCAST) {
1883         min = tr->port.min;
1884         max = tr->port.max;
1885         ttl = tr->ttl;
1886       } else {
1887         min = tr->client_port.min;
1888         max = tr->client_port.max;
1889       }
1890
1891       if (add) {
1892         GST_INFO ("adding %s:%d-%d", dest, min, max);
1893         g_signal_emit_by_name (priv->udpsink[0], "add", dest, min, NULL);
1894         g_signal_emit_by_name (priv->udpsink[1], "add", dest, max, NULL);
1895         if (ttl > 0) {
1896           GST_INFO ("setting ttl-mc %d", ttl);
1897           g_object_set (G_OBJECT (priv->udpsink[0]), "ttl-mc", ttl, NULL);
1898           g_object_set (G_OBJECT (priv->udpsink[1]), "ttl-mc", ttl, NULL);
1899         }
1900         priv->transports = g_list_prepend (priv->transports, trans);
1901       } else {
1902         GST_INFO ("removing %s:%d-%d", dest, min, max);
1903         g_signal_emit_by_name (priv->udpsink[0], "remove", dest, min, NULL);
1904         g_signal_emit_by_name (priv->udpsink[1], "remove", dest, max, NULL);
1905         priv->transports = g_list_remove (priv->transports, trans);
1906       }
1907       break;
1908     }
1909     case GST_RTSP_LOWER_TRANS_TCP:
1910       if (add) {
1911         GST_INFO ("adding TCP %s", tr->destination);
1912         priv->transports = g_list_prepend (priv->transports, trans);
1913       } else {
1914         GST_INFO ("removing TCP %s", tr->destination);
1915         priv->transports = g_list_remove (priv->transports, trans);
1916       }
1917       break;
1918     default:
1919       goto unknown_transport;
1920   }
1921   return TRUE;
1922
1923   /* ERRORS */
1924 unknown_transport:
1925   {
1926     GST_INFO ("Unknown transport %d", tr->lower_transport);
1927     return FALSE;
1928   }
1929 }
1930
1931
1932 /**
1933  * gst_rtsp_stream_add_transport:
1934  * @stream: a #GstRTSPStream
1935  * @trans: a #GstRTSPStreamTransport
1936  *
1937  * Add the transport in @trans to @stream. The media of @stream will
1938  * then also be send to the values configured in @trans.
1939  *
1940  * @stream must be joined to a bin.
1941  *
1942  * @trans must contain a valid #GstRTSPTransport.
1943  *
1944  * Returns: %TRUE if @trans was added
1945  */
1946 gboolean
1947 gst_rtsp_stream_add_transport (GstRTSPStream * stream,
1948     GstRTSPStreamTransport * trans)
1949 {
1950   GstRTSPStreamPrivate *priv;
1951   gboolean res;
1952
1953   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1954   priv = stream->priv;
1955   g_return_val_if_fail (GST_IS_RTSP_STREAM_TRANSPORT (trans), FALSE);
1956   g_return_val_if_fail (priv->is_joined, FALSE);
1957
1958   g_mutex_lock (&priv->lock);
1959   res = update_transport (stream, trans, TRUE);
1960   g_mutex_unlock (&priv->lock);
1961
1962   return res;
1963 }
1964
1965 /**
1966  * gst_rtsp_stream_remove_transport:
1967  * @stream: a #GstRTSPStream
1968  * @trans: a #GstRTSPStreamTransport
1969  *
1970  * Remove the transport in @trans from @stream. The media of @stream will
1971  * not be sent to the values configured in @trans.
1972  *
1973  * @stream must be joined to a bin.
1974  *
1975  * @trans must contain a valid #GstRTSPTransport.
1976  *
1977  * Returns: %TRUE if @trans was removed
1978  */
1979 gboolean
1980 gst_rtsp_stream_remove_transport (GstRTSPStream * stream,
1981     GstRTSPStreamTransport * trans)
1982 {
1983   GstRTSPStreamPrivate *priv;
1984   gboolean res;
1985
1986   g_return_val_if_fail (GST_IS_RTSP_STREAM (stream), FALSE);
1987   priv = stream->priv;
1988   g_return_val_if_fail (GST_IS_RTSP_STREAM_TRANSPORT (trans), FALSE);
1989   g_return_val_if_fail (priv->is_joined, FALSE);
1990
1991   g_mutex_lock (&priv->lock);
1992   res = update_transport (stream, trans, FALSE);
1993   g_mutex_unlock (&priv->lock);
1994
1995   return res;
1996 }