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     return GST_FLOW_FLUSHING;
 642   }
 643 }
 644
 645 static void
 646 gst_multiudpsink_set_clients_string (GstMultiUDPSink * sink,
 647     const gchar * string)
 648 {
 649   gchar **clients;
 650   gint i;
 651
 652   clients = g_strsplit (string, ",", 0);
 653
 654   g_mutex_lock (&sink->client_lock);
 655   /* clear all existing clients */
 656   gst_multiudpsink_clear_internal (sink, FALSE);
 657   for (i = 0; clients[i]; i++) {
 658     gchar *host, *p;
 659     gint64 port = 0;
 660
 661     host = clients[i];
 662     p = strstr (clients[i], ":");
 663     if (p != NULL) {
 664       *p = '\0';
 665       port = g_ascii_strtoll (p + 1, NULL, 10);
 666     }
 667     if (port != 0)
 668       gst_multiudpsink_add_internal (sink, host, port, FALSE);
 669   }
 670   g_mutex_unlock (&sink->client_lock);
 671
 672   g_strfreev (clients);
 673 }
 674
 675 static gchar *
 676 gst_multiudpsink_get_clients_string (GstMultiUDPSink * sink)
 677 {
 678   GString *str;
 679   GList *clients;
 680
 681   str = g_string_new ("");
 682
 683   g_mutex_lock (&sink->client_lock);
 684   clients = sink->clients;
 685   while (clients) {
 686     GstUDPClient *client;
 687     gint count;
 688
 689     client = (GstUDPClient *) clients->data;
 690
 691     clients = g_list_next (clients);
 692
 693     count = client->refcount;
 694     while (count--) {
 695       g_string_append_printf (str, "%s:%d%s", client->host, client->port,
 696           (clients || count > 1 ? "," : ""));
 697     }
 698   }
 699   g_mutex_unlock (&sink->client_lock);
 700
 701   return g_string_free (str, FALSE);
 702 }
 703
 704 static void
 705 gst_multiudpsink_setup_qos_dscp (GstMultiUDPSink * sink, GSocket * socket)
 706 {
 707   /* don't touch on -1 */
 708   if (sink->qos_dscp < 0)
 709     return;
 710
 711   if (socket == NULL)
 712     return;
 713
 714 #ifdef IP_TOS
 715   {
 716     gint tos;
 717     gint fd;
 718
 719     fd = g_socket_get_fd (socket);
 720
 721     GST_DEBUG_OBJECT (sink, "setting TOS to %d", sink->qos_dscp);
 722
 723     /* Extract and shift 6 bits of DSFIELD */
 724     tos = (sink->qos_dscp & 0x3f) << 2;
 725
 726     if (setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos)) < 0) {
 727       GST_ERROR_OBJECT (sink, "could not set TOS: %s", g_strerror (errno));
 728     }
 729 #ifdef IPV6_TCLASS
 730     if (setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos)) < 0) {
 731       GST_ERROR_OBJECT (sink, "could not set TCLASS: %s", g_strerror (errno));
 732     }
 733 #endif
 734   }
 735 #endif
 736 }
 737
 738 static void
 739 gst_multiudpsink_set_property (GObject * object, guint prop_id,
 740     const GValue * value, GParamSpec * pspec)
 741 {
 742   GstMultiUDPSink *udpsink;
 743
 744   udpsink = GST_MULTIUDPSINK (object);
 745
 746   switch (prop_id) {
 747     case PROP_SOCKET:
 748       if (udpsink->socket != NULL && udpsink->socket != udpsink->used_socket &&
 749           udpsink->close_socket) {
 750         GError *err = NULL;
 751
 752         if (!g_socket_close (udpsink->socket, &err)) {
 753           GST_ERROR ("failed to close socket %p: %s", udpsink->socket,
 754               err->message);
 755           g_clear_error (&err);
 756         }
 757       }
 758       if (udpsink->socket)
 759         g_object_unref (udpsink->socket);
 760       udpsink->socket = g_value_dup_object (value);
 761       GST_DEBUG_OBJECT (udpsink, "setting socket to %p", udpsink->socket);
 762       break;
 763     case PROP_SOCKET_V6:
 764       if (udpsink->socket_v6 != NULL
 765           && udpsink->socket_v6 != udpsink->used_socket_v6
 766           && udpsink->close_socket) {
 767         GError *err = NULL;
 768
 769         if (!g_socket_close (udpsink->socket_v6, &err)) {
 770           GST_ERROR ("failed to close socket %p: %s", udpsink->socket_v6,
 771               err->message);
 772           g_clear_error (&err);
 773         }
 774       }
 775       if (udpsink->socket_v6)
 776         g_object_unref (udpsink->socket_v6);
 777       udpsink->socket_v6 = g_value_dup_object (value);
 778       GST_DEBUG_OBJECT (udpsink, "setting socket to %p", udpsink->socket_v6);
 779       break;
 780     case PROP_CLOSE_SOCKET:
 781       udpsink->close_socket = g_value_get_boolean (value);
 782       break;
 783     case PROP_CLIENTS:
 784       gst_multiudpsink_set_clients_string (udpsink, g_value_get_string (value));
 785       break;
 786     case PROP_AUTO_MULTICAST:
 787       udpsink->auto_multicast = g_value_get_boolean (value);
 788       break;
 789     case PROP_MULTICAST_IFACE:
 790       g_free (udpsink->multi_iface);
 791
 792       if (g_value_get_string (value) == NULL)
 793         udpsink->multi_iface = g_strdup (DEFAULT_MULTICAST_IFACE);
 794       else
 795         udpsink->multi_iface = g_value_dup_string (value);
 796       break;
 797     case PROP_TTL:
 798       udpsink->ttl = g_value_get_int (value);
 799       break;
 800     case PROP_TTL_MC:
 801       udpsink->ttl_mc = g_value_get_int (value);
 802       break;
 803     case PROP_LOOP:
 804       udpsink->loop = g_value_get_boolean (value);
 805       break;
 806     case PROP_FORCE_IPV4:
 807       udpsink->force_ipv4 = g_value_get_boolean (value);
 808       break;
 809     case PROP_QOS_DSCP:
 810       udpsink->qos_dscp = g_value_get_int (value);
 811       gst_multiudpsink_setup_qos_dscp (udpsink, udpsink->used_socket);
 812       gst_multiudpsink_setup_qos_dscp (udpsink, udpsink->used_socket_v6);
 813       break;
 814     case PROP_SEND_DUPLICATES:
 815       udpsink->send_duplicates = g_value_get_boolean (value);
 816       break;
 817     case PROP_BUFFER_SIZE:
 818       udpsink->buffer_size = g_value_get_int (value);
 819       break;
 820     case PROP_BIND_ADDRESS:
 821       g_free (udpsink->bind_address);
 822       udpsink->bind_address = g_value_dup_string (value);
 823       break;
 824     case PROP_BIND_PORT:
 825       udpsink->bind_port = g_value_get_int (value);
 826       break;
 827     default:
 828       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 829       break;
 830   }
 831 }
 832
 833 static void
 834 gst_multiudpsink_get_property (GObject * object, guint prop_id, GValue * value,
 835     GParamSpec * pspec)
 836 {
 837   GstMultiUDPSink *udpsink;
 838
 839   udpsink = GST_MULTIUDPSINK (object);
 840
 841   switch (prop_id) {
 842     case PROP_BYTES_TO_SERVE:
 843       g_value_set_uint64 (value, udpsink->bytes_to_serve);
 844       break;
 845     case PROP_BYTES_SERVED:
 846       g_value_set_uint64 (value, udpsink->bytes_served);
 847       break;
 848     case PROP_SOCKET:
 849       g_value_set_object (value, udpsink->socket);
 850       break;
 851     case PROP_SOCKET_V6:
 852       g_value_set_object (value, udpsink->socket_v6);
 853       break;
 854     case PROP_CLOSE_SOCKET:
 855       g_value_set_boolean (value, udpsink->close_socket);
 856       break;
 857     case PROP_USED_SOCKET:
 858       g_value_set_object (value, udpsink->used_socket);
 859       break;
 860     case PROP_USED_SOCKET_V6:
 861       g_value_set_object (value, udpsink->used_socket_v6);
 862       break;
 863     case PROP_CLIENTS:
 864       g_value_take_string (value,
 865           gst_multiudpsink_get_clients_string (udpsink));
 866       break;
 867     case PROP_AUTO_MULTICAST:
 868       g_value_set_boolean (value, udpsink->auto_multicast);
 869       break;
 870     case PROP_MULTICAST_IFACE:
 871       g_value_set_string (value, udpsink->multi_iface);
 872       break;
 873     case PROP_TTL:
 874       g_value_set_int (value, udpsink->ttl);
 875       break;
 876     case PROP_TTL_MC:
 877       g_value_set_int (value, udpsink->ttl_mc);
 878       break;
 879     case PROP_LOOP:
 880       g_value_set_boolean (value, udpsink->loop);
 881       break;
 882     case PROP_FORCE_IPV4:
 883       g_value_set_boolean (value, udpsink->force_ipv4);
 884       break;
 885     case PROP_QOS_DSCP:
 886       g_value_set_int (value, udpsink->qos_dscp);
 887       break;
 888     case PROP_SEND_DUPLICATES:
 889       g_value_set_boolean (value, udpsink->send_duplicates);
 890       break;
 891     case PROP_BUFFER_SIZE:
 892       g_value_set_int (value, udpsink->buffer_size);
 893       break;
 894     case PROP_BIND_ADDRESS:
 895       g_value_set_string (value, udpsink->bind_address);
 896       break;
 897     case PROP_BIND_PORT:
 898       g_value_set_int (value, udpsink->bind_port);
 899       break;
 900     default:
 901       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 902       break;
 903   }
 904 }
 905
 906 static gboolean
 907 gst_multiudpsink_configure_client (GstMultiUDPSink * sink,
 908     GstUDPClient * client)
 909 {
 910   GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
 911   GInetAddress *addr = g_inet_socket_address_get_address (saddr);
 912   GSocketFamily family = g_socket_address_get_family (G_SOCKET_ADDRESS (saddr));
 913   GSocket *socket;
 914   GError *err = NULL;
 915
 916   GST_DEBUG_OBJECT (sink, "configuring client %p", client);
 917
 918   if (family == G_SOCKET_FAMILY_IPV6 && !sink->used_socket_v6)
 919     goto invalid_family;
 920
 921   /* Select socket to send from for this address */
 922   if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
 923     socket = sink->used_socket_v6;
 924   else
 925     socket = sink->used_socket;
 926
 927   if (g_inet_address_get_is_multicast (addr)) {
 928     GST_DEBUG_OBJECT (sink, "we have a multicast client %p", client);
 929     if (sink->auto_multicast) {
 930       GST_DEBUG_OBJECT (sink, "autojoining group");
 931       if (!g_socket_join_multicast_group (socket, addr, FALSE,
 932               sink->multi_iface, &err))
 933         goto join_group_failed;
 934     }
 935     GST_DEBUG_OBJECT (sink, "setting loop to %d", sink->loop);
 936     g_socket_set_multicast_loopback (socket, sink->loop);
 937     GST_DEBUG_OBJECT (sink, "setting ttl to %d", sink->ttl_mc);
 938     g_socket_set_multicast_ttl (socket, sink->ttl_mc);
 939   } else {
 940     GST_DEBUG_OBJECT (sink, "setting unicast ttl to %d", sink->ttl);
 941     g_socket_set_ttl (socket, sink->ttl);
 942   }
 943   return TRUE;
 944
 945   /* ERRORS */
 946 join_group_failed:
 947   {
 948     gst_multiudpsink_stop (GST_BASE_SINK (sink));
 949     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
 950         ("Could not join multicast group: %s",
 951             err ? err->message : "unknown reason"));
 952     g_clear_error (&err);
 953     return FALSE;
 954   }
 955 invalid_family:
 956   {
 957     gst_multiudpsink_stop (GST_BASE_SINK (sink));
 958     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
 959         ("Invalid address family (got %d)", family));
 960     return FALSE;
 961   }
 962 }
 963
 964 /* create a socket for sending to remote machine */
 965 static gboolean
 966 gst_multiudpsink_start (GstBaseSink * bsink)
 967 {
 968   GstMultiUDPSink *sink;
 969   GList *clients;
 970   GstUDPClient *client;
 971   GError *err = NULL;
 972
 973   sink = GST_MULTIUDPSINK (bsink);
 974
 975   sink->external_socket = FALSE;
 976
 977   if (sink->socket) {
 978     GST_DEBUG_OBJECT (sink, "using configured socket");
 979     if (g_socket_get_family (sink->socket) == G_SOCKET_FAMILY_IPV6) {
 980       sink->used_socket_v6 = G_SOCKET (g_object_ref (sink->socket));
 981       sink->external_socket = TRUE;
 982     } else {
 983       sink->used_socket = G_SOCKET (g_object_ref (sink->socket));
 984       sink->external_socket = TRUE;
 985     }
 986   }
 987
 988   if (sink->socket_v6) {
 989     GST_DEBUG_OBJECT (sink, "using configured IPv6 socket");
 990     g_return_val_if_fail (g_socket_get_family (sink->socket) !=
 991         G_SOCKET_FAMILY_IPV6, FALSE);
 992
 993     if (sink->used_socket_v6 && sink->used_socket_v6 != sink->socket_v6) {
 994       GST_ERROR_OBJECT (sink,
 995           "Provided different IPv6 sockets in socket and socket-v6 properties");
 996       return FALSE;
 997     }
 998
 999     sink->used_socket_v6 = G_SOCKET (g_object_ref (sink->socket_v6));
1000     sink->external_socket = TRUE;
1001   }
1002
1003   if (!sink->used_socket && !sink->used_socket_v6) {
1004     GSocketAddress *bind_addr;
1005     GInetAddress *bind_iaddr;
1006
1007     if (sink->bind_address) {
1008       GSocketFamily family;
1009
1010       bind_iaddr = g_inet_address_new_from_string (sink->bind_address);
1011       if (!bind_iaddr) {
1012         GList *results;
1013         GResolver *resolver;
1014
1015         resolver = g_resolver_get_default ();
1016         results =
1017             g_resolver_lookup_by_name (resolver, sink->bind_address,
1018             sink->cancellable, &err);
1019         if (!results) {
1020           g_object_unref (resolver);
1021           goto name_resolve;
1022         }
1023         bind_iaddr = G_INET_ADDRESS (g_object_ref (results->data));
1024         g_resolver_free_addresses (results);
1025         g_object_unref (resolver);
1026       }
1027
1028       bind_addr = g_inet_socket_address_new (bind_iaddr, sink->bind_port);
1029       g_object_unref (bind_iaddr);
1030       family = g_socket_address_get_family (G_SOCKET_ADDRESS (bind_addr));
1031
1032       if ((sink->used_socket =
1033               g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
1034                   G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
1035         g_object_unref (bind_addr);
1036         goto no_socket;
1037       }
1038
1039       g_socket_bind (sink->used_socket, bind_addr, TRUE, &err);
1040       if (err != NULL)
1041         goto bind_error;
1042     } else {
1043       /* create sender sockets if none available */
1044       if ((sink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
1045                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
1046         goto no_socket;
1047
1048       bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV4);
1049       bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
1050       g_socket_bind (sink->used_socket, bind_addr, TRUE, &err);
1051       g_object_unref (bind_addr);
1052       g_object_unref (bind_iaddr);
1053       if (err != NULL)
1054         goto bind_error;
1055
1056       if ((sink->used_socket_v6 = g_socket_new (G_SOCKET_FAMILY_IPV6,
1057                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP,
1058                   &err)) == NULL) {
1059         GST_INFO_OBJECT (sink, "Failed to create IPv6 socket: %s",
1060             err->message);
1061         g_clear_error (&err);
1062       } else {
1063         bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV6);
1064         bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
1065         g_socket_bind (sink->used_socket_v6, bind_addr, TRUE, &err);
1066         g_object_unref (bind_addr);
1067         g_object_unref (bind_iaddr);
1068         if (err != NULL)
1069           goto bind_error;
1070       }
1071     }
1072   }
1073 #ifdef SO_SNDBUF
1074   {
1075     socklen_t len;
1076     gint sndsize, ret;
1077
1078     len = sizeof (sndsize);
1079     if (sink->buffer_size != 0) {
1080       sndsize = sink->buffer_size;
1081
1082       GST_DEBUG_OBJECT (sink, "setting udp buffer of %d bytes", sndsize);
1083       /* set buffer size, Note that on Linux this is typically limited to a
1084        * maximum of around 100K. Also a minimum of 128 bytes is required on
1085        * Linux. */
1086
1087       if (sink->used_socket) {
1088         ret =
1089             setsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
1090             SO_SNDBUF, (void *) &sndsize, len);
1091         if (ret != 0) {
1092           GST_ELEMENT_WARNING (sink, RESOURCE, SETTINGS, (NULL),
1093               ("Could not create a buffer of requested %d bytes, %d: %s",
1094                   sndsize, ret, g_strerror (errno)));
1095         }
1096       }
1097
1098       if (sink->used_socket_v6) {
1099         ret =
1100             setsockopt (g_socket_get_fd (sink->used_socket_v6), SOL_SOCKET,
1101             SO_SNDBUF, (void *) &sndsize, len);
1102         if (ret != 0) {
1103           GST_ELEMENT_WARNING (sink, RESOURCE, SETTINGS, (NULL),
1104               ("Could not create a buffer of requested %d bytes, %d: %s",
1105                   sndsize, ret, g_strerror (errno)));
1106         }
1107       }
1108     }
1109
1110     /* read the value of the receive buffer. Note that on linux this returns 2x the
1111      * value we set because the kernel allocates extra memory for metadata.
1112      * The default on Linux is about 100K (which is about 50K without metadata) */
1113     if (sink->used_socket) {
1114       ret =
1115           getsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
1116           SO_SNDBUF, (void *) &sndsize, &len);
1117       if (ret == 0)
1118         GST_DEBUG_OBJECT (sink, "have UDP buffer of %d bytes", sndsize);
1119       else
1120         GST_DEBUG_OBJECT (sink, "could not get UDP buffer size");
1121     }
1122
1123     if (sink->used_socket_v6) {
1124       ret =
1125           getsockopt (g_socket_get_fd (sink->used_socket_v6), SOL_SOCKET,
1126           SO_SNDBUF, (void *) &sndsize, &len);
1127       if (ret == 0)
1128         GST_DEBUG_OBJECT (sink, "have UDPv6 buffer of %d bytes", sndsize);
1129       else
1130         GST_DEBUG_OBJECT (sink, "could not get UDPv6 buffer size");
1131     }
1132   }
1133 #endif
1134
1135 #ifdef SO_BINDTODEVICE
1136   if (sink->multi_iface) {
1137     if (sink->used_socket) {
1138       setsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
1139           SO_BINDTODEVICE, sink->multi_iface, strlen (sink->multi_iface));
1140     }
1141     if (sink->used_socket_v6) {
1142       setsockopt (g_socket_get_fd (sink->used_socket_v6), SOL_SOCKET,
1143           SO_BINDTODEVICE, sink->multi_iface, strlen (sink->multi_iface));
1144     }
1145   }
1146 #endif
1147
1148   if (sink->used_socket)
1149     g_socket_set_broadcast (sink->used_socket, TRUE);
1150   if (sink->used_socket_v6)
1151     g_socket_set_broadcast (sink->used_socket_v6, TRUE);
1152
1153   sink->bytes_to_serve = 0;
1154   sink->bytes_served = 0;
1155
1156   gst_multiudpsink_setup_qos_dscp (sink, sink->used_socket);
1157   gst_multiudpsink_setup_qos_dscp (sink, sink->used_socket_v6);
1158
1159   /* look for multicast clients and join multicast groups appropriately
1160      set also ttl and multicast loopback delivery appropriately  */
1161   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
1162     client = (GstUDPClient *) clients->data;
1163
1164     if (!gst_multiudpsink_configure_client (sink, client))
1165       return FALSE;
1166   }
1167   return TRUE;
1168
1169   /* ERRORS */
1170 no_socket:
1171   {
1172     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
1173         ("Could not create socket: %s", err->message));
1174     g_clear_error (&err);
1175     return FALSE;
1176   }
1177 bind_error:
1178   {
1179     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
1180         ("Failed to bind socket: %s", err->message));
1181     g_clear_error (&err);
1182     return FALSE;
1183   }
1184 name_resolve:
1185   {
1186     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
1187         ("Failed to resolve bind address %s: %s", sink->bind_address,
1188             err->message));
1189     g_clear_error (&err);
1190     return FALSE;
1191   }
1192 }
1193
1194 static gboolean
1195 gst_multiudpsink_stop (GstBaseSink * bsink)
1196 {
1197   GstMultiUDPSink *udpsink;
1198
1199   udpsink = GST_MULTIUDPSINK (bsink);
1200
1201   if (udpsink->used_socket) {
1202     if (udpsink->close_socket || !udpsink->external_socket) {
1203       GError *err = NULL;
1204
1205       if (!g_socket_close (udpsink->used_socket, &err)) {
1206         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
1207         g_clear_error (&err);
1208       }
1209     }
1210
1211     g_object_unref (udpsink->used_socket);
1212     udpsink->used_socket = NULL;
1213   }
1214
1215   if (udpsink->used_socket_v6) {
1216     if (udpsink->close_socket || !udpsink->external_socket) {
1217       GError *err = NULL;
1218
1219       if (!g_socket_close (udpsink->used_socket_v6, &err)) {
1220         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
1221         g_clear_error (&err);
1222       }
1223     }
1224
1225     g_object_unref (udpsink->used_socket_v6);
1226     udpsink->used_socket_v6 = NULL;
1227   }
1228
1229   return TRUE;
1230 }
1231
1232 static void
1233 gst_multiudpsink_add_internal (GstMultiUDPSink * sink, const gchar * host,
1234     gint port, gboolean lock)
1235 {
1236   GstUDPClient *client;
1237   GstUDPClient udpclient;
1238   GTimeVal now;
1239   GList *find;
1240
1241   udpclient.host = (gchar *) host;
1242   udpclient.port = port;
1243
1244   GST_DEBUG_OBJECT (sink, "adding client on host %s, port %d", host, port);
1245
1246   if (lock)
1247     g_mutex_lock (&sink->client_lock);
1248
1249   find = g_list_find_custom (sink->clients, &udpclient,
1250       (GCompareFunc) client_compare);
1251   if (find) {
1252     client = (GstUDPClient *) find->data;
1253
1254     GST_DEBUG_OBJECT (sink, "found %d existing clients with host %s, port %d",
1255         client->refcount, host, port);
1256     client->refcount++;
1257   } else {
1258     client = create_client (sink, host, port);
1259     if (!client)
1260       goto error;
1261
1262     g_get_current_time (&now);
1263     client->connect_time = GST_TIMEVAL_TO_TIME (now);
1264
1265     if (sink->used_socket)
1266       gst_multiudpsink_configure_client (sink, client);
1267
1268     GST_DEBUG_OBJECT (sink, "add client with host %s, port %d", host, port);
1269     sink->clients = g_list_prepend (sink->clients, client);
1270   }
1271
1272   if (lock)
1273     g_mutex_unlock (&sink->client_lock);
1274
1275   g_signal_emit (G_OBJECT (sink),
1276       gst_multiudpsink_signals[SIGNAL_CLIENT_ADDED], 0, host, port);
1277
1278   GST_DEBUG_OBJECT (sink, "added client on host %s, port %d", host, port);
1279   return;
1280
1281   /* ERRORS */
1282 error:
1283   {
1284     GST_DEBUG_OBJECT (sink, "did not add client on host %s, port %d", host,
1285         port);
1286     if (lock)
1287       g_mutex_unlock (&sink->client_lock);
1288     return;
1289   }
1290 }
1291
1292 void
1293 gst_multiudpsink_add (GstMultiUDPSink * sink, const gchar * host, gint port)
1294 {
1295   gst_multiudpsink_add_internal (sink, host, port, TRUE);
1296 }
1297
1298 void
1299 gst_multiudpsink_remove (GstMultiUDPSink * sink, const gchar * host, gint port)
1300 {
1301   GList *find;
1302   GstUDPClient udpclient;
1303   GstUDPClient *client;
1304   GTimeVal now;
1305
1306   udpclient.host = (gchar *) host;
1307   udpclient.port = port;
1308
1309   g_mutex_lock (&sink->client_lock);
1310   find = g_list_find_custom (sink->clients, &udpclient,
1311       (GCompareFunc) client_compare);
1312   if (!find)
1313     goto not_found;
1314
1315   client = (GstUDPClient *) find->data;
1316
1317   GST_DEBUG_OBJECT (sink, "found %d clients with host %s, port %d",
1318       client->refcount, host, port);
1319
1320   client->refcount--;
1321   if (client->refcount == 0) {
1322     GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
1323     GInetAddress *addr = g_inet_socket_address_get_address (saddr);
1324     GSocketFamily family =
1325         g_socket_address_get_family (G_SOCKET_ADDRESS (saddr));
1326     GSocket *socket;
1327
1328     /* Select socket to send from for this address */
1329     if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
1330       socket = sink->used_socket_v6;
1331     else
1332       socket = sink->used_socket;
1333
1334     GST_DEBUG_OBJECT (sink, "remove client with host %s, port %d", host, port);
1335
1336     g_get_current_time (&now);
1337     client->disconnect_time = GST_TIMEVAL_TO_TIME (now);
1338
1339     if (socket && sink->auto_multicast
1340         && g_inet_address_get_is_multicast (addr)) {
1341       GError *err = NULL;
1342
1343       if (!g_socket_leave_multicast_group (socket, addr, FALSE,
1344               sink->multi_iface, &err)) {
1345         GST_DEBUG_OBJECT (sink, "Failed to leave multicast group: %s",
1346             err->message);
1347         g_clear_error (&err);
1348       }
1349     }
1350
1351     /* Unlock to emit signal before we delete the actual client */
1352     g_mutex_unlock (&sink->client_lock);
1353     g_signal_emit (G_OBJECT (sink),
1354         gst_multiudpsink_signals[SIGNAL_CLIENT_REMOVED], 0, host, port);
1355     g_mutex_lock (&sink->client_lock);
1356
1357     sink->clients = g_list_delete_link (sink->clients, find);
1358
1359     free_client (client);
1360   }
1361   g_mutex_unlock (&sink->client_lock);
1362
1363   return;
1364
1365   /* ERRORS */
1366 not_found:
1367   {
1368     g_mutex_unlock (&sink->client_lock);
1369     GST_WARNING_OBJECT (sink, "client at host %s, port %d not found",
1370         host, port);
1371     return;
1372   }
1373 }
1374
1375 static void
1376 gst_multiudpsink_clear_internal (GstMultiUDPSink * sink, gboolean lock)
1377 {
1378   GST_DEBUG_OBJECT (sink, "clearing");
1379   /* we only need to remove the client structure, there is no additional
1380    * socket or anything to free for UDP */
1381   if (lock)
1382     g_mutex_lock (&sink->client_lock);
1383   g_list_foreach (sink->clients, (GFunc) free_client, sink);
1384   g_list_free (sink->clients);
1385   sink->clients = NULL;
1386   if (lock)
1387     g_mutex_unlock (&sink->client_lock);
1388 }
1389
1390 void
1391 gst_multiudpsink_clear (GstMultiUDPSink * sink)
1392 {
1393   gst_multiudpsink_clear_internal (sink, TRUE);
1394 }
1395
1396 GstStructure *
1397 gst_multiudpsink_get_stats (GstMultiUDPSink * sink, const gchar * host,
1398     gint port)
1399 {
1400   GstUDPClient *client;
1401   GstStructure *result = NULL;
1402   GstUDPClient udpclient;
1403   GList *find;
1404
1405   udpclient.host = (gchar *) host;
1406   udpclient.port = port;
1407
1408   g_mutex_lock (&sink->client_lock);
1409
1410   find = g_list_find_custom (sink->clients, &udpclient,
1411       (GCompareFunc) client_compare);
1412   if (!find)
1413     goto not_found;
1414
1415   GST_DEBUG_OBJECT (sink, "stats for client with host %s, port %d", host, port);
1416
1417   client = (GstUDPClient *) find->data;
1418
1419   result = gst_structure_new_empty ("multiudpsink-stats");
1420
1421   gst_structure_set (result,
1422       "bytes-sent", G_TYPE_UINT64, client->bytes_sent,
1423       "packets-sent", G_TYPE_UINT64, client->packets_sent,
1424       "connect-time", G_TYPE_UINT64, client->connect_time,
1425       "disconnect-time", G_TYPE_UINT64, client->disconnect_time, NULL);
1426
1427   g_mutex_unlock (&sink->client_lock);
1428
1429   return result;
1430
1431   /* ERRORS */
1432 not_found:
1433   {
1434     g_mutex_unlock (&sink->client_lock);
1435     GST_WARNING_OBJECT (sink, "client with host %s, port %d not found",
1436         host, port);
1437     /* Apparently (see comment in gstmultifdsink.c) returning NULL from here may
1438      * confuse/break python bindings */
1439     return gst_structure_new_empty ("multiudpsink-stats");
1440   }
1441 }
1442
1443 static gboolean
1444 gst_multiudpsink_unlock (GstBaseSink * bsink)
1445 {
1446   GstMultiUDPSink *sink;
1447
1448   sink = GST_MULTIUDPSINK (bsink);
1449
1450   g_cancellable_cancel (sink->cancellable);
1451
1452   return TRUE;
1453 }
1454
1455 static gboolean
1456 gst_multiudpsink_unlock_stop (GstBaseSink * bsink)
1457 {
1458   GstMultiUDPSink *sink;
1459
1460   sink = GST_MULTIUDPSINK (bsink);
1461
1462   g_cancellable_reset (sink->cancellable);
1463
1464   return TRUE;
1465 }