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