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   GList *addrs;
 300   GList *cur_addr;
 301   GSocketAddress *saddr;
 302   gint bound_port;
 303
 304   addrs =
 305       tcp_get_addresses (GST_ELEMENT (this), this->host,
 306       this->element.cancellable, &err);
 307   if (!addrs)
 308     goto name_resolve;
 309
 310   /* iterate over addresses until one works */
 311   cur_addr = addrs;
 312   this->server_socket =
 313       tcp_create_socket (GST_ELEMENT (this), &cur_addr, this->server_port,
 314       &saddr, &err);
 315   g_list_free_full (addrs, g_object_unref);
 316
 317   if (!this->server_socket)
 318     goto no_socket;
 319
 320   GST_DEBUG_OBJECT (this, "opened sending server socket with socket %p",
 321       this->server_socket);
 322
 323   g_socket_set_blocking (this->server_socket, FALSE);
 324
 325   /* bind it */
 326   GST_DEBUG_OBJECT (this, "binding server socket to address");
 327   if (!g_socket_bind (this->server_socket, saddr, TRUE, &err))
 328     goto bind_failed;
 329
 330   g_object_unref (saddr);
 331
 332   GST_DEBUG_OBJECT (this, "listening on server socket");
 333   g_socket_set_listen_backlog (this->server_socket, TCP_BACKLOG);
 334
 335   if (!g_socket_listen (this->server_socket, &err))
 336     goto listen_failed;
 337
 338   GST_DEBUG_OBJECT (this, "listened on server socket %p", this->server_socket);
 339
 340   if (this->server_port == 0) {
 341     saddr = g_socket_get_local_address (this->server_socket, NULL);
 342     bound_port = g_inet_socket_address_get_port ((GInetSocketAddress *) saddr);
 343     g_object_unref (saddr);
 344   } else {
 345     bound_port = this->server_port;
 346   }
 347
 348   GST_DEBUG_OBJECT (this, "listening on port %d", bound_port);
 349
 350   g_atomic_int_set (&this->current_port, bound_port);
 351
 352   g_object_notify (G_OBJECT (this), "current-port");
 353
 354   this->server_source =
 355       g_socket_create_source (this->server_socket,
 356       G_IO_IN | G_IO_OUT | G_IO_PRI | G_IO_ERR | G_IO_HUP,
 357       this->element.cancellable);
 358   g_source_set_callback (this->server_source,
 359       (GSourceFunc) gst_tcp_server_sink_socket_condition, gst_object_ref (this),
 360       (GDestroyNotify) gst_object_unref);
 361   g_source_attach (this->server_source, this->element.main_context);
 362
 363   return TRUE;
 364
 365   /* ERRORS */
 366 no_socket:
 367   {
 368     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 369         ("Failed to create socket: %s", err->message));
 370     g_clear_error (&err);
 371     return FALSE;
 372   }
 373 name_resolve:
 374   {
 375     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 376       GST_DEBUG_OBJECT (this, "Cancelled name resolution");
 377     } else {
 378       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 379           ("Failed to resolve host '%s': %s", this->host, err->message));
 380     }
 381     g_clear_error (&err);
 382     return FALSE;
 383   }
 384 bind_failed:
 385   {
 386     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 387       GST_DEBUG_OBJECT (this, "Cancelled binding");
 388     } else {
 389       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 390           ("Failed to bind on host '%s:%d': %s", this->host, this->server_port,
 391               err->message));
 392     }
 393     g_clear_error (&err);
 394     g_object_unref (saddr);
 395     gst_tcp_server_sink_close (GST_MULTI_HANDLE_SINK (&this->element));
 396     return FALSE;
 397   }
 398 listen_failed:
 399   {
 400     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 401       GST_DEBUG_OBJECT (this, "Cancelled listening");
 402     } else {
 403       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 404           ("Failed to listen on host '%s:%d': %s", this->host,
 405               this->server_port, err->message));
 406     }
 407     g_clear_error (&err);
 408     gst_tcp_server_sink_close (GST_MULTI_HANDLE_SINK (&this->element));
 409     return FALSE;
 410   }
 411 }
 412
 413 static gboolean
 414 gst_tcp_server_sink_close (GstMultiHandleSink * parent)
 415 {
 416   GstTCPServerSink *this = GST_TCP_SERVER_SINK (parent);
 417
 418   if (this->server_source) {
 419     g_source_destroy (this->server_source);
 420     g_source_unref (this->server_source);
 421     this->server_source = NULL;
 422   }
 423
 424   if (this->server_socket) {
 425     GError *err = NULL;
 426
 427     GST_DEBUG_OBJECT (this, "closing socket");
 428
 429     if (!g_socket_close (this->server_socket, &err)) {
 430       GST_ERROR_OBJECT (this, "Failed to close socket: %s", err->message);
 431       g_clear_error (&err);
 432     }
 433     g_object_unref (this->server_socket);
 434     this->server_socket = NULL;
 435
 436     g_atomic_int_set (&this->current_port, 0);
 437     g_object_notify (G_OBJECT (this), "current-port");
 438   }
 439
 440   return TRUE;
 441 }