gst/tcp/gsttcpserversink.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  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Library General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Library General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Library General Public
  16  * License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  18  * Boston, MA 02110-1301, USA.
  19  */
  20
  21 /**
  22  * SECTION:element-tcpserversink
  23  * @see_also: #multifdsink
  24  *
  25  * <refsect2>
  26  * <title>Example launch line</title>
  27  * |[
  28  * # server:
  29  * gst-launch fdsrc fd=1 ! tcpserversink port=3000
  30  * # client:
  31  * gst-launch tcpclientsrc port=3000 ! fdsink fd=2
  32  * ]|
  33  * </refsect2>
  34  */
  35
  36 #ifdef HAVE_CONFIG_H
  37 #include "config.h"
  38 #endif
  39 #include <gst/gst-i18n-plugin.h>
  40 #include <string.h>             /* memset */
  41
  42 #include "gsttcp.h"
  43 #include "gsttcpserversink.h"
  44 #include "gsttcp-marshal.h"
  45
  46 #define TCP_BACKLOG             5
  47
  48 GST_DEBUG_CATEGORY_STATIC (tcpserversink_debug);
  49 #define GST_CAT_DEFAULT (tcpserversink_debug)
  50
  51 enum
  52 {
  53   PROP_0,
  54   PROP_HOST,
  55   PROP_PORT,
  56   PROP_CURRENT_PORT
  57 };
  58
  59 static void gst_tcp_server_sink_finalize (GObject * gobject);
  60
  61 static gboolean gst_tcp_server_sink_init_send (GstMultiHandleSink * this);
  62 static gboolean gst_tcp_server_sink_close (GstMultiHandleSink * this);
  63 static void gst_tcp_server_sink_removed (GstMultiHandleSink * sink,
  64     GstMultiSinkHandle handle);
  65
  66 static void gst_tcp_server_sink_set_property (GObject * object, guint prop_id,
  67     const GValue * value, GParamSpec * pspec);
  68 static void gst_tcp_server_sink_get_property (GObject * object, guint prop_id,
  69     GValue * value, GParamSpec * pspec);
  70
  71 #define gst_tcp_server_sink_parent_class parent_class
  72 G_DEFINE_TYPE (GstTCPServerSink, gst_tcp_server_sink,
  73     GST_TYPE_MULTI_SOCKET_SINK);
  74
  75 static void
  76 gst_tcp_server_sink_class_init (GstTCPServerSinkClass * klass)
  77 {
  78   GObjectClass *gobject_class;
  79   GstElementClass *gstelement_class;
  80   GstMultiHandleSinkClass *gstmultihandlesink_class;
  81
  82   gobject_class = (GObjectClass *) klass;
  83   gstelement_class = (GstElementClass *) klass;
  84   gstmultihandlesink_class = (GstMultiHandleSinkClass *) klass;
  85
  86   gobject_class->set_property = gst_tcp_server_sink_set_property;
  87   gobject_class->get_property = gst_tcp_server_sink_get_property;
  88   gobject_class->finalize = gst_tcp_server_sink_finalize;
  89
  90   g_object_class_install_property (gobject_class, PROP_HOST,
  91       g_param_spec_string ("host", "host", "The host/IP to listen on",
  92           TCP_DEFAULT_HOST, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  93   g_object_class_install_property (gobject_class, PROP_PORT,
  94       g_param_spec_int ("port", "port",
  95           "The port to listen to (0=random available port)",
  96           0, TCP_HIGHEST_PORT, TCP_DEFAULT_PORT,
  97           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  98   /**
  99    * GstTCPServerSink:current-port:
 100    *
 101    * The port number the socket is currently bound to. Applications can use
 102    * this property to retrieve the port number actually bound to in case
 103    * the port requested was 0 (=allocate a random available port).
 104    *
 105    * Since: 1.0.2
 106    **/
 107   g_object_class_install_property (gobject_class, PROP_CURRENT_PORT,
 108       g_param_spec_int ("current-port", "current-port",
 109           "The port number the socket is currently bound to", 0,
 110           TCP_HIGHEST_PORT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 111
 112   gst_element_class_set_static_metadata (gstelement_class,
 113       "TCP server sink", "Sink/Network",
 114       "Send data as a server over the network via TCP",
 115       "Thomas Vander Stichele <thomas at apestaart dot org>");
 116
 117   gstmultihandlesink_class->init = gst_tcp_server_sink_init_send;
 118   gstmultihandlesink_class->close = gst_tcp_server_sink_close;
 119   gstmultihandlesink_class->removed = gst_tcp_server_sink_removed;
 120
 121   GST_DEBUG_CATEGORY_INIT (tcpserversink_debug, "tcpserversink", 0, "TCP sink");
 122 }
 123
 124 static void
 125 gst_tcp_server_sink_init (GstTCPServerSink * this)
 126 {
 127   this->server_port = TCP_DEFAULT_PORT;
 128   /* should support as minimum 576 for IPV4 and 1500 for IPV6 */
 129   /* this->mtu = 1500; */
 130   this->host = g_strdup (TCP_DEFAULT_HOST);
 131
 132   this->server_socket = NULL;
 133 }
 134
 135 static void
 136 gst_tcp_server_sink_finalize (GObject * gobject)
 137 {
 138   GstTCPServerSink *this = GST_TCP_SERVER_SINK (gobject);
 139
 140   if (this->server_socket)
 141     g_object_unref (this->server_socket);
 142   this->server_socket = NULL;
 143   g_free (this->host);
 144   this->host = NULL;
 145
 146   G_OBJECT_CLASS (parent_class)->finalize (gobject);
 147 }
 148
 149 /* handle a read request on the server,
 150  * which indicates a new client connection */
 151 static gboolean
 152 gst_tcp_server_sink_handle_server_read (GstTCPServerSink * sink)
 153 {
 154   GSocket *client_socket;
 155   GError *err = NULL;
 156
 157   /* wait on server socket for connections */
 158   client_socket =
 159       g_socket_accept (sink->server_socket, sink->element.cancellable, &err);
 160   if (!client_socket)
 161     goto accept_failed;
 162
 163   gst_multi_handle_sink_add (GST_MULTI_HANDLE_SINK (sink),
 164       (GstMultiSinkHandle) client_socket);
 165
 166 #ifndef GST_DISABLE_GST_DEBUG
 167   {
 168     GInetSocketAddress *addr =
 169         G_INET_SOCKET_ADDRESS (g_socket_get_remote_address (client_socket,
 170             NULL));
 171     gchar *ip =
 172         g_inet_address_to_string (g_inet_socket_address_get_address (addr));
 173
 174     GST_DEBUG_OBJECT (sink, "added new client ip %s:%u with socket %p",
 175         ip, g_inet_socket_address_get_port (addr), client_socket);
 176
 177     g_free (ip);
 178   }
 179 #endif
 180
 181   return TRUE;
 182
 183   /* ERRORS */
 184 accept_failed:
 185   {
 186     GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE, (NULL),
 187         ("Could not accept client on server socket %p: %s",
 188             sink->server_socket, err->message));
 189     g_clear_error (&err);
 190     return FALSE;
 191   }
 192 }
 193
 194 static void
 195 gst_tcp_server_sink_removed (GstMultiHandleSink * sink,
 196     GstMultiSinkHandle handle)
 197 {
 198   GError *err = NULL;
 199
 200   GST_DEBUG_OBJECT (sink, "closing socket");
 201
 202   if (!g_socket_close (handle.socket, &err)) {
 203     GST_ERROR_OBJECT (sink, "Failed to close socket: %s", err->message);
 204     g_clear_error (&err);
 205   }
 206 }
 207
 208 static gboolean
 209 gst_tcp_server_sink_socket_condition (GSocket * socket, GIOCondition condition,
 210     GstTCPServerSink * sink)
 211 {
 212   if ((condition & G_IO_ERR)) {
 213     goto error;
 214   } else if ((condition & G_IO_IN) || (condition & G_IO_PRI)) {
 215     if (!gst_tcp_server_sink_handle_server_read (sink))
 216       return FALSE;
 217   }
 218
 219   return TRUE;
 220
 221 error:
 222   GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
 223       ("client connection failed"));
 224
 225   return FALSE;
 226 }
 227
 228 static void
 229 gst_tcp_server_sink_set_property (GObject * object, guint prop_id,
 230     const GValue * value, GParamSpec * pspec)
 231 {
 232   GstTCPServerSink *sink;
 233
 234   sink = GST_TCP_SERVER_SINK (object);
 235
 236   switch (prop_id) {
 237     case PROP_HOST:
 238       if (!g_value_get_string (value)) {
 239         g_warning ("host property cannot be NULL");
 240         break;
 241       }
 242       g_free (sink->host);
 243       sink->host = g_strdup (g_value_get_string (value));
 244       break;
 245     case PROP_PORT:
 246       sink->server_port = g_value_get_int (value);
 247       break;
 248     default:
 249       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 250       break;
 251   }
 252 }
 253
 254 static void
 255 gst_tcp_server_sink_get_property (GObject * object, guint prop_id,
 256     GValue * value, GParamSpec * pspec)
 257 {
 258   GstTCPServerSink *sink;
 259
 260   sink = GST_TCP_SERVER_SINK (object);
 261
 262   switch (prop_id) {
 263     case PROP_HOST:
 264       g_value_set_string (value, sink->host);
 265       break;
 266     case PROP_PORT:
 267       g_value_set_int (value, sink->server_port);
 268       break;
 269     case PROP_CURRENT_PORT:
 270       g_value_set_int (value, g_atomic_int_get (&sink->current_port));
 271       break;
 272     default:
 273       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 274       break;
 275   }
 276 }
 277
 278
 279 /* create a socket for sending to remote machine */
 280 static gboolean
 281 gst_tcp_server_sink_init_send (GstMultiHandleSink * parent)
 282 {
 283   GstTCPServerSink *this = GST_TCP_SERVER_SINK (parent);
 284   GError *err = NULL;
 285   GInetAddress *addr;
 286   GSocketAddress *saddr;
 287   GResolver *resolver;
 288   gint bound_port;
 289
 290   /* look up name if we need to */
 291   addr = g_inet_address_new_from_string (this->host);
 292   if (!addr) {
 293     GList *results;
 294
 295     resolver = g_resolver_get_default ();
 296
 297     results =
 298         g_resolver_lookup_by_name (resolver, this->host,
 299         this->element.cancellable, &err);
 300     if (!results)
 301       goto name_resolve;
 302     addr = G_INET_ADDRESS (g_object_ref (results->data));
 303
 304     g_resolver_free_addresses (results);
 305     g_object_unref (resolver);
 306   }
 307 #ifndef GST_DISABLE_GST_DEBUG
 308   {
 309     gchar *ip = g_inet_address_to_string (addr);
 310
 311     GST_DEBUG_OBJECT (this, "IP address for host %s is %s", this->host, ip);
 312     g_free (ip);
 313   }
 314 #endif
 315   saddr = g_inet_socket_address_new (addr, this->server_port);
 316   g_object_unref (addr);
 317
 318   /* create the server listener socket */
 319   this->server_socket =
 320       g_socket_new (g_socket_address_get_family (saddr), G_SOCKET_TYPE_STREAM,
 321       G_SOCKET_PROTOCOL_TCP, &err);
 322   if (!this->server_socket)
 323     goto no_socket;
 324
 325   GST_DEBUG_OBJECT (this, "opened sending server socket with socket %p",
 326       this->server_socket);
 327
 328   g_socket_set_blocking (this->server_socket, FALSE);
 329
 330   /* bind it */
 331   GST_DEBUG_OBJECT (this, "binding server socket to address");
 332   if (!g_socket_bind (this->server_socket, saddr, TRUE, &err))
 333     goto bind_failed;
 334
 335   g_object_unref (saddr);
 336
 337   GST_DEBUG_OBJECT (this, "listening on server socket");
 338   g_socket_set_listen_backlog (this->server_socket, TCP_BACKLOG);
 339
 340   if (!g_socket_listen (this->server_socket, &err))
 341     goto listen_failed;
 342
 343   GST_DEBUG_OBJECT (this, "listened on server socket %p", this->server_socket);
 344
 345   if (this->server_port == 0) {
 346     saddr = g_socket_get_local_address (this->server_socket, NULL);
 347     bound_port = g_inet_socket_address_get_port ((GInetSocketAddress *) saddr);
 348     g_object_unref (saddr);
 349   } else {
 350     bound_port = this->server_port;
 351   }
 352
 353   GST_DEBUG_OBJECT (this, "listening on port %d", bound_port);
 354
 355   g_atomic_int_set (&this->current_port, bound_port);
 356
 357   g_object_notify (G_OBJECT (this), "current-port");
 358
 359   this->server_source =
 360       g_socket_create_source (this->server_socket,
 361       G_IO_IN | G_IO_OUT | G_IO_PRI | G_IO_ERR | G_IO_HUP,
 362       this->element.cancellable);
 363   g_source_set_callback (this->server_source,
 364       (GSourceFunc) gst_tcp_server_sink_socket_condition, gst_object_ref (this),
 365       (GDestroyNotify) gst_object_unref);
 366   g_source_attach (this->server_source, this->element.main_context);
 367
 368   return TRUE;
 369
 370   /* ERRORS */
 371 no_socket:
 372   {
 373     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 374         ("Failed to create socket: %s", err->message));
 375     g_clear_error (&err);
 376     g_object_unref (saddr);
 377     return FALSE;
 378   }
 379 name_resolve:
 380   {
 381     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 382       GST_DEBUG_OBJECT (this, "Cancelled name resolval");
 383     } else {
 384       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 385           ("Failed to resolve host '%s': %s", this->host, err->message));
 386     }
 387     g_clear_error (&err);
 388     g_object_unref (resolver);
 389     return FALSE;
 390   }
 391 bind_failed:
 392   {
 393     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 394       GST_DEBUG_OBJECT (this, "Cancelled binding");
 395     } else {
 396       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 397           ("Failed to bind on host '%s:%d': %s", this->host, this->server_port,
 398               err->message));
 399     }
 400     g_clear_error (&err);
 401     g_object_unref (saddr);
 402     gst_tcp_server_sink_close (GST_MULTI_HANDLE_SINK (&this->element));
 403     return FALSE;
 404   }
 405 listen_failed:
 406   {
 407     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 408       GST_DEBUG_OBJECT (this, "Cancelled listening");
 409     } else {
 410       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 411           ("Failed to listen on host '%s:%d': %s", this->host,
 412               this->server_port, err->message));
 413     }
 414     g_clear_error (&err);
 415     gst_tcp_server_sink_close (GST_MULTI_HANDLE_SINK (&this->element));
 416     return FALSE;
 417   }
 418 }
 419
 420 static gboolean
 421 gst_tcp_server_sink_close (GstMultiHandleSink * parent)
 422 {
 423   GstTCPServerSink *this = GST_TCP_SERVER_SINK (parent);
 424
 425   if (this->server_source) {
 426     g_source_destroy (this->server_source);
 427     g_source_unref (this->server_source);
 428     this->server_source = NULL;
 429   }
 430
 431   if (this->server_socket) {
 432     GError *err = NULL;
 433
 434     GST_DEBUG_OBJECT (this, "closing socket");
 435
 436     if (!g_socket_close (this->server_socket, &err)) {
 437       GST_ERROR_OBJECT (this, "Failed to close socket: %s", err->message);
 438       g_clear_error (&err);
 439     }
 440     g_object_unref (this->server_socket);
 441     this->server_socket = NULL;
 442
 443     g_atomic_int_set (&this->current_port, 0);
 444     g_object_notify (G_OBJECT (this), "current-port");
 445   }
 446
 447   return TRUE;
 448 }