gst/tcp/gsttcp.c

   1 /* GStreamer
   2  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
   3  * Copyright (C) <2004> Thomas Vander Stichele <thomas at apestaart dot org>
   4  *
   5  * gsttcp.c: TCP functions
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, USA.
  21  */
  22
  23 #ifdef HAVE_CONFIG_H
  24 #include "config.h"
  25 #endif
  26
  27 #include <sys/types.h>
  28 #include <sys/socket.h>
  29 #include <netinet/in.h>
  30 #include <arpa/inet.h>
  31 #include <netdb.h>
  32 #include <unistd.h>
  33
  34 #include <glib.h>
  35 #include <gst/gst.h>
  36 #include <gst/gst-i18n-plugin.h>
  37 #include <gst/dataprotocol/dataprotocol.h>
  38
  39 GST_DEBUG_CATEGORY_EXTERN (tcp_debug);
  40 #define GST_CAT_DEFAULT tcp_debug
  41
  42 #ifndef MSG_NOSIGNAL
  43 #define MSG_NOSIGNAL 0
  44 #endif
  45
  46 /* resolve host to IP address, throwing errors if it fails */
  47 /* host can already be an IP address */
  48 /* returns a newly allocated gchar * with the dotted ip address,
  49    or NULL, in which case it already fired an error. */
  50 gchar *
  51 gst_tcp_host_to_ip (GstElement * element, const gchar * host)
  52 {
  53   struct hostent *hostinfo;
  54   char **addrs;
  55   gchar *ip;
  56   struct in_addr addr;
  57
  58   GST_DEBUG_OBJECT (element, "resolving host %s", host);
  59   /* first check if it already is an IP address */
  60   if (inet_aton (host, &addr)) {
  61     ip = g_strdup (host);
  62     goto beach;
  63   }
  64
  65   /* FIXME: could do a localhost check here */
  66
  67   /* perform a name lookup */
  68   hostinfo = gethostbyname (host);
  69   if (!hostinfo) {
  70     GST_ELEMENT_ERROR (element, RESOURCE, NOT_FOUND, (NULL),
  71         ("Could not find IP address for host \"%s\".", host));
  72     return NULL;
  73   }
  74
  75   if (hostinfo->h_addrtype != AF_INET) {
  76     GST_ELEMENT_ERROR (element, RESOURCE, NOT_FOUND, (NULL),
  77         ("host \"%s\" is not an IP host", host));
  78     return NULL;
  79   }
  80
  81   addrs = hostinfo->h_addr_list;
  82   /* There could be more than one IP address, but we just return the first */
  83   ip = g_strdup (inet_ntoa (*(struct in_addr *) *addrs));
  84
  85 beach:
  86   GST_DEBUG_OBJECT (element, "resolved to IP %s", ip);
  87   return ip;
  88 }
  89
  90 /* write buffer to given socket incrementally.
  91  * Returns number of bytes written.
  92  */
  93 gint
  94 gst_tcp_socket_write (int socket, const void *buf, size_t count)
  95 {
  96   size_t bytes_written = 0;
  97
  98   while (bytes_written < count) {
  99     ssize_t wrote = send (socket, buf + bytes_written,
 100         count - bytes_written, MSG_NOSIGNAL);
 101
 102     if (wrote <= 0) {
 103       return bytes_written;
 104     }
 105     bytes_written += wrote;
 106   }
 107
 108   if (bytes_written < 0)
 109     GST_WARNING ("error while writing");
 110   else
 111     GST_LOG ("wrote %d bytes succesfully", bytes_written);
 112   return bytes_written;
 113 }
 114
 115 /* read number of bytes from a socket into a given buffer incrementally.
 116  * Returns number of bytes read with same semantics as read(2):
 117  * < 0: error, see errno
 118  * = 0: EOF
 119  * > 0: bytes read
 120  */
 121 gint
 122 gst_tcp_socket_read (int socket, void *buf, size_t count)
 123 {
 124   size_t bytes_read = 0;
 125
 126   while (bytes_read < count) {
 127     ssize_t ret = read (socket, buf + bytes_read,
 128         count - bytes_read);
 129
 130     if (ret < 0)
 131       GST_WARNING ("error while reading: %s", g_strerror (errno));
 132     if (ret <= 0)
 133       return bytes_read;
 134     bytes_read += ret;
 135   }
 136
 137   if (bytes_read < 0)
 138     GST_WARNING ("error while reading: %s", g_strerror (errno));
 139   else
 140     GST_LOG ("read %d bytes succesfully", bytes_read);
 141   return bytes_read;
 142 }
 143
 144 /* close the socket and reset the fd.  Used to clean up after errors. */
 145 void
 146 gst_tcp_socket_close (int *socket)
 147 {
 148   close (*socket);
 149   *socket = -1;
 150 }
 151
 152 /* read the gdp buffer header from the given socket
 153  * returns:
 154  * - a GstData representing a GstBuffer in which data should be read
 155  * - a GstData representing a GstEvent
 156  * - NULL, indicating a connection close or an error, to be handled with
 157  *         EOS
 158  */
 159 GstData *
 160 gst_tcp_gdp_read_header (GstElement * this, int socket)
 161 {
 162   size_t header_length = GST_DP_HEADER_LENGTH;
 163   size_t readsize;
 164   guint8 *header = NULL;
 165   ssize_t ret;
 166   GstBuffer *buffer;
 167
 168   header = g_malloc (header_length);
 169   readsize = header_length;
 170
 171   GST_LOG_OBJECT (this, "Reading %d bytes for buffer packet header", readsize);
 172   ret = gst_tcp_socket_read (socket, header, readsize);
 173   /* if we read 0 bytes, and we're blocking, we hit eos */
 174   if (ret == 0) {
 175     GST_DEBUG ("blocking read returns 0, returning NULL");
 176     g_free (header);
 177     return NULL;
 178   }
 179   if (ret < 0) {
 180     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 181     g_free (header);
 182     return NULL;
 183   }
 184   if (ret != readsize) {
 185     g_warning ("Wanted %d bytes, got %d bytes", (int) readsize, (int) ret);
 186   }
 187   g_assert (ret == readsize);
 188
 189   if (!gst_dp_validate_header (header_length, header)) {
 190     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 191         ("GDP buffer packet header does not validate"));
 192     g_free (header);
 193     return NULL;
 194   }
 195   GST_LOG_OBJECT (this, "validated buffer packet header");
 196
 197   buffer = gst_dp_buffer_from_header (header_length, header);
 198   g_free (header);
 199
 200   GST_LOG_OBJECT (this, "created new buffer %p from packet header", buffer);
 201   return GST_DATA (buffer);
 202 }
 203
 204 /* read the GDP caps packet from the given socket
 205  * returns the caps, or NULL in case of an error */
 206 GstCaps *
 207 gst_tcp_gdp_read_caps (GstElement * this, int socket)
 208 {
 209   size_t header_length = GST_DP_HEADER_LENGTH;
 210   size_t readsize;
 211   guint8 *header = NULL;
 212   guint8 *payload = NULL;
 213   ssize_t ret;
 214   GstCaps *caps;
 215   gchar *string;
 216
 217   header = g_malloc (header_length);
 218   readsize = header_length;
 219
 220   GST_LOG_OBJECT (this, "Reading %d bytes for caps packet header", readsize);
 221   ret = gst_tcp_socket_read (socket, header, readsize);
 222   if (ret < 0) {
 223     g_free (header);
 224     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 225     return NULL;
 226   }
 227   if (ret == 0) {
 228     GST_WARNING_OBJECT (this, "read returned EOF");
 229     return NULL;
 230   }
 231   if (ret != readsize) {
 232     GST_WARNING_OBJECT (this, "Tried to read %d bytes but only read %d bytes",
 233         readsize, ret);
 234     return NULL;
 235   }
 236
 237   if (!gst_dp_validate_header (header_length, header)) {
 238     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 239         ("GDP caps packet header does not validate"));
 240     g_free (header);
 241     return NULL;
 242   }
 243
 244   readsize = gst_dp_header_payload_length (header);
 245   payload = g_malloc (readsize);
 246   GST_LOG_OBJECT (this, "Reading %d bytes for caps packet payload", readsize);
 247   ret = gst_tcp_socket_read (socket, payload, readsize);
 248
 249   if (ret < 0) {
 250     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 251     g_free (header);
 252     g_free (payload);
 253     return NULL;
 254   }
 255   if (gst_dp_header_payload_type (header) != GST_DP_PAYLOAD_CAPS) {
 256     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 257         ("Header read doesn't describe CAPS payload"));
 258     g_free (header);
 259     g_free (payload);
 260     return NULL;
 261   }
 262   g_assert (ret == readsize);
 263
 264   if (!gst_dp_validate_payload (readsize, header, payload)) {
 265     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 266         ("GDP caps packet payload does not validate"));
 267     g_free (header);
 268     g_free (payload);
 269     return NULL;
 270   }
 271
 272   caps = gst_dp_caps_from_packet (header_length, header, payload);
 273   string = gst_caps_to_string (caps);
 274   GST_LOG_OBJECT (this, "retrieved GDP caps from packet payload: %s", string);
 275   g_free (string);
 276
 277   g_free (header);
 278   g_free (payload);
 279
 280   return caps;
 281 }
 282
 283 /* write a GDP header to the socket.  Return false if fails. */
 284 gboolean
 285 gst_tcp_gdp_write_header (GstElement * this, int socket, GstBuffer * buffer,
 286     gboolean fatal, const gchar * host, int port)
 287 {
 288   guint length;
 289   guint8 *header;
 290   size_t wrote;
 291
 292   if (!gst_dp_header_from_buffer (buffer, 0, &length, &header)) {
 293     if (fatal)
 294       GST_ELEMENT_ERROR (this, CORE, TOO_LAZY, (NULL),
 295           ("Could not create GDP header from buffer"));
 296     return FALSE;
 297   }
 298
 299   GST_LOG_OBJECT (this, "writing %d bytes for GDP buffer header", length);
 300   wrote = gst_tcp_socket_write (socket, header, length);
 301   g_free (header);
 302   if (wrote != length) {
 303     if (fatal)
 304       GST_ELEMENT_ERROR (this, RESOURCE, WRITE,
 305           (_("Error while sending data to \"%s:%d\"."), host, port),
 306           ("Only %d of %d bytes written: %s",
 307               wrote, GST_BUFFER_SIZE (buffer), g_strerror (errno)));
 308     return FALSE;
 309   }
 310
 311   return TRUE;
 312 }
 313
 314 /* write GDP header and payload to the given socket for the given caps.
 315  * Return false if fails. */
 316 gboolean
 317 gst_tcp_gdp_write_caps (GstElement * this, int socket, const GstCaps * caps,
 318     gboolean fatal, const char *host, int port)
 319 {
 320   guint length;
 321   guint8 *header;
 322   guint8 *payload;
 323   size_t wrote;
 324
 325   if (!gst_dp_packet_from_caps (caps, 0, &length, &header, &payload)) {
 326     if (fatal)
 327       GST_ELEMENT_ERROR (this, CORE, TOO_LAZY, (NULL),
 328           ("Could not create GDP packet from caps"));
 329     return FALSE;
 330   }
 331   GST_LOG_OBJECT (this, "writing %d bytes for GDP caps header", length);
 332   wrote = gst_tcp_socket_write (socket, header, length);
 333   if (wrote != length) {
 334     g_free (header);
 335     g_free (payload);
 336     if (fatal)
 337       GST_ELEMENT_ERROR (this, RESOURCE, WRITE,
 338           (_("Error while sending gdp header data to \"%s:%d\"."), host, port),
 339           ("Only %d of %d bytes written: %s",
 340               wrote, length, g_strerror (errno)));
 341     return FALSE;
 342   }
 343
 344   length = gst_dp_header_payload_length (header);
 345   g_free (header);
 346   GST_LOG_OBJECT (this, "writing %d bytes for GDP caps payload", length);
 347   wrote = gst_tcp_socket_write (socket, payload, length);
 348   g_free (payload);
 349   if (wrote != length) {
 350     if (fatal)
 351       GST_ELEMENT_ERROR (this, RESOURCE, WRITE,
 352           (_("Error while sending gdp payload data to \"%s:%d\"."), host, port),
 353           ("Only %d of %d bytes written: %s",
 354               wrote, length, g_strerror (errno)));
 355     return FALSE;
 356   }
 357   return TRUE;
 358 }