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