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