gst/tcp/gsttcpserversrc.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  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  */
  20
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include "config.h"
  24 #endif
  25
  26 #include <gst/gst-i18n-plugin.h>
  27 #include "gsttcp.h"
  28 #include "gsttcpserversrc.h"
  29 #include <unistd.h>
  30 #include <sys/ioctl.h>
  31
  32 #ifdef HAVE_FIONREAD_IN_SYS_FILIO
  33 #include <sys/filio.h>
  34 #endif
  35
  36 GST_DEBUG_CATEGORY (tcpserversrc_debug);
  37 #define GST_CAT_DEFAULT tcpserversrc_debug
  38
  39 #define TCP_DEFAULT_PORT                4953
  40 #define TCP_DEFAULT_HOST                NULL    /* listen on all interfaces */
  41 #define TCP_BACKLOG                     1       /* client connection queue */
  42
  43 /* elementfactory information */
  44 static GstElementDetails gst_tcpserversrc_details =
  45 GST_ELEMENT_DETAILS ("TCP Server source",
  46     "Source/Network",
  47     "Receive data as a server over the network via TCP",
  48     "Thomas Vander Stichele <thomas at apestaart dot org>");
  49
  50 /* TCPServerSrc signals and args */
  51 enum
  52 {
  53   LAST_SIGNAL
  54 };
  55
  56 enum
  57 {
  58   ARG_0,
  59   ARG_HOST,
  60   ARG_PORT,
  61   ARG_PROTOCOL
  62 };
  63
  64 static void gst_tcpserversrc_base_init (gpointer g_class);
  65 static void gst_tcpserversrc_class_init (GstTCPServerSrc * klass);
  66 static void gst_tcpserversrc_init (GstTCPServerSrc * tcpserversrc);
  67
  68 static GstData *gst_tcpserversrc_get (GstPad * pad);
  69 static GstElementStateReturn gst_tcpserversrc_change_state (GstElement *
  70     element);
  71
  72 static void gst_tcpserversrc_set_property (GObject * object, guint prop_id,
  73     const GValue * value, GParamSpec * pspec);
  74 static void gst_tcpserversrc_get_property (GObject * object, guint prop_id,
  75     GValue * value, GParamSpec * pspec);
  76 static void gst_tcpserversrc_set_clock (GstElement * element, GstClock * clock);
  77
  78 static GstElementClass *parent_class = NULL;
  79
  80 /*static guint gst_tcpserversrc_signals[LAST_SIGNAL] = { 0 }; */
  81
  82 GType
  83 gst_tcpserversrc_get_type (void)
  84 {
  85   static GType tcpserversrc_type = 0;
  86
  87
  88   if (!tcpserversrc_type) {
  89     static const GTypeInfo tcpserversrc_info = {
  90       sizeof (GstTCPServerSrcClass),
  91       gst_tcpserversrc_base_init,
  92       NULL,
  93       (GClassInitFunc) gst_tcpserversrc_class_init,
  94       NULL,
  95       NULL,
  96       sizeof (GstTCPServerSrc),
  97       0,
  98       (GInstanceInitFunc) gst_tcpserversrc_init,
  99       NULL
 100     };
 101
 102     tcpserversrc_type =
 103         g_type_register_static (GST_TYPE_ELEMENT, "GstTCPServerSrc",
 104         &tcpserversrc_info, 0);
 105   }
 106   return tcpserversrc_type;
 107 }
 108
 109 static void
 110 gst_tcpserversrc_base_init (gpointer g_class)
 111 {
 112   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 113
 114   gst_element_class_set_details (element_class, &gst_tcpserversrc_details);
 115 }
 116
 117 static void
 118 gst_tcpserversrc_class_init (GstTCPServerSrc * klass)
 119 {
 120   GObjectClass *gobject_class;
 121   GstElementClass *gstelement_class;
 122
 123   gobject_class = (GObjectClass *) klass;
 124   gstelement_class = (GstElementClass *) klass;
 125
 126   parent_class = g_type_class_ref (GST_TYPE_ELEMENT);
 127
 128   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_HOST,
 129       g_param_spec_string ("host", "Host", "The hostname to listen",
 130           TCP_DEFAULT_HOST, G_PARAM_READWRITE));
 131   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_PORT,
 132       g_param_spec_int ("port", "Port", "The port to listen to",
 133           0, 32768, TCP_DEFAULT_PORT, G_PARAM_READWRITE));
 134   g_object_class_install_property (gobject_class, ARG_PROTOCOL,
 135       g_param_spec_enum ("protocol", "Protocol", "The protocol to wrap data in",
 136           GST_TYPE_TCP_PROTOCOL_TYPE, GST_TCP_PROTOCOL_TYPE_NONE,
 137           G_PARAM_READWRITE));
 138
 139   gobject_class->set_property = gst_tcpserversrc_set_property;
 140   gobject_class->get_property = gst_tcpserversrc_get_property;
 141
 142   gstelement_class->change_state = gst_tcpserversrc_change_state;
 143   gstelement_class->set_clock = gst_tcpserversrc_set_clock;
 144
 145   GST_DEBUG_CATEGORY_INIT (tcpserversrc_debug, "tcpserversrc", 0,
 146       "TCP Server Source");
 147 }
 148
 149 static void
 150 gst_tcpserversrc_set_clock (GstElement * element, GstClock * clock)
 151 {
 152   GstTCPServerSrc *tcpserversrc;
 153
 154   tcpserversrc = GST_TCPSERVERSRC (element);
 155
 156   tcpserversrc->clock = clock;
 157 }
 158
 159 static void
 160 gst_tcpserversrc_init (GstTCPServerSrc * this)
 161 {
 162   /* create the src pad */
 163   this->srcpad = gst_pad_new ("src", GST_PAD_SRC);
 164   gst_element_add_pad (GST_ELEMENT (this), this->srcpad);
 165   gst_pad_set_get_function (this->srcpad, gst_tcpserversrc_get);
 166
 167   this->server_port = TCP_DEFAULT_PORT;
 168   this->host = TCP_DEFAULT_HOST;
 169   this->clock = NULL;
 170   this->server_sock_fd = -1;
 171   this->client_sock_fd = -1;
 172   this->curoffset = 0;
 173   this->protocol = GST_TCP_PROTOCOL_TYPE_NONE;
 174
 175   GST_FLAG_UNSET (this, GST_TCPSERVERSRC_OPEN);
 176 }
 177
 178 /* read the gdp caps packet from the socket */
 179 static GstCaps *
 180 gst_tcpserversrc_gdp_read_caps (GstTCPServerSrc * this)
 181 {
 182   size_t header_length = GST_DP_HEADER_LENGTH;
 183   size_t readsize;
 184   guint8 *header = NULL;
 185   guint8 *payload = NULL;
 186   ssize_t ret;
 187   GstCaps *caps;
 188   gchar *string;
 189
 190   header = g_malloc (header_length);
 191
 192   readsize = header_length;
 193   GST_LOG_OBJECT (this, "Reading %d bytes for caps packet header", readsize);
 194   ret = read (this->client_sock_fd, header, readsize);
 195   if (ret < 0) {
 196     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 197     g_free (header);
 198     return NULL;
 199   }
 200   g_assert (ret == readsize);
 201
 202   if (!gst_dp_validate_header (header_length, header)) {
 203     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 204         ("GDP caps packet header does not validate"));
 205     g_free (header);
 206     return NULL;
 207   }
 208
 209   readsize = gst_dp_header_payload_length (header);
 210   payload = g_malloc (readsize);
 211   GST_LOG_OBJECT (this, "Reading %d bytes for caps packet payload", readsize);
 212   ret = read (this->client_sock_fd, payload, readsize);
 213   if (ret < 0) {
 214     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 215     g_free (header);
 216     g_free (payload);
 217     return NULL;
 218   }
 219   g_assert (ret == readsize);
 220
 221   if (!gst_dp_validate_payload (readsize, header, payload)) {
 222     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 223         ("GDP caps packet payload does not validate"));
 224     g_free (header);
 225     g_free (payload);
 226     return NULL;
 227   }
 228
 229   caps = gst_dp_caps_from_packet (header_length, header, payload);
 230   string = gst_caps_to_string (caps);
 231   GST_DEBUG_OBJECT (this, "retrieved GDP caps from packet payload: %s", string);
 232   g_free (string);
 233
 234   g_free (header);
 235   g_free (payload);
 236
 237   return caps;
 238 }
 239
 240 /* read the gdp buffer header from the socket
 241  * returns a GstData,
 242  * representing the new GstBuffer to read data into, or an EOS event
 243  */
 244 static GstData *
 245 gst_tcpserversrc_gdp_read_header (GstTCPServerSrc * this)
 246 {
 247   size_t header_length = GST_DP_HEADER_LENGTH;
 248   size_t readsize;
 249   guint8 *header = NULL;
 250   ssize_t ret;
 251   GstBuffer *buffer;
 252
 253   header = g_malloc (header_length);
 254   readsize = header_length;
 255
 256   GST_LOG_OBJECT (this, "Reading %d bytes for buffer packet header", readsize);
 257   ret = read (this->client_sock_fd, header, readsize);
 258   /* if we read 0 bytes, and we're blocking, we hit eos */
 259   if (ret == 0) {
 260     GST_DEBUG ("blocking read returns 0, EOS");
 261     gst_element_set_eos (GST_ELEMENT (this));
 262     g_free (header);
 263     return GST_DATA (gst_event_new (GST_EVENT_EOS));
 264   }
 265   if (ret < 0) {
 266     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 267     g_free (header);
 268     return NULL;
 269   }
 270   if (ret != readsize) {
 271     g_warning ("Wanted %d bytes, got %d bytes", (int) readsize, (int) ret);
 272   }
 273   g_assert (ret == readsize);
 274
 275   if (!gst_dp_validate_header (header_length, header)) {
 276     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 277         ("GDP buffer packet header does not validate"));
 278     g_free (header);
 279     return NULL;
 280   }
 281   GST_LOG_OBJECT (this, "validated buffer packet header");
 282
 283   buffer = gst_dp_buffer_from_header (header_length, header);
 284
 285   GST_LOG_OBJECT (this, "created new buffer %p from packet header", buffer);
 286   return GST_DATA (buffer);
 287 }
 288
 289 static GstData *
 290 gst_tcpserversrc_get (GstPad * pad)
 291 {
 292   GstTCPServerSrc *src;
 293   size_t readsize;
 294   int ret;
 295
 296   GstData *data = NULL;
 297   GstBuffer *buf = NULL;
 298   GstCaps *caps;
 299
 300   g_return_val_if_fail (pad != NULL, NULL);
 301   g_return_val_if_fail (GST_IS_PAD (pad), NULL);
 302   src = GST_TCPSERVERSRC (GST_OBJECT_PARENT (pad));
 303   g_return_val_if_fail (GST_FLAG_IS_SET (src, GST_TCPSERVERSRC_OPEN), NULL);
 304
 305   /* read the buffer header if we're using a protocol */
 306   switch (src->protocol) {
 307       fd_set testfds;
 308
 309     case GST_TCP_PROTOCOL_TYPE_NONE:
 310       /* do a blocking select on the socket */
 311       FD_ZERO (&testfds);
 312       FD_SET (src->client_sock_fd, &testfds);
 313       ret =
 314           select (src->client_sock_fd + 1, &testfds, (fd_set *) 0, (fd_set *) 0,
 315           0);
 316       /* no action (0) is an error too in our case */
 317       if (ret <= 0) {
 318         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 319             ("select failed: %s", g_strerror (errno)));
 320         return NULL;
 321       }
 322       /* ask how much is available for reading on the socket */
 323       ret = ioctl (src->client_sock_fd, FIONREAD, &readsize);
 324       if (ret < 0) {
 325         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 326             ("ioctl failed: %s", g_strerror (errno)));
 327         return NULL;
 328       }
 329
 330       buf = gst_buffer_new_and_alloc (readsize);
 331       break;
 332     case GST_TCP_PROTOCOL_TYPE_GDP:
 333       /* if we haven't received caps yet, we should get them first */
 334       if (!src->caps_received) {
 335         gchar *string;
 336
 337         if (!(caps = gst_tcpserversrc_gdp_read_caps (src))) {
 338           GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 339               ("Could not read caps through GDP"));
 340           return NULL;
 341         }
 342         src->caps_received = TRUE;
 343         string = gst_caps_to_string (caps);
 344         GST_DEBUG_OBJECT (src, "Received caps through GDP: %s", string);
 345         g_free (string);
 346
 347         if (!gst_pad_try_set_caps (pad, caps)) {
 348           g_warning ("Could not set caps");
 349           return NULL;
 350         }
 351       }
 352
 353       /* now receive the buffer header */
 354       if (!(data = gst_tcpserversrc_gdp_read_header (src))) {
 355         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 356             ("Could not read data header through GDP"));
 357         return NULL;
 358       }
 359       if (GST_IS_EVENT (data))
 360         return data;
 361       buf = GST_BUFFER (data);
 362
 363       GST_LOG_OBJECT (src, "Going to read data from socket into buffer %p",
 364           buf);
 365       /* use this new buffer to read data into */
 366       readsize = GST_BUFFER_SIZE (buf);
 367       break;
 368     default:
 369       g_warning ("Unhandled protocol type");
 370       break;
 371   }
 372
 373   GST_LOG_OBJECT (src, "Reading %d bytes", readsize);
 374   ret =
 375       gst_tcp_socket_read (src->client_sock_fd, GST_BUFFER_DATA (buf),
 376       readsize);
 377   if (ret < 0) {
 378     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 379     gst_buffer_unref (buf);
 380     return NULL;
 381   }
 382
 383   /* if we read 0 bytes, and we're blocking, we hit eos */
 384   if (ret == 0) {
 385     GST_DEBUG ("blocking read returns 0, EOS");
 386     gst_buffer_unref (buf);
 387     gst_element_set_eos (GST_ELEMENT (src));
 388     return GST_DATA (gst_event_new (GST_EVENT_EOS));
 389   }
 390
 391   readsize = ret;
 392   GST_LOG_OBJECT (src, "Read %d bytes", readsize);
 393   GST_BUFFER_SIZE (buf) = readsize;
 394   GST_BUFFER_MAXSIZE (buf) = readsize;
 395   GST_BUFFER_OFFSET (buf) = src->curoffset;
 396   GST_BUFFER_OFFSET_END (buf) = src->curoffset + readsize;
 397   src->curoffset += readsize;
 398   return GST_DATA (buf);
 399 }
 400
 401
 402 static void
 403 gst_tcpserversrc_set_property (GObject * object, guint prop_id,
 404     const GValue * value, GParamSpec * pspec)
 405 {
 406   GstTCPServerSrc *tcpserversrc;
 407
 408   g_return_if_fail (GST_IS_TCPSERVERSRC (object));
 409   tcpserversrc = GST_TCPSERVERSRC (object);
 410
 411   switch (prop_id) {
 412     case ARG_HOST:
 413       g_free (tcpserversrc->host);
 414       tcpserversrc->host = g_strdup (g_value_get_string (value));
 415       break;
 416     case ARG_PORT:
 417       tcpserversrc->server_port = g_value_get_int (value);
 418       break;
 419     case ARG_PROTOCOL:
 420       tcpserversrc->protocol = g_value_get_enum (value);
 421       break;
 422
 423     default:
 424       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 425       break;
 426   }
 427 }
 428
 429 static void
 430 gst_tcpserversrc_get_property (GObject * object, guint prop_id, GValue * value,
 431     GParamSpec * pspec)
 432 {
 433   GstTCPServerSrc *tcpserversrc;
 434
 435   g_return_if_fail (GST_IS_TCPSERVERSRC (object));
 436   tcpserversrc = GST_TCPSERVERSRC (object);
 437
 438   switch (prop_id) {
 439     case ARG_HOST:
 440       g_value_set_string (value, tcpserversrc->host);
 441       break;
 442     case ARG_PORT:
 443       g_value_set_int (value, tcpserversrc->server_port);
 444       break;
 445     case ARG_PROTOCOL:
 446       g_value_set_enum (value, tcpserversrc->protocol);
 447       break;
 448
 449     default:
 450       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 451       break;
 452   }
 453 }
 454
 455 /* set up server */
 456 static gboolean
 457 gst_tcpserversrc_init_receive (GstTCPServerSrc * this)
 458 {
 459   int ret;
 460
 461   /* reset caps_received flag */
 462   this->caps_received = FALSE;
 463
 464   /* create the server listener socket */
 465   if ((this->server_sock_fd = socket (AF_INET, SOCK_STREAM, 0)) == -1) {
 466     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL), GST_ERROR_SYSTEM);
 467     return FALSE;
 468   }
 469   GST_DEBUG_OBJECT (this, "opened receiving server socket with fd %d",
 470       this->server_sock_fd);
 471
 472   /* make address reusable */
 473   if (setsockopt (this->server_sock_fd, SOL_SOCKET, SO_REUSEADDR, &ret,
 474           sizeof (int)) < 0) {
 475     GST_ELEMENT_ERROR (this, RESOURCE, SETTINGS, (NULL),
 476         ("Could not setsockopt: %s", g_strerror (errno)));
 477     return FALSE;
 478   }
 479
 480   /* name the socket */
 481   memset (&this->server_sin, 0, sizeof (this->server_sin));
 482   this->server_sin.sin_family = AF_INET;        /* network socket */
 483   this->server_sin.sin_port = htons (this->server_port);        /* on port */
 484   if (this->host) {
 485     gchar *host = gst_tcp_host_to_ip (GST_ELEMENT (this), this->host);
 486
 487     if (!host)
 488       return FALSE;
 489
 490     this->server_sin.sin_addr.s_addr = inet_addr (host);
 491     g_free (host);
 492   } else
 493     this->server_sin.sin_addr.s_addr = htonl (INADDR_ANY);
 494
 495   /* bind it */
 496   GST_DEBUG_OBJECT (this, "binding server socket to address");
 497   ret = bind (this->server_sock_fd, (struct sockaddr *) &this->server_sin,
 498       sizeof (this->server_sin));
 499
 500   if (ret) {
 501     switch (errno) {
 502       default:
 503         GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 504             ("bind failed: %s", g_strerror (errno)));
 505         return FALSE;
 506         break;
 507     }
 508   }
 509
 510   GST_DEBUG_OBJECT (this, "listening on server socket %d with queue of %d",
 511       this->server_sock_fd, TCP_BACKLOG);
 512   if (listen (this->server_sock_fd, TCP_BACKLOG) == -1) {
 513     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 514         ("Could not listen on server socket: %s", g_strerror (errno)));
 515     return FALSE;
 516   }
 517
 518   /* FIXME: maybe we should think about moving actual client accepting
 519      somewhere else */
 520   GST_DEBUG_OBJECT (this, "waiting for client");
 521   this->client_sock_fd =
 522       accept (this->server_sock_fd, (struct sockaddr *) &this->client_sin,
 523       &this->client_sin_len);
 524   if (this->client_sock_fd == -1) {
 525     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 526         ("Could not accept client on server socket: %s", g_strerror (errno)));
 527     return FALSE;
 528   }
 529   GST_DEBUG_OBJECT (this, "received client");
 530
 531   GST_FLAG_SET (this, GST_TCPSERVERSRC_OPEN);
 532   return TRUE;
 533 }
 534
 535 static void
 536 gst_tcpserversrc_close (GstTCPServerSrc * this)
 537 {
 538   if (this->server_sock_fd != -1) {
 539     close (this->server_sock_fd);
 540     this->server_sock_fd = -1;
 541   }
 542   if (this->client_sock_fd != -1) {
 543     close (this->client_sock_fd);
 544     this->client_sock_fd = -1;
 545   }
 546   GST_FLAG_UNSET (this, GST_TCPSERVERSRC_OPEN);
 547 }
 548
 549 static GstElementStateReturn
 550 gst_tcpserversrc_change_state (GstElement * element)
 551 {
 552   g_return_val_if_fail (GST_IS_TCPSERVERSRC (element), GST_STATE_FAILURE);
 553
 554   if (GST_STATE_PENDING (element) == GST_STATE_NULL) {
 555     if (GST_FLAG_IS_SET (element, GST_TCPSERVERSRC_OPEN))
 556       gst_tcpserversrc_close (GST_TCPSERVERSRC (element));
 557   } else {
 558     if (!GST_FLAG_IS_SET (element, GST_TCPSERVERSRC_OPEN)) {
 559       if (!gst_tcpserversrc_init_receive (GST_TCPSERVERSRC (element)))
 560         return GST_STATE_FAILURE;
 561     }
 562   }
 563
 564   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 565     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 566
 567   return GST_STATE_SUCCESS;
 568 }