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., 59 Temple Place - Suite 330,
  18  * Boston, MA 02111-1307, 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 send the packets to",
  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", "The port to send the packets to",
  95           0, TCP_HIGHEST_PORT, TCP_DEFAULT_PORT,
  96           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
  97   g_object_class_install_property (gobject_class, PROP_CURRENT_PORT,
  98       g_param_spec_int ("current-port", "current-port",
  99           "The port number the socket is currently bound to", 0,
 100           TCP_HIGHEST_PORT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 101
 102   gst_element_class_set_static_metadata (gstelement_class,
 103       "TCP server sink", "Sink/Network",
 104       "Send data as a server over the network via TCP",
 105       "Thomas Vander Stichele <thomas at apestaart dot org>");
 106
 107   gstmultihandlesink_class->init = gst_tcp_server_sink_init_send;
 108   gstmultihandlesink_class->close = gst_tcp_server_sink_close;
 109   gstmultihandlesink_class->removed = gst_tcp_server_sink_removed;
 110
 111   GST_DEBUG_CATEGORY_INIT (tcpserversink_debug, "tcpserversink", 0, "TCP sink");
 112 }
 113
 114 static void
 115 gst_tcp_server_sink_init (GstTCPServerSink * this)
 116 {
 117   this->server_port = TCP_DEFAULT_PORT;
 118   /* should support as minimum 576 for IPV4 and 1500 for IPV6 */
 119   /* this->mtu = 1500; */
 120   this->host = g_strdup (TCP_DEFAULT_HOST);
 121
 122   this->server_socket = NULL;
 123 }
 124
 125 static void
 126 gst_tcp_server_sink_finalize (GObject * gobject)
 127 {
 128   GstTCPServerSink *this = GST_TCP_SERVER_SINK (gobject);
 129
 130   if (this->server_socket)
 131     g_object_unref (this->server_socket);
 132   this->server_socket = NULL;
 133   g_free (this->host);
 134   this->host = NULL;
 135
 136   G_OBJECT_CLASS (parent_class)->finalize (gobject);
 137 }
 138
 139 /* handle a read request on the server,
 140  * which indicates a new client connection */
 141 static gboolean
 142 gst_tcp_server_sink_handle_server_read (GstTCPServerSink * sink)
 143 {
 144   GSocket *client_socket;
 145   GError *err = NULL;
 146
 147   /* wait on server socket for connections */
 148   client_socket =
 149       g_socket_accept (sink->server_socket, sink->element.cancellable, &err);
 150   if (!client_socket)
 151     goto accept_failed;
 152
 153   gst_multi_handle_sink_add (GST_MULTI_HANDLE_SINK (sink),
 154       (GstMultiSinkHandle) client_socket);
 155
 156 #ifndef GST_DISABLE_GST_DEBUG
 157   {
 158     GInetSocketAddress *addr =
 159         G_INET_SOCKET_ADDRESS (g_socket_get_remote_address (client_socket,
 160             NULL));
 161     gchar *ip =
 162         g_inet_address_to_string (g_inet_socket_address_get_address (addr));
 163
 164     GST_DEBUG_OBJECT (sink, "added new client ip %s:%u with socket %p",
 165         ip, g_inet_socket_address_get_port (addr), client_socket);
 166
 167     g_free (ip);
 168   }
 169 #endif
 170
 171   return TRUE;
 172
 173   /* ERRORS */
 174 accept_failed:
 175   {
 176     GST_ELEMENT_ERROR (sink, RESOURCE, OPEN_WRITE, (NULL),
 177         ("Could not accept client on server socket %p: %s",
 178             sink->server_socket, err->message));
 179     g_clear_error (&err);
 180     return FALSE;
 181   }
 182 }
 183
 184 static void
 185 gst_tcp_server_sink_removed (GstMultiHandleSink * sink,
 186     GstMultiSinkHandle handle)
 187 {
 188 #ifndef GST_DISABLE_GST_DEBUG
 189   GstTCPServerSink *this = GST_TCP_SERVER_SINK (sink);
 190 #endif
 191   GError *err = NULL;
 192
 193   GST_DEBUG_OBJECT (this, "closing socket");
 194
 195   if (!g_socket_close (handle.socket, &err)) {
 196     GST_ERROR_OBJECT (this, "Failed to close socket: %s", err->message);
 197     g_clear_error (&err);
 198   }
 199 }
 200
 201 static gboolean
 202 gst_tcp_server_sink_socket_condition (GSocket * socket, GIOCondition condition,
 203     GstTCPServerSink * sink)
 204 {
 205   if ((condition & G_IO_ERR)) {
 206     goto error;
 207   } else if ((condition & G_IO_IN) || (condition & G_IO_PRI)) {
 208     if (!gst_tcp_server_sink_handle_server_read (sink))
 209       return FALSE;
 210   }
 211
 212   return TRUE;
 213
 214 error:
 215   GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
 216       ("client connection failed"));
 217
 218   return FALSE;
 219 }
 220
 221 static void
 222 gst_tcp_server_sink_set_property (GObject * object, guint prop_id,
 223     const GValue * value, GParamSpec * pspec)
 224 {
 225   GstTCPServerSink *sink;
 226
 227   g_return_if_fail (GST_IS_TCP_SERVER_SINK (object));
 228   sink = GST_TCP_SERVER_SINK (object);
 229
 230   switch (prop_id) {
 231     case PROP_HOST:
 232       if (!g_value_get_string (value)) {
 233         g_warning ("host property cannot be NULL");
 234         break;
 235       }
 236       g_free (sink->host);
 237       sink->host = g_strdup (g_value_get_string (value));
 238       break;
 239     case PROP_PORT:
 240       sink->server_port = g_value_get_int (value);
 241       break;
 242     default:
 243       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 244       break;
 245   }
 246 }
 247
 248 static void
 249 gst_tcp_server_sink_get_property (GObject * object, guint prop_id,
 250     GValue * value, GParamSpec * pspec)
 251 {
 252   GstTCPServerSink *sink;
 253
 254   g_return_if_fail (GST_IS_TCP_SERVER_SINK (object));
 255   sink = GST_TCP_SERVER_SINK (object);
 256
 257   switch (prop_id) {
 258     case PROP_HOST:
 259       g_value_set_string (value, sink->host);
 260       break;
 261     case PROP_PORT:
 262       g_value_set_int (value, sink->server_port);
 263       break;
 264     case PROP_CURRENT_PORT:
 265       g_value_set_int (value, g_atomic_int_get (&sink->current_port));
 266       break;
 267     default:
 268       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 269       break;
 270   }
 271 }
 272
 273
 274 /* create a socket for sending to remote machine */
 275 static gboolean
 276 gst_tcp_server_sink_init_send (GstMultiHandleSink * parent)
 277 {
 278   GstTCPServerSink *this = GST_TCP_SERVER_SINK (parent);
 279   GError *err = NULL;
 280   GInetAddress *addr;
 281   GSocketAddress *saddr;
 282   GResolver *resolver;
 283   gint bound_port;
 284
 285   /* look up name if we need to */
 286   addr = g_inet_address_new_from_string (this->host);
 287   if (!addr) {
 288     GList *results;
 289
 290     resolver = g_resolver_get_default ();
 291
 292     results =
 293         g_resolver_lookup_by_name (resolver, this->host,
 294         this->element.cancellable, &err);
 295     if (!results)
 296       goto name_resolve;
 297     addr = G_INET_ADDRESS (g_object_ref (results->data));
 298
 299     g_resolver_free_addresses (results);
 300     g_object_unref (resolver);
 301   }
 302 #ifndef GST_DISABLE_GST_DEBUG
 303   {
 304     gchar *ip = g_inet_address_to_string (addr);
 305
 306     GST_DEBUG_OBJECT (this, "IP address for host %s is %s", this->host, ip);
 307     g_free (ip);
 308   }
 309 #endif
 310   saddr = g_inet_socket_address_new (addr, this->server_port);
 311   g_object_unref (addr);
 312
 313   /* create the server listener socket */
 314   this->server_socket =
 315       g_socket_new (g_socket_address_get_family (saddr), G_SOCKET_TYPE_STREAM,
 316       G_SOCKET_PROTOCOL_TCP, &err);
 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,
 339       "listened on server socket %p, returning from connection setup",
 340       this->server_socket);
 341
 342   if (this->server_port == 0) {
 343     saddr = g_socket_get_local_address (this->server_socket, NULL);
 344     bound_port = g_inet_socket_address_get_port ((GInetSocketAddress *) saddr);
 345     g_object_unref (saddr);
 346   } else {
 347     bound_port = this->server_port;
 348   }
 349
 350   GST_DEBUG_OBJECT (this, "listening on port %d", bound_port);
 351
 352   g_atomic_int_set (&this->current_port, bound_port);
 353
 354   g_object_notify (G_OBJECT (this), "current-port");
 355
 356   this->server_source =
 357       g_socket_create_source (this->server_socket,
 358       G_IO_IN | G_IO_OUT | G_IO_PRI | G_IO_ERR | G_IO_HUP,
 359       this->element.cancellable);
 360   g_source_set_callback (this->server_source,
 361       (GSourceFunc) gst_tcp_server_sink_socket_condition, gst_object_ref (this),
 362       (GDestroyNotify) gst_object_unref);
 363   g_source_attach (this->server_source, this->element.main_context);
 364
 365   return TRUE;
 366
 367   /* ERRORS */
 368 no_socket:
 369   {
 370     GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 371         ("Failed to create socket: %s", err->message));
 372     g_clear_error (&err);
 373     g_object_unref (saddr);
 374     return FALSE;
 375   }
 376 name_resolve:
 377   {
 378     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 379       GST_DEBUG_OBJECT (this, "Cancelled name resolval");
 380     } else {
 381       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 382           ("Failed to resolve host '%s': %s", this->host, err->message));
 383     }
 384     g_clear_error (&err);
 385     g_object_unref (resolver);
 386     return FALSE;
 387   }
 388 bind_failed:
 389   {
 390     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 391       GST_DEBUG_OBJECT (this, "Cancelled binding");
 392     } else {
 393       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 394           ("Failed to bind on host '%s:%d': %s", this->host, this->server_port,
 395               err->message));
 396     }
 397     g_clear_error (&err);
 398     g_object_unref (saddr);
 399     gst_tcp_server_sink_close (GST_MULTI_HANDLE_SINK (&this->element));
 400     return FALSE;
 401   }
 402 listen_failed:
 403   {
 404     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 405       GST_DEBUG_OBJECT (this, "Cancelled listening");
 406     } else {
 407       GST_ELEMENT_ERROR (this, RESOURCE, OPEN_READ, (NULL),
 408           ("Failed to listen on host '%s:%d': %s", this->host,
 409               this->server_port, err->message));
 410     }
 411     g_clear_error (&err);
 412     gst_tcp_server_sink_close (GST_MULTI_HANDLE_SINK (&this->element));
 413     return FALSE;
 414   }
 415 }
 416
 417 static gboolean
 418 gst_tcp_server_sink_close (GstMultiHandleSink * parent)
 419 {
 420   GstTCPServerSink *this = GST_TCP_SERVER_SINK (parent);
 421
 422   if (this->server_source) {
 423     g_source_destroy (this->server_source);
 424     g_source_unref (this->server_source);
 425     this->server_source = NULL;
 426   }
 427
 428   if (this->server_socket) {
 429     GError *err = NULL;
 430
 431     GST_DEBUG_OBJECT (this, "closing socket");
 432
 433     if (!g_socket_close (this->server_socket, &err)) {
 434       GST_ERROR_OBJECT (this, "Failed to close socket: %s", err->message);
 435       g_clear_error (&err);
 436     }
 437     g_object_unref (this->server_socket);
 438     this->server_socket = NULL;
 439
 440     g_atomic_int_set (&this->current_port, 0);
 441     g_object_notify (G_OBJECT (this), "current-port");
 442   }
 443
 444   return TRUE;
 445 }