gst-libs/gst/rtsp/gstrtspconnection.c

   1 /* GStreamer
   2  * Copyright (C) <2005-2009> Wim Taymans <wim.taymans@gmail.com>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19 /*
  20  * Unless otherwise indicated, Source Code is licensed under MIT license.
  21  * See further explanation attached in License Statement (distributed in the file
  22  * LICENSE).
  23  *
  24  * Permission is hereby granted, free of charge, to any person obtaining a copy of
  25  * this software and associated documentation files (the "Software"), to deal in
  26  * the Software without restriction, including without limitation the rights to
  27  * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
  28  * of the Software, and to permit persons to whom the Software is furnished to do
  29  * so, subject to the following conditions:
  30  *
  31  * The above copyright notice and this permission notice shall be included in all
  32  * copies or substantial portions of the Software.
  33  *
  34  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  35  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  36  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  37  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  38  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  39  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  40  * SOFTWARE.
  41  */
  42
  43 /**
  44  * SECTION:gstrtspconnection
  45  * @short_description: manage RTSP connections
  46  * @see_also: gstrtspurl
  47  *
  48  * This object manages the RTSP connection to the server. It provides function
  49  * to receive and send bytes and messages.
  50  *
  51  * Last reviewed on 2007-07-24 (0.10.14)
  52  */
  53
  54 #ifdef HAVE_CONFIG_H
  55 #  include <config.h>
  56 #endif
  57
  58 #include <stdio.h>
  59 #include <errno.h>
  60 #include <stdlib.h>
  61 #include <string.h>
  62 #include <time.h>
  63
  64 /* we include this here to get the G_OS_* defines */
  65 #include <glib.h>
  66 #include <gst/gst.h>
  67
  68 /* necessary for IP_TOS define */
  69 #if GLIB_CHECK_VERSION(2, 36, 0)
  70 #include <gio/gnetworking.h>
  71 #endif
  72
  73 #include "gstrtspconnection.h"
  74
  75 #ifdef IP_TOS
  76 union gst_sockaddr
  77 {
  78   struct sockaddr sa;
  79   struct sockaddr_in sa_in;
  80   struct sockaddr_in6 sa_in6;
  81   struct sockaddr_storage sa_stor;
  82 };
  83 #endif
  84
  85 typedef struct
  86 {
  87   gint state;
  88   guint save;
  89   guchar out[3];                /* the size must be evenly divisible by 3 */
  90   guint cout;
  91   guint coutl;
  92 } DecodeCtx;
  93
  94 #ifdef MSG_NOSIGNAL
  95 #define SEND_FLAGS MSG_NOSIGNAL
  96 #else
  97 #define SEND_FLAGS 0
  98 #endif
  99
 100 typedef enum
 101 {
 102   TUNNEL_STATE_NONE,
 103   TUNNEL_STATE_GET,
 104   TUNNEL_STATE_POST,
 105   TUNNEL_STATE_COMPLETE
 106 } GstRTSPTunnelState;
 107
 108 #define TUNNELID_LEN   24
 109
 110 struct _GstRTSPConnection
 111 {
 112   /*< private > */
 113   /* URL for the remote connection */
 114   GstRTSPUrl *url;
 115
 116   gboolean server;
 117   GSocketClient *client;
 118   GIOStream *stream0;
 119   GIOStream *stream1;
 120
 121   GInputStream *input_stream;
 122   GOutputStream *output_stream;
 123   /* this is a read source we add on the write socket in tunneled mode to be
 124    * able to detect when client disconnects the GET channel */
 125   GInputStream *control_stream;
 126
 127   /* connection state */
 128   GSocket *read_socket;
 129   GSocket *write_socket;
 130   GSocket *socket0, *socket1;
 131   gboolean manual_http;
 132   gboolean may_cancel;
 133   GCancellable *cancellable;
 134
 135   gchar tunnelid[TUNNELID_LEN];
 136   gboolean tunneled;
 137   GstRTSPTunnelState tstate;
 138
 139   /* the remote and local ip */
 140   gchar *remote_ip;
 141   gchar *local_ip;
 142
 143   gint read_ahead;
 144
 145   gchar *initial_buffer;
 146   gsize initial_buffer_offset;
 147
 148   gboolean remember_session_id; /* remember the session id or not */
 149
 150   /* Session state */
 151   gint cseq;                    /* sequence number */
 152   gchar session_id[512];        /* session id */
 153   gint timeout;                 /* session timeout in seconds */
 154   GTimer *timer;                /* timeout timer */
 155
 156   /* Authentication */
 157   GstRTSPAuthMethod auth_method;
 158   gchar *username;
 159   gchar *passwd;
 160   GHashTable *auth_params;
 161
 162   /* TLS */
 163   GTlsDatabase *tls_database;
 164
 165   DecodeCtx ctx;
 166   DecodeCtx *ctxp;
 167
 168   gchar *proxy_host;
 169   guint proxy_port;
 170 };
 171
 172 enum
 173 {
 174   STATE_START = 0,
 175   STATE_DATA_HEADER,
 176   STATE_DATA_BODY,
 177   STATE_READ_LINES,
 178   STATE_END,
 179   STATE_LAST
 180 };
 181
 182 enum
 183 {
 184   READ_AHEAD_EOH = -1,          /* end of headers */
 185   READ_AHEAD_CRLF = -2,
 186   READ_AHEAD_CRLFCR = -3
 187 };
 188
 189 /* a structure for constructing RTSPMessages */
 190 typedef struct
 191 {
 192   gint state;
 193   GstRTSPResult status;
 194   guint8 buffer[4096];
 195   guint offset;
 196
 197   guint line;
 198   guint8 *body_data;
 199   glong body_len;
 200 } GstRTSPBuilder;
 201
 202 static void
 203 build_reset (GstRTSPBuilder * builder)
 204 {
 205   g_free (builder->body_data);
 206   memset (builder, 0, sizeof (GstRTSPBuilder));
 207 }
 208
 209 static gboolean
 210 tls_accept_certificate (GTlsConnection * conn, GTlsCertificate * peer_cert,
 211     GTlsCertificateFlags errors, GstRTSPConnection * rtspconn)
 212 {
 213   GError *error = NULL;
 214   gboolean accept = FALSE;
 215
 216   if (rtspconn->tls_database) {
 217     GSocketConnectable *peer_identity;
 218     GTlsCertificateFlags validation_flags;
 219
 220     GST_DEBUG ("TLS peer certificate not accepted, checking user database...");
 221
 222     peer_identity =
 223         g_tls_client_connection_get_server_identity (G_TLS_CLIENT_CONNECTION
 224         (conn));
 225
 226     errors =
 227         g_tls_database_verify_chain (rtspconn->tls_database, peer_cert,
 228         G_TLS_DATABASE_PURPOSE_AUTHENTICATE_SERVER, peer_identity,
 229         g_tls_connection_get_interaction (conn), G_TLS_DATABASE_VERIFY_NONE,
 230         NULL, &error);
 231
 232     if (error)
 233       goto verify_error;
 234
 235     validation_flags = gst_rtsp_connection_get_tls_validation_flags (rtspconn);
 236
 237     accept = ((errors & validation_flags) == 0);
 238     if (accept)
 239       GST_DEBUG ("Peer certificate accepted");
 240     else
 241       GST_DEBUG ("Peer certificate not accepted (errors: 0x%08X)", errors);
 242   }
 243
 244   return accept;
 245
 246 /* ERRORS */
 247 verify_error:
 248   {
 249     GST_ERROR ("An error occurred while verifying the peer certificate: %s",
 250         error->message);
 251     g_clear_error (&error);
 252     return FALSE;
 253   }
 254 }
 255
 256 static void
 257 socket_client_event (GSocketClient * client, GSocketClientEvent event,
 258     GSocketConnectable * connectable, GIOStream * connection,
 259     GstRTSPConnection * rtspconn)
 260 {
 261   if (event == G_SOCKET_CLIENT_TLS_HANDSHAKING) {
 262     GST_DEBUG ("TLS handshaking about to start...");
 263
 264     g_signal_connect (connection, "accept-certificate",
 265         (GCallback) tls_accept_certificate, rtspconn);
 266   }
 267 }
 268
 269 /**
 270  * gst_rtsp_connection_create:
 271  * @url: a #GstRTSPUrl
 272  * @conn: (out) (transfer full): storage for a #GstRTSPConnection
 273  *
 274  * Create a newly allocated #GstRTSPConnection from @url and store it in @conn.
 275  * The connection will not yet attempt to connect to @url, use
 276  * gst_rtsp_connection_connect().
 277  *
 278  * A copy of @url will be made.
 279  *
 280  * Returns: #GST_RTSP_OK when @conn contains a valid connection.
 281  */
 282 GstRTSPResult
 283 gst_rtsp_connection_create (const GstRTSPUrl * url, GstRTSPConnection ** conn)
 284 {
 285   GstRTSPConnection *newconn;
 286
 287   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
 288   g_return_val_if_fail (url != NULL, GST_RTSP_EINVAL);
 289
 290   newconn = g_new0 (GstRTSPConnection, 1);
 291
 292   newconn->may_cancel = TRUE;
 293   newconn->cancellable = g_cancellable_new ();
 294   newconn->client = g_socket_client_new ();
 295
 296   if (url->transports & GST_RTSP_LOWER_TRANS_TLS)
 297     g_socket_client_set_tls (newconn->client, TRUE);
 298
 299   g_signal_connect (newconn->client, "event", (GCallback) socket_client_event,
 300       newconn);
 301
 302   newconn->url = gst_rtsp_url_copy (url);
 303   newconn->timer = g_timer_new ();
 304   newconn->timeout = 60;
 305   newconn->cseq = 1;
 306
 307   newconn->remember_session_id = TRUE;
 308
 309   newconn->auth_method = GST_RTSP_AUTH_NONE;
 310   newconn->username = NULL;
 311   newconn->passwd = NULL;
 312   newconn->auth_params = NULL;
 313
 314   *conn = newconn;
 315
 316   return GST_RTSP_OK;
 317 }
 318
 319 static gboolean
 320 collect_addresses (GSocket * socket, gchar ** ip, guint16 * port,
 321     gboolean remote, GError ** error)
 322 {
 323   GSocketAddress *addr;
 324
 325   if (remote)
 326     addr = g_socket_get_remote_address (socket, error);
 327   else
 328     addr = g_socket_get_local_address (socket, error);
 329   if (!addr)
 330     return FALSE;
 331
 332   if (ip)
 333     *ip = g_inet_address_to_string (g_inet_socket_address_get_address
 334         (G_INET_SOCKET_ADDRESS (addr)));
 335   if (port)
 336     *port = g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (addr));
 337
 338   g_object_unref (addr);
 339
 340   return TRUE;
 341 }
 342
 343
 344 /**
 345  * gst_rtsp_connection_create_from_socket:
 346  * @socket: a #GSocket
 347  * @ip: the IP address of the other end
 348  * @port: the port used by the other end
 349  * @initial_buffer: data already read from @fd
 350  * @conn: (out) (transfer full): storage for a #GstRTSPConnection
 351  *
 352  * Create a new #GstRTSPConnection for handling communication on the existing
 353  * socket @socket. The @initial_buffer contains zero terminated data already
 354  * read from @socket which should be used before starting to read new data.
 355  *
 356  * Returns: #GST_RTSP_OK when @conn contains a valid connection.
 357  */
 358 GstRTSPResult
 359 gst_rtsp_connection_create_from_socket (GSocket * socket, const gchar * ip,
 360     guint16 port, const gchar * initial_buffer, GstRTSPConnection ** conn)
 361 {
 362   GstRTSPConnection *newconn = NULL;
 363   GstRTSPUrl *url;
 364   GstRTSPResult res;
 365   GError *err = NULL;
 366   gchar *local_ip;
 367   GIOStream *stream;
 368
 369   g_return_val_if_fail (G_IS_SOCKET (socket), GST_RTSP_EINVAL);
 370   g_return_val_if_fail (ip != NULL, GST_RTSP_EINVAL);
 371   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
 372
 373   if (!collect_addresses (socket, &local_ip, NULL, FALSE, &err))
 374     goto getnameinfo_failed;
 375
 376   /* create a url for the client address */
 377   url = g_new0 (GstRTSPUrl, 1);
 378   url->host = g_strdup (ip);
 379   url->port = port;
 380
 381   /* now create the connection object */
 382   GST_RTSP_CHECK (gst_rtsp_connection_create (url, &newconn), newconn_failed);
 383   gst_rtsp_url_free (url);
 384
 385   stream = G_IO_STREAM (g_socket_connection_factory_create_connection (socket));
 386
 387   /* both read and write initially */
 388   newconn->server = TRUE;
 389   newconn->socket0 = socket;
 390   newconn->stream0 = stream;
 391   newconn->write_socket = newconn->read_socket = newconn->socket0;
 392   newconn->input_stream = g_io_stream_get_input_stream (stream);
 393   newconn->output_stream = g_io_stream_get_output_stream (stream);
 394   newconn->control_stream = NULL;
 395   newconn->remote_ip = g_strdup (ip);
 396   newconn->local_ip = local_ip;
 397   newconn->initial_buffer = g_strdup (initial_buffer);
 398
 399   *conn = newconn;
 400
 401   return GST_RTSP_OK;
 402
 403   /* ERRORS */
 404 getnameinfo_failed:
 405   {
 406     GST_ERROR ("failed to get local address: %s", err->message);
 407     g_clear_error (&err);
 408     return GST_RTSP_ERROR;
 409   }
 410 newconn_failed:
 411   {
 412     GST_ERROR ("failed to make connection");
 413     g_free (local_ip);
 414     gst_rtsp_url_free (url);
 415     return res;
 416   }
 417 }
 418
 419 /**
 420  * gst_rtsp_connection_accept:
 421  * @socket: a socket
 422  * @conn: (out) (transfer full): storage for a #GstRTSPConnection
 423  * @cancellable: a #GCancellable to cancel the operation
 424  *
 425  * Accept a new connection on @socket and create a new #GstRTSPConnection for
 426  * handling communication on new socket.
 427  *
 428  * Returns: #GST_RTSP_OK when @conn contains a valid connection.
 429  */
 430 GstRTSPResult
 431 gst_rtsp_connection_accept (GSocket * socket, GstRTSPConnection ** conn,
 432     GCancellable * cancellable)
 433 {
 434   GError *err = NULL;
 435   gchar *ip;
 436   guint16 port;
 437   GSocket *client_sock;
 438   GstRTSPResult ret;
 439
 440   g_return_val_if_fail (G_IS_SOCKET (socket), GST_RTSP_EINVAL);
 441   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
 442
 443   client_sock = g_socket_accept (socket, cancellable, &err);
 444   if (!client_sock)
 445     goto accept_failed;
 446
 447   /* get the remote ip address and port */
 448   if (!collect_addresses (client_sock, &ip, &port, TRUE, &err))
 449     goto getnameinfo_failed;
 450
 451   ret =
 452       gst_rtsp_connection_create_from_socket (client_sock, ip, port, NULL,
 453       conn);
 454   g_object_unref (client_sock);
 455   g_free (ip);
 456
 457   return ret;
 458
 459   /* ERRORS */
 460 accept_failed:
 461   {
 462     GST_DEBUG ("Accepting client failed: %s", err->message);
 463     g_clear_error (&err);
 464     return GST_RTSP_ESYS;
 465   }
 466 getnameinfo_failed:
 467   {
 468     GST_DEBUG ("getnameinfo failed: %s", err->message);
 469     g_clear_error (&err);
 470     if (!g_socket_close (client_sock, &err)) {
 471       GST_DEBUG ("Closing socket failed: %s", err->message);
 472       g_clear_error (&err);
 473     }
 474     g_object_unref (client_sock);
 475     return GST_RTSP_ERROR;
 476   }
 477 }
 478
 479 /**
 480  * gst_rtsp_connection_get_tls:
 481  * @conn: a #GstRTSPConnection
 482  * @error: #GError for error reporting, or NULL to ignore.
 483  *
 484  * Get the TLS connection of @conn.
 485  *
 486  * For client side this will return the #GTlsClientConnection when connected
 487  * over TLS.
 488  *
 489  * For server side connections, this function will create a GTlsServerConnection
 490  * when called the first time and will return that same connection on subsequent
 491  * calls. The server is then responsible for configuring the TLS connection.
 492  *
 493  * Returns: (transfer none): the TLS connection for @conn.
 494  *
 495  * Since: 1.2
 496  */
 497 GTlsConnection *
 498 gst_rtsp_connection_get_tls (GstRTSPConnection * conn, GError ** error)
 499 {
 500   GTlsConnection *result;
 501
 502   if (G_IS_TLS_CONNECTION (conn->stream0)) {
 503     /* we already had one, return it */
 504     result = G_TLS_CONNECTION (conn->stream0);
 505   } else if (conn->server) {
 506     /* no TLS connection but we are server, make one */
 507     result = (GTlsConnection *)
 508         g_tls_server_connection_new (conn->stream0, NULL, error);
 509     if (result) {
 510       g_object_unref (conn->stream0);
 511       conn->stream0 = G_IO_STREAM (result);
 512       conn->input_stream = g_io_stream_get_input_stream (conn->stream0);
 513       conn->output_stream = g_io_stream_get_output_stream (conn->stream0);
 514     }
 515   } else {
 516     /* client */
 517     result = NULL;
 518     g_set_error (error, GST_LIBRARY_ERROR, GST_LIBRARY_ERROR_FAILED,
 519         "client not connected with TLS");
 520   }
 521   return result;
 522 }
 523
 524 /**
 525  * gst_rtsp_connection_set_tls_validation_flags:
 526  * @conn: a #GstRTSPConnection
 527  * @flags: the validation flags.
 528  *
 529  * Sets the TLS validation flags to be used to verify the peer
 530  * certificate when a TLS connection is established.
 531  *
 532  * Returns: TRUE if the validation flags are set correctly, or FALSE if
 533  * @conn is NULL or is not a TLS connection.
 534  *
 535  * Since: 1.2.1
 536  */
 537 gboolean
 538 gst_rtsp_connection_set_tls_validation_flags (GstRTSPConnection * conn,
 539     GTlsCertificateFlags flags)
 540 {
 541   gboolean res = FALSE;
 542
 543   g_return_val_if_fail (conn != NULL, FALSE);
 544
 545   res = g_socket_client_get_tls (conn->client);
 546   if (res)
 547     g_socket_client_set_tls_validation_flags (conn->client, flags);
 548
 549   return res;
 550 }
 551
 552 /**
 553  * gst_rtsp_connection_get_tls_validation_flags:
 554  * @conn: a #GstRTSPConnection
 555  *
 556  * Gets the TLS validation flags used to verify the peer certificate
 557  * when a TLS connection is established.
 558  *
 559  * Returns: the validationg flags.
 560  *
 561  * Since: 1.2.1
 562  */
 563 GTlsCertificateFlags
 564 gst_rtsp_connection_get_tls_validation_flags (GstRTSPConnection * conn)
 565 {
 566   g_return_val_if_fail (conn != NULL, 0);
 567
 568   return g_socket_client_get_tls_validation_flags (conn->client);
 569 }
 570
 571 /**
 572  * gst_rtsp_connection_set_tls_database:
 573  * @conn: a #GstRTSPConnection
 574  * @database: a #GTlsDatabase
 575  *
 576  * Sets the anchor certificate authorities database. This certificate
 577  * database will be used to verify the server's certificate in case it
 578  * can't be verified with the default certificate database first.
 579  *
 580  * Since: 1.4
 581  */
 582 void
 583 gst_rtsp_connection_set_tls_database (GstRTSPConnection * conn,
 584     GTlsDatabase * database)
 585 {
 586   GTlsDatabase *old_db;
 587
 588   g_return_if_fail (conn != NULL);
 589
 590   if (database)
 591     g_object_ref (database);
 592
 593   old_db = conn->tls_database;
 594   conn->tls_database = database;
 595
 596   if (old_db)
 597     g_object_unref (old_db);
 598 }
 599
 600 /**
 601  * gst_rtsp_connection_get_tls_database:
 602  * @conn: a #GstRTSPConnection
 603  *
 604  * Gets the anchor certificate authorities database that will be used
 605  * after a server certificate can't be verified with the default
 606  * certificate database.
 607  *
 608  * Returns: (transfer full): the anchor certificate authorities database, or NULL if no
 609  * database has been previously set. Use g_object_unref() to release the
 610  * certificate database.
 611  *
 612  * Since: 1.4
 613  */
 614 GTlsDatabase *
 615 gst_rtsp_connection_get_tls_database (GstRTSPConnection * conn)
 616 {
 617   GTlsDatabase *result;
 618
 619   g_return_val_if_fail (conn != NULL, NULL);
 620
 621   if ((result = conn->tls_database))
 622     g_object_ref (result);
 623
 624   return result;
 625 }
 626
 627 static GstRTSPResult
 628 setup_tunneling (GstRTSPConnection * conn, GTimeVal * timeout, gchar * uri)
 629 {
 630   gint i;
 631   GstRTSPResult res;
 632   gchar *value;
 633   guint16 url_port;
 634   GstRTSPMessage *msg;
 635   GstRTSPMessage response;
 636   gboolean old_http;
 637   GstRTSPUrl *url;
 638   GError *error = NULL;
 639   GSocketConnection *connection;
 640   GSocket *socket;
 641
 642   memset (&response, 0, sizeof (response));
 643   gst_rtsp_message_init (&response);
 644
 645   url = conn->url;
 646
 647   /* create a random sessionid */
 648   for (i = 0; i < TUNNELID_LEN; i++)
 649     conn->tunnelid[i] = g_random_int_range ('a', 'z');
 650   conn->tunnelid[TUNNELID_LEN - 1] = '\0';
 651
 652   /* create the GET request for the read connection */
 653   GST_RTSP_CHECK (gst_rtsp_message_new_request (&msg, GST_RTSP_GET, uri),
 654       no_message);
 655   msg->type = GST_RTSP_MESSAGE_HTTP_REQUEST;
 656
 657   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_X_SESSIONCOOKIE,
 658       conn->tunnelid);
 659   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_ACCEPT,
 660       "application/x-rtsp-tunnelled");
 661   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_CACHE_CONTROL, "no-cache");
 662   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_PRAGMA, "no-cache");
 663
 664   /* we need to temporarily set conn->tunneled to FALSE to prevent the HTTP
 665    * request from being base64 encoded */
 666   conn->tunneled = FALSE;
 667   GST_RTSP_CHECK (gst_rtsp_connection_send (conn, msg, timeout), write_failed);
 668   gst_rtsp_message_free (msg);
 669   conn->tunneled = TRUE;
 670
 671   /* receive the response to the GET request */
 672   /* we need to temporarily set manual_http to TRUE since
 673    * gst_rtsp_connection_receive() will treat the HTTP response as a parsing
 674    * failure otherwise */
 675   old_http = conn->manual_http;
 676   conn->manual_http = TRUE;
 677   GST_RTSP_CHECK (gst_rtsp_connection_receive (conn, &response, timeout),
 678       read_failed);
 679   conn->manual_http = old_http;
 680
 681   if (response.type != GST_RTSP_MESSAGE_HTTP_RESPONSE ||
 682       response.type_data.response.code != GST_RTSP_STS_OK)
 683     goto wrong_result;
 684
 685   if (gst_rtsp_message_get_header (&response, GST_RTSP_HDR_X_SERVER_IP_ADDRESS,
 686           &value, 0) == GST_RTSP_OK) {
 687     g_free (url->host);
 688     url->host = g_strdup (value);
 689     g_free (conn->remote_ip);
 690     conn->remote_ip = g_strdup (value);
 691   }
 692
 693   gst_rtsp_url_get_port (url, &url_port);
 694   uri = g_strdup_printf ("http://%s:%d%s%s%s", url->host, url_port,
 695       url->abspath, url->query ? "?" : "", url->query ? url->query : "");
 696
 697   /* connect to the host/port */
 698   if (conn->proxy_host) {
 699     connection = g_socket_client_connect_to_host (conn->client,
 700         conn->proxy_host, conn->proxy_port, conn->cancellable, &error);
 701   } else {
 702     connection = g_socket_client_connect_to_uri (conn->client,
 703         uri, 0, conn->cancellable, &error);
 704   }
 705   if (connection == NULL)
 706     goto connect_failed;
 707
 708   socket = g_socket_connection_get_socket (connection);
 709
 710   /* get remote address */
 711   g_free (conn->remote_ip);
 712   conn->remote_ip = NULL;
 713
 714   if (!collect_addresses (socket, &conn->remote_ip, NULL, TRUE, &error))
 715     goto remote_address_failed;
 716
 717   /* this is now our writing socket */
 718   conn->stream1 = G_IO_STREAM (connection);
 719   conn->socket1 = socket;
 720   conn->write_socket = conn->socket1;
 721   conn->output_stream = g_io_stream_get_output_stream (conn->stream1);
 722   conn->control_stream = NULL;
 723
 724   /* create the POST request for the write connection */
 725   GST_RTSP_CHECK (gst_rtsp_message_new_request (&msg, GST_RTSP_POST, uri),
 726       no_message);
 727   msg->type = GST_RTSP_MESSAGE_HTTP_REQUEST;
 728
 729   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_X_SESSIONCOOKIE,
 730       conn->tunnelid);
 731   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_ACCEPT,
 732       "application/x-rtsp-tunnelled");
 733   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_CACHE_CONTROL, "no-cache");
 734   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_PRAGMA, "no-cache");
 735   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_EXPIRES,
 736       "Sun, 9 Jan 1972 00:00:00 GMT");
 737   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_CONTENT_LENGTH, "32767");
 738
 739   /* we need to temporarily set conn->tunneled to FALSE to prevent the HTTP
 740    * request from being base64 encoded */
 741   conn->tunneled = FALSE;
 742   GST_RTSP_CHECK (gst_rtsp_connection_send (conn, msg, timeout), write_failed);
 743   gst_rtsp_message_free (msg);
 744   conn->tunneled = TRUE;
 745
 746 exit:
 747   gst_rtsp_message_unset (&response);
 748   g_free (uri);
 749
 750   return res;
 751
 752   /* ERRORS */
 753 no_message:
 754   {
 755     GST_ERROR ("failed to create request (%d)", res);
 756     goto exit;
 757   }
 758 write_failed:
 759   {
 760     GST_ERROR ("write failed (%d)", res);
 761     gst_rtsp_message_free (msg);
 762     conn->tunneled = TRUE;
 763     goto exit;
 764   }
 765 read_failed:
 766   {
 767     GST_ERROR ("read failed (%d)", res);
 768     conn->manual_http = FALSE;
 769     goto exit;
 770   }
 771 wrong_result:
 772   {
 773     GST_ERROR ("got failure response %d %s", response.type_data.response.code,
 774         response.type_data.response.reason);
 775     res = GST_RTSP_ERROR;
 776     goto exit;
 777   }
 778 connect_failed:
 779   {
 780     GST_ERROR ("failed to connect: %s", error->message);
 781     res = GST_RTSP_ERROR;
 782     g_clear_error (&error);
 783     goto exit;
 784   }
 785 remote_address_failed:
 786   {
 787     GST_ERROR ("failed to resolve address: %s", error->message);
 788     g_object_unref (connection);
 789     g_clear_error (&error);
 790     return GST_RTSP_ERROR;
 791   }
 792 }
 793
 794 /**
 795  * gst_rtsp_connection_connect:
 796  * @conn: a #GstRTSPConnection
 797  * @timeout: a #GTimeVal timeout
 798  *
 799  * Attempt to connect to the url of @conn made with
 800  * gst_rtsp_connection_create(). If @timeout is #NULL this function can block
 801  * forever. If @timeout contains a valid timeout, this function will return
 802  * #GST_RTSP_ETIMEOUT after the timeout expired.
 803  *
 804  * This function can be cancelled with gst_rtsp_connection_flush().
 805  *
 806  * Returns: #GST_RTSP_OK when a connection could be made.
 807  */
 808 GstRTSPResult
 809 gst_rtsp_connection_connect (GstRTSPConnection * conn, GTimeVal * timeout)
 810 {
 811   GstRTSPResult res;
 812   GSocketConnection *connection;
 813   GSocket *socket;
 814   GError *error = NULL;
 815   gchar *uri, *remote_ip;
 816   GstClockTime to;
 817   guint16 url_port;
 818   GstRTSPUrl *url;
 819
 820   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
 821   g_return_val_if_fail (conn->url != NULL, GST_RTSP_EINVAL);
 822   g_return_val_if_fail (conn->stream0 == NULL, GST_RTSP_EINVAL);
 823
 824   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : 0;
 825   g_socket_client_set_timeout (conn->client,
 826       (to + GST_SECOND - 1) / GST_SECOND);
 827
 828   url = conn->url;
 829
 830   gst_rtsp_url_get_port (url, &url_port);
 831
 832   if (conn->tunneled) {
 833     uri = g_strdup_printf ("http://%s:%d%s%s%s", url->host, url_port,
 834         url->abspath, url->query ? "?" : "", url->query ? url->query : "");
 835   } else {
 836     uri = gst_rtsp_url_get_request_uri (url);
 837   }
 838
 839   if (conn->proxy_host) {
 840     connection = g_socket_client_connect_to_host (conn->client,
 841         conn->proxy_host, conn->proxy_port, conn->cancellable, &error);
 842   } else {
 843     connection = g_socket_client_connect_to_uri (conn->client,
 844         uri, url_port, conn->cancellable, &error);
 845   }
 846   if (connection == NULL)
 847     goto connect_failed;
 848
 849   /* get remote address */
 850   socket = g_socket_connection_get_socket (connection);
 851
 852   if (!collect_addresses (socket, &remote_ip, NULL, TRUE, &error))
 853     goto remote_address_failed;
 854
 855   g_free (conn->remote_ip);
 856   conn->remote_ip = remote_ip;
 857   conn->stream0 = G_IO_STREAM (connection);
 858   conn->socket0 = socket;
 859   /* this is our read socket */
 860   conn->read_socket = conn->socket0;
 861   conn->write_socket = conn->socket0;
 862   conn->input_stream = g_io_stream_get_input_stream (conn->stream0);
 863   conn->output_stream = g_io_stream_get_output_stream (conn->stream0);
 864   conn->control_stream = NULL;
 865
 866   if (conn->tunneled) {
 867     res = setup_tunneling (conn, timeout, uri);
 868     if (res != GST_RTSP_OK)
 869       goto tunneling_failed;
 870   }
 871   g_free (uri);
 872
 873   return GST_RTSP_OK;
 874
 875   /* ERRORS */
 876 connect_failed:
 877   {
 878     GST_ERROR ("failed to connect: %s", error->message);
 879     g_clear_error (&error);
 880     return GST_RTSP_ERROR;
 881   }
 882 remote_address_failed:
 883   {
 884     GST_ERROR ("failed to connect: %s", error->message);
 885     g_object_unref (connection);
 886     g_clear_error (&error);
 887     return GST_RTSP_ERROR;
 888   }
 889 tunneling_failed:
 890   {
 891     GST_ERROR ("failed to setup tunneling");
 892     return res;
 893   }
 894 }
 895
 896 static void
 897 auth_digest_compute_hex_urp (const gchar * username,
 898     const gchar * realm, const gchar * password, gchar hex_urp[33])
 899 {
 900   GChecksum *md5_context = g_checksum_new (G_CHECKSUM_MD5);
 901   const gchar *digest_string;
 902
 903   g_checksum_update (md5_context, (const guchar *) username, strlen (username));
 904   g_checksum_update (md5_context, (const guchar *) ":", 1);
 905   g_checksum_update (md5_context, (const guchar *) realm, strlen (realm));
 906   g_checksum_update (md5_context, (const guchar *) ":", 1);
 907   g_checksum_update (md5_context, (const guchar *) password, strlen (password));
 908   digest_string = g_checksum_get_string (md5_context);
 909
 910   memset (hex_urp, 0, 33);
 911   memcpy (hex_urp, digest_string, strlen (digest_string));
 912
 913   g_checksum_free (md5_context);
 914 }
 915
 916 static void
 917 auth_digest_compute_response (const gchar * method,
 918     const gchar * uri, const gchar * hex_a1, const gchar * nonce,
 919     gchar response[33])
 920 {
 921   char hex_a2[33] = { 0, };
 922   GChecksum *md5_context = g_checksum_new (G_CHECKSUM_MD5);
 923   const gchar *digest_string;
 924
 925   /* compute A2 */
 926   g_checksum_update (md5_context, (const guchar *) method, strlen (method));
 927   g_checksum_update (md5_context, (const guchar *) ":", 1);
 928   g_checksum_update (md5_context, (const guchar *) uri, strlen (uri));
 929   digest_string = g_checksum_get_string (md5_context);
 930   memcpy (hex_a2, digest_string, strlen (digest_string));
 931
 932   /* compute KD */
 933   g_checksum_reset (md5_context);
 934   g_checksum_update (md5_context, (const guchar *) hex_a1, strlen (hex_a1));
 935   g_checksum_update (md5_context, (const guchar *) ":", 1);
 936   g_checksum_update (md5_context, (const guchar *) nonce, strlen (nonce));
 937   g_checksum_update (md5_context, (const guchar *) ":", 1);
 938
 939   g_checksum_update (md5_context, (const guchar *) hex_a2, 32);
 940   digest_string = g_checksum_get_string (md5_context);
 941   memset (response, 0, 33);
 942   memcpy (response, digest_string, strlen (digest_string));
 943
 944   g_checksum_free (md5_context);
 945 }
 946
 947 static void
 948 add_auth_header (GstRTSPConnection * conn, GstRTSPMessage * message)
 949 {
 950   switch (conn->auth_method) {
 951     case GST_RTSP_AUTH_BASIC:{
 952       gchar *user_pass;
 953       gchar *user_pass64;
 954       gchar *auth_string;
 955
 956       if (conn->username == NULL || conn->passwd == NULL)
 957         break;
 958
 959       user_pass = g_strdup_printf ("%s:%s", conn->username, conn->passwd);
 960       user_pass64 = g_base64_encode ((guchar *) user_pass, strlen (user_pass));
 961       auth_string = g_strdup_printf ("Basic %s", user_pass64);
 962
 963       gst_rtsp_message_take_header (message, GST_RTSP_HDR_AUTHORIZATION,
 964           auth_string);
 965
 966       g_free (user_pass);
 967       g_free (user_pass64);
 968       break;
 969     }
 970     case GST_RTSP_AUTH_DIGEST:{
 971       gchar response[33], hex_urp[33];
 972       gchar *auth_string, *auth_string2;
 973       gchar *realm;
 974       gchar *nonce;
 975       gchar *opaque;
 976       const gchar *uri;
 977       const gchar *method;
 978
 979       /* we need to have some params set */
 980       if (conn->auth_params == NULL || conn->username == NULL ||
 981           conn->passwd == NULL)
 982         break;
 983
 984       /* we need the realm and nonce */
 985       realm = (gchar *) g_hash_table_lookup (conn->auth_params, "realm");
 986       nonce = (gchar *) g_hash_table_lookup (conn->auth_params, "nonce");
 987       if (realm == NULL || nonce == NULL)
 988         break;
 989
 990       auth_digest_compute_hex_urp (conn->username, realm, conn->passwd,
 991           hex_urp);
 992
 993       method = gst_rtsp_method_as_text (message->type_data.request.method);
 994       uri = message->type_data.request.uri;
 995
 996       /* Assume no qop, algorithm=md5, stale=false */
 997       /* For algorithm MD5, a1 = urp. */
 998       auth_digest_compute_response (method, uri, hex_urp, nonce, response);
 999       auth_string = g_strdup_printf ("Digest username=\"%s\", "
1000           "realm=\"%s\", nonce=\"%s\", uri=\"%s\", response=\"%s\"",
1001           conn->username, realm, nonce, uri, response);
1002
1003       opaque = (gchar *) g_hash_table_lookup (conn->auth_params, "opaque");
1004       if (opaque) {
1005         auth_string2 = g_strdup_printf ("%s, opaque=\"%s\"", auth_string,
1006             opaque);
1007         g_free (auth_string);
1008         auth_string = auth_string2;
1009       }
1010       gst_rtsp_message_take_header (message, GST_RTSP_HDR_AUTHORIZATION,
1011           auth_string);
1012       break;
1013     }
1014     default:
1015       /* Nothing to do */
1016       break;
1017   }
1018 }
1019
1020 static void
1021 gen_date_string (gchar * date_string, guint len)
1022 {
1023   static const char wkdays[7][4] =
1024       { "Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat" };
1025   static const char months[12][4] =
1026       { "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct",
1027     "Nov", "Dec"
1028   };
1029   struct tm tm;
1030   time_t t;
1031
1032   time (&t);
1033
1034 #ifdef HAVE_GMTIME_R
1035   gmtime_r (&t, &tm);
1036 #else
1037   tm = *gmtime (&t);
1038 #endif
1039
1040   g_snprintf (date_string, len, "%s, %02d %s %04d %02d:%02d:%02d GMT",
1041       wkdays[tm.tm_wday], tm.tm_mday, months[tm.tm_mon], tm.tm_year + 1900,
1042       tm.tm_hour, tm.tm_min, tm.tm_sec);
1043 }
1044
1045 static GstRTSPResult
1046 write_bytes (GOutputStream * stream, const guint8 * buffer, guint * idx,
1047     guint size, gboolean block, GCancellable * cancellable)
1048 {
1049   guint left;
1050   gssize r;
1051   GError *err = NULL;
1052
1053   if (G_UNLIKELY (*idx > size))
1054     return GST_RTSP_ERROR;
1055
1056   left = size - *idx;
1057
1058   while (left) {
1059     if (block)
1060       r = g_output_stream_write (stream, (gchar *) & buffer[*idx], left,
1061           cancellable, &err);
1062     else
1063       r = g_pollable_output_stream_write_nonblocking (G_POLLABLE_OUTPUT_STREAM
1064           (stream), (gchar *) & buffer[*idx], left, cancellable, &err);
1065     if (G_UNLIKELY (r < 0))
1066       goto error;
1067
1068     left -= r;
1069     *idx += r;
1070   }
1071   return GST_RTSP_OK;
1072
1073   /* ERRORS */
1074 error:
1075   {
1076     if (G_UNLIKELY (r == 0))
1077       return GST_RTSP_EEOF;
1078
1079     GST_DEBUG ("%s", err->message);
1080     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
1081       g_clear_error (&err);
1082       return GST_RTSP_EINTR;
1083     } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
1084       g_clear_error (&err);
1085       return GST_RTSP_EINTR;
1086     } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
1087       g_clear_error (&err);
1088       return GST_RTSP_ETIMEOUT;
1089     }
1090     g_clear_error (&err);
1091     return GST_RTSP_ESYS;
1092   }
1093 }
1094
1095 static gint
1096 fill_raw_bytes (GstRTSPConnection * conn, guint8 * buffer, guint size,
1097     gboolean block, GError ** err)
1098 {
1099   gint out = 0;
1100
1101   if (G_UNLIKELY (conn->initial_buffer != NULL)) {
1102     gsize left = strlen (&conn->initial_buffer[conn->initial_buffer_offset]);
1103
1104     out = MIN (left, size);
1105     memcpy (buffer, &conn->initial_buffer[conn->initial_buffer_offset], out);
1106
1107     if (left == (gsize) out) {
1108       g_free (conn->initial_buffer);
1109       conn->initial_buffer = NULL;
1110       conn->initial_buffer_offset = 0;
1111     } else
1112       conn->initial_buffer_offset += out;
1113   }
1114
1115   if (G_LIKELY (size > (guint) out)) {
1116     gssize r;
1117     gsize count = size - out;
1118     if (block)
1119       r = g_input_stream_read (conn->input_stream, (gchar *) & buffer[out],
1120           count, conn->may_cancel ? conn->cancellable : NULL, err);
1121     else
1122       r = g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM
1123           (conn->input_stream), (gchar *) & buffer[out], count,
1124           conn->may_cancel ? conn->cancellable : NULL, err);
1125
1126     if (G_UNLIKELY (r < 0)) {
1127       if (out == 0) {
1128         /* propagate the error */
1129         out = r;
1130       } else {
1131         /* we have some data ignore error */
1132         g_clear_error (err);
1133       }
1134     } else
1135       out += r;
1136   }
1137
1138   return out;
1139 }
1140
1141 static gint
1142 fill_bytes (GstRTSPConnection * conn, guint8 * buffer, guint size,
1143     gboolean block, GError ** err)
1144 {
1145   DecodeCtx *ctx = conn->ctxp;
1146   gint out = 0;
1147
1148   if (ctx) {
1149     while (size > 0) {
1150       guint8 in[sizeof (ctx->out) * 4 / 3];
1151       gint r;
1152
1153       while (size > 0 && ctx->cout < ctx->coutl) {
1154         /* we have some leftover bytes */
1155         *buffer++ = ctx->out[ctx->cout++];
1156         size--;
1157         out++;
1158       }
1159
1160       /* got what we needed? */
1161       if (size == 0)
1162         break;
1163
1164       /* try to read more bytes */
1165       r = fill_raw_bytes (conn, in, sizeof (in), block, err);
1166       if (r <= 0) {
1167         if (out == 0)
1168           out = r;
1169         break;
1170       }
1171
1172       ctx->cout = 0;
1173       ctx->coutl =
1174           g_base64_decode_step ((gchar *) in, r, ctx->out, &ctx->state,
1175           &ctx->save);
1176     }
1177   } else {
1178     out = fill_raw_bytes (conn, buffer, size, block, err);
1179   }
1180
1181   return out;
1182 }
1183
1184 static GstRTSPResult
1185 read_bytes (GstRTSPConnection * conn, guint8 * buffer, guint * idx, guint size,
1186     gboolean block)
1187 {
1188   guint left;
1189   gint r;
1190   GError *err = NULL;
1191
1192   if (G_UNLIKELY (*idx > size))
1193     return GST_RTSP_ERROR;
1194
1195   left = size - *idx;
1196
1197   while (left) {
1198     r = fill_bytes (conn, &buffer[*idx], left, block, &err);
1199     if (G_UNLIKELY (r <= 0))
1200       goto error;
1201
1202     left -= r;
1203     *idx += r;
1204   }
1205   return GST_RTSP_OK;
1206
1207   /* ERRORS */
1208 error:
1209   {
1210     if (G_UNLIKELY (r == 0))
1211       return GST_RTSP_EEOF;
1212
1213     GST_DEBUG ("%s", err->message);
1214     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
1215       g_clear_error (&err);
1216       return GST_RTSP_EINTR;
1217     } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
1218       g_clear_error (&err);
1219       return GST_RTSP_EINTR;
1220     } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
1221       g_clear_error (&err);
1222       return GST_RTSP_ETIMEOUT;
1223     }
1224     g_clear_error (&err);
1225     return GST_RTSP_ESYS;
1226   }
1227 }
1228
1229 /* The code below tries to handle clients using \r, \n or \r\n to indicate the
1230  * end of a line. It even does its best to handle clients which mix them (even
1231  * though this is a really stupid idea (tm).) It also handles Line White Space
1232  * (LWS), where a line end followed by whitespace is considered LWS. This is
1233  * the method used in RTSP (and HTTP) to break long lines.
1234  */
1235 static GstRTSPResult
1236 read_line (GstRTSPConnection * conn, guint8 * buffer, guint * idx, guint size,
1237     gboolean block)
1238 {
1239   GstRTSPResult res;
1240
1241   while (TRUE) {
1242     guint8 c;
1243     guint i;
1244
1245     if (conn->read_ahead == READ_AHEAD_EOH) {
1246       /* the last call to read_line() already determined that we have reached
1247        * the end of the headers, so convey that information now */
1248       conn->read_ahead = 0;
1249       break;
1250     } else if (conn->read_ahead == READ_AHEAD_CRLF) {
1251       /* the last call to read_line() left off after having read \r\n */
1252       c = '\n';
1253     } else if (conn->read_ahead == READ_AHEAD_CRLFCR) {
1254       /* the last call to read_line() left off after having read \r\n\r */
1255       c = '\r';
1256     } else if (conn->read_ahead != 0) {
1257       /* the last call to read_line() left us with a character to start with */
1258       c = (guint8) conn->read_ahead;
1259       conn->read_ahead = 0;
1260     } else {
1261       /* read the next character */
1262       i = 0;
1263       res = read_bytes (conn, &c, &i, 1, block);
1264       if (G_UNLIKELY (res != GST_RTSP_OK))
1265         return res;
1266     }
1267
1268     /* special treatment of line endings */
1269     if (c == '\r' || c == '\n') {
1270       guint8 read_ahead;
1271
1272     retry:
1273       /* need to read ahead one more character to know what to do... */
1274       i = 0;
1275       res = read_bytes (conn, &read_ahead, &i, 1, block);
1276       if (G_UNLIKELY (res != GST_RTSP_OK))
1277         return res;
1278
1279       if (read_ahead == ' ' || read_ahead == '\t') {
1280         if (conn->read_ahead == READ_AHEAD_CRLFCR) {
1281           /* got \r\n\r followed by whitespace, treat it as a normal line
1282            * followed by one starting with LWS */
1283           conn->read_ahead = read_ahead;
1284           break;
1285         } else {
1286           /* got LWS, change the line ending to a space and continue */
1287           c = ' ';
1288           conn->read_ahead = read_ahead;
1289         }
1290       } else if (conn->read_ahead == READ_AHEAD_CRLFCR) {
1291         if (read_ahead == '\r' || read_ahead == '\n') {
1292           /* got \r\n\r\r or \r\n\r\n, treat it as the end of the headers */
1293           conn->read_ahead = READ_AHEAD_EOH;
1294           break;
1295         } else {
1296           /* got \r\n\r followed by something else, this is not really
1297            * supported since we have probably just eaten the first character
1298            * of the body or the next message, so just ignore the second \r
1299            * and live with it... */
1300           conn->read_ahead = read_ahead;
1301           break;
1302         }
1303       } else if (conn->read_ahead == READ_AHEAD_CRLF) {
1304         if (read_ahead == '\r') {
1305           /* got \r\n\r so far, need one more character... */
1306           conn->read_ahead = READ_AHEAD_CRLFCR;
1307           goto retry;
1308         } else if (read_ahead == '\n') {
1309           /* got \r\n\n, treat it as the end of the headers */
1310           conn->read_ahead = READ_AHEAD_EOH;
1311           break;
1312         } else {
1313           /* found the end of a line, keep read_ahead for the next line */
1314           conn->read_ahead = read_ahead;
1315           break;
1316         }
1317       } else if (c == read_ahead) {
1318         /* got double \r or \n, treat it as the end of the headers */
1319         conn->read_ahead = READ_AHEAD_EOH;
1320         break;
1321       } else if (c == '\r' && read_ahead == '\n') {
1322         /* got \r\n so far, still need more to know what to do... */
1323         conn->read_ahead = READ_AHEAD_CRLF;
1324         goto retry;
1325       } else {
1326         /* found the end of a line, keep read_ahead for the next line */
1327         conn->read_ahead = read_ahead;
1328         break;
1329       }
1330     }
1331
1332     if (G_LIKELY (*idx < size - 1))
1333       buffer[(*idx)++] = c;
1334   }
1335   buffer[*idx] = '\0';
1336
1337   return GST_RTSP_OK;
1338 }
1339
1340 /**
1341  * gst_rtsp_connection_write:
1342  * @conn: a #GstRTSPConnection
1343  * @data: the data to write
1344  * @size: the size of @data
1345  * @timeout: a timeout value or #NULL
1346  *
1347  * Attempt to write @size bytes of @data to the connected @conn, blocking up to
1348  * the specified @timeout. @timeout can be #NULL, in which case this function
1349  * might block forever.
1350  *
1351  * This function can be cancelled with gst_rtsp_connection_flush().
1352  *
1353  * Returns: #GST_RTSP_OK on success.
1354  */
1355 GstRTSPResult
1356 gst_rtsp_connection_write (GstRTSPConnection * conn, const guint8 * data,
1357     guint size, GTimeVal * timeout)
1358 {
1359   guint offset;
1360   GstClockTime to;
1361   GstRTSPResult res;
1362
1363   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
1364   g_return_val_if_fail (data != NULL || size == 0, GST_RTSP_EINVAL);
1365   g_return_val_if_fail (conn->output_stream != NULL, GST_RTSP_EINVAL);
1366
1367   offset = 0;
1368
1369   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : 0;
1370
1371   g_socket_set_timeout (conn->write_socket, (to + GST_SECOND - 1) / GST_SECOND);
1372   res =
1373       write_bytes (conn->output_stream, data, &offset, size, TRUE,
1374       conn->cancellable);
1375   g_socket_set_timeout (conn->write_socket, 0);
1376
1377   return res;
1378 }
1379
1380 static GString *
1381 message_to_string (GstRTSPConnection * conn, GstRTSPMessage * message)
1382 {
1383   GString *str = NULL;
1384
1385   str = g_string_new ("");
1386
1387   switch (message->type) {
1388     case GST_RTSP_MESSAGE_REQUEST:
1389       /* create request string, add CSeq */
1390       g_string_append_printf (str, "%s %s RTSP/1.0\r\n"
1391           "CSeq: %d\r\n",
1392           gst_rtsp_method_as_text (message->type_data.request.method),
1393           message->type_data.request.uri, conn->cseq++);
1394       /* add session id if we have one */
1395       if (conn->session_id[0] != '\0') {
1396         gst_rtsp_message_remove_header (message, GST_RTSP_HDR_SESSION, -1);
1397         gst_rtsp_message_add_header (message, GST_RTSP_HDR_SESSION,
1398             conn->session_id);
1399       }
1400       /* add any authentication headers */
1401       add_auth_header (conn, message);
1402       break;
1403     case GST_RTSP_MESSAGE_RESPONSE:
1404       /* create response string */
1405       g_string_append_printf (str, "RTSP/1.0 %d %s\r\n",
1406           message->type_data.response.code, message->type_data.response.reason);
1407       break;
1408     case GST_RTSP_MESSAGE_HTTP_REQUEST:
1409       /* create request string */
1410       g_string_append_printf (str, "%s %s HTTP/%s\r\n",
1411           gst_rtsp_method_as_text (message->type_data.request.method),
1412           message->type_data.request.uri,
1413           gst_rtsp_version_as_text (message->type_data.request.version));
1414       /* add any authentication headers */
1415       add_auth_header (conn, message);
1416       break;
1417     case GST_RTSP_MESSAGE_HTTP_RESPONSE:
1418       /* create response string */
1419       g_string_append_printf (str, "HTTP/%s %d %s\r\n",
1420           gst_rtsp_version_as_text (message->type_data.request.version),
1421           message->type_data.response.code, message->type_data.response.reason);
1422       break;
1423     case GST_RTSP_MESSAGE_DATA:
1424     {
1425       guint8 data_header[4];
1426
1427       /* prepare data header */
1428       data_header[0] = '$';
1429       data_header[1] = message->type_data.data.channel;
1430       data_header[2] = (message->body_size >> 8) & 0xff;
1431       data_header[3] = message->body_size & 0xff;
1432
1433       /* create string with header and data */
1434       str = g_string_append_len (str, (gchar *) data_header, 4);
1435       str =
1436           g_string_append_len (str, (gchar *) message->body,
1437           message->body_size);
1438       break;
1439     }
1440     default:
1441       g_string_free (str, TRUE);
1442       g_return_val_if_reached (NULL);
1443       break;
1444   }
1445
1446   /* append headers and body */
1447   if (message->type != GST_RTSP_MESSAGE_DATA) {
1448     gchar date_string[100];
1449
1450     gen_date_string (date_string, sizeof (date_string));
1451
1452     /* add date header */
1453     gst_rtsp_message_remove_header (message, GST_RTSP_HDR_DATE, -1);
1454     gst_rtsp_message_add_header (message, GST_RTSP_HDR_DATE, date_string);
1455
1456     /* append headers */
1457     gst_rtsp_message_append_headers (message, str);
1458
1459     /* append Content-Length and body if needed */
1460     if (message->body != NULL && message->body_size > 0) {
1461       gchar *len;
1462
1463       len = g_strdup_printf ("%d", message->body_size);
1464       g_string_append_printf (str, "%s: %s\r\n",
1465           gst_rtsp_header_as_text (GST_RTSP_HDR_CONTENT_LENGTH), len);
1466       g_free (len);
1467       /* header ends here */
1468       g_string_append (str, "\r\n");
1469       str =
1470           g_string_append_len (str, (gchar *) message->body,
1471           message->body_size);
1472     } else {
1473       /* just end headers */
1474       g_string_append (str, "\r\n");
1475     }
1476   }
1477
1478   return str;
1479 }
1480
1481 /**
1482  * gst_rtsp_connection_send:
1483  * @conn: a #GstRTSPConnection
1484  * @message: the message to send
1485  * @timeout: a timeout value or #NULL
1486  *
1487  * Attempt to send @message to the connected @conn, blocking up to
1488  * the specified @timeout. @timeout can be #NULL, in which case this function
1489  * might block forever.
1490  *
1491  * This function can be cancelled with gst_rtsp_connection_flush().
1492  *
1493  * Returns: #GST_RTSP_OK on success.
1494  */
1495 GstRTSPResult
1496 gst_rtsp_connection_send (GstRTSPConnection * conn, GstRTSPMessage * message,
1497     GTimeVal * timeout)
1498 {
1499   GString *string = NULL;
1500   GstRTSPResult res;
1501   gchar *str;
1502   gsize len;
1503
1504   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
1505   g_return_val_if_fail (message != NULL, GST_RTSP_EINVAL);
1506
1507   if (G_UNLIKELY (!(string = message_to_string (conn, message))))
1508     goto no_message;
1509
1510   if (conn->tunneled) {
1511     str = g_base64_encode ((const guchar *) string->str, string->len);
1512     g_string_free (string, TRUE);
1513     len = strlen (str);
1514   } else {
1515     str = string->str;
1516     len = string->len;
1517     g_string_free (string, FALSE);
1518   }
1519
1520   /* write request */
1521   res = gst_rtsp_connection_write (conn, (guint8 *) str, len, timeout);
1522
1523   g_free (str);
1524
1525   return res;
1526
1527 no_message:
1528   {
1529     g_warning ("Wrong message");
1530     return GST_RTSP_EINVAL;
1531   }
1532 }
1533
1534 static GstRTSPResult
1535 parse_string (gchar * dest, gint size, gchar ** src)
1536 {
1537   GstRTSPResult res = GST_RTSP_OK;
1538   gint idx;
1539
1540   idx = 0;
1541   /* skip spaces */
1542   while (g_ascii_isspace (**src))
1543     (*src)++;
1544
1545   while (!g_ascii_isspace (**src) && **src != '\0') {
1546     if (idx < size - 1)
1547       dest[idx++] = **src;
1548     else
1549       res = GST_RTSP_EPARSE;
1550     (*src)++;
1551   }
1552   if (size > 0)
1553     dest[idx] = '\0';
1554
1555   return res;
1556 }
1557
1558 static GstRTSPResult
1559 parse_protocol_version (gchar * protocol, GstRTSPMsgType * type,
1560     GstRTSPVersion * version)
1561 {
1562   GstRTSPResult res = GST_RTSP_OK;
1563   gchar *ver;
1564
1565   if (G_LIKELY ((ver = strchr (protocol, '/')) != NULL)) {
1566     guint major;
1567     guint minor;
1568     gchar dummychar;
1569
1570     *ver++ = '\0';
1571
1572     /* the version number must be formatted as X.Y with nothing following */
1573     if (sscanf (ver, "%u.%u%c", &major, &minor, &dummychar) != 2)
1574       res = GST_RTSP_EPARSE;
1575
1576     if (g_ascii_strcasecmp (protocol, "RTSP") == 0) {
1577       if (major != 1 || minor != 0) {
1578         *version = GST_RTSP_VERSION_INVALID;
1579         res = GST_RTSP_ERROR;
1580       }
1581     } else if (g_ascii_strcasecmp (protocol, "HTTP") == 0) {
1582       if (*type == GST_RTSP_MESSAGE_REQUEST)
1583         *type = GST_RTSP_MESSAGE_HTTP_REQUEST;
1584       else if (*type == GST_RTSP_MESSAGE_RESPONSE)
1585         *type = GST_RTSP_MESSAGE_HTTP_RESPONSE;
1586
1587       if (major == 1 && minor == 1) {
1588         *version = GST_RTSP_VERSION_1_1;
1589       } else if (major != 1 || minor != 0) {
1590         *version = GST_RTSP_VERSION_INVALID;
1591         res = GST_RTSP_ERROR;
1592       }
1593     } else
1594       res = GST_RTSP_EPARSE;
1595   } else
1596     res = GST_RTSP_EPARSE;
1597
1598   return res;
1599 }
1600
1601 static GstRTSPResult
1602 parse_response_status (guint8 * buffer, GstRTSPMessage * msg)
1603 {
1604   GstRTSPResult res = GST_RTSP_OK;
1605   GstRTSPResult res2;
1606   gchar versionstr[20];
1607   gchar codestr[4];
1608   gint code;
1609   gchar *bptr;
1610
1611   bptr = (gchar *) buffer;
1612
1613   if (parse_string (versionstr, sizeof (versionstr), &bptr) != GST_RTSP_OK)
1614     res = GST_RTSP_EPARSE;
1615
1616   if (parse_string (codestr, sizeof (codestr), &bptr) != GST_RTSP_OK)
1617     res = GST_RTSP_EPARSE;
1618   code = atoi (codestr);
1619   if (G_UNLIKELY (*codestr == '\0' || code < 0 || code >= 600))
1620     res = GST_RTSP_EPARSE;
1621
1622   while (g_ascii_isspace (*bptr))
1623     bptr++;
1624
1625   if (G_UNLIKELY (gst_rtsp_message_init_response (msg, code, bptr,
1626               NULL) != GST_RTSP_OK))
1627     res = GST_RTSP_EPARSE;
1628
1629   res2 = parse_protocol_version (versionstr, &msg->type,
1630       &msg->type_data.response.version);
1631   if (G_LIKELY (res == GST_RTSP_OK))
1632     res = res2;
1633
1634   return res;
1635 }
1636
1637 static GstRTSPResult
1638 parse_request_line (guint8 * buffer, GstRTSPMessage * msg)
1639 {
1640   GstRTSPResult res = GST_RTSP_OK;
1641   GstRTSPResult res2;
1642   gchar versionstr[20];
1643   gchar methodstr[20];
1644   gchar urlstr[4096];
1645   gchar *bptr;
1646   GstRTSPMethod method;
1647
1648   bptr = (gchar *) buffer;
1649
1650   if (parse_string (methodstr, sizeof (methodstr), &bptr) != GST_RTSP_OK)
1651     res = GST_RTSP_EPARSE;
1652   method = gst_rtsp_find_method (methodstr);
1653
1654   if (parse_string (urlstr, sizeof (urlstr), &bptr) != GST_RTSP_OK)
1655     res = GST_RTSP_EPARSE;
1656   if (G_UNLIKELY (*urlstr == '\0'))
1657     res = GST_RTSP_EPARSE;
1658
1659   if (parse_string (versionstr, sizeof (versionstr), &bptr) != GST_RTSP_OK)
1660     res = GST_RTSP_EPARSE;
1661
1662   if (G_UNLIKELY (*bptr != '\0'))
1663     res = GST_RTSP_EPARSE;
1664
1665   if (G_UNLIKELY (gst_rtsp_message_init_request (msg, method,
1666               urlstr) != GST_RTSP_OK))
1667     res = GST_RTSP_EPARSE;
1668
1669   res2 = parse_protocol_version (versionstr, &msg->type,
1670       &msg->type_data.request.version);
1671   if (G_LIKELY (res == GST_RTSP_OK))
1672     res = res2;
1673
1674   if (G_LIKELY (msg->type == GST_RTSP_MESSAGE_REQUEST)) {
1675     /* GET and POST are not allowed as RTSP methods */
1676     if (msg->type_data.request.method == GST_RTSP_GET ||
1677         msg->type_data.request.method == GST_RTSP_POST) {
1678       msg->type_data.request.method = GST_RTSP_INVALID;
1679       if (res == GST_RTSP_OK)
1680         res = GST_RTSP_ERROR;
1681     }
1682   } else if (msg->type == GST_RTSP_MESSAGE_HTTP_REQUEST) {
1683     /* only GET and POST are allowed as HTTP methods */
1684     if (msg->type_data.request.method != GST_RTSP_GET &&
1685         msg->type_data.request.method != GST_RTSP_POST) {
1686       msg->type_data.request.method = GST_RTSP_INVALID;
1687       if (res == GST_RTSP_OK)
1688         res = GST_RTSP_ERROR;
1689     }
1690   }
1691
1692   return res;
1693 }
1694
1695 /* parsing lines means reading a Key: Value pair */
1696 static GstRTSPResult
1697 parse_line (guint8 * buffer, GstRTSPMessage * msg)
1698 {
1699   GstRTSPHeaderField field;
1700   gchar *line = (gchar *) buffer;
1701   gchar *value;
1702
1703   if ((value = strchr (line, ':')) == NULL || value == line)
1704     goto parse_error;
1705
1706   /* trim space before the colon */
1707   if (value[-1] == ' ')
1708     value[-1] = '\0';
1709
1710   /* replace the colon with a NUL */
1711   *value++ = '\0';
1712
1713   /* find the header */
1714   field = gst_rtsp_find_header_field (line);
1715   if (field == GST_RTSP_HDR_INVALID)
1716     goto done;
1717
1718   /* split up the value in multiple key:value pairs if it contains comma(s) */
1719   while (*value != '\0') {
1720     gchar *next_value;
1721     gchar *comma = NULL;
1722     gboolean quoted = FALSE;
1723     guint comment = 0;
1724
1725     /* trim leading space */
1726     if (*value == ' ')
1727       value++;
1728
1729     /* for headers which may not appear multiple times, and thus may not
1730      * contain multiple values on the same line, we can short-circuit the loop
1731      * below and the entire value results in just one key:value pair*/
1732     if (!gst_rtsp_header_allow_multiple (field))
1733       next_value = value + strlen (value);
1734     else
1735       next_value = value;
1736
1737     /* find the next value, taking special care of quotes and comments */
1738     while (*next_value != '\0') {
1739       if ((quoted || comment != 0) && *next_value == '\\' &&
1740           next_value[1] != '\0')
1741         next_value++;
1742       else if (comment == 0 && *next_value == '"')
1743         quoted = !quoted;
1744       else if (!quoted && *next_value == '(')
1745         comment++;
1746       else if (comment != 0 && *next_value == ')')
1747         comment--;
1748       else if (!quoted && comment == 0) {
1749         /* To quote RFC 2068: "User agents MUST take special care in parsing
1750          * the WWW-Authenticate field value if it contains more than one
1751          * challenge, or if more than one WWW-Authenticate header field is
1752          * provided, since the contents of a challenge may itself contain a
1753          * comma-separated list of authentication parameters."
1754          *
1755          * What this means is that we cannot just look for an unquoted comma
1756          * when looking for multiple values in Proxy-Authenticate and
1757          * WWW-Authenticate headers. Instead we need to look for the sequence
1758          * "comma [space] token space token" before we can split after the
1759          * comma...
1760          */
1761         if (field == GST_RTSP_HDR_PROXY_AUTHENTICATE ||
1762             field == GST_RTSP_HDR_WWW_AUTHENTICATE) {
1763           if (*next_value == ',') {
1764             if (next_value[1] == ' ') {
1765               /* skip any space following the comma so we do not mistake it for
1766                * separating between two tokens */
1767               next_value++;
1768             }
1769             comma = next_value;
1770           } else if (*next_value == ' ' && next_value[1] != ',' &&
1771               next_value[1] != '=' && comma != NULL) {
1772             next_value = comma;
1773             comma = NULL;
1774             break;
1775           }
1776         } else if (*next_value == ',')
1777           break;
1778       }
1779
1780       next_value++;
1781     }
1782
1783     /* trim space */
1784     if (value != next_value && next_value[-1] == ' ')
1785       next_value[-1] = '\0';
1786
1787     if (*next_value != '\0')
1788       *next_value++ = '\0';
1789
1790     /* add the key:value pair */
1791     if (*value != '\0')
1792       gst_rtsp_message_add_header (msg, field, value);
1793
1794     value = next_value;
1795   }
1796
1797 done:
1798   return GST_RTSP_OK;
1799
1800   /* ERRORS */
1801 parse_error:
1802   {
1803     return GST_RTSP_EPARSE;
1804   }
1805 }
1806
1807 /* convert all consecutive whitespace to a single space */
1808 static void
1809 normalize_line (guint8 * buffer)
1810 {
1811   while (*buffer) {
1812     if (g_ascii_isspace (*buffer)) {
1813       guint8 *tmp;
1814
1815       *buffer++ = ' ';
1816       for (tmp = buffer; g_ascii_isspace (*tmp); tmp++) {
1817       }
1818       if (buffer != tmp)
1819         memmove (buffer, tmp, strlen ((gchar *) tmp) + 1);
1820     } else {
1821       buffer++;
1822     }
1823   }
1824 }
1825
1826 /* returns:
1827  *  GST_RTSP_OK when a complete message was read.
1828  *  GST_RTSP_EEOF: when the read socket is closed
1829  *  GST_RTSP_EINTR: when more data is needed.
1830  *  GST_RTSP_..: some other error occured.
1831  */
1832 static GstRTSPResult
1833 build_next (GstRTSPBuilder * builder, GstRTSPMessage * message,
1834     GstRTSPConnection * conn, gboolean block)
1835 {
1836   GstRTSPResult res;
1837
1838   while (TRUE) {
1839     switch (builder->state) {
1840       case STATE_START:
1841       {
1842         guint8 c;
1843
1844         builder->offset = 0;
1845         res =
1846             read_bytes (conn, (guint8 *) builder->buffer, &builder->offset, 1,
1847             block);
1848         if (res != GST_RTSP_OK)
1849           goto done;
1850
1851         c = builder->buffer[0];
1852
1853         /* we have 1 bytes now and we can see if this is a data message or
1854          * not */
1855         if (c == '$') {
1856           /* data message, prepare for the header */
1857           builder->state = STATE_DATA_HEADER;
1858           conn->may_cancel = FALSE;
1859         } else if (c == '\n' || c == '\r') {
1860           /* skip \n and \r */
1861           builder->offset = 0;
1862         } else {
1863           builder->line = 0;
1864           builder->state = STATE_READ_LINES;
1865           conn->may_cancel = FALSE;
1866         }
1867         break;
1868       }
1869       case STATE_DATA_HEADER:
1870       {
1871         res =
1872             read_bytes (conn, (guint8 *) builder->buffer, &builder->offset, 4,
1873             block);
1874         if (res != GST_RTSP_OK)
1875           goto done;
1876
1877         gst_rtsp_message_init_data (message, builder->buffer[1]);
1878
1879         builder->body_len = (builder->buffer[2] << 8) | builder->buffer[3];
1880         builder->body_data = g_malloc (builder->body_len + 1);
1881         builder->body_data[builder->body_len] = '\0';
1882         builder->offset = 0;
1883         builder->state = STATE_DATA_BODY;
1884         break;
1885       }
1886       case STATE_DATA_BODY:
1887       {
1888         res =
1889             read_bytes (conn, builder->body_data, &builder->offset,
1890             builder->body_len, block);
1891         if (res != GST_RTSP_OK)
1892           goto done;
1893
1894         /* we have the complete body now, store in the message adjusting the
1895          * length to include the trailing '\0' */
1896         gst_rtsp_message_take_body (message,
1897             (guint8 *) builder->body_data, builder->body_len + 1);
1898         builder->body_data = NULL;
1899         builder->body_len = 0;
1900
1901         builder->state = STATE_END;
1902         break;
1903       }
1904       case STATE_READ_LINES:
1905       {
1906         res = read_line (conn, builder->buffer, &builder->offset,
1907             sizeof (builder->buffer), block);
1908         if (res != GST_RTSP_OK)
1909           goto done;
1910
1911         /* we have a regular response */
1912         if (builder->buffer[0] == '\0') {
1913           gchar *hdrval;
1914
1915           /* empty line, end of message header */
1916           /* see if there is a Content-Length header, but ignore it if this
1917            * is a POST request with an x-sessioncookie header */
1918           if (gst_rtsp_message_get_header (message,
1919                   GST_RTSP_HDR_CONTENT_LENGTH, &hdrval, 0) == GST_RTSP_OK &&
1920               (message->type != GST_RTSP_MESSAGE_HTTP_REQUEST ||
1921                   message->type_data.request.method != GST_RTSP_POST ||
1922                   gst_rtsp_message_get_header (message,
1923                       GST_RTSP_HDR_X_SESSIONCOOKIE, NULL, 0) != GST_RTSP_OK)) {
1924             /* there is, prepare to read the body */
1925             builder->body_len = atol (hdrval);
1926             builder->body_data = g_try_malloc (builder->body_len + 1);
1927             /* we can't do much here, we need the length to know how many bytes
1928              * we need to read next and when allocation fails, something is
1929              * probably wrong with the length. */
1930             if (builder->body_data == NULL)
1931               goto invalid_body_len;
1932
1933             builder->body_data[builder->body_len] = '\0';
1934             builder->offset = 0;
1935             builder->state = STATE_DATA_BODY;
1936           } else {
1937             builder->state = STATE_END;
1938           }
1939           break;
1940         }
1941
1942         /* we have a line */
1943         normalize_line (builder->buffer);
1944         if (builder->line == 0) {
1945           /* first line, check for response status */
1946           if (memcmp (builder->buffer, "RTSP", 4) == 0 ||
1947               memcmp (builder->buffer, "HTTP", 4) == 0) {
1948             builder->status = parse_response_status (builder->buffer, message);
1949           } else {
1950             builder->status = parse_request_line (builder->buffer, message);
1951           }
1952         } else {
1953           /* else just parse the line */
1954           res = parse_line (builder->buffer, message);
1955           if (res != GST_RTSP_OK)
1956             builder->status = res;
1957         }
1958         builder->line++;
1959         builder->offset = 0;
1960         break;
1961       }
1962       case STATE_END:
1963       {
1964         gchar *session_cookie;
1965         gchar *session_id;
1966
1967         conn->may_cancel = TRUE;
1968
1969         if (message->type == GST_RTSP_MESSAGE_DATA) {
1970           /* data messages don't have headers */
1971           res = GST_RTSP_OK;
1972           goto done;
1973         }
1974
1975         /* save the tunnel session in the connection */
1976         if (message->type == GST_RTSP_MESSAGE_HTTP_REQUEST &&
1977             !conn->manual_http &&
1978             conn->tstate == TUNNEL_STATE_NONE &&
1979             gst_rtsp_message_get_header (message, GST_RTSP_HDR_X_SESSIONCOOKIE,
1980                 &session_cookie, 0) == GST_RTSP_OK) {
1981           strncpy (conn->tunnelid, session_cookie, TUNNELID_LEN);
1982           conn->tunnelid[TUNNELID_LEN - 1] = '\0';
1983           conn->tunneled = TRUE;
1984         }
1985
1986         /* save session id in the connection for further use */
1987         if (message->type == GST_RTSP_MESSAGE_RESPONSE &&
1988             gst_rtsp_message_get_header (message, GST_RTSP_HDR_SESSION,
1989                 &session_id, 0) == GST_RTSP_OK) {
1990           gint maxlen, i;
1991
1992           maxlen = sizeof (conn->session_id) - 1;
1993           /* the sessionid can have attributes marked with ;
1994            * Make sure we strip them */
1995           for (i = 0; session_id[i] != '\0'; i++) {
1996             if (session_id[i] == ';') {
1997               maxlen = i;
1998               /* parse timeout */
1999               do {
2000                 i++;
2001               } while (g_ascii_isspace (session_id[i]));
2002               if (g_str_has_prefix (&session_id[i], "timeout=")) {
2003                 gint to;
2004
2005                 /* if we parsed something valid, configure */
2006                 if ((to = atoi (&session_id[i + 8])) > 0)
2007                   conn->timeout = to;
2008               }
2009               break;
2010             }
2011           }
2012
2013           /* make sure to not overflow */
2014           if (conn->remember_session_id) {
2015             strncpy (conn->session_id, session_id, maxlen);
2016             conn->session_id[maxlen] = '\0';
2017           }
2018         }
2019         res = builder->status;
2020         goto done;
2021       }
2022       default:
2023         res = GST_RTSP_ERROR;
2024         break;
2025     }
2026   }
2027 done:
2028   return res;
2029
2030   /* ERRORS */
2031 invalid_body_len:
2032   {
2033     GST_DEBUG ("could not allocate body");
2034     return GST_RTSP_ERROR;
2035   }
2036 }
2037
2038 /**
2039  * gst_rtsp_connection_read:
2040  * @conn: a #GstRTSPConnection
2041  * @data: the data to read
2042  * @size: the size of @data
2043  * @timeout: a timeout value or #NULL
2044  *
2045  * Attempt to read @size bytes into @data from the connected @conn, blocking up to
2046  * the specified @timeout. @timeout can be #NULL, in which case this function
2047  * might block forever.
2048  *
2049  * This function can be cancelled with gst_rtsp_connection_flush().
2050  *
2051  * Returns: #GST_RTSP_OK on success.
2052  */
2053 GstRTSPResult
2054 gst_rtsp_connection_read (GstRTSPConnection * conn, guint8 * data, guint size,
2055     GTimeVal * timeout)
2056 {
2057   guint offset;
2058   GstClockTime to;
2059   GstRTSPResult res;
2060
2061   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2062   g_return_val_if_fail (data != NULL, GST_RTSP_EINVAL);
2063   g_return_val_if_fail (conn->read_socket != NULL, GST_RTSP_EINVAL);
2064
2065   if (G_UNLIKELY (size == 0))
2066     return GST_RTSP_OK;
2067
2068   offset = 0;
2069
2070   /* configure timeout if any */
2071   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : 0;
2072
2073   g_socket_set_timeout (conn->read_socket, (to + GST_SECOND - 1) / GST_SECOND);
2074   res = read_bytes (conn, data, &offset, size, TRUE);
2075   g_socket_set_timeout (conn->read_socket, 0);
2076
2077   return res;
2078 }
2079
2080 static GstRTSPMessage *
2081 gen_tunnel_reply (GstRTSPConnection * conn, GstRTSPStatusCode code,
2082     const GstRTSPMessage * request)
2083 {
2084   GstRTSPMessage *msg;
2085   GstRTSPResult res;
2086
2087   if (gst_rtsp_status_as_text (code) == NULL)
2088     code = GST_RTSP_STS_INTERNAL_SERVER_ERROR;
2089
2090   GST_RTSP_CHECK (gst_rtsp_message_new_response (&msg, code, NULL, request),
2091       no_message);
2092
2093   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_SERVER,
2094       "GStreamer RTSP Server");
2095   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_CONNECTION, "close");
2096   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_CACHE_CONTROL, "no-store");
2097   gst_rtsp_message_add_header (msg, GST_RTSP_HDR_PRAGMA, "no-cache");
2098
2099   if (code == GST_RTSP_STS_OK) {
2100     /* add the local ip address to the tunnel reply, this is where the client
2101      * should send the POST request to */
2102     if (conn->local_ip)
2103       gst_rtsp_message_add_header (msg, GST_RTSP_HDR_X_SERVER_IP_ADDRESS,
2104           conn->local_ip);
2105     gst_rtsp_message_add_header (msg, GST_RTSP_HDR_CONTENT_TYPE,
2106         "application/x-rtsp-tunnelled");
2107   }
2108
2109   return msg;
2110
2111   /* ERRORS */
2112 no_message:
2113   {
2114     return NULL;
2115   }
2116 }
2117
2118 /**
2119  * gst_rtsp_connection_receive:
2120  * @conn: a #GstRTSPConnection
2121  * @message: the message to read
2122  * @timeout: a timeout value or #NULL
2123  *
2124  * Attempt to read into @message from the connected @conn, blocking up to
2125  * the specified @timeout. @timeout can be #NULL, in which case this function
2126  * might block forever.
2127  *
2128  * This function can be cancelled with gst_rtsp_connection_flush().
2129  *
2130  * Returns: #GST_RTSP_OK on success.
2131  */
2132 GstRTSPResult
2133 gst_rtsp_connection_receive (GstRTSPConnection * conn, GstRTSPMessage * message,
2134     GTimeVal * timeout)
2135 {
2136   GstRTSPResult res;
2137   GstRTSPBuilder builder;
2138   GstClockTime to;
2139
2140   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2141   g_return_val_if_fail (message != NULL, GST_RTSP_EINVAL);
2142   g_return_val_if_fail (conn->read_socket != NULL, GST_RTSP_EINVAL);
2143
2144   /* configure timeout if any */
2145   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : 0;
2146
2147   g_socket_set_timeout (conn->read_socket, (to + GST_SECOND - 1) / GST_SECOND);
2148   memset (&builder, 0, sizeof (GstRTSPBuilder));
2149   res = build_next (&builder, message, conn, TRUE);
2150   g_socket_set_timeout (conn->read_socket, 0);
2151
2152   if (G_UNLIKELY (res != GST_RTSP_OK))
2153     goto read_error;
2154
2155   if (!conn->manual_http) {
2156     if (message->type == GST_RTSP_MESSAGE_HTTP_REQUEST) {
2157       if (conn->tstate == TUNNEL_STATE_NONE &&
2158           message->type_data.request.method == GST_RTSP_GET) {
2159         GstRTSPMessage *response;
2160
2161         conn->tstate = TUNNEL_STATE_GET;
2162
2163         /* tunnel GET request, we can reply now */
2164         response = gen_tunnel_reply (conn, GST_RTSP_STS_OK, message);
2165         res = gst_rtsp_connection_send (conn, response, timeout);
2166         gst_rtsp_message_free (response);
2167         if (res == GST_RTSP_OK)
2168           res = GST_RTSP_ETGET;
2169         goto cleanup;
2170       } else if (conn->tstate == TUNNEL_STATE_NONE &&
2171           message->type_data.request.method == GST_RTSP_POST) {
2172         conn->tstate = TUNNEL_STATE_POST;
2173
2174         /* tunnel POST request, the caller now has to link the two
2175          * connections. */
2176         res = GST_RTSP_ETPOST;
2177         goto cleanup;
2178       } else {
2179         res = GST_RTSP_EPARSE;
2180         goto cleanup;
2181       }
2182     } else if (message->type == GST_RTSP_MESSAGE_HTTP_RESPONSE) {
2183       res = GST_RTSP_EPARSE;
2184       goto cleanup;
2185     }
2186   }
2187
2188   /* we have a message here */
2189   build_reset (&builder);
2190
2191   return GST_RTSP_OK;
2192
2193   /* ERRORS */
2194 read_error:
2195 cleanup:
2196   {
2197     build_reset (&builder);
2198     gst_rtsp_message_unset (message);
2199     return res;
2200   }
2201 }
2202
2203 /**
2204  * gst_rtsp_connection_close:
2205  * @conn: a #GstRTSPConnection
2206  *
2207  * Close the connected @conn. After this call, the connection is in the same
2208  * state as when it was first created.
2209  *
2210  * Returns: #GST_RTSP_OK on success.
2211  */
2212 GstRTSPResult
2213 gst_rtsp_connection_close (GstRTSPConnection * conn)
2214 {
2215   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2216
2217   /* last unref closes the connection we don't want to explicitly close here
2218    * because these sockets might have been provided at construction */
2219   if (conn->stream0) {
2220     g_object_unref (conn->stream0);
2221     conn->stream0 = NULL;
2222     conn->socket0 = NULL;
2223   }
2224   if (conn->stream1) {
2225     g_object_unref (conn->stream1);
2226     conn->stream1 = NULL;
2227     conn->socket1 = NULL;
2228   }
2229
2230   /* these were owned by the stream */
2231   conn->input_stream = NULL;
2232   conn->output_stream = NULL;
2233
2234   g_free (conn->remote_ip);
2235   conn->remote_ip = NULL;
2236   g_free (conn->local_ip);
2237   conn->local_ip = NULL;
2238
2239   conn->read_ahead = 0;
2240
2241   g_free (conn->initial_buffer);
2242   conn->initial_buffer = NULL;
2243   conn->initial_buffer_offset = 0;
2244
2245   conn->write_socket = NULL;
2246   conn->read_socket = NULL;
2247   conn->tunneled = FALSE;
2248   conn->tstate = TUNNEL_STATE_NONE;
2249   conn->ctxp = NULL;
2250   g_free (conn->username);
2251   conn->username = NULL;
2252   g_free (conn->passwd);
2253   conn->passwd = NULL;
2254   gst_rtsp_connection_clear_auth_params (conn);
2255   conn->timeout = 60;
2256   conn->cseq = 0;
2257   conn->session_id[0] = '\0';
2258
2259   return GST_RTSP_OK;
2260 }
2261
2262 /**
2263  * gst_rtsp_connection_free:
2264  * @conn: a #GstRTSPConnection
2265  *
2266  * Close and free @conn.
2267  *
2268  * Returns: #GST_RTSP_OK on success.
2269  */
2270 GstRTSPResult
2271 gst_rtsp_connection_free (GstRTSPConnection * conn)
2272 {
2273   GstRTSPResult res;
2274
2275   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2276
2277   res = gst_rtsp_connection_close (conn);
2278
2279   if (conn->cancellable)
2280     g_object_unref (conn->cancellable);
2281   if (conn->client)
2282     g_object_unref (conn->client);
2283   if (conn->tls_database)
2284     g_object_unref (conn->tls_database);
2285
2286   g_timer_destroy (conn->timer);
2287   gst_rtsp_url_free (conn->url);
2288   g_free (conn->proxy_host);
2289   g_free (conn);
2290
2291   return res;
2292 }
2293
2294 /**
2295  * gst_rtsp_connection_poll:
2296  * @conn: a #GstRTSPConnection
2297  * @events: a bitmask of #GstRTSPEvent flags to check
2298  * @revents: location for result flags
2299  * @timeout: a timeout
2300  *
2301  * Wait up to the specified @timeout for the connection to become available for
2302  * at least one of the operations specified in @events. When the function returns
2303  * with #GST_RTSP_OK, @revents will contain a bitmask of available operations on
2304  * @conn.
2305  *
2306  * @timeout can be #NULL, in which case this function might block forever.
2307  *
2308  * This function can be cancelled with gst_rtsp_connection_flush().
2309  *
2310  * Returns: #GST_RTSP_OK on success.
2311  */
2312 GstRTSPResult
2313 gst_rtsp_connection_poll (GstRTSPConnection * conn, GstRTSPEvent events,
2314     GstRTSPEvent * revents, GTimeVal * timeout)
2315 {
2316   GstClockTime to;
2317   GMainContext *ctx;
2318   GSource *rs, *ws, *ts;
2319   GIOCondition condition;
2320
2321   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2322   g_return_val_if_fail (events != 0, GST_RTSP_EINVAL);
2323   g_return_val_if_fail (revents != NULL, GST_RTSP_EINVAL);
2324   g_return_val_if_fail (conn->read_socket != NULL, GST_RTSP_EINVAL);
2325   g_return_val_if_fail (conn->write_socket != NULL, GST_RTSP_EINVAL);
2326
2327   ctx = g_main_context_new ();
2328
2329   /* configure timeout if any */
2330   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : GST_CLOCK_TIME_NONE;
2331
2332   if (timeout) {
2333     ts = g_timeout_source_new (to / GST_MSECOND);
2334     g_source_set_dummy_callback (ts);
2335     g_source_attach (ts, ctx);
2336     g_source_unref (ts);
2337   }
2338
2339   rs = g_socket_create_source (conn->read_socket,
2340       G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, conn->cancellable);
2341   g_source_set_dummy_callback (rs);
2342   g_source_attach (rs, ctx);
2343   g_source_unref (rs);
2344
2345   ws = g_socket_create_source (conn->write_socket,
2346       G_IO_OUT | G_IO_ERR | G_IO_HUP, conn->cancellable);
2347   g_source_set_dummy_callback (ws);
2348   g_source_attach (ws, ctx);
2349   g_source_unref (ws);
2350
2351   /* Returns after handling all pending events */
2352   g_main_context_iteration (ctx, TRUE);
2353
2354   g_main_context_unref (ctx);
2355
2356   condition =
2357       g_socket_condition_check (conn->read_socket,
2358       G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP);
2359   condition |=
2360       g_socket_condition_check (conn->write_socket,
2361       G_IO_OUT | G_IO_ERR | G_IO_HUP);
2362
2363   *revents = 0;
2364   if (events & GST_RTSP_EV_READ) {
2365     if ((condition & G_IO_IN) || (condition & G_IO_PRI))
2366       *revents |= GST_RTSP_EV_READ;
2367   }
2368   if (events & GST_RTSP_EV_WRITE) {
2369     if ((condition & G_IO_OUT))
2370       *revents |= GST_RTSP_EV_WRITE;
2371   }
2372
2373   if (*revents == 0)
2374     return GST_RTSP_ETIMEOUT;
2375
2376   return GST_RTSP_OK;
2377 }
2378
2379 /**
2380  * gst_rtsp_connection_next_timeout:
2381  * @conn: a #GstRTSPConnection
2382  * @timeout: a timeout
2383  *
2384  * Calculate the next timeout for @conn, storing the result in @timeout.
2385  *
2386  * Returns: #GST_RTSP_OK.
2387  */
2388 GstRTSPResult
2389 gst_rtsp_connection_next_timeout (GstRTSPConnection * conn, GTimeVal * timeout)
2390 {
2391   gdouble elapsed;
2392   glong sec;
2393   gulong usec;
2394   gint ctimeout;
2395
2396   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2397   g_return_val_if_fail (timeout != NULL, GST_RTSP_EINVAL);
2398
2399   ctimeout = conn->timeout;
2400   if (ctimeout >= 20) {
2401     /* Because we should act before the timeout we timeout 5
2402      * seconds in advance. */
2403     ctimeout -= 5;
2404   } else if (ctimeout >= 5) {
2405     /* else timeout 20% earlier */
2406     ctimeout -= ctimeout / 5;
2407   } else if (ctimeout >= 1) {
2408     /* else timeout 1 second earlier */
2409     ctimeout -= 1;
2410   }
2411
2412   elapsed = g_timer_elapsed (conn->timer, &usec);
2413   if (elapsed >= ctimeout) {
2414     sec = 0;
2415     usec = 0;
2416   } else {
2417     sec = ctimeout - elapsed;
2418     if (usec <= G_USEC_PER_SEC)
2419       usec = G_USEC_PER_SEC - usec;
2420     else
2421       usec = 0;
2422   }
2423
2424   timeout->tv_sec = sec;
2425   timeout->tv_usec = usec;
2426
2427   return GST_RTSP_OK;
2428 }
2429
2430 /**
2431  * gst_rtsp_connection_reset_timeout:
2432  * @conn: a #GstRTSPConnection
2433  *
2434  * Reset the timeout of @conn.
2435  *
2436  * Returns: #GST_RTSP_OK.
2437  */
2438 GstRTSPResult
2439 gst_rtsp_connection_reset_timeout (GstRTSPConnection * conn)
2440 {
2441   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2442
2443   g_timer_start (conn->timer);
2444
2445   return GST_RTSP_OK;
2446 }
2447
2448 /**
2449  * gst_rtsp_connection_flush:
2450  * @conn: a #GstRTSPConnection
2451  * @flush: start or stop the flush
2452  *
2453  * Start or stop the flushing action on @conn. When flushing, all current
2454  * and future actions on @conn will return #GST_RTSP_EINTR until the connection
2455  * is set to non-flushing mode again.
2456  *
2457  * Returns: #GST_RTSP_OK.
2458  */
2459 GstRTSPResult
2460 gst_rtsp_connection_flush (GstRTSPConnection * conn, gboolean flush)
2461 {
2462   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2463
2464   if (flush)
2465     g_cancellable_cancel (conn->cancellable);
2466   else
2467     g_cancellable_reset (conn->cancellable);
2468
2469   return GST_RTSP_OK;
2470 }
2471
2472 /**
2473  * gst_rtsp_connection_set_proxy:
2474  * @conn: a #GstRTSPConnection
2475  * @host: the proxy host
2476  * @port: the proxy port
2477  *
2478  * Set the proxy host and port.
2479  *
2480  * Returns: #GST_RTSP_OK.
2481  */
2482 GstRTSPResult
2483 gst_rtsp_connection_set_proxy (GstRTSPConnection * conn,
2484     const gchar * host, guint port)
2485 {
2486   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2487
2488   g_free (conn->proxy_host);
2489   conn->proxy_host = g_strdup (host);
2490   conn->proxy_port = port;
2491
2492   return GST_RTSP_OK;
2493 }
2494
2495 /**
2496  * gst_rtsp_connection_set_auth:
2497  * @conn: a #GstRTSPConnection
2498  * @method: authentication method
2499  * @user: the user
2500  * @pass: the password
2501  *
2502  * Configure @conn for authentication mode @method with @user and @pass as the
2503  * user and password respectively.
2504  *
2505  * Returns: #GST_RTSP_OK.
2506  */
2507 GstRTSPResult
2508 gst_rtsp_connection_set_auth (GstRTSPConnection * conn,
2509     GstRTSPAuthMethod method, const gchar * user, const gchar * pass)
2510 {
2511   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2512
2513   if (method == GST_RTSP_AUTH_DIGEST && ((user == NULL || pass == NULL)
2514           || g_strrstr (user, ":") != NULL))
2515     return GST_RTSP_EINVAL;
2516
2517   /* Make sure the username and passwd are being set for authentication */
2518   if (method == GST_RTSP_AUTH_NONE && (user == NULL || pass == NULL))
2519     return GST_RTSP_EINVAL;
2520
2521   /* ":" chars are not allowed in usernames for basic auth */
2522   if (method == GST_RTSP_AUTH_BASIC && g_strrstr (user, ":") != NULL)
2523     return GST_RTSP_EINVAL;
2524
2525   g_free (conn->username);
2526   g_free (conn->passwd);
2527
2528   conn->auth_method = method;
2529   conn->username = g_strdup (user);
2530   conn->passwd = g_strdup (pass);
2531
2532   return GST_RTSP_OK;
2533 }
2534
2535 /**
2536  * str_case_hash:
2537  * @key: ASCII string to hash
2538  *
2539  * Hashes @key in a case-insensitive manner.
2540  *
2541  * Returns: the hash code.
2542  **/
2543 static guint
2544 str_case_hash (gconstpointer key)
2545 {
2546   const char *p = key;
2547   guint h = g_ascii_toupper (*p);
2548
2549   if (h)
2550     for (p += 1; *p != '\0'; p++)
2551       h = (h << 5) - h + g_ascii_toupper (*p);
2552
2553   return h;
2554 }
2555
2556 /**
2557  * str_case_equal:
2558  * @v1: an ASCII string
2559  * @v2: another ASCII string
2560  *
2561  * Compares @v1 and @v2 in a case-insensitive manner
2562  *
2563  * Returns: %TRUE if they are equal (modulo case)
2564  **/
2565 static gboolean
2566 str_case_equal (gconstpointer v1, gconstpointer v2)
2567 {
2568   const char *string1 = v1;
2569   const char *string2 = v2;
2570
2571   return g_ascii_strcasecmp (string1, string2) == 0;
2572 }
2573
2574 /**
2575  * gst_rtsp_connection_set_auth_param:
2576  * @conn: a #GstRTSPConnection
2577  * @param: authentication directive
2578  * @value: value
2579  *
2580  * Setup @conn with authentication directives. This is not necesary for
2581  * methods #GST_RTSP_AUTH_NONE and #GST_RTSP_AUTH_BASIC. For
2582  * #GST_RTSP_AUTH_DIGEST, directives should be taken from the digest challenge
2583  * in the WWW-Authenticate response header and can include realm, domain,
2584  * nonce, opaque, stale, algorithm, qop as per RFC2617.
2585  */
2586 void
2587 gst_rtsp_connection_set_auth_param (GstRTSPConnection * conn,
2588     const gchar * param, const gchar * value)
2589 {
2590   g_return_if_fail (conn != NULL);
2591   g_return_if_fail (param != NULL);
2592
2593   if (conn->auth_params == NULL) {
2594     conn->auth_params =
2595         g_hash_table_new_full (str_case_hash, str_case_equal, g_free, g_free);
2596   }
2597   g_hash_table_insert (conn->auth_params, g_strdup (param), g_strdup (value));
2598 }
2599
2600 /**
2601  * gst_rtsp_connection_clear_auth_params:
2602  * @conn: a #GstRTSPConnection
2603  *
2604  * Clear the list of authentication directives stored in @conn.
2605  */
2606 void
2607 gst_rtsp_connection_clear_auth_params (GstRTSPConnection * conn)
2608 {
2609   g_return_if_fail (conn != NULL);
2610
2611   if (conn->auth_params != NULL) {
2612     g_hash_table_destroy (conn->auth_params);
2613     conn->auth_params = NULL;
2614   }
2615 }
2616
2617 static GstRTSPResult
2618 set_qos_dscp (GSocket * socket, guint qos_dscp)
2619 {
2620 #ifndef IP_TOS
2621   GST_FIXME ("IP_TOS socket option is not defined, not setting dscp");
2622   return GST_RTSP_OK;
2623 #else
2624   gint fd;
2625   union gst_sockaddr sa;
2626   socklen_t slen = sizeof (sa);
2627   gint af;
2628   gint tos;
2629
2630   if (!socket)
2631     return GST_RTSP_OK;
2632
2633   fd = g_socket_get_fd (socket);
2634   if (getsockname (fd, &sa.sa, &slen) < 0)
2635     goto no_getsockname;
2636
2637   af = sa.sa.sa_family;
2638
2639   /* if this is an IPv4-mapped address then do IPv4 QoS */
2640   if (af == AF_INET6) {
2641     if (IN6_IS_ADDR_V4MAPPED (&sa.sa_in6.sin6_addr))
2642       af = AF_INET;
2643   }
2644
2645   /* extract and shift 6 bits of the DSCP */
2646   tos = (qos_dscp & 0x3f) << 2;
2647
2648   switch (af) {
2649     case AF_INET:
2650       if (setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos)) < 0)
2651         goto no_setsockopt;
2652       break;
2653     case AF_INET6:
2654 #ifdef IPV6_TCLASS
2655       if (setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos)) < 0)
2656         goto no_setsockopt;
2657       break;
2658 #endif
2659     default:
2660       goto wrong_family;
2661   }
2662
2663   return GST_RTSP_OK;
2664
2665   /* ERRORS */
2666 no_getsockname:
2667 no_setsockopt:
2668   {
2669     return GST_RTSP_ESYS;
2670   }
2671 wrong_family:
2672   {
2673     return GST_RTSP_ERROR;
2674   }
2675 #endif
2676 }
2677
2678 /**
2679  * gst_rtsp_connection_set_qos_dscp:
2680  * @conn: a #GstRTSPConnection
2681  * @qos_dscp: DSCP value
2682  *
2683  * Configure @conn to use the specified DSCP value.
2684  *
2685  * Returns: #GST_RTSP_OK on success.
2686  */
2687 GstRTSPResult
2688 gst_rtsp_connection_set_qos_dscp (GstRTSPConnection * conn, guint qos_dscp)
2689 {
2690   GstRTSPResult res;
2691
2692   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2693   g_return_val_if_fail (conn->read_socket != NULL, GST_RTSP_EINVAL);
2694   g_return_val_if_fail (conn->write_socket != NULL, GST_RTSP_EINVAL);
2695
2696   res = set_qos_dscp (conn->socket0, qos_dscp);
2697   if (res == GST_RTSP_OK)
2698     res = set_qos_dscp (conn->socket1, qos_dscp);
2699
2700   return res;
2701 }
2702
2703
2704 /**
2705  * gst_rtsp_connection_get_url:
2706  * @conn: a #GstRTSPConnection
2707  *
2708  * Retrieve the URL of the other end of @conn.
2709  *
2710  * Returns: The URL. This value remains valid until the
2711  * connection is freed.
2712  */
2713 GstRTSPUrl *
2714 gst_rtsp_connection_get_url (const GstRTSPConnection * conn)
2715 {
2716   g_return_val_if_fail (conn != NULL, NULL);
2717
2718   return conn->url;
2719 }
2720
2721 /**
2722  * gst_rtsp_connection_get_ip:
2723  * @conn: a #GstRTSPConnection
2724  *
2725  * Retrieve the IP address of the other end of @conn.
2726  *
2727  * Returns: The IP address as a string. this value remains valid until the
2728  * connection is closed.
2729  */
2730 const gchar *
2731 gst_rtsp_connection_get_ip (const GstRTSPConnection * conn)
2732 {
2733   g_return_val_if_fail (conn != NULL, NULL);
2734
2735   return conn->remote_ip;
2736 }
2737
2738 /**
2739  * gst_rtsp_connection_set_ip:
2740  * @conn: a #GstRTSPConnection
2741  * @ip: an ip address
2742  *
2743  * Set the IP address of the server.
2744  */
2745 void
2746 gst_rtsp_connection_set_ip (GstRTSPConnection * conn, const gchar * ip)
2747 {
2748   g_return_if_fail (conn != NULL);
2749
2750   g_free (conn->remote_ip);
2751   conn->remote_ip = g_strdup (ip);
2752 }
2753
2754 /**
2755  * gst_rtsp_connection_get_readfd:
2756  * @conn: a #GstRTSPConnection
2757  *
2758  * Get the file descriptor for reading.
2759  *
2760  * Returns: (transfer none): the file descriptor used for reading or %NULL on
2761  * error. The file descriptor remains valid until the connection is closed.
2762  */
2763 GSocket *
2764 gst_rtsp_connection_get_read_socket (const GstRTSPConnection * conn)
2765 {
2766   g_return_val_if_fail (conn != NULL, NULL);
2767   g_return_val_if_fail (conn->read_socket != NULL, NULL);
2768
2769   return conn->read_socket;
2770 }
2771
2772 /**
2773  * gst_rtsp_connection_get_write_socket:
2774  * @conn: a #GstRTSPConnection
2775  *
2776  * Get the file descriptor for writing.
2777  *
2778  * Returns: (transfer none): the file descriptor used for writing or NULL on
2779  * error. The file descriptor remains valid until the connection is closed.
2780  */
2781 GSocket *
2782 gst_rtsp_connection_get_write_socket (const GstRTSPConnection * conn)
2783 {
2784   g_return_val_if_fail (conn != NULL, NULL);
2785   g_return_val_if_fail (conn->write_socket != NULL, NULL);
2786
2787   return conn->write_socket;
2788 }
2789
2790 /**
2791  * gst_rtsp_connection_set_http_mode:
2792  * @conn: a #GstRTSPConnection
2793  * @enable: %TRUE to enable manual HTTP mode
2794  *
2795  * By setting the HTTP mode to %TRUE the message parsing will support HTTP
2796  * messages in addition to the RTSP messages. It will also disable the
2797  * automatic handling of setting up an HTTP tunnel.
2798  */
2799 void
2800 gst_rtsp_connection_set_http_mode (GstRTSPConnection * conn, gboolean enable)
2801 {
2802   g_return_if_fail (conn != NULL);
2803
2804   conn->manual_http = enable;
2805 }
2806
2807 /**
2808  * gst_rtsp_connection_set_tunneled:
2809  * @conn: a #GstRTSPConnection
2810  * @tunneled: the new state
2811  *
2812  * Set the HTTP tunneling state of the connection. This must be configured before
2813  * the @conn is connected.
2814  */
2815 void
2816 gst_rtsp_connection_set_tunneled (GstRTSPConnection * conn, gboolean tunneled)
2817 {
2818   g_return_if_fail (conn != NULL);
2819   g_return_if_fail (conn->read_socket == NULL);
2820   g_return_if_fail (conn->write_socket == NULL);
2821
2822   conn->tunneled = tunneled;
2823 }
2824
2825 /**
2826  * gst_rtsp_connection_is_tunneled:
2827  * @conn: a #GstRTSPConnection
2828  *
2829  * Get the tunneling state of the connection.
2830  *
2831  * Returns: if @conn is using HTTP tunneling.
2832  */
2833 gboolean
2834 gst_rtsp_connection_is_tunneled (const GstRTSPConnection * conn)
2835 {
2836   g_return_val_if_fail (conn != NULL, FALSE);
2837
2838   return conn->tunneled;
2839 }
2840
2841 /**
2842  * gst_rtsp_connection_get_tunnelid:
2843  * @conn: a #GstRTSPConnection
2844  *
2845  * Get the tunnel session id the connection.
2846  *
2847  * Returns: returns a non-empty string if @conn is being tunneled over HTTP.
2848  */
2849 const gchar *
2850 gst_rtsp_connection_get_tunnelid (const GstRTSPConnection * conn)
2851 {
2852   g_return_val_if_fail (conn != NULL, NULL);
2853
2854   if (!conn->tunneled)
2855     return NULL;
2856
2857   return conn->tunnelid;
2858 }
2859
2860 /**
2861  * gst_rtsp_connection_do_tunnel:
2862  * @conn: a #GstRTSPConnection
2863  * @conn2: a #GstRTSPConnection or %NULL
2864  *
2865  * If @conn received the first tunnel connection and @conn2 received
2866  * the second tunnel connection, link the two connections together so that
2867  * @conn manages the tunneled connection.
2868  *
2869  * After this call, @conn2 cannot be used anymore and must be freed with
2870  * gst_rtsp_connection_free().
2871  *
2872  * If @conn2 is %NULL then only the base64 decoding context will be setup for
2873  * @conn.
2874  *
2875  * Returns: return GST_RTSP_OK on success.
2876  */
2877 GstRTSPResult
2878 gst_rtsp_connection_do_tunnel (GstRTSPConnection * conn,
2879     GstRTSPConnection * conn2)
2880 {
2881   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2882
2883   if (conn2 != NULL) {
2884     g_return_val_if_fail (conn->tstate == TUNNEL_STATE_GET, GST_RTSP_EINVAL);
2885     g_return_val_if_fail (conn2->tstate == TUNNEL_STATE_POST, GST_RTSP_EINVAL);
2886     g_return_val_if_fail (!memcmp (conn2->tunnelid, conn->tunnelid,
2887             TUNNELID_LEN), GST_RTSP_EINVAL);
2888
2889     /* both connections have socket0 as the read/write socket. start by taking the
2890      * socket from conn2 and set it as the socket in conn */
2891     conn->socket1 = conn2->socket0;
2892     conn->stream1 = conn2->stream0;
2893     conn->input_stream = conn2->input_stream;
2894     conn->control_stream = g_io_stream_get_input_stream (conn->stream0);
2895
2896     /* clean up some of the state of conn2 */
2897     g_cancellable_cancel (conn2->cancellable);
2898     conn2->write_socket = conn2->read_socket = NULL;
2899     conn2->socket0 = NULL;
2900     conn2->stream0 = NULL;
2901     conn2->input_stream = NULL;
2902     conn2->output_stream = NULL;
2903     conn2->control_stream = NULL;
2904     g_cancellable_reset (conn2->cancellable);
2905
2906     /* We make socket0 the write socket and socket1 the read socket. */
2907     conn->write_socket = conn->socket0;
2908     conn->read_socket = conn->socket1;
2909
2910     conn->tstate = TUNNEL_STATE_COMPLETE;
2911
2912     g_free (conn->initial_buffer);
2913     conn->initial_buffer = conn2->initial_buffer;
2914     conn2->initial_buffer = NULL;
2915     conn->initial_buffer_offset = conn2->initial_buffer_offset;
2916   }
2917
2918   /* we need base64 decoding for the readfd */
2919   conn->ctx.state = 0;
2920   conn->ctx.save = 0;
2921   conn->ctx.cout = 0;
2922   conn->ctx.coutl = 0;
2923   conn->ctxp = &conn->ctx;
2924
2925   return GST_RTSP_OK;
2926 }
2927
2928 /**
2929  * gst_rtsp_connection_set_remember_session_id:
2930  * @conn: a #GstRTSPConnection
2931  * @remember: %TRUE if the connection should remember the session id
2932  *
2933  * Sets if the #GstRTSPConnection should remember the session id from the last
2934  * response received and force it onto any further requests.
2935  *
2936  * The default value is %TRUE
2937  */
2938
2939 void
2940 gst_rtsp_connection_set_remember_session_id (GstRTSPConnection * conn,
2941     gboolean remember)
2942 {
2943   conn->remember_session_id = remember;
2944   if (!remember)
2945     conn->session_id[0] = '\0';
2946 }
2947
2948 /**
2949  * gst_rtsp_connection_get_remember_session_id:
2950  * @conn: a #GstRTSPConnection
2951  *
2952  * Returns: %TRUE if the #GstRTSPConnection remembers the session id in the
2953  * last response to set it on any further request.
2954  */
2955
2956 gboolean
2957 gst_rtsp_connection_get_remember_session_id (GstRTSPConnection * conn)
2958 {
2959   return conn->remember_session_id;
2960 }
2961
2962
2963 #define READ_ERR    (G_IO_HUP | G_IO_ERR | G_IO_NVAL)
2964 #define READ_COND   (G_IO_IN | READ_ERR)
2965 #define WRITE_ERR   (G_IO_HUP | G_IO_ERR | G_IO_NVAL)
2966 #define WRITE_COND  (G_IO_OUT | WRITE_ERR)
2967
2968 typedef struct
2969 {
2970   guint8 *data;
2971   guint size;
2972   guint id;
2973 } GstRTSPRec;
2974
2975 /* async functions */
2976 struct _GstRTSPWatch
2977 {
2978   GSource source;
2979
2980   GstRTSPConnection *conn;
2981
2982   GstRTSPBuilder builder;
2983   GstRTSPMessage message;
2984
2985   GSource *readsrc;
2986   GSource *writesrc;
2987   GSource *controlsrc;
2988
2989   gboolean keep_running;
2990
2991   /* queued message for transmission */
2992   guint id;
2993   GMutex mutex;
2994   GQueue *messages;
2995   gsize messages_bytes;
2996   guint8 *write_data;
2997   guint write_off;
2998   guint write_size;
2999   guint write_id;
3000   gsize max_bytes;
3001   guint max_messages;
3002
3003   GstRTSPWatchFuncs funcs;
3004
3005   gpointer user_data;
3006   GDestroyNotify notify;
3007 };
3008
3009 static gboolean
3010 gst_rtsp_source_prepare (GSource * source, gint * timeout)
3011 {
3012   GstRTSPWatch *watch = (GstRTSPWatch *) source;
3013
3014   if (watch->conn->initial_buffer != NULL)
3015     return TRUE;
3016
3017   *timeout = (watch->conn->timeout * 1000);
3018
3019   return FALSE;
3020 }
3021
3022 static gboolean
3023 gst_rtsp_source_check (GSource * source)
3024 {
3025   return FALSE;
3026 }
3027
3028 static gboolean
3029 gst_rtsp_source_dispatch_read_get_channel (GPollableInputStream * stream,
3030     GstRTSPWatch * watch)
3031 {
3032   gssize count;
3033   guint8 buffer[1024];
3034   GError *error = NULL;
3035
3036   /* try to read in order to be able to detect errors, we read 1k in case some
3037    * client actually decides to send data on the GET channel */
3038   count = g_pollable_input_stream_read_nonblocking (stream, buffer, 1024, NULL,
3039       &error);
3040   if (count == 0) {
3041     /* other end closed the socket */
3042     goto eof;
3043   }
3044
3045   if (count < 0) {
3046     GST_DEBUG ("%s", error->message);
3047     if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK) ||
3048         g_error_matches (error, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
3049       g_clear_error (&error);
3050       goto done;
3051     }
3052     g_clear_error (&error);
3053     goto read_error;
3054   }
3055
3056   /* client sent data on the GET channel, ignore it */
3057
3058 done:
3059   return TRUE;
3060
3061   /* ERRORS */
3062 eof:
3063   {
3064     if (watch->funcs.closed)
3065       watch->funcs.closed (watch, watch->user_data);
3066
3067     /* the read connection was closed, stop the watch now */
3068     watch->keep_running = FALSE;
3069
3070     return FALSE;
3071   }
3072 read_error:
3073   {
3074     if (watch->funcs.error_full)
3075       watch->funcs.error_full (watch, GST_RTSP_ESYS, &watch->message,
3076           0, watch->user_data);
3077     else if (watch->funcs.error)
3078       watch->funcs.error (watch, GST_RTSP_ESYS, watch->user_data);
3079
3080     goto eof;
3081   }
3082 }
3083
3084 static gboolean
3085 gst_rtsp_source_dispatch_read (GPollableInputStream * stream,
3086     GstRTSPWatch * watch)
3087 {
3088   GstRTSPResult res = GST_RTSP_ERROR;
3089   GstRTSPConnection *conn = watch->conn;
3090
3091   /* if this connection was already closed, stop now */
3092   if (G_POLLABLE_INPUT_STREAM (conn->input_stream) != stream)
3093     goto eof;
3094
3095   res = build_next (&watch->builder, &watch->message, conn, FALSE);
3096   if (res == GST_RTSP_EINTR)
3097     goto done;
3098   else if (G_UNLIKELY (res == GST_RTSP_EEOF)) {
3099     if (watch->readsrc) {
3100       g_source_remove_child_source ((GSource *) watch, watch->readsrc);
3101       g_source_unref (watch->readsrc);
3102       watch->readsrc = NULL;
3103     }
3104
3105     if (conn->stream1) {
3106       g_object_unref (conn->stream1);
3107       conn->stream1 = NULL;
3108       conn->socket1 = NULL;
3109       conn->input_stream = NULL;
3110     }
3111
3112     /* When we are in tunnelled mode, the read socket can be closed and we
3113      * should be prepared for a new POST method to reopen it */
3114     if (conn->tstate == TUNNEL_STATE_COMPLETE) {
3115       /* remove the read connection for the tunnel */
3116       /* we accept a new POST request */
3117       conn->tstate = TUNNEL_STATE_GET;
3118       /* and signal that we lost our tunnel */
3119       if (watch->funcs.tunnel_lost)
3120         res = watch->funcs.tunnel_lost (watch, watch->user_data);
3121       goto read_done;
3122     } else
3123       goto eof;
3124   } else if (G_LIKELY (res == GST_RTSP_OK)) {
3125     if (!conn->manual_http &&
3126         watch->message.type == GST_RTSP_MESSAGE_HTTP_REQUEST) {
3127       if (conn->tstate == TUNNEL_STATE_NONE &&
3128           watch->message.type_data.request.method == GST_RTSP_GET) {
3129         GstRTSPMessage *response;
3130         GstRTSPStatusCode code;
3131
3132         conn->tstate = TUNNEL_STATE_GET;
3133
3134         if (watch->funcs.tunnel_start)
3135           code = watch->funcs.tunnel_start (watch, watch->user_data);
3136         else
3137           code = GST_RTSP_STS_OK;
3138
3139         /* queue the response */
3140         response = gen_tunnel_reply (conn, code, &watch->message);
3141         if (watch->funcs.tunnel_http_response)
3142           watch->funcs.tunnel_http_response (watch, &watch->message, response,
3143               watch->user_data);
3144         gst_rtsp_watch_send_message (watch, response, NULL);
3145         gst_rtsp_message_free (response);
3146         goto read_done;
3147       } else if (conn->tstate == TUNNEL_STATE_NONE &&
3148           watch->message.type_data.request.method == GST_RTSP_POST) {
3149         conn->tstate = TUNNEL_STATE_POST;
3150
3151         /* in the callback the connection should be tunneled with the
3152          * GET connection */
3153         if (watch->funcs.tunnel_complete) {
3154           watch->funcs.tunnel_complete (watch, watch->user_data);
3155         }
3156         goto read_done;
3157       }
3158     }
3159   } else
3160     goto read_error;
3161
3162   if (!conn->manual_http) {
3163     /* if manual HTTP support is not enabled, then restore the message to
3164      * what it would have looked like without the support for parsing HTTP
3165      * messages being present */
3166     if (watch->message.type == GST_RTSP_MESSAGE_HTTP_REQUEST) {
3167       watch->message.type = GST_RTSP_MESSAGE_REQUEST;
3168       watch->message.type_data.request.method = GST_RTSP_INVALID;
3169       if (watch->message.type_data.request.version != GST_RTSP_VERSION_1_0)
3170         watch->message.type_data.request.version = GST_RTSP_VERSION_INVALID;
3171       res = GST_RTSP_EPARSE;
3172     } else if (watch->message.type == GST_RTSP_MESSAGE_HTTP_RESPONSE) {
3173       watch->message.type = GST_RTSP_MESSAGE_RESPONSE;
3174       if (watch->message.type_data.response.version != GST_RTSP_VERSION_1_0)
3175         watch->message.type_data.response.version = GST_RTSP_VERSION_INVALID;
3176       res = GST_RTSP_EPARSE;
3177     }
3178   }
3179   if (G_LIKELY (res != GST_RTSP_OK))
3180     goto read_error;
3181
3182   if (watch->funcs.message_received)
3183     watch->funcs.message_received (watch, &watch->message, watch->user_data);
3184
3185 read_done:
3186   gst_rtsp_message_unset (&watch->message);
3187   build_reset (&watch->builder);
3188
3189 done:
3190   return TRUE;
3191
3192   /* ERRORS */
3193 eof:
3194   {
3195     if (watch->funcs.closed)
3196       watch->funcs.closed (watch, watch->user_data);
3197
3198     /* we closed the read connection, stop the watch now */
3199     watch->keep_running = FALSE;
3200
3201     /* always stop when the input returns EOF in non-tunneled mode */
3202     return FALSE;
3203   }
3204 read_error:
3205   {
3206     if (watch->funcs.error_full)
3207       watch->funcs.error_full (watch, res, &watch->message,
3208           0, watch->user_data);
3209     else if (watch->funcs.error)
3210       watch->funcs.error (watch, res, watch->user_data);
3211
3212     goto eof;
3213   }
3214 }
3215
3216 static gboolean
3217 gst_rtsp_source_dispatch (GSource * source, GSourceFunc callback G_GNUC_UNUSED,
3218     gpointer user_data G_GNUC_UNUSED)
3219 {
3220   GstRTSPWatch *watch = (GstRTSPWatch *) source;
3221   GstRTSPConnection *conn = watch->conn;
3222
3223   if (conn->initial_buffer != NULL) {
3224     gst_rtsp_source_dispatch_read (G_POLLABLE_INPUT_STREAM (conn->input_stream),
3225         watch);
3226   }
3227   return watch->keep_running;
3228 }
3229
3230 static gboolean
3231 gst_rtsp_source_dispatch_write (GPollableOutputStream * stream,
3232     GstRTSPWatch * watch)
3233 {
3234   GstRTSPResult res = GST_RTSP_ERROR;
3235   GstRTSPConnection *conn = watch->conn;
3236
3237   /* if this connection was already closed, stop now */
3238   if (G_POLLABLE_OUTPUT_STREAM (conn->output_stream) != stream)
3239     goto eof;
3240
3241   g_mutex_lock (&watch->mutex);
3242   do {
3243     if (watch->write_data == NULL) {
3244       GstRTSPRec *rec;
3245
3246       /* get a new message from the queue */
3247       rec = g_queue_pop_tail (watch->messages);
3248       if (rec == NULL) {
3249         if (watch->writesrc) {
3250           g_source_remove_child_source ((GSource *) watch, watch->writesrc);
3251           g_source_unref (watch->writesrc);
3252           watch->writesrc = NULL;
3253           /* we create and add the write source again when we actually have
3254            * something to write */
3255
3256           /* since write source is now removed we add read source on the write
3257            * socket instead to be able to detect when client closes get channel
3258            * in tunneled mode */
3259           if (watch->conn->control_stream) {
3260             watch->controlsrc =
3261                 g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM
3262                 (watch->conn->control_stream), NULL);
3263             g_source_set_callback (watch->controlsrc,
3264                 (GSourceFunc) gst_rtsp_source_dispatch_read_get_channel, watch,
3265                 NULL);
3266             g_source_add_child_source ((GSource *) watch, watch->controlsrc);
3267           } else {
3268             watch->controlsrc = NULL;
3269           }
3270         }
3271         break;
3272       }
3273
3274       watch->messages_bytes -= rec->size;
3275
3276       watch->write_off = 0;
3277       watch->write_data = rec->data;
3278       watch->write_size = rec->size;
3279       watch->write_id = rec->id;
3280
3281       g_slice_free (GstRTSPRec, rec);
3282     }
3283
3284     res = write_bytes (conn->output_stream, watch->write_data,
3285         &watch->write_off, watch->write_size, FALSE, conn->cancellable);
3286     g_mutex_unlock (&watch->mutex);
3287
3288     if (res == GST_RTSP_EINTR)
3289       goto write_blocked;
3290     else if (G_LIKELY (res == GST_RTSP_OK)) {
3291       if (watch->funcs.message_sent)
3292         watch->funcs.message_sent (watch, watch->write_id, watch->user_data);
3293     } else {
3294       goto write_error;
3295     }
3296     g_mutex_lock (&watch->mutex);
3297
3298     g_free (watch->write_data);
3299     watch->write_data = NULL;
3300   } while (TRUE);
3301   g_mutex_unlock (&watch->mutex);
3302
3303 write_blocked:
3304   return TRUE;
3305
3306   /* ERRORS */
3307 eof:
3308   {
3309     return FALSE;
3310   }
3311 write_error:
3312   {
3313     if (watch->funcs.error_full)
3314       watch->funcs.error_full (watch, res, NULL,
3315           watch->write_id, watch->user_data);
3316     else if (watch->funcs.error)
3317       watch->funcs.error (watch, res, watch->user_data);
3318
3319     return FALSE;
3320   }
3321 }
3322
3323 static void
3324 gst_rtsp_rec_free (gpointer data)
3325 {
3326   GstRTSPRec *rec = data;
3327
3328   g_free (rec->data);
3329   g_slice_free (GstRTSPRec, rec);
3330 }
3331
3332 static void
3333 gst_rtsp_source_finalize (GSource * source)
3334 {
3335   GstRTSPWatch *watch = (GstRTSPWatch *) source;
3336
3337   build_reset (&watch->builder);
3338   gst_rtsp_message_unset (&watch->message);
3339
3340   g_queue_foreach (watch->messages, (GFunc) gst_rtsp_rec_free, NULL);
3341   g_queue_free (watch->messages);
3342   watch->messages = NULL;
3343   watch->messages_bytes = 0;
3344   g_free (watch->write_data);
3345
3346   if (watch->readsrc)
3347     g_source_unref (watch->readsrc);
3348   if (watch->writesrc)
3349     g_source_unref (watch->writesrc);
3350   if (watch->controlsrc)
3351     g_source_unref (watch->controlsrc);
3352
3353   g_mutex_clear (&watch->mutex);
3354
3355   if (watch->notify)
3356     watch->notify (watch->user_data);
3357 }
3358
3359 static GSourceFuncs gst_rtsp_source_funcs = {
3360   gst_rtsp_source_prepare,
3361   gst_rtsp_source_check,
3362   gst_rtsp_source_dispatch,
3363   gst_rtsp_source_finalize,
3364   NULL,
3365   NULL
3366 };
3367
3368 /**
3369  * gst_rtsp_watch_new:
3370  * @conn: a #GstRTSPConnection
3371  * @funcs: watch functions
3372  * @user_data: user data to pass to @funcs
3373  * @notify: notify when @user_data is not referenced anymore
3374  *
3375  * Create a watch object for @conn. The functions provided in @funcs will be
3376  * called with @user_data when activity happened on the watch.
3377  *
3378  * The new watch is usually created so that it can be attached to a
3379  * maincontext with gst_rtsp_watch_attach().
3380  *
3381  * @conn must exist for the entire lifetime of the watch.
3382  *
3383  * Returns: a #GstRTSPWatch that can be used for asynchronous RTSP
3384  * communication. Free with gst_rtsp_watch_unref () after usage.
3385  */
3386 GstRTSPWatch *
3387 gst_rtsp_watch_new (GstRTSPConnection * conn,
3388     GstRTSPWatchFuncs * funcs, gpointer user_data, GDestroyNotify notify)
3389 {
3390   GstRTSPWatch *result;
3391
3392   g_return_val_if_fail (conn != NULL, NULL);
3393   g_return_val_if_fail (funcs != NULL, NULL);
3394   g_return_val_if_fail (conn->read_socket != NULL, NULL);
3395   g_return_val_if_fail (conn->write_socket != NULL, NULL);
3396
3397   result = (GstRTSPWatch *) g_source_new (&gst_rtsp_source_funcs,
3398       sizeof (GstRTSPWatch));
3399
3400   result->conn = conn;
3401   result->builder.state = STATE_START;
3402
3403   g_mutex_init (&result->mutex);
3404   result->messages = g_queue_new ();
3405
3406   gst_rtsp_watch_reset (result);
3407   result->keep_running = TRUE;
3408
3409   result->funcs = *funcs;
3410   result->user_data = user_data;
3411   result->notify = notify;
3412
3413   return result;
3414 }
3415
3416 /**
3417  * gst_rtsp_watch_reset:
3418  * @watch: a #GstRTSPWatch
3419  *
3420  * Reset @watch, this is usually called after gst_rtsp_connection_do_tunnel()
3421  * when the file descriptors of the connection might have changed.
3422  */
3423 void
3424 gst_rtsp_watch_reset (GstRTSPWatch * watch)
3425 {
3426   if (watch->readsrc) {
3427     g_source_remove_child_source ((GSource *) watch, watch->readsrc);
3428     g_source_unref (watch->readsrc);
3429   }
3430   if (watch->writesrc) {
3431     g_source_remove_child_source ((GSource *) watch, watch->writesrc);
3432     g_source_unref (watch->writesrc);
3433     watch->writesrc = NULL;
3434   }
3435   if (watch->controlsrc) {
3436     g_source_remove_child_source ((GSource *) watch, watch->controlsrc);
3437     g_source_unref (watch->controlsrc);
3438     watch->controlsrc = NULL;
3439   }
3440
3441   if (watch->conn->input_stream) {
3442     watch->readsrc =
3443         g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM
3444         (watch->conn->input_stream), NULL);
3445     g_source_set_callback (watch->readsrc,
3446         (GSourceFunc) gst_rtsp_source_dispatch_read, watch, NULL);
3447     g_source_add_child_source ((GSource *) watch, watch->readsrc);
3448   } else {
3449     watch->readsrc = NULL;
3450   }
3451
3452   /* we create and add the write source when we actually have something to
3453    * write */
3454
3455   /* when write source is not added we add read source on the write socket
3456    * instead to be able to detect when client closes get channel in tunneled
3457    * mode */
3458   if (watch->conn->control_stream) {
3459     watch->controlsrc =
3460         g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM
3461         (watch->conn->control_stream), NULL);
3462     g_source_set_callback (watch->controlsrc,
3463         (GSourceFunc) gst_rtsp_source_dispatch_read_get_channel, watch, NULL);
3464     g_source_add_child_source ((GSource *) watch, watch->controlsrc);
3465   } else {
3466     watch->controlsrc = NULL;
3467   }
3468 }
3469
3470 /**
3471  * gst_rtsp_watch_attach:
3472  * @watch: a #GstRTSPWatch
3473  * @context: a GMainContext (if NULL, the default context will be used)
3474  *
3475  * Adds a #GstRTSPWatch to a context so that it will be executed within that context.
3476  *
3477  * Returns: the ID (greater than 0) for the watch within the GMainContext.
3478  */
3479 guint
3480 gst_rtsp_watch_attach (GstRTSPWatch * watch, GMainContext * context)
3481 {
3482   g_return_val_if_fail (watch != NULL, 0);
3483
3484   return g_source_attach ((GSource *) watch, context);
3485 }
3486
3487 /**
3488  * gst_rtsp_watch_unref:
3489  * @watch: a #GstRTSPWatch
3490  *
3491  * Decreases the reference count of @watch by one. If the resulting reference
3492  * count is zero the watch and associated memory will be destroyed.
3493  */
3494 void
3495 gst_rtsp_watch_unref (GstRTSPWatch * watch)
3496 {
3497   g_return_if_fail (watch != NULL);
3498
3499   g_source_unref ((GSource *) watch);
3500 }
3501
3502 /**
3503  * gst_rtsp_watch_set_send_backlog:
3504  * @watch: a #GstRTSPWatch
3505  * @bytes: maximum bytes
3506  * @messages: maximum messages
3507  *
3508  * Set the maximum amount of bytes and messages that will be queued in @watch.
3509  * When the maximum amounts are exceeded, gst_rtsp_watch_write_data() and
3510  * gst_rtsp_watch_send_message() will return #GST_RTSP_ENOMEM.
3511  *
3512  * A value of 0 for @bytes or @messages means no limits.
3513  *
3514  * Since: 1.2
3515  */
3516 void
3517 gst_rtsp_watch_set_send_backlog (GstRTSPWatch * watch,
3518     gsize bytes, guint messages)
3519 {
3520   g_return_if_fail (watch != NULL);
3521
3522   g_mutex_lock (&watch->mutex);
3523   watch->max_bytes = bytes;
3524   watch->max_messages = messages;
3525   g_mutex_unlock (&watch->mutex);
3526
3527   GST_DEBUG ("set backlog to bytes %" G_GSIZE_FORMAT ", messages %u",
3528       bytes, messages);
3529 }
3530
3531 /**
3532  * gst_rtsp_watch_get_send_backlog:
3533  * @watch: a #GstRTSPWatch
3534  * @bytes: (out) (allow-none): maximum bytes
3535  * @messages: (out) (allow-none): maximum messages
3536  *
3537  * Get the maximum amount of bytes and messages that will be queued in @watch.
3538  * See gst_rtsp_watch_set_send_backlog().
3539  *
3540  * Since: 1.2
3541  */
3542 void
3543 gst_rtsp_watch_get_send_backlog (GstRTSPWatch * watch,
3544     gsize * bytes, guint * messages)
3545 {
3546   g_return_if_fail (watch != NULL);
3547
3548   g_mutex_lock (&watch->mutex);
3549   if (bytes)
3550     *bytes = watch->max_bytes;
3551   if (messages)
3552     *messages = watch->max_messages;
3553   g_mutex_unlock (&watch->mutex);
3554 }
3555
3556 /**
3557  * gst_rtsp_watch_write_data:
3558  * @watch: a #GstRTSPWatch
3559  * @data: (array length=size) (transfer full): the data to queue
3560  * @size: the size of @data
3561  * @id: (out) (allow-none): location for a message ID or %NULL
3562  *
3563  * Write @data using the connection of the @watch. If it cannot be sent
3564  * immediately, it will be queued for transmission in @watch. The contents of
3565  * @message will then be serialized and transmitted when the connection of the
3566  * @watch becomes writable. In case the @message is queued, the ID returned in
3567  * @id will be non-zero and used as the ID argument in the message_sent
3568  * callback.
3569  *
3570  * This function will take ownership of @data and g_free() it after use.
3571  *
3572  * If the amount of queued data exceeds the limits set with
3573  * gst_rtsp_watch_set_send_backlog(), this function will return
3574  * #GST_RTSP_ENOMEM.
3575  *
3576  * Returns: #GST_RTSP_OK on success. #GST_RTSP_ENOMEM when the backlog limits
3577  * are reached.
3578  */
3579 GstRTSPResult
3580 gst_rtsp_watch_write_data (GstRTSPWatch * watch, const guint8 * data,
3581     guint size, guint * id)
3582 {
3583   GstRTSPResult res;
3584   GstRTSPRec *rec;
3585   guint off = 0;
3586   GMainContext *context = NULL;
3587
3588   g_return_val_if_fail (watch != NULL, GST_RTSP_EINVAL);
3589   g_return_val_if_fail (data != NULL, GST_RTSP_EINVAL);
3590   g_return_val_if_fail (size != 0, GST_RTSP_EINVAL);
3591
3592   g_mutex_lock (&watch->mutex);
3593
3594   /* try to send the message synchronously first */
3595   if (watch->messages->length == 0 && watch->write_data == NULL) {
3596     res =
3597         write_bytes (watch->conn->output_stream, data, &off, size,
3598         FALSE, watch->conn->cancellable);
3599     if (res != GST_RTSP_EINTR) {
3600       if (id != NULL)
3601         *id = 0;
3602       g_free ((gpointer) data);
3603       goto done;
3604     }
3605   }
3606
3607   /* check limits */
3608   if ((watch->max_bytes != 0 && watch->messages_bytes >= watch->max_bytes) ||
3609       (watch->max_messages != 0
3610           && watch->messages->length >= watch->max_messages))
3611     goto too_much_backlog;
3612
3613   /* make a record with the data and id for sending async */
3614   rec = g_slice_new (GstRTSPRec);
3615   if (off == 0) {
3616     rec->data = (guint8 *) data;
3617     rec->size = size;
3618   } else {
3619     rec->data = g_memdup (data + off, size - off);
3620     rec->size = size - off;
3621     g_free ((gpointer) data);
3622   }
3623
3624   do {
3625     /* make sure rec->id is never 0 */
3626     rec->id = ++watch->id;
3627   } while (G_UNLIKELY (rec->id == 0));
3628
3629   /* add the record to a queue. */
3630   g_queue_push_head (watch->messages, rec);
3631   watch->messages_bytes += rec->size;
3632
3633   /* make sure the main context will now also check for writability on the
3634    * socket */
3635   context = ((GSource *) watch)->context;
3636   if (!watch->writesrc) {
3637     /* remove the read source on the write socket, we will be able to detect
3638      * errors while writing */
3639     if (watch->controlsrc) {
3640       g_source_remove_child_source ((GSource *) watch, watch->controlsrc);
3641       g_source_unref (watch->controlsrc);
3642       watch->controlsrc = NULL;
3643     }
3644
3645     watch->writesrc =
3646         g_pollable_output_stream_create_source (G_POLLABLE_OUTPUT_STREAM
3647         (watch->conn->output_stream), NULL);
3648     g_source_set_callback (watch->writesrc,
3649         (GSourceFunc) gst_rtsp_source_dispatch_write, watch, NULL);
3650     g_source_add_child_source ((GSource *) watch, watch->writesrc);
3651   }
3652
3653   if (id != NULL)
3654     *id = rec->id;
3655   res = GST_RTSP_OK;
3656
3657 done:
3658   g_mutex_unlock (&watch->mutex);
3659
3660   if (context)
3661     g_main_context_wakeup (context);
3662
3663   return res;
3664
3665   /* ERRORS */
3666 too_much_backlog:
3667   {
3668     GST_WARNING ("too much backlog: max_bytes %" G_GSIZE_FORMAT ", current %"
3669         G_GSIZE_FORMAT ", max_messages %u, current %u", watch->max_bytes,
3670         watch->messages_bytes, watch->max_messages, watch->messages->length);
3671     g_mutex_unlock (&watch->mutex);
3672     g_free ((gpointer) data);
3673     return GST_RTSP_ENOMEM;
3674   }
3675 }
3676
3677 /**
3678  * gst_rtsp_watch_send_message:
3679  * @watch: a #GstRTSPWatch
3680  * @message: a #GstRTSPMessage
3681  * @id: (out) (allow-none): location for a message ID or %NULL
3682  *
3683  * Send a @message using the connection of the @watch. If it cannot be sent
3684  * immediately, it will be queued for transmission in @watch. The contents of
3685  * @message will then be serialized and transmitted when the connection of the
3686  * @watch becomes writable. In case the @message is queued, the ID returned in
3687  * @id will be non-zero and used as the ID argument in the message_sent
3688  * callback.
3689  *
3690  * Returns: #GST_RTSP_OK on success.
3691  */
3692 GstRTSPResult
3693 gst_rtsp_watch_send_message (GstRTSPWatch * watch, GstRTSPMessage * message,
3694     guint * id)
3695 {
3696   GString *str;
3697   guint size;
3698
3699   g_return_val_if_fail (watch != NULL, GST_RTSP_EINVAL);
3700   g_return_val_if_fail (message != NULL, GST_RTSP_EINVAL);
3701
3702   /* make a record with the message as a string and id */
3703   str = message_to_string (watch->conn, message);
3704   size = str->len;
3705   return gst_rtsp_watch_write_data (watch,
3706       (guint8 *) g_string_free (str, FALSE), size, id);
3707 }