gst/rtsp-sink/gstrtspclientsink.c

   1 /* GStreamer
   2  * Copyright (C) <2005,2006> Wim Taymans <wim at fluendo dot com>
   3  *               <2006> Lutz Mueller <lutz at topfrose dot de>
   4  *               <2015> Jan Schmidt <jan at centricular dot com>
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  19  * Boston, MA 02110-1301, USA.
  20  */
  21 /*
  22  * Unless otherwise indicated, Source Code is licensed under MIT license.
  23  * See further explanation attached in License Statement (distributed in the file
  24  * LICENSE).
  25  *
  26  * Permission is hereby granted, free of charge, to any person obtaining a copy of
  27  * this software and associated documentation files (the "Software"), to deal in
  28  * the Software without restriction, including without limitation the rights to
  29  * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
  30  * of the Software, and to permit persons to whom the Software is furnished to do
  31  * so, subject to the following conditions:
  32  *
  33  * The above copyright notice and this permission notice shall be included in all
  34  * copies or substantial portions of the Software.
  35  *
  36  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  37  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  38  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  39  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  40  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  41  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  42  * SOFTWARE.
  43  */
  44 /**
  45  * SECTION:element-rtspclientsink
  46  *
  47  * Makes a connection to an RTSP server and send data via RTSP RECORD.
  48  * rtspclientsink strictly follows RFC 2326
  49  *
  50  * RTSP supports transport over TCP or UDP in unicast or multicast mode. By
  51  * default rtspclientsink will negotiate a connection in the following order:
  52  * UDP unicast/UDP multicast/TCP. The order cannot be changed but the allowed
  53  * protocols can be controlled with the #GstRTSPClientSink:protocols property.
  54  *
  55  * rtspclientsink will internally instantiate an RTP session manager element
  56  * that will handle the RTCP messages to and from the server, jitter removal,
  57  * and packet reordering.
  58  * This feature is implemented using the gstrtpbin element.
  59  *
  60  * rtspclientsink accepts any stream for which there is an installed payloader,
  61  * creates the payloader and manages payload-types, as well as RTX setup.
  62  * The new-payloader signal is fired when a payloader is created, in case
  63  * an app wants to do custom configuration (such as for MTU).
  64  *
  65  * <refsect2>
  66  * <title>Example launch line</title>
  67  * |[
  68  * gst-launch-1.0 videotestsrc ! jpegenc ! rtspclientsink location=rtsp://some.server/url
  69  * ]| Establish a connection to an RTSP server and send JPEG encoded video packets
  70  * </refsect2>
  71  */
  72
  73 /* FIXMEs
  74  * - Handle EOS properly and shutdown. The problem with EOS is we don't know
  75  *   when the server has received all data, so we don't know when to do teardown.
  76  *   At the moment, we forward EOS to the app as soon as we stop sending. Is there
  77  *   a way to know from the receiver that it's got all data? Some session timeout?
  78  * - Implement extension support for Real / WMS if they support RECORD?
  79  * - Add support for network clock synchronised streaming?
  80  * - Fix crypto key nego so SAVP/SAVPF profiles work.
  81  * - Test (&fix?) HTTP tunnel support
  82  * - Add an address pool object for GstRTSPStreams to use for multicast
  83  * - Test multicast UDP transport
  84  */
  85
  86 #ifdef HAVE_CONFIG_H
  87 #include "config.h"
  88 #endif
  89
  90 #ifdef HAVE_UNISTD_H
  91 #include <unistd.h>
  92 #endif /* HAVE_UNISTD_H */
  93 #include <stdlib.h>
  94 #include <string.h>
  95 #include <stdio.h>
  96 #include <stdarg.h>
  97
  98 #include <gst/net/gstnet.h>
  99 #include <gst/sdp/gstsdpmessage.h>
 100 #include <gst/sdp/gstmikey.h>
 101 #include <gst/rtp/rtp.h>
 102
 103 #include "gstrtspclientsink.h"
 104
 105 GST_DEBUG_CATEGORY_STATIC (rtsp_client_sink_debug);
 106 #define GST_CAT_DEFAULT (rtsp_client_sink_debug)
 107
 108 static GstStaticPadTemplate rtptemplate = GST_STATIC_PAD_TEMPLATE ("stream_%u",
 109     GST_PAD_SINK,
 110     GST_PAD_REQUEST,
 111     GST_STATIC_CAPS_ANY);       /* Actual caps come from available set of payloaders */
 112
 113 enum
 114 {
 115   SIGNAL_HANDLE_REQUEST,
 116   SIGNAL_NEW_MANAGER,
 117   SIGNAL_NEW_PAYLOADER,
 118   SIGNAL_REQUEST_RTCP_KEY,
 119   LAST_SIGNAL
 120 };
 121
 122 enum _GstRTSPClientSinkNtpTimeSource
 123 {
 124   NTP_TIME_SOURCE_NTP,
 125   NTP_TIME_SOURCE_UNIX,
 126   NTP_TIME_SOURCE_RUNNING_TIME,
 127   NTP_TIME_SOURCE_CLOCK_TIME
 128 };
 129
 130 #define GST_TYPE_RTSP_CLIENT_SINK_NTP_TIME_SOURCE (gst_rtsp_client_sink_ntp_time_source_get_type())
 131 static GType
 132 gst_rtsp_client_sink_ntp_time_source_get_type (void)
 133 {
 134   static GType ntp_time_source_type = 0;
 135   static const GEnumValue ntp_time_source_values[] = {
 136     {NTP_TIME_SOURCE_NTP, "NTP time based on realtime clock", "ntp"},
 137     {NTP_TIME_SOURCE_UNIX, "UNIX time based on realtime clock", "unix"},
 138     {NTP_TIME_SOURCE_RUNNING_TIME,
 139           "Running time based on pipeline clock",
 140         "running-time"},
 141     {NTP_TIME_SOURCE_CLOCK_TIME, "Pipeline clock time", "clock-time"},
 142     {0, NULL, NULL},
 143   };
 144
 145   if (!ntp_time_source_type) {
 146     ntp_time_source_type =
 147         g_enum_register_static ("GstRTSPClientSinkNtpTimeSource",
 148         ntp_time_source_values);
 149   }
 150   return ntp_time_source_type;
 151 }
 152
 153 #define DEFAULT_LOCATION         NULL
 154 #define DEFAULT_PROTOCOLS        GST_RTSP_LOWER_TRANS_UDP | GST_RTSP_LOWER_TRANS_UDP_MCAST | GST_RTSP_LOWER_TRANS_TCP
 155 #define DEFAULT_DEBUG            FALSE
 156 #define DEFAULT_RETRY            20
 157 #define DEFAULT_TIMEOUT          5000000
 158 #define DEFAULT_UDP_BUFFER_SIZE  0x80000
 159 #define DEFAULT_TCP_TIMEOUT      20000000
 160 #define DEFAULT_LATENCY_MS       2000
 161 #define DEFAULT_DO_RTSP_KEEP_ALIVE       TRUE
 162 #define DEFAULT_PROXY            NULL
 163 #define DEFAULT_RTP_BLOCKSIZE    0
 164 #define DEFAULT_USER_ID          NULL
 165 #define DEFAULT_USER_PW          NULL
 166 #define DEFAULT_PORT_RANGE       NULL
 167 #define DEFAULT_UDP_RECONNECT    TRUE
 168 #define DEFAULT_MULTICAST_IFACE  NULL
 169 #define DEFAULT_TLS_VALIDATION_FLAGS     G_TLS_CERTIFICATE_VALIDATE_ALL
 170 #define DEFAULT_TLS_DATABASE     NULL
 171 #define DEFAULT_TLS_INTERACTION     NULL
 172 #define DEFAULT_NTP_TIME_SOURCE  NTP_TIME_SOURCE_NTP
 173 #define DEFAULT_USER_AGENT       "GStreamer/" PACKAGE_VERSION
 174 #define DEFAULT_PROFILES         GST_RTSP_PROFILE_AVP
 175 #define DEFAULT_RTX_TIME_MS      500
 176
 177 enum
 178 {
 179   PROP_0,
 180   PROP_LOCATION,
 181   PROP_PROTOCOLS,
 182   PROP_DEBUG,
 183   PROP_RETRY,
 184   PROP_TIMEOUT,
 185   PROP_TCP_TIMEOUT,
 186   PROP_LATENCY,
 187   PROP_RTX_TIME,
 188   PROP_DO_RTSP_KEEP_ALIVE,
 189   PROP_PROXY,
 190   PROP_PROXY_ID,
 191   PROP_PROXY_PW,
 192   PROP_RTP_BLOCKSIZE,
 193   PROP_USER_ID,
 194   PROP_USER_PW,
 195   PROP_PORT_RANGE,
 196   PROP_UDP_BUFFER_SIZE,
 197   PROP_UDP_RECONNECT,
 198   PROP_MULTICAST_IFACE,
 199   PROP_SDES,
 200   PROP_TLS_VALIDATION_FLAGS,
 201   PROP_TLS_DATABASE,
 202   PROP_TLS_INTERACTION,
 203   PROP_NTP_TIME_SOURCE,
 204   PROP_USER_AGENT,
 205   PROP_PROFILES
 206 };
 207
 208 static void gst_rtsp_client_sink_finalize (GObject * object);
 209
 210 static void gst_rtsp_client_sink_set_property (GObject * object, guint prop_id,
 211     const GValue * value, GParamSpec * pspec);
 212 static void gst_rtsp_client_sink_get_property (GObject * object, guint prop_id,
 213     GValue * value, GParamSpec * pspec);
 214
 215 static GstClock *gst_rtsp_client_sink_provide_clock (GstElement * element);
 216
 217 static void gst_rtsp_client_sink_uri_handler_init (gpointer g_iface,
 218     gpointer iface_data);
 219
 220 static gboolean gst_rtsp_client_sink_set_proxy (GstRTSPClientSink * rtsp,
 221     const gchar * proxy);
 222 static void gst_rtsp_client_sink_set_tcp_timeout (GstRTSPClientSink *
 223     rtsp_client_sink, guint64 timeout);
 224
 225 static GstStateChangeReturn gst_rtsp_client_sink_change_state (GstElement *
 226     element, GstStateChange transition);
 227 static void gst_rtsp_client_sink_handle_message (GstBin * bin,
 228     GstMessage * message);
 229
 230 static gboolean gst_rtsp_client_sink_setup_auth (GstRTSPClientSink * sink,
 231     GstRTSPMessage * response);
 232
 233 static gboolean gst_rtsp_client_sink_loop_send_cmd (GstRTSPClientSink * sink,
 234     gint cmd, gint mask);
 235
 236 static GstRTSPResult gst_rtsp_client_sink_open (GstRTSPClientSink * sink,
 237     gboolean async);
 238 static GstRTSPResult gst_rtsp_client_sink_record (GstRTSPClientSink * sink,
 239     gboolean async);
 240 static GstRTSPResult gst_rtsp_client_sink_pause (GstRTSPClientSink * sink,
 241     gboolean async);
 242 static GstRTSPResult gst_rtsp_client_sink_close (GstRTSPClientSink * sink,
 243     gboolean async, gboolean only_close);
 244 static gboolean gst_rtsp_client_sink_collect_streams (GstRTSPClientSink * sink);
 245
 246 static gboolean gst_rtsp_client_sink_uri_set_uri (GstURIHandler * handler,
 247     const gchar * uri, GError ** error);
 248 static gchar *gst_rtsp_client_sink_uri_get_uri (GstURIHandler * handler);
 249
 250 static gboolean gst_rtsp_client_sink_loop (GstRTSPClientSink * sink);
 251 static void gst_rtsp_client_sink_connection_flush (GstRTSPClientSink * sink,
 252     gboolean flush);
 253
 254 static GstPad *gst_rtsp_client_sink_request_new_pad (GstElement * element,
 255     GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
 256 static void gst_rtsp_client_sink_release_pad (GstElement * element,
 257     GstPad * pad);
 258
 259 /* commands we send to out loop to notify it of events */
 260 #define CMD_OPEN        (1 << 0)
 261 #define CMD_RECORD      (1 << 1)
 262 #define CMD_PAUSE       (1 << 2)
 263 #define CMD_CLOSE       (1 << 3)
 264 #define CMD_WAIT        (1 << 4)
 265 #define CMD_RECONNECT   (1 << 5)
 266 #define CMD_LOOP        (1 << 6)
 267
 268 /* mask for all commands */
 269 #define CMD_ALL         ((CMD_LOOP << 1) - 1)
 270
 271 #define GST_ELEMENT_PROGRESS(el, type, code, text)      \
 272 G_STMT_START {                                          \
 273   gchar *__txt = _gst_element_error_printf text;        \
 274   gst_element_post_message (GST_ELEMENT_CAST (el),      \
 275       gst_message_new_progress (GST_OBJECT_CAST (el),   \
 276           GST_PROGRESS_TYPE_ ##type, code, __txt));     \
 277   g_free (__txt);                                       \
 278 } G_STMT_END
 279
 280 static guint gst_rtsp_client_sink_signals[LAST_SIGNAL] = { 0 };
 281
 282 #define gst_rtsp_client_sink_parent_class parent_class
 283 G_DEFINE_TYPE_WITH_CODE (GstRTSPClientSink, gst_rtsp_client_sink, GST_TYPE_BIN,
 284     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER,
 285         gst_rtsp_client_sink_uri_handler_init));
 286
 287 #ifndef GST_DISABLE_GST_DEBUG
 288 static inline const gchar *
 289 cmd_to_string (guint cmd)
 290 {
 291   switch (cmd) {
 292     case CMD_OPEN:
 293       return "OPEN";
 294     case CMD_RECORD:
 295       return "RECORD";
 296     case CMD_PAUSE:
 297       return "PAUSE";
 298     case CMD_CLOSE:
 299       return "CLOSE";
 300     case CMD_WAIT:
 301       return "WAIT";
 302     case CMD_RECONNECT:
 303       return "RECONNECT";
 304     case CMD_LOOP:
 305       return "LOOP";
 306   }
 307
 308   return "unknown";
 309 }
 310 #endif
 311
 312 static void
 313 gst_rtsp_client_sink_class_init (GstRTSPClientSinkClass * klass)
 314 {
 315   GObjectClass *gobject_class;
 316   GstElementClass *gstelement_class;
 317   GstBinClass *gstbin_class;
 318
 319   gobject_class = (GObjectClass *) klass;
 320   gstelement_class = (GstElementClass *) klass;
 321   gstbin_class = (GstBinClass *) klass;
 322
 323   GST_DEBUG_CATEGORY_INIT (rtsp_client_sink_debug, "rtspclientsink", 0,
 324       "RTSP sink element");
 325
 326   gobject_class->set_property = gst_rtsp_client_sink_set_property;
 327   gobject_class->get_property = gst_rtsp_client_sink_get_property;
 328
 329   gobject_class->finalize = gst_rtsp_client_sink_finalize;
 330
 331   g_object_class_install_property (gobject_class, PROP_LOCATION,
 332       g_param_spec_string ("location", "RTSP Location",
 333           "Location of the RTSP url to read",
 334           DEFAULT_LOCATION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 335
 336   g_object_class_install_property (gobject_class, PROP_PROTOCOLS,
 337       g_param_spec_flags ("protocols", "Protocols",
 338           "Allowed lower transport protocols", GST_TYPE_RTSP_LOWER_TRANS,
 339           DEFAULT_PROTOCOLS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 340
 341   g_object_class_install_property (gobject_class, PROP_PROFILES,
 342       g_param_spec_flags ("profiles", "Profiles",
 343           "Allowed RTSP profiles", GST_TYPE_RTSP_PROFILE,
 344           DEFAULT_PROFILES, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 345
 346   g_object_class_install_property (gobject_class, PROP_DEBUG,
 347       g_param_spec_boolean ("debug", "Debug",
 348           "Dump request and response messages to stdout",
 349           DEFAULT_DEBUG, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 350
 351   g_object_class_install_property (gobject_class, PROP_RETRY,
 352       g_param_spec_uint ("retry", "Retry",
 353           "Max number of retries when allocating RTP ports.",
 354           0, G_MAXUINT16, DEFAULT_RETRY,
 355           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 356
 357   g_object_class_install_property (gobject_class, PROP_TIMEOUT,
 358       g_param_spec_uint64 ("timeout", "Timeout",
 359           "Retry TCP transport after UDP timeout microseconds (0 = disabled)",
 360           0, G_MAXUINT64, DEFAULT_TIMEOUT,
 361           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 362
 363   g_object_class_install_property (gobject_class, PROP_TCP_TIMEOUT,
 364       g_param_spec_uint64 ("tcp-timeout", "TCP Timeout",
 365           "Fail after timeout microseconds on TCP connections (0 = disabled)",
 366           0, G_MAXUINT64, DEFAULT_TCP_TIMEOUT,
 367           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 368
 369   g_object_class_install_property (gobject_class, PROP_LATENCY,
 370       g_param_spec_uint ("latency", "Buffer latency in ms",
 371           "Amount of ms to buffer", 0, G_MAXUINT, DEFAULT_LATENCY_MS,
 372           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 373
 374   g_object_class_install_property (gobject_class, PROP_RTX_TIME,
 375       g_param_spec_uint ("rtx-time", "Retransmission buffer in ms",
 376           "Amount of ms to buffer for retransmission. 0 disables retransmission",
 377           0, G_MAXUINT, DEFAULT_RTX_TIME_MS,
 378           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 379
 380   /**
 381    * GstRTSPClientSink:do-rtsp-keep-alive:
 382    *
 383    * Enable RTSP keep alive support. Some old server don't like RTSP
 384    * keep alive and then this property needs to be set to FALSE.
 385    */
 386   g_object_class_install_property (gobject_class, PROP_DO_RTSP_KEEP_ALIVE,
 387       g_param_spec_boolean ("do-rtsp-keep-alive", "Do RTSP Keep Alive",
 388           "Send RTSP keep alive packets, disable for old incompatible server.",
 389           DEFAULT_DO_RTSP_KEEP_ALIVE,
 390           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 391
 392   /**
 393    * GstRTSPClientSink:proxy:
 394    *
 395    * Set the proxy parameters. This has to be a string of the format
 396    * [http://][user:passwd@]host[:port].
 397    */
 398   g_object_class_install_property (gobject_class, PROP_PROXY,
 399       g_param_spec_string ("proxy", "Proxy",
 400           "Proxy settings for HTTP tunneling. Format: [http://][user:passwd@]host[:port]",
 401           DEFAULT_PROXY, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 402   /**
 403    * GstRTSPClientSink:proxy-id:
 404    *
 405    * Sets the proxy URI user id for authentication. If the URI set via the
 406    * "proxy" property contains a user-id already, that will take precedence.
 407    *
 408    */
 409   g_object_class_install_property (gobject_class, PROP_PROXY_ID,
 410       g_param_spec_string ("proxy-id", "proxy-id",
 411           "HTTP proxy URI user id for authentication", "",
 412           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 413   /**
 414    * GstRTSPClientSink:proxy-pw:
 415    *
 416    * Sets the proxy URI password for authentication. If the URI set via the
 417    * "proxy" property contains a password already, that will take precedence.
 418    *
 419    */
 420   g_object_class_install_property (gobject_class, PROP_PROXY_PW,
 421       g_param_spec_string ("proxy-pw", "proxy-pw",
 422           "HTTP proxy URI user password for authentication", "",
 423           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 424
 425   /**
 426    * GstRTSPClientSink:rtp-blocksize:
 427    *
 428    * RTP package size to suggest to server.
 429    */
 430   g_object_class_install_property (gobject_class, PROP_RTP_BLOCKSIZE,
 431       g_param_spec_uint ("rtp-blocksize", "RTP Blocksize",
 432           "RTP package size to suggest to server (0 = disabled)",
 433           0, 65536, DEFAULT_RTP_BLOCKSIZE,
 434           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 435
 436   g_object_class_install_property (gobject_class,
 437       PROP_USER_ID,
 438       g_param_spec_string ("user-id", "user-id",
 439           "RTSP location URI user id for authentication", DEFAULT_USER_ID,
 440           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 441   g_object_class_install_property (gobject_class, PROP_USER_PW,
 442       g_param_spec_string ("user-pw", "user-pw",
 443           "RTSP location URI user password for authentication", DEFAULT_USER_PW,
 444           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 445
 446   /**
 447    * GstRTSPClientSink:port-range:
 448    *
 449    * Configure the client port numbers that can be used to receive
 450    * RTCP.
 451    */
 452   g_object_class_install_property (gobject_class, PROP_PORT_RANGE,
 453       g_param_spec_string ("port-range", "Port range",
 454           "Client port range that can be used to receive RTCP data, "
 455           "eg. 3000-3005 (NULL = no restrictions)", DEFAULT_PORT_RANGE,
 456           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 457
 458   /**
 459    * GstRTSPClientSink:udp-buffer-size:
 460    *
 461    * Size of the kernel UDP receive buffer in bytes.
 462    */
 463   g_object_class_install_property (gobject_class, PROP_UDP_BUFFER_SIZE,
 464       g_param_spec_int ("udp-buffer-size", "UDP Buffer Size",
 465           "Size of the kernel UDP receive buffer in bytes, 0=default",
 466           0, G_MAXINT, DEFAULT_UDP_BUFFER_SIZE,
 467           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 468
 469   g_object_class_install_property (gobject_class, PROP_UDP_RECONNECT,
 470       g_param_spec_boolean ("udp-reconnect", "Reconnect to the server",
 471           "Reconnect to the server if RTSP connection is closed when doing UDP",
 472           DEFAULT_UDP_RECONNECT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 473
 474   g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
 475       g_param_spec_string ("multicast-iface", "Multicast Interface",
 476           "The network interface on which to join the multicast group",
 477           DEFAULT_MULTICAST_IFACE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 478
 479   g_object_class_install_property (gobject_class, PROP_SDES,
 480       g_param_spec_boxed ("sdes", "SDES",
 481           "The SDES items of this session",
 482           GST_TYPE_STRUCTURE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 483
 484   /**
 485    * GstRTSPClientSink::tls-validation-flags:
 486    *
 487    * TLS certificate validation flags used to validate server
 488    * certificate.
 489    *
 490    */
 491   g_object_class_install_property (gobject_class, PROP_TLS_VALIDATION_FLAGS,
 492       g_param_spec_flags ("tls-validation-flags", "TLS validation flags",
 493           "TLS certificate validation flags used to validate the server certificate",
 494           G_TYPE_TLS_CERTIFICATE_FLAGS, DEFAULT_TLS_VALIDATION_FLAGS,
 495           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 496
 497   /**
 498    * GstRTSPClientSink::tls-database:
 499    *
 500    * TLS database with anchor certificate authorities used to validate
 501    * the server certificate.
 502    *
 503    */
 504   g_object_class_install_property (gobject_class, PROP_TLS_DATABASE,
 505       g_param_spec_object ("tls-database", "TLS database",
 506           "TLS database with anchor certificate authorities used to validate the server certificate",
 507           G_TYPE_TLS_DATABASE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 508
 509   /**
 510    * GstRTSPClientSink::tls-interaction:
 511    *
 512    * A #GTlsInteraction object to be used when the connection or certificate
 513    * database need to interact with the user. This will be used to prompt the
 514    * user for passwords where necessary.
 515    *
 516    */
 517   g_object_class_install_property (gobject_class, PROP_TLS_INTERACTION,
 518       g_param_spec_object ("tls-interaction", "TLS interaction",
 519           "A GTlsInteraction object to prompt the user for password or certificate",
 520           G_TYPE_TLS_INTERACTION, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 521
 522   /**
 523    * GstRTSPClientSink::ntp-time-source:
 524    *
 525    * allows to select the time source that should be used
 526    * for the NTP time in outgoing packets
 527    *
 528    */
 529   g_object_class_install_property (gobject_class, PROP_NTP_TIME_SOURCE,
 530       g_param_spec_enum ("ntp-time-source", "NTP Time Source",
 531           "NTP time source for RTCP packets",
 532           GST_TYPE_RTSP_CLIENT_SINK_NTP_TIME_SOURCE, DEFAULT_NTP_TIME_SOURCE,
 533           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 534
 535   /**
 536    * GstRTSPClientSink::user-agent:
 537    *
 538    * The string to set in the User-Agent header.
 539    *
 540    */
 541   g_object_class_install_property (gobject_class, PROP_USER_AGENT,
 542       g_param_spec_string ("user-agent", "User Agent",
 543           "The User-Agent string to send to the server",
 544           DEFAULT_USER_AGENT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 545
 546   /**
 547    * GstRTSPClientSink::handle-request:
 548    * @rtsp_client_sink: a #GstRTSPClientSink
 549    * @request: a #GstRTSPMessage
 550    * @response: a #GstRTSPMessage
 551    *
 552    * Handle a server request in @request and prepare @response.
 553    *
 554    * This signal is called from the streaming thread, you should therefore not
 555    * do any state changes on @rtsp_client_sink because this might deadlock. If you want
 556    * to modify the state as a result of this signal, post a
 557    * #GST_MESSAGE_REQUEST_STATE message on the bus or signal the main thread
 558    * in some other way.
 559    *
 560    */
 561   gst_rtsp_client_sink_signals[SIGNAL_HANDLE_REQUEST] =
 562       g_signal_new ("handle-request", G_TYPE_FROM_CLASS (klass), 0,
 563       0, NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2,
 564       G_TYPE_POINTER, G_TYPE_POINTER);
 565
 566   /**
 567    * GstRTSPClientSink::new-manager:
 568    * @rtsp_client_sink: a #GstRTSPClientSink
 569    * @manager: a #GstElement
 570    *
 571    * Emitted after a new manager (like rtpbin) was created and the default
 572    * properties were configured.
 573    *
 574    */
 575   gst_rtsp_client_sink_signals[SIGNAL_NEW_MANAGER] =
 576       g_signal_new_class_handler ("new-manager", G_TYPE_FROM_CLASS (klass),
 577       G_SIGNAL_RUN_FIRST | G_SIGNAL_RUN_CLEANUP, 0, NULL, NULL,
 578       g_cclosure_marshal_generic, G_TYPE_NONE, 1, GST_TYPE_ELEMENT);
 579
 580   /**
 581    * GstRTSPClientSink::new-payloader:
 582    * @rtsp_client_sink: a #GstRTSPClientSink
 583    * @payloader: a #GstElement
 584    *
 585    * Emitted after a new RTP payloader was created and the default
 586    * properties were configured.
 587    *
 588    */
 589   gst_rtsp_client_sink_signals[SIGNAL_NEW_PAYLOADER] =
 590       g_signal_new_class_handler ("new-payloader", G_TYPE_FROM_CLASS (klass),
 591       G_SIGNAL_RUN_FIRST | G_SIGNAL_RUN_CLEANUP, 0, NULL, NULL,
 592       g_cclosure_marshal_generic, G_TYPE_NONE, 1, GST_TYPE_ELEMENT);
 593
 594   /**
 595    * GstRTSPClientSink::request-rtcp-key:
 596    * @rtsp_client_sink: a #GstRTSPClientSink
 597    * @num: the stream number
 598    *
 599    * Signal emitted to get the crypto parameters relevant to the RTCP
 600    * stream. User should provide the key and the RTCP encryption ciphers
 601    * and authentication, and return them wrapped in a GstCaps.
 602    *
 603    */
 604   gst_rtsp_client_sink_signals[SIGNAL_REQUEST_RTCP_KEY] =
 605       g_signal_new ("request-rtcp-key", G_TYPE_FROM_CLASS (klass),
 606       G_SIGNAL_RUN_LAST, 0, NULL, NULL, NULL, GST_TYPE_CAPS, 1, G_TYPE_UINT);
 607
 608   gstelement_class->provide_clock = gst_rtsp_client_sink_provide_clock;
 609   gstelement_class->change_state = gst_rtsp_client_sink_change_state;
 610   gstelement_class->request_new_pad =
 611       GST_DEBUG_FUNCPTR (gst_rtsp_client_sink_request_new_pad);
 612   gstelement_class->release_pad =
 613       GST_DEBUG_FUNCPTR (gst_rtsp_client_sink_release_pad);
 614
 615   gst_element_class_add_pad_template (gstelement_class,
 616       gst_static_pad_template_get (&rtptemplate));
 617
 618   gst_element_class_set_static_metadata (gstelement_class,
 619       "RTSP RECORD client", "Sink/Network",
 620       "Send data over the network via RTSP RECORD(RFC 2326)",
 621       "Jan Schmidt <jan@centricular.com>");
 622
 623   gstbin_class->handle_message = gst_rtsp_client_sink_handle_message;
 624 }
 625
 626 static void
 627 gst_rtsp_client_sink_init (GstRTSPClientSink * sink)
 628 {
 629   sink->conninfo.location = g_strdup (DEFAULT_LOCATION);
 630   sink->protocols = DEFAULT_PROTOCOLS;
 631   sink->debug = DEFAULT_DEBUG;
 632   sink->retry = DEFAULT_RETRY;
 633   sink->udp_timeout = DEFAULT_TIMEOUT;
 634   gst_rtsp_client_sink_set_tcp_timeout (sink, DEFAULT_TCP_TIMEOUT);
 635   sink->latency = DEFAULT_LATENCY_MS;
 636   sink->rtx_time = DEFAULT_RTX_TIME_MS;
 637   sink->do_rtsp_keep_alive = DEFAULT_DO_RTSP_KEEP_ALIVE;
 638   gst_rtsp_client_sink_set_proxy (sink, DEFAULT_PROXY);
 639   sink->rtp_blocksize = DEFAULT_RTP_BLOCKSIZE;
 640   sink->user_id = g_strdup (DEFAULT_USER_ID);
 641   sink->user_pw = g_strdup (DEFAULT_USER_PW);
 642   sink->client_port_range.min = 0;
 643   sink->client_port_range.max = 0;
 644   sink->udp_buffer_size = DEFAULT_UDP_BUFFER_SIZE;
 645   sink->udp_reconnect = DEFAULT_UDP_RECONNECT;
 646   sink->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
 647   sink->sdes = NULL;
 648   sink->tls_validation_flags = DEFAULT_TLS_VALIDATION_FLAGS;
 649   sink->tls_database = DEFAULT_TLS_DATABASE;
 650   sink->tls_interaction = DEFAULT_TLS_INTERACTION;
 651   sink->ntp_time_source = DEFAULT_NTP_TIME_SOURCE;
 652   sink->user_agent = g_strdup (DEFAULT_USER_AGENT);
 653
 654   sink->profiles = DEFAULT_PROFILES;
 655
 656   /* protects the streaming thread in interleaved mode or the polling
 657    * thread in UDP mode. */
 658   g_rec_mutex_init (&sink->stream_rec_lock);
 659
 660   /* protects our state changes from multiple invocations */
 661   g_rec_mutex_init (&sink->state_rec_lock);
 662
 663   g_mutex_init (&sink->send_lock);
 664
 665   g_mutex_init (&sink->preroll_lock);
 666   g_cond_init (&sink->preroll_cond);
 667
 668   sink->state = GST_RTSP_STATE_INVALID;
 669
 670   sink->internal_bin = (GstBin *) gst_bin_new ("rtspbin");
 671   gst_element_set_locked_state (GST_ELEMENT_CAST (sink->internal_bin), TRUE);
 672   gst_bin_add (GST_BIN (sink), GST_ELEMENT_CAST (sink->internal_bin));
 673
 674   sink->next_dyn_pt = 96;
 675
 676   gst_sdp_message_init (&sink->cursdp);
 677
 678   GST_OBJECT_FLAG_SET (sink, GST_ELEMENT_FLAG_SINK);
 679 }
 680
 681 static void
 682 gst_rtsp_client_sink_finalize (GObject * object)
 683 {
 684   GstRTSPClientSink *rtsp_client_sink;
 685
 686   rtsp_client_sink = GST_RTSP_CLIENT_SINK (object);
 687
 688   gst_sdp_message_uninit (&rtsp_client_sink->cursdp);
 689
 690   g_free (rtsp_client_sink->conninfo.location);
 691   gst_rtsp_url_free (rtsp_client_sink->conninfo.url);
 692   g_free (rtsp_client_sink->conninfo.url_str);
 693   g_free (rtsp_client_sink->user_id);
 694   g_free (rtsp_client_sink->user_pw);
 695   g_free (rtsp_client_sink->multi_iface);
 696   g_free (rtsp_client_sink->user_agent);
 697
 698   if (rtsp_client_sink->uri_sdp) {
 699     gst_sdp_message_free (rtsp_client_sink->uri_sdp);
 700     rtsp_client_sink->uri_sdp = NULL;
 701   }
 702   if (rtsp_client_sink->provided_clock)
 703     gst_object_unref (rtsp_client_sink->provided_clock);
 704
 705   if (rtsp_client_sink->sdes)
 706     gst_structure_free (rtsp_client_sink->sdes);
 707
 708   if (rtsp_client_sink->tls_database)
 709     g_object_unref (rtsp_client_sink->tls_database);
 710
 711   if (rtsp_client_sink->tls_interaction)
 712     g_object_unref (rtsp_client_sink->tls_interaction);
 713
 714   /* free locks */
 715   g_rec_mutex_clear (&rtsp_client_sink->stream_rec_lock);
 716   g_rec_mutex_clear (&rtsp_client_sink->state_rec_lock);
 717
 718   g_mutex_clear (&rtsp_client_sink->send_lock);
 719
 720   g_mutex_clear (&rtsp_client_sink->preroll_lock);
 721   g_cond_clear (&rtsp_client_sink->preroll_cond);
 722
 723   G_OBJECT_CLASS (parent_class)->finalize (object);
 724 }
 725
 726 static gboolean
 727 gst_rtp_payloader_filter_func (GstPluginFeature * feature, gpointer user_data)
 728 {
 729   GstElementFactory *factory = NULL;
 730   const gchar *klass;
 731
 732   if (!GST_IS_ELEMENT_FACTORY (feature))
 733     return FALSE;
 734
 735   factory = GST_ELEMENT_FACTORY (feature);
 736
 737   if (gst_plugin_feature_get_rank (feature) == GST_RANK_NONE)
 738     return FALSE;
 739
 740   if (!gst_element_factory_list_is_type (factory,
 741           GST_ELEMENT_FACTORY_TYPE_PAYLOADER))
 742     return FALSE;
 743
 744   klass =
 745       gst_element_factory_get_metadata (factory, GST_ELEMENT_METADATA_KLASS);
 746   if (strstr (klass, "Codec") == NULL)
 747     return FALSE;
 748   if (strstr (klass, "RTP") == NULL)
 749     return FALSE;
 750
 751   return TRUE;
 752 }
 753
 754 static gint
 755 compare_ranks (GstPluginFeature * f1, GstPluginFeature * f2)
 756 {
 757   gint diff;
 758   const gchar *rname1, *rname2;
 759   GstRank rank1, rank2;
 760
 761   rname1 = gst_plugin_feature_get_name (f1);
 762   rname2 = gst_plugin_feature_get_name (f2);
 763
 764   rank1 = gst_plugin_feature_get_rank (f1);
 765   rank2 = gst_plugin_feature_get_rank (f2);
 766
 767   /* HACK: Prefer rtpmp4apay over rtpmp4gpay */
 768   if (g_str_equal (rname1, "rtpmp4apay"))
 769     rank1 = GST_RANK_SECONDARY + 1;
 770   if (g_str_equal (rname2, "rtpmp4apay"))
 771     rank2 = GST_RANK_SECONDARY + 1;
 772
 773   diff = rank2 - rank1;
 774   if (diff != 0)
 775     return diff;
 776
 777   diff = strcmp (rname2, rname1);
 778
 779   return diff;
 780 }
 781
 782 static GList *
 783 gst_rtsp_client_sink_get_factories (void)
 784 {
 785   static GList *payloader_factories = NULL;
 786
 787   if (g_once_init_enter (&payloader_factories)) {
 788     GList *all_factories;
 789
 790     all_factories =
 791         gst_registry_feature_filter (gst_registry_get (),
 792         gst_rtp_payloader_filter_func, FALSE, NULL);
 793
 794     all_factories = g_list_sort (all_factories, (GCompareFunc) compare_ranks);
 795
 796     g_once_init_leave (&payloader_factories, all_factories);
 797   }
 798
 799   return payloader_factories;
 800 }
 801
 802 static GstCaps *
 803 gst_rtsp_client_sink_get_payloader_caps (void)
 804 {
 805   /* Cached caps result */
 806   static GstCaps *ret;
 807
 808   if (g_once_init_enter (&ret)) {
 809     GList *factories, *cur;
 810     GstCaps *caps = gst_caps_new_empty ();
 811
 812     factories = gst_rtsp_client_sink_get_factories ();
 813     for (cur = factories; cur != NULL; cur = g_list_next (cur)) {
 814       GstElementFactory *factory = GST_ELEMENT_FACTORY (cur->data);
 815       const GList *tmp;
 816
 817       for (tmp = gst_element_factory_get_static_pad_templates (factory);
 818           tmp; tmp = g_list_next (tmp)) {
 819         GstStaticPadTemplate *template = tmp->data;
 820
 821         if (template->direction == GST_PAD_SINK) {
 822           GstCaps *static_caps = gst_static_pad_template_get_caps (template);
 823
 824           GST_LOG ("Found pad template %s on factory %s",
 825               template->name_template, gst_plugin_feature_get_name (factory));
 826
 827           if (static_caps)
 828             caps = gst_caps_merge (caps, static_caps);
 829
 830           /* Early out, any is absorbing */
 831           if (gst_caps_is_any (caps))
 832             goto out;
 833         }
 834       }
 835     }
 836   out:
 837     g_once_init_leave (&ret, caps);
 838   }
 839
 840   /* Return cached result */
 841   return gst_caps_ref (ret);
 842 }
 843
 844 static GstElement *
 845 gst_rtsp_client_sink_make_payloader (GstCaps * caps)
 846 {
 847   GList *factories, *cur;
 848
 849   factories = gst_rtsp_client_sink_get_factories ();
 850   for (cur = factories; cur != NULL; cur = g_list_next (cur)) {
 851     GstElementFactory *factory = GST_ELEMENT_FACTORY (cur->data);
 852     const GList *tmp;
 853
 854     for (tmp = gst_element_factory_get_static_pad_templates (factory);
 855         tmp; tmp = g_list_next (tmp)) {
 856       GstStaticPadTemplate *template = tmp->data;
 857
 858       if (template->direction == GST_PAD_SINK) {
 859         GstCaps *static_caps = gst_static_pad_template_get_caps (template);
 860         GstElement *payloader = NULL;
 861
 862         if (gst_caps_can_intersect (static_caps, caps)) {
 863           GST_DEBUG ("caps %" GST_PTR_FORMAT " intersects with template %"
 864               GST_PTR_FORMAT " for payloader %s", caps, static_caps,
 865               gst_plugin_feature_get_name (factory));
 866           payloader = gst_element_factory_create (factory, NULL);
 867         }
 868
 869         gst_caps_unref (static_caps);
 870
 871         if (payloader)
 872           return payloader;
 873       }
 874     }
 875   }
 876
 877   return NULL;
 878 }
 879
 880 static GstRTSPStream *
 881 gst_rtsp_client_sink_create_stream (GstRTSPClientSink * sink,
 882     GstRTSPStreamContext * context, GstElement * payloader, GstPad * pad)
 883 {
 884   GstRTSPStream *stream = NULL;
 885   guint pt, aux_pt;
 886
 887   GST_OBJECT_LOCK (sink);
 888
 889   g_object_get (G_OBJECT (payloader), "pt", &pt, NULL);
 890   if (pt >= 96 && pt <= sink->next_dyn_pt) {
 891     /* Payloader has a dynamic PT, but one that's already used */
 892     /* FIXME: Create a caps->ptmap instead? */
 893     pt = sink->next_dyn_pt;
 894
 895     if (pt > 127)
 896       goto no_free_pt;
 897
 898     GST_DEBUG_OBJECT (sink, "Assigning pt %u to stream %d", pt, context->index);
 899
 900     sink->next_dyn_pt++;
 901   } else {
 902     GST_DEBUG_OBJECT (sink, "Keeping existing pt %u for stream %d",
 903         pt, context->index);
 904   }
 905
 906   aux_pt = sink->next_dyn_pt;
 907   if (aux_pt > 127)
 908     goto no_free_pt;
 909   sink->next_dyn_pt++;
 910
 911   GST_OBJECT_UNLOCK (sink);
 912
 913
 914   g_object_set (G_OBJECT (payloader), "pt", pt, NULL);
 915
 916   stream = gst_rtsp_stream_new (context->index, payloader, pad);
 917
 918   gst_rtsp_stream_set_client_side (stream, TRUE);
 919   gst_rtsp_stream_set_retransmission_time (stream,
 920       (GstClockTime) (sink->rtx_time) * GST_MSECOND);
 921   gst_rtsp_stream_set_protocols (stream, sink->protocols);
 922   gst_rtsp_stream_set_profiles (stream, sink->profiles);
 923   gst_rtsp_stream_set_retransmission_pt (stream, aux_pt);
 924   gst_rtsp_stream_set_buffer_size (stream, sink->udp_buffer_size);
 925   if (sink->rtp_blocksize > 0)
 926     gst_rtsp_stream_set_mtu (stream, sink->rtp_blocksize);
 927
 928 #if 0
 929   if (priv->pool)
 930     gst_rtsp_stream_set_address_pool (stream, priv->pool);
 931 #endif
 932
 933   return stream;
 934 no_free_pt:
 935   GST_OBJECT_UNLOCK (sink);
 936
 937   GST_ELEMENT_ERROR (sink, RESOURCE, NO_SPACE_LEFT, (NULL),
 938       ("Ran out of dynamic payload types."));
 939
 940   if (stream)
 941     g_object_unref (stream);
 942   return NULL;
 943 }
 944
 945 static GstPadProbeReturn
 946 handle_payloader_block (GstPad * pad, GstPadProbeInfo * info,
 947     GstRTSPStreamContext * context)
 948 {
 949   GstRTSPClientSink *sink = context->parent;
 950
 951   GST_INFO_OBJECT (sink, "Block on pad %" GST_PTR_FORMAT, pad);
 952
 953   g_mutex_lock (&sink->preroll_lock);
 954   context->prerolled = TRUE;
 955   g_cond_broadcast (&sink->preroll_cond);
 956   g_mutex_unlock (&sink->preroll_lock);
 957
 958   GST_INFO_OBJECT (sink, "Announced preroll on pad %" GST_PTR_FORMAT, pad);
 959
 960   return GST_PAD_PROBE_OK;
 961 }
 962
 963 static gboolean
 964 gst_rtsp_client_sink_setup_payloader (GstRTSPClientSink * sink, GstPad * pad,
 965     GstCaps * caps)
 966 {
 967   GstRTSPStreamContext *context;
 968
 969   GstElement *payloader;
 970   GstPad *sinkpad, *srcpad, *ghostsink;
 971
 972   context = gst_pad_get_element_private (pad);
 973
 974   /* Find the payloader. FIXME: Allow user to provide payloader via pad property */
 975   payloader = gst_rtsp_client_sink_make_payloader (caps);
 976   if (payloader == NULL)
 977     return FALSE;
 978
 979   GST_DEBUG_OBJECT (sink, "Configuring payloader %" GST_PTR_FORMAT
 980       " for pad %" GST_PTR_FORMAT, payloader, pad);
 981
 982   sinkpad = gst_element_get_static_pad (payloader, "sink");
 983   if (sinkpad == NULL)
 984     goto no_sinkpad;
 985
 986   srcpad = gst_element_get_static_pad (payloader, "src");
 987   if (srcpad == NULL)
 988     goto no_srcpad;
 989
 990   gst_bin_add (GST_BIN (sink->internal_bin), payloader);
 991   ghostsink = gst_ghost_pad_new (NULL, sinkpad);
 992   gst_pad_set_active (ghostsink, TRUE);
 993   gst_element_add_pad (GST_ELEMENT (sink->internal_bin), ghostsink);
 994
 995   g_signal_emit (sink, gst_rtsp_client_sink_signals[SIGNAL_NEW_PAYLOADER], 0,
 996       payloader);
 997
 998   GST_RTSP_STATE_LOCK (sink);
 999   context->payloader_block_id =
1000       gst_pad_add_probe (srcpad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM,
1001       (GstPadProbeCallback) handle_payloader_block, context, NULL);
1002   context->payloader = payloader;
1003
1004   payloader = gst_object_ref (payloader);
1005
1006   gst_ghost_pad_set_target (GST_GHOST_PAD (pad), ghostsink);
1007   gst_object_unref (GST_OBJECT (sinkpad));
1008   GST_RTSP_STATE_UNLOCK (sink);
1009
1010   gst_element_sync_state_with_parent (payloader);
1011
1012   gst_object_unref (payloader);
1013   gst_object_unref (GST_OBJECT (srcpad));
1014
1015   return TRUE;
1016
1017 no_sinkpad:
1018   GST_ERROR_OBJECT (sink,
1019       "Could not find sink pad on payloader %" GST_PTR_FORMAT, payloader);
1020   gst_object_unref (payloader);
1021   return FALSE;
1022
1023 no_srcpad:
1024   GST_ERROR_OBJECT (sink,
1025       "Could not find src pad on payloader %" GST_PTR_FORMAT, payloader);
1026   gst_object_unref (GST_OBJECT (sinkpad));
1027   gst_object_unref (payloader);
1028   return TRUE;
1029 }
1030
1031 static gboolean
1032 gst_rtsp_client_sink_sinkpad_event (GstPad * pad, GstObject * parent,
1033     GstEvent * event)
1034 {
1035   if (GST_EVENT_TYPE (event) == GST_EVENT_CAPS) {
1036     GstPad *target = gst_ghost_pad_get_target (GST_GHOST_PAD (pad));
1037     if (target == NULL) {
1038       GstCaps *caps;
1039
1040       /* No target yet - choose a payloader and configure it */
1041       gst_event_parse_caps (event, &caps);
1042
1043       GST_DEBUG_OBJECT (parent,
1044           "Have set caps event on pad %" GST_PTR_FORMAT
1045           " caps %" GST_PTR_FORMAT, pad, caps);
1046
1047       if (!gst_rtsp_client_sink_setup_payloader (GST_RTSP_CLIENT_SINK (parent),
1048               pad, caps)) {
1049         gst_event_unref (event);
1050         return FALSE;
1051       }
1052     }
1053   }
1054
1055   return gst_pad_event_default (pad, parent, event);
1056 }
1057
1058 static gboolean
1059 gst_rtsp_client_sink_sinkpad_query (GstPad * pad, GstObject * parent,
1060     GstQuery * query)
1061 {
1062   if (GST_QUERY_TYPE (query) == GST_QUERY_CAPS) {
1063     GstPad *target = gst_ghost_pad_get_target (GST_GHOST_PAD (pad));
1064     if (target == NULL) {
1065       /* No target yet - return the union of all payloader caps */
1066       GstCaps *caps = gst_rtsp_client_sink_get_payloader_caps ();
1067
1068       GST_TRACE_OBJECT (parent, "Returning payloader caps %" GST_PTR_FORMAT,
1069           caps);
1070
1071       gst_query_set_caps_result (query, caps);
1072       gst_caps_unref (caps);
1073
1074       return TRUE;
1075     }
1076   }
1077
1078   return gst_pad_query_default (pad, parent, query);
1079 }
1080
1081 static GstPad *
1082 gst_rtsp_client_sink_request_new_pad (GstElement * element,
1083     GstPadTemplate * templ, const gchar * name, const GstCaps * caps)
1084 {
1085   GstRTSPClientSink *sink = GST_RTSP_CLIENT_SINK (element);
1086   GstPad *pad;
1087   GstRTSPStreamContext *context;
1088   guint idx = (guint) - 1;
1089   gchar *tmpname;
1090
1091   g_mutex_lock (&sink->preroll_lock);
1092   if (sink->streams_collected) {
1093     GST_WARNING_OBJECT (element, "Can't add streams to a running session");
1094     g_mutex_unlock (&sink->preroll_lock);
1095     return NULL;
1096   }
1097   g_mutex_unlock (&sink->preroll_lock);
1098
1099   GST_OBJECT_LOCK (sink);
1100   if (name) {
1101     if (!sscanf (name, "sink_%u", &idx)) {
1102       GST_OBJECT_UNLOCK (sink);
1103       GST_ERROR_OBJECT (element, "Invalid sink pad name %s", name);
1104       return NULL;
1105     }
1106
1107     if (idx >= sink->next_pad_id)
1108       sink->next_pad_id = idx + 1;
1109   }
1110   if (idx == (guint) - 1) {
1111     idx = sink->next_pad_id;
1112     sink->next_pad_id++;
1113   }
1114   GST_OBJECT_UNLOCK (sink);
1115
1116   tmpname = g_strdup_printf ("sink_%u", idx);
1117   pad = gst_ghost_pad_new_no_target_from_template (tmpname, templ);
1118   g_free (tmpname);
1119
1120   GST_DEBUG_OBJECT (element, "Creating request pad %" GST_PTR_FORMAT, pad);
1121
1122   gst_pad_set_event_function (pad,
1123       GST_DEBUG_FUNCPTR (gst_rtsp_client_sink_sinkpad_event));
1124   gst_pad_set_query_function (pad,
1125       GST_DEBUG_FUNCPTR (gst_rtsp_client_sink_sinkpad_query));
1126
1127   context = g_new0 (GstRTSPStreamContext, 1);
1128   context->parent = sink;
1129   context->index = idx;
1130
1131   gst_pad_set_element_private (pad, context);
1132
1133   /* The rest of the context is configured on a caps set */
1134   gst_pad_set_active (pad, TRUE);
1135   gst_element_add_pad (element, pad);
1136
1137   (void) gst_rtsp_client_sink_get_factories ();
1138
1139   GST_RTSP_STATE_LOCK (sink);
1140   sink->contexts = g_list_prepend (sink->contexts, context);
1141   GST_RTSP_STATE_UNLOCK (sink);
1142
1143   return pad;
1144 }
1145
1146 static void
1147 gst_rtsp_client_sink_release_pad (GstElement * element, GstPad * pad)
1148 {
1149   GstRTSPClientSink *sink = GST_RTSP_CLIENT_SINK (element);
1150   GstRTSPStreamContext *context;
1151
1152   context = gst_pad_get_element_private (pad);
1153
1154   GST_RTSP_STATE_LOCK (sink);
1155   sink->contexts = g_list_remove (sink->contexts, context);
1156   GST_RTSP_STATE_UNLOCK (sink);
1157
1158   /* FIXME: Shut down and clean up streaming on this pad,
1159    * do teardown if needed */
1160   GST_LOG_OBJECT (sink,
1161       "Cleaning up payloader and stream for released pad %" GST_PTR_FORMAT,
1162       pad);
1163
1164   if (context->stream_transport) {
1165     gst_rtsp_stream_transport_set_active (context->stream_transport, FALSE);
1166     gst_object_unref (context->stream_transport);
1167     context->stream_transport = NULL;
1168   }
1169   if (context->stream) {
1170     if (context->joined) {
1171       gst_rtsp_stream_leave_bin (context->stream,
1172           GST_BIN (sink->internal_bin), sink->rtpbin);
1173       context->joined = FALSE;
1174     }
1175     gst_object_unref (context->stream);
1176     context->stream = NULL;
1177   }
1178   if (context->srtcpparams)
1179     gst_caps_unref (context->srtcpparams);
1180
1181   g_free (context->conninfo.location);
1182   context->conninfo.location = NULL;
1183
1184   g_free (context);
1185
1186   gst_element_remove_pad (element, pad);
1187 }
1188
1189 static GstClock *
1190 gst_rtsp_client_sink_provide_clock (GstElement * element)
1191 {
1192   GstRTSPClientSink *sink = GST_RTSP_CLIENT_SINK (element);
1193   GstClock *clock;
1194
1195   if ((clock = sink->provided_clock) != NULL)
1196     gst_object_ref (clock);
1197
1198   return clock;
1199 }
1200
1201 /* a proxy string of the format [user:passwd@]host[:port] */
1202 static gboolean
1203 gst_rtsp_client_sink_set_proxy (GstRTSPClientSink * rtsp, const gchar * proxy)
1204 {
1205   gchar *p, *at, *col;
1206
1207   g_free (rtsp->proxy_user);
1208   rtsp->proxy_user = NULL;
1209   g_free (rtsp->proxy_passwd);
1210   rtsp->proxy_passwd = NULL;
1211   g_free (rtsp->proxy_host);
1212   rtsp->proxy_host = NULL;
1213   rtsp->proxy_port = 0;
1214
1215   p = (gchar *) proxy;
1216
1217   if (p == NULL)
1218     return TRUE;
1219
1220   /* we allow http:// in front but ignore it */
1221   if (g_str_has_prefix (p, "http://"))
1222     p += 7;
1223
1224   at = strchr (p, '@');
1225   if (at) {
1226     /* look for user:passwd */
1227     col = strchr (proxy, ':');
1228     if (col == NULL || col > at)
1229       return FALSE;
1230
1231     rtsp->proxy_user = g_strndup (p, col - p);
1232     col++;
1233     rtsp->proxy_passwd = g_strndup (col, at - col);
1234
1235     /* move to host */
1236     p = at + 1;
1237   } else {
1238     if (rtsp->prop_proxy_id != NULL && *rtsp->prop_proxy_id != '\0')
1239       rtsp->proxy_user = g_strdup (rtsp->prop_proxy_id);
1240     if (rtsp->prop_proxy_pw != NULL && *rtsp->prop_proxy_pw != '\0')
1241       rtsp->proxy_passwd = g_strdup (rtsp->prop_proxy_pw);
1242     if (rtsp->proxy_user != NULL || rtsp->proxy_passwd != NULL) {
1243       GST_LOG_OBJECT (rtsp, "set proxy user/pw from properties: %s:%s",
1244           GST_STR_NULL (rtsp->proxy_user), GST_STR_NULL (rtsp->proxy_passwd));
1245     }
1246   }
1247   col = strchr (p, ':');
1248
1249   if (col) {
1250     /* everything before the colon is the hostname */
1251     rtsp->proxy_host = g_strndup (p, col - p);
1252     p = col + 1;
1253     rtsp->proxy_port = strtoul (p, (char **) &p, 10);
1254   } else {
1255     rtsp->proxy_host = g_strdup (p);
1256     rtsp->proxy_port = 8080;
1257   }
1258   return TRUE;
1259 }
1260
1261 static void
1262 gst_rtsp_client_sink_set_tcp_timeout (GstRTSPClientSink * rtsp_client_sink,
1263     guint64 timeout)
1264 {
1265   rtsp_client_sink->tcp_timeout.tv_sec = timeout / G_USEC_PER_SEC;
1266   rtsp_client_sink->tcp_timeout.tv_usec = timeout % G_USEC_PER_SEC;
1267
1268   if (timeout != 0)
1269     rtsp_client_sink->ptcp_timeout = &rtsp_client_sink->tcp_timeout;
1270   else
1271     rtsp_client_sink->ptcp_timeout = NULL;
1272 }
1273
1274 static void
1275 gst_rtsp_client_sink_set_property (GObject * object, guint prop_id,
1276     const GValue * value, GParamSpec * pspec)
1277 {
1278   GstRTSPClientSink *rtsp_client_sink;
1279
1280   rtsp_client_sink = GST_RTSP_CLIENT_SINK (object);
1281
1282   switch (prop_id) {
1283     case PROP_LOCATION:
1284       gst_rtsp_client_sink_uri_set_uri (GST_URI_HANDLER (rtsp_client_sink),
1285           g_value_get_string (value), NULL);
1286       break;
1287     case PROP_PROTOCOLS:
1288       rtsp_client_sink->protocols = g_value_get_flags (value);
1289       break;
1290     case PROP_PROFILES:
1291       rtsp_client_sink->profiles = g_value_get_flags (value);
1292       break;
1293     case PROP_DEBUG:
1294       rtsp_client_sink->debug = g_value_get_boolean (value);
1295       break;
1296     case PROP_RETRY:
1297       rtsp_client_sink->retry = g_value_get_uint (value);
1298       break;
1299     case PROP_TIMEOUT:
1300       rtsp_client_sink->udp_timeout = g_value_get_uint64 (value);
1301       break;
1302     case PROP_TCP_TIMEOUT:
1303       gst_rtsp_client_sink_set_tcp_timeout (rtsp_client_sink,
1304           g_value_get_uint64 (value));
1305       break;
1306     case PROP_LATENCY:
1307       rtsp_client_sink->latency = g_value_get_uint (value);
1308       break;
1309     case PROP_RTX_TIME:
1310       rtsp_client_sink->rtx_time = g_value_get_uint (value);
1311       break;
1312     case PROP_DO_RTSP_KEEP_ALIVE:
1313       rtsp_client_sink->do_rtsp_keep_alive = g_value_get_boolean (value);
1314       break;
1315     case PROP_PROXY:
1316       gst_rtsp_client_sink_set_proxy (rtsp_client_sink,
1317           g_value_get_string (value));
1318       break;
1319     case PROP_PROXY_ID:
1320       if (rtsp_client_sink->prop_proxy_id)
1321         g_free (rtsp_client_sink->prop_proxy_id);
1322       rtsp_client_sink->prop_proxy_id = g_value_dup_string (value);
1323       break;
1324     case PROP_PROXY_PW:
1325       if (rtsp_client_sink->prop_proxy_pw)
1326         g_free (rtsp_client_sink->prop_proxy_pw);
1327       rtsp_client_sink->prop_proxy_pw = g_value_dup_string (value);
1328       break;
1329     case PROP_RTP_BLOCKSIZE:
1330       rtsp_client_sink->rtp_blocksize = g_value_get_uint (value);
1331       break;
1332     case PROP_USER_ID:
1333       if (rtsp_client_sink->user_id)
1334         g_free (rtsp_client_sink->user_id);
1335       rtsp_client_sink->user_id = g_value_dup_string (value);
1336       break;
1337     case PROP_USER_PW:
1338       if (rtsp_client_sink->user_pw)
1339         g_free (rtsp_client_sink->user_pw);
1340       rtsp_client_sink->user_pw = g_value_dup_string (value);
1341       break;
1342     case PROP_PORT_RANGE:
1343     {
1344       const gchar *str;
1345
1346       str = g_value_get_string (value);
1347       if (str) {
1348         sscanf (str, "%u-%u",
1349             &rtsp_client_sink->client_port_range.min,
1350             &rtsp_client_sink->client_port_range.max);
1351       } else {
1352         rtsp_client_sink->client_port_range.min = 0;
1353         rtsp_client_sink->client_port_range.max = 0;
1354       }
1355       break;
1356     }
1357     case PROP_UDP_BUFFER_SIZE:
1358       rtsp_client_sink->udp_buffer_size = g_value_get_int (value);
1359       break;
1360     case PROP_UDP_RECONNECT:
1361       rtsp_client_sink->udp_reconnect = g_value_get_boolean (value);
1362       break;
1363     case PROP_MULTICAST_IFACE:
1364       g_free (rtsp_client_sink->multi_iface);
1365
1366       if (g_value_get_string (value) == NULL)
1367         rtsp_client_sink->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
1368       else
1369         rtsp_client_sink->multi_iface = g_value_dup_string (value);
1370       break;
1371     case PROP_SDES:
1372       rtsp_client_sink->sdes = g_value_dup_boxed (value);
1373       break;
1374     case PROP_TLS_VALIDATION_FLAGS:
1375       rtsp_client_sink->tls_validation_flags = g_value_get_flags (value);
1376       break;
1377     case PROP_TLS_DATABASE:
1378       g_clear_object (&rtsp_client_sink->tls_database);
1379       rtsp_client_sink->tls_database = g_value_dup_object (value);
1380       break;
1381     case PROP_TLS_INTERACTION:
1382       g_clear_object (&rtsp_client_sink->tls_interaction);
1383       rtsp_client_sink->tls_interaction = g_value_dup_object (value);
1384       break;
1385     case PROP_NTP_TIME_SOURCE:
1386       rtsp_client_sink->ntp_time_source = g_value_get_enum (value);
1387       break;
1388     case PROP_USER_AGENT:
1389       g_free (rtsp_client_sink->user_agent);
1390       rtsp_client_sink->user_agent = g_value_dup_string (value);
1391       break;
1392     default:
1393       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1394       break;
1395   }
1396 }
1397
1398 static void
1399 gst_rtsp_client_sink_get_property (GObject * object, guint prop_id,
1400     GValue * value, GParamSpec * pspec)
1401 {
1402   GstRTSPClientSink *rtsp_client_sink;
1403
1404   rtsp_client_sink = GST_RTSP_CLIENT_SINK (object);
1405
1406   switch (prop_id) {
1407     case PROP_LOCATION:
1408       g_value_set_string (value, rtsp_client_sink->conninfo.location);
1409       break;
1410     case PROP_PROTOCOLS:
1411       g_value_set_flags (value, rtsp_client_sink->protocols);
1412       break;
1413     case PROP_PROFILES:
1414       g_value_set_flags (value, rtsp_client_sink->profiles);
1415       break;
1416     case PROP_DEBUG:
1417       g_value_set_boolean (value, rtsp_client_sink->debug);
1418       break;
1419     case PROP_RETRY:
1420       g_value_set_uint (value, rtsp_client_sink->retry);
1421       break;
1422     case PROP_TIMEOUT:
1423       g_value_set_uint64 (value, rtsp_client_sink->udp_timeout);
1424       break;
1425     case PROP_TCP_TIMEOUT:
1426     {
1427       guint64 timeout;
1428
1429       timeout = rtsp_client_sink->tcp_timeout.tv_sec * G_USEC_PER_SEC +
1430           rtsp_client_sink->tcp_timeout.tv_usec;
1431       g_value_set_uint64 (value, timeout);
1432       break;
1433     }
1434     case PROP_LATENCY:
1435       g_value_set_uint (value, rtsp_client_sink->latency);
1436       break;
1437     case PROP_RTX_TIME:
1438       g_value_set_uint (value, rtsp_client_sink->rtx_time);
1439       break;
1440     case PROP_DO_RTSP_KEEP_ALIVE:
1441       g_value_set_boolean (value, rtsp_client_sink->do_rtsp_keep_alive);
1442       break;
1443     case PROP_PROXY:
1444     {
1445       gchar *str;
1446
1447       if (rtsp_client_sink->proxy_host) {
1448         str =
1449             g_strdup_printf ("%s:%d", rtsp_client_sink->proxy_host,
1450             rtsp_client_sink->proxy_port);
1451       } else {
1452         str = NULL;
1453       }
1454       g_value_take_string (value, str);
1455       break;
1456     }
1457     case PROP_PROXY_ID:
1458       g_value_set_string (value, rtsp_client_sink->prop_proxy_id);
1459       break;
1460     case PROP_PROXY_PW:
1461       g_value_set_string (value, rtsp_client_sink->prop_proxy_pw);
1462       break;
1463     case PROP_RTP_BLOCKSIZE:
1464       g_value_set_uint (value, rtsp_client_sink->rtp_blocksize);
1465       break;
1466     case PROP_USER_ID:
1467       g_value_set_string (value, rtsp_client_sink->user_id);
1468       break;
1469     case PROP_USER_PW:
1470       g_value_set_string (value, rtsp_client_sink->user_pw);
1471       break;
1472     case PROP_PORT_RANGE:
1473     {
1474       gchar *str;
1475
1476       if (rtsp_client_sink->client_port_range.min != 0) {
1477         str = g_strdup_printf ("%u-%u", rtsp_client_sink->client_port_range.min,
1478             rtsp_client_sink->client_port_range.max);
1479       } else {
1480         str = NULL;
1481       }
1482       g_value_take_string (value, str);
1483       break;
1484     }
1485     case PROP_UDP_BUFFER_SIZE:
1486       g_value_set_int (value, rtsp_client_sink->udp_buffer_size);
1487       break;
1488     case PROP_UDP_RECONNECT:
1489       g_value_set_boolean (value, rtsp_client_sink->udp_reconnect);
1490       break;
1491     case PROP_MULTICAST_IFACE:
1492       g_value_set_string (value, rtsp_client_sink->multi_iface);
1493       break;
1494     case PROP_SDES:
1495       g_value_set_boxed (value, rtsp_client_sink->sdes);
1496       break;
1497     case PROP_TLS_VALIDATION_FLAGS:
1498       g_value_set_flags (value, rtsp_client_sink->tls_validation_flags);
1499       break;
1500     case PROP_TLS_DATABASE:
1501       g_value_set_object (value, rtsp_client_sink->tls_database);
1502       break;
1503     case PROP_TLS_INTERACTION:
1504       g_value_set_object (value, rtsp_client_sink->tls_interaction);
1505       break;
1506     case PROP_NTP_TIME_SOURCE:
1507       g_value_set_enum (value, rtsp_client_sink->ntp_time_source);
1508       break;
1509     case PROP_USER_AGENT:
1510       g_value_set_string (value, rtsp_client_sink->user_agent);
1511       break;
1512     default:
1513       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1514       break;
1515   }
1516 }
1517
1518 static const gchar *
1519 get_aggregate_control (GstRTSPClientSink * sink)
1520 {
1521   const gchar *base;
1522
1523   if (sink->control)
1524     base = sink->control;
1525   else if (sink->content_base)
1526     base = sink->content_base;
1527   else if (sink->conninfo.url_str)
1528     base = sink->conninfo.url_str;
1529   else
1530     base = "/";
1531
1532   return base;
1533 }
1534
1535 static void
1536 gst_rtsp_client_sink_cleanup (GstRTSPClientSink * sink)
1537 {
1538   GList *walk;
1539
1540   GST_DEBUG_OBJECT (sink, "cleanup");
1541
1542   gst_element_set_state (GST_ELEMENT (sink->internal_bin), GST_STATE_NULL);
1543
1544   /* Clean up any left over stream objects */
1545   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
1546     GstRTSPStreamContext *context = (GstRTSPStreamContext *) (walk->data);
1547     if (context->stream_transport) {
1548       gst_rtsp_stream_transport_set_active (context->stream_transport, FALSE);
1549       gst_object_unref (context->stream_transport);
1550       context->stream_transport = NULL;
1551     }
1552
1553     if (context->stream) {
1554       if (context->joined) {
1555         gst_rtsp_stream_leave_bin (context->stream,
1556             GST_BIN (sink->internal_bin), sink->rtpbin);
1557         context->joined = FALSE;
1558       }
1559       gst_object_unref (context->stream);
1560       context->stream = NULL;
1561     }
1562
1563     if (context->srtcpparams)
1564       gst_caps_unref (context->srtcpparams);
1565     g_free (context->conninfo.location);
1566     context->conninfo.location = NULL;
1567   }
1568
1569   if (sink->rtpbin) {
1570     gst_element_set_state (sink->rtpbin, GST_STATE_NULL);
1571     gst_bin_remove (GST_BIN_CAST (sink->internal_bin), sink->rtpbin);
1572     sink->rtpbin = NULL;
1573   }
1574
1575   g_free (sink->content_base);
1576   sink->content_base = NULL;
1577
1578   g_free (sink->control);
1579   sink->control = NULL;
1580
1581   if (sink->range)
1582     gst_rtsp_range_free (sink->range);
1583   sink->range = NULL;
1584
1585   /* don't clear the SDP when it was used in the url */
1586   if (sink->uri_sdp && !sink->from_sdp) {
1587     gst_sdp_message_free (sink->uri_sdp);
1588     sink->uri_sdp = NULL;
1589   }
1590
1591   if (sink->provided_clock) {
1592     gst_object_unref (sink->provided_clock);
1593     sink->provided_clock = NULL;
1594   }
1595
1596   g_free (sink->server_ip);
1597   sink->server_ip = NULL;
1598
1599   sink->next_pad_id = 0;
1600   sink->next_dyn_pt = 96;
1601 }
1602
1603 static GstRTSPResult
1604 gst_rtsp_client_sink_connection_send (GstRTSPClientSink * sink,
1605     GstRTSPConnection * conn, GstRTSPMessage * message, GTimeVal * timeout)
1606 {
1607   GstRTSPResult ret;
1608
1609   if (conn)
1610     ret = gst_rtsp_connection_send (conn, message, timeout);
1611   else
1612     ret = GST_RTSP_ERROR;
1613
1614   return ret;
1615 }
1616
1617 static GstRTSPResult
1618 gst_rtsp_client_sink_connection_receive (GstRTSPClientSink * sink,
1619     GstRTSPConnection * conn, GstRTSPMessage * message, GTimeVal * timeout)
1620 {
1621   GstRTSPResult ret;
1622
1623   if (conn)
1624     ret = gst_rtsp_connection_receive (conn, message, timeout);
1625   else
1626     ret = GST_RTSP_ERROR;
1627
1628   return ret;
1629 }
1630
1631 static GstRTSPResult
1632 gst_rtsp_conninfo_connect (GstRTSPClientSink * sink, GstRTSPConnInfo * info,
1633     gboolean async)
1634 {
1635   GstRTSPResult res;
1636
1637   if (info->connection == NULL) {
1638     if (info->url == NULL) {
1639       GST_DEBUG_OBJECT (sink, "parsing uri (%s)...", info->location);
1640       if ((res = gst_rtsp_url_parse (info->location, &info->url)) < 0)
1641         goto parse_error;
1642     }
1643
1644     /* create connection */
1645     GST_DEBUG_OBJECT (sink, "creating connection (%s)...", info->location);
1646     if ((res = gst_rtsp_connection_create (info->url, &info->connection)) < 0)
1647       goto could_not_create;
1648
1649     if (info->url_str)
1650       g_free (info->url_str);
1651     info->url_str = gst_rtsp_url_get_request_uri (info->url);
1652
1653     GST_DEBUG_OBJECT (sink, "sanitized uri %s", info->url_str);
1654
1655     if (info->url->transports & GST_RTSP_LOWER_TRANS_TLS) {
1656       if (!gst_rtsp_connection_set_tls_validation_flags (info->connection,
1657               sink->tls_validation_flags))
1658         GST_WARNING_OBJECT (sink, "Unable to set TLS validation flags");
1659
1660       if (sink->tls_database)
1661         gst_rtsp_connection_set_tls_database (info->connection,
1662             sink->tls_database);
1663
1664       if (sink->tls_interaction)
1665         gst_rtsp_connection_set_tls_interaction (info->connection,
1666             sink->tls_interaction);
1667     }
1668
1669     if (info->url->transports & GST_RTSP_LOWER_TRANS_HTTP)
1670       gst_rtsp_connection_set_tunneled (info->connection, TRUE);
1671
1672     if (sink->proxy_host) {
1673       GST_DEBUG_OBJECT (sink, "setting proxy %s:%d", sink->proxy_host,
1674           sink->proxy_port);
1675       gst_rtsp_connection_set_proxy (info->connection, sink->proxy_host,
1676           sink->proxy_port);
1677     }
1678   }
1679
1680   if (!info->connected) {
1681     /* connect */
1682     if (async)
1683       GST_ELEMENT_PROGRESS (sink, CONTINUE, "connect",
1684           ("Connecting to %s", info->location));
1685     GST_DEBUG_OBJECT (sink, "connecting (%s)...", info->location);
1686     if ((res =
1687             gst_rtsp_connection_connect (info->connection,
1688                 sink->ptcp_timeout)) < 0)
1689       goto could_not_connect;
1690
1691     info->connected = TRUE;
1692   }
1693   return GST_RTSP_OK;
1694
1695   /* ERRORS */
1696 parse_error:
1697   {
1698     GST_ERROR_OBJECT (sink, "No valid RTSP URL was provided");
1699     return res;
1700   }
1701 could_not_create:
1702   {
1703     gchar *str = gst_rtsp_strresult (res);
1704     GST_ERROR_OBJECT (sink, "Could not create connection. (%s)", str);
1705     g_free (str);
1706     return res;
1707   }
1708 could_not_connect:
1709   {
1710     gchar *str = gst_rtsp_strresult (res);
1711     GST_ERROR_OBJECT (sink, "Could not connect to server. (%s)", str);
1712     g_free (str);
1713     return res;
1714   }
1715 }
1716
1717 static GstRTSPResult
1718 gst_rtsp_conninfo_close (GstRTSPClientSink * sink, GstRTSPConnInfo * info,
1719     gboolean free)
1720 {
1721   GST_RTSP_STATE_LOCK (sink);
1722   if (info->connected) {
1723     GST_DEBUG_OBJECT (sink, "closing connection...");
1724     gst_rtsp_connection_close (info->connection);
1725     info->connected = FALSE;
1726   }
1727   if (free && info->connection) {
1728     /* free connection */
1729     GST_DEBUG_OBJECT (sink, "freeing connection...");
1730     gst_rtsp_connection_free (info->connection);
1731     info->connection = NULL;
1732   }
1733   GST_RTSP_STATE_UNLOCK (sink);
1734   return GST_RTSP_OK;
1735 }
1736
1737 static GstRTSPResult
1738 gst_rtsp_conninfo_reconnect (GstRTSPClientSink * sink, GstRTSPConnInfo * info,
1739     gboolean async)
1740 {
1741   GstRTSPResult res;
1742
1743   GST_DEBUG_OBJECT (sink, "reconnecting connection...");
1744   gst_rtsp_conninfo_close (sink, info, FALSE);
1745   res = gst_rtsp_conninfo_connect (sink, info, async);
1746
1747   return res;
1748 }
1749
1750 static void
1751 gst_rtsp_client_sink_connection_flush (GstRTSPClientSink * sink, gboolean flush)
1752 {
1753   GList *walk;
1754
1755   GST_DEBUG_OBJECT (sink, "set flushing %d", flush);
1756   g_mutex_lock (&sink->preroll_lock);
1757   if (sink->conninfo.connection && sink->conninfo.flushing != flush) {
1758     GST_DEBUG_OBJECT (sink, "connection flush");
1759     gst_rtsp_connection_flush (sink->conninfo.connection, flush);
1760     sink->conninfo.flushing = flush;
1761   }
1762   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
1763     GstRTSPStreamContext *stream = (GstRTSPStreamContext *) walk->data;
1764     if (stream->conninfo.connection && stream->conninfo.flushing != flush) {
1765       GST_DEBUG_OBJECT (sink, "stream %p flush", stream);
1766       gst_rtsp_connection_flush (stream->conninfo.connection, flush);
1767       stream->conninfo.flushing = flush;
1768     }
1769   }
1770   g_cond_broadcast (&sink->preroll_cond);
1771   g_mutex_unlock (&sink->preroll_lock);
1772 }
1773
1774 static GstRTSPResult
1775 gst_rtsp_client_sink_init_request (GstRTSPClientSink * sink,
1776     GstRTSPMessage * msg, GstRTSPMethod method, const gchar * uri)
1777 {
1778   GstRTSPResult res;
1779
1780   res = gst_rtsp_message_init_request (msg, method, uri);
1781   if (res < 0)
1782     return res;
1783
1784   /* set user-agent */
1785   if (sink->user_agent)
1786     gst_rtsp_message_add_header (msg, GST_RTSP_HDR_USER_AGENT,
1787         sink->user_agent);
1788
1789   return res;
1790 }
1791
1792 /* FIXME, handle server request, reply with OK, for now */
1793 static GstRTSPResult
1794 gst_rtsp_client_sink_handle_request (GstRTSPClientSink * sink,
1795     GstRTSPConnection * conn, GstRTSPMessage * request)
1796 {
1797   GstRTSPMessage response = { 0 };
1798   GstRTSPResult res;
1799
1800   GST_DEBUG_OBJECT (sink, "got server request message");
1801
1802   if (sink->debug)
1803     gst_rtsp_message_dump (request);
1804
1805   /* default implementation, send OK */
1806   GST_DEBUG_OBJECT (sink, "prepare OK reply");
1807   res =
1808       gst_rtsp_message_init_response (&response, GST_RTSP_STS_OK, "OK",
1809       request);
1810   if (res < 0)
1811     goto send_error;
1812
1813   /* let app parse and reply */
1814   g_signal_emit (sink, gst_rtsp_client_sink_signals[SIGNAL_HANDLE_REQUEST],
1815       0, request, &response);
1816
1817   if (sink->debug)
1818     gst_rtsp_message_dump (&response);
1819
1820   res = gst_rtsp_client_sink_connection_send (sink, conn, &response, NULL);
1821   if (res < 0)
1822     goto send_error;
1823
1824   gst_rtsp_message_unset (&response);
1825
1826   return GST_RTSP_OK;
1827
1828   /* ERRORS */
1829 send_error:
1830   {
1831     gst_rtsp_message_unset (&response);
1832     return res;
1833   }
1834 }
1835
1836 /* send server keep-alive */
1837 static GstRTSPResult
1838 gst_rtsp_client_sink_send_keep_alive (GstRTSPClientSink * sink)
1839 {
1840   GstRTSPMessage request = { 0 };
1841   GstRTSPResult res;
1842   GstRTSPMethod method;
1843   const gchar *control;
1844
1845   if (sink->do_rtsp_keep_alive == FALSE) {
1846     GST_DEBUG_OBJECT (sink, "do-rtsp-keep-alive is FALSE, not sending.");
1847     gst_rtsp_connection_reset_timeout (sink->conninfo.connection);
1848     return GST_RTSP_OK;
1849   }
1850
1851   GST_DEBUG_OBJECT (sink, "creating server keep-alive");
1852
1853   /* find a method to use for keep-alive */
1854   if (sink->methods & GST_RTSP_GET_PARAMETER)
1855     method = GST_RTSP_GET_PARAMETER;
1856   else
1857     method = GST_RTSP_OPTIONS;
1858
1859   control = get_aggregate_control (sink);
1860   if (control == NULL)
1861     goto no_control;
1862
1863   res = gst_rtsp_client_sink_init_request (sink, &request, method, control);
1864   if (res < 0)
1865     goto send_error;
1866
1867   if (sink->debug)
1868     gst_rtsp_message_dump (&request);
1869
1870   res =
1871       gst_rtsp_client_sink_connection_send (sink, sink->conninfo.connection,
1872       &request, NULL);
1873   if (res < 0)
1874     goto send_error;
1875
1876   gst_rtsp_connection_reset_timeout (sink->conninfo.connection);
1877   gst_rtsp_message_unset (&request);
1878
1879   return GST_RTSP_OK;
1880
1881   /* ERRORS */
1882 no_control:
1883   {
1884     GST_WARNING_OBJECT (sink, "no control url to send keepalive");
1885     return GST_RTSP_OK;
1886   }
1887 send_error:
1888   {
1889     gchar *str = gst_rtsp_strresult (res);
1890
1891     gst_rtsp_message_unset (&request);
1892     GST_ELEMENT_WARNING (sink, RESOURCE, WRITE, (NULL),
1893         ("Could not send keep-alive. (%s)", str));
1894     g_free (str);
1895     return res;
1896   }
1897 }
1898
1899 static GstFlowReturn
1900 gst_rtsp_client_sink_loop_rx (GstRTSPClientSink * sink)
1901 {
1902   GstRTSPResult res;
1903   GstRTSPMessage message = { 0 };
1904   gint retry = 0;
1905
1906   while (TRUE) {
1907     GTimeVal tv_timeout;
1908
1909     /* get the next timeout interval */
1910     gst_rtsp_connection_next_timeout (sink->conninfo.connection, &tv_timeout);
1911
1912     GST_DEBUG_OBJECT (sink, "doing receive with timeout %d seconds",
1913         (gint) tv_timeout.tv_sec);
1914
1915     gst_rtsp_message_unset (&message);
1916
1917     /* we should continue reading the TCP socket because the server might
1918      * send us requests. When the session timeout expires, we need to send a
1919      * keep-alive request to keep the session open. */
1920     res =
1921         gst_rtsp_client_sink_connection_receive (sink,
1922         sink->conninfo.connection, &message, &tv_timeout);
1923
1924     switch (res) {
1925       case GST_RTSP_OK:
1926         GST_DEBUG_OBJECT (sink, "we received a server message");
1927         break;
1928       case GST_RTSP_EINTR:
1929         /* we got interrupted, see what we have to do */
1930         goto interrupt;
1931       case GST_RTSP_ETIMEOUT:
1932         /* send keep-alive, ignore the result, a warning will be posted. */
1933         GST_DEBUG_OBJECT (sink, "timeout, sending keep-alive");
1934         if ((res =
1935                 gst_rtsp_client_sink_send_keep_alive (sink)) == GST_RTSP_EINTR)
1936           goto interrupt;
1937         continue;
1938       case GST_RTSP_EEOF:
1939         /* server closed the connection. not very fatal for UDP, reconnect and
1940          * see what happens. */
1941         GST_ELEMENT_WARNING (sink, RESOURCE, READ, (NULL),
1942             ("The server closed the connection."));
1943         if (sink->udp_reconnect) {
1944           if ((res =
1945                   gst_rtsp_conninfo_reconnect (sink, &sink->conninfo,
1946                       FALSE)) < 0)
1947             goto connect_error;
1948         } else {
1949           goto server_eof;
1950         }
1951         continue;
1952       case GST_RTSP_ENET:
1953         GST_DEBUG_OBJECT (sink, "An ethernet problem occured.");
1954       default:
1955         GST_ELEMENT_WARNING (sink, RESOURCE, READ, (NULL),
1956             ("Unhandled return value %d.", res));
1957         goto receive_error;
1958     }
1959
1960     switch (message.type) {
1961       case GST_RTSP_MESSAGE_REQUEST:
1962         /* server sends us a request message, handle it */
1963         res =
1964             gst_rtsp_client_sink_handle_request (sink,
1965             sink->conninfo.connection, &message);
1966         if (res == GST_RTSP_EEOF)
1967           goto server_eof;
1968         else if (res < 0)
1969           goto handle_request_failed;
1970         break;
1971       case GST_RTSP_MESSAGE_RESPONSE:
1972         /* we ignore response and data messages */
1973         GST_DEBUG_OBJECT (sink, "ignoring response message");
1974         if (sink->debug)
1975           gst_rtsp_message_dump (&message);
1976         if (message.type_data.response.code == GST_RTSP_STS_UNAUTHORIZED) {
1977           GST_DEBUG_OBJECT (sink, "but is Unauthorized response ...");
1978           if (gst_rtsp_client_sink_setup_auth (sink, &message) && !(retry++)) {
1979             GST_DEBUG_OBJECT (sink, "so retrying keep-alive");
1980             if ((res =
1981                     gst_rtsp_client_sink_send_keep_alive (sink)) ==
1982                 GST_RTSP_EINTR)
1983               goto interrupt;
1984           }
1985         } else {
1986           retry = 0;
1987         }
1988         break;
1989       case GST_RTSP_MESSAGE_DATA:
1990         /* we ignore response and data messages */
1991         GST_DEBUG_OBJECT (sink, "ignoring data message");
1992         break;
1993       default:
1994         GST_WARNING_OBJECT (sink, "ignoring unknown message type %d",
1995             message.type);
1996         break;
1997     }
1998   }
1999   g_assert_not_reached ();
2000
2001   /* we get here when the connection got interrupted */
2002 interrupt:
2003   {
2004     gst_rtsp_message_unset (&message);
2005     GST_DEBUG_OBJECT (sink, "got interrupted");
2006     return GST_FLOW_FLUSHING;
2007   }
2008 connect_error:
2009   {
2010     gchar *str = gst_rtsp_strresult (res);
2011     GstFlowReturn ret;
2012
2013     sink->conninfo.connected = FALSE;
2014     if (res != GST_RTSP_EINTR) {
2015       GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_READ_WRITE, (NULL),
2016           ("Could not connect to server. (%s)", str));
2017       g_free (str);
2018       ret = GST_FLOW_ERROR;
2019     } else {
2020       ret = GST_FLOW_FLUSHING;
2021     }
2022     return ret;
2023   }
2024 receive_error:
2025   {
2026     gchar *str = gst_rtsp_strresult (res);
2027
2028     GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
2029         ("Could not receive message. (%s)", str));
2030     g_free (str);
2031     return GST_FLOW_ERROR;
2032   }
2033 handle_request_failed:
2034   {
2035     gchar *str = gst_rtsp_strresult (res);
2036     GstFlowReturn ret;
2037
2038     gst_rtsp_message_unset (&message);
2039     if (res != GST_RTSP_EINTR) {
2040       GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
2041           ("Could not handle server message. (%s)", str));
2042       g_free (str);
2043       ret = GST_FLOW_ERROR;
2044     } else {
2045       ret = GST_FLOW_FLUSHING;
2046     }
2047     return ret;
2048   }
2049 server_eof:
2050   {
2051     GST_DEBUG_OBJECT (sink, "we got an eof from the server");
2052     GST_ELEMENT_WARNING (sink, RESOURCE, READ, (NULL),
2053         ("The server closed the connection."));
2054     sink->conninfo.connected = FALSE;
2055     gst_rtsp_message_unset (&message);
2056     return GST_FLOW_EOS;
2057   }
2058 }
2059
2060 static GstRTSPResult
2061 gst_rtsp_client_sink_reconnect (GstRTSPClientSink * sink, gboolean async)
2062 {
2063   GstRTSPResult res = GST_RTSP_OK;
2064   gboolean restart = FALSE;
2065
2066   GST_DEBUG_OBJECT (sink, "doing reconnect");
2067
2068   GST_FIXME_OBJECT (sink, "Reconnection is not yet implemented");
2069
2070   /* no need to restart, we're done */
2071   if (!restart)
2072     goto done;
2073
2074   /* we can try only TCP now */
2075   sink->cur_protocols = GST_RTSP_LOWER_TRANS_TCP;
2076
2077   /* close and cleanup our state */
2078   if ((res = gst_rtsp_client_sink_close (sink, async, FALSE)) < 0)
2079     goto done;
2080
2081   /* see if we have TCP left to try. Also don't try TCP when we were configured
2082    * with an SDP. */
2083   if (!(sink->protocols & GST_RTSP_LOWER_TRANS_TCP) || sink->from_sdp)
2084     goto no_protocols;
2085
2086   /* We post a warning message now to inform the user
2087    * that nothing happened. It's most likely a firewall thing. */
2088   GST_ELEMENT_WARNING (sink, RESOURCE, READ, (NULL),
2089       ("Could not receive any UDP packets for %.4f seconds, maybe your "
2090           "firewall is blocking it. Retrying using a TCP connection.",
2091           gst_guint64_to_gdouble (sink->udp_timeout / 1000000.0)));
2092
2093   /* open new connection using tcp */
2094   if (gst_rtsp_client_sink_open (sink, async) < 0)
2095     goto open_failed;
2096
2097   /* start recording */
2098   if (gst_rtsp_client_sink_record (sink, async) < 0)
2099     goto play_failed;
2100
2101 done:
2102   return res;
2103
2104   /* ERRORS */
2105 no_protocols:
2106   {
2107     sink->cur_protocols = 0;
2108     /* no transport possible, post an error and stop */
2109     GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
2110         ("Could not receive any UDP packets for %.4f seconds, maybe your "
2111             "firewall is blocking it. No other protocols to try.",
2112             gst_guint64_to_gdouble (sink->udp_timeout / 1000000.0)));
2113     return GST_RTSP_ERROR;
2114   }
2115 open_failed:
2116   {
2117     GST_DEBUG_OBJECT (sink, "open failed");
2118     return GST_RTSP_OK;
2119   }
2120 play_failed:
2121   {
2122     GST_DEBUG_OBJECT (sink, "play failed");
2123     return GST_RTSP_OK;
2124   }
2125 }
2126
2127 static void
2128 gst_rtsp_client_sink_loop_start_cmd (GstRTSPClientSink * sink, gint cmd)
2129 {
2130   switch (cmd) {
2131     case CMD_OPEN:
2132       GST_ELEMENT_PROGRESS (sink, START, "open", ("Opening Stream"));
2133       break;
2134     case CMD_RECORD:
2135       GST_ELEMENT_PROGRESS (sink, START, "request", ("Sending RECORD request"));
2136       break;
2137     case CMD_PAUSE:
2138       GST_ELEMENT_PROGRESS (sink, START, "request", ("Sending PAUSE request"));
2139       break;
2140     case CMD_CLOSE:
2141       GST_ELEMENT_PROGRESS (sink, START, "close", ("Closing Stream"));
2142       break;
2143     default:
2144       break;
2145   }
2146 }
2147
2148 static void
2149 gst_rtsp_client_sink_loop_complete_cmd (GstRTSPClientSink * sink, gint cmd)
2150 {
2151   switch (cmd) {
2152     case CMD_OPEN:
2153       GST_ELEMENT_PROGRESS (sink, COMPLETE, "open", ("Opened Stream"));
2154       break;
2155     case CMD_RECORD:
2156       GST_ELEMENT_PROGRESS (sink, COMPLETE, "request", ("Sent RECORD request"));
2157       break;
2158     case CMD_PAUSE:
2159       GST_ELEMENT_PROGRESS (sink, COMPLETE, "request", ("Sent PAUSE request"));
2160       break;
2161     case CMD_CLOSE:
2162       GST_ELEMENT_PROGRESS (sink, COMPLETE, "close", ("Closed Stream"));
2163       break;
2164     default:
2165       break;
2166   }
2167 }
2168
2169 static void
2170 gst_rtsp_client_sink_loop_cancel_cmd (GstRTSPClientSink * sink, gint cmd)
2171 {
2172   switch (cmd) {
2173     case CMD_OPEN:
2174       GST_ELEMENT_PROGRESS (sink, CANCELED, "open", ("Open canceled"));
2175       break;
2176     case CMD_RECORD:
2177       GST_ELEMENT_PROGRESS (sink, CANCELED, "request", ("RECORD canceled"));
2178       break;
2179     case CMD_PAUSE:
2180       GST_ELEMENT_PROGRESS (sink, CANCELED, "request", ("PAUSE canceled"));
2181       break;
2182     case CMD_CLOSE:
2183       GST_ELEMENT_PROGRESS (sink, CANCELED, "close", ("Close canceled"));
2184       break;
2185     default:
2186       break;
2187   }
2188 }
2189
2190 static void
2191 gst_rtsp_client_sink_loop_error_cmd (GstRTSPClientSink * sink, gint cmd)
2192 {
2193   switch (cmd) {
2194     case CMD_OPEN:
2195       GST_ELEMENT_PROGRESS (sink, ERROR, "open", ("Open failed"));
2196       break;
2197     case CMD_RECORD:
2198       GST_ELEMENT_PROGRESS (sink, ERROR, "request", ("RECORD failed"));
2199       break;
2200     case CMD_PAUSE:
2201       GST_ELEMENT_PROGRESS (sink, ERROR, "request", ("PAUSE failed"));
2202       break;
2203     case CMD_CLOSE:
2204       GST_ELEMENT_PROGRESS (sink, ERROR, "close", ("Close failed"));
2205       break;
2206     default:
2207       break;
2208   }
2209 }
2210
2211 static void
2212 gst_rtsp_client_sink_loop_end_cmd (GstRTSPClientSink * sink, gint cmd,
2213     GstRTSPResult ret)
2214 {
2215   if (ret == GST_RTSP_OK)
2216     gst_rtsp_client_sink_loop_complete_cmd (sink, cmd);
2217   else if (ret == GST_RTSP_EINTR)
2218     gst_rtsp_client_sink_loop_cancel_cmd (sink, cmd);
2219   else
2220     gst_rtsp_client_sink_loop_error_cmd (sink, cmd);
2221 }
2222
2223 static gboolean
2224 gst_rtsp_client_sink_loop_send_cmd (GstRTSPClientSink * sink, gint cmd,
2225     gint mask)
2226 {
2227   gint old;
2228   gboolean flushed = FALSE;
2229
2230   /* start new request */
2231   gst_rtsp_client_sink_loop_start_cmd (sink, cmd);
2232
2233   GST_DEBUG_OBJECT (sink, "sending cmd %s", cmd_to_string (cmd));
2234
2235   GST_OBJECT_LOCK (sink);
2236   old = sink->pending_cmd;
2237   if (old == CMD_RECONNECT) {
2238     GST_DEBUG_OBJECT (sink, "ignore, we were reconnecting");
2239     cmd = CMD_RECONNECT;
2240   }
2241   if (old != CMD_WAIT) {
2242     sink->pending_cmd = CMD_WAIT;
2243     GST_OBJECT_UNLOCK (sink);
2244     /* cancel previous request */
2245     GST_DEBUG_OBJECT (sink, "cancel previous request %s", cmd_to_string (old));
2246     gst_rtsp_client_sink_loop_cancel_cmd (sink, old);
2247     GST_OBJECT_LOCK (sink);
2248   }
2249   sink->pending_cmd = cmd;
2250   /* interrupt if allowed */
2251   if (sink->busy_cmd & mask) {
2252     GST_DEBUG_OBJECT (sink, "connection flush busy %s",
2253         cmd_to_string (sink->busy_cmd));
2254     gst_rtsp_client_sink_connection_flush (sink, TRUE);
2255     flushed = TRUE;
2256   } else {
2257     GST_DEBUG_OBJECT (sink, "not interrupting busy cmd %s",
2258         cmd_to_string (sink->busy_cmd));
2259   }
2260   if (sink->task)
2261     gst_task_start (sink->task);
2262   GST_OBJECT_UNLOCK (sink);
2263
2264   return flushed;
2265 }
2266
2267 static gboolean
2268 gst_rtsp_client_sink_loop (GstRTSPClientSink * sink)
2269 {
2270   GstFlowReturn ret;
2271
2272   if (!sink->conninfo.connection || !sink->conninfo.connected)
2273     goto no_connection;
2274
2275   ret = gst_rtsp_client_sink_loop_rx (sink);
2276   if (ret != GST_FLOW_OK)
2277     goto pause;
2278
2279   return TRUE;
2280
2281   /* ERRORS */
2282 no_connection:
2283   {
2284     GST_WARNING_OBJECT (sink, "we are not connected");
2285     ret = GST_FLOW_FLUSHING;
2286     goto pause;
2287   }
2288 pause:
2289   {
2290     const gchar *reason = gst_flow_get_name (ret);
2291
2292     GST_DEBUG_OBJECT (sink, "pausing task, reason %s", reason);
2293     gst_rtsp_client_sink_loop_send_cmd (sink, CMD_WAIT, CMD_LOOP);
2294     return FALSE;
2295   }
2296 }
2297
2298 #ifndef GST_DISABLE_GST_DEBUG
2299 static const gchar *
2300 gst_rtsp_auth_method_to_string (GstRTSPAuthMethod method)
2301 {
2302   gint index = 0;
2303
2304   while (method != 0) {
2305     index++;
2306     method >>= 1;
2307   }
2308   switch (index) {
2309     case 0:
2310       return "None";
2311     case 1:
2312       return "Basic";
2313     case 2:
2314       return "Digest";
2315   }
2316
2317   return "Unknown";
2318 }
2319 #endif
2320
2321 static const gchar *
2322 gst_rtsp_client_sink_skip_lws (const gchar * s)
2323 {
2324   while (g_ascii_isspace (*s))
2325     s++;
2326   return s;
2327 }
2328
2329 static const gchar *
2330 gst_rtsp_client_sink_unskip_lws (const gchar * s, const gchar * start)
2331 {
2332   while (s > start && g_ascii_isspace (*(s - 1)))
2333     s--;
2334   return s;
2335 }
2336
2337 static const gchar *
2338 gst_rtsp_client_sink_skip_commas (const gchar * s)
2339 {
2340   /* The grammar allows for multiple commas */
2341   while (g_ascii_isspace (*s) || *s == ',')
2342     s++;
2343   return s;
2344 }
2345
2346 static const gchar *
2347 gst_rtsp_client_sink_skip_item (const gchar * s)
2348 {
2349   gboolean quoted = FALSE;
2350   const gchar *start = s;
2351
2352   /* A list item ends at the last non-whitespace character
2353    * before a comma which is not inside a quoted-string. Or at
2354    * the end of the string.
2355    */
2356   while (*s) {
2357     if (*s == '"')
2358       quoted = !quoted;
2359     else if (quoted) {
2360       if (*s == '\\' && *(s + 1))
2361         s++;
2362     } else {
2363       if (*s == ',')
2364         break;
2365     }
2366     s++;
2367   }
2368
2369   return gst_rtsp_client_sink_unskip_lws (s, start);
2370 }
2371
2372 static void
2373 gst_rtsp_decode_quoted_string (gchar * quoted_string)
2374 {
2375   gchar *src, *dst;
2376
2377   src = quoted_string + 1;
2378   dst = quoted_string;
2379   while (*src && *src != '"') {
2380     if (*src == '\\' && *(src + 1))
2381       src++;
2382     *dst++ = *src++;
2383   }
2384   *dst = '\0';
2385 }
2386
2387 /* Extract the authentication tokens that the server provided for each method
2388  * into an array of structures and give those to the connection object.
2389  */
2390 static void
2391 gst_rtsp_client_sink_parse_digest_challenge (GstRTSPConnection * conn,
2392     const gchar * header, gboolean * stale)
2393 {
2394   GSList *list = NULL, *iter;
2395   const gchar *end;
2396   gchar *item, *eq, *name_end, *value;
2397
2398   g_return_if_fail (stale != NULL);
2399
2400   gst_rtsp_connection_clear_auth_params (conn);
2401   *stale = FALSE;
2402
2403   /* Parse a header whose content is described by RFC2616 as
2404    * "#something", where "something" does not itself contain commas,
2405    * except as part of quoted-strings, into a list of allocated strings.
2406    */
2407   header = gst_rtsp_client_sink_skip_commas (header);
2408   while (*header) {
2409     end = gst_rtsp_client_sink_skip_item (header);
2410     list = g_slist_prepend (list, g_strndup (header, end - header));
2411     header = gst_rtsp_client_sink_skip_commas (end);
2412   }
2413   if (!list)
2414     return;
2415
2416   list = g_slist_reverse (list);
2417   for (iter = list; iter; iter = iter->next) {
2418     item = iter->data;
2419
2420     eq = strchr (item, '=');
2421     if (eq) {
2422       name_end = (gchar *) gst_rtsp_client_sink_unskip_lws (eq, item);
2423       if (name_end == item) {
2424         /* That's no good... */
2425         g_free (item);
2426         continue;
2427       }
2428
2429       *name_end = '\0';
2430
2431       value = (gchar *) gst_rtsp_client_sink_skip_lws (eq + 1);
2432       if (*value == '"')
2433         gst_rtsp_decode_quoted_string (value);
2434     } else
2435       value = NULL;
2436
2437     if (value && strcmp (item, "stale") == 0 && strcmp (value, "TRUE") == 0)
2438       *stale = TRUE;
2439     gst_rtsp_connection_set_auth_param (conn, item, value);
2440     g_free (item);
2441   }
2442
2443   g_slist_free (list);
2444 }
2445
2446 /* Parse a WWW-Authenticate Response header and determine the
2447  * available authentication methods
2448  *
2449  * This code should also cope with the fact that each WWW-Authenticate
2450  * header can contain multiple challenge methods + tokens
2451  *
2452  * At the moment, for Basic auth, we just do a minimal check and don't
2453  * even parse out the realm */
2454 static void
2455 gst_rtsp_client_sink_parse_auth_hdr (gchar * hdr, GstRTSPAuthMethod * methods,
2456     GstRTSPConnection * conn, gboolean * stale)
2457 {
2458   gchar *start;
2459
2460   g_return_if_fail (hdr != NULL);
2461   g_return_if_fail (methods != NULL);
2462   g_return_if_fail (stale != NULL);
2463
2464   /* Skip whitespace at the start of the string */
2465   for (start = hdr; start[0] != '\0' && g_ascii_isspace (start[0]); start++);
2466
2467   if (g_ascii_strncasecmp (start, "basic", 5) == 0)
2468     *methods |= GST_RTSP_AUTH_BASIC;
2469   else if (g_ascii_strncasecmp (start, "digest ", 7) == 0) {
2470     *methods |= GST_RTSP_AUTH_DIGEST;
2471     gst_rtsp_client_sink_parse_digest_challenge (conn, &start[7], stale);
2472   }
2473 }
2474
2475 /**
2476  * gst_rtsp_client_sink_setup_auth:
2477  * @src: the rtsp source
2478  *
2479  * Configure a username and password and auth method on the
2480  * connection object based on a response we received from the
2481  * peer.
2482  *
2483  * Currently, this requires that a username and password were supplied
2484  * in the uri. In the future, they may be requested on demand by sending
2485  * a message up the bus.
2486  *
2487  * Returns: TRUE if authentication information could be set up correctly.
2488  */
2489 static gboolean
2490 gst_rtsp_client_sink_setup_auth (GstRTSPClientSink * sink,
2491     GstRTSPMessage * response)
2492 {
2493   gchar *user = NULL;
2494   gchar *pass = NULL;
2495   GstRTSPAuthMethod avail_methods = GST_RTSP_AUTH_NONE;
2496   GstRTSPAuthMethod method;
2497   GstRTSPResult auth_result;
2498   GstRTSPUrl *url;
2499   GstRTSPConnection *conn;
2500   gchar *hdr;
2501   gboolean stale = FALSE;
2502
2503   conn = sink->conninfo.connection;
2504
2505   /* Identify the available auth methods and see if any are supported */
2506   if (gst_rtsp_message_get_header (response, GST_RTSP_HDR_WWW_AUTHENTICATE,
2507           &hdr, 0) == GST_RTSP_OK) {
2508     gst_rtsp_client_sink_parse_auth_hdr (hdr, &avail_methods, conn, &stale);
2509   }
2510
2511   if (avail_methods == GST_RTSP_AUTH_NONE)
2512     goto no_auth_available;
2513
2514   /* For digest auth, if the response indicates that the session
2515    * data are stale, we just update them in the connection object and
2516    * return TRUE to retry the request */
2517   if (stale)
2518     sink->tried_url_auth = FALSE;
2519
2520   url = gst_rtsp_connection_get_url (conn);
2521
2522   /* Do we have username and password available? */
2523   if (url != NULL && !sink->tried_url_auth && url->user != NULL
2524       && url->passwd != NULL) {
2525     user = url->user;
2526     pass = url->passwd;
2527     sink->tried_url_auth = TRUE;
2528     GST_DEBUG_OBJECT (sink,
2529         "Attempting authentication using credentials from the URL");
2530   } else {
2531     user = sink->user_id;
2532     pass = sink->user_pw;
2533     GST_DEBUG_OBJECT (sink,
2534         "Attempting authentication using credentials from the properties");
2535   }
2536
2537   /* FIXME: If the url didn't contain username and password or we tried them
2538    * already, request a username and passwd from the application via some kind
2539    * of credentials request message */
2540
2541   /* If we don't have a username and passwd at this point, bail out. */
2542   if (user == NULL || pass == NULL)
2543     goto no_user_pass;
2544
2545   /* Try to configure for each available authentication method, strongest to
2546    * weakest */
2547   for (method = GST_RTSP_AUTH_MAX; method != GST_RTSP_AUTH_NONE; method >>= 1) {
2548     /* Check if this method is available on the server */
2549     if ((method & avail_methods) == 0)
2550       continue;
2551
2552     /* Pass the credentials to the connection to try on the next request */
2553     auth_result = gst_rtsp_connection_set_auth (conn, method, user, pass);
2554     /* INVAL indicates an invalid username/passwd were supplied, so we'll just
2555      * ignore it and end up retrying later */
2556     if (auth_result == GST_RTSP_OK || auth_result == GST_RTSP_EINVAL) {
2557       GST_DEBUG_OBJECT (sink, "Attempting %s authentication",
2558           gst_rtsp_auth_method_to_string (method));
2559       break;
2560     }
2561   }
2562
2563   if (method == GST_RTSP_AUTH_NONE)
2564     goto no_auth_available;
2565
2566   return TRUE;
2567
2568 no_auth_available:
2569   {
2570     /* Output an error indicating that we couldn't connect because there were
2571      * no supported authentication protocols */
2572     GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_READ, (NULL),
2573         ("No supported authentication protocol was found"));
2574     return FALSE;
2575   }
2576 no_user_pass:
2577   {
2578     /* We don't fire an error message, we just return FALSE and let the
2579      * normal NOT_AUTHORIZED error be propagated */
2580     return FALSE;
2581   }
2582 }
2583
2584 static GstRTSPResult
2585 gst_rtsp_client_sink_try_send (GstRTSPClientSink * sink,
2586     GstRTSPConnection * conn, GstRTSPMessage * request,
2587     GstRTSPMessage * response, GstRTSPStatusCode * code)
2588 {
2589   GstRTSPResult res;
2590   GstRTSPStatusCode thecode;
2591   gchar *content_base = NULL;
2592   gint try = 0;
2593
2594 again:
2595   GST_DEBUG_OBJECT (sink, "sending message");
2596
2597   if (sink->debug)
2598     gst_rtsp_message_dump (request);
2599
2600   g_mutex_lock (&sink->send_lock);
2601
2602   res =
2603       gst_rtsp_client_sink_connection_send (sink, conn, request,
2604       sink->ptcp_timeout);
2605   if (res < 0) {
2606     g_mutex_unlock (&sink->send_lock);
2607     goto send_error;
2608   }
2609
2610   gst_rtsp_connection_reset_timeout (conn);
2611
2612   /* See if we should handle the response */
2613   if (response == NULL) {
2614     g_mutex_unlock (&sink->send_lock);
2615     return GST_RTSP_OK;
2616   }
2617 next:
2618   res =
2619       gst_rtsp_client_sink_connection_receive (sink, conn, response,
2620       sink->ptcp_timeout);
2621
2622   g_mutex_unlock (&sink->send_lock);
2623
2624   if (res < 0)
2625     goto receive_error;
2626
2627   if (sink->debug)
2628     gst_rtsp_message_dump (response);
2629
2630
2631   switch (response->type) {
2632     case GST_RTSP_MESSAGE_REQUEST:
2633       res = gst_rtsp_client_sink_handle_request (sink, conn, response);
2634       if (res == GST_RTSP_EEOF)
2635         goto server_eof;
2636       else if (res < 0)
2637         goto handle_request_failed;
2638       g_mutex_lock (&sink->send_lock);
2639       goto next;
2640     case GST_RTSP_MESSAGE_RESPONSE:
2641       /* ok, a response is good */
2642       GST_DEBUG_OBJECT (sink, "received response message");
2643       break;
2644     case GST_RTSP_MESSAGE_DATA:
2645       /* we ignore data messages */
2646       GST_DEBUG_OBJECT (sink, "ignoring data message");
2647       g_mutex_lock (&sink->send_lock);
2648       goto next;
2649     default:
2650       GST_WARNING_OBJECT (sink, "ignoring unknown message type %d",
2651           response->type);
2652       g_mutex_lock (&sink->send_lock);
2653       goto next;
2654   }
2655
2656   thecode = response->type_data.response.code;
2657
2658   GST_DEBUG_OBJECT (sink, "got response message %d", thecode);
2659
2660   /* if the caller wanted the result code, we store it. */
2661   if (code)
2662     *code = thecode;
2663
2664   /* If the request didn't succeed, bail out before doing any more */
2665   if (thecode != GST_RTSP_STS_OK)
2666     return GST_RTSP_OK;
2667
2668   /* store new content base if any */
2669   gst_rtsp_message_get_header (response, GST_RTSP_HDR_CONTENT_BASE,
2670       &content_base, 0);
2671   if (content_base) {
2672     g_free (sink->content_base);
2673     sink->content_base = g_strdup (content_base);
2674   }
2675
2676   return GST_RTSP_OK;
2677
2678   /* ERRORS */
2679 send_error:
2680   {
2681     gchar *str = gst_rtsp_strresult (res);
2682
2683     if (res != GST_RTSP_EINTR) {
2684       GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
2685           ("Could not send message. (%s)", str));
2686     } else {
2687       GST_WARNING_OBJECT (sink, "send interrupted");
2688     }
2689     g_free (str);
2690     return res;
2691   }
2692 receive_error:
2693   {
2694     switch (res) {
2695       case GST_RTSP_EEOF:
2696         GST_WARNING_OBJECT (sink, "server closed connection");
2697         if ((try == 0) && !sink->interleaved && sink->udp_reconnect) {
2698           try++;
2699           /* if reconnect succeeds, try again */
2700           if ((res =
2701                   gst_rtsp_conninfo_reconnect (sink, &sink->conninfo,
2702                       FALSE)) == 0)
2703             goto again;
2704         }
2705         /* only try once after reconnect, then fallthrough and error out */
2706       default:
2707       {
2708         gchar *str = gst_rtsp_strresult (res);
2709
2710         if (res != GST_RTSP_EINTR) {
2711           GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
2712               ("Could not receive message. (%s)", str));
2713         } else {
2714           GST_WARNING_OBJECT (sink, "receive interrupted");
2715         }
2716         g_free (str);
2717         break;
2718       }
2719     }
2720     return res;
2721   }
2722 handle_request_failed:
2723   {
2724     /* ERROR was posted */
2725     gst_rtsp_message_unset (response);
2726     return res;
2727   }
2728 server_eof:
2729   {
2730     GST_DEBUG_OBJECT (sink, "we got an eof from the server");
2731     GST_ELEMENT_WARNING (sink, RESOURCE, READ, (NULL),
2732         ("The server closed the connection."));
2733     gst_rtsp_message_unset (response);
2734     return res;
2735   }
2736 }
2737
2738 static void
2739 gst_rtsp_client_sink_set_state (GstRTSPClientSink * sink, GstState state)
2740 {
2741   GST_DEBUG_OBJECT (sink, "Setting internal state to %s",
2742       gst_element_state_get_name (state));
2743   gst_element_set_state (GST_ELEMENT (sink->internal_bin), state);
2744 }
2745
2746 /**
2747  * gst_rtsp_client_sink_send:
2748  * @src: the rtsp source
2749  * @conn: the connection to send on
2750  * @request: must point to a valid request
2751  * @response: must point to an empty #GstRTSPMessage
2752  * @code: an optional code result
2753  *
2754  * send @request and retrieve the response in @response. optionally @code can be
2755  * non-NULL in which case it will contain the status code of the response.
2756  *
2757  * If This function returns #GST_RTSP_OK, @response will contain a valid response
2758  * message that should be cleaned with gst_rtsp_message_unset() after usage.
2759  *
2760  * If @code is NULL, this function will return #GST_RTSP_ERROR (with an invalid
2761  * @response message) if the response code was not 200 (OK).
2762  *
2763  * If the attempt results in an authentication failure, then this will attempt
2764  * to retrieve authentication credentials via gst_rtsp_client_sink_setup_auth and retry
2765  * the request.
2766  *
2767  * Returns: #GST_RTSP_OK if the processing was successful.
2768  */
2769 static GstRTSPResult
2770 gst_rtsp_client_sink_send (GstRTSPClientSink * sink, GstRTSPConnection * conn,
2771     GstRTSPMessage * request, GstRTSPMessage * response,
2772     GstRTSPStatusCode * code)
2773 {
2774   GstRTSPStatusCode int_code = GST_RTSP_STS_OK;
2775   GstRTSPResult res = GST_RTSP_ERROR;
2776   gint count;
2777   gboolean retry;
2778   GstRTSPMethod method = GST_RTSP_INVALID;
2779
2780   count = 0;
2781   do {
2782     retry = FALSE;
2783
2784     /* make sure we don't loop forever */
2785     if (count++ > 8)
2786       break;
2787
2788     /* save method so we can disable it when the server complains */
2789     method = request->type_data.request.method;
2790
2791     if ((res =
2792             gst_rtsp_client_sink_try_send (sink, conn, request, response,
2793                 &int_code)) < 0)
2794       goto error;
2795
2796     switch (int_code) {
2797       case GST_RTSP_STS_UNAUTHORIZED:
2798         if (gst_rtsp_client_sink_setup_auth (sink, response)) {
2799           /* Try the request/response again after configuring the auth info
2800            * and loop again */
2801           retry = TRUE;
2802         }
2803         break;
2804       default:
2805         break;
2806     }
2807   } while (retry == TRUE);
2808
2809   /* If the user requested the code, let them handle errors, otherwise
2810    * post an error below */
2811   if (code != NULL)
2812     *code = int_code;
2813   else if (int_code != GST_RTSP_STS_OK)
2814     goto error_response;
2815
2816   return res;
2817
2818   /* ERRORS */
2819 error:
2820   {
2821     GST_DEBUG_OBJECT (sink, "got error %d", res);
2822     return res;
2823   }
2824 error_response:
2825   {
2826     res = GST_RTSP_ERROR;
2827
2828     switch (response->type_data.response.code) {
2829       case GST_RTSP_STS_NOT_FOUND:
2830         GST_ELEMENT_ERROR (sink, RESOURCE, NOT_FOUND, (NULL), ("%s",
2831                 response->type_data.response.reason));
2832         break;
2833       case GST_RTSP_STS_UNAUTHORIZED:
2834         GST_ELEMENT_ERROR (sink, RESOURCE, NOT_AUTHORIZED, (NULL), ("%s",
2835                 response->type_data.response.reason));
2836         break;
2837       case GST_RTSP_STS_MOVED_PERMANENTLY:
2838       case GST_RTSP_STS_MOVE_TEMPORARILY:
2839       {
2840         gchar *new_location;
2841         GstRTSPLowerTrans transports;
2842
2843         GST_DEBUG_OBJECT (sink, "got redirection");
2844         /* if we don't have a Location Header, we must error */
2845         if (gst_rtsp_message_get_header (response, GST_RTSP_HDR_LOCATION,
2846                 &new_location, 0) < 0)
2847           break;
2848
2849         /* When we receive a redirect result, we go back to the INIT state after
2850          * parsing the new URI. The caller should do the needed steps to issue
2851          * a new setup when it detects this state change. */
2852         GST_DEBUG_OBJECT (sink, "redirection to %s", new_location);
2853
2854         /* save current transports */
2855         if (sink->conninfo.url)
2856           transports = sink->conninfo.url->transports;
2857         else
2858           transports = GST_RTSP_LOWER_TRANS_UNKNOWN;
2859
2860         gst_rtsp_client_sink_uri_set_uri (GST_URI_HANDLER (sink), new_location,
2861             NULL);
2862
2863         /* set old transports */
2864         if (sink->conninfo.url && transports != GST_RTSP_LOWER_TRANS_UNKNOWN)
2865           sink->conninfo.url->transports = transports;
2866
2867         sink->need_redirect = TRUE;
2868         sink->state = GST_RTSP_STATE_INIT;
2869         res = GST_RTSP_OK;
2870         break;
2871       }
2872       case GST_RTSP_STS_NOT_ACCEPTABLE:
2873       case GST_RTSP_STS_NOT_IMPLEMENTED:
2874       case GST_RTSP_STS_METHOD_NOT_ALLOWED:
2875         GST_WARNING_OBJECT (sink, "got NOT IMPLEMENTED, disable method %s",
2876             gst_rtsp_method_as_text (method));
2877         sink->methods &= ~method;
2878         res = GST_RTSP_OK;
2879         break;
2880       default:
2881         GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
2882             ("Got error response: %d (%s).", response->type_data.response.code,
2883                 response->type_data.response.reason));
2884         break;
2885     }
2886     /* if we return ERROR we should unset the response ourselves */
2887     if (res == GST_RTSP_ERROR)
2888       gst_rtsp_message_unset (response);
2889
2890     return res;
2891   }
2892 }
2893
2894 /* parse the response and collect all the supported methods. We need this
2895  * information so that we don't try to send an unsupported request to the
2896  * server.
2897  */
2898 static gboolean
2899 gst_rtsp_client_sink_parse_methods (GstRTSPClientSink * sink,
2900     GstRTSPMessage * response)
2901 {
2902   GstRTSPHeaderField field;
2903   gchar *respoptions;
2904   gint indx = 0;
2905
2906   /* reset supported methods */
2907   sink->methods = 0;
2908
2909   /* Try Allow Header first */
2910   field = GST_RTSP_HDR_ALLOW;
2911   while (TRUE) {
2912     respoptions = NULL;
2913     gst_rtsp_message_get_header (response, field, &respoptions, indx);
2914     if (indx == 0 && !respoptions) {
2915       /* if no Allow header was found then try the Public header... */
2916       field = GST_RTSP_HDR_PUBLIC;
2917       gst_rtsp_message_get_header (response, field, &respoptions, indx);
2918     }
2919     if (!respoptions)
2920       break;
2921
2922     sink->methods |= gst_rtsp_options_from_text (respoptions);
2923
2924     indx++;
2925   }
2926
2927   if (sink->methods == 0) {
2928     /* neither Allow nor Public are required, assume the server supports
2929      * at least SETUP. */
2930     GST_DEBUG_OBJECT (sink, "could not get OPTIONS");
2931     sink->methods = GST_RTSP_SETUP;
2932   }
2933
2934   /* Even if the server replied, and didn't say it supports
2935    * RECORD|ANNOUNCE, try anyway by assuming it does */
2936   sink->methods |= GST_RTSP_ANNOUNCE | GST_RTSP_RECORD;
2937
2938   if (!(sink->methods & GST_RTSP_SETUP))
2939     goto no_setup;
2940
2941   return TRUE;
2942
2943   /* ERRORS */
2944 no_setup:
2945   {
2946     GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_READ, (NULL),
2947         ("Server does not support SETUP."));
2948     return FALSE;
2949   }
2950 }
2951
2952 static GstRTSPResult
2953 gst_rtsp_client_sink_connect_to_server (GstRTSPClientSink * sink,
2954     gboolean async)
2955 {
2956   GstRTSPResult res;
2957   GstRTSPMessage request = { 0 };
2958   GstRTSPMessage response = { 0 };
2959   GSocket *conn_socket;
2960   GSocketAddress *sa;
2961   GInetAddress *ia;
2962
2963   sink->need_redirect = FALSE;
2964
2965   /* can't continue without a valid url */
2966   if (G_UNLIKELY (sink->conninfo.url == NULL)) {
2967     res = GST_RTSP_EINVAL;
2968     goto no_url;
2969   }
2970   sink->tried_url_auth = FALSE;
2971
2972   if ((res = gst_rtsp_conninfo_connect (sink, &sink->conninfo, async)) < 0)
2973     goto connect_failed;
2974
2975   conn_socket = gst_rtsp_connection_get_read_socket (sink->conninfo.connection);
2976   sa = g_socket_get_remote_address (conn_socket, NULL);
2977   ia = g_inet_socket_address_get_address (G_INET_SOCKET_ADDRESS (sa));
2978
2979   sink->server_ip = g_inet_address_to_string (ia);
2980
2981   g_object_unref (sa);
2982
2983   /* create OPTIONS */
2984   GST_DEBUG_OBJECT (sink, "create options...");
2985   res =
2986       gst_rtsp_client_sink_init_request (sink, &request, GST_RTSP_OPTIONS,
2987       sink->conninfo.url_str);
2988   if (res < 0)
2989     goto create_request_failed;
2990
2991   /* send OPTIONS */
2992   GST_DEBUG_OBJECT (sink, "send options...");
2993
2994   if (async)
2995     GST_ELEMENT_PROGRESS (sink, CONTINUE, "open",
2996         ("Retrieving server options"));
2997
2998   if ((res =
2999           gst_rtsp_client_sink_send (sink, sink->conninfo.connection, &request,
3000               &response, NULL)) < 0)
3001     goto send_error;
3002
3003   /* parse OPTIONS */
3004   if (!gst_rtsp_client_sink_parse_methods (sink, &response))
3005     goto methods_error;
3006
3007   /* FIXME: Do we need to handle REDIRECT responses for OPTIONS? */
3008
3009   /* clean up any messages */
3010   gst_rtsp_message_unset (&request);
3011   gst_rtsp_message_unset (&response);
3012
3013   return res;
3014
3015   /* ERRORS */
3016 no_url:
3017   {
3018     GST_ELEMENT_ERROR (sink, RESOURCE, NOT_FOUND, (NULL),
3019         ("No valid RTSP URL was provided"));
3020     goto cleanup_error;
3021   }
3022 connect_failed:
3023   {
3024     gchar *str = gst_rtsp_strresult (res);
3025
3026     if (res != GST_RTSP_EINTR) {
3027       GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_READ_WRITE, (NULL),
3028           ("Failed to connect. (%s)", str));
3029     } else {
3030       GST_WARNING_OBJECT (sink, "connect interrupted");
3031     }
3032     g_free (str);
3033     goto cleanup_error;
3034   }
3035 create_request_failed:
3036   {
3037     gchar *str = gst_rtsp_strresult (res);
3038
3039     GST_ELEMENT_ERROR (sink, LIBRARY, INIT, (NULL),
3040         ("Could not create request. (%s)", str));
3041     g_free (str);
3042     goto cleanup_error;
3043   }
3044 send_error:
3045   {
3046     /* Don't post a message - the rtsp_send method will have
3047      * taken care of it because we passed NULL for the response code */
3048     goto cleanup_error;
3049   }
3050 methods_error:
3051   {
3052     /* error was posted */
3053     res = GST_RTSP_ERROR;
3054     goto cleanup_error;
3055   }
3056 cleanup_error:
3057   {
3058     if (sink->conninfo.connection) {
3059       GST_DEBUG_OBJECT (sink, "free connection");
3060       gst_rtsp_conninfo_close (sink, &sink->conninfo, TRUE);
3061     }
3062     gst_rtsp_message_unset (&request);
3063     gst_rtsp_message_unset (&response);
3064     return res;
3065   }
3066 }
3067
3068 static GstRTSPResult
3069 gst_rtsp_client_sink_open (GstRTSPClientSink * sink, gboolean async)
3070 {
3071   GstRTSPResult ret;
3072
3073   sink->methods =
3074       GST_RTSP_SETUP | GST_RTSP_RECORD | GST_RTSP_PAUSE | GST_RTSP_TEARDOWN;
3075
3076   if ((ret = gst_rtsp_client_sink_connect_to_server (sink, async)) < 0)
3077     goto open_failed;
3078
3079   if (async)
3080     gst_rtsp_client_sink_loop_end_cmd (sink, CMD_OPEN, ret);
3081
3082   /* Collect all our input streams and create
3083    * stream objects before actually returning */
3084   gst_rtsp_client_sink_collect_streams (sink);
3085
3086   return ret;
3087
3088   /* ERRORS */
3089 open_failed:
3090   {
3091     GST_WARNING_OBJECT (sink, "Failed to connect to server");
3092     sink->open_error = TRUE;
3093     if (async)
3094       gst_rtsp_client_sink_loop_end_cmd (sink, CMD_OPEN, ret);
3095     return ret;
3096   }
3097 }
3098
3099 static GstRTSPResult
3100 gst_rtsp_client_sink_close (GstRTSPClientSink * sink, gboolean async,
3101     gboolean only_close)
3102 {
3103   GstRTSPMessage request = { 0 };
3104   GstRTSPMessage response = { 0 };
3105   GstRTSPResult res = GST_RTSP_OK;
3106   GList *walk;
3107   const gchar *control;
3108
3109   GST_DEBUG_OBJECT (sink, "TEARDOWN...");
3110
3111   gst_rtsp_client_sink_set_state (sink, GST_STATE_NULL);
3112
3113   if (sink->state < GST_RTSP_STATE_READY) {
3114     GST_DEBUG_OBJECT (sink, "not ready, doing cleanup");
3115     goto close;
3116   }
3117
3118   if (only_close)
3119     goto close;
3120
3121   /* construct a control url */
3122   control = get_aggregate_control (sink);
3123
3124   if (!(sink->methods & (GST_RTSP_RECORD | GST_RTSP_TEARDOWN)))
3125     goto not_supported;
3126
3127   /* stop streaming */
3128   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
3129     GstRTSPStreamContext *context = (GstRTSPStreamContext *) walk->data;
3130
3131     if (context->stream_transport)
3132       gst_rtsp_stream_transport_set_active (context->stream_transport, FALSE);
3133
3134     if (context->joined) {
3135       gst_rtsp_stream_leave_bin (context->stream, GST_BIN (sink->internal_bin),
3136           sink->rtpbin);
3137       context->joined = FALSE;
3138     }
3139   }
3140
3141   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
3142     GstRTSPStreamContext *context = (GstRTSPStreamContext *) walk->data;
3143     const gchar *setup_url;
3144     GstRTSPConnInfo *info;
3145
3146     GST_DEBUG_OBJECT (sink, "Looking at stream %p for teardown",
3147         context->stream);
3148
3149     /* try aggregate control first but do non-aggregate control otherwise */
3150     if (control)
3151       setup_url = control;
3152     else if ((setup_url = context->conninfo.location) == NULL) {
3153       GST_DEBUG_OBJECT (sink, "Skipping TEARDOWN stream %p - no setup URL",
3154           context->stream);
3155       continue;
3156     }
3157
3158     if (sink->conninfo.connection) {
3159       info = &sink->conninfo;
3160     } else if (context->conninfo.connection) {
3161       info = &context->conninfo;
3162     } else {
3163       continue;
3164     }
3165     if (!info->connected)
3166       goto next;
3167
3168     /* do TEARDOWN */
3169     GST_DEBUG_OBJECT (sink, "Sending teardown for stream %p at URL %s",
3170         context->stream, setup_url);
3171     res =
3172         gst_rtsp_client_sink_init_request (sink, &request, GST_RTSP_TEARDOWN,
3173         setup_url);
3174     if (res < 0)
3175       goto create_request_failed;
3176
3177     if (async)
3178       GST_ELEMENT_PROGRESS (sink, CONTINUE, "close", ("Closing stream"));
3179
3180     if ((res =
3181             gst_rtsp_client_sink_send (sink, info->connection, &request,
3182                 &response, NULL)) < 0)
3183       goto send_error;
3184
3185     /* FIXME, parse result? */
3186     gst_rtsp_message_unset (&request);
3187     gst_rtsp_message_unset (&response);
3188
3189   next:
3190     /* early exit when we did aggregate control */
3191     if (control)
3192       break;
3193   }
3194
3195 close:
3196   /* close connections */
3197   GST_DEBUG_OBJECT (sink, "closing connection...");
3198   gst_rtsp_conninfo_close (sink, &sink->conninfo, TRUE);
3199   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
3200     GstRTSPStreamContext *stream = (GstRTSPStreamContext *) walk->data;
3201     gst_rtsp_conninfo_close (sink, &stream->conninfo, TRUE);
3202   }
3203
3204   /* cleanup */
3205   gst_rtsp_client_sink_cleanup (sink);
3206
3207   sink->state = GST_RTSP_STATE_INVALID;
3208
3209   if (async)
3210     gst_rtsp_client_sink_loop_end_cmd (sink, CMD_CLOSE, res);
3211
3212   return res;
3213
3214   /* ERRORS */
3215 create_request_failed:
3216   {
3217     gchar *str = gst_rtsp_strresult (res);
3218
3219     GST_ELEMENT_ERROR (sink, LIBRARY, INIT, (NULL),
3220         ("Could not create request. (%s)", str));
3221     g_free (str);
3222     goto close;
3223   }
3224 send_error:
3225   {
3226     gchar *str = gst_rtsp_strresult (res);
3227
3228     gst_rtsp_message_unset (&request);
3229     if (res != GST_RTSP_EINTR) {
3230       GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
3231           ("Could not send message. (%s)", str));
3232     } else {
3233       GST_WARNING_OBJECT (sink, "TEARDOWN interrupted");
3234     }
3235     g_free (str);
3236     goto close;
3237   }
3238 not_supported:
3239   {
3240     GST_DEBUG_OBJECT (sink,
3241         "TEARDOWN and PLAY not supported, can't do TEARDOWN");
3242     goto close;
3243   }
3244 }
3245
3246 static gboolean
3247 gst_rtsp_client_sink_configure_manager (GstRTSPClientSink * sink)
3248 {
3249   GstElement *rtpbin;
3250   GstStateChangeReturn ret;
3251
3252   rtpbin = sink->rtpbin;
3253
3254   if (rtpbin == NULL) {
3255     GObjectClass *klass;
3256
3257     rtpbin = gst_element_factory_make ("rtpbin", NULL);
3258     if (rtpbin == NULL)
3259       goto no_rtpbin;
3260
3261     gst_bin_add (GST_BIN_CAST (sink->internal_bin), rtpbin);
3262
3263     sink->rtpbin = rtpbin;
3264
3265     /* Any more settings we should configure on rtpbin here? */
3266     g_object_set (sink->rtpbin, "latency", sink->latency, NULL);
3267
3268     klass = G_OBJECT_GET_CLASS (G_OBJECT (rtpbin));
3269
3270     if (g_object_class_find_property (klass, "ntp-time-source")) {
3271       g_object_set (sink->rtpbin, "ntp-time-source", sink->ntp_time_source,
3272           NULL);
3273     }
3274
3275     if (sink->sdes && g_object_class_find_property (klass, "sdes")) {
3276       g_object_set (sink->rtpbin, "sdes", sink->sdes, NULL);
3277     }
3278
3279     g_signal_emit (sink, gst_rtsp_client_sink_signals[SIGNAL_NEW_MANAGER], 0,
3280         sink->rtpbin);
3281   }
3282
3283   ret = gst_element_set_state (rtpbin, GST_STATE_PAUSED);
3284   if (ret == GST_STATE_CHANGE_FAILURE)
3285     goto start_manager_failure;
3286
3287   return TRUE;
3288
3289 no_rtpbin:
3290   {
3291     GST_WARNING ("no rtpbin element");
3292     g_warning ("failed to create element 'rtpbin', check your installation");
3293     return FALSE;
3294   }
3295 start_manager_failure:
3296   {
3297     GST_DEBUG_OBJECT (sink, "could not start session manager");
3298     gst_bin_remove (GST_BIN_CAST (sink->internal_bin), rtpbin);
3299     return FALSE;
3300   }
3301 }
3302
3303 static GstElement *
3304 request_aux_sender (GstElement * rtpbin, guint sessid, GstRTSPClientSink * sink)
3305 {
3306   GstRTSPStream *stream = NULL;
3307   GstElement *ret = NULL;
3308   GList *walk;
3309
3310   GST_RTSP_STATE_LOCK (sink);
3311   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
3312     GstRTSPStreamContext *context = (GstRTSPStreamContext *) walk->data;
3313
3314     if (sessid == gst_rtsp_stream_get_index (context->stream)) {
3315       stream = context->stream;
3316       break;
3317     }
3318   }
3319
3320   if (stream != NULL) {
3321     GST_DEBUG_OBJECT (sink, "Creating aux sender for stream %u", sessid);
3322     ret = gst_rtsp_stream_request_aux_sender (stream, sessid);
3323   }
3324
3325   GST_RTSP_STATE_UNLOCK (sink);
3326
3327   return ret;
3328 }
3329
3330 static gboolean
3331 gst_rtsp_client_sink_collect_streams (GstRTSPClientSink * sink)
3332 {
3333   GstRTSPStreamContext *context;
3334   GList *walk;
3335   const gchar *base;
3336   gboolean has_slash;
3337
3338   GST_DEBUG_OBJECT (sink, "Collecting stream information");
3339
3340   if (!gst_rtsp_client_sink_configure_manager (sink))
3341     return FALSE;
3342
3343   base = get_aggregate_control (sink);
3344   /* check if the base ends with / */
3345   has_slash = g_str_has_suffix (base, "/");
3346
3347   g_mutex_lock (&sink->preroll_lock);
3348   while (sink->contexts == NULL && !sink->conninfo.flushing) {
3349     g_cond_wait (&sink->preroll_cond, &sink->preroll_lock);
3350   }
3351   g_mutex_unlock (&sink->preroll_lock);
3352
3353   /* FIXME: Need different locking - need to protect against pad releases
3354    * and potential state changes ruining things here */
3355   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
3356     GstPad *srcpad;
3357
3358     context = (GstRTSPStreamContext *) walk->data;
3359     if (context->stream)
3360       continue;
3361
3362     g_mutex_lock (&sink->preroll_lock);
3363     while (!context->prerolled && !sink->conninfo.flushing) {
3364       GST_DEBUG_OBJECT (sink, "Waiting for caps on stream %d", context->index);
3365       g_cond_wait (&sink->preroll_cond, &sink->preroll_lock);
3366     }
3367     if (sink->conninfo.flushing) {
3368       g_mutex_unlock (&sink->preroll_lock);
3369       break;
3370     }
3371     g_mutex_unlock (&sink->preroll_lock);
3372
3373     if (context->payloader == NULL)
3374       continue;
3375
3376     srcpad = gst_element_get_static_pad (context->payloader, "src");
3377
3378     GST_DEBUG_OBJECT (sink, "Creating stream object for stream %d",
3379         context->index);
3380     context->stream =
3381         gst_rtsp_client_sink_create_stream (sink, context, context->payloader,
3382         srcpad);
3383
3384     /* concatenate the two strings, insert / when not present */
3385     g_free (context->conninfo.location);
3386     context->conninfo.location =
3387         g_strdup_printf ("%s%sstream=%d", base, has_slash ? "" : "/",
3388         context->index);
3389
3390     if (sink->rtx_time > 0) {
3391       /* enable retransmission by setting rtprtxsend as the "aux" element of rtpbin */
3392       g_signal_connect (sink->rtpbin, "request-aux-sender",
3393           (GCallback) request_aux_sender, sink);
3394     }
3395
3396     if (!gst_rtsp_stream_join_bin (context->stream,
3397             GST_BIN (sink->internal_bin), sink->rtpbin, GST_STATE_PAUSED)) {
3398       goto join_bin_failed;
3399     }
3400     context->joined = TRUE;
3401
3402     /* Let the stream object receive data */
3403     gst_pad_remove_probe (srcpad, context->payloader_block_id);
3404
3405     gst_object_unref (srcpad);
3406   }
3407
3408   /* Now wait for the preroll of the rtp bin */
3409   g_mutex_lock (&sink->preroll_lock);
3410   while (!sink->prerolled && !sink->conninfo.flushing) {
3411     GST_LOG_OBJECT (sink, "Waiting for preroll before continuing");
3412     g_cond_wait (&sink->preroll_cond, &sink->preroll_lock);
3413   }
3414   GST_LOG_OBJECT (sink, "Marking streams as collected");
3415   sink->streams_collected = TRUE;
3416   g_mutex_unlock (&sink->preroll_lock);
3417
3418   return TRUE;
3419
3420 join_bin_failed:
3421
3422   GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
3423       ("Could not start stream %d", context->index));
3424   return FALSE;
3425 }
3426
3427 static GstRTSPResult
3428 gst_rtsp_client_sink_create_transports_string (GstRTSPClientSink * sink,
3429     GstRTSPStreamContext * context, GSocketFamily family,
3430     GstRTSPLowerTrans protocols, GstRTSPProfile profiles, gchar ** transports)
3431 {
3432   GString *result;
3433   GstRTSPStream *stream = context->stream;
3434   gboolean first = TRUE;
3435
3436   /* the default RTSP transports */
3437   result = g_string_new ("RTP");
3438
3439   while (profiles != 0) {
3440     if (!first)
3441       g_string_append (result, ",RTP");
3442
3443     if (profiles & GST_RTSP_PROFILE_SAVPF) {
3444       g_string_append (result, "/SAVPF");
3445       profiles &= ~GST_RTSP_PROFILE_SAVPF;
3446     } else if (profiles & GST_RTSP_PROFILE_SAVP) {
3447       g_string_append (result, "/SAVP");
3448       profiles &= ~GST_RTSP_PROFILE_SAVP;
3449     } else if (profiles & GST_RTSP_PROFILE_AVPF) {
3450       g_string_append (result, "/AVPF");
3451       profiles &= ~GST_RTSP_PROFILE_AVPF;
3452     } else if (profiles & GST_RTSP_PROFILE_AVP) {
3453       g_string_append (result, "/AVP");
3454       profiles &= ~GST_RTSP_PROFILE_AVP;
3455     } else {
3456       GST_WARNING_OBJECT (sink, "Unimplemented profile(s) 0x%x", profiles);
3457       break;
3458     }
3459
3460     if (protocols & GST_RTSP_LOWER_TRANS_UDP) {
3461       GstRTSPRange ports;
3462
3463       GST_DEBUG_OBJECT (sink, "adding UDP unicast");
3464       gst_rtsp_stream_get_server_port (stream, &ports, family);
3465
3466       g_string_append_printf (result, "/UDP;unicast;client_port=%d-%d",
3467           ports.min, ports.max);
3468     } else if (protocols & GST_RTSP_LOWER_TRANS_UDP_MCAST) {
3469       GstRTSPAddress *addr =
3470           gst_rtsp_stream_get_multicast_address (stream, family);
3471       if (addr) {
3472         GST_DEBUG_OBJECT (sink, "adding UDP multicast");
3473         g_string_append_printf (result, "/UDP;multicast;client_port=%d-%d",
3474             addr->port, addr->port + addr->n_ports - 1);
3475         gst_rtsp_address_free (addr);
3476       }
3477     } else if (protocols & GST_RTSP_LOWER_TRANS_TCP) {
3478       GST_DEBUG_OBJECT (sink, "adding TCP");
3479       g_string_append_printf (result, "/TCP;unicast;interleaved=%d-%d",
3480           sink->free_channel, sink->free_channel + 1);
3481     }
3482
3483     g_string_append (result, ";mode=RECORD");
3484     /* FIXME: Support appending too:
3485        if (sink->append)
3486        g_string_append (result, ";append");
3487      */
3488
3489     first = FALSE;
3490   }
3491
3492   if (first) {
3493     /* No valid transport could be constructed */
3494     GST_ERROR_OBJECT (sink, "No supported profiles configured");
3495     goto fail;
3496   }
3497
3498   *transports = g_string_free (result, FALSE);
3499
3500   GST_DEBUG_OBJECT (sink, "prepared transports %s", GST_STR_NULL (*transports));
3501
3502   return GST_RTSP_OK;
3503 fail:
3504   g_string_free (result, TRUE);
3505   return GST_RTSP_ERROR;
3506 }
3507
3508 static GstCaps *
3509 signal_get_srtcp_params (GstRTSPClientSink * sink,
3510     GstRTSPStreamContext * context)
3511 {
3512   GstCaps *caps = NULL;
3513
3514   g_signal_emit (sink, gst_rtsp_client_sink_signals[SIGNAL_REQUEST_RTCP_KEY], 0,
3515       context->index, &caps);
3516
3517   if (caps != NULL)
3518     GST_DEBUG_OBJECT (sink, "SRTP parameters received");
3519
3520   return caps;
3521 }
3522
3523 static gchar *
3524 gst_rtsp_client_sink_stream_make_keymgmt (GstRTSPClientSink * sink,
3525     GstRTSPStreamContext * context)
3526 {
3527   gchar *base64, *result = NULL;
3528   GstMIKEYMessage *mikey_msg;
3529
3530   context->srtcpparams = signal_get_srtcp_params (sink, context);
3531   if (context->srtcpparams == NULL)
3532     context->srtcpparams = gst_rtsp_stream_get_caps (context->stream);
3533
3534   mikey_msg = gst_mikey_message_new_from_caps (context->srtcpparams);
3535   if (mikey_msg) {
3536     guint send_ssrc;
3537
3538     /* add policy '0' for our SSRC */
3539     gst_rtsp_stream_get_ssrc (context->stream, &send_ssrc);
3540     GST_LOG_OBJECT (sink, "Stream %p ssrc %x", context->stream, send_ssrc);
3541     gst_mikey_message_add_cs_srtp (mikey_msg, 0, send_ssrc, 0);
3542
3543     base64 = gst_mikey_message_base64_encode (mikey_msg);
3544     gst_mikey_message_unref (mikey_msg);
3545
3546     if (base64) {
3547       result = gst_sdp_make_keymgmt (context->conninfo.location, base64);
3548       g_free (base64);
3549     }
3550   }
3551
3552   return result;
3553 }
3554
3555 /* masks to be kept in sync with the hardcoded protocol order of preference
3556  * in code below */
3557 static const guint protocol_masks[] = {
3558   GST_RTSP_LOWER_TRANS_UDP,
3559   GST_RTSP_LOWER_TRANS_UDP_MCAST,
3560   GST_RTSP_LOWER_TRANS_TCP,
3561   0
3562 };
3563
3564 /* Same for profile_masks */
3565 static const guint profile_masks[] = {
3566   GST_RTSP_PROFILE_SAVPF,
3567   GST_RTSP_PROFILE_SAVP,
3568   GST_RTSP_PROFILE_AVPF,
3569   GST_RTSP_PROFILE_AVP,
3570   0
3571 };
3572
3573 static gboolean
3574 do_send_data (GstBuffer * buffer, guint8 channel,
3575     GstRTSPStreamContext * context)
3576 {
3577   GstRTSPClientSink *sink = context->parent;
3578   GstRTSPMessage message = { 0 };
3579   GstRTSPResult res = GST_RTSP_OK;
3580   GstMapInfo map_info;
3581   guint8 *data;
3582   guint usize;
3583
3584   gst_rtsp_message_init_data (&message, channel);
3585
3586   /* FIXME, need some sort of iovec RTSPMessage here */
3587   if (!gst_buffer_map (buffer, &map_info, GST_MAP_READ))
3588     return FALSE;
3589
3590   gst_rtsp_message_take_body (&message, map_info.data, map_info.size);
3591
3592   res =
3593       gst_rtsp_client_sink_try_send (sink, sink->conninfo.connection, &message,
3594       NULL, NULL);
3595
3596   gst_rtsp_message_steal_body (&message, &data, &usize);
3597   gst_buffer_unmap (buffer, &map_info);
3598
3599   gst_rtsp_message_unset (&message);
3600
3601   return res == GST_RTSP_OK;
3602 }
3603
3604 static GstRTSPResult
3605 gst_rtsp_client_sink_setup_streams (GstRTSPClientSink * sink, gboolean async)
3606 {
3607   GstRTSPResult res = GST_RTSP_ERROR;
3608   GstRTSPMessage request = { 0 };
3609   GstRTSPMessage response = { 0 };
3610   GstRTSPLowerTrans protocols;
3611   GstRTSPStatusCode code;
3612   GSocketFamily family;
3613   GSocketAddress *sa;
3614   GSocket *conn_socket;
3615   GstRTSPUrl *url;
3616   GList *walk;
3617   gchar *hval;
3618
3619   if (sink->conninfo.connection) {
3620     url = gst_rtsp_connection_get_url (sink->conninfo.connection);
3621     /* we initially allow all configured lower transports. based on the URL
3622      * transports and the replies from the server we narrow them down. */
3623     protocols = url->transports & sink->cur_protocols;
3624   } else {
3625     url = NULL;
3626     protocols = sink->cur_protocols;
3627   }
3628
3629   if (protocols == 0)
3630     goto no_protocols;
3631
3632   GST_RTSP_STATE_LOCK (sink);
3633
3634   if (G_UNLIKELY (sink->contexts == NULL))
3635     goto no_streams;
3636
3637   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
3638     GstRTSPStreamContext *context = (GstRTSPStreamContext *) walk->data;
3639     GstRTSPStream *stream;
3640
3641     GstRTSPConnection *conn;
3642     GstRTSPProfile profiles;
3643     GstRTSPProfile cur_profile;
3644     gchar *transports;
3645     gint retry = 0;
3646     guint profile_mask = 0;
3647     guint mask = 0;
3648     GstCaps *caps;
3649     const GstSDPMedia *media;
3650
3651     stream = context->stream;
3652     profiles = gst_rtsp_stream_get_profiles (stream);
3653
3654     caps = gst_rtsp_stream_get_caps (stream);
3655     if (caps == NULL) {
3656       GST_DEBUG_OBJECT (sink, "skipping stream %p, no caps", stream);
3657       continue;
3658     }
3659     gst_caps_unref (caps);
3660     media = gst_sdp_message_get_media (&sink->cursdp, context->sdp_index);
3661     if (media == NULL) {
3662       GST_DEBUG_OBJECT (sink, "skipping stream %p, no SDP info", stream);
3663       continue;
3664     }
3665
3666     /* skip setup if we have no URL for it */
3667     if (context->conninfo.location == NULL) {
3668       GST_DEBUG_OBJECT (sink, "skipping stream %p, no setup", stream);
3669       continue;
3670     }
3671
3672     if (sink->conninfo.connection == NULL) {
3673       if (!gst_rtsp_conninfo_connect (sink, &context->conninfo, async)) {
3674         GST_DEBUG_OBJECT (sink, "skipping stream %p, failed to connect",
3675             stream);
3676         continue;
3677       }
3678       conn = context->conninfo.connection;
3679     } else {
3680       conn = sink->conninfo.connection;
3681     }
3682     GST_DEBUG_OBJECT (sink, "doing setup of stream %p with %s", stream,
3683         context->conninfo.location);
3684
3685     conn_socket = gst_rtsp_connection_get_read_socket (conn);
3686     sa = g_socket_get_local_address (conn_socket, NULL);
3687     family = g_socket_address_get_family (sa);
3688     g_object_unref (sa);
3689
3690   next_protocol:
3691     /* first selectable profile */
3692     while (profile_masks[profile_mask]
3693         && !(profiles & profile_masks[profile_mask]))
3694       profile_mask++;
3695     if (!profile_masks[profile_mask])
3696       goto no_profiles;
3697
3698     /* first selectable protocol */
3699     while (protocol_masks[mask] && !(protocols & protocol_masks[mask]))
3700       mask++;
3701     if (!protocol_masks[mask])
3702       goto no_protocols;
3703
3704   retry:
3705     GST_DEBUG_OBJECT (sink, "protocols = 0x%x, protocol mask = 0x%x", protocols,
3706         protocol_masks[mask]);
3707     /* create a string with first transport in line */
3708     transports = NULL;
3709     cur_profile = profiles & profile_masks[profile_mask];
3710     res = gst_rtsp_client_sink_create_transports_string (sink, context, family,
3711         protocols & protocol_masks[mask], cur_profile, &transports);
3712     if (res < 0 || transports == NULL)
3713       goto setup_transport_failed;
3714
3715     if (strlen (transports) == 0) {
3716       g_free (transports);
3717       GST_DEBUG_OBJECT (sink, "no transports found");
3718       mask++;
3719       profile_mask = 0;
3720       goto next_protocol;
3721     }
3722
3723     GST_DEBUG_OBJECT (sink, "transport is %s", GST_STR_NULL (transports));
3724
3725     /* create SETUP request */
3726     res =
3727         gst_rtsp_client_sink_init_request (sink, &request, GST_RTSP_SETUP,
3728         context->conninfo.location);
3729     if (res < 0) {
3730       g_free (transports);
3731       goto create_request_failed;
3732     }
3733
3734     /* select transport */
3735     gst_rtsp_message_take_header (&request, GST_RTSP_HDR_TRANSPORT, transports);
3736
3737     /* set up keys */
3738     if (cur_profile == GST_RTSP_PROFILE_SAVP ||
3739         cur_profile == GST_RTSP_PROFILE_SAVPF) {
3740       hval = gst_rtsp_client_sink_stream_make_keymgmt (sink, context);
3741       gst_rtsp_message_take_header (&request, GST_RTSP_HDR_KEYMGMT, hval);
3742     }
3743
3744     /* if the user wants a non default RTP packet size we add the blocksize
3745      * parameter */
3746     if (sink->rtp_blocksize > 0) {
3747       hval = g_strdup_printf ("%d", sink->rtp_blocksize);
3748       gst_rtsp_message_take_header (&request, GST_RTSP_HDR_BLOCKSIZE, hval);
3749     }
3750
3751     if (async)
3752       GST_ELEMENT_PROGRESS (sink, CONTINUE, "request", ("SETUP stream %d",
3753               context->index));
3754
3755     /* handle the code ourselves */
3756     res = gst_rtsp_client_sink_send (sink, conn, &request, &response, &code);
3757     if (res < 0)
3758       goto send_error;
3759
3760     switch (code) {
3761       case GST_RTSP_STS_OK:
3762         break;
3763       case GST_RTSP_STS_UNSUPPORTED_TRANSPORT:
3764         gst_rtsp_message_unset (&request);
3765         gst_rtsp_message_unset (&response);
3766
3767         /* Try another profile. If no more, move to the next protocol */
3768         profile_mask++;
3769         while (profile_masks[profile_mask]
3770             && !(profiles & profile_masks[profile_mask]))
3771           profile_mask++;
3772         if (profile_masks[profile_mask])
3773           goto retry;
3774
3775         /* select next available protocol, give up on this stream if none */
3776         /* Reset profiles to try: */
3777         profile_mask = 0;
3778
3779         mask++;
3780         while (protocol_masks[mask] && !(protocols & protocol_masks[mask]))
3781           mask++;
3782         if (!protocol_masks[mask])
3783           continue;
3784         else
3785           goto retry;
3786       default:
3787         goto response_error;
3788     }
3789
3790     /* parse response transport */
3791     {
3792       gchar *resptrans = NULL;
3793       GstRTSPTransport *transport;
3794
3795       gst_rtsp_message_get_header (&response, GST_RTSP_HDR_TRANSPORT,
3796           &resptrans, 0);
3797       if (!resptrans) {
3798         goto no_transport;
3799       }
3800
3801       gst_rtsp_transport_new (&transport);
3802
3803       /* parse transport, go to next stream on parse error */
3804       if (gst_rtsp_transport_parse (resptrans, transport) != GST_RTSP_OK) {
3805         GST_WARNING_OBJECT (sink, "failed to parse transport %s", resptrans);
3806         goto next;
3807       }
3808
3809       /* update allowed transports for other streams. once the transport of
3810        * one stream has been determined, we make sure that all other streams
3811        * are configured in the same way */
3812       switch (transport->lower_transport) {
3813         case GST_RTSP_LOWER_TRANS_TCP:
3814           GST_DEBUG_OBJECT (sink, "stream %p as TCP interleaved", stream);
3815           protocols = GST_RTSP_LOWER_TRANS_TCP;
3816           sink->interleaved = TRUE;
3817           /* update free channels */
3818           sink->free_channel =
3819               MAX (transport->interleaved.min, sink->free_channel);
3820           sink->free_channel =
3821               MAX (transport->interleaved.max, sink->free_channel);
3822           sink->free_channel++;
3823           break;
3824         case GST_RTSP_LOWER_TRANS_UDP_MCAST:
3825           /* only allow multicast for other streams */
3826           GST_DEBUG_OBJECT (sink, "stream %p as UDP multicast", stream);
3827           protocols = GST_RTSP_LOWER_TRANS_UDP_MCAST;
3828           break;
3829         case GST_RTSP_LOWER_TRANS_UDP:
3830           /* only allow unicast for other streams */
3831           GST_DEBUG_OBJECT (sink, "stream %p as UDP unicast", stream);
3832           protocols = GST_RTSP_LOWER_TRANS_UDP;
3833           /* Update transport with server destination if not provided by the server */
3834           if (transport->destination == NULL) {
3835             transport->destination = g_strdup (sink->server_ip);
3836           }
3837           break;
3838         default:
3839           GST_DEBUG_OBJECT (sink, "stream %p unknown transport %d", stream,
3840               transport->lower_transport);
3841           break;
3842       }
3843
3844       if (!retry) {
3845         GST_DEBUG ("Configuring the stream transport for stream %d",
3846             context->index);
3847         if (context->stream_transport == NULL)
3848           context->stream_transport =
3849               gst_rtsp_stream_transport_new (stream, transport);
3850         else
3851           gst_rtsp_stream_transport_set_transport (context->stream_transport,
3852               transport);
3853
3854         if (transport->lower_transport == GST_RTSP_LOWER_TRANS_TCP) {
3855           /* our callbacks to send data on this TCP connection */
3856           gst_rtsp_stream_transport_set_callbacks (context->stream_transport,
3857               (GstRTSPSendFunc) do_send_data,
3858               (GstRTSPSendFunc) do_send_data, context, NULL);
3859         }
3860
3861         /* The stream_transport now owns the transport */
3862         transport = NULL;
3863
3864         gst_rtsp_stream_transport_set_active (context->stream_transport, TRUE);
3865       }
3866     next:
3867       if (transport)
3868         gst_rtsp_transport_free (transport);
3869       /* clean up used RTSP messages */
3870       gst_rtsp_message_unset (&request);
3871       gst_rtsp_message_unset (&response);
3872     }
3873   }
3874   GST_RTSP_STATE_UNLOCK (sink);
3875
3876   /* store the transport protocol that was configured */
3877   sink->cur_protocols = protocols;
3878
3879   return res;
3880
3881 no_streams:
3882   {
3883     GST_RTSP_STATE_UNLOCK (sink);
3884     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
3885         ("SDP contains no streams"));
3886     return GST_RTSP_ERROR;
3887   }
3888 setup_transport_failed:
3889   {
3890     GST_RTSP_STATE_UNLOCK (sink);
3891     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
3892         ("Could not setup transport."));
3893     res = GST_RTSP_ERROR;
3894     goto cleanup_error;
3895   }
3896 no_profiles:
3897   {
3898     GST_RTSP_STATE_UNLOCK (sink);
3899     /* no transport possible, post an error and stop */
3900     GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
3901         ("Could not connect to server, no profiles left"));
3902     return GST_RTSP_ERROR;
3903   }
3904 no_protocols:
3905   {
3906     GST_RTSP_STATE_UNLOCK (sink);
3907     /* no transport possible, post an error and stop */
3908     GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
3909         ("Could not connect to server, no protocols left"));
3910     return GST_RTSP_ERROR;
3911   }
3912 no_transport:
3913   {
3914     GST_RTSP_STATE_UNLOCK (sink);
3915     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
3916         ("Server did not select transport."));
3917     res = GST_RTSP_ERROR;
3918     goto cleanup_error;
3919   }
3920 create_request_failed:
3921   {
3922     gchar *str = gst_rtsp_strresult (res);
3923
3924     GST_RTSP_STATE_UNLOCK (sink);
3925     GST_ELEMENT_ERROR (sink, LIBRARY, INIT, (NULL),
3926         ("Could not create request. (%s)", str));
3927     g_free (str);
3928     goto cleanup_error;
3929   }
3930 send_error:
3931   {
3932     gchar *str = gst_rtsp_strresult (res);
3933
3934     GST_RTSP_STATE_UNLOCK (sink);
3935     if (res != GST_RTSP_EINTR) {
3936       GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
3937           ("Could not send message. (%s)", str));
3938     } else {
3939       GST_WARNING_OBJECT (sink, "send interrupted");
3940     }
3941     g_free (str);
3942     goto cleanup_error;
3943   }
3944 response_error:
3945   {
3946     const gchar *str = gst_rtsp_status_as_text (code);
3947
3948     GST_RTSP_STATE_UNLOCK (sink);
3949     GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
3950         ("Error (%d): %s", code, GST_STR_NULL (str)));
3951     res = GST_RTSP_ERROR;
3952     goto cleanup_error;
3953   }
3954 cleanup_error:
3955   {
3956     gst_rtsp_message_unset (&request);
3957     gst_rtsp_message_unset (&response);
3958     return res;
3959   }
3960 }
3961
3962 static GstRTSPResult
3963 gst_rtsp_client_sink_ensure_open (GstRTSPClientSink * sink, gboolean async)
3964 {
3965   GstRTSPResult res = GST_RTSP_OK;
3966
3967   if (sink->state < GST_RTSP_STATE_READY) {
3968     res = GST_RTSP_ERROR;
3969     if (sink->open_error) {
3970       GST_DEBUG_OBJECT (sink, "the stream was in error");
3971       goto done;
3972     }
3973     if (async)
3974       gst_rtsp_client_sink_loop_start_cmd (sink, CMD_OPEN);
3975
3976     if ((res = gst_rtsp_client_sink_open (sink, async)) < 0) {
3977       GST_DEBUG_OBJECT (sink, "failed to open stream");
3978       goto done;
3979     }
3980   }
3981
3982 done:
3983   return res;
3984 }
3985
3986 static GstRTSPResult
3987 gst_rtsp_client_sink_record (GstRTSPClientSink * sink, gboolean async)
3988 {
3989   GstRTSPMessage request = { 0 };
3990   GstRTSPMessage response = { 0 };
3991   GstRTSPResult res = GST_RTSP_OK;
3992   GstSDPMessage *sdp;
3993   guint sdp_index = 0;
3994   GstSDPInfo info = { 0, };
3995
3996   const gchar *proto;
3997   gchar *sess_id, *client_ip, *str;
3998   GSocketAddress *sa;
3999   GInetAddress *ia;
4000   GSocket *conn_socket;
4001   GList *walk;
4002
4003   /* Wait for streams to preroll */
4004   g_mutex_lock (&sink->preroll_lock);
4005   while (sink->in_async) {
4006     GST_LOG_OBJECT (sink, "Waiting for ASYNC_DONE preroll");
4007     g_cond_wait (&sink->preroll_cond, &sink->preroll_lock);
4008   }
4009   g_mutex_unlock (&sink->preroll_lock);
4010
4011   if (sink->state == GST_RTSP_STATE_PLAYING) {
4012     /* Already recording, don't send another request */
4013     GST_LOG_OBJECT (sink, "Already in RECORD. Skipping duplicate request.");
4014     goto done;
4015   }
4016
4017   /* Send announce, then setup for all streams */
4018   gst_sdp_message_init (&sink->cursdp);
4019   sdp = &sink->cursdp;
4020
4021   /* some standard things first */
4022   gst_sdp_message_set_version (sdp, "0");
4023
4024   /* session ID doesn't have to be super-unique in this case */
4025   sess_id = g_strdup_printf ("%u", g_random_int ());
4026
4027   if (sink->conninfo.connection == NULL)
4028     return GST_RTSP_ERROR;
4029
4030   conn_socket = gst_rtsp_connection_get_read_socket (sink->conninfo.connection);
4031
4032   sa = g_socket_get_local_address (conn_socket, NULL);
4033   ia = g_inet_socket_address_get_address (G_INET_SOCKET_ADDRESS (sa));
4034   client_ip = g_inet_address_to_string (ia);
4035   if (g_socket_address_get_family (sa) == G_SOCKET_FAMILY_IPV6) {
4036     info.is_ipv6 = TRUE;
4037     proto = "IP6";
4038   } else if (g_socket_address_get_family (sa) == G_SOCKET_FAMILY_IPV4)
4039     proto = "IP4";
4040   else
4041     g_assert_not_reached ();
4042   g_object_unref (sa);
4043
4044   /* FIXME: Should this actually be the server's IP or ours? */
4045   info.server_ip = sink->server_ip;
4046
4047   gst_sdp_message_set_origin (sdp, "-", sess_id, "1", "IN", proto, client_ip);
4048
4049   gst_sdp_message_set_session_name (sdp, "Session streamed with GStreamer");
4050   gst_sdp_message_set_information (sdp, "rtspclientsink");
4051   gst_sdp_message_add_time (sdp, "0", "0", NULL);
4052   gst_sdp_message_add_attribute (sdp, "tool", "GStreamer");
4053
4054   /* add stream */
4055   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
4056     GstRTSPStreamContext *context = (GstRTSPStreamContext *) walk->data;
4057
4058     gst_rtsp_sdp_from_stream (sdp, &info, context->stream);
4059     context->sdp_index = sdp_index++;
4060   }
4061
4062   g_free (sess_id);
4063   g_free (client_ip);
4064
4065   /* send ANNOUNCE request */
4066   GST_DEBUG_OBJECT (sink, "create ANNOUNCE request...");
4067   res =
4068       gst_rtsp_client_sink_init_request (sink, &request, GST_RTSP_ANNOUNCE,
4069       sink->conninfo.url_str);
4070   if (res < 0)
4071     goto create_request_failed;
4072
4073   gst_rtsp_message_add_header (&request, GST_RTSP_HDR_CONTENT_TYPE,
4074       "application/sdp");
4075
4076   /* add SDP to the request body */
4077   str = gst_sdp_message_as_text (sdp);
4078   gst_rtsp_message_take_body (&request, (guint8 *) str, strlen (str));
4079
4080   /* send ANNOUNCE */
4081   GST_DEBUG_OBJECT (sink, "sending announce...");
4082
4083   if (async)
4084     GST_ELEMENT_PROGRESS (sink, CONTINUE, "record",
4085         ("Sending server stream info"));
4086
4087   if ((res =
4088           gst_rtsp_client_sink_send (sink, sink->conninfo.connection, &request,
4089               &response, NULL)) < 0)
4090     goto send_error;
4091
4092   /* send setup for all streams */
4093   if ((res = gst_rtsp_client_sink_setup_streams (sink, async)) < 0)
4094     goto setup_failed;
4095
4096   res = gst_rtsp_client_sink_init_request (sink, &request, GST_RTSP_RECORD,
4097       sink->conninfo.url_str);
4098
4099   if (res < 0)
4100     goto create_request_failed;
4101
4102 #if 0                           /* FIXME: Configure a range based on input segments? */
4103   if (src->need_range) {
4104     hval = gen_range_header (src, segment);
4105
4106     gst_rtsp_message_take_header (&request, GST_RTSP_HDR_RANGE, hval);
4107   }
4108
4109   if (segment->rate != 1.0) {
4110     gchar hval[G_ASCII_DTOSTR_BUF_SIZE];
4111
4112     g_ascii_dtostr (hval, sizeof (hval), segment->rate);
4113     if (src->skip)
4114       gst_rtsp_message_add_header (&request, GST_RTSP_HDR_SCALE, hval);
4115     else
4116       gst_rtsp_message_add_header (&request, GST_RTSP_HDR_SPEED, hval);
4117   }
4118 #endif
4119
4120   if (async)
4121     GST_ELEMENT_PROGRESS (sink, CONTINUE, "record", ("Starting recording"));
4122   if ((res =
4123           gst_rtsp_client_sink_send (sink, sink->conninfo.connection, &request,
4124               &response, NULL)) < 0)
4125     goto send_error;
4126
4127 #if 0                           /* FIXME: Check if servers return these for record: */
4128   /* parse the RTP-Info header field (if ANY) to get the base seqnum and timestamp
4129    * for the RTP packets. If this is not present, we assume all starts from 0...
4130    * This is info for the RTP session manager that we pass to it in caps. */
4131   hval_idx = 0;
4132   while (gst_rtsp_message_get_header (&response, GST_RTSP_HDR_RTP_INFO,
4133           &hval, hval_idx++) == GST_RTSP_OK)
4134     gst_rtspsrc_parse_rtpinfo (src, hval);
4135
4136   /* some servers indicate RTCP parameters in PLAY response,
4137    * rather than properly in SDP */
4138   if (gst_rtsp_message_get_header (&response, GST_RTSP_HDR_RTCP_INTERVAL,
4139           &hval, 0) == GST_RTSP_OK)
4140     gst_rtspsrc_handle_rtcp_interval (src, hval);
4141 #endif
4142
4143   gst_rtsp_client_sink_set_state (sink, GST_STATE_PLAYING);
4144   sink->state = GST_RTSP_STATE_PLAYING;
4145
4146   /* clean up any messages */
4147   gst_rtsp_message_unset (&request);
4148   gst_rtsp_message_unset (&response);
4149
4150 done:
4151   return res;
4152
4153 create_request_failed:
4154   {
4155     gchar *str = gst_rtsp_strresult (res);
4156
4157     GST_ELEMENT_ERROR (sink, LIBRARY, INIT, (NULL),
4158         ("Could not create request. (%s)", str));
4159     g_free (str);
4160     goto cleanup_error;
4161   }
4162 send_error:
4163   {
4164     /* Don't post a message - the rtsp_send method will have
4165      * taken care of it because we passed NULL for the response code */
4166     goto cleanup_error;
4167   }
4168 setup_failed:
4169   {
4170     GST_ERROR_OBJECT (sink, "setup failed");
4171     goto cleanup_error;
4172   }
4173 cleanup_error:
4174   {
4175     if (sink->conninfo.connection) {
4176       GST_DEBUG_OBJECT (sink, "free connection");
4177       gst_rtsp_conninfo_close (sink, &sink->conninfo, TRUE);
4178     }
4179     gst_rtsp_message_unset (&request);
4180     gst_rtsp_message_unset (&response);
4181     return res;
4182   }
4183 }
4184
4185 static GstRTSPResult
4186 gst_rtsp_client_sink_pause (GstRTSPClientSink * sink, gboolean async)
4187 {
4188   GstRTSPResult res = GST_RTSP_OK;
4189   GstRTSPMessage request = { 0 };
4190   GstRTSPMessage response = { 0 };
4191   GList *walk;
4192   const gchar *control;
4193
4194   GST_DEBUG_OBJECT (sink, "PAUSE...");
4195
4196   if ((res = gst_rtsp_client_sink_ensure_open (sink, async)) < 0)
4197     goto open_failed;
4198
4199   if (!(sink->methods & GST_RTSP_PAUSE))
4200     goto not_supported;
4201
4202   if (sink->state == GST_RTSP_STATE_READY)
4203     goto was_paused;
4204
4205   if (!sink->conninfo.connection || !sink->conninfo.connected)
4206     goto no_connection;
4207
4208   /* construct a control url */
4209   control = get_aggregate_control (sink);
4210
4211   /* loop over the streams. We might exit the loop early when we could do an
4212    * aggregate control */
4213   for (walk = sink->contexts; walk; walk = g_list_next (walk)) {
4214     GstRTSPStreamContext *stream = (GstRTSPStreamContext *) walk->data;
4215     GstRTSPConnection *conn;
4216     const gchar *setup_url;
4217
4218     /* try aggregate control first but do non-aggregate control otherwise */
4219     if (control)
4220       setup_url = control;
4221     else if ((setup_url = stream->conninfo.location) == NULL)
4222       continue;
4223
4224     if (sink->conninfo.connection) {
4225       conn = sink->conninfo.connection;
4226     } else if (stream->conninfo.connection) {
4227       conn = stream->conninfo.connection;
4228     } else {
4229       continue;
4230     }
4231
4232     if (async)
4233       GST_ELEMENT_PROGRESS (sink, CONTINUE, "request",
4234           ("Sending PAUSE request"));
4235
4236     if ((res =
4237             gst_rtsp_client_sink_init_request (sink, &request, GST_RTSP_PAUSE,
4238                 setup_url)) < 0)
4239       goto create_request_failed;
4240
4241     if ((res =
4242             gst_rtsp_client_sink_send (sink, conn, &request, &response,
4243                 NULL)) < 0)
4244       goto send_error;
4245
4246     gst_rtsp_message_unset (&request);
4247     gst_rtsp_message_unset (&response);
4248
4249     /* exit early when we did agregate control */
4250     if (control)
4251       break;
4252   }
4253
4254   /* change element states now */
4255   gst_rtsp_client_sink_set_state (sink, GST_STATE_PAUSED);
4256
4257 no_connection:
4258   sink->state = GST_RTSP_STATE_READY;
4259
4260 done:
4261   if (async)
4262     gst_rtsp_client_sink_loop_end_cmd (sink, CMD_PAUSE, res);
4263
4264   return res;
4265
4266   /* ERRORS */
4267 open_failed:
4268   {
4269     GST_DEBUG_OBJECT (sink, "failed to open stream");
4270     goto done;
4271   }
4272 not_supported:
4273   {
4274     GST_DEBUG_OBJECT (sink, "PAUSE is not supported");
4275     goto done;
4276   }
4277 was_paused:
4278   {
4279     GST_DEBUG_OBJECT (sink, "we were already PAUSED");
4280     goto done;
4281   }
4282 create_request_failed:
4283   {
4284     gchar *str = gst_rtsp_strresult (res);
4285
4286     GST_ELEMENT_ERROR (sink, LIBRARY, INIT, (NULL),
4287         ("Could not create request. (%s)", str));
4288     g_free (str);
4289     goto done;
4290   }
4291 send_error:
4292   {
4293     gchar *str = gst_rtsp_strresult (res);
4294
4295     gst_rtsp_message_unset (&request);
4296     if (res != GST_RTSP_EINTR) {
4297       GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
4298           ("Could not send message. (%s)", str));
4299     } else {
4300       GST_WARNING_OBJECT (sink, "PAUSE interrupted");
4301     }
4302     g_free (str);
4303     goto done;
4304   }
4305 }
4306
4307 static void
4308 gst_rtsp_client_sink_handle_message (GstBin * bin, GstMessage * message)
4309 {
4310   GstRTSPClientSink *rtsp_client_sink;
4311
4312   rtsp_client_sink = GST_RTSP_CLIENT_SINK (bin);
4313
4314   switch (GST_MESSAGE_TYPE (message)) {
4315     case GST_MESSAGE_ELEMENT:
4316     {
4317       const GstStructure *s = gst_message_get_structure (message);
4318
4319       if (gst_structure_has_name (s, "GstUDPSrcTimeout")) {
4320         gboolean ignore_timeout;
4321
4322         GST_DEBUG_OBJECT (bin, "timeout on UDP port");
4323
4324         GST_OBJECT_LOCK (rtsp_client_sink);
4325         ignore_timeout = rtsp_client_sink->ignore_timeout;
4326         rtsp_client_sink->ignore_timeout = TRUE;
4327         GST_OBJECT_UNLOCK (rtsp_client_sink);
4328
4329         /* we only act on the first udp timeout message, others are irrelevant
4330          * and can be ignored. */
4331         if (!ignore_timeout)
4332           gst_rtsp_client_sink_loop_send_cmd (rtsp_client_sink, CMD_RECONNECT,
4333               CMD_LOOP);
4334         /* eat and free */
4335         gst_message_unref (message);
4336         return;
4337       } else if (gst_structure_has_name (s, "GstRTSPStreamBlocking")) {
4338         /* An RTSPStream has prerolled */
4339         g_cond_broadcast (&rtsp_client_sink->preroll_cond);
4340       }
4341       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
4342       break;
4343     }
4344     case GST_MESSAGE_ASYNC_START:{
4345       GstObject *sender;
4346
4347       sender = GST_MESSAGE_SRC (message);
4348
4349       GST_LOG_OBJECT (rtsp_client_sink,
4350           "Have async-start from %" GST_PTR_FORMAT, sender);
4351       if (sender == GST_OBJECT (rtsp_client_sink->internal_bin)) {
4352         GST_LOG_OBJECT (rtsp_client_sink, "child bin is now ASYNC");
4353       }
4354       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
4355       break;
4356     }
4357     case GST_MESSAGE_ASYNC_DONE:
4358     {
4359       GstObject *sender;
4360       gboolean need_async_done;
4361
4362       sender = GST_MESSAGE_SRC (message);
4363       GST_LOG_OBJECT (rtsp_client_sink, "Have async-done from %" GST_PTR_FORMAT,
4364           sender);
4365
4366       g_mutex_lock (&rtsp_client_sink->preroll_lock);
4367       if (sender == GST_OBJECT_CAST (rtsp_client_sink->internal_bin)) {
4368         GST_LOG_OBJECT (rtsp_client_sink, "child bin is no longer ASYNC");
4369       }
4370       need_async_done = rtsp_client_sink->in_async;
4371       if (rtsp_client_sink->in_async) {
4372         rtsp_client_sink->in_async = FALSE;
4373         g_cond_broadcast (&rtsp_client_sink->preroll_cond);
4374       }
4375       g_mutex_unlock (&rtsp_client_sink->preroll_lock);
4376
4377       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
4378
4379       if (need_async_done) {
4380         GST_DEBUG_OBJECT (rtsp_client_sink, "Posting ASYNC-DONE");
4381         gst_element_post_message (GST_ELEMENT_CAST (rtsp_client_sink),
4382             gst_message_new_async_done (GST_OBJECT_CAST (rtsp_client_sink),
4383                 GST_CLOCK_TIME_NONE));
4384       }
4385       break;
4386     }
4387     case GST_MESSAGE_ERROR:
4388     {
4389       GstObject *sender;
4390
4391       sender = GST_MESSAGE_SRC (message);
4392
4393       GST_DEBUG_OBJECT (rtsp_client_sink, "got error from %s",
4394           GST_ELEMENT_NAME (sender));
4395
4396       /* FIXME: Ignore errors on RTCP? */
4397       /* fatal but not our message, forward */
4398       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
4399       break;
4400     }
4401     case GST_MESSAGE_STATE_CHANGED:
4402     {
4403       if (GST_MESSAGE_SRC (message) ==
4404           (GstObject *) rtsp_client_sink->internal_bin) {
4405         GstState newstate, pending;
4406         gst_message_parse_state_changed (message, NULL, &newstate, &pending);
4407         g_mutex_lock (&rtsp_client_sink->preroll_lock);
4408         rtsp_client_sink->prerolled = (newstate >= GST_STATE_PAUSED)
4409             && pending == GST_STATE_VOID_PENDING;
4410         g_cond_broadcast (&rtsp_client_sink->preroll_cond);
4411         g_mutex_unlock (&rtsp_client_sink->preroll_lock);
4412         GST_DEBUG_OBJECT (bin,
4413             "Internal bin changed state to %s (pending %s). Prerolled now %d",
4414             gst_element_state_get_name (newstate),
4415             gst_element_state_get_name (pending), rtsp_client_sink->prerolled);
4416       }
4417     }
4418     default:
4419     {
4420       GST_BIN_CLASS (parent_class)->handle_message (bin, message);
4421       break;
4422     }
4423   }
4424 }
4425
4426 /* the thread where everything happens */
4427 static void
4428 gst_rtsp_client_sink_thread (GstRTSPClientSink * sink)
4429 {
4430   gint cmd;
4431
4432   GST_OBJECT_LOCK (sink);
4433   cmd = sink->pending_cmd;
4434   if (cmd == CMD_RECONNECT || cmd == CMD_RECORD || cmd == CMD_PAUSE
4435       || cmd == CMD_LOOP || cmd == CMD_OPEN)
4436     sink->pending_cmd = CMD_LOOP;
4437   else
4438     sink->pending_cmd = CMD_WAIT;
4439   GST_DEBUG_OBJECT (sink, "got command %s", cmd_to_string (cmd));
4440
4441   /* we got the message command, so ensure communication is possible again */
4442   gst_rtsp_client_sink_connection_flush (sink, FALSE);
4443
4444   sink->busy_cmd = cmd;
4445   GST_OBJECT_UNLOCK (sink);
4446
4447   switch (cmd) {
4448     case CMD_OPEN:
4449       gst_rtsp_client_sink_open (sink, TRUE);
4450       break;
4451     case CMD_RECORD:
4452       gst_rtsp_client_sink_record (sink, TRUE);
4453       break;
4454     case CMD_PAUSE:
4455       gst_rtsp_client_sink_pause (sink, TRUE);
4456       break;
4457     case CMD_CLOSE:
4458       gst_rtsp_client_sink_close (sink, TRUE, FALSE);
4459       break;
4460     case CMD_LOOP:
4461       gst_rtsp_client_sink_loop (sink);
4462       break;
4463     case CMD_RECONNECT:
4464       gst_rtsp_client_sink_reconnect (sink, FALSE);
4465       break;
4466     default:
4467       break;
4468   }
4469
4470   GST_OBJECT_LOCK (sink);
4471   /* and go back to sleep */
4472   if (sink->pending_cmd == CMD_WAIT) {
4473     if (sink->task)
4474       gst_task_pause (sink->task);
4475   }
4476   /* reset waiting */
4477   sink->busy_cmd = CMD_WAIT;
4478   GST_OBJECT_UNLOCK (sink);
4479 }
4480
4481 static gboolean
4482 gst_rtsp_client_sink_start (GstRTSPClientSink * sink)
4483 {
4484   GST_DEBUG_OBJECT (sink, "starting");
4485
4486   sink->streams_collected = FALSE;
4487   sink->in_async = TRUE;
4488   gst_element_set_locked_state (GST_ELEMENT (sink->internal_bin), TRUE);
4489
4490   gst_rtsp_client_sink_set_state (sink, GST_STATE_READY);
4491
4492   GST_OBJECT_LOCK (sink);
4493   sink->pending_cmd = CMD_WAIT;
4494
4495   if (sink->task == NULL) {
4496     sink->task =
4497         gst_task_new ((GstTaskFunction) gst_rtsp_client_sink_thread, sink,
4498         NULL);
4499     if (sink->task == NULL)
4500       goto task_error;
4501
4502     gst_task_set_lock (sink->task, GST_RTSP_STREAM_GET_LOCK (sink));
4503   }
4504   GST_OBJECT_UNLOCK (sink);
4505
4506   return TRUE;
4507
4508   /* ERRORS */
4509 task_error:
4510   {
4511     GST_OBJECT_UNLOCK (sink);
4512     GST_ERROR_OBJECT (sink, "failed to create task");
4513     return FALSE;
4514   }
4515 }
4516
4517 static gboolean
4518 gst_rtsp_client_sink_stop (GstRTSPClientSink * sink)
4519 {
4520   GstTask *task;
4521
4522   GST_DEBUG_OBJECT (sink, "stopping");
4523
4524   /* also cancels pending task */
4525   gst_rtsp_client_sink_loop_send_cmd (sink, CMD_WAIT, CMD_ALL & ~CMD_CLOSE);
4526
4527   GST_OBJECT_LOCK (sink);
4528   if ((task = sink->task)) {
4529     sink->task = NULL;
4530     GST_OBJECT_UNLOCK (sink);
4531
4532     gst_task_stop (task);
4533
4534     /* make sure it is not running */
4535     GST_RTSP_STREAM_LOCK (sink);
4536     GST_RTSP_STREAM_UNLOCK (sink);
4537
4538     /* now wait for the task to finish */
4539     gst_task_join (task);
4540
4541     /* and free the task */
4542     gst_object_unref (GST_OBJECT (task));
4543
4544     GST_OBJECT_LOCK (sink);
4545   }
4546   GST_OBJECT_UNLOCK (sink);
4547
4548   /* ensure synchronously all is closed and clean */
4549   gst_rtsp_client_sink_close (sink, FALSE, TRUE);
4550
4551   return TRUE;
4552 }
4553
4554 static GstStateChangeReturn
4555 gst_rtsp_client_sink_change_state (GstElement * element,
4556     GstStateChange transition)
4557 {
4558   GstRTSPClientSink *rtsp_client_sink;
4559   GstStateChangeReturn ret;
4560
4561   rtsp_client_sink = GST_RTSP_CLIENT_SINK (element);
4562
4563   switch (transition) {
4564     case GST_STATE_CHANGE_NULL_TO_READY:
4565       if (!gst_rtsp_client_sink_start (rtsp_client_sink))
4566         goto start_failed;
4567       break;
4568     case GST_STATE_CHANGE_READY_TO_PAUSED:
4569       /* init some state */
4570       rtsp_client_sink->cur_protocols = rtsp_client_sink->protocols;
4571       /* first attempt, don't ignore timeouts */
4572       rtsp_client_sink->ignore_timeout = FALSE;
4573       rtsp_client_sink->open_error = FALSE;
4574
4575       gst_rtsp_client_sink_set_state (rtsp_client_sink, GST_STATE_PAUSED);
4576
4577       g_mutex_lock (&rtsp_client_sink->preroll_lock);
4578       if (rtsp_client_sink->in_async) {
4579         GST_DEBUG_OBJECT (rtsp_client_sink, "Posting ASYNC-START");
4580         gst_element_post_message (GST_ELEMENT_CAST (rtsp_client_sink),
4581             gst_message_new_async_start (GST_OBJECT_CAST (rtsp_client_sink)));
4582       }
4583       g_mutex_unlock (&rtsp_client_sink->preroll_lock);
4584
4585       break;
4586     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
4587       /* fall-through */
4588     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
4589       /* unblock the tcp tasks and make the loop waiting */
4590       if (gst_rtsp_client_sink_loop_send_cmd (rtsp_client_sink, CMD_WAIT,
4591               CMD_LOOP)) {
4592         /* make sure it is waiting before we send PLAY below */
4593         GST_RTSP_STREAM_LOCK (rtsp_client_sink);
4594         GST_RTSP_STREAM_UNLOCK (rtsp_client_sink);
4595       }
4596       break;
4597     case GST_STATE_CHANGE_PAUSED_TO_READY:
4598       gst_rtsp_client_sink_set_state (rtsp_client_sink, GST_STATE_READY);
4599       break;
4600     default:
4601       break;
4602   }
4603
4604   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
4605   if (ret == GST_STATE_CHANGE_FAILURE)
4606     goto done;
4607
4608   switch (transition) {
4609     case GST_STATE_CHANGE_NULL_TO_READY:
4610       ret = GST_STATE_CHANGE_SUCCESS;
4611       break;
4612     case GST_STATE_CHANGE_READY_TO_PAUSED:
4613       /* Return ASYNC and preroll input streams */
4614       g_mutex_lock (&rtsp_client_sink->preroll_lock);
4615       if (rtsp_client_sink->in_async)
4616         ret = GST_STATE_CHANGE_ASYNC;
4617       g_mutex_unlock (&rtsp_client_sink->preroll_lock);
4618       gst_rtsp_client_sink_loop_send_cmd (rtsp_client_sink, CMD_OPEN, 0);
4619       break;
4620     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:{
4621       GST_DEBUG_OBJECT (rtsp_client_sink,
4622           "Switching to playing -sending RECORD");
4623       gst_rtsp_client_sink_loop_send_cmd (rtsp_client_sink, CMD_RECORD, 0);
4624       ret = GST_STATE_CHANGE_SUCCESS;
4625       break;
4626     }
4627     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
4628       /* send pause request and keep the idle task around */
4629       gst_rtsp_client_sink_loop_send_cmd (rtsp_client_sink, CMD_PAUSE,
4630           CMD_LOOP);
4631       ret = GST_STATE_CHANGE_NO_PREROLL;
4632       break;
4633     case GST_STATE_CHANGE_PAUSED_TO_READY:
4634       gst_rtsp_client_sink_loop_send_cmd (rtsp_client_sink, CMD_CLOSE,
4635           CMD_PAUSE);
4636       ret = GST_STATE_CHANGE_SUCCESS;
4637       break;
4638     case GST_STATE_CHANGE_READY_TO_NULL:
4639       gst_rtsp_client_sink_stop (rtsp_client_sink);
4640       ret = GST_STATE_CHANGE_SUCCESS;
4641       break;
4642     default:
4643       break;
4644   }
4645
4646 done:
4647   return ret;
4648
4649 start_failed:
4650   {
4651     GST_DEBUG_OBJECT (rtsp_client_sink, "start failed");
4652     return GST_STATE_CHANGE_FAILURE;
4653   }
4654 }
4655
4656 /*** GSTURIHANDLER INTERFACE *************************************************/
4657
4658 static GstURIType
4659 gst_rtsp_client_sink_uri_get_type (GType type)
4660 {
4661   return GST_URI_SINK;
4662 }
4663
4664 static const gchar *const *
4665 gst_rtsp_client_sink_uri_get_protocols (GType type)
4666 {
4667   static const gchar *protocols[] =
4668       { "rtsp", "rtspu", "rtspt", "rtsph", "rtsp-sdp",
4669     "rtsps", "rtspsu", "rtspst", "rtspsh", NULL
4670   };
4671
4672   return protocols;
4673 }
4674
4675 static gchar *
4676 gst_rtsp_client_sink_uri_get_uri (GstURIHandler * handler)
4677 {
4678   GstRTSPClientSink *sink = GST_RTSP_CLIENT_SINK (handler);
4679
4680   /* FIXME: make thread-safe */
4681   return g_strdup (sink->conninfo.location);
4682 }
4683
4684 static gboolean
4685 gst_rtsp_client_sink_uri_set_uri (GstURIHandler * handler, const gchar * uri,
4686     GError ** error)
4687 {
4688   GstRTSPClientSink *sink;
4689   GstRTSPResult res;
4690   GstSDPResult sres;
4691   GstRTSPUrl *newurl = NULL;
4692   GstSDPMessage *sdp = NULL;
4693
4694   sink = GST_RTSP_CLIENT_SINK (handler);
4695
4696   /* same URI, we're fine */
4697   if (sink->conninfo.location && uri && !strcmp (uri, sink->conninfo.location))
4698     goto was_ok;
4699
4700   if (g_str_has_prefix (uri, "rtsp-sdp://")) {
4701     sres = gst_sdp_message_new (&sdp);
4702     if (sres < 0)
4703       goto sdp_failed;
4704
4705     GST_DEBUG_OBJECT (sink, "parsing SDP message");
4706     sres = gst_sdp_message_parse_uri (uri, sdp);
4707     if (sres < 0)
4708       goto invalid_sdp;
4709   } else {
4710     /* try to parse */
4711     GST_DEBUG_OBJECT (sink, "parsing URI");
4712     if ((res = gst_rtsp_url_parse (uri, &newurl)) < 0)
4713       goto parse_error;
4714   }
4715
4716   /* if worked, free previous and store new url object along with the original
4717    * location. */
4718   GST_DEBUG_OBJECT (sink, "configuring URI");
4719   g_free (sink->conninfo.location);
4720   sink->conninfo.location = g_strdup (uri);
4721   gst_rtsp_url_free (sink->conninfo.url);
4722   sink->conninfo.url = newurl;
4723   g_free (sink->conninfo.url_str);
4724   if (newurl)
4725     sink->conninfo.url_str = gst_rtsp_url_get_request_uri (sink->conninfo.url);
4726   else
4727     sink->conninfo.url_str = NULL;
4728
4729   if (sink->uri_sdp)
4730     gst_sdp_message_free (sink->uri_sdp);
4731   sink->uri_sdp = sdp;
4732   sink->from_sdp = sdp != NULL;
4733
4734   GST_DEBUG_OBJECT (sink, "set uri: %s", GST_STR_NULL (uri));
4735   GST_DEBUG_OBJECT (sink, "request uri is: %s",
4736       GST_STR_NULL (sink->conninfo.url_str));
4737
4738   return TRUE;
4739
4740   /* Special cases */
4741 was_ok:
4742   {
4743     GST_DEBUG_OBJECT (sink, "URI was ok: '%s'", GST_STR_NULL (uri));
4744     return TRUE;
4745   }
4746 sdp_failed:
4747   {
4748     GST_ERROR_OBJECT (sink, "Could not create new SDP (%d)", sres);
4749     g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
4750         "Could not create SDP");
4751     return FALSE;
4752   }
4753 invalid_sdp:
4754   {
4755     GST_ERROR_OBJECT (sink, "Not a valid SDP (%d) '%s'", sres,
4756         GST_STR_NULL (uri));
4757     gst_sdp_message_free (sdp);
4758     g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
4759         "Invalid SDP");
4760     return FALSE;
4761   }
4762 parse_error:
4763   {
4764     GST_ERROR_OBJECT (sink, "Not a valid RTSP url '%s' (%d)",
4765         GST_STR_NULL (uri), res);
4766     g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
4767         "Invalid RTSP URI");
4768     return FALSE;
4769   }
4770 }
4771
4772 static void
4773 gst_rtsp_client_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
4774 {
4775   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
4776
4777   iface->get_type = gst_rtsp_client_sink_uri_get_type;
4778   iface->get_protocols = gst_rtsp_client_sink_uri_get_protocols;
4779   iface->get_uri = gst_rtsp_client_sink_uri_get_uri;
4780   iface->set_uri = gst_rtsp_client_sink_uri_set_uri;
4781 }