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