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  * Copyright (C) <2011> Collabora Ltd.
   5  *     Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
   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 /**
  24  * SECTION:element-tcpclientsrc
  25  * @see_also: #tcpclientsink
  26  *
  27  * <refsect2>
  28  * <title>Example launch line</title>
  29  * |[
  30  * # server:
  31  * nc -l -p 3000
  32  * # client:
  33  * gst-launch tcpclientsrc port=3000 ! fdsink fd=2
  34  * ]| everything you type in the server is shown on the client
  35  * </refsect2>
  36  */
  37
  38 #ifdef HAVE_CONFIG_H
  39 #include "config.h"
  40 #endif
  41
  42 #include <gst/gst-i18n-plugin.h>
  43 #include "gsttcpclientsrc.h"
  44 #include "gsttcp.h"
  45
  46 GST_DEBUG_CATEGORY_STATIC (tcpclientsrc_debug);
  47 #define GST_CAT_DEFAULT tcpclientsrc_debug
  48
  49 #define MAX_READ_SIZE                   4 * 1024
  50
  51
  52 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
  53     GST_PAD_SRC,
  54     GST_PAD_ALWAYS,
  55     GST_STATIC_CAPS_ANY);
  56
  57
  58 enum
  59 {
  60   PROP_0,
  61   PROP_HOST,
  62   PROP_PORT
  63 };
  64
  65 #define gst_tcp_client_src_parent_class parent_class
  66 G_DEFINE_TYPE (GstTCPClientSrc, gst_tcp_client_src, GST_TYPE_PUSH_SRC);
  67
  68
  69 static void gst_tcp_client_src_finalize (GObject * gobject);
  70
  71 static GstCaps *gst_tcp_client_src_getcaps (GstBaseSrc * psrc,
  72     GstCaps * filter);
  73
  74 static GstFlowReturn gst_tcp_client_src_create (GstPushSrc * psrc,
  75     GstBuffer ** outbuf);
  76 static gboolean gst_tcp_client_src_stop (GstBaseSrc * bsrc);
  77 static gboolean gst_tcp_client_src_start (GstBaseSrc * bsrc);
  78 static gboolean gst_tcp_client_src_unlock (GstBaseSrc * bsrc);
  79 static gboolean gst_tcp_client_src_unlock_stop (GstBaseSrc * bsrc);
  80
  81 static void gst_tcp_client_src_set_property (GObject * object, guint prop_id,
  82     const GValue * value, GParamSpec * pspec);
  83 static void gst_tcp_client_src_get_property (GObject * object, guint prop_id,
  84     GValue * value, GParamSpec * pspec);
  85
  86 static void
  87 gst_tcp_client_src_class_init (GstTCPClientSrcClass * klass)
  88 {
  89   GObjectClass *gobject_class;
  90   GstElementClass *gstelement_class;
  91   GstBaseSrcClass *gstbasesrc_class;
  92   GstPushSrcClass *gstpush_src_class;
  93
  94   gobject_class = (GObjectClass *) klass;
  95   gstelement_class = (GstElementClass *) klass;
  96   gstbasesrc_class = (GstBaseSrcClass *) klass;
  97   gstpush_src_class = (GstPushSrcClass *) klass;
  98
  99   gobject_class->set_property = gst_tcp_client_src_set_property;
 100   gobject_class->get_property = gst_tcp_client_src_get_property;
 101   gobject_class->finalize = gst_tcp_client_src_finalize;
 102
 103   g_object_class_install_property (gobject_class, PROP_HOST,
 104       g_param_spec_string ("host", "Host",
 105           "The host IP address to receive packets from", TCP_DEFAULT_HOST,
 106           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 107   g_object_class_install_property (gobject_class, PROP_PORT,
 108       g_param_spec_int ("port", "Port", "The port to receive packets from", 0,
 109           TCP_HIGHEST_PORT, TCP_DEFAULT_PORT,
 110           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 111
 112   gst_element_class_add_pad_template (gstelement_class,
 113       gst_static_pad_template_get (&srctemplate));
 114
 115   gst_element_class_set_details_simple (gstelement_class,
 116       "TCP client source", "Source/Network",
 117       "Receive data as a client over the network via TCP",
 118       "Thomas Vander Stichele <thomas at apestaart dot org>");
 119
 120   gstbasesrc_class->get_caps = gst_tcp_client_src_getcaps;
 121   gstbasesrc_class->start = gst_tcp_client_src_start;
 122   gstbasesrc_class->stop = gst_tcp_client_src_stop;
 123   gstbasesrc_class->unlock = gst_tcp_client_src_unlock;
 124   gstbasesrc_class->unlock_stop = gst_tcp_client_src_unlock_stop;
 125
 126   gstpush_src_class->create = gst_tcp_client_src_create;
 127
 128   GST_DEBUG_CATEGORY_INIT (tcpclientsrc_debug, "tcpclientsrc", 0,
 129       "TCP Client Source");
 130 }
 131
 132 static void
 133 gst_tcp_client_src_init (GstTCPClientSrc * this)
 134 {
 135   this->port = TCP_DEFAULT_PORT;
 136   this->host = g_strdup (TCP_DEFAULT_HOST);
 137   this->socket = NULL;
 138   this->cancellable = g_cancellable_new ();
 139
 140   GST_OBJECT_FLAG_UNSET (this, GST_TCP_CLIENT_SRC_OPEN);
 141 }
 142
 143 static void
 144 gst_tcp_client_src_finalize (GObject * gobject)
 145 {
 146   GstTCPClientSrc *this = GST_TCP_CLIENT_SRC (gobject);
 147
 148   if (this->cancellable)
 149     g_object_unref (this->cancellable);
 150   this->cancellable = NULL;
 151   if (this->socket)
 152     g_object_unref (this->socket);
 153   this->socket = NULL;
 154   g_free (this->host);
 155   this->host = NULL;
 156
 157   G_OBJECT_CLASS (parent_class)->finalize (gobject);
 158 }
 159
 160 static GstCaps *
 161 gst_tcp_client_src_getcaps (GstBaseSrc * bsrc, GstCaps * filter)
 162 {
 163   GstTCPClientSrc *src;
 164   GstCaps *caps = NULL;
 165
 166   src = GST_TCP_CLIENT_SRC (bsrc);
 167
 168   caps = (filter ? gst_caps_ref (filter) : gst_caps_new_any ());
 169
 170   GST_DEBUG_OBJECT (src, "returning caps %" GST_PTR_FORMAT, caps);
 171   g_assert (GST_IS_CAPS (caps));
 172   return caps;
 173 }
 174
 175 static GstFlowReturn
 176 gst_tcp_client_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
 177 {
 178   GstTCPClientSrc *src;
 179   GstFlowReturn ret = GST_FLOW_OK;
 180   gssize rret;
 181   GError *err = NULL;
 182   guint8 *data;
 183   gssize avail, read;
 184
 185   src = GST_TCP_CLIENT_SRC (psrc);
 186
 187   if (!GST_OBJECT_FLAG_IS_SET (src, GST_TCP_CLIENT_SRC_OPEN))
 188     goto wrong_state;
 189
 190   GST_LOG_OBJECT (src, "asked for a buffer");
 191
 192   /* read the buffer header */
 193   avail = g_socket_get_available_bytes (src->socket);
 194   if (avail < 0) {
 195     goto get_available_error;
 196   } else if (avail == 0) {
 197     GIOCondition condition;
 198
 199     if (!g_socket_condition_wait (src->socket,
 200             G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, src->cancellable, &err))
 201       goto select_error;
 202
 203     condition =
 204         g_socket_condition_check (src->socket,
 205         G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP);
 206
 207     if ((condition & G_IO_ERR)) {
 208       GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 209           ("Socket in error state"));
 210       *outbuf = NULL;
 211       ret = GST_FLOW_ERROR;
 212       goto done;
 213     } else if ((condition & G_IO_HUP)) {
 214       GST_DEBUG_OBJECT (src, "Connection closed");
 215       *outbuf = NULL;
 216       ret = GST_FLOW_EOS;
 217       goto done;
 218     }
 219     avail = g_socket_get_available_bytes (src->socket);
 220     if (avail < 0)
 221       goto get_available_error;
 222   }
 223
 224   if (avail > 0) {
 225     read = MIN (avail, MAX_READ_SIZE);
 226     *outbuf = gst_buffer_new_and_alloc (read);
 227     data = gst_buffer_map (*outbuf, NULL, NULL, GST_MAP_READWRITE);
 228     rret =
 229         g_socket_receive (src->socket, (gchar *) data, read,
 230         src->cancellable, &err);
 231   } else {
 232     /* Connection closed */
 233     *outbuf = NULL;
 234     rret = 0;
 235   }
 236
 237   if (rret == 0) {
 238     GST_DEBUG_OBJECT (src, "Connection closed");
 239     ret = GST_FLOW_EOS;
 240     if (*outbuf) {
 241       gst_buffer_unmap (*outbuf, data, read);
 242       gst_buffer_unref (*outbuf);
 243     }
 244     *outbuf = NULL;
 245   } else if (rret < 0) {
 246     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 247       ret = GST_FLOW_WRONG_STATE;
 248       GST_DEBUG_OBJECT (src, "Cancelled reading from socket");
 249     } else {
 250       ret = GST_FLOW_ERROR;
 251       GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 252           ("Failed to read from socket: %s", err->message));
 253     }
 254     gst_buffer_unmap (*outbuf, data, read);
 255     gst_buffer_unref (*outbuf);
 256     *outbuf = NULL;
 257   } else {
 258     ret = GST_FLOW_OK;
 259     gst_buffer_unmap (*outbuf, data, rret);
 260
 261     GST_LOG_OBJECT (src,
 262         "Returning buffer from _get of size %" G_GSIZE_FORMAT ", ts %"
 263         GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT
 264         ", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT,
 265         gst_buffer_get_size (*outbuf),
 266         GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (*outbuf)),
 267         GST_TIME_ARGS (GST_BUFFER_DURATION (*outbuf)),
 268         GST_BUFFER_OFFSET (*outbuf), GST_BUFFER_OFFSET_END (*outbuf));
 269   }
 270   g_clear_error (&err);
 271
 272 done:
 273   return ret;
 274
 275 select_error:
 276   {
 277     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 278         ("Select failed: %s", err->message));
 279     g_clear_error (&err);
 280     return GST_FLOW_ERROR;
 281   }
 282 get_available_error:
 283   {
 284     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 285         ("Failed to get available bytes from socket"));
 286     return GST_FLOW_ERROR;
 287   }
 288 wrong_state:
 289   {
 290     GST_DEBUG_OBJECT (src, "connection to closed, cannot read data");
 291     return GST_FLOW_WRONG_STATE;
 292   }
 293 }
 294
 295 static void
 296 gst_tcp_client_src_set_property (GObject * object, guint prop_id,
 297     const GValue * value, GParamSpec * pspec)
 298 {
 299   GstTCPClientSrc *tcpclientsrc = GST_TCP_CLIENT_SRC (object);
 300
 301   switch (prop_id) {
 302     case PROP_HOST:
 303       if (!g_value_get_string (value)) {
 304         g_warning ("host property cannot be NULL");
 305         break;
 306       }
 307       g_free (tcpclientsrc->host);
 308       tcpclientsrc->host = g_strdup (g_value_get_string (value));
 309       break;
 310     case PROP_PORT:
 311       tcpclientsrc->port = g_value_get_int (value);
 312       break;
 313
 314     default:
 315       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 316       break;
 317   }
 318 }
 319
 320 static void
 321 gst_tcp_client_src_get_property (GObject * object, guint prop_id,
 322     GValue * value, GParamSpec * pspec)
 323 {
 324   GstTCPClientSrc *tcpclientsrc = GST_TCP_CLIENT_SRC (object);
 325
 326   switch (prop_id) {
 327     case PROP_HOST:
 328       g_value_set_string (value, tcpclientsrc->host);
 329       break;
 330     case PROP_PORT:
 331       g_value_set_int (value, tcpclientsrc->port);
 332       break;
 333     default:
 334       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 335       break;
 336   }
 337 }
 338
 339 /* create a socket for connecting to remote server */
 340 static gboolean
 341 gst_tcp_client_src_start (GstBaseSrc * bsrc)
 342 {
 343   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
 344   GError *err = NULL;
 345   GInetAddress *addr;
 346   GSocketAddress *saddr;
 347   GResolver *resolver;
 348
 349   /* look up name if we need to */
 350   addr = g_inet_address_new_from_string (src->host);
 351   if (!addr) {
 352     GList *results;
 353
 354     resolver = g_resolver_get_default ();
 355
 356     results =
 357         g_resolver_lookup_by_name (resolver, src->host, src->cancellable, &err);
 358     if (!results)
 359       goto name_resolve;
 360     addr = G_INET_ADDRESS (g_object_ref (results->data));
 361
 362     g_resolver_free_addresses (results);
 363     g_object_unref (resolver);
 364   }
 365 #ifndef GST_DISABLE_GST_DEBUG
 366   {
 367     gchar *ip = g_inet_address_to_string (addr);
 368
 369     GST_DEBUG_OBJECT (src, "IP address for host %s is %s", src->host, ip);
 370     g_free (ip);
 371   }
 372 #endif
 373
 374   saddr = g_inet_socket_address_new (addr, src->port);
 375   g_object_unref (addr);
 376
 377   /* create receiving client socket */
 378   GST_DEBUG_OBJECT (src, "opening receiving client socket to %s:%d",
 379       src->host, src->port);
 380
 381   src->socket =
 382       g_socket_new (g_socket_address_get_family (saddr), G_SOCKET_TYPE_STREAM,
 383       G_SOCKET_PROTOCOL_TCP, &err);
 384   if (!src->socket)
 385     goto no_socket;
 386
 387   GST_DEBUG_OBJECT (src, "opened receiving client socket");
 388   GST_OBJECT_FLAG_SET (src, GST_TCP_CLIENT_SRC_OPEN);
 389
 390   /* connect to server */
 391   if (!g_socket_connect (src->socket, saddr, src->cancellable, &err))
 392     goto connect_failed;
 393
 394   g_object_unref (saddr);
 395
 396   return TRUE;
 397
 398 no_socket:
 399   {
 400     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 401         ("Failed to create socket: %s", err->message));
 402     g_clear_error (&err);
 403     g_object_unref (saddr);
 404     return FALSE;
 405   }
 406 name_resolve:
 407   {
 408     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 409       GST_DEBUG_OBJECT (src, "Cancelled name resolval");
 410     } else {
 411       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 412           ("Failed to resolve host '%s': %s", src->host, err->message));
 413     }
 414     g_clear_error (&err);
 415     g_object_unref (resolver);
 416     return FALSE;
 417   }
 418 connect_failed:
 419   {
 420     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 421       GST_DEBUG_OBJECT (src, "Cancelled connecting");
 422     } else {
 423       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 424           ("Failed to connect to host '%s:%d': %s", src->host, src->port,
 425               err->message));
 426     }
 427     g_clear_error (&err);
 428     g_object_unref (saddr);
 429     gst_tcp_client_src_stop (GST_BASE_SRC (src));
 430     return FALSE;
 431   }
 432 }
 433
 434 /* close the socket and associated resources
 435  * unset OPEN flag
 436  * used both to recover from errors and go to NULL state */
 437 static gboolean
 438 gst_tcp_client_src_stop (GstBaseSrc * bsrc)
 439 {
 440   GstTCPClientSrc *src;
 441   GError *err = NULL;
 442
 443   src = GST_TCP_CLIENT_SRC (bsrc);
 444
 445   if (src->socket) {
 446     GST_DEBUG_OBJECT (src, "closing socket");
 447
 448     if (!g_socket_close (src->socket, &err)) {
 449       GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
 450       g_clear_error (&err);
 451     }
 452     g_object_unref (src->socket);
 453     src->socket = NULL;
 454   }
 455
 456   GST_OBJECT_FLAG_UNSET (src, GST_TCP_CLIENT_SRC_OPEN);
 457
 458   return TRUE;
 459 }
 460
 461 /* will be called only between calls to start() and stop() */
 462 static gboolean
 463 gst_tcp_client_src_unlock (GstBaseSrc * bsrc)
 464 {
 465   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
 466
 467   GST_DEBUG_OBJECT (src, "set to flushing");
 468   g_cancellable_cancel (src->cancellable);
 469
 470   return TRUE;
 471 }
 472
 473 /* will be called only between calls to start() and stop() */
 474 static gboolean
 475 gst_tcp_client_src_unlock_stop (GstBaseSrc * bsrc)
 476 {
 477   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
 478
 479   GST_DEBUG_OBJECT (src, "unset flushing");
 480   g_cancellable_reset (src->cancellable);
 481
 482   return TRUE;
 483 }