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