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_static_pad_template (gstelement_class, &sink_template);
 134
 135   gst_element_class_set_static_metadata (gstelement_class, "UDP packet sender",
 136       "Sink/Network",
 137       "Send data over the network via UDP with packet destinations picked up "
 138       "dynamically from meta on the buffers passed",
 139       "Philippe Khalaf <burger@speedy.org>");
 140
 141   gstbasesink_class->render = gst_dynudpsink_render;
 142   gstbasesink_class->start = gst_dynudpsink_start;
 143   gstbasesink_class->stop = gst_dynudpsink_stop;
 144   gstbasesink_class->unlock = gst_dynudpsink_unlock;
 145   gstbasesink_class->unlock_stop = gst_dynudpsink_unlock_stop;
 146
 147   klass->get_stats = gst_dynudpsink_get_stats;
 148
 149   GST_DEBUG_CATEGORY_INIT (dynudpsink_debug, "dynudpsink", 0, "UDP sink");
 150 }
 151
 152 static void
 153 gst_dynudpsink_init (GstDynUDPSink * sink)
 154 {
 155   sink->socket = UDP_DEFAULT_SOCKET;
 156   sink->socket_v6 = UDP_DEFAULT_SOCKET;
 157   sink->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
 158   sink->external_socket = FALSE;
 159   sink->bind_address = UDP_DEFAULT_BIND_ADDRESS;
 160   sink->bind_port = UDP_DEFAULT_BIND_PORT;
 161
 162   sink->used_socket = NULL;
 163   sink->used_socket_v6 = NULL;
 164 }
 165
 166 static void
 167 gst_dynudpsink_finalize (GObject * object)
 168 {
 169   GstDynUDPSink *sink;
 170
 171   sink = GST_DYNUDPSINK (object);
 172
 173   if (sink->socket)
 174     g_object_unref (sink->socket);
 175   sink->socket = NULL;
 176
 177   if (sink->socket_v6)
 178     g_object_unref (sink->socket_v6);
 179   sink->socket_v6 = NULL;
 180
 181   if (sink->used_socket)
 182     g_object_unref (sink->used_socket);
 183   sink->used_socket = NULL;
 184
 185   if (sink->used_socket_v6)
 186     g_object_unref (sink->used_socket_v6);
 187   sink->used_socket_v6 = NULL;
 188
 189   g_free (sink->bind_address);
 190   sink->bind_address = NULL;
 191
 192   G_OBJECT_CLASS (parent_class)->finalize (object);
 193 }
 194
 195 static GstFlowReturn
 196 gst_dynudpsink_render (GstBaseSink * bsink, GstBuffer * buffer)
 197 {
 198   GstDynUDPSink *sink;
 199   gssize ret;
 200   GstMapInfo map;
 201   GstNetAddressMeta *meta;
 202   GSocketAddress *addr;
 203   GError *err = NULL;
 204   GSocketFamily family;
 205   GSocket *socket;
 206
 207   meta = gst_buffer_get_net_address_meta (buffer);
 208
 209   if (meta == NULL) {
 210     GST_DEBUG ("Received buffer without GstNetAddressMeta, skipping");
 211     return GST_FLOW_OK;
 212   }
 213
 214   sink = GST_DYNUDPSINK (bsink);
 215
 216   /* let's get the address from the metadata */
 217   addr = meta->addr;
 218
 219   family = g_socket_address_get_family (addr);
 220   if (family == G_SOCKET_FAMILY_IPV6 && !sink->used_socket_v6)
 221     goto invalid_family;
 222
 223   gst_buffer_map (buffer, &map, GST_MAP_READ);
 224
 225   GST_DEBUG ("about to send %" G_GSIZE_FORMAT " bytes", map.size);
 226
 227 #ifndef GST_DISABLE_GST_DEBUG
 228   {
 229     gchar *host;
 230
 231     host =
 232         g_inet_address_to_string (g_inet_socket_address_get_address
 233         (G_INET_SOCKET_ADDRESS (addr)));
 234     GST_DEBUG ("sending %" G_GSIZE_FORMAT " bytes to client %s port %d",
 235         map.size, host,
 236         g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (addr)));
 237     g_free (host);
 238   }
 239 #endif
 240
 241   /* Select socket to send from for this address */
 242   if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
 243     socket = sink->used_socket_v6;
 244   else
 245     socket = sink->used_socket;
 246
 247   ret =
 248       g_socket_send_to (socket, addr, (gchar *) map.data, map.size,
 249       sink->cancellable, &err);
 250   gst_buffer_unmap (buffer, &map);
 251
 252   if (ret < 0)
 253     goto send_error;
 254
 255   GST_DEBUG ("sent %" G_GSSIZE_FORMAT " bytes", ret);
 256
 257   return GST_FLOW_OK;
 258
 259 send_error:
 260   {
 261     GstFlowReturn flow_ret;
 262
 263     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 264       GST_DEBUG_OBJECT (sink, "send cancelled");
 265       flow_ret = GST_FLOW_FLUSHING;
 266     } else {
 267       GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
 268           ("send error: %s", err->message));
 269       flow_ret = GST_FLOW_ERROR;
 270     }
 271     g_clear_error (&err);
 272     return flow_ret;
 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 static void
 370 gst_dynudpsink_create_cancellable (GstDynUDPSink * sink)
 371 {
 372   GPollFD pollfd;
 373
 374   sink->cancellable = g_cancellable_new ();
 375   sink->made_cancel_fd = g_cancellable_make_pollfd (sink->cancellable, &pollfd);
 376 }
 377
 378 static void
 379 gst_dynudpsink_free_cancellable (GstDynUDPSink * sink)
 380 {
 381   if (sink->made_cancel_fd) {
 382     g_cancellable_release_fd (sink->cancellable);
 383     sink->made_cancel_fd = FALSE;
 384   }
 385   g_object_unref (sink->cancellable);
 386   sink->cancellable = NULL;
 387 }
 388
 389 /* create a socket for sending to remote machine */
 390 static gboolean
 391 gst_dynudpsink_start (GstBaseSink * bsink)
 392 {
 393   GstDynUDPSink *udpsink;
 394   GError *err = NULL;
 395
 396   udpsink = GST_DYNUDPSINK (bsink);
 397
 398   gst_dynudpsink_create_cancellable (udpsink);
 399
 400   udpsink->external_socket = FALSE;
 401
 402   if (udpsink->socket) {
 403     if (g_socket_get_family (udpsink->socket) == G_SOCKET_FAMILY_IPV6) {
 404       udpsink->used_socket_v6 = G_SOCKET (g_object_ref (udpsink->socket));
 405       udpsink->external_socket = TRUE;
 406     } else {
 407       udpsink->used_socket = G_SOCKET (g_object_ref (udpsink->socket));
 408       udpsink->external_socket = TRUE;
 409     }
 410   }
 411
 412   if (udpsink->socket_v6) {
 413     g_return_val_if_fail (g_socket_get_family (udpsink->socket) !=
 414         G_SOCKET_FAMILY_IPV6, FALSE);
 415
 416     if (udpsink->used_socket_v6
 417         && udpsink->used_socket_v6 != udpsink->socket_v6) {
 418       GST_ERROR_OBJECT (udpsink,
 419           "Provided different IPv6 sockets in socket and socket-v6 properties");
 420       return FALSE;
 421     }
 422
 423     udpsink->used_socket_v6 = G_SOCKET (g_object_ref (udpsink->socket_v6));
 424     udpsink->external_socket = TRUE;
 425   }
 426
 427   if (!udpsink->used_socket && !udpsink->used_socket_v6) {
 428     GSocketAddress *bind_addr;
 429     GInetAddress *bind_iaddr;
 430
 431     if (udpsink->bind_address) {
 432       GSocketFamily family;
 433
 434       bind_iaddr = g_inet_address_new_from_string (udpsink->bind_address);
 435       if (!bind_iaddr) {
 436         GList *results;
 437         GResolver *resolver;
 438
 439         resolver = g_resolver_get_default ();
 440         results =
 441             g_resolver_lookup_by_name (resolver, udpsink->bind_address,
 442             udpsink->cancellable, &err);
 443         if (!results) {
 444           g_object_unref (resolver);
 445           goto name_resolve;
 446         }
 447         bind_iaddr = G_INET_ADDRESS (g_object_ref (results->data));
 448         g_resolver_free_addresses (results);
 449         g_object_unref (resolver);
 450       }
 451
 452       bind_addr = g_inet_socket_address_new (bind_iaddr, udpsink->bind_port);
 453       g_object_unref (bind_iaddr);
 454       family = g_socket_address_get_family (G_SOCKET_ADDRESS (bind_addr));
 455
 456       if ((udpsink->used_socket =
 457               g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
 458                   G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
 459         g_object_unref (bind_addr);
 460         goto no_socket;
 461       }
 462
 463       g_socket_bind (udpsink->used_socket, bind_addr, TRUE, &err);
 464       if (err != NULL)
 465         goto bind_error;
 466     } else {
 467       /* create sender sockets if none available */
 468       if ((udpsink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
 469                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 470         goto no_socket;
 471
 472       bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV4);
 473       bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
 474       g_socket_bind (udpsink->used_socket, bind_addr, TRUE, &err);
 475       g_object_unref (bind_addr);
 476       g_object_unref (bind_iaddr);
 477       if (err != NULL)
 478         goto bind_error;
 479
 480       if ((udpsink->used_socket_v6 = g_socket_new (G_SOCKET_FAMILY_IPV6,
 481                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP,
 482                   &err)) == NULL) {
 483         GST_INFO_OBJECT (udpsink, "Failed to create IPv6 socket: %s",
 484             err->message);
 485         g_clear_error (&err);
 486       } else {
 487         bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV6);
 488         bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
 489         g_socket_bind (udpsink->used_socket_v6, bind_addr, TRUE, &err);
 490         g_object_unref (bind_addr);
 491         g_object_unref (bind_iaddr);
 492         if (err != NULL)
 493           goto bind_error;
 494       }
 495     }
 496   }
 497
 498   if (udpsink->used_socket)
 499     g_socket_set_broadcast (udpsink->used_socket, TRUE);
 500   if (udpsink->used_socket_v6)
 501     g_socket_set_broadcast (udpsink->used_socket_v6, TRUE);
 502
 503   return TRUE;
 504
 505   /* ERRORS */
 506 no_socket:
 507   {
 508     GST_ERROR_OBJECT (udpsink, "Failed to create IPv4 socket: %s",
 509         err->message);
 510     g_clear_error (&err);
 511     return FALSE;
 512   }
 513 bind_error:
 514   {
 515     GST_ELEMENT_ERROR (udpsink, RESOURCE, FAILED, (NULL),
 516         ("Failed to bind socket: %s", err->message));
 517     g_clear_error (&err);
 518     return FALSE;
 519   }
 520 name_resolve:
 521   {
 522     GST_ELEMENT_ERROR (udpsink, RESOURCE, FAILED, (NULL),
 523         ("Failed to resolve bind address %s: %s", udpsink->bind_address,
 524             err->message));
 525     g_clear_error (&err);
 526     return FALSE;
 527   }
 528 }
 529
 530 static GstStructure *
 531 gst_dynudpsink_get_stats (GstDynUDPSink * sink, const gchar * host, gint port)
 532 {
 533   return NULL;
 534 }
 535
 536 static gboolean
 537 gst_dynudpsink_stop (GstBaseSink * bsink)
 538 {
 539   GstDynUDPSink *udpsink;
 540
 541   udpsink = GST_DYNUDPSINK (bsink);
 542
 543   if (udpsink->used_socket) {
 544     if (udpsink->close_socket || !udpsink->external_socket) {
 545       GError *err = NULL;
 546
 547       if (!g_socket_close (udpsink->used_socket, &err)) {
 548         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
 549         g_clear_error (&err);
 550       }
 551     }
 552
 553     g_object_unref (udpsink->used_socket);
 554     udpsink->used_socket = NULL;
 555   }
 556
 557   if (udpsink->used_socket_v6) {
 558     if (udpsink->close_socket || !udpsink->external_socket) {
 559       GError *err = NULL;
 560
 561       if (!g_socket_close (udpsink->used_socket_v6, &err)) {
 562         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
 563         g_clear_error (&err);
 564       }
 565     }
 566
 567     g_object_unref (udpsink->used_socket_v6);
 568     udpsink->used_socket_v6 = NULL;
 569   }
 570
 571   gst_dynudpsink_free_cancellable (udpsink);
 572
 573   return TRUE;
 574 }
 575
 576 static gboolean
 577 gst_dynudpsink_unlock (GstBaseSink * bsink)
 578 {
 579   GstDynUDPSink *udpsink;
 580
 581   udpsink = GST_DYNUDPSINK (bsink);
 582
 583   g_cancellable_cancel (udpsink->cancellable);
 584
 585   return TRUE;
 586 }
 587
 588 static gboolean
 589 gst_dynudpsink_unlock_stop (GstBaseSink * bsink)
 590 {
 591   GstDynUDPSink *udpsink;
 592
 593   udpsink = GST_DYNUDPSINK (bsink);
 594
 595   gst_dynudpsink_free_cancellable (udpsink);
 596   gst_dynudpsink_create_cancellable (udpsink);
 597
 598   return TRUE;
 599 }