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