gst/udp/gstdynudpsink.c

   1 /* GStreamer
   2  * Copyright (C) <2005> Philippe Khalaf <burger@speedy.org>
   3  * Copyright (C) <2005> Nokia Corporation <kai.vehmanen@nokia.com>
   4  * Copyright (C) <2006> Joni Valtanen <joni.valtanen@movial.fi>
   5  * Copyright (C) <2012> Collabora Ltd.
   6  *   Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Library General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Library General Public
  19  * License along with this library; if not, write to the
  20  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  21  * Boston, MA 02110-1301, USA.
  22  */
  23
  24 /* FIXME 0.11: suppress warnings for deprecated API such as GValueArray
  25  * with newer GLib versions (>= 2.31.0) */
  26 #define GLIB_DISABLE_DEPRECATION_WARNINGS
  27
  28 #ifdef HAVE_CONFIG_H
  29 #include "config.h"
  30 #endif
  31 #include "gstdynudpsink.h"
  32
  33 #include <gst/net/gstnetaddressmeta.h>
  34
  35 GST_DEBUG_CATEGORY_STATIC (dynudpsink_debug);
  36 #define GST_CAT_DEFAULT (dynudpsink_debug)
  37
  38 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
  39     GST_PAD_SINK,
  40     GST_PAD_ALWAYS,
  41     GST_STATIC_CAPS_ANY);
  42
  43 /* DynUDPSink signals and args */
  44 enum
  45 {
  46   /* methods */
  47   SIGNAL_GET_STATS,
  48
  49   /* signals */
  50
  51   /* FILL ME */
  52   LAST_SIGNAL
  53 };
  54
  55 #define UDP_DEFAULT_SOCKET              NULL
  56 #define UDP_DEFAULT_CLOSE_SOCKET        TRUE
  57 #define UDP_DEFAULT_BIND_ADDRESS        NULL
  58 #define UDP_DEFAULT_BIND_PORT           0
  59
  60 enum
  61 {
  62   PROP_0,
  63   PROP_SOCKET,
  64   PROP_SOCKET_V6,
  65   PROP_CLOSE_SOCKET,
  66   PROP_BIND_ADDRESS,
  67   PROP_BIND_PORT
  68 };
  69
  70 static void gst_dynudpsink_finalize (GObject * object);
  71
  72 static GstFlowReturn gst_dynudpsink_render (GstBaseSink * sink,
  73     GstBuffer * buffer);
  74 static gboolean gst_dynudpsink_stop (GstBaseSink * bsink);
  75 static gboolean gst_dynudpsink_start (GstBaseSink * bsink);
  76 static gboolean gst_dynudpsink_unlock (GstBaseSink * bsink);
  77 static gboolean gst_dynudpsink_unlock_stop (GstBaseSink * bsink);
  78
  79 static void gst_dynudpsink_set_property (GObject * object, guint prop_id,
  80     const GValue * value, GParamSpec * pspec);
  81 static void gst_dynudpsink_get_property (GObject * object, guint prop_id,
  82     GValue * value, GParamSpec * pspec);
  83 static GstStructure *gst_dynudpsink_get_stats (GstDynUDPSink * sink,
  84     const gchar * host, gint port);
  85
  86 static guint gst_dynudpsink_signals[LAST_SIGNAL] = { 0 };
  87
  88 #define gst_dynudpsink_parent_class parent_class
  89 G_DEFINE_TYPE (GstDynUDPSink, gst_dynudpsink, GST_TYPE_BASE_SINK);
  90
  91 static void
  92 gst_dynudpsink_class_init (GstDynUDPSinkClass * klass)
  93 {
  94   GObjectClass *gobject_class;
  95   GstElementClass *gstelement_class;
  96   GstBaseSinkClass *gstbasesink_class;
  97
  98   gobject_class = (GObjectClass *) klass;
  99   gstelement_class = (GstElementClass *) klass;
 100   gstbasesink_class = (GstBaseSinkClass *) klass;
 101
 102   parent_class = g_type_class_peek_parent (klass);
 103
 104   gobject_class->set_property = gst_dynudpsink_set_property;
 105   gobject_class->get_property = gst_dynudpsink_get_property;
 106   gobject_class->finalize = gst_dynudpsink_finalize;
 107
 108   gst_dynudpsink_signals[SIGNAL_GET_STATS] =
 109       g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
 110       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 111       G_STRUCT_OFFSET (GstDynUDPSinkClass, get_stats),
 112       NULL, NULL, g_cclosure_marshal_generic, GST_TYPE_STRUCTURE, 2,
 113       G_TYPE_STRING, G_TYPE_INT);
 114
 115   g_object_class_install_property (gobject_class, PROP_SOCKET,
 116       g_param_spec_object ("socket", "Socket",
 117           "Socket to use for UDP sending. (NULL == allocate)",
 118           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 119   g_object_class_install_property (gobject_class, PROP_SOCKET_V6,
 120       g_param_spec_object ("socket-v6", "Socket IPv6",
 121           "Socket to use for UDPv6 sending. (NULL == allocate)",
 122           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 123   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 124       g_param_spec_boolean ("close-socket", "Close socket",
 125           "Close socket if passed as property on state change",
 126           UDP_DEFAULT_CLOSE_SOCKET,
 127           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 128   g_object_class_install_property (gobject_class, PROP_BIND_ADDRESS,
 129       g_param_spec_string ("bind-address", "Bind Address",
 130           "Address to bind the socket to", UDP_DEFAULT_BIND_ADDRESS,
 131           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 132   g_object_class_install_property (gobject_class, PROP_BIND_PORT,
 133       g_param_spec_int ("bind-port", "Bind Port",
 134           "Port to bind the socket to", 0, G_MAXUINT16,
 135           UDP_DEFAULT_BIND_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 136
 137   gst_element_class_add_pad_template (gstelement_class,
 138       gst_static_pad_template_get (&sink_template));
 139
 140   gst_element_class_set_static_metadata (gstelement_class, "UDP packet sender",
 141       "Sink/Network",
 142       "Send data over the network via UDP",
 143       "Philippe Khalaf <burger@speedy.org>");
 144
 145   gstbasesink_class->render = gst_dynudpsink_render;
 146   gstbasesink_class->start = gst_dynudpsink_start;
 147   gstbasesink_class->stop = gst_dynudpsink_stop;
 148   gstbasesink_class->unlock = gst_dynudpsink_unlock;
 149   gstbasesink_class->unlock_stop = gst_dynudpsink_unlock_stop;
 150
 151   klass->get_stats = gst_dynudpsink_get_stats;
 152
 153   GST_DEBUG_CATEGORY_INIT (dynudpsink_debug, "dynudpsink", 0, "UDP sink");
 154 }
 155
 156 static void
 157 gst_dynudpsink_init (GstDynUDPSink * sink)
 158 {
 159   sink->socket = UDP_DEFAULT_SOCKET;
 160   sink->socket_v6 = UDP_DEFAULT_SOCKET;
 161   sink->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
 162   sink->external_socket = FALSE;
 163   sink->bind_address = UDP_DEFAULT_BIND_ADDRESS;
 164   sink->bind_port = UDP_DEFAULT_BIND_PORT;
 165
 166   sink->used_socket = NULL;
 167   sink->used_socket_v6 = NULL;
 168   sink->cancellable = g_cancellable_new ();
 169 }
 170
 171 static void
 172 gst_dynudpsink_finalize (GObject * object)
 173 {
 174   GstDynUDPSink *sink;
 175
 176   sink = GST_DYNUDPSINK (object);
 177
 178   if (sink->cancellable)
 179     g_object_unref (sink->cancellable);
 180   sink->cancellable = NULL;
 181
 182   if (sink->socket)
 183     g_object_unref (sink->socket);
 184   sink->socket = NULL;
 185
 186   if (sink->socket_v6)
 187     g_object_unref (sink->socket_v6);
 188   sink->socket_v6 = NULL;
 189
 190   if (sink->used_socket)
 191     g_object_unref (sink->used_socket);
 192   sink->used_socket = NULL;
 193
 194   if (sink->used_socket_v6)
 195     g_object_unref (sink->used_socket_v6);
 196   sink->used_socket_v6 = NULL;
 197
 198   g_free (sink->bind_address);
 199   sink->bind_address = NULL;
 200
 201   G_OBJECT_CLASS (parent_class)->finalize (object);
 202 }
 203
 204 static GstFlowReturn
 205 gst_dynudpsink_render (GstBaseSink * bsink, GstBuffer * buffer)
 206 {
 207   GstDynUDPSink *sink;
 208   gssize ret;
 209   GstMapInfo map;
 210   GstNetAddressMeta *meta;
 211   GSocketAddress *addr;
 212   GError *err = NULL;
 213   GSocketFamily family;
 214   GSocket *socket;
 215
 216   meta = gst_buffer_get_net_address_meta (buffer);
 217
 218   if (meta == NULL) {
 219     GST_DEBUG ("Received buffer without GstNetAddressMeta, skipping");
 220     return GST_FLOW_OK;
 221   }
 222
 223   sink = GST_DYNUDPSINK (bsink);
 224
 225   /* let's get the address from the metadata */
 226   addr = meta->addr;
 227
 228   family = g_socket_address_get_family (addr);
 229   if (family == G_SOCKET_FAMILY_IPV6 && !sink->used_socket_v6)
 230     goto invalid_family;
 231
 232   gst_buffer_map (buffer, &map, GST_MAP_READ);
 233
 234   GST_DEBUG ("about to send %" G_GSIZE_FORMAT " bytes", map.size);
 235
 236 #ifndef GST_DISABLE_GST_DEBUG
 237   {
 238     gchar *host;
 239
 240     host =
 241         g_inet_address_to_string (g_inet_socket_address_get_address
 242         (G_INET_SOCKET_ADDRESS (addr)));
 243     GST_DEBUG ("sending %" G_GSIZE_FORMAT " bytes to client %s port %d",
 244         map.size, host,
 245         g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (addr)));
 246     g_free (host);
 247   }
 248 #endif
 249
 250   /* Select socket to send from for this address */
 251   if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
 252     socket = sink->used_socket_v6;
 253   else
 254     socket = sink->used_socket;
 255
 256   ret =
 257       g_socket_send_to (socket, addr, (gchar *) map.data, map.size,
 258       sink->cancellable, &err);
 259   gst_buffer_unmap (buffer, &map);
 260
 261   if (ret < 0)
 262     goto send_error;
 263
 264   GST_DEBUG ("sent %" G_GSSIZE_FORMAT " bytes", ret);
 265
 266   return GST_FLOW_OK;
 267
 268 send_error:
 269   {
 270     GST_DEBUG ("got send error %s", err->message);
 271     g_clear_error (&err);
 272     return GST_FLOW_ERROR;
 273   }
 274 invalid_family:
 275   {
 276     GST_DEBUG ("invalid address family (got %d)", family);
 277     return GST_FLOW_ERROR;
 278   }
 279 }
 280
 281 static void
 282 gst_dynudpsink_set_property (GObject * object, guint prop_id,
 283     const GValue * value, GParamSpec * pspec)
 284 {
 285   GstDynUDPSink *udpsink;
 286
 287   udpsink = GST_DYNUDPSINK (object);
 288
 289   switch (prop_id) {
 290     case PROP_SOCKET:
 291       if (udpsink->socket != NULL && udpsink->socket != udpsink->used_socket &&
 292           udpsink->close_socket) {
 293         GError *err = NULL;
 294
 295         if (!g_socket_close (udpsink->socket, &err)) {
 296           GST_ERROR ("failed to close socket %p: %s", udpsink->socket,
 297               err->message);
 298           g_clear_error (&err);
 299         }
 300       }
 301       if (udpsink->socket)
 302         g_object_unref (udpsink->socket);
 303       udpsink->socket = g_value_dup_object (value);
 304       GST_DEBUG ("setting socket to %p", udpsink->socket);
 305       break;
 306     case PROP_SOCKET_V6:
 307       if (udpsink->socket_v6 != NULL
 308           && udpsink->socket_v6 != udpsink->used_socket_v6
 309           && udpsink->close_socket) {
 310         GError *err = NULL;
 311
 312         if (!g_socket_close (udpsink->socket_v6, &err)) {
 313           GST_ERROR ("failed to close socket %p: %s", udpsink->socket_v6,
 314               err->message);
 315           g_clear_error (&err);
 316         }
 317       }
 318       if (udpsink->socket_v6)
 319         g_object_unref (udpsink->socket_v6);
 320       udpsink->socket_v6 = g_value_dup_object (value);
 321       GST_DEBUG ("setting socket v6 to %p", udpsink->socket_v6);
 322       break;
 323     case PROP_CLOSE_SOCKET:
 324       udpsink->close_socket = g_value_get_boolean (value);
 325       break;
 326     case PROP_BIND_ADDRESS:
 327       g_free (udpsink->bind_address);
 328       udpsink->bind_address = g_value_dup_string (value);
 329       break;
 330     case PROP_BIND_PORT:
 331       udpsink->bind_port = g_value_get_int (value);
 332       break;
 333     default:
 334       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 335       break;
 336   }
 337 }
 338
 339 static void
 340 gst_dynudpsink_get_property (GObject * object, guint prop_id, GValue * value,
 341     GParamSpec * pspec)
 342 {
 343   GstDynUDPSink *udpsink;
 344
 345   udpsink = GST_DYNUDPSINK (object);
 346
 347   switch (prop_id) {
 348     case PROP_SOCKET:
 349       g_value_set_object (value, udpsink->socket);
 350       break;
 351     case PROP_SOCKET_V6:
 352       g_value_set_object (value, udpsink->socket_v6);
 353       break;
 354     case PROP_CLOSE_SOCKET:
 355       g_value_set_boolean (value, udpsink->close_socket);
 356       break;
 357     case PROP_BIND_ADDRESS:
 358       g_value_set_string (value, udpsink->bind_address);
 359       break;
 360     case PROP_BIND_PORT:
 361       g_value_set_int (value, udpsink->bind_port);
 362       break;
 363     default:
 364       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 365       break;
 366   }
 367 }
 368
 369 /* create a socket for sending to remote machine */
 370 static gboolean
 371 gst_dynudpsink_start (GstBaseSink * bsink)
 372 {
 373   GstDynUDPSink *udpsink;
 374   GError *err = NULL;
 375
 376   udpsink = GST_DYNUDPSINK (bsink);
 377
 378   udpsink->external_socket = FALSE;
 379
 380   if (udpsink->socket) {
 381     if (g_socket_get_family (udpsink->socket) == G_SOCKET_FAMILY_IPV6) {
 382       udpsink->used_socket_v6 = G_SOCKET (g_object_ref (udpsink->socket));
 383       udpsink->external_socket = TRUE;
 384     } else {
 385       udpsink->used_socket = G_SOCKET (g_object_ref (udpsink->socket));
 386       udpsink->external_socket = TRUE;
 387     }
 388   }
 389
 390   if (udpsink->socket_v6) {
 391     g_return_val_if_fail (g_socket_get_family (udpsink->socket) !=
 392         G_SOCKET_FAMILY_IPV6, FALSE);
 393
 394     if (udpsink->used_socket_v6
 395         && udpsink->used_socket_v6 != udpsink->socket_v6) {
 396       GST_ERROR_OBJECT (udpsink,
 397           "Provided different IPv6 sockets in socket and socket-v6 properties");
 398       return FALSE;
 399     }
 400
 401     udpsink->used_socket_v6 = G_SOCKET (g_object_ref (udpsink->socket_v6));
 402     udpsink->external_socket = TRUE;
 403   }
 404
 405   if (!udpsink->used_socket && !udpsink->used_socket_v6) {
 406     GSocketAddress *bind_addr;
 407     GInetAddress *bind_iaddr;
 408
 409     if (udpsink->bind_address) {
 410       GSocketFamily family;
 411
 412       bind_iaddr = g_inet_address_new_from_string (udpsink->bind_address);
 413       if (!bind_iaddr) {
 414         GList *results;
 415         GResolver *resolver;
 416
 417         resolver = g_resolver_get_default ();
 418         results =
 419             g_resolver_lookup_by_name (resolver, udpsink->bind_address,
 420             udpsink->cancellable, &err);
 421         if (!results) {
 422           g_object_unref (resolver);
 423           goto name_resolve;
 424         }
 425         bind_iaddr = G_INET_ADDRESS (g_object_ref (results->data));
 426         g_resolver_free_addresses (results);
 427         g_object_unref (resolver);
 428       }
 429
 430       bind_addr = g_inet_socket_address_new (bind_iaddr, udpsink->bind_port);
 431       g_object_unref (bind_iaddr);
 432       family = g_socket_address_get_family (G_SOCKET_ADDRESS (bind_addr));
 433
 434       if ((udpsink->used_socket =
 435               g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
 436                   G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
 437         g_object_unref (bind_addr);
 438         goto no_socket;
 439       }
 440
 441       g_socket_bind (udpsink->used_socket, bind_addr, TRUE, &err);
 442       if (err != NULL)
 443         goto bind_error;
 444     } else {
 445       /* create sender sockets if none available */
 446       if ((udpsink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
 447                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 448         goto no_socket;
 449
 450       bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV4);
 451       bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
 452       g_socket_bind (udpsink->used_socket, bind_addr, TRUE, &err);
 453       g_object_unref (bind_addr);
 454       g_object_unref (bind_iaddr);
 455       if (err != NULL)
 456         goto bind_error;
 457
 458       if ((udpsink->used_socket_v6 = g_socket_new (G_SOCKET_FAMILY_IPV6,
 459                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP,
 460                   &err)) == NULL) {
 461         GST_INFO_OBJECT (udpsink, "Failed to create IPv6 socket: %s",
 462             err->message);
 463         g_clear_error (&err);
 464       } else {
 465         bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV6);
 466         bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
 467         g_socket_bind (udpsink->used_socket_v6, bind_addr, TRUE, &err);
 468         g_object_unref (bind_addr);
 469         g_object_unref (bind_iaddr);
 470         if (err != NULL)
 471           goto bind_error;
 472       }
 473     }
 474   }
 475
 476   if (udpsink->used_socket)
 477     g_socket_set_broadcast (udpsink->used_socket, TRUE);
 478   if (udpsink->used_socket_v6)
 479     g_socket_set_broadcast (udpsink->used_socket_v6, TRUE);
 480
 481   return TRUE;
 482
 483   /* ERRORS */
 484 no_socket:
 485   {
 486     GST_ERROR_OBJECT (udpsink, "Failed to create IPv4 socket: %s",
 487         err->message);
 488     g_clear_error (&err);
 489     return FALSE;
 490   }
 491 bind_error:
 492   {
 493     GST_ELEMENT_ERROR (udpsink, RESOURCE, FAILED, (NULL),
 494         ("Failed to bind socket: %s", err->message));
 495     g_clear_error (&err);
 496     return FALSE;
 497   }
 498 name_resolve:
 499   {
 500     GST_ELEMENT_ERROR (udpsink, RESOURCE, FAILED, (NULL),
 501         ("Failed to resolve bind address %s: %s", udpsink->bind_address,
 502             err->message));
 503     g_clear_error (&err);
 504     return FALSE;
 505   }
 506 }
 507
 508 static GstStructure *
 509 gst_dynudpsink_get_stats (GstDynUDPSink * sink, const gchar * host, gint port)
 510 {
 511   return NULL;
 512 }
 513
 514 static gboolean
 515 gst_dynudpsink_stop (GstBaseSink * bsink)
 516 {
 517   GstDynUDPSink *udpsink;
 518
 519   udpsink = GST_DYNUDPSINK (bsink);
 520
 521   if (udpsink->used_socket) {
 522     if (udpsink->close_socket || !udpsink->external_socket) {
 523       GError *err = NULL;
 524
 525       if (!g_socket_close (udpsink->used_socket, &err)) {
 526         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
 527         g_clear_error (&err);
 528       }
 529     }
 530
 531     g_object_unref (udpsink->used_socket);
 532     udpsink->used_socket = NULL;
 533   }
 534
 535   if (udpsink->used_socket_v6) {
 536     if (udpsink->close_socket || !udpsink->external_socket) {
 537       GError *err = NULL;
 538
 539       if (!g_socket_close (udpsink->used_socket_v6, &err)) {
 540         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
 541         g_clear_error (&err);
 542       }
 543     }
 544
 545     g_object_unref (udpsink->used_socket_v6);
 546     udpsink->used_socket_v6 = NULL;
 547   }
 548
 549   return TRUE;
 550 }
 551
 552 static gboolean
 553 gst_dynudpsink_unlock (GstBaseSink * bsink)
 554 {
 555   GstDynUDPSink *udpsink;
 556
 557   udpsink = GST_DYNUDPSINK (bsink);
 558
 559   g_cancellable_cancel (udpsink->cancellable);
 560
 561   return TRUE;
 562 }
 563
 564 static gboolean
 565 gst_dynudpsink_unlock_stop (GstBaseSink * bsink)
 566 {
 567   GstDynUDPSink *udpsink;
 568
 569   udpsink = GST_DYNUDPSINK (bsink);
 570
 571   g_cancellable_reset (udpsink->cancellable);
 572
 573   return TRUE;
 574 }