gst/tcp/gsttcpclientsink.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., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 /**
  24  * SECTION:element-tcpclientsink
  25  * @title: tcpclientsink
  26  * @see_also: #tcpclientsink
  27  *
  28  * ## Example launch line (server):
  29  * |[
  30  * nc -l -p 3000
  31  * ]|
  32  * ## Example launch line (client):
  33  * |[
  34  * gst-launch-1.0 fdsink fd=1 ! tcpclientsink port=3000
  35  * ]|
  36  *  everything you type in the client is shown on the server (fd=1 means
  37  * standard input which is the command line input file descriptor)
  38  *
  39  */
  40
  41 #ifdef HAVE_CONFIG_H
  42 #include "config.h"
  43 #endif
  44 #include <gst/gst-i18n-plugin.h>
  45
  46 #include "gsttcp.h"
  47 #include "gsttcpelements.h"
  48 #include "gsttcpclientsink.h"
  49
  50 /* TCPClientSink signals and args */
  51 enum
  52 {
  53   FRAME_ENCODED,
  54   /* FILL ME */
  55   LAST_SIGNAL
  56 };
  57
  58 GST_DEBUG_CATEGORY_STATIC (tcpclientsink_debug);
  59 #define GST_CAT_DEFAULT (tcpclientsink_debug)
  60
  61 enum
  62 {
  63   PROP_0,
  64   PROP_HOST,
  65   PROP_PORT
  66 };
  67
  68 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
  69     GST_PAD_SINK,
  70     GST_PAD_ALWAYS,
  71     GST_STATIC_CAPS_ANY);
  72
  73 static void gst_tcp_client_sink_finalize (GObject * gobject);
  74
  75 static gboolean gst_tcp_client_sink_setcaps (GstBaseSink * bsink,
  76     GstCaps * caps);
  77 static GstFlowReturn gst_tcp_client_sink_render (GstBaseSink * bsink,
  78     GstBuffer * buf);
  79 static gboolean gst_tcp_client_sink_start (GstBaseSink * bsink);
  80 static gboolean gst_tcp_client_sink_stop (GstBaseSink * bsink);
  81 static gboolean gst_tcp_client_sink_unlock (GstBaseSink * bsink);
  82 static gboolean gst_tcp_client_sink_unlock_stop (GstBaseSink * bsink);
  83
  84 static void gst_tcp_client_sink_set_property (GObject * object, guint prop_id,
  85     const GValue * value, GParamSpec * pspec);
  86 static void gst_tcp_client_sink_get_property (GObject * object, guint prop_id,
  87     GValue * value, GParamSpec * pspec);
  88
  89
  90 /*static guint gst_tcp_client_sink_signals[LAST_SIGNAL] = { 0 }; */
  91
  92 #define gst_tcp_client_sink_parent_class parent_class
  93 G_DEFINE_TYPE (GstTCPClientSink, gst_tcp_client_sink, GST_TYPE_BASE_SINK);
  94 GST_ELEMENT_REGISTER_DEFINE_WITH_CODE (tcpclientsink, "tcpclientsink",
  95     GST_RANK_NONE, GST_TYPE_TCP_CLIENT_SINK, tcp_element_init (plugin));
  96
  97 static void
  98 gst_tcp_client_sink_class_init (GstTCPClientSinkClass * klass)
  99 {
 100   GObjectClass *gobject_class;
 101   GstElementClass *gstelement_class;
 102   GstBaseSinkClass *gstbasesink_class;
 103
 104   gobject_class = (GObjectClass *) klass;
 105   gstelement_class = (GstElementClass *) klass;
 106   gstbasesink_class = (GstBaseSinkClass *) klass;
 107
 108   parent_class = g_type_class_peek_parent (klass);
 109
 110   gobject_class->set_property = gst_tcp_client_sink_set_property;
 111   gobject_class->get_property = gst_tcp_client_sink_get_property;
 112   gobject_class->finalize = gst_tcp_client_sink_finalize;
 113
 114   g_object_class_install_property (gobject_class, PROP_HOST,
 115       g_param_spec_string ("host", "Host", "The host/IP to send the packets to",
 116           TCP_DEFAULT_HOST, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 117   g_object_class_install_property (gobject_class, PROP_PORT,
 118       g_param_spec_int ("port", "Port", "The port to send the packets to",
 119           0, TCP_HIGHEST_PORT, TCP_DEFAULT_PORT,
 120           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 121
 122   gst_element_class_add_static_pad_template (gstelement_class, &sinktemplate);
 123
 124   gst_element_class_set_static_metadata (gstelement_class,
 125       "TCP client sink", "Sink/Network",
 126       "Send data as a client over the network via TCP",
 127       "Thomas Vander Stichele <thomas at apestaart dot org>");
 128
 129   gstbasesink_class->start = gst_tcp_client_sink_start;
 130   gstbasesink_class->stop = gst_tcp_client_sink_stop;
 131   gstbasesink_class->set_caps = gst_tcp_client_sink_setcaps;
 132   gstbasesink_class->render = gst_tcp_client_sink_render;
 133   gstbasesink_class->unlock = gst_tcp_client_sink_unlock;
 134   gstbasesink_class->unlock_stop = gst_tcp_client_sink_unlock_stop;
 135
 136   GST_DEBUG_CATEGORY_INIT (tcpclientsink_debug, "tcpclientsink", 0, "TCP sink");
 137 }
 138
 139 static void
 140 gst_tcp_client_sink_init (GstTCPClientSink * this)
 141 {
 142   this->host = g_strdup (TCP_DEFAULT_HOST);
 143   this->port = TCP_DEFAULT_PORT;
 144
 145   this->socket = NULL;
 146   this->cancellable = g_cancellable_new ();
 147
 148   GST_OBJECT_FLAG_UNSET (this, GST_TCP_CLIENT_SINK_OPEN);
 149 }
 150
 151 static void
 152 gst_tcp_client_sink_finalize (GObject * gobject)
 153 {
 154   GstTCPClientSink *this = GST_TCP_CLIENT_SINK (gobject);
 155
 156   if (this->cancellable)
 157     g_object_unref (this->cancellable);
 158   this->cancellable = NULL;
 159
 160   if (this->socket)
 161     g_object_unref (this->socket);
 162   this->socket = NULL;
 163
 164   g_free (this->host);
 165   this->host = NULL;
 166
 167   G_OBJECT_CLASS (parent_class)->finalize (gobject);
 168 }
 169
 170 static gboolean
 171 gst_tcp_client_sink_setcaps (GstBaseSink * bsink, GstCaps * caps)
 172 {
 173   return TRUE;
 174 }
 175
 176 static GstFlowReturn
 177 gst_tcp_client_sink_render (GstBaseSink * bsink, GstBuffer * buf)
 178 {
 179   GstTCPClientSink *sink;
 180   GstMapInfo map;
 181   gsize written = 0;
 182   gssize rret;
 183   GError *err = NULL;
 184
 185   sink = GST_TCP_CLIENT_SINK (bsink);
 186
 187   g_return_val_if_fail (GST_OBJECT_FLAG_IS_SET (sink, GST_TCP_CLIENT_SINK_OPEN),
 188       GST_FLOW_FLUSHING);
 189
 190   gst_buffer_map (buf, &map, GST_MAP_READ);
 191   GST_LOG_OBJECT (sink, "writing %" G_GSIZE_FORMAT " bytes for buffer data",
 192       map.size);
 193
 194   /* write buffer data */
 195   while (written < map.size) {
 196     rret =
 197         g_socket_send (sink->socket, (gchar *) map.data + written,
 198         map.size - written, sink->cancellable, &err);
 199     if (rret < 0)
 200       goto write_error;
 201     written += rret;
 202   }
 203   gst_buffer_unmap (buf, &map);
 204
 205   sink->data_written += written;
 206
 207   return GST_FLOW_OK;
 208
 209   /* ERRORS */
 210 write_error:
 211   {
 212     GstFlowReturn ret;
 213
 214     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 215       ret = GST_FLOW_FLUSHING;
 216       GST_DEBUG_OBJECT (sink, "Cancelled reading from socket");
 217     } else {
 218       GST_ELEMENT_ERROR (sink, RESOURCE, WRITE,
 219           (_("Error while sending data to \"%s:%d\"."), sink->host, sink->port),
 220           ("Only %" G_GSIZE_FORMAT " of %" G_GSIZE_FORMAT " bytes written: %s",
 221               written, map.size, err->message));
 222       ret = GST_FLOW_ERROR;
 223     }
 224     gst_buffer_unmap (buf, &map);
 225     g_clear_error (&err);
 226     return ret;
 227   }
 228 }
 229
 230 static void
 231 gst_tcp_client_sink_set_property (GObject * object, guint prop_id,
 232     const GValue * value, GParamSpec * pspec)
 233 {
 234   GstTCPClientSink *tcpclientsink;
 235
 236   g_return_if_fail (GST_IS_TCP_CLIENT_SINK (object));
 237   tcpclientsink = GST_TCP_CLIENT_SINK (object);
 238
 239   switch (prop_id) {
 240     case PROP_HOST:
 241       if (!g_value_get_string (value)) {
 242         g_warning ("host property cannot be NULL");
 243         break;
 244       }
 245       g_free (tcpclientsink->host);
 246       tcpclientsink->host = g_value_dup_string (value);
 247       break;
 248     case PROP_PORT:
 249       tcpclientsink->port = g_value_get_int (value);
 250       break;
 251
 252     default:
 253       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 254       break;
 255   }
 256 }
 257
 258 static void
 259 gst_tcp_client_sink_get_property (GObject * object, guint prop_id,
 260     GValue * value, GParamSpec * pspec)
 261 {
 262   GstTCPClientSink *tcpclientsink;
 263
 264   g_return_if_fail (GST_IS_TCP_CLIENT_SINK (object));
 265   tcpclientsink = GST_TCP_CLIENT_SINK (object);
 266
 267   switch (prop_id) {
 268     case PROP_HOST:
 269       g_value_set_string (value, tcpclientsink->host);
 270       break;
 271     case PROP_PORT:
 272       g_value_set_int (value, tcpclientsink->port);
 273       break;
 274     default:
 275       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 276       break;
 277   }
 278 }
 279
 280
 281 /* create a socket for sending to remote machine */
 282 static gboolean
 283 gst_tcp_client_sink_start (GstBaseSink * bsink)
 284 {
 285   GstTCPClientSink *this = GST_TCP_CLIENT_SINK (bsink);
 286   GError *err = NULL;
 287   GInetAddress *addr;
 288   GSocketAddress *saddr;
 289   GResolver *resolver;
 290
 291   if (GST_OBJECT_FLAG_IS_SET (this, GST_TCP_CLIENT_SINK_OPEN))
 292     return TRUE;
 293
 294   /* look up name if we need to */
 295   addr = g_inet_address_new_from_string (this->host);
 296   if (!addr) {
 297     GList *results;
 298
 299     resolver = g_resolver_get_default ();
 300
 301     results =
 302         g_resolver_lookup_by_name (resolver, this->host, this->cancellable,
 303         &err);
 304     if (!results)
 305       goto name_resolve;
 306     addr = G_INET_ADDRESS (g_object_ref (results->data));
 307
 308     g_resolver_free_addresses (results);
 309     g_object_unref (resolver);
 310   }
 311 #ifndef GST_DISABLE_GST_DEBUG
 312   {
 313     gchar *ip = g_inet_address_to_string (addr);
 314
 315     GST_DEBUG_OBJECT (this, "IP address for host %s is %s", this->host, ip);
 316     g_free (ip);
 317   }
 318 #endif
 319   saddr = g_inet_socket_address_new (addr, this->port);
 320   g_object_unref (addr);
 321
 322   /* create sending client socket */
 323   GST_DEBUG_OBJECT (this, "opening sending client socket to %s:%d", this->host,
 324       this->port);
 325   this->socket =
 326       g_socket_new (g_socket_address_get_family (saddr), G_SOCKET_TYPE_STREAM,
 327       G_SOCKET_PROTOCOL_TCP, &err);
 328   if (!this->socket)
 329     goto no_socket;
 330
 331   GST_DEBUG_OBJECT (this, "opened sending client socket");
 332
 333   /* connect to server */
 334   if (!g_socket_connect (this->socket, saddr, this->cancellable, &err))
 335     goto connect_failed;
 336
 337   g_object_unref (saddr);
 338
 339   GST_OBJECT_FLAG_SET (this, GST_TCP_CLIENT_SINK_OPEN);
 340
 341   this->data_written = 0;
 342
 343   return TRUE;
 344 no_socket:
 345   {
 346     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 347         ("Failed to create socket: %s", err->message));
 348     g_clear_error (&err);
 349     g_object_unref (saddr);
 350     return FALSE;
 351   }
 352 name_resolve:
 353   {
 354     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 355       GST_DEBUG_OBJECT (this, "Cancelled name resolval");
 356     } else {
 357       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 358           ("Failed to resolve host '%s': %s", this->host, err->message));
 359     }
 360     g_clear_error (&err);
 361     g_object_unref (resolver);
 362     return FALSE;
 363   }
 364 connect_failed:
 365   {
 366     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 367       GST_DEBUG_OBJECT (this, "Cancelled connecting");
 368     } else {
 369       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 370           ("Failed to connect to host '%s:%d': %s", this->host, this->port,
 371               err->message));
 372     }
 373     g_clear_error (&err);
 374     g_object_unref (saddr);
 375     /* pretend we opened ok for proper cleanup to happen */
 376     GST_OBJECT_FLAG_SET (this, GST_TCP_CLIENT_SINK_OPEN);
 377     gst_tcp_client_sink_stop (GST_BASE_SINK (this));
 378     return FALSE;
 379   }
 380 }
 381
 382 static gboolean
 383 gst_tcp_client_sink_stop (GstBaseSink * bsink)
 384 {
 385   GstTCPClientSink *this = GST_TCP_CLIENT_SINK (bsink);
 386   GError *err = NULL;
 387
 388   if (!GST_OBJECT_FLAG_IS_SET (this, GST_TCP_CLIENT_SINK_OPEN))
 389     return TRUE;
 390
 391   if (this->socket) {
 392     GST_DEBUG_OBJECT (this, "closing socket");
 393
 394     if (!g_socket_close (this->socket, &err)) {
 395       GST_ERROR_OBJECT (this, "Failed to close socket: %s", err->message);
 396       g_clear_error (&err);
 397     }
 398     g_object_unref (this->socket);
 399     this->socket = NULL;
 400   }
 401
 402   GST_OBJECT_FLAG_UNSET (this, GST_TCP_CLIENT_SINK_OPEN);
 403
 404   return TRUE;
 405 }
 406
 407 /* will be called only between calls to start() and stop() */
 408 static gboolean
 409 gst_tcp_client_sink_unlock (GstBaseSink * bsink)
 410 {
 411   GstTCPClientSink *sink = GST_TCP_CLIENT_SINK (bsink);
 412
 413   GST_DEBUG_OBJECT (sink, "set to flushing");
 414   g_cancellable_cancel (sink->cancellable);
 415
 416   return TRUE;
 417 }
 418
 419 /* will be called only between calls to start() and stop() */
 420 static gboolean
 421 gst_tcp_client_sink_unlock_stop (GstBaseSink * bsink)
 422 {
 423   GstTCPClientSink *sink = GST_TCP_CLIENT_SINK (bsink);
 424
 425   GST_DEBUG_OBJECT (sink, "unset flushing");
 426   g_object_unref (sink->cancellable);
 427   sink->cancellable = g_cancellable_new ();
 428
 429   return TRUE;
 430 }