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_static_metadata (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   GstMapInfo map;
 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     gst_buffer_map (*outbuf, &map, GST_MAP_READWRITE);
 228     rret =
 229         g_socket_receive (src->socket, (gchar *) map.data, read,
 230         src->cancellable, &err);
 231   } else {
 232     /* Connection closed */
 233     *outbuf = NULL;
 234     read = 0;
 235     rret = 0;
 236   }
 237
 238   if (rret == 0) {
 239     GST_DEBUG_OBJECT (src, "Connection closed");
 240     ret = GST_FLOW_EOS;
 241     if (*outbuf) {
 242       gst_buffer_unmap (*outbuf, &map);
 243       gst_buffer_unref (*outbuf);
 244     }
 245     *outbuf = NULL;
 246   } else if (rret < 0) {
 247     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 248       ret = GST_FLOW_FLUSHING;
 249       GST_DEBUG_OBJECT (src, "Cancelled reading from socket");
 250     } else {
 251       ret = GST_FLOW_ERROR;
 252       GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 253           ("Failed to read from socket: %s", err->message));
 254     }
 255     gst_buffer_unmap (*outbuf, &map);
 256     gst_buffer_unref (*outbuf);
 257     *outbuf = NULL;
 258   } else {
 259     ret = GST_FLOW_OK;
 260     gst_buffer_unmap (*outbuf, &map);
 261     gst_buffer_resize (*outbuf, 0, rret);
 262
 263     GST_LOG_OBJECT (src,
 264         "Returning buffer from _get of size %" G_GSIZE_FORMAT ", ts %"
 265         GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT
 266         ", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT,
 267         gst_buffer_get_size (*outbuf),
 268         GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (*outbuf)),
 269         GST_TIME_ARGS (GST_BUFFER_DURATION (*outbuf)),
 270         GST_BUFFER_OFFSET (*outbuf), GST_BUFFER_OFFSET_END (*outbuf));
 271   }
 272   g_clear_error (&err);
 273
 274 done:
 275   return ret;
 276
 277 select_error:
 278   {
 279     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 280         ("Select failed: %s", err->message));
 281     g_clear_error (&err);
 282     return GST_FLOW_ERROR;
 283   }
 284 get_available_error:
 285   {
 286     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 287         ("Failed to get available bytes from socket"));
 288     return GST_FLOW_ERROR;
 289   }
 290 wrong_state:
 291   {
 292     GST_DEBUG_OBJECT (src, "connection to closed, cannot read data");
 293     return GST_FLOW_FLUSHING;
 294   }
 295 }
 296
 297 static void
 298 gst_tcp_client_src_set_property (GObject * object, guint prop_id,
 299     const GValue * value, GParamSpec * pspec)
 300 {
 301   GstTCPClientSrc *tcpclientsrc = GST_TCP_CLIENT_SRC (object);
 302
 303   switch (prop_id) {
 304     case PROP_HOST:
 305       if (!g_value_get_string (value)) {
 306         g_warning ("host property cannot be NULL");
 307         break;
 308       }
 309       g_free (tcpclientsrc->host);
 310       tcpclientsrc->host = g_strdup (g_value_get_string (value));
 311       break;
 312     case PROP_PORT:
 313       tcpclientsrc->port = g_value_get_int (value);
 314       break;
 315
 316     default:
 317       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 318       break;
 319   }
 320 }
 321
 322 static void
 323 gst_tcp_client_src_get_property (GObject * object, guint prop_id,
 324     GValue * value, GParamSpec * pspec)
 325 {
 326   GstTCPClientSrc *tcpclientsrc = GST_TCP_CLIENT_SRC (object);
 327
 328   switch (prop_id) {
 329     case PROP_HOST:
 330       g_value_set_string (value, tcpclientsrc->host);
 331       break;
 332     case PROP_PORT:
 333       g_value_set_int (value, tcpclientsrc->port);
 334       break;
 335     default:
 336       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 337       break;
 338   }
 339 }
 340
 341 /* create a socket for connecting to remote server */
 342 static gboolean
 343 gst_tcp_client_src_start (GstBaseSrc * bsrc)
 344 {
 345   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
 346   GError *err = NULL;
 347   GInetAddress *addr;
 348   GSocketAddress *saddr;
 349   GResolver *resolver;
 350
 351   /* look up name if we need to */
 352   addr = g_inet_address_new_from_string (src->host);
 353   if (!addr) {
 354     GList *results;
 355
 356     resolver = g_resolver_get_default ();
 357
 358     results =
 359         g_resolver_lookup_by_name (resolver, src->host, src->cancellable, &err);
 360     if (!results)
 361       goto name_resolve;
 362     addr = G_INET_ADDRESS (g_object_ref (results->data));
 363
 364     g_resolver_free_addresses (results);
 365     g_object_unref (resolver);
 366   }
 367 #ifndef GST_DISABLE_GST_DEBUG
 368   {
 369     gchar *ip = g_inet_address_to_string (addr);
 370
 371     GST_DEBUG_OBJECT (src, "IP address for host %s is %s", src->host, ip);
 372     g_free (ip);
 373   }
 374 #endif
 375
 376   saddr = g_inet_socket_address_new (addr, src->port);
 377   g_object_unref (addr);
 378
 379   /* create receiving client socket */
 380   GST_DEBUG_OBJECT (src, "opening receiving client socket to %s:%d",
 381       src->host, src->port);
 382
 383   src->socket =
 384       g_socket_new (g_socket_address_get_family (saddr), G_SOCKET_TYPE_STREAM,
 385       G_SOCKET_PROTOCOL_TCP, &err);
 386   if (!src->socket)
 387     goto no_socket;
 388
 389   GST_DEBUG_OBJECT (src, "opened receiving client socket");
 390   GST_OBJECT_FLAG_SET (src, GST_TCP_CLIENT_SRC_OPEN);
 391
 392   /* connect to server */
 393   if (!g_socket_connect (src->socket, saddr, src->cancellable, &err))
 394     goto connect_failed;
 395
 396   g_object_unref (saddr);
 397
 398   return TRUE;
 399
 400 no_socket:
 401   {
 402     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 403         ("Failed to create socket: %s", err->message));
 404     g_clear_error (&err);
 405     g_object_unref (saddr);
 406     return FALSE;
 407   }
 408 name_resolve:
 409   {
 410     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 411       GST_DEBUG_OBJECT (src, "Cancelled name resolval");
 412     } else {
 413       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 414           ("Failed to resolve host '%s': %s", src->host, err->message));
 415     }
 416     g_clear_error (&err);
 417     g_object_unref (resolver);
 418     return FALSE;
 419   }
 420 connect_failed:
 421   {
 422     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 423       GST_DEBUG_OBJECT (src, "Cancelled connecting");
 424     } else {
 425       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 426           ("Failed to connect to host '%s:%d': %s", src->host, src->port,
 427               err->message));
 428     }
 429     g_clear_error (&err);
 430     g_object_unref (saddr);
 431     gst_tcp_client_src_stop (GST_BASE_SRC (src));
 432     return FALSE;
 433   }
 434 }
 435
 436 /* close the socket and associated resources
 437  * unset OPEN flag
 438  * used both to recover from errors and go to NULL state */
 439 static gboolean
 440 gst_tcp_client_src_stop (GstBaseSrc * bsrc)
 441 {
 442   GstTCPClientSrc *src;
 443   GError *err = NULL;
 444
 445   src = GST_TCP_CLIENT_SRC (bsrc);
 446
 447   if (src->socket) {
 448     GST_DEBUG_OBJECT (src, "closing socket");
 449
 450     if (!g_socket_close (src->socket, &err)) {
 451       GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
 452       g_clear_error (&err);
 453     }
 454     g_object_unref (src->socket);
 455     src->socket = NULL;
 456   }
 457
 458   GST_OBJECT_FLAG_UNSET (src, GST_TCP_CLIENT_SRC_OPEN);
 459
 460   return TRUE;
 461 }
 462
 463 /* will be called only between calls to start() and stop() */
 464 static gboolean
 465 gst_tcp_client_src_unlock (GstBaseSrc * bsrc)
 466 {
 467   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
 468
 469   GST_DEBUG_OBJECT (src, "set to flushing");
 470   g_cancellable_cancel (src->cancellable);
 471
 472   return TRUE;
 473 }
 474
 475 /* will be called only between calls to start() and stop() */
 476 static gboolean
 477 gst_tcp_client_src_unlock_stop (GstBaseSrc * bsrc)
 478 {
 479   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
 480
 481   GST_DEBUG_OBJECT (src, "unset flushing");
 482   g_cancellable_reset (src->cancellable);
 483
 484   return TRUE;
 485 }