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