gst/tcp/gsttcpclientsrc.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 "gsttcpclientsrc.h"
  29 #include <string.h>             /* memset */
  30 #include <unistd.h>
  31 #include <sys/socket.h>
  32 #include <netinet/in.h>
  33 #include <arpa/inet.h>
  34 #include <sys/ioctl.h>
  35
  36 GST_DEBUG_CATEGORY (tcpclientsrc_debug);
  37 #define GST_CAT_DEFAULT tcpclientsrc_debug
  38
  39 #define TCP_DEFAULT_PORT                4953
  40 #define TCP_DEFAULT_HOST                "localhost"
  41 #define MAX_READ_SIZE                   4 * 1024
  42
  43 /* elementfactory information */
  44 static GstElementDetails gst_tcpclientsrc_details =
  45 GST_ELEMENT_DETAILS ("TCP Client source",
  46     "Source/Network",
  47     "Receive data as a client over the network via TCP",
  48     "Thomas Vander Stichele <thomas at apestaart dot org>");
  49
  50 /* TCPClientSrc 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_tcpclientsrc_base_init (gpointer g_class);
  65 static void gst_tcpclientsrc_class_init (GstTCPClientSrc * klass);
  66 static void gst_tcpclientsrc_init (GstTCPClientSrc * tcpclientsrc);
  67
  68 static GstData *gst_tcpclientsrc_get (GstPad * pad);
  69 static GstElementStateReturn gst_tcpclientsrc_change_state (GstElement *
  70     element);
  71
  72 static void gst_tcpclientsrc_set_property (GObject * object, guint prop_id,
  73     const GValue * value, GParamSpec * pspec);
  74 static void gst_tcpclientsrc_get_property (GObject * object, guint prop_id,
  75     GValue * value, GParamSpec * pspec);
  76 static void gst_tcpclientsrc_set_clock (GstElement * element, GstClock * clock);
  77
  78 static GstElementClass *parent_class = NULL;
  79
  80 /*static guint gst_tcpclientsrc_signals[LAST_SIGNAL] = { 0 }; */
  81
  82 GType
  83 gst_tcpclientsrc_get_type (void)
  84 {
  85   static GType tcpclientsrc_type = 0;
  86
  87
  88   if (!tcpclientsrc_type) {
  89     static const GTypeInfo tcpclientsrc_info = {
  90       sizeof (GstTCPClientSrcClass),
  91       gst_tcpclientsrc_base_init,
  92       NULL,
  93       (GClassInitFunc) gst_tcpclientsrc_class_init,
  94       NULL,
  95       NULL,
  96       sizeof (GstTCPClientSrc),
  97       0,
  98       (GInstanceInitFunc) gst_tcpclientsrc_init,
  99       NULL
 100     };
 101
 102     tcpclientsrc_type =
 103         g_type_register_static (GST_TYPE_ELEMENT, "GstTCPClientSrc",
 104         &tcpclientsrc_info, 0);
 105   }
 106   return tcpclientsrc_type;
 107 }
 108
 109 static void
 110 gst_tcpclientsrc_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_tcpclientsrc_details);
 115 }
 116
 117 static void
 118 gst_tcpclientsrc_class_init (GstTCPClientSrc * 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",
 130           "The host IP address to receive packets from", TCP_DEFAULT_HOST,
 131           G_PARAM_READWRITE));
 132   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_PORT,
 133       g_param_spec_int ("port", "Port", "The port to receive packets from", 0,
 134           32768, TCP_DEFAULT_PORT, G_PARAM_READWRITE));
 135   g_object_class_install_property (gobject_class, ARG_PROTOCOL,
 136       g_param_spec_enum ("protocol", "Protocol", "The protocol to wrap data in",
 137           GST_TYPE_TCP_PROTOCOL_TYPE, GST_TCP_PROTOCOL_TYPE_GDP,
 138           G_PARAM_READWRITE));
 139
 140   gobject_class->set_property = gst_tcpclientsrc_set_property;
 141   gobject_class->get_property = gst_tcpclientsrc_get_property;
 142
 143   gstelement_class->change_state = gst_tcpclientsrc_change_state;
 144   gstelement_class->set_clock = gst_tcpclientsrc_set_clock;
 145
 146   GST_DEBUG_CATEGORY_INIT (tcpclientsrc_debug, "tcpclientsrc", 0,
 147       "TCP Client Source");
 148 }
 149
 150 static void
 151 gst_tcpclientsrc_set_clock (GstElement * element, GstClock * clock)
 152 {
 153   GstTCPClientSrc *tcpclientsrc;
 154
 155   tcpclientsrc = GST_TCPCLIENTSRC (element);
 156
 157   tcpclientsrc->clock = clock;
 158 }
 159
 160 static void
 161 gst_tcpclientsrc_init (GstTCPClientSrc * this)
 162 {
 163   /* create the src pad */
 164   this->srcpad = gst_pad_new ("src", GST_PAD_SRC);
 165   gst_element_add_pad (GST_ELEMENT (this), this->srcpad);
 166   gst_pad_set_get_function (this->srcpad, gst_tcpclientsrc_get);
 167
 168   this->port = TCP_DEFAULT_PORT;
 169   this->host = g_strdup (TCP_DEFAULT_HOST);
 170   this->clock = NULL;
 171   this->sock_fd = -1;
 172   this->protocol = GST_TCP_PROTOCOL_TYPE_GDP;
 173   this->curoffset = 0;
 174
 175   GST_FLAG_UNSET (this, GST_TCPCLIENTSRC_OPEN);
 176 }
 177
 178 static GstData *
 179 gst_tcpclientsrc_get (GstPad * pad)
 180 {
 181   GstTCPClientSrc *src;
 182   size_t readsize;
 183   int ret;
 184
 185   GstData *data = NULL;
 186   GstBuffer *buf = NULL;
 187   GstCaps *caps;
 188
 189   g_return_val_if_fail (pad != NULL, NULL);
 190   g_return_val_if_fail (GST_IS_PAD (pad), NULL);
 191   src = GST_TCPCLIENTSRC (GST_OBJECT_PARENT (pad));
 192   g_return_val_if_fail (GST_FLAG_IS_SET (src, GST_TCPCLIENTSRC_OPEN), NULL);
 193
 194   /* if we have a left over buffer after a discont, return that */
 195   if (src->buffer_after_discont) {
 196     buf = src->buffer_after_discont;
 197     GST_LOG_OBJECT (src,
 198         "Returning buffer after discont of size %d with timestamp %"
 199         GST_TIME_FORMAT " and duration %" GST_TIME_FORMAT,
 200         GST_BUFFER_SIZE (buf), GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)),
 201         GST_TIME_ARGS (GST_BUFFER_DURATION (buf)));
 202     src->buffer_after_discont = NULL;
 203     return GST_DATA (buf);
 204   }
 205
 206   /* read the buffer header if we're using a protocol */
 207   switch (src->protocol) {
 208       fd_set testfds;
 209
 210     case GST_TCP_PROTOCOL_TYPE_NONE:
 211       /* do a blocking select on the socket */
 212       FD_ZERO (&testfds);
 213       FD_SET (src->sock_fd, &testfds);
 214       ret = select (src->sock_fd + 1, &testfds, (fd_set *) 0, (fd_set *) 0, 0);
 215       /* no action (0) is an error too in our case */
 216       if (ret <= 0) {
 217         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 218             ("select failed: %s", g_strerror (errno)));
 219         return NULL;
 220       }
 221
 222       /* ask how much is available for reading on the socket */
 223       ret = ioctl (src->sock_fd, FIONREAD, &readsize);
 224       if (ret < 0) {
 225         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 226             ("ioctl failed: %s", g_strerror (errno)));
 227         return NULL;
 228       }
 229       buf = gst_buffer_new_and_alloc (readsize);
 230       break;
 231     case GST_TCP_PROTOCOL_TYPE_GDP:
 232       /* if we haven't received caps yet, we should get them first */
 233       if (!src->caps_received) {
 234         gchar *string;
 235
 236         if (!(caps = gst_tcp_gdp_read_caps (GST_ELEMENT (src), src->sock_fd))) {
 237           GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 238               ("Could not read caps through GDP"));
 239           return NULL;
 240         }
 241         src->caps_received = TRUE;
 242         string = gst_caps_to_string (caps);
 243         GST_DEBUG_OBJECT (src, "Received caps through GDP: %s", string);
 244         g_free (string);
 245
 246         if (!gst_pad_try_set_caps (pad, caps)) {
 247           g_warning ("Could not set caps");
 248           return NULL;
 249         }
 250       }
 251
 252       /* now receive the buffer header */
 253       if (!(data = gst_tcp_gdp_read_header (GST_ELEMENT (src), src->sock_fd))) {
 254         GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 255             ("Could not read data header through GDP"));
 256         return NULL;
 257       }
 258       if (GST_IS_EVENT (data))
 259         return data;
 260       buf = GST_BUFFER (data);
 261
 262       GST_LOG_OBJECT (src, "Going to read data from socket into buffer %p",
 263           buf);
 264       /* use this new buffer to read data into */
 265       readsize = GST_BUFFER_SIZE (buf);
 266       break;
 267     default:
 268       g_warning ("Unhandled protocol type");
 269       break;
 270   }
 271
 272   GST_LOG_OBJECT (src, "Reading %d bytes", readsize);
 273   ret = gst_tcp_socket_read (src->sock_fd, GST_BUFFER_DATA (buf), readsize);
 274   if (ret < 0) {
 275     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), GST_ERROR_SYSTEM);
 276     return NULL;
 277   }
 278
 279   /* if we read 0 bytes, and we're blocking, we hit eos */
 280   if (ret == 0) {
 281     GST_DEBUG ("blocking read returns 0, EOS");
 282     gst_buffer_unref (buf);
 283     gst_element_set_eos (GST_ELEMENT (src));
 284     return GST_DATA (gst_event_new (GST_EVENT_EOS));
 285   }
 286
 287   readsize = ret;
 288   GST_BUFFER_SIZE (buf) = readsize;
 289   GST_BUFFER_MAXSIZE (buf) = readsize;
 290   GST_BUFFER_OFFSET (buf) = src->curoffset;
 291   GST_BUFFER_OFFSET_END (buf) = src->curoffset + readsize;
 292
 293   /* if this is our first buffer, we need to send a discont with the
 294    * given timestamp or the current offset, and store the buffer for
 295    * the next iteration through the get loop */
 296   if (src->send_discont) {
 297     GstClockTime timestamp;
 298     GstEvent *event;
 299
 300     src->send_discont = FALSE;
 301     src->buffer_after_discont = buf;
 302     /* if the timestamp is valid, send a timed discont
 303      * taking into account the incoming buffer's timestamps */
 304     timestamp = GST_BUFFER_TIMESTAMP (buf);
 305     if (GST_CLOCK_TIME_IS_VALID (timestamp)) {
 306       GST_DEBUG_OBJECT (src,
 307           "sending discontinuous with timestamp %" GST_TIME_FORMAT,
 308           GST_TIME_ARGS (timestamp));
 309       event =
 310           gst_event_new_discontinuous (FALSE, GST_FORMAT_TIME, timestamp, NULL);
 311       return GST_DATA (event);
 312     }
 313     /* otherwise, send an offset discont */
 314     GST_DEBUG_OBJECT (src, "sending discontinuous with offset %d",
 315         src->curoffset);
 316     event =
 317         gst_event_new_discontinuous (FALSE, GST_FORMAT_BYTES, src->curoffset,
 318         NULL);
 319     return GST_DATA (event);
 320   }
 321
 322   src->curoffset += readsize;
 323   GST_LOG_OBJECT (src,
 324       "Returning buffer of size %d with timestamp %" GST_TIME_FORMAT
 325       " and duration %" GST_TIME_FORMAT, readsize,
 326       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)),
 327       GST_TIME_ARGS (GST_BUFFER_DURATION (buf)));
 328   return GST_DATA (buf);
 329 }
 330
 331
 332 static void
 333 gst_tcpclientsrc_set_property (GObject * object, guint prop_id,
 334     const GValue * value, GParamSpec * pspec)
 335 {
 336   GstTCPClientSrc *tcpclientsrc;
 337
 338   /* it's not null if we got it, but it might not be ours */
 339   g_return_if_fail (GST_IS_TCPCLIENTSRC (object));
 340   tcpclientsrc = GST_TCPCLIENTSRC (object);
 341
 342   switch (prop_id) {
 343     case ARG_PORT:
 344       tcpclientsrc->port = g_value_get_int (value);
 345       break;
 346     case ARG_HOST:
 347       /* FIXME: create a setter and handle changes correctly */
 348       g_free (tcpclientsrc->host);
 349       tcpclientsrc->host = g_strdup (g_value_get_string (value));
 350       break;
 351     case ARG_PROTOCOL:
 352       tcpclientsrc->protocol = g_value_get_enum (value);
 353       break;
 354     default:
 355       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 356       break;
 357   }
 358 }
 359
 360 static void
 361 gst_tcpclientsrc_get_property (GObject * object, guint prop_id, GValue * value,
 362     GParamSpec * pspec)
 363 {
 364   GstTCPClientSrc *tcpclientsrc;
 365
 366   g_return_if_fail (GST_IS_TCPCLIENTSRC (object));
 367   tcpclientsrc = GST_TCPCLIENTSRC (object);
 368
 369   switch (prop_id) {
 370     case ARG_PORT:
 371       g_value_set_int (value, tcpclientsrc->port);
 372       break;
 373     case ARG_HOST:
 374       g_value_set_string (value, tcpclientsrc->host);
 375       break;
 376     case ARG_PROTOCOL:
 377       g_value_set_enum (value, tcpclientsrc->protocol);
 378       break;
 379
 380     default:
 381       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 382       break;
 383   }
 384 }
 385
 386 /* create a socket for connecting to remote server */
 387 static gboolean
 388 gst_tcpclientsrc_init_receive (GstTCPClientSrc * this)
 389 {
 390   int ret;
 391   gchar *ip;
 392
 393   /* create receiving client socket */
 394   GST_DEBUG_OBJECT (this, "opening receiving client socket to %s:%d",
 395       this->host, this->port);
 396   if ((this->sock_fd = socket (AF_INET, SOCK_STREAM, 0)) == -1) {
 397     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL), GST_ERROR_SYSTEM);
 398     return FALSE;
 399   }
 400   GST_DEBUG_OBJECT (this, "opened receiving client socket with fd %d",
 401       this->sock_fd);
 402
 403   /* look up name if we need to */
 404   ip = gst_tcp_host_to_ip (GST_ELEMENT (this), this->host);
 405   if (!ip)
 406     return FALSE;
 407   GST_DEBUG_OBJECT (this, "IP address for host %s is %s", this->host, ip);
 408
 409   /* connect to server */
 410   memset (&this->server_sin, 0, sizeof (this->server_sin));
 411   this->server_sin.sin_family = AF_INET;        /* network socket */
 412   this->server_sin.sin_port = htons (this->port);       /* on port */
 413   this->server_sin.sin_addr.s_addr = inet_addr (ip);    /* on host ip */
 414
 415   GST_DEBUG_OBJECT (this, "connecting to server");
 416   ret = connect (this->sock_fd, (struct sockaddr *) &this->server_sin,
 417       sizeof (this->server_sin));
 418
 419   if (ret) {
 420     switch (errno) {
 421       case ECONNREFUSED:
 422         GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ,
 423             (_("Connection to %s:%d refused."), this->host, this->port),
 424             (NULL));
 425         return FALSE;
 426         break;
 427       default:
 428         GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 429             ("connect to %s:%d failed: %s", this->host, this->port,
 430                 g_strerror (errno)));
 431         return FALSE;
 432         break;
 433     }
 434   }
 435
 436   this->send_discont = TRUE;
 437   this->buffer_after_discont = NULL;
 438   GST_FLAG_SET (this, GST_TCPCLIENTSRC_OPEN);
 439
 440   return TRUE;
 441 }
 442
 443 static void
 444 gst_tcpclientsrc_close (GstTCPClientSrc * this)
 445 {
 446   if (this->sock_fd != -1) {
 447     close (this->sock_fd);
 448     this->sock_fd = -1;
 449   }
 450   GST_FLAG_UNSET (this, GST_TCPCLIENTSRC_OPEN);
 451 }
 452
 453 static GstElementStateReturn
 454 gst_tcpclientsrc_change_state (GstElement * element)
 455 {
 456   g_return_val_if_fail (GST_IS_TCPCLIENTSRC (element), GST_STATE_FAILURE);
 457
 458   if (GST_STATE_PENDING (element) == GST_STATE_NULL) {
 459     if (GST_FLAG_IS_SET (element, GST_TCPCLIENTSRC_OPEN))
 460       gst_tcpclientsrc_close (GST_TCPCLIENTSRC (element));
 461   } else {
 462     if (!GST_FLAG_IS_SET (element, GST_TCPCLIENTSRC_OPEN)) {
 463       if (!gst_tcpclientsrc_init_receive (GST_TCPCLIENTSRC (element)))
 464         return GST_STATE_FAILURE;
 465     }
 466   }
 467
 468   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 469     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 470
 471   return GST_STATE_SUCCESS;
 472 }