gst/udp/gstmultiudpsink.c

   1 /* GStreamer
   2  * Copyright (C) <2007> Wim Taymans <wim.taymans@gmail.com>
   3  * Copyright (C) <2009> Jarkko Palviainen <jarkko.palviainen@sesca.com>
   4  * Copyright (C) <2012> Collabora Ltd.
   5  *   Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 /**
  24  * SECTION:element-multiudpsink
  25  * @see_also: udpsink, multifdsink
  26  *
  27  * multiudpsink is a network sink that sends UDP packets to multiple
  28  * clients.
  29  * It can be combined with rtp payload encoders to implement RTP streaming.
  30  */
  31
  32 /* FIXME 0.11: suppress warnings for deprecated API such as GValueArray
  33  * with newer GLib versions (>= 2.31.0) */
  34 #define GLIB_DISABLE_DEPRECATION_WARNINGS
  35
  36 #ifdef HAVE_CONFIG_H
  37 #include "config.h"
  38 #endif
  39 #include "gstmultiudpsink.h"
  40
  41 #include <string.h>
  42 #ifdef HAVE_SYS_SOCKET_H
  43 #include <sys/socket.h>
  44 #endif
  45
  46 #ifndef G_OS_WIN32
  47 #include <netinet/in.h>
  48 #endif
  49
  50 #include "gst/glib-compat-private.h"
  51
  52 GST_DEBUG_CATEGORY_STATIC (multiudpsink_debug);
  53 #define GST_CAT_DEFAULT (multiudpsink_debug)
  54
  55 #define UDP_MAX_SIZE 65507
  56
  57 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
  58     GST_PAD_SINK,
  59     GST_PAD_ALWAYS,
  60     GST_STATIC_CAPS_ANY);
  61
  62 /* MultiUDPSink signals and args */
  63 enum
  64 {
  65   /* methods */
  66   SIGNAL_ADD,
  67   SIGNAL_REMOVE,
  68   SIGNAL_CLEAR,
  69   SIGNAL_GET_STATS,
  70
  71   /* signals */
  72   SIGNAL_CLIENT_ADDED,
  73   SIGNAL_CLIENT_REMOVED,
  74
  75   /* FILL ME */
  76   LAST_SIGNAL
  77 };
  78
  79 #define DEFAULT_SOCKET             NULL
  80 #define DEFAULT_CLOSE_SOCKET       TRUE
  81 #define DEFAULT_USED_SOCKET        NULL
  82 #define DEFAULT_CLIENTS            NULL
  83 /* FIXME, this should be disabled by default, we don't need to join a multicast
  84  * group for sending, if this socket is also used for receiving, it should
  85  * be configured in the element that does the receive. */
  86 #define DEFAULT_AUTO_MULTICAST     TRUE
  87 #define DEFAULT_MULTICAST_IFACE    NULL
  88 #define DEFAULT_TTL                64
  89 #define DEFAULT_TTL_MC             1
  90 #define DEFAULT_LOOP               TRUE
  91 #define DEFAULT_FORCE_IPV4         FALSE
  92 #define DEFAULT_QOS_DSCP           -1
  93 #define DEFAULT_SEND_DUPLICATES    TRUE
  94 #define DEFAULT_BUFFER_SIZE        0
  95 #define DEFAULT_BIND_ADDRESS       NULL
  96 #define DEFAULT_BIND_PORT          0
  97
  98 enum
  99 {
 100   PROP_0,
 101   PROP_BYTES_TO_SERVE,
 102   PROP_BYTES_SERVED,
 103   PROP_SOCKET,
 104   PROP_SOCKET_V6,
 105   PROP_CLOSE_SOCKET,
 106   PROP_USED_SOCKET,
 107   PROP_USED_SOCKET_V6,
 108   PROP_CLIENTS,
 109   PROP_AUTO_MULTICAST,
 110   PROP_MULTICAST_IFACE,
 111   PROP_TTL,
 112   PROP_TTL_MC,
 113   PROP_LOOP,
 114   PROP_FORCE_IPV4,
 115   PROP_QOS_DSCP,
 116   PROP_SEND_DUPLICATES,
 117   PROP_BUFFER_SIZE,
 118   PROP_BIND_ADDRESS,
 119   PROP_BIND_PORT,
 120   PROP_LAST
 121 };
 122
 123 static void gst_multiudpsink_finalize (GObject * object);
 124
 125 static GstFlowReturn gst_multiudpsink_render (GstBaseSink * sink,
 126     GstBuffer * buffer);
 127
 128 static gboolean gst_multiudpsink_start (GstBaseSink * bsink);
 129 static gboolean gst_multiudpsink_stop (GstBaseSink * bsink);
 130 static gboolean gst_multiudpsink_unlock (GstBaseSink * bsink);
 131 static gboolean gst_multiudpsink_unlock_stop (GstBaseSink * bsink);
 132
 133 static void gst_multiudpsink_set_property (GObject * object, guint prop_id,
 134     const GValue * value, GParamSpec * pspec);
 135 static void gst_multiudpsink_get_property (GObject * object, guint prop_id,
 136     GValue * value, GParamSpec * pspec);
 137
 138 static void gst_multiudpsink_add_internal (GstMultiUDPSink * sink,
 139     const gchar * host, gint port, gboolean lock);
 140 static void gst_multiudpsink_clear_internal (GstMultiUDPSink * sink,
 141     gboolean lock);
 142
 143 static guint gst_multiudpsink_signals[LAST_SIGNAL] = { 0 };
 144
 145 #define gst_multiudpsink_parent_class parent_class
 146 G_DEFINE_TYPE (GstMultiUDPSink, gst_multiudpsink, GST_TYPE_BASE_SINK);
 147
 148 static void
 149 gst_multiudpsink_class_init (GstMultiUDPSinkClass * klass)
 150 {
 151   GObjectClass *gobject_class;
 152   GstElementClass *gstelement_class;
 153   GstBaseSinkClass *gstbasesink_class;
 154
 155   gobject_class = (GObjectClass *) klass;
 156   gstelement_class = (GstElementClass *) klass;
 157   gstbasesink_class = (GstBaseSinkClass *) klass;
 158
 159   gobject_class->set_property = gst_multiudpsink_set_property;
 160   gobject_class->get_property = gst_multiudpsink_get_property;
 161   gobject_class->finalize = gst_multiudpsink_finalize;
 162
 163   /**
 164    * GstMultiUDPSink::add:
 165    * @gstmultiudpsink: the sink on which the signal is emitted
 166    * @host: the hostname/IP address of the client to add
 167    * @port: the port of the client to add
 168    *
 169    * Add a client with destination @host and @port to the list of
 170    * clients. When the same host/port pair is added multiple times, the
 171    * send-duplicates property defines if the packets are sent multiple times to
 172    * the same host/port pair or not.
 173    *
 174    * When a host/port pair is added multiple times, an equal amount of remove
 175    * calls must be performed to actually remove the host/port pair from the list
 176    * of destinations.
 177    */
 178   gst_multiudpsink_signals[SIGNAL_ADD] =
 179       g_signal_new ("add", G_TYPE_FROM_CLASS (klass),
 180       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 181       G_STRUCT_OFFSET (GstMultiUDPSinkClass, add),
 182       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2,
 183       G_TYPE_STRING, G_TYPE_INT);
 184   /**
 185    * GstMultiUDPSink::remove:
 186    * @gstmultiudpsink: the sink on which the signal is emitted
 187    * @host: the hostname/IP address of the client to remove
 188    * @port: the port of the client to remove
 189    *
 190    * Remove the client with destination @host and @port from the list of
 191    * clients.
 192    */
 193   gst_multiudpsink_signals[SIGNAL_REMOVE] =
 194       g_signal_new ("remove", G_TYPE_FROM_CLASS (klass),
 195       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 196       G_STRUCT_OFFSET (GstMultiUDPSinkClass, remove),
 197       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2,
 198       G_TYPE_STRING, G_TYPE_INT);
 199   /**
 200    * GstMultiUDPSink::clear:
 201    * @gstmultiudpsink: the sink on which the signal is emitted
 202    *
 203    * Clear the list of clients.
 204    */
 205   gst_multiudpsink_signals[SIGNAL_CLEAR] =
 206       g_signal_new ("clear", G_TYPE_FROM_CLASS (klass),
 207       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 208       G_STRUCT_OFFSET (GstMultiUDPSinkClass, clear),
 209       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 0);
 210   /**
 211    * GstMultiUDPSink::get-stats:
 212    * @gstmultiudpsink: the sink on which the signal is emitted
 213    * @host: the hostname/IP address of the client to get stats on
 214    * @port: the port of the client to get stats on
 215    *
 216    * Get the statistics of the client with destination @host and @port.
 217    *
 218    * Returns: a GstStructure: bytes_sent, packets_sent,
 219    *           connect_time (in epoch seconds), disconnect_time (in epoch seconds)
 220    */
 221   gst_multiudpsink_signals[SIGNAL_GET_STATS] =
 222       g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
 223       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 224       G_STRUCT_OFFSET (GstMultiUDPSinkClass, get_stats),
 225       NULL, NULL, g_cclosure_marshal_generic, GST_TYPE_STRUCTURE, 2,
 226       G_TYPE_STRING, G_TYPE_INT);
 227   /**
 228    * GstMultiUDPSink::client-added:
 229    * @gstmultiudpsink: the sink emitting the signal
 230    * @host: the hostname/IP address of the added client
 231    * @port: the port of the added client
 232    *
 233    * Signal emited when a new client is added to the list of
 234    * clients.
 235    */
 236   gst_multiudpsink_signals[SIGNAL_CLIENT_ADDED] =
 237       g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
 238       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiUDPSinkClass, client_added),
 239       NULL, NULL, g_cclosure_marshal_generic, G_TYPE_NONE, 2,
 240       G_TYPE_STRING, G_TYPE_INT);
 241   /**
 242    * GstMultiUDPSink::client-removed:
 243    * @gstmultiudpsink: the sink emitting the signal
 244    * @host: the hostname/IP address of the removed client
 245    * @port: the port of the removed client
 246    *
 247    * Signal emited when a client is removed from the list of
 248    * clients.
 249    */
 250   gst_multiudpsink_signals[SIGNAL_CLIENT_REMOVED] =
 251       g_signal_new ("client-removed", G_TYPE_FROM_CLASS (klass),
 252       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiUDPSinkClass,
 253           client_removed), NULL, NULL, g_cclosure_marshal_generic,
 254       G_TYPE_NONE, 2, G_TYPE_STRING, G_TYPE_INT);
 255
 256   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BYTES_TO_SERVE,
 257       g_param_spec_uint64 ("bytes-to-serve", "Bytes to serve",
 258           "Number of bytes received to serve to clients", 0, G_MAXUINT64, 0,
 259           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 260   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BYTES_SERVED,
 261       g_param_spec_uint64 ("bytes-served", "Bytes served",
 262           "Total number of bytes sent to all clients", 0, G_MAXUINT64, 0,
 263           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 264   g_object_class_install_property (gobject_class, PROP_SOCKET,
 265       g_param_spec_object ("socket", "Socket Handle",
 266           "Socket to use for UDP sending. (NULL == allocate)",
 267           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 268   g_object_class_install_property (gobject_class, PROP_SOCKET_V6,
 269       g_param_spec_object ("socket-v6", "Socket Handle IPv6",
 270           "Socket to use for UDPv6 sending. (NULL == allocate)",
 271           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 272   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 273       g_param_spec_boolean ("close-socket", "Close socket",
 274           "Close socket if passed as property on state change",
 275           DEFAULT_CLOSE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 276   g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
 277       g_param_spec_object ("used-socket", "Used Socket Handle",
 278           "Socket currently in use for UDP sending. (NULL == no socket)",
 279           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 280   g_object_class_install_property (gobject_class, PROP_USED_SOCKET_V6,
 281       g_param_spec_object ("used-socket-v6", "Used Socket Handle IPv6",
 282           "Socket currently in use for UDPv6 sending. (NULL == no socket)",
 283           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 284   g_object_class_install_property (gobject_class, PROP_CLIENTS,
 285       g_param_spec_string ("clients", "Clients",
 286           "A comma separated list of host:port pairs with destinations",
 287           DEFAULT_CLIENTS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 288   g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
 289       g_param_spec_boolean ("auto-multicast",
 290           "Automatically join/leave multicast groups",
 291           "Automatically join/leave the multicast groups, FALSE means user"
 292           " has to do it himself", DEFAULT_AUTO_MULTICAST,
 293           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 294   g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
 295       g_param_spec_string ("multicast-iface", "Multicast Interface",
 296           "The network interface on which to join the multicast group",
 297           DEFAULT_MULTICAST_IFACE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 298   g_object_class_install_property (gobject_class, PROP_TTL,
 299       g_param_spec_int ("ttl", "Unicast TTL",
 300           "Used for setting the unicast TTL parameter",
 301           0, 255, DEFAULT_TTL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 302   g_object_class_install_property (gobject_class, PROP_TTL_MC,
 303       g_param_spec_int ("ttl-mc", "Multicast TTL",
 304           "Used for setting the multicast TTL parameter",
 305           0, 255, DEFAULT_TTL_MC, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 306   g_object_class_install_property (gobject_class, PROP_LOOP,
 307       g_param_spec_boolean ("loop", "Multicast Loopback",
 308           "Used for setting the multicast loop parameter. TRUE = enable,"
 309           " FALSE = disable", DEFAULT_LOOP,
 310           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 311   /**
 312    * GstMultiUDPSink::force-ipv4
 313    *
 314    * Force the use of an IPv4 socket.
 315    *
 316    * Since: 1.0.2
 317    */
 318   g_object_class_install_property (gobject_class, PROP_FORCE_IPV4,
 319       g_param_spec_boolean ("force-ipv4", "Force IPv4",
 320           "Forcing the use of an IPv4 socket (DEPRECATED, has no effect anymore)",
 321           DEFAULT_FORCE_IPV4, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 322
 323   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_QOS_DSCP,
 324       g_param_spec_int ("qos-dscp", "QoS diff srv code point",
 325           "Quality of Service, differentiated services code point (-1 default)",
 326           -1, 63, DEFAULT_QOS_DSCP,
 327           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 328   /**
 329    * GstMultiUDPSink::send-duplicates
 330    *
 331    * When a host/port pair is added mutliple times, send the packet to the host
 332    * multiple times as well.
 333    *
 334    * Since: 0.10.26
 335    */
 336   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SEND_DUPLICATES,
 337       g_param_spec_boolean ("send-duplicates", "Send Duplicates",
 338           "When a distination/port pair is added multiple times, send packets "
 339           "multiple times as well", DEFAULT_SEND_DUPLICATES,
 340           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 341
 342   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
 343       g_param_spec_int ("buffer-size", "Buffer Size",
 344           "Size of the kernel send buffer in bytes, 0=default", 0, G_MAXINT,
 345           DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 346
 347   g_object_class_install_property (gobject_class, PROP_BIND_ADDRESS,
 348       g_param_spec_string ("bind-address", "Bind Address",
 349           "Address to bind the socket to", DEFAULT_BIND_ADDRESS,
 350           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 351   g_object_class_install_property (gobject_class, PROP_BIND_PORT,
 352       g_param_spec_int ("bind-port", "Bind Port",
 353           "Port to bind the socket to", 0, G_MAXUINT16,
 354           DEFAULT_BIND_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 355
 356   gst_element_class_add_pad_template (gstelement_class,
 357       gst_static_pad_template_get (&sink_template));
 358
 359   gst_element_class_set_static_metadata (gstelement_class, "UDP packet sender",
 360       "Sink/Network",
 361       "Send data over the network via UDP",
 362       "Wim Taymans <wim.taymans@gmail.com>");
 363
 364   gstbasesink_class->render = gst_multiudpsink_render;
 365   gstbasesink_class->start = gst_multiudpsink_start;
 366   gstbasesink_class->stop = gst_multiudpsink_stop;
 367   gstbasesink_class->unlock = gst_multiudpsink_unlock;
 368   gstbasesink_class->unlock_stop = gst_multiudpsink_unlock_stop;
 369   klass->add = gst_multiudpsink_add;
 370   klass->remove = gst_multiudpsink_remove;
 371   klass->clear = gst_multiudpsink_clear;
 372   klass->get_stats = gst_multiudpsink_get_stats;
 373
 374   GST_DEBUG_CATEGORY_INIT (multiudpsink_debug, "multiudpsink", 0, "UDP sink");
 375 }
 376
 377
 378 static void
 379 gst_multiudpsink_init (GstMultiUDPSink * sink)
 380 {
 381   guint max_mem;
 382
 383   g_mutex_init (&sink->client_lock);
 384   sink->socket = DEFAULT_SOCKET;
 385   sink->socket_v6 = DEFAULT_SOCKET;
 386   sink->used_socket = DEFAULT_USED_SOCKET;
 387   sink->used_socket_v6 = DEFAULT_USED_SOCKET;
 388   sink->close_socket = DEFAULT_CLOSE_SOCKET;
 389   sink->external_socket = (sink->socket != NULL);
 390   sink->auto_multicast = DEFAULT_AUTO_MULTICAST;
 391   sink->ttl = DEFAULT_TTL;
 392   sink->ttl_mc = DEFAULT_TTL_MC;
 393   sink->loop = DEFAULT_LOOP;
 394   sink->force_ipv4 = DEFAULT_FORCE_IPV4;
 395   sink->qos_dscp = DEFAULT_QOS_DSCP;
 396   sink->send_duplicates = DEFAULT_SEND_DUPLICATES;
 397   sink->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
 398
 399   sink->cancellable = g_cancellable_new ();
 400
 401   /* allocate OutputVector and MapInfo for use in the render function, buffers can
 402    * hold up to a maximum amount of memory so we can create a maximally sized
 403    * array for them.  */
 404   max_mem = gst_buffer_get_max_memory ();
 405
 406   sink->vec = g_new (GOutputVector, max_mem);
 407   sink->map = g_new (GstMapInfo, max_mem);
 408 }
 409
 410 static GstUDPClient *
 411 create_client (GstMultiUDPSink * sink, const gchar * host, gint port)
 412 {
 413   GstUDPClient *client;
 414   GInetAddress *addr;
 415   GResolver *resolver;
 416   GError *err = NULL;
 417
 418   addr = g_inet_address_new_from_string (host);
 419   if (!addr) {
 420     GList *results;
 421
 422     resolver = g_resolver_get_default ();
 423     results =
 424         g_resolver_lookup_by_name (resolver, host, sink->cancellable, &err);
 425     if (!results)
 426       goto name_resolve;
 427     addr = G_INET_ADDRESS (g_object_ref (results->data));
 428
 429     g_resolver_free_addresses (results);
 430     g_object_unref (resolver);
 431   }
 432 #ifndef GST_DISABLE_GST_DEBUG
 433   {
 434     gchar *ip = g_inet_address_to_string (addr);
 435
 436     GST_DEBUG_OBJECT (sink, "IP address for host %s is %s", host, ip);
 437     g_free (ip);
 438   }
 439 #endif
 440
 441   client = g_slice_new0 (GstUDPClient);
 442   client->refcount = 1;
 443   client->host = g_strdup (host);
 444   client->port = port;
 445   client->addr = g_inet_socket_address_new (addr, port);
 446   g_object_unref (addr);
 447
 448   return client;
 449
 450 name_resolve:
 451   {
 452     g_object_unref (resolver);
 453
 454     return NULL;
 455   }
 456 }
 457
 458 static void
 459 free_client (GstUDPClient * client)
 460 {
 461   g_object_unref (client->addr);
 462   g_free (client->host);
 463   g_slice_free (GstUDPClient, client);
 464 }
 465
 466 static gint
 467 client_compare (GstUDPClient * a, GstUDPClient * b)
 468 {
 469   if ((a->port == b->port) && (strcmp (a->host, b->host) == 0))
 470     return 0;
 471
 472   return 1;
 473 }
 474
 475 static void
 476 gst_multiudpsink_finalize (GObject * object)
 477 {
 478   GstMultiUDPSink *sink;
 479
 480   sink = GST_MULTIUDPSINK (object);
 481
 482   g_list_foreach (sink->clients, (GFunc) free_client, NULL);
 483   g_list_free (sink->clients);
 484
 485   if (sink->socket)
 486     g_object_unref (sink->socket);
 487   sink->socket = NULL;
 488
 489   if (sink->socket_v6)
 490     g_object_unref (sink->socket_v6);
 491   sink->socket_v6 = NULL;
 492
 493   if (sink->used_socket)
 494     g_object_unref (sink->used_socket);
 495   sink->used_socket = NULL;
 496
 497   if (sink->used_socket_v6)
 498     g_object_unref (sink->used_socket_v6);
 499   sink->used_socket_v6 = NULL;
 500
 501   if (sink->cancellable)
 502     g_object_unref (sink->cancellable);
 503   sink->cancellable = NULL;
 504
 505   g_free (sink->multi_iface);
 506   sink->multi_iface = NULL;
 507
 508   g_free (sink->vec);
 509   sink->vec = NULL;
 510   g_free (sink->map);
 511   sink->map = NULL;
 512
 513   g_free (sink->bind_address);
 514   sink->bind_address = NULL;
 515
 516   g_mutex_clear (&sink->client_lock);
 517
 518   G_OBJECT_CLASS (parent_class)->finalize (object);
 519 }
 520
 521 static GstFlowReturn
 522 gst_multiudpsink_render (GstBaseSink * bsink, GstBuffer * buffer)
 523 {
 524   GstMultiUDPSink *sink;
 525   GList *clients;
 526   GOutputVector *vec;
 527   GstMapInfo *map;
 528   guint n_mem, i;
 529   gsize size;
 530   GstMemory *mem;
 531   gint num, no_clients;
 532   GError *err = NULL;
 533
 534   sink = GST_MULTIUDPSINK (bsink);
 535
 536   n_mem = gst_buffer_n_memory (buffer);
 537   if (n_mem == 0)
 538     goto no_data;
 539
 540   /* allocated on the stack, the max number of memory blocks is limited so this
 541    * should not cause stack overflows */
 542   vec = sink->vec;
 543   map = sink->map;
 544
 545   size = 0;
 546   for (i = 0; i < n_mem; i++) {
 547     mem = gst_buffer_get_memory (buffer, i);
 548     gst_memory_map (mem, &map[i], GST_MAP_READ);
 549
 550     vec[i].buffer = map[i].data;
 551     vec[i].size = map[i].size;
 552
 553     size += map[i].size;
 554   }
 555
 556   sink->bytes_to_serve += size;
 557
 558   /* grab lock while iterating and sending to clients, this should be
 559    * fast as UDP never blocks */
 560   g_mutex_lock (&sink->client_lock);
 561   GST_LOG_OBJECT (bsink, "about to send %" G_GSIZE_FORMAT " bytes in %u blocks",
 562       size, n_mem);
 563
 564   no_clients = 0;
 565   num = 0;
 566   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
 567     GstUDPClient *client;
 568     GSocket *socket;
 569     GSocketFamily family;
 570     gint count;
 571
 572     client = (GstUDPClient *) clients->data;
 573     no_clients++;
 574     GST_LOG_OBJECT (sink, "sending %" G_GSIZE_FORMAT " bytes to client %p",
 575         size, client);
 576
 577     family = g_socket_address_get_family (G_SOCKET_ADDRESS (client->addr));
 578     /* Select socket to send from for this address */
 579     if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
 580       socket = sink->used_socket_v6;
 581     else
 582       socket = sink->used_socket;
 583
 584     count = sink->send_duplicates ? client->refcount : 1;
 585
 586     while (count--) {
 587       gssize ret;
 588
 589       ret =
 590           g_socket_send_message (socket, client->addr, vec, n_mem,
 591           NULL, 0, 0, sink->cancellable, &err);
 592
 593       if (G_UNLIKELY (ret < 0)) {
 594         if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED))
 595           goto flushing;
 596
 597         /* we continue after posting a warning, next packets might be ok
 598          * again */
 599         if (size > UDP_MAX_SIZE) {
 600           GST_ELEMENT_WARNING (sink, RESOURCE, WRITE,
 601               ("Attempting to send a UDP packet larger than maximum size "
 602                   "(%" G_GSIZE_FORMAT " > %d)", size, UDP_MAX_SIZE),
 603               ("Reason: %s", err ? err->message : "unknown reason"));
 604         } else {
 605           GST_ELEMENT_WARNING (sink, RESOURCE, WRITE,
 606               ("Error sending UDP packet"), ("Reason: %s",
 607                   err ? err->message : "unknown reason"));
 608         }
 609         g_clear_error (&err);
 610       } else {
 611         num++;
 612         client->bytes_sent += ret;
 613         client->packets_sent++;
 614         sink->bytes_served += ret;
 615       }
 616     }
 617   }
 618   g_mutex_unlock (&sink->client_lock);
 619
 620   /* unmap all memory again */
 621   for (i = 0; i < n_mem; i++) {
 622     gst_memory_unmap (map[i].memory, &map[i]);
 623     gst_memory_unref (map[i].memory);
 624   }
 625
 626   GST_LOG_OBJECT (sink, "sent %" G_GSIZE_FORMAT " bytes to %d (of %d) clients",
 627       size, num, no_clients);
 628
 629   return GST_FLOW_OK;
 630
 631 no_data:
 632   {
 633     return GST_FLOW_OK;
 634   }
 635 flushing:
 636   {
 637     GST_DEBUG ("we are flushing");
 638     g_mutex_unlock (&sink->client_lock);
 639     g_clear_error (&err);
 640
 641     /* unmap all memory */
 642     for (i = 0; i < n_mem; i++) {
 643       gst_memory_unmap (map[i].memory, &map[i]);
 644       gst_memory_unref (map[i].memory);
 645     }
 646
 647     return GST_FLOW_FLUSHING;
 648   }
 649 }
 650
 651 static void
 652 gst_multiudpsink_set_clients_string (GstMultiUDPSink * sink,
 653     const gchar * string)
 654 {
 655   gchar **clients;
 656   gint i;
 657
 658   clients = g_strsplit (string, ",", 0);
 659
 660   g_mutex_lock (&sink->client_lock);
 661   /* clear all existing clients */
 662   gst_multiudpsink_clear_internal (sink, FALSE);
 663   for (i = 0; clients[i]; i++) {
 664     gchar *host, *p;
 665     gint64 port = 0;
 666
 667     host = clients[i];
 668     p = strstr (clients[i], ":");
 669     if (p != NULL) {
 670       *p = '\0';
 671       port = g_ascii_strtoll (p + 1, NULL, 10);
 672     }
 673     if (port != 0)
 674       gst_multiudpsink_add_internal (sink, host, port, FALSE);
 675   }
 676   g_mutex_unlock (&sink->client_lock);
 677
 678   g_strfreev (clients);
 679 }
 680
 681 static gchar *
 682 gst_multiudpsink_get_clients_string (GstMultiUDPSink * sink)
 683 {
 684   GString *str;
 685   GList *clients;
 686
 687   str = g_string_new ("");
 688
 689   g_mutex_lock (&sink->client_lock);
 690   clients = sink->clients;
 691   while (clients) {
 692     GstUDPClient *client;
 693     gint count;
 694
 695     client = (GstUDPClient *) clients->data;
 696
 697     clients = g_list_next (clients);
 698
 699     count = client->refcount;
 700     while (count--) {
 701       g_string_append_printf (str, "%s:%d%s", client->host, client->port,
 702           (clients || count > 1 ? "," : ""));
 703     }
 704   }
 705   g_mutex_unlock (&sink->client_lock);
 706
 707   return g_string_free (str, FALSE);
 708 }
 709
 710 static void
 711 gst_multiudpsink_setup_qos_dscp (GstMultiUDPSink * sink, GSocket * socket)
 712 {
 713   /* don't touch on -1 */
 714   if (sink->qos_dscp < 0)
 715     return;
 716
 717   if (socket == NULL)
 718     return;
 719
 720 #ifdef IP_TOS
 721   {
 722     gint tos;
 723     gint fd;
 724
 725     fd = g_socket_get_fd (socket);
 726
 727     GST_DEBUG_OBJECT (sink, "setting TOS to %d", sink->qos_dscp);
 728
 729     /* Extract and shift 6 bits of DSFIELD */
 730     tos = (sink->qos_dscp & 0x3f) << 2;
 731
 732     if (setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos)) < 0) {
 733       GST_ERROR_OBJECT (sink, "could not set TOS: %s", g_strerror (errno));
 734     }
 735 #ifdef IPV6_TCLASS
 736     if (setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos)) < 0) {
 737       GST_ERROR_OBJECT (sink, "could not set TCLASS: %s", g_strerror (errno));
 738     }
 739 #endif
 740   }
 741 #endif
 742 }
 743
 744 static void
 745 gst_multiudpsink_set_property (GObject * object, guint prop_id,
 746     const GValue * value, GParamSpec * pspec)
 747 {
 748   GstMultiUDPSink *udpsink;
 749
 750   udpsink = GST_MULTIUDPSINK (object);
 751
 752   switch (prop_id) {
 753     case PROP_SOCKET:
 754       if (udpsink->socket != NULL && udpsink->socket != udpsink->used_socket &&
 755           udpsink->close_socket) {
 756         GError *err = NULL;
 757
 758         if (!g_socket_close (udpsink->socket, &err)) {
 759           GST_ERROR ("failed to close socket %p: %s", udpsink->socket,
 760               err->message);
 761           g_clear_error (&err);
 762         }
 763       }
 764       if (udpsink->socket)
 765         g_object_unref (udpsink->socket);
 766       udpsink->socket = g_value_dup_object (value);
 767       GST_DEBUG_OBJECT (udpsink, "setting socket to %p", udpsink->socket);
 768       break;
 769     case PROP_SOCKET_V6:
 770       if (udpsink->socket_v6 != NULL
 771           && udpsink->socket_v6 != udpsink->used_socket_v6
 772           && udpsink->close_socket) {
 773         GError *err = NULL;
 774
 775         if (!g_socket_close (udpsink->socket_v6, &err)) {
 776           GST_ERROR ("failed to close socket %p: %s", udpsink->socket_v6,
 777               err->message);
 778           g_clear_error (&err);
 779         }
 780       }
 781       if (udpsink->socket_v6)
 782         g_object_unref (udpsink->socket_v6);
 783       udpsink->socket_v6 = g_value_dup_object (value);
 784       GST_DEBUG_OBJECT (udpsink, "setting socket to %p", udpsink->socket_v6);
 785       break;
 786     case PROP_CLOSE_SOCKET:
 787       udpsink->close_socket = g_value_get_boolean (value);
 788       break;
 789     case PROP_CLIENTS:
 790       gst_multiudpsink_set_clients_string (udpsink, g_value_get_string (value));
 791       break;
 792     case PROP_AUTO_MULTICAST:
 793       udpsink->auto_multicast = g_value_get_boolean (value);
 794       break;
 795     case PROP_MULTICAST_IFACE:
 796       g_free (udpsink->multi_iface);
 797
 798       if (g_value_get_string (value) == NULL)
 799         udpsink->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
 800       else
 801         udpsink->multi_iface = g_value_dup_string (value);
 802       break;
 803     case PROP_TTL:
 804       udpsink->ttl = g_value_get_int (value);
 805       break;
 806     case PROP_TTL_MC:
 807       udpsink->ttl_mc = g_value_get_int (value);
 808       break;
 809     case PROP_LOOP:
 810       udpsink->loop = g_value_get_boolean (value);
 811       break;
 812     case PROP_FORCE_IPV4:
 813       udpsink->force_ipv4 = g_value_get_boolean (value);
 814       break;
 815     case PROP_QOS_DSCP:
 816       udpsink->qos_dscp = g_value_get_int (value);
 817       gst_multiudpsink_setup_qos_dscp (udpsink, udpsink->used_socket);
 818       gst_multiudpsink_setup_qos_dscp (udpsink, udpsink->used_socket_v6);
 819       break;
 820     case PROP_SEND_DUPLICATES:
 821       udpsink->send_duplicates = g_value_get_boolean (value);
 822       break;
 823     case PROP_BUFFER_SIZE:
 824       udpsink->buffer_size = g_value_get_int (value);
 825       break;
 826     case PROP_BIND_ADDRESS:
 827       g_free (udpsink->bind_address);
 828       udpsink->bind_address = g_value_dup_string (value);
 829       break;
 830     case PROP_BIND_PORT:
 831       udpsink->bind_port = g_value_get_int (value);
 832       break;
 833     default:
 834       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 835       break;
 836   }
 837 }
 838
 839 static void
 840 gst_multiudpsink_get_property (GObject * object, guint prop_id, GValue * value,
 841     GParamSpec * pspec)
 842 {
 843   GstMultiUDPSink *udpsink;
 844
 845   udpsink = GST_MULTIUDPSINK (object);
 846
 847   switch (prop_id) {
 848     case PROP_BYTES_TO_SERVE:
 849       g_value_set_uint64 (value, udpsink->bytes_to_serve);
 850       break;
 851     case PROP_BYTES_SERVED:
 852       g_value_set_uint64 (value, udpsink->bytes_served);
 853       break;
 854     case PROP_SOCKET:
 855       g_value_set_object (value, udpsink->socket);
 856       break;
 857     case PROP_SOCKET_V6:
 858       g_value_set_object (value, udpsink->socket_v6);
 859       break;
 860     case PROP_CLOSE_SOCKET:
 861       g_value_set_boolean (value, udpsink->close_socket);
 862       break;
 863     case PROP_USED_SOCKET:
 864       g_value_set_object (value, udpsink->used_socket);
 865       break;
 866     case PROP_USED_SOCKET_V6:
 867       g_value_set_object (value, udpsink->used_socket_v6);
 868       break;
 869     case PROP_CLIENTS:
 870       g_value_take_string (value,
 871           gst_multiudpsink_get_clients_string (udpsink));
 872       break;
 873     case PROP_AUTO_MULTICAST:
 874       g_value_set_boolean (value, udpsink->auto_multicast);
 875       break;
 876     case PROP_MULTICAST_IFACE:
 877       g_value_set_string (value, udpsink->multi_iface);
 878       break;
 879     case PROP_TTL:
 880       g_value_set_int (value, udpsink->ttl);
 881       break;
 882     case PROP_TTL_MC:
 883       g_value_set_int (value, udpsink->ttl_mc);
 884       break;
 885     case PROP_LOOP:
 886       g_value_set_boolean (value, udpsink->loop);
 887       break;
 888     case PROP_FORCE_IPV4:
 889       g_value_set_boolean (value, udpsink->force_ipv4);
 890       break;
 891     case PROP_QOS_DSCP:
 892       g_value_set_int (value, udpsink->qos_dscp);
 893       break;
 894     case PROP_SEND_DUPLICATES:
 895       g_value_set_boolean (value, udpsink->send_duplicates);
 896       break;
 897     case PROP_BUFFER_SIZE:
 898       g_value_set_int (value, udpsink->buffer_size);
 899       break;
 900     case PROP_BIND_ADDRESS:
 901       g_value_set_string (value, udpsink->bind_address);
 902       break;
 903     case PROP_BIND_PORT:
 904       g_value_set_int (value, udpsink->bind_port);
 905       break;
 906     default:
 907       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 908       break;
 909   }
 910 }
 911
 912 static gboolean
 913 gst_multiudpsink_configure_client (GstMultiUDPSink * sink,
 914     GstUDPClient * client)
 915 {
 916   GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
 917   GInetAddress *addr = g_inet_socket_address_get_address (saddr);
 918   GSocketFamily family = g_socket_address_get_family (G_SOCKET_ADDRESS (saddr));
 919   GSocket *socket;
 920   GError *err = NULL;
 921
 922   GST_DEBUG_OBJECT (sink, "configuring client %p", client);
 923
 924   if (family == G_SOCKET_FAMILY_IPV6 && !sink->used_socket_v6)
 925     goto invalid_family;
 926
 927   /* Select socket to send from for this address */
 928   if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
 929     socket = sink->used_socket_v6;
 930   else
 931     socket = sink->used_socket;
 932
 933   if (g_inet_address_get_is_multicast (addr)) {
 934     GST_DEBUG_OBJECT (sink, "we have a multicast client %p", client);
 935     if (sink->auto_multicast) {
 936       GST_DEBUG_OBJECT (sink, "autojoining group");
 937       if (!g_socket_join_multicast_group (socket, addr, FALSE,
 938               sink->multi_iface, &err))
 939         goto join_group_failed;
 940     }
 941     GST_DEBUG_OBJECT (sink, "setting loop to %d", sink->loop);
 942     g_socket_set_multicast_loopback (socket, sink->loop);
 943     GST_DEBUG_OBJECT (sink, "setting ttl to %d", sink->ttl_mc);
 944     g_socket_set_multicast_ttl (socket, sink->ttl_mc);
 945   } else {
 946     GST_DEBUG_OBJECT (sink, "setting unicast ttl to %d", sink->ttl);
 947     g_socket_set_ttl (socket, sink->ttl);
 948   }
 949   return TRUE;
 950
 951   /* ERRORS */
 952 join_group_failed:
 953   {
 954     gst_multiudpsink_stop (GST_BASE_SINK (sink));
 955     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
 956         ("Could not join multicast group: %s",
 957             err ? err->message : "unknown reason"));
 958     g_clear_error (&err);
 959     return FALSE;
 960   }
 961 invalid_family:
 962   {
 963     gst_multiudpsink_stop (GST_BASE_SINK (sink));
 964     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
 965         ("Invalid address family (got %d)", family));
 966     return FALSE;
 967   }
 968 }
 969
 970 /* create a socket for sending to remote machine */
 971 static gboolean
 972 gst_multiudpsink_start (GstBaseSink * bsink)
 973 {
 974   GstMultiUDPSink *sink;
 975   GList *clients;
 976   GstUDPClient *client;
 977   GError *err = NULL;
 978
 979   sink = GST_MULTIUDPSINK (bsink);
 980
 981   sink->external_socket = FALSE;
 982
 983   if (sink->socket) {
 984     GST_DEBUG_OBJECT (sink, "using configured socket");
 985     if (g_socket_get_family (sink->socket) == G_SOCKET_FAMILY_IPV6) {
 986       sink->used_socket_v6 = G_SOCKET (g_object_ref (sink->socket));
 987       sink->external_socket = TRUE;
 988     } else {
 989       sink->used_socket = G_SOCKET (g_object_ref (sink->socket));
 990       sink->external_socket = TRUE;
 991     }
 992   }
 993
 994   if (sink->socket_v6) {
 995     GST_DEBUG_OBJECT (sink, "using configured IPv6 socket");
 996     g_return_val_if_fail (g_socket_get_family (sink->socket) !=
 997         G_SOCKET_FAMILY_IPV6, FALSE);
 998
 999     if (sink->used_socket_v6 && sink->used_socket_v6 != sink->socket_v6) {
1000       GST_ERROR_OBJECT (sink,
1001           "Provided different IPv6 sockets in socket and socket-v6 properties");
1002       return FALSE;
1003     }
1004
1005     sink->used_socket_v6 = G_SOCKET (g_object_ref (sink->socket_v6));
1006     sink->external_socket = TRUE;
1007   }
1008
1009   if (!sink->used_socket && !sink->used_socket_v6) {
1010     GSocketAddress *bind_addr;
1011     GInetAddress *bind_iaddr;
1012
1013     if (sink->bind_address) {
1014       GSocketFamily family;
1015
1016       bind_iaddr = g_inet_address_new_from_string (sink->bind_address);
1017       if (!bind_iaddr) {
1018         GList *results;
1019         GResolver *resolver;
1020
1021         resolver = g_resolver_get_default ();
1022         results =
1023             g_resolver_lookup_by_name (resolver, sink->bind_address,
1024             sink->cancellable, &err);
1025         if (!results) {
1026           g_object_unref (resolver);
1027           goto name_resolve;
1028         }
1029         bind_iaddr = G_INET_ADDRESS (g_object_ref (results->data));
1030         g_resolver_free_addresses (results);
1031         g_object_unref (resolver);
1032       }
1033
1034       bind_addr = g_inet_socket_address_new (bind_iaddr, sink->bind_port);
1035       g_object_unref (bind_iaddr);
1036       family = g_socket_address_get_family (G_SOCKET_ADDRESS (bind_addr));
1037
1038       if ((sink->used_socket =
1039               g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
1040                   G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
1041         g_object_unref (bind_addr);
1042         goto no_socket;
1043       }
1044
1045       g_socket_bind (sink->used_socket, bind_addr, TRUE, &err);
1046       if (err != NULL)
1047         goto bind_error;
1048     } else {
1049       /* create sender sockets if none available */
1050       if ((sink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
1051                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
1052         goto no_socket;
1053
1054       bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV4);
1055       bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
1056       g_socket_bind (sink->used_socket, bind_addr, TRUE, &err);
1057       g_object_unref (bind_addr);
1058       g_object_unref (bind_iaddr);
1059       if (err != NULL)
1060         goto bind_error;
1061
1062       if ((sink->used_socket_v6 = g_socket_new (G_SOCKET_FAMILY_IPV6,
1063                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP,
1064                   &err)) == NULL) {
1065         GST_INFO_OBJECT (sink, "Failed to create IPv6 socket: %s",
1066             err->message);
1067         g_clear_error (&err);
1068       } else {
1069         bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV6);
1070         bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
1071         g_socket_bind (sink->used_socket_v6, bind_addr, TRUE, &err);
1072         g_object_unref (bind_addr);
1073         g_object_unref (bind_iaddr);
1074         if (err != NULL)
1075           goto bind_error;
1076       }
1077     }
1078   }
1079 #ifdef SO_SNDBUF
1080   {
1081     socklen_t len;
1082     gint sndsize, ret;
1083
1084     len = sizeof (sndsize);
1085     if (sink->buffer_size != 0) {
1086       sndsize = sink->buffer_size;
1087
1088       GST_DEBUG_OBJECT (sink, "setting udp buffer of %d bytes", sndsize);
1089       /* set buffer size, Note that on Linux this is typically limited to a
1090        * maximum of around 100K. Also a minimum of 128 bytes is required on
1091        * Linux. */
1092
1093       if (sink->used_socket) {
1094         ret =
1095             setsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
1096             SO_SNDBUF, (void *) &sndsize, len);
1097         if (ret != 0) {
1098           GST_ELEMENT_WARNING (sink, RESOURCE, SETTINGS, (NULL),
1099               ("Could not create a buffer of requested %d bytes, %d: %s",
1100                   sndsize, ret, g_strerror (errno)));
1101         }
1102       }
1103
1104       if (sink->used_socket_v6) {
1105         ret =
1106             setsockopt (g_socket_get_fd (sink->used_socket_v6), SOL_SOCKET,
1107             SO_SNDBUF, (void *) &sndsize, len);
1108         if (ret != 0) {
1109           GST_ELEMENT_WARNING (sink, RESOURCE, SETTINGS, (NULL),
1110               ("Could not create a buffer of requested %d bytes, %d: %s",
1111                   sndsize, ret, g_strerror (errno)));
1112         }
1113       }
1114     }
1115
1116     /* read the value of the receive buffer. Note that on linux this returns 2x the
1117      * value we set because the kernel allocates extra memory for metadata.
1118      * The default on Linux is about 100K (which is about 50K without metadata) */
1119     if (sink->used_socket) {
1120       ret =
1121           getsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
1122           SO_SNDBUF, (void *) &sndsize, &len);
1123       if (ret == 0)
1124         GST_DEBUG_OBJECT (sink, "have UDP buffer of %d bytes", sndsize);
1125       else
1126         GST_DEBUG_OBJECT (sink, "could not get UDP buffer size");
1127     }
1128
1129     if (sink->used_socket_v6) {
1130       ret =
1131           getsockopt (g_socket_get_fd (sink->used_socket_v6), SOL_SOCKET,
1132           SO_SNDBUF, (void *) &sndsize, &len);
1133       if (ret == 0)
1134         GST_DEBUG_OBJECT (sink, "have UDPv6 buffer of %d bytes", sndsize);
1135       else
1136         GST_DEBUG_OBJECT (sink, "could not get UDPv6 buffer size");
1137     }
1138   }
1139 #endif
1140
1141 #ifdef SO_BINDTODEVICE
1142   if (sink->multi_iface) {
1143     if (sink->used_socket) {
1144       setsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
1145           SO_BINDTODEVICE, sink->multi_iface, strlen (sink->multi_iface));
1146     }
1147     if (sink->used_socket_v6) {
1148       setsockopt (g_socket_get_fd (sink->used_socket_v6), SOL_SOCKET,
1149           SO_BINDTODEVICE, sink->multi_iface, strlen (sink->multi_iface));
1150     }
1151   }
1152 #endif
1153
1154   if (sink->used_socket)
1155     g_socket_set_broadcast (sink->used_socket, TRUE);
1156   if (sink->used_socket_v6)
1157     g_socket_set_broadcast (sink->used_socket_v6, TRUE);
1158
1159   sink->bytes_to_serve = 0;
1160   sink->bytes_served = 0;
1161
1162   gst_multiudpsink_setup_qos_dscp (sink, sink->used_socket);
1163   gst_multiudpsink_setup_qos_dscp (sink, sink->used_socket_v6);
1164
1165   /* look for multicast clients and join multicast groups appropriately
1166      set also ttl and multicast loopback delivery appropriately  */
1167   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
1168     client = (GstUDPClient *) clients->data;
1169
1170     if (!gst_multiudpsink_configure_client (sink, client))
1171       return FALSE;
1172   }
1173   return TRUE;
1174
1175   /* ERRORS */
1176 no_socket:
1177   {
1178     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
1179         ("Could not create socket: %s", err->message));
1180     g_clear_error (&err);
1181     return FALSE;
1182   }
1183 bind_error:
1184   {
1185     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
1186         ("Failed to bind socket: %s", err->message));
1187     g_clear_error (&err);
1188     return FALSE;
1189   }
1190 name_resolve:
1191   {
1192     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
1193         ("Failed to resolve bind address %s: %s", sink->bind_address,
1194             err->message));
1195     g_clear_error (&err);
1196     return FALSE;
1197   }
1198 }
1199
1200 static gboolean
1201 gst_multiudpsink_stop (GstBaseSink * bsink)
1202 {
1203   GstMultiUDPSink *udpsink;
1204
1205   udpsink = GST_MULTIUDPSINK (bsink);
1206
1207   if (udpsink->used_socket) {
1208     if (udpsink->close_socket || !udpsink->external_socket) {
1209       GError *err = NULL;
1210
1211       if (!g_socket_close (udpsink->used_socket, &err)) {
1212         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
1213         g_clear_error (&err);
1214       }
1215     }
1216
1217     g_object_unref (udpsink->used_socket);
1218     udpsink->used_socket = NULL;
1219   }
1220
1221   if (udpsink->used_socket_v6) {
1222     if (udpsink->close_socket || !udpsink->external_socket) {
1223       GError *err = NULL;
1224
1225       if (!g_socket_close (udpsink->used_socket_v6, &err)) {
1226         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
1227         g_clear_error (&err);
1228       }
1229     }
1230
1231     g_object_unref (udpsink->used_socket_v6);
1232     udpsink->used_socket_v6 = NULL;
1233   }
1234
1235   return TRUE;
1236 }
1237
1238 static void
1239 gst_multiudpsink_add_internal (GstMultiUDPSink * sink, const gchar * host,
1240     gint port, gboolean lock)
1241 {
1242   GstUDPClient *client;
1243   GstUDPClient udpclient;
1244   GTimeVal now;
1245   GList *find;
1246
1247   udpclient.host = (gchar *) host;
1248   udpclient.port = port;
1249
1250   GST_DEBUG_OBJECT (sink, "adding client on host %s, port %d", host, port);
1251
1252   if (lock)
1253     g_mutex_lock (&sink->client_lock);
1254
1255   find = g_list_find_custom (sink->clients, &udpclient,
1256       (GCompareFunc) client_compare);
1257   if (find) {
1258     client = (GstUDPClient *) find->data;
1259
1260     GST_DEBUG_OBJECT (sink, "found %d existing clients with host %s, port %d",
1261         client->refcount, host, port);
1262     client->refcount++;
1263   } else {
1264     client = create_client (sink, host, port);
1265     if (!client)
1266       goto error;
1267
1268     g_get_current_time (&now);
1269     client->connect_time = GST_TIMEVAL_TO_TIME (now);
1270
1271     if (sink->used_socket)
1272       gst_multiudpsink_configure_client (sink, client);
1273
1274     GST_DEBUG_OBJECT (sink, "add client with host %s, port %d", host, port);
1275     sink->clients = g_list_prepend (sink->clients, client);
1276   }
1277
1278   if (lock)
1279     g_mutex_unlock (&sink->client_lock);
1280
1281   g_signal_emit (G_OBJECT (sink),
1282       gst_multiudpsink_signals[SIGNAL_CLIENT_ADDED], 0, host, port);
1283
1284   GST_DEBUG_OBJECT (sink, "added client on host %s, port %d", host, port);
1285   return;
1286
1287   /* ERRORS */
1288 error:
1289   {
1290     GST_DEBUG_OBJECT (sink, "did not add client on host %s, port %d", host,
1291         port);
1292     if (lock)
1293       g_mutex_unlock (&sink->client_lock);
1294     return;
1295   }
1296 }
1297
1298 void
1299 gst_multiudpsink_add (GstMultiUDPSink * sink, const gchar * host, gint port)
1300 {
1301   gst_multiudpsink_add_internal (sink, host, port, TRUE);
1302 }
1303
1304 void
1305 gst_multiudpsink_remove (GstMultiUDPSink * sink, const gchar * host, gint port)
1306 {
1307   GList *find;
1308   GstUDPClient udpclient;
1309   GstUDPClient *client;
1310   GTimeVal now;
1311
1312   udpclient.host = (gchar *) host;
1313   udpclient.port = port;
1314
1315   g_mutex_lock (&sink->client_lock);
1316   find = g_list_find_custom (sink->clients, &udpclient,
1317       (GCompareFunc) client_compare);
1318   if (!find)
1319     goto not_found;
1320
1321   client = (GstUDPClient *) find->data;
1322
1323   GST_DEBUG_OBJECT (sink, "found %d clients with host %s, port %d",
1324       client->refcount, host, port);
1325
1326   client->refcount--;
1327   if (client->refcount == 0) {
1328     GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
1329     GInetAddress *addr = g_inet_socket_address_get_address (saddr);
1330     GSocketFamily family =
1331         g_socket_address_get_family (G_SOCKET_ADDRESS (saddr));
1332     GSocket *socket;
1333
1334     /* Select socket to send from for this address */
1335     if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
1336       socket = sink->used_socket_v6;
1337     else
1338       socket = sink->used_socket;
1339
1340     GST_DEBUG_OBJECT (sink, "remove client with host %s, port %d", host, port);
1341
1342     g_get_current_time (&now);
1343     client->disconnect_time = GST_TIMEVAL_TO_TIME (now);
1344
1345     if (socket && sink->auto_multicast
1346         && g_inet_address_get_is_multicast (addr)) {
1347       GError *err = NULL;
1348
1349       if (!g_socket_leave_multicast_group (socket, addr, FALSE,
1350               sink->multi_iface, &err)) {
1351         GST_DEBUG_OBJECT (sink, "Failed to leave multicast group: %s",
1352             err->message);
1353         g_clear_error (&err);
1354       }
1355     }
1356
1357     /* Unlock to emit signal before we delete the actual client */
1358     g_mutex_unlock (&sink->client_lock);
1359     g_signal_emit (G_OBJECT (sink),
1360         gst_multiudpsink_signals[SIGNAL_CLIENT_REMOVED], 0, host, port);
1361     g_mutex_lock (&sink->client_lock);
1362
1363     sink->clients = g_list_delete_link (sink->clients, find);
1364
1365     free_client (client);
1366   }
1367   g_mutex_unlock (&sink->client_lock);
1368
1369   return;
1370
1371   /* ERRORS */
1372 not_found:
1373   {
1374     g_mutex_unlock (&sink->client_lock);
1375     GST_WARNING_OBJECT (sink, "client at host %s, port %d not found",
1376         host, port);
1377     return;
1378   }
1379 }
1380
1381 static void
1382 gst_multiudpsink_clear_internal (GstMultiUDPSink * sink, gboolean lock)
1383 {
1384   GST_DEBUG_OBJECT (sink, "clearing");
1385   /* we only need to remove the client structure, there is no additional
1386    * socket or anything to free for UDP */
1387   if (lock)
1388     g_mutex_lock (&sink->client_lock);
1389   g_list_foreach (sink->clients, (GFunc) free_client, sink);
1390   g_list_free (sink->clients);
1391   sink->clients = NULL;
1392   if (lock)
1393     g_mutex_unlock (&sink->client_lock);
1394 }
1395
1396 void
1397 gst_multiudpsink_clear (GstMultiUDPSink * sink)
1398 {
1399   gst_multiudpsink_clear_internal (sink, TRUE);
1400 }
1401
1402 GstStructure *
1403 gst_multiudpsink_get_stats (GstMultiUDPSink * sink, const gchar * host,
1404     gint port)
1405 {
1406   GstUDPClient *client;
1407   GstStructure *result = NULL;
1408   GstUDPClient udpclient;
1409   GList *find;
1410
1411   udpclient.host = (gchar *) host;
1412   udpclient.port = port;
1413
1414   g_mutex_lock (&sink->client_lock);
1415
1416   find = g_list_find_custom (sink->clients, &udpclient,
1417       (GCompareFunc) client_compare);
1418   if (!find)
1419     goto not_found;
1420
1421   GST_DEBUG_OBJECT (sink, "stats for client with host %s, port %d", host, port);
1422
1423   client = (GstUDPClient *) find->data;
1424
1425   result = gst_structure_new_empty ("multiudpsink-stats");
1426
1427   gst_structure_set (result,
1428       "bytes-sent", G_TYPE_UINT64, client->bytes_sent,
1429       "packets-sent", G_TYPE_UINT64, client->packets_sent,
1430       "connect-time", G_TYPE_UINT64, client->connect_time,
1431       "disconnect-time", G_TYPE_UINT64, client->disconnect_time, NULL);
1432
1433   g_mutex_unlock (&sink->client_lock);
1434
1435   return result;
1436
1437   /* ERRORS */
1438 not_found:
1439   {
1440     g_mutex_unlock (&sink->client_lock);
1441     GST_WARNING_OBJECT (sink, "client with host %s, port %d not found",
1442         host, port);
1443     /* Apparently (see comment in gstmultifdsink.c) returning NULL from here may
1444      * confuse/break python bindings */
1445     return gst_structure_new_empty ("multiudpsink-stats");
1446   }
1447 }
1448
1449 static gboolean
1450 gst_multiudpsink_unlock (GstBaseSink * bsink)
1451 {
1452   GstMultiUDPSink *sink;
1453
1454   sink = GST_MULTIUDPSINK (bsink);
1455
1456   g_cancellable_cancel (sink->cancellable);
1457
1458   return TRUE;
1459 }
1460
1461 static gboolean
1462 gst_multiudpsink_unlock_stop (GstBaseSink * bsink)
1463 {
1464   GstMultiUDPSink *sink;
1465
1466   sink = GST_MULTIUDPSINK (bsink);
1467
1468   g_cancellable_reset (sink->cancellable);
1469
1470   return TRUE;
1471 }