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_PORT,
  60   ARG_HOST,
  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_PORT,
 129       g_param_spec_int ("port", "Port", "The port to listen to",
 130           0, 32768, TCP_DEFAULT_PORT, G_PARAM_READWRITE));
 131   g_object_class_install_property (gobject_class, ARG_PROTOCOL,
 132       g_param_spec_enum ("protocol", "Protocol", "The protocol to wrap data in",
 133           GST_TYPE_TCP_PROTOCOL_TYPE, GST_TCP_PROTOCOL_TYPE_GDP,
 134           G_PARAM_READWRITE));
 135   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_HOST,
 136       g_param_spec_string ("host", "Host", "The hostname to listen",
 137           TCP_DEFAULT_HOST, 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_GDP;
 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     return NULL;
 198   }
 199   g_assert (ret == readsize);
 200
 201   if (!gst_dp_validate_header (header_length, header)) {
 202     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 203         ("GDP caps packet header does not validate"));
 204     g_free (header);
 205     return NULL;
 206   }
 207
 208   readsize = gst_dp_header_payload_length (header);
 209   payload = g_malloc (readsize);
 210   GST_LOG_OBJECT (this, "Reading %d bytes for caps packet payload", readsize);
 211   ret = read (this->client_sock_fd, payload, readsize);
 212   if (ret < 0) {
 213     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 214     g_free (header);
 215     return NULL;
 216   }
 217   g_assert (ret == readsize);
 218
 219   if (!gst_dp_validate_payload (readsize, header, payload)) {
 220     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 221         ("GDP caps packet payload does not validate"));
 222     g_free (header);
 223     g_free (payload);
 224     return NULL;
 225   }
 226
 227   caps = gst_dp_caps_from_packet (header_length, header, payload);
 228   string = gst_caps_to_string (caps);
 229   GST_DEBUG_OBJECT (this, "retrieved GDP caps from packet payload: %s", string);
 230
 231   g_free (header);
 232   g_free (payload);
 233   g_free (string);
 234   return caps;
 235 }
 236
 237 /* read the gdp buffer header from the socket
 238  * returns a GstData,
 239  * representing the new GstBuffer to read data into, or an EOS event
 240  */
 241 static GstData *
 242 gst_tcpserversrc_gdp_read_header (GstTCPServerSrc * this)
 243 {
 244   size_t header_length = GST_DP_HEADER_LENGTH;
 245   size_t readsize;
 246   guint8 *header = NULL;
 247   ssize_t ret;
 248   GstBuffer *buffer;
 249
 250   header = g_malloc (header_length);
 251
 252   readsize = header_length;
 253   GST_LOG_OBJECT (this, "Reading %d bytes for buffer packet header", readsize);
 254   ret = read (this->client_sock_fd, header, readsize);
 255   /* if we read 0 bytes, and we're blocking, we hit eos */
 256   if (ret == 0) {
 257     GST_DEBUG ("blocking read returns 0, EOS");
 258     gst_element_set_eos (GST_ELEMENT (this));
 259     return GST_DATA (gst_event_new (GST_EVENT_EOS));
 260   }
 261   if (ret < 0) {
 262     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 263     return NULL;
 264   }
 265   if (ret != readsize) {
 266     g_warning ("Wanted %d bytes, got %d bytes", (int) readsize, (int) ret);
 267   }
 268   g_assert (ret == readsize);
 269
 270   if (!gst_dp_validate_header (header_length, header)) {
 271     GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL),
 272         ("GDP buffer packet header does not validate"));
 273     g_free (header);
 274     return NULL;
 275   }
 276   GST_LOG_OBJECT (this, "validated buffer packet header");
 277
 278   buffer = gst_dp_buffer_from_header (header_length, header);
 279
 280   GST_LOG_OBJECT (this, "created new buffer %p from packet header", buffer);
 281   return GST_DATA (buffer);
 282 }
 283
 284 static GstData *
 285 gst_tcpserversrc_get (GstPad * pad)
 286 {
 287   GstTCPServerSrc *src;
 288   size_t readsize;
 289   int ret;
 290
 291   GstData *data = NULL;
 292   GstBuffer *buf = NULL;
 293   GstCaps *caps;
 294
 295   g_return_val_if_fail (pad != NULL, NULL);
 296   g_return_val_if_fail (GST_IS_PAD (pad), NULL);
 297   src = GST_TCPSERVERSRC (GST_OBJECT_PARENT (pad));
 298   g_return_val_if_fail (GST_FLAG_IS_SET (src, GST_TCPSERVERSRC_OPEN), NULL);
 299
 300   /* read the buffer header if we're using a protocol */
 301   switch (src->protocol) {
 302       fd_set testfds;
 303
 304     case GST_TCP_PROTOCOL_TYPE_NONE:
 305       /* do a blocking select on the socket */
 306       FD_ZERO (&testfds);
 307       FD_SET (src->client_sock_fd, &testfds);
 308       ret =
 309           select (src->client_sock_fd + 1, &testfds, (fd_set *) 0, (fd_set *) 0,
 310           0);
 311       /* no action (0) is an error too in our case */
 312       if (ret <= 0) {
 313         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 314             ("select failed: %s", g_strerror (errno)));
 315         return NULL;
 316       }
 317       /* ask how much is available for reading on the socket */
 318       ret = ioctl (src->client_sock_fd, FIONREAD, &readsize);
 319       if (ret < 0) {
 320         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 321             ("ioctl failed: %s", g_strerror (errno)));
 322         return NULL;
 323       }
 324
 325       buf = gst_buffer_new_and_alloc (readsize);
 326       break;
 327     case GST_TCP_PROTOCOL_TYPE_GDP:
 328       /* if we haven't received caps yet, we should get them first */
 329       if (!src->caps_received) {
 330         gchar *string;
 331
 332         if (!(caps = gst_tcpserversrc_gdp_read_caps (src))) {
 333           GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 334               ("Could not read caps through GDP"));
 335           return NULL;
 336         }
 337         src->caps_received = TRUE;
 338         string = gst_caps_to_string (caps);
 339         GST_DEBUG_OBJECT (src, "Received caps through GDP: %s", string);
 340         g_free (string);
 341
 342         if (!gst_pad_try_set_caps (pad, caps)) {
 343           g_warning ("Could not set caps");
 344           return NULL;
 345         }
 346       }
 347
 348       /* now receive the buffer header */
 349       if (!(data = gst_tcpserversrc_gdp_read_header (src))) {
 350         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 351             ("Could not read data header through GDP"));
 352         return NULL;
 353       }
 354       if (GST_IS_EVENT (data))
 355         return data;
 356       buf = GST_BUFFER (data);
 357
 358       GST_LOG_OBJECT (src, "Going to read data from socket into buffer %p",
 359           buf);
 360       /* use this new buffer to read data into */
 361       readsize = GST_BUFFER_SIZE (buf);
 362       break;
 363     default:
 364       g_warning ("Unhandled protocol type");
 365       break;
 366   }
 367
 368   GST_LOG_OBJECT (src, "Reading %d bytes", readsize);
 369   ret =
 370       gst_tcp_socket_read (src->client_sock_fd, GST_BUFFER_DATA (buf),
 371       readsize);
 372   if (ret < 0) {
 373     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 374     return NULL;
 375   }
 376
 377   /* if we read 0 bytes, and we're blocking, we hit eos */
 378   if (ret == 0) {
 379     GST_DEBUG ("blocking read returns 0, EOS");
 380     gst_buffer_unref (buf);
 381     gst_element_set_eos (GST_ELEMENT (src));
 382     return GST_DATA (gst_event_new (GST_EVENT_EOS));
 383   }
 384
 385   readsize = ret;
 386   GST_LOG_OBJECT (src, "Read %d bytes", readsize);
 387   GST_BUFFER_SIZE (buf) = readsize;
 388   GST_BUFFER_MAXSIZE (buf) = readsize;
 389   GST_BUFFER_OFFSET (buf) = src->curoffset;
 390   GST_BUFFER_OFFSET_END (buf) = src->curoffset + readsize;
 391   src->curoffset += readsize;
 392   return GST_DATA (buf);
 393 }
 394
 395
 396 static void
 397 gst_tcpserversrc_set_property (GObject * object, guint prop_id,
 398     const GValue * value, GParamSpec * pspec)
 399 {
 400   GstTCPServerSrc *tcpserversrc;
 401
 402   g_return_if_fail (GST_IS_TCPSERVERSRC (object));
 403   tcpserversrc = GST_TCPSERVERSRC (object);
 404
 405   switch (prop_id) {
 406     case ARG_PORT:
 407       tcpserversrc->server_port = g_value_get_int (value);
 408       break;
 409     case ARG_PROTOCOL:
 410       tcpserversrc->protocol = g_value_get_enum (value);
 411       break;
 412     case ARG_HOST:
 413       if (tcpserversrc->host)
 414         g_free (tcpserversrc->host);
 415       tcpserversrc->host = g_strdup (g_value_get_string (value));
 416       break;
 417     default:
 418       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 419       break;
 420   }
 421 }
 422
 423 static void
 424 gst_tcpserversrc_get_property (GObject * object, guint prop_id, GValue * value,
 425     GParamSpec * pspec)
 426 {
 427   GstTCPServerSrc *tcpserversrc;
 428
 429   g_return_if_fail (GST_IS_TCPSERVERSRC (object));
 430   tcpserversrc = GST_TCPSERVERSRC (object);
 431
 432   switch (prop_id) {
 433     case ARG_PORT:
 434       g_value_set_int (value, tcpserversrc->server_port);
 435       break;
 436     case ARG_PROTOCOL:
 437       g_value_set_enum (value, tcpserversrc->protocol);
 438       break;
 439     case ARG_HOST:
 440       g_value_set_string (value, tcpserversrc->host);
 441       break;
 442     default:
 443       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 444       break;
 445   }
 446 }
 447
 448 /* set up server */
 449 static gboolean
 450 gst_tcpserversrc_init_receive (GstTCPServerSrc * this)
 451 {
 452   int ret;
 453
 454   /* reset caps_received flag */
 455   this->caps_received = FALSE;
 456
 457   /* create the server listener socket */
 458   if ((this->server_sock_fd = socket (AF_INET, SOCK_STREAM, 0)) == -1) {
 459     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL), GST_ERROR_SYSTEM);
 460     return FALSE;
 461   }
 462   GST_DEBUG_OBJECT (this, "opened receiving server socket with fd %d",
 463       this->server_sock_fd);
 464
 465   /* make address reusable */
 466   if (setsockopt (this->server_sock_fd, SOL_SOCKET, SO_REUSEADDR, &ret,
 467           sizeof (int)) < 0) {
 468     GST_ELEMENT_ERROR (this, RESOURCE, SETTINGS, (NULL),
 469         ("Could not setsockopt: %s", g_strerror (errno)));
 470     return FALSE;
 471   }
 472
 473   /* name the socket */
 474   memset (&this->server_sin, 0, sizeof (this->server_sin));
 475   this->server_sin.sin_family = AF_INET;        /* network socket */
 476   this->server_sin.sin_port = htons (this->server_port);        /* on port */
 477   if (this->host) {
 478     gchar *host = gst_tcp_host_to_ip (GST_ELEMENT (this), this->host);
 479
 480     if (!host)
 481       return FALSE;
 482
 483     this->server_sin.sin_addr.s_addr = inet_addr (host);
 484     g_free (host);
 485   } else
 486     this->server_sin.sin_addr.s_addr = htonl (INADDR_ANY);
 487
 488   /* bind it */
 489   GST_DEBUG_OBJECT (this, "binding server socket to address");
 490   ret = bind (this->server_sock_fd, (struct sockaddr *) &this->server_sin,
 491       sizeof (this->server_sin));
 492
 493   if (ret) {
 494     switch (errno) {
 495       default:
 496         GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 497             ("bind failed: %s", g_strerror (errno)));
 498         return FALSE;
 499         break;
 500     }
 501   }
 502
 503   GST_DEBUG_OBJECT (this, "listening on server socket %d with queue of %d",
 504       this->server_sock_fd, TCP_BACKLOG);
 505   if (listen (this->server_sock_fd, TCP_BACKLOG) == -1) {
 506     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 507         ("Could not listen on server socket: %s", g_strerror (errno)));
 508     return FALSE;
 509   }
 510
 511   /* FIXME: maybe we should think about moving actual client accepting
 512      somewhere else */
 513   GST_DEBUG_OBJECT (this, "waiting for client");
 514   this->client_sock_fd =
 515       accept (this->server_sock_fd, (struct sockaddr *) &this->client_sin,
 516       &this->client_sin_len);
 517   if (this->client_sock_fd == -1) {
 518     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 519         ("Could not accept client on server socket: %s", g_strerror (errno)));
 520     return FALSE;
 521   }
 522   GST_DEBUG_OBJECT (this, "received client");
 523
 524   GST_FLAG_SET (this, GST_TCPSERVERSRC_OPEN);
 525   return TRUE;
 526 }
 527
 528 static void
 529 gst_tcpserversrc_close (GstTCPServerSrc * this)
 530 {
 531   if (this->server_sock_fd != -1) {
 532     close (this->server_sock_fd);
 533     this->server_sock_fd = -1;
 534   }
 535   if (this->client_sock_fd != -1) {
 536     close (this->client_sock_fd);
 537     this->client_sock_fd = -1;
 538   }
 539   GST_FLAG_UNSET (this, GST_TCPSERVERSRC_OPEN);
 540 }
 541
 542 static GstElementStateReturn
 543 gst_tcpserversrc_change_state (GstElement * element)
 544 {
 545   g_return_val_if_fail (GST_IS_TCPSERVERSRC (element), GST_STATE_FAILURE);
 546
 547   if (GST_STATE_PENDING (element) == GST_STATE_NULL) {
 548     if (GST_FLAG_IS_SET (element, GST_TCPSERVERSRC_OPEN))
 549       gst_tcpserversrc_close (GST_TCPSERVERSRC (element));
 550   } else {
 551     if (!GST_FLAG_IS_SET (element, GST_TCPSERVERSRC_OPEN)) {
 552       if (!gst_tcpserversrc_init_receive (GST_TCPSERVERSRC (element)))
 553         return GST_STATE_FAILURE;
 554     }
 555   }
 556
 557   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 558     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 559
 560   return GST_STATE_SUCCESS;
 561 }