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