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., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, 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  * <refsect2>
  49  * <para>
  50  * This object manages the RTSP connection to the server. It provides function
  51  * to receive and send bytes and messages.
  52  * </para>
  53  * </refsect2>
  54  *
  55  * Last reviewed on 2007-07-24 (0.10.14)
  56  */
  57
  58 #ifdef HAVE_CONFIG_H
  59 #  include <config.h>
  60 #endif
  61
  62 #include <stdio.h>
  63 #include <errno.h>
  64 #include <stdlib.h>
  65 #include <string.h>
  66 #include <time.h>
  67
  68 #ifdef HAVE_UNISTD_H
  69 #include <unistd.h>
  70 #endif
  71
  72 /* we include this here to get the G_OS_* defines */
  73 #include <glib.h>
  74 #include <gst/gst.h>
  75
  76 #ifdef G_OS_WIN32
  77 /* ws2_32.dll has getaddrinfo and freeaddrinfo on Windows XP and later.
  78  * minwg32 headers check WINVER before allowing the use of these */
  79 #ifndef WINVER
  80 #define WINVER 0x0501
  81 #endif
  82 #include <winsock2.h>
  83 #include <ws2tcpip.h>
  84 #define EINPROGRESS WSAEINPROGRESS
  85 #else
  86 #include <sys/ioctl.h>
  87 #include <netdb.h>
  88 #include <sys/socket.h>
  89 #include <fcntl.h>
  90 #include <netinet/in.h>
  91 #endif
  92
  93 #ifdef HAVE_FIONREAD_IN_SYS_FILIO
  94 #include <sys/filio.h>
  95 #endif
  96
  97 #include "gstrtspconnection.h"
  98 #include "gstrtspbase64.h"
  99
 100 union gst_sockaddr
 101 {
 102   struct sockaddr sa;
 103   struct sockaddr_in sa_in;
 104   struct sockaddr_in6 sa_in6;
 105   struct sockaddr_storage sa_stor;
 106 };
 107
 108 typedef struct
 109 {
 110   gint state;
 111   guint save;
 112   guchar out[3];                /* the size must be evenly divisible by 3 */
 113   guint cout;
 114   guint coutl;
 115 } DecodeCtx;
 116
 117 static GstRTSPResult read_line (gint fd, guint8 * buffer, guint * idx,
 118     guint size, DecodeCtx * ctxp);
 119 static GstRTSPResult parse_key_value (guint8 * buffer, gchar * key,
 120     guint keysize, gchar ** value);
 121 static void parse_string (gchar * dest, gint size, gchar ** src);
 122
 123 #ifdef G_OS_WIN32
 124 #define READ_SOCKET(fd, buf, len) recv (fd, (char *)buf, len, 0)
 125 #define WRITE_SOCKET(fd, buf, len) send (fd, (const char *)buf, len, 0)
 126 #define SETSOCKOPT(sock, level, name, val, len) setsockopt (sock, level, name, (const char *)val, len)
 127 #define CLOSE_SOCKET(sock) closesocket (sock)
 128 #define ERRNO_IS_EAGAIN (WSAGetLastError () == WSAEWOULDBLOCK)
 129 #define ERRNO_IS_EINTR (WSAGetLastError () == WSAEINTR)
 130 /* According to Microsoft's connect() documentation this one returns
 131  * WSAEWOULDBLOCK and not WSAEINPROGRESS. */
 132 #define ERRNO_IS_EINPROGRESS (WSAGetLastError () == WSAEWOULDBLOCK)
 133 #else
 134 #define READ_SOCKET(fd, buf, len) read (fd, buf, len)
 135 #define WRITE_SOCKET(fd, buf, len) write (fd, buf, len)
 136 #define SETSOCKOPT(sock, level, name, val, len) setsockopt (sock, level, name, val, len)
 137 #define CLOSE_SOCKET(sock) close (sock)
 138 #define ERRNO_IS_EAGAIN (errno == EAGAIN)
 139 #define ERRNO_IS_EINTR (errno == EINTR)
 140 #define ERRNO_IS_EINPROGRESS (errno == EINPROGRESS)
 141 #endif
 142
 143 #define ADD_POLLFD(fdset, pfd, fd)        \
 144 G_STMT_START {                            \
 145   (pfd)->fd = fd;                         \
 146   gst_poll_add_fd (fdset, pfd);           \
 147 } G_STMT_END
 148
 149 #define REMOVE_POLLFD(fdset, pfd)          \
 150 G_STMT_START {                             \
 151   if ((pfd)->fd != -1) {                   \
 152     GST_DEBUG ("remove fd %d", (pfd)->fd); \
 153     gst_poll_remove_fd (fdset, pfd);       \
 154     CLOSE_SOCKET ((pfd)->fd);              \
 155     (pfd)->fd = -1;                        \
 156   }                                        \
 157 } G_STMT_END
 158
 159 typedef enum
 160 {
 161   TUNNEL_STATE_NONE,
 162   TUNNEL_STATE_GET,
 163   TUNNEL_STATE_POST,
 164   TUNNEL_STATE_COMPLETE
 165 } GstRTSPTunnelState;
 166
 167 #define TUNNELID_LEN   24
 168
 169 struct _GstRTSPConnection
 170 {
 171   /*< private > */
 172   /* URL for the connection */
 173   GstRTSPUrl *url;
 174
 175   /* connection state */
 176   GstPollFD fd0;
 177   GstPollFD fd1;
 178
 179   GstPollFD *readfd;
 180   GstPollFD *writefd;
 181
 182   gchar tunnelid[TUNNELID_LEN];
 183   gboolean tunneled;
 184   GstRTSPTunnelState tstate;
 185
 186   GstPoll *fdset;
 187   gchar *ip;
 188
 189   /* Session state */
 190   gint cseq;                    /* sequence number */
 191   gchar session_id[512];        /* session id */
 192   gint timeout;                 /* session timeout in seconds */
 193   GTimer *timer;                /* timeout timer */
 194
 195   /* Authentication */
 196   GstRTSPAuthMethod auth_method;
 197   gchar *username;
 198   gchar *passwd;
 199   GHashTable *auth_params;
 200
 201   DecodeCtx ctx;
 202   DecodeCtx *ctxp;
 203
 204   gchar *proxy_host;
 205   guint proxy_port;
 206 };
 207
 208 enum
 209 {
 210   STATE_START = 0,
 211   STATE_DATA_HEADER,
 212   STATE_DATA_BODY,
 213   STATE_READ_LINES,
 214   STATE_END,
 215   STATE_LAST
 216 };
 217
 218 /* a structure for constructing RTSPMessages */
 219 typedef struct
 220 {
 221   gint state;
 222   guint8 buffer[4096];
 223   guint offset;
 224
 225   guint line;
 226   guint8 *body_data;
 227   glong body_len;
 228 } GstRTSPBuilder;
 229
 230 static void
 231 build_reset (GstRTSPBuilder * builder)
 232 {
 233   g_free (builder->body_data);
 234   memset (builder, 0, sizeof (GstRTSPBuilder));
 235 }
 236
 237 /**
 238  * gst_rtsp_connection_create:
 239  * @url: a #GstRTSPUrl
 240  * @conn: storage for a #GstRTSPConnection
 241  *
 242  * Create a newly allocated #GstRTSPConnection from @url and store it in @conn.
 243  * The connection will not yet attempt to connect to @url, use
 244  * gst_rtsp_connection_connect().
 245  *
 246  * A copy of @url will be made.
 247  *
 248  * Returns: #GST_RTSP_OK when @conn contains a valid connection.
 249  */
 250 GstRTSPResult
 251 gst_rtsp_connection_create (const GstRTSPUrl * url, GstRTSPConnection ** conn)
 252 {
 253   GstRTSPConnection *newconn;
 254 #ifdef G_OS_WIN32
 255   WSADATA w;
 256   int error;
 257 #endif
 258
 259   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
 260
 261 #ifdef G_OS_WIN32
 262   error = WSAStartup (0x0202, &w);
 263
 264   if (error)
 265     goto startup_error;
 266
 267   if (w.wVersion != 0x0202)
 268     goto version_error;
 269 #endif
 270
 271   newconn = g_new0 (GstRTSPConnection, 1);
 272
 273   if ((newconn->fdset = gst_poll_new (TRUE)) == NULL)
 274     goto no_fdset;
 275
 276   newconn->url = gst_rtsp_url_copy (url);
 277   newconn->fd0.fd = -1;
 278   newconn->fd1.fd = -1;
 279   newconn->timer = g_timer_new ();
 280   newconn->timeout = 60;
 281   newconn->cseq = 1;
 282
 283   newconn->auth_method = GST_RTSP_AUTH_NONE;
 284   newconn->username = NULL;
 285   newconn->passwd = NULL;
 286   newconn->auth_params = NULL;
 287
 288   *conn = newconn;
 289
 290   return GST_RTSP_OK;
 291
 292   /* ERRORS */
 293 #ifdef G_OS_WIN32
 294 startup_error:
 295   {
 296     g_warning ("Error %d on WSAStartup", error);
 297     return GST_RTSP_EWSASTART;
 298   }
 299 version_error:
 300   {
 301     g_warning ("Windows sockets are not version 0x202 (current 0x%x)",
 302         w.wVersion);
 303     WSACleanup ();
 304     return GST_RTSP_EWSAVERSION;
 305   }
 306 #endif
 307 no_fdset:
 308   {
 309     g_free (newconn);
 310 #ifdef G_OS_WIN32
 311     WSACleanup ();
 312 #endif
 313     return GST_RTSP_ESYS;
 314   }
 315 }
 316
 317 /**
 318  * gst_rtsp_connection_accept:
 319  * @sock: a socket
 320  * @conn: storage for a #GstRTSPConnection
 321  *
 322  * Accept a new connection on @sock and create a new #GstRTSPConnection for
 323  * handling communication on new socket.
 324  *
 325  * Returns: #GST_RTSP_OK when @conn contains a valid connection.
 326  *
 327  * Since: 0.10.23
 328  */
 329 GstRTSPResult
 330 gst_rtsp_connection_accept (gint sock, GstRTSPConnection ** conn)
 331 {
 332   int fd;
 333   GstRTSPConnection *newconn = NULL;
 334   union gst_sockaddr sa;
 335   socklen_t slen = sizeof (sa);
 336   gchar ip[INET6_ADDRSTRLEN];
 337   GstRTSPUrl *url;
 338 #ifdef G_OS_WIN32
 339   gulong flags = 1;
 340 #endif
 341
 342   memset (&sa, 0, slen);
 343
 344 #ifndef G_OS_WIN32
 345   fd = accept (sock, &sa.sa, &slen);
 346 #else
 347   fd = accept (sock, &sa.sa, (gint *) & slen);
 348 #endif /* G_OS_WIN32 */
 349   if (fd == -1)
 350     goto accept_failed;
 351
 352   if (getnameinfo (&sa.sa, slen, ip, sizeof (ip), NULL, 0, NI_NUMERICHOST) != 0)
 353     goto getnameinfo_failed;
 354   if (sa.sa.sa_family != AF_INET && sa.sa.sa_family != AF_INET6)
 355     goto wrong_family;
 356
 357   /* set to non-blocking mode so that we can cancel the communication */
 358 #ifndef G_OS_WIN32
 359   fcntl (fd, F_SETFL, O_NONBLOCK);
 360 #else
 361   ioctlsocket (fd, FIONBIO, &flags);
 362 #endif /* G_OS_WIN32 */
 363
 364   /* create a url for the client address */
 365   url = g_new0 (GstRTSPUrl, 1);
 366   url->host = g_strdup (ip);
 367   if (sa.sa.sa_family == AF_INET)
 368     url->port = sa.sa_in.sin_port;
 369   else
 370     url->port = sa.sa_in6.sin6_port;
 371
 372   /* now create the connection object */
 373   gst_rtsp_connection_create (url, &newconn);
 374   gst_rtsp_url_free (url);
 375
 376   ADD_POLLFD (newconn->fdset, &newconn->fd0, fd);
 377
 378   /* both read and write initially */
 379   newconn->readfd = &newconn->fd0;
 380   newconn->writefd = &newconn->fd0;
 381
 382   *conn = newconn;
 383
 384   return GST_RTSP_OK;
 385
 386   /* ERRORS */
 387 accept_failed:
 388   {
 389     return GST_RTSP_ESYS;
 390   }
 391 getnameinfo_failed:
 392 wrong_family:
 393   {
 394     close (fd);
 395     return GST_RTSP_ERROR;
 396   }
 397 }
 398
 399 static gchar *
 400 do_resolve (const gchar * host)
 401 {
 402   static gchar ip[INET6_ADDRSTRLEN];
 403   struct addrinfo *aires;
 404   struct addrinfo *ai;
 405   gint aierr;
 406
 407   aierr = getaddrinfo (host, NULL, NULL, &aires);
 408   if (aierr != 0)
 409     goto no_addrinfo;
 410
 411   for (ai = aires; ai; ai = ai->ai_next) {
 412     if (ai->ai_family == AF_INET || ai->ai_family == AF_INET6) {
 413       break;
 414     }
 415   }
 416   if (ai == NULL)
 417     goto no_family;
 418
 419   aierr = getnameinfo (ai->ai_addr, ai->ai_addrlen, ip, sizeof (ip), NULL, 0,
 420       NI_NUMERICHOST | NI_NUMERICSERV);
 421   if (aierr != 0)
 422     goto no_address;
 423
 424   freeaddrinfo (aires);
 425
 426   return g_strdup (ip);
 427
 428   /* ERRORS */
 429 no_addrinfo:
 430   {
 431     GST_ERROR ("no addrinfo found for %s: %s", host, gai_strerror (aierr));
 432     return NULL;
 433   }
 434 no_family:
 435   {
 436     GST_ERROR ("no family found for %s", host);
 437     freeaddrinfo (aires);
 438     return NULL;
 439   }
 440 no_address:
 441   {
 442     GST_ERROR ("no address found for %s: %s", host, gai_strerror (aierr));
 443     freeaddrinfo (aires);
 444     return NULL;
 445   }
 446 }
 447
 448 static GstRTSPResult
 449 do_connect (const gchar * ip, guint16 port, GstPollFD * fdout,
 450     GstPoll * fdset, GTimeVal * timeout)
 451 {
 452   gint fd;
 453   struct addrinfo hints;
 454   struct addrinfo *aires;
 455   struct addrinfo *ai;
 456   gint aierr;
 457   gchar service[NI_MAXSERV];
 458   gint ret;
 459 #ifdef G_OS_WIN32
 460   unsigned long flags = 1;
 461 #endif /* G_OS_WIN32 */
 462   GstClockTime to;
 463   gint retval;
 464
 465   memset (&hints, 0, sizeof hints);
 466   hints.ai_flags = AI_NUMERICHOST;
 467   hints.ai_family = AF_UNSPEC;
 468   hints.ai_socktype = SOCK_STREAM;
 469   g_snprintf (service, sizeof (service) - 1, "%hu", port);
 470   service[sizeof (service) - 1] = '\0';
 471
 472   aierr = getaddrinfo (ip, service, &hints, &aires);
 473   if (aierr != 0)
 474     goto no_addrinfo;
 475
 476   for (ai = aires; ai; ai = ai->ai_next) {
 477     if (ai->ai_family == AF_INET || ai->ai_family == AF_INET6) {
 478       break;
 479     }
 480   }
 481   if (ai == NULL)
 482     goto no_family;
 483
 484   fd = socket (ai->ai_family, SOCK_STREAM, 0);
 485   if (fd == -1)
 486     goto no_socket;
 487
 488   /* set to non-blocking mode so that we can cancel the connect */
 489 #ifndef G_OS_WIN32
 490   fcntl (fd, F_SETFL, O_NONBLOCK);
 491 #else
 492   ioctlsocket (fd, FIONBIO, &flags);
 493 #endif /* G_OS_WIN32 */
 494
 495   /* add the socket to our fdset */
 496   ADD_POLLFD (fdset, fdout, fd);
 497
 498   /* we are going to connect ASYNC now */
 499   ret = connect (fd, ai->ai_addr, ai->ai_addrlen);
 500   if (ret == 0)
 501     goto done;
 502   if (!ERRNO_IS_EINPROGRESS)
 503     goto sys_error;
 504
 505   /* wait for connect to complete up to the specified timeout or until we got
 506    * interrupted. */
 507   gst_poll_fd_ctl_write (fdset, fdout, TRUE);
 508
 509   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : GST_CLOCK_TIME_NONE;
 510
 511   do {
 512     retval = gst_poll_wait (fdset, to);
 513   } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
 514
 515   if (retval == 0)
 516     goto timeout;
 517   else if (retval == -1)
 518     goto sys_error;
 519
 520   /* we can still have an error connecting on windows */
 521   if (gst_poll_fd_has_error (fdset, fdout)) {
 522     socklen_t len = sizeof (errno);
 523 #ifndef G_OS_WIN32
 524     getsockopt (fd, SOL_SOCKET, SO_ERROR, &errno, &len);
 525 #else
 526     getsockopt (fd, SOL_SOCKET, SO_ERROR, (char *) &errno, &len);
 527 #endif
 528     goto sys_error;
 529   }
 530
 531   gst_poll_fd_ignored (fdset, fdout);
 532
 533 done:
 534   freeaddrinfo (aires);
 535
 536   return GST_RTSP_OK;
 537
 538   /* ERRORS */
 539 no_addrinfo:
 540   {
 541     GST_ERROR ("no addrinfo found for %s: %s", ip, gai_strerror (aierr));
 542     return GST_RTSP_ERROR;
 543   }
 544 no_family:
 545   {
 546     GST_ERROR ("no family found for %s", ip);
 547     freeaddrinfo (aires);
 548     return GST_RTSP_ERROR;
 549   }
 550 no_socket:
 551   {
 552     GST_ERROR ("no socket %d (%s)", errno, g_strerror (errno));
 553     freeaddrinfo (aires);
 554     return GST_RTSP_ESYS;
 555   }
 556 sys_error:
 557   {
 558     GST_ERROR ("system error %d (%s)", errno, g_strerror (errno));
 559     REMOVE_POLLFD (fdset, fdout);
 560     freeaddrinfo (aires);
 561     return GST_RTSP_ESYS;
 562   }
 563 timeout:
 564   {
 565     GST_ERROR ("timeout");
 566     REMOVE_POLLFD (fdset, fdout);
 567     freeaddrinfo (aires);
 568     return GST_RTSP_ETIMEOUT;
 569   }
 570 }
 571
 572 static GstRTSPResult
 573 setup_tunneling (GstRTSPConnection * conn, GTimeVal * timeout)
 574 {
 575   gint i;
 576   GstRTSPResult res;
 577   gchar *str;
 578   guint idx, line;
 579   gint retval;
 580   GstClockTime to;
 581   gchar *ip, *url_port_str;
 582   guint16 port, url_port;
 583   gchar codestr[4], *resultstr;
 584   gint code;
 585   GstRTSPUrl *url;
 586   gchar *hostparam;
 587
 588   /* create a random sessionid */
 589   for (i = 0; i < TUNNELID_LEN; i++)
 590     conn->tunnelid[i] = g_random_int_range ('a', 'z');
 591   conn->tunnelid[TUNNELID_LEN - 1] = '\0';
 592
 593   url = conn->url;
 594   /* get the port from the url */
 595   gst_rtsp_url_get_port (url, &url_port);
 596
 597   if (conn->proxy_host) {
 598     hostparam = g_strdup_printf ("Host: %s:%d\r\n", url->host, url_port);
 599     url_port_str = g_strdup_printf (":%d", url_port);
 600     ip = conn->proxy_host;
 601     port = conn->proxy_port;
 602   } else {
 603     hostparam = NULL;
 604     url_port_str = NULL;
 605     ip = conn->ip;
 606     port = url_port;
 607   }
 608
 609   /* */
 610   str = g_strdup_printf ("GET %s%s%s%s%s%s HTTP/1.0\r\n"
 611       "%s"
 612       "x-sessioncookie: %s\r\n"
 613       "Accept: application/x-rtsp-tunnelled\r\n"
 614       "Pragma: no-cache\r\n"
 615       "Cache-Control: no-cache\r\n" "\r\n",
 616       conn->proxy_host ? "http://" : "",
 617       conn->proxy_host ? url->host : "",
 618       conn->proxy_host ? url_port_str : "",
 619       url->abspath, url->query ? "?" : "", url->query ? url->query : "",
 620       hostparam ? hostparam : "", conn->tunnelid);
 621
 622   /* we start by writing to this fd */
 623   conn->writefd = &conn->fd0;
 624
 625   res = gst_rtsp_connection_write (conn, (guint8 *) str, strlen (str), timeout);
 626   g_free (str);
 627   if (res != GST_RTSP_OK)
 628     goto write_failed;
 629
 630   gst_poll_fd_ctl_write (conn->fdset, &conn->fd0, FALSE);
 631   gst_poll_fd_ctl_read (conn->fdset, &conn->fd0, TRUE);
 632
 633   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : GST_CLOCK_TIME_NONE;
 634
 635   line = 0;
 636   while (TRUE) {
 637     guint8 buffer[4096];
 638
 639     idx = 0;
 640     while (TRUE) {
 641       res = read_line (conn->fd0.fd, buffer, &idx, sizeof (buffer), NULL);
 642       if (res == GST_RTSP_EEOF)
 643         goto eof;
 644       if (res == GST_RTSP_OK)
 645         break;
 646       if (res != GST_RTSP_EINTR)
 647         goto read_error;
 648
 649       do {
 650         retval = gst_poll_wait (conn->fdset, to);
 651       } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
 652
 653       /* check for timeout */
 654       if (retval == 0)
 655         goto timeout;
 656
 657       if (retval == -1) {
 658         if (errno == EBUSY)
 659           goto stopped;
 660         else
 661           goto select_error;
 662       }
 663     }
 664
 665     /* check for last line */
 666     if (buffer[0] == '\r')
 667       buffer[0] = '\0';
 668     if (buffer[0] == '\0')
 669       break;
 670
 671     if (line == 0) {
 672       /* first line, parse response */
 673       gchar versionstr[20];
 674       gchar *bptr;
 675
 676       bptr = (gchar *) buffer;
 677
 678       parse_string (versionstr, sizeof (versionstr), &bptr);
 679       parse_string (codestr, sizeof (codestr), &bptr);
 680       code = atoi (codestr);
 681
 682       while (g_ascii_isspace (*bptr))
 683         bptr++;
 684
 685       resultstr = bptr;
 686
 687       if (code != GST_RTSP_STS_OK)
 688         goto wrong_result;
 689     } else {
 690       gchar key[32];
 691       gchar *value;
 692
 693       /* other lines, parse key/value */
 694       res = parse_key_value (buffer, key, sizeof (key), &value);
 695       if (res == GST_RTSP_OK) {
 696         /* we got a new ip address */
 697         if (g_ascii_strcasecmp (key, "x-server-ip-address") == 0) {
 698           if (conn->proxy_host) {
 699             /* if we use a proxy we need to change the destination url */
 700             g_free (url->host);
 701             url->host = g_strdup (value);
 702             g_free (hostparam);
 703             g_free (url_port_str);
 704             hostparam =
 705                 g_strdup_printf ("Host: %s:%d\r\n", url->host, url_port);
 706             url_port_str = g_strdup_printf (":%d", url_port);
 707           } else {
 708             /* and resolve the new ip address */
 709             if (!(ip = do_resolve (conn->ip)))
 710               goto not_resolved;
 711             g_free (conn->ip);
 712             conn->ip = ip;
 713           }
 714         }
 715       }
 716     }
 717     line++;
 718   }
 719
 720   /* connect to the host/port */
 721   res = do_connect (ip, port, &conn->fd1, conn->fdset, timeout);
 722   if (res != GST_RTSP_OK)
 723     goto connect_failed;
 724
 725   /* this is now our writing socket */
 726   conn->writefd = &conn->fd1;
 727
 728   /* */
 729   str = g_strdup_printf ("POST %s%s%s%s%s%s HTTP/1.0\r\n"
 730       "%s"
 731       "x-sessioncookie: %s\r\n"
 732       "Content-Type: application/x-rtsp-tunnelled\r\n"
 733       "Pragma: no-cache\r\n"
 734       "Cache-Control: no-cache\r\n"
 735       "Content-Length: 32767\r\n"
 736       "Expires: Sun, 9 Jan 1972 00:00:00 GMT\r\n"
 737       "\r\n",
 738       conn->proxy_host ? "http://" : "",
 739       conn->proxy_host ? url->host : "",
 740       conn->proxy_host ? url_port_str : "",
 741       url->abspath, url->query ? "?" : "", url->query ? url->query : "",
 742       hostparam ? hostparam : "", conn->tunnelid);
 743
 744   res = gst_rtsp_connection_write (conn, (guint8 *) str, strlen (str), timeout);
 745   g_free (str);
 746   if (res != GST_RTSP_OK)
 747     goto write_failed;
 748
 749 exit:
 750   g_free (hostparam);
 751   g_free (url_port_str);
 752
 753   return res;
 754
 755   /* ERRORS */
 756 write_failed:
 757   {
 758     GST_ERROR ("write failed (%d)", res);
 759     goto exit;
 760   }
 761 eof:
 762   {
 763     res = GST_RTSP_EEOF;
 764     goto exit;
 765   }
 766 read_error:
 767   {
 768     goto exit;
 769   }
 770 timeout:
 771   {
 772     res = GST_RTSP_ETIMEOUT;
 773     goto exit;
 774   }
 775 select_error:
 776   {
 777     res = GST_RTSP_ESYS;
 778     goto exit;
 779   }
 780 stopped:
 781   {
 782     res = GST_RTSP_EINTR;
 783     goto exit;
 784   }
 785 wrong_result:
 786   {
 787     GST_ERROR ("got failure response %d %s", code, resultstr);
 788     res = GST_RTSP_ERROR;
 789     goto exit;
 790   }
 791 not_resolved:
 792   {
 793     GST_ERROR ("could not resolve %s", conn->ip);
 794     res = GST_RTSP_ENET;
 795     goto exit;
 796   }
 797 connect_failed:
 798   {
 799     GST_ERROR ("failed to connect");
 800     goto exit;
 801   }
 802 }
 803
 804 /**
 805  * gst_rtsp_connection_connect:
 806  * @conn: a #GstRTSPConnection
 807  * @timeout: a #GTimeVal timeout
 808  *
 809  * Attempt to connect to the url of @conn made with
 810  * gst_rtsp_connection_create(). If @timeout is #NULL this function can block
 811  * forever. If @timeout contains a valid timeout, this function will return
 812  * #GST_RTSP_ETIMEOUT after the timeout expired.
 813  *
 814  * This function can be cancelled with gst_rtsp_connection_flush().
 815  *
 816  * Returns: #GST_RTSP_OK when a connection could be made.
 817  */
 818 GstRTSPResult
 819 gst_rtsp_connection_connect (GstRTSPConnection * conn, GTimeVal * timeout)
 820 {
 821   GstRTSPResult res;
 822   gchar *ip;
 823   guint16 port;
 824   GstRTSPUrl *url;
 825
 826   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
 827   g_return_val_if_fail (conn->url != NULL, GST_RTSP_EINVAL);
 828   g_return_val_if_fail (conn->fd0.fd < 0, GST_RTSP_EINVAL);
 829
 830   url = conn->url;
 831
 832   if (conn->proxy_host && conn->tunneled) {
 833     if (!(ip = do_resolve (conn->proxy_host))) {
 834       GST_ERROR ("could not resolve %s", conn->proxy_host);
 835       goto not_resolved;
 836     }
 837     port = conn->proxy_port;
 838     g_free (conn->proxy_host);
 839     conn->proxy_host = ip;
 840   } else {
 841     if (!(ip = do_resolve (url->host))) {
 842       GST_ERROR ("could not resolve %s", url->host);
 843       goto not_resolved;
 844     }
 845     /* get the port from the url */
 846     gst_rtsp_url_get_port (url, &port);
 847
 848     g_free (conn->ip);
 849     conn->ip = ip;
 850   }
 851
 852   /* connect to the host/port */
 853   res = do_connect (ip, port, &conn->fd0, conn->fdset, timeout);
 854   if (res != GST_RTSP_OK)
 855     goto connect_failed;
 856
 857   /* this is our read URL */
 858   conn->readfd = &conn->fd0;
 859
 860   if (conn->tunneled) {
 861     res = setup_tunneling (conn, timeout);
 862     if (res != GST_RTSP_OK)
 863       goto tunneling_failed;
 864   } else {
 865     conn->writefd = &conn->fd0;
 866   }
 867
 868   return GST_RTSP_OK;
 869
 870 not_resolved:
 871   {
 872     return GST_RTSP_ENET;
 873   }
 874 connect_failed:
 875   {
 876     GST_ERROR ("failed to connect");
 877     return res;
 878   }
 879 tunneling_failed:
 880   {
 881     GST_ERROR ("failed to setup tunneling");
 882     return res;
 883   }
 884 }
 885
 886 static void
 887 auth_digest_compute_hex_urp (const gchar * username,
 888     const gchar * realm, const gchar * password, gchar hex_urp[33])
 889 {
 890   GChecksum *md5_context = g_checksum_new (G_CHECKSUM_MD5);
 891   const gchar *digest_string;
 892
 893   g_checksum_update (md5_context, (const guchar *) username, strlen (username));
 894   g_checksum_update (md5_context, (const guchar *) ":", 1);
 895   g_checksum_update (md5_context, (const guchar *) realm, strlen (realm));
 896   g_checksum_update (md5_context, (const guchar *) ":", 1);
 897   g_checksum_update (md5_context, (const guchar *) password, strlen (password));
 898   digest_string = g_checksum_get_string (md5_context);
 899
 900   memset (hex_urp, 0, 33);
 901   memcpy (hex_urp, digest_string, strlen (digest_string));
 902
 903   g_checksum_free (md5_context);
 904 }
 905
 906 static void
 907 auth_digest_compute_response (const gchar * method,
 908     const gchar * uri, const gchar * hex_a1, const gchar * nonce,
 909     gchar response[33])
 910 {
 911   char hex_a2[33] = { 0, };
 912   GChecksum *md5_context = g_checksum_new (G_CHECKSUM_MD5);
 913   const gchar *digest_string;
 914
 915   /* compute A2 */
 916   g_checksum_update (md5_context, (const guchar *) method, strlen (method));
 917   g_checksum_update (md5_context, (const guchar *) ":", 1);
 918   g_checksum_update (md5_context, (const guchar *) uri, strlen (uri));
 919   digest_string = g_checksum_get_string (md5_context);
 920   memcpy (hex_a2, digest_string, strlen (digest_string));
 921
 922   /* compute KD */
 923 #if GLIB_CHECK_VERSION (2, 18, 0)
 924   g_checksum_reset (md5_context);
 925 #else
 926   g_checksum_free (md5_context);
 927   md5_context = g_checksum_new (G_CHECKSUM_MD5);
 928 #endif
 929   g_checksum_update (md5_context, (const guchar *) hex_a1, strlen (hex_a1));
 930   g_checksum_update (md5_context, (const guchar *) ":", 1);
 931   g_checksum_update (md5_context, (const guchar *) nonce, strlen (nonce));
 932   g_checksum_update (md5_context, (const guchar *) ":", 1);
 933
 934   g_checksum_update (md5_context, (const guchar *) hex_a2, 32);
 935   digest_string = g_checksum_get_string (md5_context);
 936   memset (response, 0, 33);
 937   memcpy (response, digest_string, strlen (digest_string));
 938
 939   g_checksum_free (md5_context);
 940 }
 941
 942 static void
 943 add_auth_header (GstRTSPConnection * conn, GstRTSPMessage * message)
 944 {
 945   switch (conn->auth_method) {
 946     case GST_RTSP_AUTH_BASIC:{
 947       gchar *user_pass;
 948       gchar *user_pass64;
 949       gchar *auth_string;
 950
 951       user_pass = g_strdup_printf ("%s:%s", conn->username, conn->passwd);
 952       user_pass64 = g_base64_encode ((guchar *) user_pass, strlen (user_pass));
 953       auth_string = g_strdup_printf ("Basic %s", user_pass64);
 954
 955       gst_rtsp_message_take_header (message, GST_RTSP_HDR_AUTHORIZATION,
 956           auth_string);
 957
 958       g_free (user_pass);
 959       g_free (user_pass64);
 960       break;
 961     }
 962     case GST_RTSP_AUTH_DIGEST:{
 963       gchar response[33], hex_urp[33];
 964       gchar *auth_string, *auth_string2;
 965       gchar *realm;
 966       gchar *nonce;
 967       gchar *opaque;
 968       const gchar *uri;
 969       const gchar *method;
 970
 971       /* we need to have some params set */
 972       if (conn->auth_params == NULL)
 973         break;
 974
 975       /* we need the realm and nonce */
 976       realm = (gchar *) g_hash_table_lookup (conn->auth_params, "realm");
 977       nonce = (gchar *) g_hash_table_lookup (conn->auth_params, "nonce");
 978       if (realm == NULL || nonce == NULL)
 979         break;
 980
 981       auth_digest_compute_hex_urp (conn->username, realm, conn->passwd,
 982           hex_urp);
 983
 984       method = gst_rtsp_method_as_text (message->type_data.request.method);
 985       uri = message->type_data.request.uri;
 986
 987       /* Assume no qop, algorithm=md5, stale=false */
 988       /* For algorithm MD5, a1 = urp. */
 989       auth_digest_compute_response (method, uri, hex_urp, nonce, response);
 990       auth_string = g_strdup_printf ("Digest username=\"%s\", "
 991           "realm=\"%s\", nonce=\"%s\", uri=\"%s\", response=\"%s\"",
 992           conn->username, realm, nonce, uri, response);
 993
 994       opaque = (gchar *) g_hash_table_lookup (conn->auth_params, "opaque");
 995       if (opaque) {
 996         auth_string2 = g_strdup_printf ("%s, opaque=\"%s\"", auth_string,
 997             opaque);
 998         g_free (auth_string);
 999         auth_string = auth_string2;
1000       }
1001       gst_rtsp_message_take_header (message, GST_RTSP_HDR_AUTHORIZATION,
1002           auth_string);
1003       break;
1004     }
1005     default:
1006       /* Nothing to do */
1007       break;
1008   }
1009 }
1010
1011 static void
1012 gen_date_string (gchar * date_string, guint len)
1013 {
1014   GTimeVal tv;
1015   time_t t;
1016 #ifdef HAVE_GMTIME_R
1017   struct tm tm_;
1018 #endif
1019
1020   g_get_current_time (&tv);
1021   t = (time_t) tv.tv_sec;
1022
1023 #ifdef HAVE_GMTIME_R
1024   strftime (date_string, len, "%a, %d %b %Y %H:%M:%S GMT", gmtime_r (&t, &tm_));
1025 #else
1026   strftime (date_string, len, "%a, %d %b %Y %H:%M:%S GMT", gmtime (&t));
1027 #endif
1028 }
1029
1030 static GstRTSPResult
1031 write_bytes (gint fd, const guint8 * buffer, guint * idx, guint size)
1032 {
1033   guint left;
1034
1035   if (G_UNLIKELY (*idx > size))
1036     return GST_RTSP_ERROR;
1037
1038   left = size - *idx;
1039
1040   while (left) {
1041     gint r;
1042
1043     r = WRITE_SOCKET (fd, &buffer[*idx], left);
1044     if (G_UNLIKELY (r == 0)) {
1045       return GST_RTSP_EINTR;
1046     } else if (G_UNLIKELY (r < 0)) {
1047       if (ERRNO_IS_EAGAIN)
1048         return GST_RTSP_EINTR;
1049       if (!ERRNO_IS_EINTR)
1050         return GST_RTSP_ESYS;
1051     } else {
1052       left -= r;
1053       *idx += r;
1054     }
1055   }
1056   return GST_RTSP_OK;
1057 }
1058
1059 static gint
1060 fill_bytes (gint fd, guint8 * buffer, guint size, DecodeCtx * ctx)
1061 {
1062   gint out = 0;
1063
1064   if (ctx) {
1065     while (size > 0) {
1066       guint8 in[sizeof (ctx->out) * 4 / 3];
1067       gint r;
1068
1069       while (size > 0 && ctx->cout < ctx->coutl) {
1070         /* we have some leftover bytes */
1071         *buffer++ = ctx->out[ctx->cout++];
1072         size--;
1073         out++;
1074       }
1075
1076       /* got what we needed? */
1077       if (size == 0)
1078         break;
1079
1080       /* try to read more bytes */
1081       r = READ_SOCKET (fd, in, sizeof (in));
1082       if (r <= 0) {
1083         if (out == 0)
1084           out = r;
1085         break;
1086       }
1087
1088       ctx->cout = 0;
1089       ctx->coutl =
1090           g_base64_decode_step ((gchar *) in, r, ctx->out, &ctx->state,
1091           &ctx->save);
1092     }
1093   } else {
1094     out = READ_SOCKET (fd, buffer, size);
1095   }
1096
1097   return out;
1098 }
1099
1100 static GstRTSPResult
1101 read_bytes (gint fd, guint8 * buffer, guint * idx, guint size, DecodeCtx * ctx)
1102 {
1103   guint left;
1104
1105   if (G_UNLIKELY (*idx > size))
1106     return GST_RTSP_ERROR;
1107
1108   left = size - *idx;
1109
1110   while (left) {
1111     gint r;
1112
1113     r = fill_bytes (fd, &buffer[*idx], left, ctx);
1114     if (G_UNLIKELY (r == 0)) {
1115       return GST_RTSP_EEOF;
1116     } else if (G_UNLIKELY (r < 0)) {
1117       if (ERRNO_IS_EAGAIN)
1118         return GST_RTSP_EINTR;
1119       if (!ERRNO_IS_EINTR)
1120         return GST_RTSP_ESYS;
1121     } else {
1122       left -= r;
1123       *idx += r;
1124     }
1125   }
1126   return GST_RTSP_OK;
1127 }
1128
1129 static GstRTSPResult
1130 read_line (gint fd, guint8 * buffer, guint * idx, guint size, DecodeCtx * ctx)
1131 {
1132   while (TRUE) {
1133     guint8 c;
1134     gint r;
1135
1136     r = fill_bytes (fd, &c, 1, ctx);
1137     if (G_UNLIKELY (r == 0)) {
1138       return GST_RTSP_EEOF;
1139     } else if (G_UNLIKELY (r < 0)) {
1140       if (ERRNO_IS_EAGAIN)
1141         return GST_RTSP_EINTR;
1142       if (!ERRNO_IS_EINTR)
1143         return GST_RTSP_ESYS;
1144     } else {
1145       if (c == '\n')            /* end on \n */
1146         break;
1147       if (c == '\r')            /* ignore \r */
1148         continue;
1149
1150       if (G_LIKELY (*idx < size - 1))
1151         buffer[(*idx)++] = c;
1152     }
1153   }
1154   buffer[*idx] = '\0';
1155
1156   return GST_RTSP_OK;
1157 }
1158
1159 /**
1160  * gst_rtsp_connection_write:
1161  * @conn: a #GstRTSPConnection
1162  * @data: the data to write
1163  * @size: the size of @data
1164  * @timeout: a timeout value or #NULL
1165  *
1166  * Attempt to write @size bytes of @data to the connected @conn, blocking up to
1167  * the specified @timeout. @timeout can be #NULL, in which case this function
1168  * might block forever.
1169  *
1170  * This function can be cancelled with gst_rtsp_connection_flush().
1171  *
1172  * Returns: #GST_RTSP_OK on success.
1173  */
1174 GstRTSPResult
1175 gst_rtsp_connection_write (GstRTSPConnection * conn, const guint8 * data,
1176     guint size, GTimeVal * timeout)
1177 {
1178   guint offset;
1179   gint retval;
1180   GstClockTime to;
1181   GstRTSPResult res;
1182
1183   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
1184   g_return_val_if_fail (data != NULL || size == 0, GST_RTSP_EINVAL);
1185   g_return_val_if_fail (conn->writefd != NULL, GST_RTSP_EINVAL);
1186
1187   gst_poll_set_controllable (conn->fdset, TRUE);
1188   gst_poll_fd_ctl_write (conn->fdset, conn->writefd, TRUE);
1189   gst_poll_fd_ctl_read (conn->fdset, conn->readfd, FALSE);
1190   /* clear all previous poll results */
1191   gst_poll_fd_ignored (conn->fdset, conn->writefd);
1192   gst_poll_fd_ignored (conn->fdset, conn->readfd);
1193
1194   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : GST_CLOCK_TIME_NONE;
1195
1196   offset = 0;
1197
1198   while (TRUE) {
1199     /* try to write */
1200     res = write_bytes (conn->writefd->fd, data, &offset, size);
1201     if (G_LIKELY (res == GST_RTSP_OK))
1202       break;
1203     if (G_UNLIKELY (res != GST_RTSP_EINTR))
1204       goto write_error;
1205
1206     /* not all is written, wait until we can write more */
1207     do {
1208       retval = gst_poll_wait (conn->fdset, to);
1209     } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
1210
1211     if (G_UNLIKELY (retval == 0))
1212       goto timeout;
1213
1214     if (G_UNLIKELY (retval == -1)) {
1215       if (errno == EBUSY)
1216         goto stopped;
1217       else
1218         goto select_error;
1219     }
1220   }
1221   return GST_RTSP_OK;
1222
1223   /* ERRORS */
1224 timeout:
1225   {
1226     return GST_RTSP_ETIMEOUT;
1227   }
1228 select_error:
1229   {
1230     return GST_RTSP_ESYS;
1231   }
1232 stopped:
1233   {
1234     return GST_RTSP_EINTR;
1235   }
1236 write_error:
1237   {
1238     return res;
1239   }
1240 }
1241
1242 static GString *
1243 message_to_string (GstRTSPConnection * conn, GstRTSPMessage * message)
1244 {
1245   GString *str = NULL;
1246
1247   str = g_string_new ("");
1248
1249   switch (message->type) {
1250     case GST_RTSP_MESSAGE_REQUEST:
1251       /* create request string, add CSeq */
1252       g_string_append_printf (str, "%s %s RTSP/1.0\r\n"
1253           "CSeq: %d\r\n",
1254           gst_rtsp_method_as_text (message->type_data.request.method),
1255           message->type_data.request.uri, conn->cseq++);
1256       /* add session id if we have one */
1257       if (conn->session_id[0] != '\0') {
1258         gst_rtsp_message_remove_header (message, GST_RTSP_HDR_SESSION, -1);
1259         gst_rtsp_message_add_header (message, GST_RTSP_HDR_SESSION,
1260             conn->session_id);
1261       }
1262       /* add any authentication headers */
1263       add_auth_header (conn, message);
1264       break;
1265     case GST_RTSP_MESSAGE_RESPONSE:
1266       /* create response string */
1267       g_string_append_printf (str, "RTSP/1.0 %d %s\r\n",
1268           message->type_data.response.code, message->type_data.response.reason);
1269       break;
1270     case GST_RTSP_MESSAGE_DATA:
1271     {
1272       guint8 data_header[4];
1273
1274       /* prepare data header */
1275       data_header[0] = '$';
1276       data_header[1] = message->type_data.data.channel;
1277       data_header[2] = (message->body_size >> 8) & 0xff;
1278       data_header[3] = message->body_size & 0xff;
1279
1280       /* create string with header and data */
1281       str = g_string_append_len (str, (gchar *) data_header, 4);
1282       str =
1283           g_string_append_len (str, (gchar *) message->body,
1284           message->body_size);
1285       break;
1286     }
1287     default:
1288       g_string_free (str, TRUE);
1289       g_return_val_if_reached (NULL);
1290       break;
1291   }
1292
1293   /* append headers and body */
1294   if (message->type != GST_RTSP_MESSAGE_DATA) {
1295     gchar date_string[100];
1296
1297     gen_date_string (date_string, sizeof (date_string));
1298
1299     /* add date header */
1300     gst_rtsp_message_remove_header (message, GST_RTSP_HDR_DATE, -1);
1301     gst_rtsp_message_add_header (message, GST_RTSP_HDR_DATE, date_string);
1302
1303     /* append headers */
1304     gst_rtsp_message_append_headers (message, str);
1305
1306     /* append Content-Length and body if needed */
1307     if (message->body != NULL && message->body_size > 0) {
1308       gchar *len;
1309
1310       len = g_strdup_printf ("%d", message->body_size);
1311       g_string_append_printf (str, "%s: %s\r\n",
1312           gst_rtsp_header_as_text (GST_RTSP_HDR_CONTENT_LENGTH), len);
1313       g_free (len);
1314       /* header ends here */
1315       g_string_append (str, "\r\n");
1316       str =
1317           g_string_append_len (str, (gchar *) message->body,
1318           message->body_size);
1319     } else {
1320       /* just end headers */
1321       g_string_append (str, "\r\n");
1322     }
1323   }
1324
1325   return str;
1326 }
1327
1328 /**
1329  * gst_rtsp_connection_send:
1330  * @conn: a #GstRTSPConnection
1331  * @message: the message to send
1332  * @timeout: a timeout value or #NULL
1333  *
1334  * Attempt to send @message to the connected @conn, blocking up to
1335  * the specified @timeout. @timeout can be #NULL, in which case this function
1336  * might block forever.
1337  *
1338  * This function can be cancelled with gst_rtsp_connection_flush().
1339  *
1340  * Returns: #GST_RTSP_OK on success.
1341  */
1342 GstRTSPResult
1343 gst_rtsp_connection_send (GstRTSPConnection * conn, GstRTSPMessage * message,
1344     GTimeVal * timeout)
1345 {
1346   GString *string = NULL;
1347   GstRTSPResult res;
1348   gchar *str;
1349   gsize len;
1350
1351   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
1352   g_return_val_if_fail (message != NULL, GST_RTSP_EINVAL);
1353
1354   if (G_UNLIKELY (!(string = message_to_string (conn, message))))
1355     goto no_message;
1356
1357   if (conn->tunneled) {
1358     str = g_base64_encode ((const guchar *) string->str, string->len);
1359     g_string_free (string, TRUE);
1360     len = strlen (str);
1361   } else {
1362     str = string->str;
1363     len = string->len;
1364     g_string_free (string, FALSE);
1365   }
1366
1367   /* write request */
1368   res = gst_rtsp_connection_write (conn, (guint8 *) str, len, timeout);
1369
1370   g_free (str);
1371
1372   return res;
1373
1374 no_message:
1375   {
1376     g_warning ("Wrong message");
1377     return GST_RTSP_EINVAL;
1378   }
1379 }
1380
1381 static void
1382 parse_string (gchar * dest, gint size, gchar ** src)
1383 {
1384   gint idx;
1385
1386   idx = 0;
1387   /* skip spaces */
1388   while (g_ascii_isspace (**src))
1389     (*src)++;
1390
1391   while (!g_ascii_isspace (**src) && **src != '\0') {
1392     if (idx < size - 1)
1393       dest[idx++] = **src;
1394     (*src)++;
1395   }
1396   if (size > 0)
1397     dest[idx] = '\0';
1398 }
1399
1400 static void
1401 parse_key (gchar * dest, gint size, gchar ** src)
1402 {
1403   gint idx;
1404
1405   idx = 0;
1406   while (**src != ':' && **src != '\0') {
1407     if (idx < size - 1)
1408       dest[idx++] = **src;
1409     (*src)++;
1410   }
1411   if (size > 0)
1412     dest[idx] = '\0';
1413 }
1414
1415 static GstRTSPResult
1416 parse_response_status (guint8 * buffer, GstRTSPMessage * msg)
1417 {
1418   GstRTSPResult res;
1419   gchar versionstr[20];
1420   gchar codestr[4];
1421   gint code;
1422   gchar *bptr;
1423
1424   bptr = (gchar *) buffer;
1425
1426   parse_string (versionstr, sizeof (versionstr), &bptr);
1427   parse_string (codestr, sizeof (codestr), &bptr);
1428   code = atoi (codestr);
1429
1430   while (g_ascii_isspace (*bptr))
1431     bptr++;
1432
1433   if (strcmp (versionstr, "RTSP/1.0") == 0)
1434     GST_RTSP_CHECK (gst_rtsp_message_init_response (msg, code, bptr, NULL),
1435         parse_error);
1436   else if (strncmp (versionstr, "RTSP/", 5) == 0) {
1437     GST_RTSP_CHECK (gst_rtsp_message_init_response (msg, code, bptr, NULL),
1438         parse_error);
1439     msg->type_data.response.version = GST_RTSP_VERSION_INVALID;
1440   } else
1441     goto parse_error;
1442
1443   return GST_RTSP_OK;
1444
1445 parse_error:
1446   {
1447     return GST_RTSP_EPARSE;
1448   }
1449 }
1450
1451 static GstRTSPResult
1452 parse_request_line (GstRTSPConnection * conn, guint8 * buffer,
1453     GstRTSPMessage * msg)
1454 {
1455   GstRTSPResult res = GST_RTSP_OK;
1456   gchar versionstr[20];
1457   gchar methodstr[20];
1458   gchar urlstr[4096];
1459   gchar *bptr;
1460   GstRTSPMethod method;
1461   GstRTSPTunnelState tstate = TUNNEL_STATE_NONE;
1462
1463   bptr = (gchar *) buffer;
1464
1465   parse_string (methodstr, sizeof (methodstr), &bptr);
1466   method = gst_rtsp_find_method (methodstr);
1467   if (method == GST_RTSP_INVALID) {
1468     /* a tunnel request is allowed when we don't have one yet */
1469     if (conn->tstate != TUNNEL_STATE_NONE)
1470       goto invalid_method;
1471     /* we need GET or POST for a valid tunnel request */
1472     if (!strcmp (methodstr, "GET"))
1473       tstate = TUNNEL_STATE_GET;
1474     else if (!strcmp (methodstr, "POST"))
1475       tstate = TUNNEL_STATE_POST;
1476     else
1477       goto invalid_method;
1478   }
1479
1480   parse_string (urlstr, sizeof (urlstr), &bptr);
1481   if (G_UNLIKELY (*urlstr == '\0'))
1482     goto invalid_url;
1483
1484   parse_string (versionstr, sizeof (versionstr), &bptr);
1485
1486   if (G_UNLIKELY (*bptr != '\0'))
1487     goto invalid_version;
1488
1489   if (strcmp (versionstr, "RTSP/1.0") == 0) {
1490     res = gst_rtsp_message_init_request (msg, method, urlstr);
1491   } else if (strncmp (versionstr, "RTSP/", 5) == 0) {
1492     res = gst_rtsp_message_init_request (msg, method, urlstr);
1493     msg->type_data.request.version = GST_RTSP_VERSION_INVALID;
1494   } else if (strcmp (versionstr, "HTTP/1.0") == 0) {
1495     /* tunnel request, we need a tunnel method */
1496     if (tstate == TUNNEL_STATE_NONE) {
1497       res = GST_RTSP_EPARSE;
1498     } else {
1499       conn->tstate = tstate;
1500     }
1501   } else {
1502     res = GST_RTSP_EPARSE;
1503   }
1504
1505   return res;
1506
1507   /* ERRORS */
1508 invalid_method:
1509   {
1510     GST_ERROR ("invalid method %s", methodstr);
1511     return GST_RTSP_EPARSE;
1512   }
1513 invalid_url:
1514   {
1515     GST_ERROR ("invalid url %s", urlstr);
1516     return GST_RTSP_EPARSE;
1517   }
1518 invalid_version:
1519   {
1520     GST_ERROR ("invalid version");
1521     return GST_RTSP_EPARSE;
1522   }
1523 }
1524
1525 static GstRTSPResult
1526 parse_key_value (guint8 * buffer, gchar * key, guint keysize, gchar ** value)
1527 {
1528   gchar *bptr;
1529
1530   bptr = (gchar *) buffer;
1531
1532   /* read key */
1533   parse_key (key, keysize, &bptr);
1534   if (G_UNLIKELY (*bptr != ':'))
1535     goto no_column;
1536
1537   bptr++;
1538   while (g_ascii_isspace (*bptr))
1539     bptr++;
1540
1541   *value = bptr;
1542
1543   return GST_RTSP_OK;
1544
1545   /* ERRORS */
1546 no_column:
1547   {
1548     return GST_RTSP_EPARSE;
1549   }
1550 }
1551
1552 /* parsing lines means reading a Key: Value pair */
1553 static GstRTSPResult
1554 parse_line (GstRTSPConnection * conn, guint8 * buffer, GstRTSPMessage * msg)
1555 {
1556   GstRTSPResult res;
1557   gchar key[32];
1558   gchar *value;
1559   GstRTSPHeaderField field;
1560
1561   res = parse_key_value (buffer, key, sizeof (key), &value);
1562   if (G_UNLIKELY (res != GST_RTSP_OK))
1563     goto parse_error;
1564
1565   if (conn->tstate == TUNNEL_STATE_GET || conn->tstate == TUNNEL_STATE_POST) {
1566     /* save the tunnel session in the connection */
1567     if (!strcmp (key, "x-sessioncookie")) {
1568       strncpy (conn->tunnelid, value, TUNNELID_LEN);
1569       conn->tunnelid[TUNNELID_LEN - 1] = '\0';
1570       conn->tunneled = TRUE;
1571     }
1572   } else {
1573     field = gst_rtsp_find_header_field (key);
1574     if (field != GST_RTSP_HDR_INVALID)
1575       gst_rtsp_message_add_header (msg, field, value);
1576   }
1577
1578   return GST_RTSP_OK;
1579
1580   /* ERRORS */
1581 parse_error:
1582   {
1583     return res;
1584   }
1585 }
1586
1587 /* returns:
1588  *  GST_RTSP_OK when a complete message was read.
1589  *  GST_RTSP_EEOF: when the socket is closed
1590  *  GST_RTSP_EINTR: when more data is needed.
1591  *  GST_RTSP_..: some other error occured.
1592  */
1593 static GstRTSPResult
1594 build_next (GstRTSPBuilder * builder, GstRTSPMessage * message,
1595     GstRTSPConnection * conn)
1596 {
1597   GstRTSPResult res;
1598
1599   while (TRUE) {
1600     switch (builder->state) {
1601       case STATE_START:
1602         builder->offset = 0;
1603         res =
1604             read_bytes (conn->readfd->fd, (guint8 *) builder->buffer,
1605             &builder->offset, 1, conn->ctxp);
1606         if (res != GST_RTSP_OK)
1607           goto done;
1608
1609         /* we have 1 bytes now and we can see if this is a data message or
1610          * not */
1611         if (builder->buffer[0] == '$') {
1612           /* data message, prepare for the header */
1613           builder->state = STATE_DATA_HEADER;
1614         } else {
1615           builder->line = 0;
1616           builder->state = STATE_READ_LINES;
1617         }
1618         break;
1619       case STATE_DATA_HEADER:
1620       {
1621         res =
1622             read_bytes (conn->readfd->fd, (guint8 *) builder->buffer,
1623             &builder->offset, 4, conn->ctxp);
1624         if (res != GST_RTSP_OK)
1625           goto done;
1626
1627         gst_rtsp_message_init_data (message, builder->buffer[1]);
1628
1629         builder->body_len = (builder->buffer[2] << 8) | builder->buffer[3];
1630         builder->body_data = g_malloc (builder->body_len + 1);
1631         builder->body_data[builder->body_len] = '\0';
1632         builder->offset = 0;
1633         builder->state = STATE_DATA_BODY;
1634         break;
1635       }
1636       case STATE_DATA_BODY:
1637       {
1638         res =
1639             read_bytes (conn->readfd->fd, builder->body_data, &builder->offset,
1640             builder->body_len, conn->ctxp);
1641         if (res != GST_RTSP_OK)
1642           goto done;
1643
1644         /* we have the complete body now, store in the message adjusting the
1645          * length to include the traling '\0' */
1646         gst_rtsp_message_take_body (message,
1647             (guint8 *) builder->body_data, builder->body_len + 1);
1648         builder->body_data = NULL;
1649         builder->body_len = 0;
1650
1651         builder->state = STATE_END;
1652         break;
1653       }
1654       case STATE_READ_LINES:
1655       {
1656         res = read_line (conn->readfd->fd, builder->buffer, &builder->offset,
1657             sizeof (builder->buffer), conn->ctxp);
1658         if (res != GST_RTSP_OK)
1659           goto done;
1660
1661         /* we have a regular response */
1662         if (builder->buffer[0] == '\r') {
1663           builder->buffer[0] = '\0';
1664         }
1665
1666         if (builder->buffer[0] == '\0') {
1667           gchar *hdrval;
1668
1669           /* empty line, end of message header */
1670           /* see if there is a Content-Length header */
1671           if (gst_rtsp_message_get_header (message,
1672                   GST_RTSP_HDR_CONTENT_LENGTH, &hdrval, 0) == GST_RTSP_OK) {
1673             /* there is, prepare to read the body */
1674             builder->body_len = atol (hdrval);
1675             builder->body_data = g_malloc (builder->body_len + 1);
1676             builder->body_data[builder->body_len] = '\0';
1677             builder->offset = 0;
1678             builder->state = STATE_DATA_BODY;
1679           } else {
1680             builder->state = STATE_END;
1681           }
1682           break;
1683         }
1684
1685         /* we have a line */
1686         if (builder->line == 0) {
1687           /* first line, check for response status */
1688           if (memcmp (builder->buffer, "RTSP", 4) == 0) {
1689             res = parse_response_status (builder->buffer, message);
1690           } else {
1691             res = parse_request_line (conn, builder->buffer, message);
1692           }
1693           /* the first line must parse without errors */
1694           if (res != GST_RTSP_OK)
1695             goto done;
1696         } else {
1697           /* else just parse the line, ignore errors */
1698           parse_line (conn, builder->buffer, message);
1699         }
1700         builder->line++;
1701         builder->offset = 0;
1702         break;
1703       }
1704       case STATE_END:
1705       {
1706         gchar *session_id;
1707
1708         if (conn->tstate == TUNNEL_STATE_GET) {
1709           res = GST_RTSP_ETGET;
1710           goto done;
1711         } else if (conn->tstate == TUNNEL_STATE_POST) {
1712           res = GST_RTSP_ETPOST;
1713           goto done;
1714         }
1715
1716         if (message->type == GST_RTSP_MESSAGE_DATA) {
1717           /* data messages don't have headers */
1718           res = GST_RTSP_OK;
1719           goto done;
1720         }
1721
1722         /* save session id in the connection for further use */
1723         if (message->type == GST_RTSP_MESSAGE_RESPONSE &&
1724             gst_rtsp_message_get_header (message, GST_RTSP_HDR_SESSION,
1725                 &session_id, 0) == GST_RTSP_OK) {
1726           gint maxlen, i;
1727
1728           maxlen = sizeof (conn->session_id) - 1;
1729           /* the sessionid can have attributes marked with ;
1730            * Make sure we strip them */
1731           for (i = 0; session_id[i] != '\0'; i++) {
1732             if (session_id[i] == ';') {
1733               maxlen = i;
1734               /* parse timeout */
1735               do {
1736                 i++;
1737               } while (g_ascii_isspace (session_id[i]));
1738               if (g_str_has_prefix (&session_id[i], "timeout=")) {
1739                 gint to;
1740
1741                 /* if we parsed something valid, configure */
1742                 if ((to = atoi (&session_id[i + 8])) > 0)
1743                   conn->timeout = to;
1744               }
1745               break;
1746             }
1747           }
1748
1749           /* make sure to not overflow */
1750           strncpy (conn->session_id, session_id, maxlen);
1751           conn->session_id[maxlen] = '\0';
1752         }
1753         res = GST_RTSP_OK;
1754         goto done;
1755       }
1756       default:
1757         res = GST_RTSP_ERROR;
1758         break;
1759     }
1760   }
1761 done:
1762   return res;
1763 }
1764
1765 /**
1766  * gst_rtsp_connection_read:
1767  * @conn: a #GstRTSPConnection
1768  * @data: the data to read
1769  * @size: the size of @data
1770  * @timeout: a timeout value or #NULL
1771  *
1772  * Attempt to read @size bytes into @data from the connected @conn, blocking up to
1773  * the specified @timeout. @timeout can be #NULL, in which case this function
1774  * might block forever.
1775  *
1776  * This function can be cancelled with gst_rtsp_connection_flush().
1777  *
1778  * Returns: #GST_RTSP_OK on success.
1779  */
1780 GstRTSPResult
1781 gst_rtsp_connection_read (GstRTSPConnection * conn, guint8 * data, guint size,
1782     GTimeVal * timeout)
1783 {
1784   guint offset;
1785   gint retval;
1786   GstClockTime to;
1787   GstRTSPResult res;
1788
1789   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
1790   g_return_val_if_fail (data != NULL, GST_RTSP_EINVAL);
1791   g_return_val_if_fail (conn->readfd != NULL, GST_RTSP_EINVAL);
1792
1793   if (G_UNLIKELY (size == 0))
1794     return GST_RTSP_OK;
1795
1796   offset = 0;
1797
1798   /* configure timeout if any */
1799   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : GST_CLOCK_TIME_NONE;
1800
1801   gst_poll_set_controllable (conn->fdset, TRUE);
1802   gst_poll_fd_ctl_write (conn->fdset, conn->writefd, FALSE);
1803   gst_poll_fd_ctl_read (conn->fdset, conn->readfd, TRUE);
1804
1805   while (TRUE) {
1806     res = read_bytes (conn->readfd->fd, data, &offset, size, conn->ctxp);
1807     if (G_UNLIKELY (res == GST_RTSP_EEOF))
1808       goto eof;
1809     if (G_LIKELY (res == GST_RTSP_OK))
1810       break;
1811     if (G_UNLIKELY (res != GST_RTSP_EINTR))
1812       goto read_error;
1813
1814     do {
1815       retval = gst_poll_wait (conn->fdset, to);
1816     } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
1817
1818     /* check for timeout */
1819     if (G_UNLIKELY (retval == 0))
1820       goto select_timeout;
1821
1822     if (G_UNLIKELY (retval == -1)) {
1823       if (errno == EBUSY)
1824         goto stopped;
1825       else
1826         goto select_error;
1827     }
1828     gst_poll_set_controllable (conn->fdset, FALSE);
1829   }
1830   return GST_RTSP_OK;
1831
1832   /* ERRORS */
1833 select_error:
1834   {
1835     return GST_RTSP_ESYS;
1836   }
1837 select_timeout:
1838   {
1839     return GST_RTSP_ETIMEOUT;
1840   }
1841 stopped:
1842   {
1843     return GST_RTSP_EINTR;
1844   }
1845 eof:
1846   {
1847     return GST_RTSP_EEOF;
1848   }
1849 read_error:
1850   {
1851     return res;
1852   }
1853 }
1854
1855 static GString *
1856 gen_tunnel_reply (GstRTSPConnection * conn, GstRTSPStatusCode code)
1857 {
1858   GString *str;
1859   gchar date_string[100];
1860   const gchar *status;
1861
1862   gen_date_string (date_string, sizeof (date_string));
1863
1864   status = gst_rtsp_status_as_text (code);
1865   if (status == NULL) {
1866     code = GST_RTSP_STS_INTERNAL_SERVER_ERROR;
1867     status = "Internal Server Error";
1868   }
1869
1870   str = g_string_new ("");
1871
1872   /* */
1873   g_string_append_printf (str, "HTTP/1.0 %d %s\r\n", code, status);
1874   g_string_append_printf (str,
1875       "Server: GStreamer RTSP Server\r\n"
1876       "Date: %s\r\n"
1877       "Connection: close\r\n"
1878       "Cache-Control: no-store\r\n" "Pragma: no-cache\r\n", date_string);
1879   if (code == GST_RTSP_STS_OK) {
1880     if (conn->ip)
1881       g_string_append_printf (str, "x-server-ip-address: %s\r\n", conn->ip);
1882     g_string_append_printf (str,
1883         "Content-Type: application/x-rtsp-tunnelled\r\n");
1884   }
1885   g_string_append_printf (str, "\r\n");
1886   return str;
1887 }
1888
1889 /**
1890  * gst_rtsp_connection_receive:
1891  * @conn: a #GstRTSPConnection
1892  * @message: the message to read
1893  * @timeout: a timeout value or #NULL
1894  *
1895  * Attempt to read into @message from the connected @conn, blocking up to
1896  * the specified @timeout. @timeout can be #NULL, in which case this function
1897  * might block forever.
1898  *
1899  * This function can be cancelled with gst_rtsp_connection_flush().
1900  *
1901  * Returns: #GST_RTSP_OK on success.
1902  */
1903 GstRTSPResult
1904 gst_rtsp_connection_receive (GstRTSPConnection * conn, GstRTSPMessage * message,
1905     GTimeVal * timeout)
1906 {
1907   GstRTSPResult res;
1908   GstRTSPBuilder builder;
1909   gint retval;
1910   GstClockTime to;
1911
1912   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
1913   g_return_val_if_fail (message != NULL, GST_RTSP_EINVAL);
1914   g_return_val_if_fail (conn->readfd != NULL, GST_RTSP_EINVAL);
1915
1916   /* configure timeout if any */
1917   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : GST_CLOCK_TIME_NONE;
1918
1919   gst_poll_set_controllable (conn->fdset, TRUE);
1920   gst_poll_fd_ctl_write (conn->fdset, conn->writefd, FALSE);
1921   gst_poll_fd_ctl_read (conn->fdset, conn->readfd, TRUE);
1922
1923   memset (&builder, 0, sizeof (GstRTSPBuilder));
1924   while (TRUE) {
1925     res = build_next (&builder, message, conn);
1926     if (G_UNLIKELY (res == GST_RTSP_EEOF))
1927       goto eof;
1928     if (G_LIKELY (res == GST_RTSP_OK))
1929       break;
1930     if (res == GST_RTSP_ETGET) {
1931       GString *str;
1932
1933       /* tunnel GET request, we can reply now */
1934       str = gen_tunnel_reply (conn, GST_RTSP_STS_OK);
1935       res =
1936           gst_rtsp_connection_write (conn, (guint8 *) str->str, str->len,
1937           timeout);
1938       g_string_free (str, TRUE);
1939     } else if (res == GST_RTSP_ETPOST) {
1940       /* tunnel POST request, return the value, the caller now has to link the
1941        * two connections. */
1942       break;
1943     } else if (G_UNLIKELY (res != GST_RTSP_EINTR))
1944       goto read_error;
1945
1946     do {
1947       retval = gst_poll_wait (conn->fdset, to);
1948     } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
1949
1950     /* check for timeout */
1951     if (G_UNLIKELY (retval == 0))
1952       goto select_timeout;
1953
1954     if (G_UNLIKELY (retval == -1)) {
1955       if (errno == EBUSY)
1956         goto stopped;
1957       else
1958         goto select_error;
1959     }
1960     gst_poll_set_controllable (conn->fdset, FALSE);
1961   }
1962
1963   /* we have a message here */
1964   build_reset (&builder);
1965
1966   return GST_RTSP_OK;
1967
1968   /* ERRORS */
1969 select_error:
1970   {
1971     res = GST_RTSP_ESYS;
1972     goto cleanup;
1973   }
1974 select_timeout:
1975   {
1976     res = GST_RTSP_ETIMEOUT;
1977     goto cleanup;
1978   }
1979 stopped:
1980   {
1981     res = GST_RTSP_EINTR;
1982     goto cleanup;
1983   }
1984 eof:
1985   {
1986     res = GST_RTSP_EEOF;
1987     goto cleanup;
1988   }
1989 read_error:
1990 cleanup:
1991   {
1992     build_reset (&builder);
1993     gst_rtsp_message_unset (message);
1994     return res;
1995   }
1996 }
1997
1998 /**
1999  * gst_rtsp_connection_close:
2000  * @conn: a #GstRTSPConnection
2001  *
2002  * Close the connected @conn. After this call, the connection is in the same
2003  * state as when it was first created.
2004  *
2005  * Returns: #GST_RTSP_OK on success.
2006  */
2007 GstRTSPResult
2008 gst_rtsp_connection_close (GstRTSPConnection * conn)
2009 {
2010   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2011
2012   g_free (conn->ip);
2013   conn->ip = NULL;
2014
2015   REMOVE_POLLFD (conn->fdset, &conn->fd0);
2016   REMOVE_POLLFD (conn->fdset, &conn->fd1);
2017   conn->writefd = NULL;
2018   conn->readfd = NULL;
2019   conn->tunneled = FALSE;
2020   conn->tstate = TUNNEL_STATE_NONE;
2021   conn->ctxp = NULL;
2022   g_free (conn->username);
2023   conn->username = NULL;
2024   g_free (conn->passwd);
2025   conn->passwd = NULL;
2026   gst_rtsp_connection_clear_auth_params (conn);
2027   conn->timeout = 60;
2028   conn->cseq = 0;
2029   conn->session_id[0] = '\0';
2030
2031   return GST_RTSP_OK;
2032 }
2033
2034 /**
2035  * gst_rtsp_connection_free:
2036  * @conn: a #GstRTSPConnection
2037  *
2038  * Close and free @conn.
2039  *
2040  * Returns: #GST_RTSP_OK on success.
2041  */
2042 GstRTSPResult
2043 gst_rtsp_connection_free (GstRTSPConnection * conn)
2044 {
2045   GstRTSPResult res;
2046
2047   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2048
2049   res = gst_rtsp_connection_close (conn);
2050   gst_poll_free (conn->fdset);
2051   g_timer_destroy (conn->timer);
2052   gst_rtsp_url_free (conn->url);
2053   g_free (conn->proxy_host);
2054   g_free (conn);
2055 #ifdef G_OS_WIN32
2056   WSACleanup ();
2057 #endif
2058
2059   return res;
2060 }
2061
2062 /**
2063  * gst_rtsp_connection_poll:
2064  * @conn: a #GstRTSPConnection
2065  * @events: a bitmask of #GstRTSPEvent flags to check
2066  * @revents: location for result flags
2067  * @timeout: a timeout
2068  *
2069  * Wait up to the specified @timeout for the connection to become available for
2070  * at least one of the operations specified in @events. When the function returns
2071  * with #GST_RTSP_OK, @revents will contain a bitmask of available operations on
2072  * @conn.
2073  *
2074  * @timeout can be #NULL, in which case this function might block forever.
2075  *
2076  * This function can be cancelled with gst_rtsp_connection_flush().
2077  *
2078  * Returns: #GST_RTSP_OK on success.
2079  *
2080  * Since: 0.10.15
2081  */
2082 GstRTSPResult
2083 gst_rtsp_connection_poll (GstRTSPConnection * conn, GstRTSPEvent events,
2084     GstRTSPEvent * revents, GTimeVal * timeout)
2085 {
2086   GstClockTime to;
2087   gint retval;
2088
2089   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2090   g_return_val_if_fail (events != 0, GST_RTSP_EINVAL);
2091   g_return_val_if_fail (revents != NULL, GST_RTSP_EINVAL);
2092   g_return_val_if_fail (conn->readfd != NULL, GST_RTSP_EINVAL);
2093   g_return_val_if_fail (conn->writefd != NULL, GST_RTSP_EINVAL);
2094
2095   gst_poll_set_controllable (conn->fdset, TRUE);
2096
2097   /* add fd to writer set when asked to */
2098   gst_poll_fd_ctl_write (conn->fdset, conn->writefd,
2099       events & GST_RTSP_EV_WRITE);
2100
2101   /* add fd to reader set when asked to */
2102   gst_poll_fd_ctl_read (conn->fdset, conn->readfd, events & GST_RTSP_EV_READ);
2103
2104   /* configure timeout if any */
2105   to = timeout ? GST_TIMEVAL_TO_TIME (*timeout) : GST_CLOCK_TIME_NONE;
2106
2107   do {
2108     retval = gst_poll_wait (conn->fdset, to);
2109   } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
2110
2111   if (G_UNLIKELY (retval == 0))
2112     goto select_timeout;
2113
2114   if (G_UNLIKELY (retval == -1)) {
2115     if (errno == EBUSY)
2116       goto stopped;
2117     else
2118       goto select_error;
2119   }
2120
2121   *revents = 0;
2122   if (events & GST_RTSP_EV_READ) {
2123     if (gst_poll_fd_can_read (conn->fdset, conn->readfd))
2124       *revents |= GST_RTSP_EV_READ;
2125   }
2126   if (events & GST_RTSP_EV_WRITE) {
2127     if (gst_poll_fd_can_write (conn->fdset, conn->writefd))
2128       *revents |= GST_RTSP_EV_WRITE;
2129   }
2130   return GST_RTSP_OK;
2131
2132   /* ERRORS */
2133 select_timeout:
2134   {
2135     return GST_RTSP_ETIMEOUT;
2136   }
2137 select_error:
2138   {
2139     return GST_RTSP_ESYS;
2140   }
2141 stopped:
2142   {
2143     return GST_RTSP_EINTR;
2144   }
2145 }
2146
2147 /**
2148  * gst_rtsp_connection_next_timeout:
2149  * @conn: a #GstRTSPConnection
2150  * @timeout: a timeout
2151  *
2152  * Calculate the next timeout for @conn, storing the result in @timeout.
2153  *
2154  * Returns: #GST_RTSP_OK.
2155  */
2156 GstRTSPResult
2157 gst_rtsp_connection_next_timeout (GstRTSPConnection * conn, GTimeVal * timeout)
2158 {
2159   gdouble elapsed;
2160   glong sec;
2161   gulong usec;
2162
2163   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2164   g_return_val_if_fail (timeout != NULL, GST_RTSP_EINVAL);
2165
2166   elapsed = g_timer_elapsed (conn->timer, &usec);
2167   if (elapsed >= conn->timeout) {
2168     sec = 0;
2169     usec = 0;
2170   } else {
2171     sec = conn->timeout - elapsed;
2172   }
2173
2174   timeout->tv_sec = sec;
2175   timeout->tv_usec = usec;
2176
2177   return GST_RTSP_OK;
2178 }
2179
2180 /**
2181  * gst_rtsp_connection_reset_timeout:
2182  * @conn: a #GstRTSPConnection
2183  *
2184  * Reset the timeout of @conn.
2185  *
2186  * Returns: #GST_RTSP_OK.
2187  */
2188 GstRTSPResult
2189 gst_rtsp_connection_reset_timeout (GstRTSPConnection * conn)
2190 {
2191   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2192
2193   g_timer_start (conn->timer);
2194
2195   return GST_RTSP_OK;
2196 }
2197
2198 /**
2199  * gst_rtsp_connection_flush:
2200  * @conn: a #GstRTSPConnection
2201  * @flush: start or stop the flush
2202  *
2203  * Start or stop the flushing action on @conn. When flushing, all current
2204  * and future actions on @conn will return #GST_RTSP_EINTR until the connection
2205  * is set to non-flushing mode again.
2206  *
2207  * Returns: #GST_RTSP_OK.
2208  */
2209 GstRTSPResult
2210 gst_rtsp_connection_flush (GstRTSPConnection * conn, gboolean flush)
2211 {
2212   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2213
2214   gst_poll_set_flushing (conn->fdset, flush);
2215
2216   return GST_RTSP_OK;
2217 }
2218
2219 /**
2220  * gst_rtsp_connection_set_proxy:
2221  * @conn: a #GstRTSPConnection
2222  * @host: the proxy host
2223  * @port: the proxy port
2224  *
2225  * Set the proxy host and port.
2226  *
2227  * Returns: #GST_RTSP_OK.
2228  *
2229  * Since: 0.10.23
2230  */
2231 GstRTSPResult
2232 gst_rtsp_connection_set_proxy (GstRTSPConnection * conn,
2233     const gchar * host, guint port)
2234 {
2235   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2236
2237   g_free (conn->proxy_host);
2238   conn->proxy_host = g_strdup (host);
2239   conn->proxy_port = port;
2240
2241   return GST_RTSP_OK;
2242 }
2243
2244 /**
2245  * gst_rtsp_connection_set_auth:
2246  * @conn: a #GstRTSPConnection
2247  * @method: authentication method
2248  * @user: the user
2249  * @pass: the password
2250  *
2251  * Configure @conn for authentication mode @method with @user and @pass as the
2252  * user and password respectively.
2253  *
2254  * Returns: #GST_RTSP_OK.
2255  */
2256 GstRTSPResult
2257 gst_rtsp_connection_set_auth (GstRTSPConnection * conn,
2258     GstRTSPAuthMethod method, const gchar * user, const gchar * pass)
2259 {
2260   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2261
2262   if (method == GST_RTSP_AUTH_DIGEST && ((user == NULL || pass == NULL)
2263           || g_strrstr (user, ":") != NULL))
2264     return GST_RTSP_EINVAL;
2265
2266   /* Make sure the username and passwd are being set for authentication */
2267   if (method == GST_RTSP_AUTH_NONE && (user == NULL || pass == NULL))
2268     return GST_RTSP_EINVAL;
2269
2270   /* ":" chars are not allowed in usernames for basic auth */
2271   if (method == GST_RTSP_AUTH_BASIC && g_strrstr (user, ":") != NULL)
2272     return GST_RTSP_EINVAL;
2273
2274   g_free (conn->username);
2275   g_free (conn->passwd);
2276
2277   conn->auth_method = method;
2278   conn->username = g_strdup (user);
2279   conn->passwd = g_strdup (pass);
2280
2281   return GST_RTSP_OK;
2282 }
2283
2284 /**
2285  * str_case_hash:
2286  * @key: ASCII string to hash
2287  *
2288  * Hashes @key in a case-insensitive manner.
2289  *
2290  * Returns: the hash code.
2291  **/
2292 static guint
2293 str_case_hash (gconstpointer key)
2294 {
2295   const char *p = key;
2296   guint h = g_ascii_toupper (*p);
2297
2298   if (h)
2299     for (p += 1; *p != '\0'; p++)
2300       h = (h << 5) - h + g_ascii_toupper (*p);
2301
2302   return h;
2303 }
2304
2305 /**
2306  * str_case_equal:
2307  * @v1: an ASCII string
2308  * @v2: another ASCII string
2309  *
2310  * Compares @v1 and @v2 in a case-insensitive manner
2311  *
2312  * Returns: %TRUE if they are equal (modulo case)
2313  **/
2314 static gboolean
2315 str_case_equal (gconstpointer v1, gconstpointer v2)
2316 {
2317   const char *string1 = v1;
2318   const char *string2 = v2;
2319
2320   return g_ascii_strcasecmp (string1, string2) == 0;
2321 }
2322
2323 /**
2324  * gst_rtsp_connection_set_auth_param:
2325  * @conn: a #GstRTSPConnection
2326  * @param: authentication directive
2327  * @value: value
2328  *
2329  * Setup @conn with authentication directives. This is not necesary for
2330  * methods #GST_RTSP_AUTH_NONE and #GST_RTSP_AUTH_BASIC. For
2331  * #GST_RTSP_AUTH_DIGEST, directives should be taken from the digest challenge
2332  * in the WWW-Authenticate response header and can include realm, domain,
2333  * nonce, opaque, stale, algorithm, qop as per RFC2617.
2334  *
2335  * Since: 0.10.20
2336  */
2337 void
2338 gst_rtsp_connection_set_auth_param (GstRTSPConnection * conn,
2339     const gchar * param, const gchar * value)
2340 {
2341   g_return_if_fail (conn != NULL);
2342   g_return_if_fail (param != NULL);
2343
2344   if (conn->auth_params == NULL) {
2345     conn->auth_params =
2346         g_hash_table_new_full (str_case_hash, str_case_equal, g_free, g_free);
2347   }
2348   g_hash_table_insert (conn->auth_params, g_strdup (param), g_strdup (value));
2349 }
2350
2351 /**
2352  * gst_rtsp_connection_clear_auth_params:
2353  * @conn: a #GstRTSPConnection
2354  *
2355  * Clear the list of authentication directives stored in @conn.
2356  *
2357  * Since: 0.10.20
2358  */
2359 void
2360 gst_rtsp_connection_clear_auth_params (GstRTSPConnection * conn)
2361 {
2362   g_return_if_fail (conn != NULL);
2363
2364   if (conn->auth_params != NULL) {
2365     g_hash_table_destroy (conn->auth_params);
2366     conn->auth_params = NULL;
2367   }
2368 }
2369
2370 static GstRTSPResult
2371 set_qos_dscp (gint fd, guint qos_dscp)
2372 {
2373   union gst_sockaddr sa;
2374   socklen_t slen = sizeof (sa);
2375   gint af;
2376   gint tos;
2377
2378   if (fd == -1)
2379     return GST_RTSP_OK;
2380
2381   if (getsockname (fd, &sa.sa, &slen) < 0)
2382     goto no_getsockname;
2383
2384   af = sa.sa.sa_family;
2385
2386   /* if this is an IPv4-mapped address then do IPv4 QoS */
2387   if (af == AF_INET6) {
2388     if (IN6_IS_ADDR_V4MAPPED (&sa.sa_in6.sin6_addr))
2389       af = AF_INET;
2390   }
2391
2392   /* extract and shift 6 bits of the DSCP */
2393   tos = (qos_dscp & 0x3f) << 2;
2394
2395   switch (af) {
2396     case AF_INET:
2397       if (SETSOCKOPT (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos)) < 0)
2398         goto no_setsockopt;
2399       break;
2400     case AF_INET6:
2401 #ifdef IPV6_TCLASS
2402       if (SETSOCKOPT (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos)) < 0)
2403         goto no_setsockopt;
2404       break;
2405 #endif
2406     default:
2407       goto wrong_family;
2408   }
2409
2410   return GST_RTSP_OK;
2411
2412   /* ERRORS */
2413 no_getsockname:
2414 no_setsockopt:
2415   {
2416     return GST_RTSP_ESYS;
2417   }
2418
2419 wrong_family:
2420   {
2421     return GST_RTSP_ERROR;
2422   }
2423 }
2424
2425 /**
2426  * gst_rtsp_connection_set_qos_dscp:
2427  * @conn: a #GstRTSPConnection
2428  * @qos_dscp: DSCP value
2429  *
2430  * Configure @conn to use the specified DSCP value.
2431  *
2432  * Returns: #GST_RTSP_OK on success.
2433  *
2434  * Since: 0.10.20
2435  */
2436 GstRTSPResult
2437 gst_rtsp_connection_set_qos_dscp (GstRTSPConnection * conn, guint qos_dscp)
2438 {
2439   GstRTSPResult res;
2440
2441   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2442   g_return_val_if_fail (conn->readfd != NULL, GST_RTSP_EINVAL);
2443   g_return_val_if_fail (conn->writefd != NULL, GST_RTSP_EINVAL);
2444
2445   res = set_qos_dscp (conn->fd0.fd, qos_dscp);
2446   if (res == GST_RTSP_OK)
2447     res = set_qos_dscp (conn->fd1.fd, qos_dscp);
2448
2449   return res;
2450 }
2451
2452
2453 /**
2454  * gst_rtsp_connection_get_url:
2455  * @conn: a #GstRTSPConnection
2456  *
2457  * Retrieve the URL of the other end of @conn.
2458  *
2459  * Returns: The URL. This value remains valid until the
2460  * connection is freed.
2461  *
2462  * Since: 0.10.23
2463  */
2464 GstRTSPUrl *
2465 gst_rtsp_connection_get_url (const GstRTSPConnection * conn)
2466 {
2467   g_return_val_if_fail (conn != NULL, NULL);
2468
2469   return conn->url;
2470 }
2471
2472 /**
2473  * gst_rtsp_connection_get_ip:
2474  * @conn: a #GstRTSPConnection
2475  *
2476  * Retrieve the IP address of the other end of @conn.
2477  *
2478  * Returns: The IP address as a string. this value remains valid until the
2479  * connection is closed.
2480  *
2481  * Since: 0.10.20
2482  */
2483 const gchar *
2484 gst_rtsp_connection_get_ip (const GstRTSPConnection * conn)
2485 {
2486   g_return_val_if_fail (conn != NULL, NULL);
2487
2488   return conn->ip;
2489 }
2490
2491 /**
2492  * gst_rtsp_connection_set_ip:
2493  * @conn: a #GstRTSPConnection
2494  * @ip: an ip address
2495  *
2496  * Set the IP address of the server.
2497  *
2498  * Since: 0.10.23
2499  */
2500 void
2501 gst_rtsp_connection_set_ip (GstRTSPConnection * conn, const gchar * ip)
2502 {
2503   g_return_if_fail (conn != NULL);
2504
2505   g_free (conn->ip);
2506   conn->ip = g_strdup (ip);
2507 }
2508
2509 /**
2510  * gst_rtsp_connection_get_readfd:
2511  * @conn: a #GstRTSPConnection
2512  *
2513  * Get the file descriptor for reading.
2514  *
2515  * Returns: the file descriptor used for reading or -1 on error. The file
2516  * descriptor remains valid until the connection is closed.
2517  *
2518  * Since: 0.10.23
2519  */
2520 gint
2521 gst_rtsp_connection_get_readfd (const GstRTSPConnection * conn)
2522 {
2523   g_return_val_if_fail (conn != NULL, -1);
2524   g_return_val_if_fail (conn->readfd != NULL, -1);
2525
2526   return conn->readfd->fd;
2527 }
2528
2529 /**
2530  * gst_rtsp_connection_get_writefd:
2531  * @conn: a #GstRTSPConnection
2532  *
2533  * Get the file descriptor for writing.
2534  *
2535  * Returns: the file descriptor used for writing or -1 on error. The file
2536  * descriptor remains valid until the connection is closed.
2537  *
2538  * Since: 0.10.23
2539  */
2540 gint
2541 gst_rtsp_connection_get_writefd (const GstRTSPConnection * conn)
2542 {
2543   g_return_val_if_fail (conn != NULL, -1);
2544   g_return_val_if_fail (conn->writefd != NULL, -1);
2545
2546   return conn->writefd->fd;
2547 }
2548
2549
2550 /**
2551  * gst_rtsp_connection_set_tunneled:
2552  * @conn: a #GstRTSPConnection
2553  * @tunneled: the new state
2554  *
2555  * Set the HTTP tunneling state of the connection. This must be configured before
2556  * the @conn is connected.
2557  *
2558  * Since: 0.10.23
2559  */
2560 void
2561 gst_rtsp_connection_set_tunneled (GstRTSPConnection * conn, gboolean tunneled)
2562 {
2563   g_return_if_fail (conn != NULL);
2564   g_return_if_fail (conn->readfd == NULL);
2565   g_return_if_fail (conn->writefd == NULL);
2566
2567   conn->tunneled = tunneled;
2568 }
2569
2570 /**
2571  * gst_rtsp_connection_is_tunneled:
2572  * @conn: a #GstRTSPConnection
2573  *
2574  * Get the tunneling state of the connection.
2575  *
2576  * Returns: if @conn is using HTTP tunneling.
2577  *
2578  * Since: 0.10.23
2579  */
2580 gboolean
2581 gst_rtsp_connection_is_tunneled (const GstRTSPConnection * conn)
2582 {
2583   g_return_val_if_fail (conn != NULL, FALSE);
2584
2585   return conn->tunneled;
2586 }
2587
2588 /**
2589  * gst_rtsp_connection_get_tunnelid:
2590  * @conn: a #GstRTSPConnection
2591  *
2592  * Get the tunnel session id the connection.
2593  *
2594  * Returns: returns a non-empty string if @conn is being tunneled over HTTP.
2595  *
2596  * Since: 0.10.23
2597  */
2598 const gchar *
2599 gst_rtsp_connection_get_tunnelid (const GstRTSPConnection * conn)
2600 {
2601   g_return_val_if_fail (conn != NULL, NULL);
2602
2603   if (!conn->tunneled)
2604     return NULL;
2605
2606   return conn->tunnelid;
2607 }
2608
2609 /**
2610  * gst_rtsp_connection_do_tunnel:
2611  * @conn: a #GstRTSPConnection
2612  * @conn2: a #GstRTSPConnection
2613  *
2614  * If @conn received the first tunnel connection and @conn2 received
2615  * the second tunnel connection, link the two connections together so that
2616  * @conn manages the tunneled connection.
2617  *
2618  * After this call, @conn2 cannot be used anymore and must be freed with
2619  * gst_rtsp_connection_free().
2620  *
2621  * Returns: return GST_RTSP_OK on success.
2622  *
2623  * Since: 0.10.23
2624  */
2625 GstRTSPResult
2626 gst_rtsp_connection_do_tunnel (GstRTSPConnection * conn,
2627     GstRTSPConnection * conn2)
2628 {
2629   g_return_val_if_fail (conn != NULL, GST_RTSP_EINVAL);
2630   g_return_val_if_fail (conn2 != NULL, GST_RTSP_EINVAL);
2631   g_return_val_if_fail (conn->tstate == TUNNEL_STATE_GET, GST_RTSP_EINVAL);
2632   g_return_val_if_fail (conn2->tstate == TUNNEL_STATE_POST, GST_RTSP_EINVAL);
2633   g_return_val_if_fail (!memcmp (conn2->tunnelid, conn->tunnelid, TUNNELID_LEN),
2634       GST_RTSP_EINVAL);
2635
2636   /* both connections have fd0 as the read/write socket. start by taking the
2637    * socket from conn2 and set it as the socket in conn */
2638   conn->fd1 = conn2->fd0;
2639
2640   /* clean up some of the state of conn2 */
2641   gst_poll_remove_fd (conn2->fdset, &conn2->fd0);
2642   conn2->fd0.fd = -1;
2643   conn2->readfd = conn2->writefd = NULL;
2644
2645   /* We make fd0 the write socket and fd1 the read socket. */
2646   conn->writefd = &conn->fd0;
2647   conn->readfd = &conn->fd1;
2648
2649   conn->tstate = TUNNEL_STATE_COMPLETE;
2650
2651   /* we need base64 decoding for the readfd */
2652   conn->ctx.state = 0;
2653   conn->ctx.save = 0;
2654   conn->ctx.cout = 0;
2655   conn->ctx.coutl = 0;
2656   conn->ctxp = &conn->ctx;
2657
2658   return GST_RTSP_OK;
2659 }
2660
2661 #define READ_COND   (G_IO_IN | G_IO_HUP | G_IO_ERR)
2662 #define WRITE_COND  (G_IO_OUT | G_IO_ERR)
2663
2664 typedef struct
2665 {
2666   guint8 *data;
2667   guint size;
2668   guint id;
2669 } GstRTSPRec;
2670
2671 /* async functions */
2672 struct _GstRTSPWatch
2673 {
2674   GSource source;
2675
2676   GstRTSPConnection *conn;
2677
2678   GstRTSPBuilder builder;
2679   GstRTSPMessage message;
2680
2681   GPollFD readfd;
2682   GPollFD writefd;
2683   gboolean write_added;
2684
2685   /* queued message for transmission */
2686   guint id;
2687   GAsyncQueue *messages;
2688   guint8 *write_data;
2689   guint write_off;
2690   guint write_size;
2691   guint write_id;
2692
2693   GstRTSPWatchFuncs funcs;
2694
2695   gpointer user_data;
2696   GDestroyNotify notify;
2697 };
2698
2699 static gboolean
2700 gst_rtsp_source_prepare (GSource * source, gint * timeout)
2701 {
2702   GstRTSPWatch *watch = (GstRTSPWatch *) source;
2703
2704   *timeout = (watch->conn->timeout * 1000);
2705
2706   return FALSE;
2707 }
2708
2709 static gboolean
2710 gst_rtsp_source_check (GSource * source)
2711 {
2712   GstRTSPWatch *watch = (GstRTSPWatch *) source;
2713
2714   if (watch->readfd.revents & READ_COND)
2715     return TRUE;
2716
2717   if (watch->writefd.revents & WRITE_COND)
2718     return TRUE;
2719
2720   return FALSE;
2721 }
2722
2723 static gboolean
2724 gst_rtsp_source_dispatch (GSource * source, GSourceFunc callback G_GNUC_UNUSED,
2725     gpointer user_data G_GNUC_UNUSED)
2726 {
2727   GstRTSPWatch *watch = (GstRTSPWatch *) source;
2728   GstRTSPResult res;
2729
2730   /* first read as much as we can */
2731   if (watch->readfd.revents & READ_COND) {
2732     do {
2733       res = build_next (&watch->builder, &watch->message, watch->conn);
2734       if (res == GST_RTSP_EINTR)
2735         break;
2736       if (G_UNLIKELY (res == GST_RTSP_EEOF))
2737         goto eof;
2738       if (res == GST_RTSP_ETGET) {
2739         GString *str;
2740         GstRTSPStatusCode code;
2741         guint size;
2742
2743         if (watch->funcs.tunnel_start)
2744           code = watch->funcs.tunnel_start (watch, watch->user_data);
2745         else
2746           code = GST_RTSP_STS_OK;
2747
2748         /* queue the response string */
2749         str = gen_tunnel_reply (watch->conn, code);
2750         size = str->len;
2751         gst_rtsp_watch_queue_data (watch, (guint8 *) g_string_free (str, FALSE),
2752             size);
2753       } else if (res == GST_RTSP_ETPOST) {
2754         /* in the callback the connection should be tunneled with the
2755          * GET connection */
2756         if (watch->funcs.tunnel_complete)
2757           watch->funcs.tunnel_complete (watch, watch->user_data);
2758       } else if (G_UNLIKELY (res != GST_RTSP_OK))
2759         goto error;
2760
2761       if (G_LIKELY (res == GST_RTSP_OK)) {
2762         if (watch->funcs.message_received)
2763           watch->funcs.message_received (watch, &watch->message,
2764               watch->user_data);
2765
2766         gst_rtsp_message_unset (&watch->message);
2767       }
2768       build_reset (&watch->builder);
2769     } while (FALSE);
2770   }
2771
2772   if (watch->writefd.revents & WRITE_COND) {
2773     do {
2774       if (watch->write_data == NULL) {
2775         GstRTSPRec *rec;
2776
2777         /* get a new message from the queue */
2778         rec = g_async_queue_try_pop (watch->messages);
2779         if (rec == NULL)
2780           goto done;
2781
2782         watch->write_off = 0;
2783         watch->write_data = rec->data;
2784         watch->write_size = rec->size;
2785         watch->write_id = rec->id;
2786
2787         g_slice_free (GstRTSPRec, rec);
2788       }
2789
2790       res = write_bytes (watch->writefd.fd, watch->write_data,
2791           &watch->write_off, watch->write_size);
2792       if (res == GST_RTSP_EINTR)
2793         break;
2794       if (G_UNLIKELY (res != GST_RTSP_OK))
2795         goto error;
2796
2797       if (watch->funcs.message_sent)
2798         watch->funcs.message_sent (watch, watch->write_id, watch->user_data);
2799
2800     done:
2801       if (g_async_queue_length (watch->messages) == 0 && watch->write_added) {
2802         g_source_remove_poll ((GSource *) watch, &watch->writefd);
2803         watch->write_added = FALSE;
2804         watch->writefd.revents = 0;
2805       }
2806       g_free (watch->write_data);
2807       watch->write_data = NULL;
2808     } while (FALSE);
2809   }
2810
2811   return TRUE;
2812
2813   /* ERRORS */
2814 eof:
2815   {
2816     if (watch->funcs.closed)
2817       watch->funcs.closed (watch, watch->user_data);
2818     return FALSE;
2819   }
2820 error:
2821   {
2822     if (watch->funcs.error)
2823       watch->funcs.error (watch, res, watch->user_data);
2824     return FALSE;
2825   }
2826 }
2827
2828 static void
2829 gst_rtsp_rec_free (gpointer data)
2830 {
2831   GstRTSPRec *rec = data;
2832
2833   g_free (rec->data);
2834   g_slice_free (GstRTSPRec, rec);
2835 }
2836
2837 static void
2838 gst_rtsp_source_finalize (GSource * source)
2839 {
2840   GstRTSPWatch *watch = (GstRTSPWatch *) source;
2841
2842   build_reset (&watch->builder);
2843   gst_rtsp_message_unset (&watch->message);
2844
2845   g_async_queue_unref (watch->messages);
2846   watch->messages = NULL;
2847
2848   g_free (watch->write_data);
2849
2850   if (watch->notify)
2851     watch->notify (watch->user_data);
2852 }
2853
2854 static GSourceFuncs gst_rtsp_source_funcs = {
2855   gst_rtsp_source_prepare,
2856   gst_rtsp_source_check,
2857   gst_rtsp_source_dispatch,
2858   gst_rtsp_source_finalize,
2859   NULL,
2860   NULL
2861 };
2862
2863 /**
2864  * gst_rtsp_watch_new:
2865  * @conn: a #GstRTSPConnection
2866  * @funcs: watch functions
2867  * @user_data: user data to pass to @funcs
2868  * @notify: notify when @user_data is not referenced anymore
2869  *
2870  * Create a watch object for @conn. The functions provided in @funcs will be
2871  * called with @user_data when activity happened on the watch.
2872  *
2873  * The new watch is usually created so that it can be attached to a
2874  * maincontext with gst_rtsp_watch_attach().
2875  *
2876  * @conn must exist for the entire lifetime of the watch.
2877  *
2878  * Returns: a #GstRTSPWatch that can be used for asynchronous RTSP
2879  * communication. Free with gst_rtsp_watch_unref () after usage.
2880  *
2881  * Since: 0.10.23
2882  */
2883 GstRTSPWatch *
2884 gst_rtsp_watch_new (GstRTSPConnection * conn,
2885     GstRTSPWatchFuncs * funcs, gpointer user_data, GDestroyNotify notify)
2886 {
2887   GstRTSPWatch *result;
2888
2889   g_return_val_if_fail (conn != NULL, NULL);
2890   g_return_val_if_fail (funcs != NULL, NULL);
2891   g_return_val_if_fail (conn->readfd != NULL, NULL);
2892   g_return_val_if_fail (conn->writefd != NULL, NULL);
2893
2894   result = (GstRTSPWatch *) g_source_new (&gst_rtsp_source_funcs,
2895       sizeof (GstRTSPWatch));
2896
2897   result->conn = conn;
2898   result->builder.state = STATE_START;
2899
2900   result->messages = g_async_queue_new_full (gst_rtsp_rec_free);
2901
2902   result->readfd.fd = -1;
2903   result->writefd.fd = -1;
2904
2905   gst_rtsp_watch_reset (result);
2906
2907   result->funcs = *funcs;
2908   result->user_data = user_data;
2909   result->notify = notify;
2910
2911   /* only add the read fd, the write fd is only added when we have data
2912    * to send. */
2913   g_source_add_poll ((GSource *) result, &result->readfd);
2914
2915   return result;
2916 }
2917
2918 /**
2919  * gst_rtsp_watch_reset:
2920  * @watch: a #GstRTSPWatch
2921  *
2922  * Reset @watch, this is usually called after gst_rtsp_connection_do_tunnel()
2923  * when the file descriptors of the connection might have changed.
2924  *
2925  * Since: 0.10.23
2926  */
2927 void
2928 gst_rtsp_watch_reset (GstRTSPWatch * watch)
2929 {
2930   if (watch->readfd.fd != -1)
2931     g_source_remove_poll ((GSource *) watch, &watch->readfd);
2932   if (watch->writefd.fd != -1)
2933     g_source_remove_poll ((GSource *) watch, &watch->writefd);
2934
2935   watch->readfd.fd = watch->conn->readfd->fd;
2936   watch->readfd.events = READ_COND;
2937   watch->readfd.revents = 0;
2938
2939   watch->writefd.fd = watch->conn->writefd->fd;
2940   watch->writefd.events = WRITE_COND;
2941   watch->writefd.revents = 0;
2942   watch->write_added = FALSE;
2943
2944   g_source_add_poll ((GSource *) watch, &watch->readfd);
2945 }
2946
2947 /**
2948  * gst_rtsp_watch_attach:
2949  * @watch: a #GstRTSPWatch
2950  * @context: a GMainContext (if NULL, the default context will be used)
2951  *
2952  * Adds a #GstRTSPWatch to a context so that it will be executed within that context.
2953  *
2954  * Returns: the ID (greater than 0) for the watch within the GMainContext.
2955  *
2956  * Since: 0.10.23
2957  */
2958 guint
2959 gst_rtsp_watch_attach (GstRTSPWatch * watch, GMainContext * context)
2960 {
2961   g_return_val_if_fail (watch != NULL, 0);
2962
2963   return g_source_attach ((GSource *) watch, context);
2964 }
2965
2966 /**
2967  * gst_rtsp_watch_unref:
2968  * @watch: a #GstRTSPWatch
2969  *
2970  * Decreases the reference count of @watch by one. If the resulting reference
2971  * count is zero the watch and associated memory will be destroyed.
2972  *
2973  * Since: 0.10.23
2974  */
2975 void
2976 gst_rtsp_watch_unref (GstRTSPWatch * watch)
2977 {
2978   g_return_if_fail (watch != NULL);
2979
2980   g_source_unref ((GSource *) watch);
2981 }
2982
2983 /**
2984  * gst_rtsp_watch_queue_data:
2985  * @watch: a #GstRTSPWatch
2986  * @data: the data to queue
2987  * @size: the size of @data
2988  *
2989  * Queue @data for transmission in @watch. It will be transmitted when the
2990  * connection of the @watch becomes writable.
2991  *
2992  * This function will take ownership of @data and g_free() it after use.
2993  *
2994  * The return value of this function will be used as the id argument in the
2995  * message_sent callback.
2996  *
2997  * Returns: an id.
2998  *
2999  * Since: 0.10.24
3000  */
3001 guint
3002 gst_rtsp_watch_queue_data (GstRTSPWatch * watch, const guint8 * data,
3003     guint size)
3004 {
3005   GstRTSPRec *rec;
3006
3007   g_return_val_if_fail (watch != NULL, GST_RTSP_EINVAL);
3008   g_return_val_if_fail (data != NULL, GST_RTSP_EINVAL);
3009   g_return_val_if_fail (size != 0, GST_RTSP_EINVAL);
3010
3011   /* make a record with the data and id */
3012   rec = g_slice_new (GstRTSPRec);
3013   rec->data = (guint8 *) data;
3014   rec->size = size;
3015   do {
3016     /* make sure rec->id is never 0 */
3017     rec->id = ++watch->id;
3018   } while (G_UNLIKELY (rec->id == 0));
3019
3020   /* add the record to a queue. FIXME we would like to have an upper limit here */
3021   g_async_queue_push (watch->messages, rec);
3022
3023   /* FIXME: does the following need to be made thread-safe? (this might be
3024    * called from a streaming thread, like appsink's render function) */
3025   /* make sure the main context will now also check for writability on the
3026    * socket */
3027   if (!watch->write_added) {
3028     g_source_add_poll ((GSource *) watch, &watch->writefd);
3029     watch->write_added = TRUE;
3030   }
3031
3032   return rec->id;
3033 }
3034
3035 /**
3036  * gst_rtsp_watch_queue_message:
3037  * @watch: a #GstRTSPWatch
3038  * @message: a #GstRTSPMessage
3039  *
3040  * Queue a @message for transmission in @watch. The contents of this
3041  * message will be serialized and transmitted when the connection of the
3042  * @watch becomes writable.
3043  *
3044  * The return value of this function will be used as the id argument in the
3045  * message_sent callback.
3046  *
3047  * Returns: an id.
3048  *
3049  * Since: 0.10.23
3050  */
3051 guint
3052 gst_rtsp_watch_queue_message (GstRTSPWatch * watch, GstRTSPMessage * message)
3053 {
3054   GString *str;
3055   guint size;
3056
3057   g_return_val_if_fail (watch != NULL, GST_RTSP_EINVAL);
3058   g_return_val_if_fail (message != NULL, GST_RTSP_EINVAL);
3059
3060   /* make a record with the message as a string and id */
3061   str = message_to_string (watch->conn, message);
3062   size = str->len;
3063   return gst_rtsp_watch_queue_data (watch,
3064       (guint8 *) g_string_free (str, FALSE), size);
3065 }