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., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, 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 #ifdef HAVE_CONFIG_H
  33 #include "config.h"
  34 #endif
  35 #include "gstudp-marshal.h"
  36 #include "gstmultiudpsink.h"
  37
  38 #include <string.h>
  39
  40 #include "gst/glib-compat-private.h"
  41
  42 GST_DEBUG_CATEGORY_STATIC (multiudpsink_debug);
  43 #define GST_CAT_DEFAULT (multiudpsink_debug)
  44
  45 #define UDP_MAX_SIZE 65507
  46
  47 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
  48     GST_PAD_SINK,
  49     GST_PAD_ALWAYS,
  50     GST_STATIC_CAPS_ANY);
  51
  52 /* MultiUDPSink signals and args */
  53 enum
  54 {
  55   /* methods */
  56   SIGNAL_ADD,
  57   SIGNAL_REMOVE,
  58   SIGNAL_CLEAR,
  59   SIGNAL_GET_STATS,
  60
  61   /* signals */
  62   SIGNAL_CLIENT_ADDED,
  63   SIGNAL_CLIENT_REMOVED,
  64
  65   /* FILL ME */
  66   LAST_SIGNAL
  67 };
  68
  69 #define DEFAULT_SOCKET             NULL
  70 #define DEFAULT_CLOSE_SOCKET       TRUE
  71 #define DEFAULT_USED_SOCKET        NULL
  72 #define DEFAULT_CLIENTS            NULL
  73 #define DEFAULT_FAMILY             G_SOCKET_FAMILY_IPV6
  74 /* FIXME, this should be disabled by default, we don't need to join a multicast
  75  * group for sending, if this socket is also used for receiving, it should
  76  * be configured in the element that does the receive. */
  77 #define DEFAULT_AUTO_MULTICAST     TRUE
  78 #define DEFAULT_TTL                64
  79 #define DEFAULT_TTL_MC             1
  80 #define DEFAULT_LOOP               TRUE
  81 #define DEFAULT_QOS_DSCP           -1
  82 #define DEFAULT_SEND_DUPLICATES    TRUE
  83 #define DEFAULT_BUFFER_SIZE        0
  84
  85 enum
  86 {
  87   PROP_0,
  88   PROP_BYTES_TO_SERVE,
  89   PROP_BYTES_SERVED,
  90   PROP_SOCKET,
  91   PROP_CLOSE_SOCKET,
  92   PROP_USED_SOCKET,
  93   PROP_CLIENTS,
  94   PROP_AUTO_MULTICAST,
  95   PROP_TTL,
  96   PROP_TTL_MC,
  97   PROP_LOOP,
  98   PROP_QOS_DSCP,
  99   PROP_SEND_DUPLICATES,
 100   PROP_BUFFER_SIZE,
 101   PROP_LAST
 102 };
 103
 104 static void gst_multiudpsink_finalize (GObject * object);
 105
 106 static GstFlowReturn gst_multiudpsink_render (GstBaseSink * sink,
 107     GstBuffer * buffer);
 108
 109 static gboolean gst_multiudpsink_start (GstBaseSink * bsink);
 110 static gboolean gst_multiudpsink_stop (GstBaseSink * bsink);
 111 static gboolean gst_multiudpsink_unlock (GstBaseSink * bsink);
 112 static gboolean gst_multiudpsink_unlock_stop (GstBaseSink * bsink);
 113
 114 static void gst_multiudpsink_set_property (GObject * object, guint prop_id,
 115     const GValue * value, GParamSpec * pspec);
 116 static void gst_multiudpsink_get_property (GObject * object, guint prop_id,
 117     GValue * value, GParamSpec * pspec);
 118
 119 static void gst_multiudpsink_add_internal (GstMultiUDPSink * sink,
 120     const gchar * host, gint port, gboolean lock);
 121 static void gst_multiudpsink_clear_internal (GstMultiUDPSink * sink,
 122     gboolean lock);
 123
 124 static guint gst_multiudpsink_signals[LAST_SIGNAL] = { 0 };
 125
 126 #define gst_multiudpsink_parent_class parent_class
 127 G_DEFINE_TYPE (GstMultiUDPSink, gst_multiudpsink, GST_TYPE_BASE_SINK);
 128
 129 static void
 130 gst_multiudpsink_class_init (GstMultiUDPSinkClass * klass)
 131 {
 132   GObjectClass *gobject_class;
 133   GstElementClass *gstelement_class;
 134   GstBaseSinkClass *gstbasesink_class;
 135
 136   gobject_class = (GObjectClass *) klass;
 137   gstelement_class = (GstElementClass *) klass;
 138   gstbasesink_class = (GstBaseSinkClass *) klass;
 139
 140   gobject_class->set_property = gst_multiudpsink_set_property;
 141   gobject_class->get_property = gst_multiudpsink_get_property;
 142   gobject_class->finalize = gst_multiudpsink_finalize;
 143
 144   /**
 145    * GstMultiUDPSink::add:
 146    * @gstmultiudpsink: the sink on which the signal is emitted
 147    * @host: the hostname/IP address of the client to add
 148    * @port: the port of the client to add
 149    *
 150    * Add a client with destination @host and @port to the list of
 151    * clients. When the same host/port pair is added multiple times, the
 152    * send-duplicates property defines if the packets are sent multiple times to
 153    * the same host/port pair or not.
 154    *
 155    * When a host/port pair is added multiple times, an equal amount of remove
 156    * calls must be performed to actually remove the host/port pair from the list
 157    * of destinations.
 158    */
 159   gst_multiudpsink_signals[SIGNAL_ADD] =
 160       g_signal_new ("add", G_TYPE_FROM_CLASS (klass),
 161       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 162       G_STRUCT_OFFSET (GstMultiUDPSinkClass, add),
 163       NULL, NULL, gst_udp_marshal_VOID__STRING_INT, G_TYPE_NONE, 2,
 164       G_TYPE_STRING, G_TYPE_INT);
 165   /**
 166    * GstMultiUDPSink::remove:
 167    * @gstmultiudpsink: the sink on which the signal is emitted
 168    * @host: the hostname/IP address of the client to remove
 169    * @port: the port of the client to remove
 170    *
 171    * Remove the client with destination @host and @port from the list of
 172    * clients.
 173    */
 174   gst_multiudpsink_signals[SIGNAL_REMOVE] =
 175       g_signal_new ("remove", G_TYPE_FROM_CLASS (klass),
 176       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 177       G_STRUCT_OFFSET (GstMultiUDPSinkClass, remove),
 178       NULL, NULL, gst_udp_marshal_VOID__STRING_INT, G_TYPE_NONE, 2,
 179       G_TYPE_STRING, G_TYPE_INT);
 180   /**
 181    * GstMultiUDPSink::clear:
 182    * @gstmultiudpsink: the sink on which the signal is emitted
 183    *
 184    * Clear the list of clients.
 185    */
 186   gst_multiudpsink_signals[SIGNAL_CLEAR] =
 187       g_signal_new ("clear", G_TYPE_FROM_CLASS (klass),
 188       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 189       G_STRUCT_OFFSET (GstMultiUDPSinkClass, clear),
 190       NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 191   /**
 192    * GstMultiUDPSink::get-stats:
 193    * @gstmultiudpsink: the sink on which the signal is emitted
 194    * @host: the hostname/IP address of the client to get stats on
 195    * @port: the port of the client to get stats on
 196    *
 197    * Get the statistics of the client with destination @host and @port.
 198    *
 199    * Returns: a GstStructure: bytes_sent, packets_sent,
 200    *           connect_time (in epoch seconds), disconnect_time (in epoch seconds)
 201    */
 202   gst_multiudpsink_signals[SIGNAL_GET_STATS] =
 203       g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
 204       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 205       G_STRUCT_OFFSET (GstMultiUDPSinkClass, get_stats),
 206       NULL, NULL, gst_udp_marshal_BOXED__STRING_INT, GST_TYPE_STRUCTURE, 2,
 207       G_TYPE_STRING, G_TYPE_INT);
 208   /**
 209    * GstMultiUDPSink::client-added:
 210    * @gstmultiudpsink: the sink emitting the signal
 211    * @host: the hostname/IP address of the added client
 212    * @port: the port of the added client
 213    *
 214    * Signal emited when a new client is added to the list of
 215    * clients.
 216    */
 217   gst_multiudpsink_signals[SIGNAL_CLIENT_ADDED] =
 218       g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
 219       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiUDPSinkClass, client_added),
 220       NULL, NULL, gst_udp_marshal_VOID__STRING_INT, G_TYPE_NONE, 2,
 221       G_TYPE_STRING, G_TYPE_INT);
 222   /**
 223    * GstMultiUDPSink::client-removed:
 224    * @gstmultiudpsink: the sink emitting the signal
 225    * @host: the hostname/IP address of the removed client
 226    * @port: the port of the removed client
 227    *
 228    * Signal emited when a client is removed from the list of
 229    * clients.
 230    */
 231   gst_multiudpsink_signals[SIGNAL_CLIENT_REMOVED] =
 232       g_signal_new ("client-removed", G_TYPE_FROM_CLASS (klass),
 233       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiUDPSinkClass,
 234           client_removed), NULL, NULL, gst_udp_marshal_VOID__STRING_INT,
 235       G_TYPE_NONE, 2, G_TYPE_STRING, G_TYPE_INT);
 236
 237   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BYTES_TO_SERVE,
 238       g_param_spec_uint64 ("bytes-to-serve", "Bytes to serve",
 239           "Number of bytes received to serve to clients", 0, G_MAXUINT64, 0,
 240           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 241   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BYTES_SERVED,
 242       g_param_spec_uint64 ("bytes-served", "Bytes served",
 243           "Total number of bytes sent to all clients", 0, G_MAXUINT64, 0,
 244           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 245   g_object_class_install_property (gobject_class, PROP_SOCKET,
 246       g_param_spec_object ("socket", "Socket Handle",
 247           "Socket to use for UDP sending. (NULL == allocate)",
 248           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 249   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 250       g_param_spec_boolean ("close-socket", "Close socket",
 251           "Close socket if passed as property on state change",
 252           DEFAULT_CLOSE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 253   g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
 254       g_param_spec_object ("used-socket", "Used Socket Handle",
 255           "Socket currently in use for UDP sending. (NULL == no socket)",
 256           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 257   g_object_class_install_property (gobject_class, PROP_CLIENTS,
 258       g_param_spec_string ("clients", "Clients",
 259           "A comma separated list of host:port pairs with destinations",
 260           DEFAULT_CLIENTS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 261   g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
 262       g_param_spec_boolean ("auto-multicast",
 263           "Automatically join/leave multicast groups",
 264           "Automatically join/leave the multicast groups, FALSE means user"
 265           " has to do it himself", DEFAULT_AUTO_MULTICAST,
 266           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 267   g_object_class_install_property (gobject_class, PROP_TTL,
 268       g_param_spec_int ("ttl", "Unicast TTL",
 269           "Used for setting the unicast TTL parameter",
 270           0, 255, DEFAULT_TTL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 271   g_object_class_install_property (gobject_class, PROP_TTL_MC,
 272       g_param_spec_int ("ttl-mc", "Multicast TTL",
 273           "Used for setting the multicast TTL parameter",
 274           0, 255, DEFAULT_TTL_MC, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 275   g_object_class_install_property (gobject_class, PROP_LOOP,
 276       g_param_spec_boolean ("loop", "Multicast Loopback",
 277           "Used for setting the multicast loop parameter. TRUE = enable,"
 278           " FALSE = disable", DEFAULT_LOOP,
 279           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 280   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_QOS_DSCP,
 281       g_param_spec_int ("qos-dscp", "QoS diff srv code point",
 282           "Quality of Service, differentiated services code point (-1 default)",
 283           -1, 63, DEFAULT_QOS_DSCP,
 284           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 285   /**
 286    * GstMultiUDPSink::send-duplicates
 287    *
 288    * When a host/port pair is added mutliple times, send the packet to the host
 289    * multiple times as well.
 290    *
 291    * Since: 0.10.26
 292    */
 293   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SEND_DUPLICATES,
 294       g_param_spec_boolean ("send-duplicates", "Send Duplicates",
 295           "When a distination/port pair is added multiple times, send packets "
 296           "multiple times as well", DEFAULT_SEND_DUPLICATES,
 297           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 298
 299   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
 300       g_param_spec_int ("buffer-size", "Buffer Size",
 301           "Size of the kernel send buffer in bytes, 0=default", 0, G_MAXINT,
 302           DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 303
 304   gst_element_class_add_pad_template (gstelement_class,
 305       gst_static_pad_template_get (&sink_template));
 306
 307   gst_element_class_set_details_simple (gstelement_class, "UDP packet sender",
 308       "Sink/Network",
 309       "Send data over the network via UDP",
 310       "Wim Taymans <wim.taymans@gmail.com>");
 311
 312   gstbasesink_class->render = gst_multiudpsink_render;
 313   gstbasesink_class->start = gst_multiudpsink_start;
 314   gstbasesink_class->stop = gst_multiudpsink_stop;
 315   gstbasesink_class->unlock = gst_multiudpsink_unlock;
 316   gstbasesink_class->unlock_stop = gst_multiudpsink_unlock_stop;
 317   klass->add = gst_multiudpsink_add;
 318   klass->remove = gst_multiudpsink_remove;
 319   klass->clear = gst_multiudpsink_clear;
 320   klass->get_stats = gst_multiudpsink_get_stats;
 321
 322   GST_DEBUG_CATEGORY_INIT (multiudpsink_debug, "multiudpsink", 0, "UDP sink");
 323 }
 324
 325
 326 static void
 327 gst_multiudpsink_init (GstMultiUDPSink * sink)
 328 {
 329   g_mutex_init (&sink->client_lock);
 330   sink->socket = DEFAULT_SOCKET;
 331   sink->used_socket = DEFAULT_USED_SOCKET;
 332   sink->close_socket = DEFAULT_CLOSE_SOCKET;
 333   sink->external_socket = (sink->socket != NULL);
 334   sink->auto_multicast = DEFAULT_AUTO_MULTICAST;
 335   sink->ttl = DEFAULT_TTL;
 336   sink->ttl_mc = DEFAULT_TTL_MC;
 337   sink->loop = DEFAULT_LOOP;
 338   sink->qos_dscp = DEFAULT_QOS_DSCP;
 339   sink->family = DEFAULT_FAMILY;
 340   sink->send_duplicates = DEFAULT_SEND_DUPLICATES;
 341
 342   sink->cancellable = g_cancellable_new ();
 343 }
 344
 345 static GstUDPClient *
 346 create_client (GstMultiUDPSink * sink, const gchar * host, gint port)
 347 {
 348   GstUDPClient *client;
 349   GInetAddress *addr;
 350   GResolver *resolver;
 351   GError *err = NULL;
 352
 353   addr = g_inet_address_new_from_string (host);
 354   if (!addr) {
 355     GList *results;
 356
 357     resolver = g_resolver_get_default ();
 358     results =
 359         g_resolver_lookup_by_name (resolver, host, sink->cancellable, &err);
 360     if (!results)
 361       goto name_resolve;
 362     addr = G_INET_ADDRESS (g_object_ref (results->data));
 363
 364     g_resolver_free_addresses (results);
 365     g_object_unref (resolver);
 366   }
 367 #ifndef GST_DISABLE_GST_DEBUG
 368   {
 369     gchar *ip = g_inet_address_to_string (addr);
 370
 371     GST_DEBUG_OBJECT (sink, "IP address for host %s is %s", host, ip);
 372     g_free (ip);
 373   }
 374 #endif
 375
 376   client = g_slice_new0 (GstUDPClient);
 377   client->refcount = 1;
 378   client->host = g_strdup (host);
 379   client->port = port;
 380   client->addr = g_inet_socket_address_new (addr, port);
 381   g_object_unref (addr);
 382
 383   return client;
 384
 385 name_resolve:
 386   {
 387     g_object_unref (resolver);
 388
 389     return NULL;
 390   }
 391 }
 392
 393 static void
 394 free_client (GstUDPClient * client)
 395 {
 396   g_object_unref (client->addr);
 397   g_free (client->host);
 398   g_slice_free (GstUDPClient, client);
 399 }
 400
 401 static gint
 402 client_compare (GstUDPClient * a, GstUDPClient * b)
 403 {
 404   if ((a->port == b->port) && (strcmp (a->host, b->host) == 0))
 405     return 0;
 406
 407   return 1;
 408 }
 409
 410 static void
 411 gst_multiudpsink_finalize (GObject * object)
 412 {
 413   GstMultiUDPSink *sink;
 414
 415   sink = GST_MULTIUDPSINK (object);
 416
 417   g_list_foreach (sink->clients, (GFunc) free_client, NULL);
 418   g_list_free (sink->clients);
 419
 420   if (sink->socket)
 421     g_object_unref (sink->socket);
 422   sink->socket = NULL;
 423
 424   if (sink->used_socket)
 425     g_object_unref (sink->used_socket);
 426   sink->used_socket = NULL;
 427
 428   if (sink->cancellable)
 429     g_object_unref (sink->cancellable);
 430   sink->cancellable = NULL;
 431
 432   g_mutex_clear (&sink->client_lock);
 433
 434   G_OBJECT_CLASS (parent_class)->finalize (object);
 435 }
 436
 437 static GstFlowReturn
 438 gst_multiudpsink_render (GstBaseSink * bsink, GstBuffer * buffer)
 439 {
 440   GstMultiUDPSink *sink;
 441   GList *clients;
 442   GOutputVector *vec;
 443   GstMapInfo *map;
 444   guint n_mem, i;
 445   gsize size;
 446   GstMemory *mem;
 447   gint num, no_clients;
 448   GError *err = NULL;
 449
 450   sink = GST_MULTIUDPSINK (bsink);
 451
 452   n_mem = gst_buffer_n_memory (buffer);
 453   if (n_mem == 0)
 454     goto no_data;
 455
 456   vec = g_new (GOutputVector, n_mem);
 457   map = g_new (GstMapInfo, n_mem);
 458
 459   size = 0;
 460   for (i = 0; i < n_mem; i++) {
 461     mem = gst_buffer_get_memory (buffer, i);
 462     gst_memory_map (mem, &map[i], GST_MAP_READ);
 463
 464     if (map[i].size > UDP_MAX_SIZE) {
 465       GST_WARNING ("Attempting to send a UDP packet larger than maximum "
 466           "size (%" G_GSIZE_FORMAT " > %d)", map[i].size, UDP_MAX_SIZE);
 467     }
 468
 469     vec[i].buffer = map[i].data;
 470     vec[i].size = map[i].size;
 471
 472     size += map[i].size;
 473   }
 474
 475   sink->bytes_to_serve += size;
 476
 477   /* grab lock while iterating and sending to clients, this should be
 478    * fast as UDP never blocks */
 479   g_mutex_lock (&sink->client_lock);
 480   GST_LOG_OBJECT (bsink, "about to send %d bytes", size);
 481
 482   no_clients = 0;
 483   num = 0;
 484   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
 485     GstUDPClient *client;
 486     gint count;
 487
 488     client = (GstUDPClient *) clients->data;
 489     no_clients++;
 490     GST_LOG_OBJECT (sink, "sending %d bytes to client %p", size, client);
 491
 492     count = sink->send_duplicates ? client->refcount : 1;
 493
 494     while (count--) {
 495       gssize ret;
 496
 497       ret =
 498           g_socket_send_message (sink->used_socket, client->addr, vec, n_mem,
 499           NULL, 0, 0, sink->cancellable, &err);
 500
 501       if (ret < 0)
 502         goto send_error;
 503
 504       num++;
 505       client->bytes_sent += ret;
 506       client->packets_sent++;
 507       sink->bytes_served += ret;
 508     }
 509   }
 510   g_mutex_unlock (&sink->client_lock);
 511
 512   /* unmap all memory again */
 513   for (i = 0; i < n_mem; i++) {
 514     gst_memory_unmap (map[i].memory, &map[i]);
 515     gst_memory_unref (map[i].memory);
 516   }
 517
 518   g_free (vec);
 519   g_free (map);
 520
 521   GST_LOG_OBJECT (sink, "sent %d bytes to %d (of %d) clients", size, num,
 522       no_clients);
 523
 524   return GST_FLOW_OK;
 525
 526 no_data:
 527   {
 528     return GST_FLOW_OK;
 529   }
 530 send_error:
 531   {
 532     GST_DEBUG ("got send error %s", err->message);
 533     g_clear_error (&err);
 534     return GST_FLOW_ERROR;
 535   }
 536 }
 537
 538 static void
 539 gst_multiudpsink_set_clients_string (GstMultiUDPSink * sink,
 540     const gchar * string)
 541 {
 542   gchar **clients;
 543   gint i;
 544
 545   clients = g_strsplit (string, ",", 0);
 546
 547   g_mutex_lock (&sink->client_lock);
 548   /* clear all existing clients */
 549   gst_multiudpsink_clear_internal (sink, FALSE);
 550   for (i = 0; clients[i]; i++) {
 551     gchar *host, *p;
 552     gint64 port = 0;
 553
 554     host = clients[i];
 555     p = strstr (clients[i], ":");
 556     if (p != NULL) {
 557       *p = '\0';
 558       port = g_ascii_strtoll (p + 1, NULL, 10);
 559     }
 560     if (port != 0)
 561       gst_multiudpsink_add_internal (sink, host, port, FALSE);
 562   }
 563   g_mutex_unlock (&sink->client_lock);
 564
 565   g_strfreev (clients);
 566 }
 567
 568 static gchar *
 569 gst_multiudpsink_get_clients_string (GstMultiUDPSink * sink)
 570 {
 571   GString *str;
 572   GList *clients;
 573
 574   str = g_string_new ("");
 575
 576   g_mutex_lock (&sink->client_lock);
 577   clients = sink->clients;
 578   while (clients) {
 579     GstUDPClient *client;
 580     gint count;
 581
 582     client = (GstUDPClient *) clients->data;
 583
 584     clients = g_list_next (clients);
 585
 586     count = client->refcount;
 587     while (count--) {
 588       g_string_append_printf (str, "%s:%d%s", client->host, client->port,
 589           (clients || count > 1 ? "," : ""));
 590     }
 591   }
 592   g_mutex_unlock (&sink->client_lock);
 593
 594   return g_string_free (str, FALSE);
 595 }
 596
 597 static void
 598 gst_multiudpsink_setup_qos_dscp (GstMultiUDPSink * sink)
 599 {
 600   /* don't touch on -1 */
 601   if (sink->qos_dscp < 0)
 602     return;
 603
 604   if (sink->used_socket == NULL)
 605     return;
 606
 607 #ifdef IP_TOS
 608   {
 609     gint tos;
 610     gint fd;
 611
 612     fd = g_socket_get_fd (sink->used_socket);
 613
 614     GST_DEBUG_OBJECT (sink, "setting TOS to %d", sink->qos_dscp);
 615
 616     /* Extract and shift 6 bits of DSFIELD */
 617     tos = (sink->qos_dscp & 0x3f) << 2;
 618
 619     if (setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos)) < 0) {
 620       GST_ERROR_OBJECT (sink, "could not set TOS: %s", g_strerror (errno));
 621     }
 622 #ifdef IPV6_TCLASS
 623     if (setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos)) < 0) {
 624       GST_ERROR_OBJECT (sink, "could not set TCLASS: %s", g_strerror (errno));
 625     }
 626   }
 627 #endif
 628 #endif
 629 }
 630
 631 static void
 632 gst_multiudpsink_set_property (GObject * object, guint prop_id,
 633     const GValue * value, GParamSpec * pspec)
 634 {
 635   GstMultiUDPSink *udpsink;
 636
 637   udpsink = GST_MULTIUDPSINK (object);
 638
 639   switch (prop_id) {
 640     case PROP_SOCKET:
 641       if (udpsink->socket != NULL && udpsink->socket != udpsink->used_socket &&
 642           udpsink->close_socket) {
 643         GError *err = NULL;
 644
 645         if (!g_socket_close (udpsink->socket, &err)) {
 646           GST_ERROR ("failed to close socket %p: %s", udpsink->socket,
 647               err->message);
 648           g_clear_error (&err);
 649         }
 650       }
 651       if (udpsink->socket)
 652         g_object_unref (udpsink->socket);
 653       udpsink->socket = g_value_dup_object (value);
 654       GST_DEBUG_OBJECT (udpsink, "setting socket to %p", udpsink->socket);
 655       break;
 656     case PROP_CLOSE_SOCKET:
 657       udpsink->close_socket = g_value_get_boolean (value);
 658       break;
 659     case PROP_CLIENTS:
 660       gst_multiudpsink_set_clients_string (udpsink, g_value_get_string (value));
 661       break;
 662     case PROP_AUTO_MULTICAST:
 663       udpsink->auto_multicast = g_value_get_boolean (value);
 664       break;
 665     case PROP_TTL:
 666       udpsink->ttl = g_value_get_int (value);
 667       break;
 668     case PROP_TTL_MC:
 669       udpsink->ttl_mc = g_value_get_int (value);
 670       break;
 671     case PROP_LOOP:
 672       udpsink->loop = g_value_get_boolean (value);
 673       break;
 674     case PROP_QOS_DSCP:
 675       udpsink->qos_dscp = g_value_get_int (value);
 676       gst_multiudpsink_setup_qos_dscp (udpsink);
 677       break;
 678     case PROP_SEND_DUPLICATES:
 679       udpsink->send_duplicates = g_value_get_boolean (value);
 680       break;
 681     case PROP_BUFFER_SIZE:
 682       udpsink->buffer_size = g_value_get_int (value);
 683       break;
 684     default:
 685       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 686       break;
 687   }
 688 }
 689
 690 static void
 691 gst_multiudpsink_get_property (GObject * object, guint prop_id, GValue * value,
 692     GParamSpec * pspec)
 693 {
 694   GstMultiUDPSink *udpsink;
 695
 696   udpsink = GST_MULTIUDPSINK (object);
 697
 698   switch (prop_id) {
 699     case PROP_BYTES_TO_SERVE:
 700       g_value_set_uint64 (value, udpsink->bytes_to_serve);
 701       break;
 702     case PROP_BYTES_SERVED:
 703       g_value_set_uint64 (value, udpsink->bytes_served);
 704       break;
 705     case PROP_SOCKET:
 706       g_value_set_object (value, udpsink->socket);
 707       break;
 708     case PROP_CLOSE_SOCKET:
 709       g_value_set_boolean (value, udpsink->close_socket);
 710       break;
 711     case PROP_USED_SOCKET:
 712       g_value_set_object (value, udpsink->used_socket);
 713       break;
 714     case PROP_CLIENTS:
 715       g_value_take_string (value,
 716           gst_multiudpsink_get_clients_string (udpsink));
 717       break;
 718     case PROP_AUTO_MULTICAST:
 719       g_value_set_boolean (value, udpsink->auto_multicast);
 720       break;
 721     case PROP_TTL:
 722       g_value_set_int (value, udpsink->ttl);
 723       break;
 724     case PROP_TTL_MC:
 725       g_value_set_int (value, udpsink->ttl_mc);
 726       break;
 727     case PROP_LOOP:
 728       g_value_set_boolean (value, udpsink->loop);
 729       break;
 730     case PROP_QOS_DSCP:
 731       g_value_set_int (value, udpsink->qos_dscp);
 732       break;
 733     case PROP_SEND_DUPLICATES:
 734       g_value_set_boolean (value, udpsink->send_duplicates);
 735       break;
 736     case PROP_BUFFER_SIZE:
 737       g_value_set_int (value, udpsink->buffer_size);
 738       break;
 739     default:
 740       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 741       break;
 742   }
 743 }
 744
 745 static gboolean
 746 gst_multiudpsink_configure_client (GstMultiUDPSink * sink,
 747     GstUDPClient * client)
 748 {
 749   GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
 750   GInetAddress *addr = g_inet_socket_address_get_address (saddr);
 751   GError *err = NULL;
 752
 753   GST_DEBUG_OBJECT (sink, "configuring client %p", client);
 754
 755   if (g_inet_address_get_is_multicast (addr)) {
 756     GST_DEBUG_OBJECT (sink, "we have a multicast client %p", client);
 757     if (sink->auto_multicast) {
 758       GST_DEBUG_OBJECT (sink, "autojoining group");
 759       if (!g_socket_join_multicast_group (sink->used_socket, addr, FALSE, NULL,
 760               &err))
 761         goto join_group_failed;
 762     }
 763     GST_DEBUG_OBJECT (sink, "setting loop to %d", sink->loop);
 764     g_socket_set_multicast_loopback (sink->used_socket, sink->loop);
 765     GST_DEBUG_OBJECT (sink, "setting ttl to %d", sink->ttl_mc);
 766     g_socket_set_multicast_ttl (sink->used_socket, sink->ttl_mc);
 767   } else {
 768     GST_DEBUG_OBJECT (sink, "setting unicast ttl to %d", sink->ttl);
 769     g_socket_set_ttl (sink->used_socket, sink->ttl);
 770   }
 771   return TRUE;
 772
 773   /* ERRORS */
 774 join_group_failed:
 775   {
 776     gst_multiudpsink_stop (GST_BASE_SINK (sink));
 777     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
 778         ("Could not join multicast group: %s", err->message));
 779     g_clear_error (&err);
 780     return FALSE;
 781   }
 782 }
 783
 784 /* create a socket for sending to remote machine */
 785 static gboolean
 786 gst_multiudpsink_start (GstBaseSink * bsink)
 787 {
 788   GstMultiUDPSink *sink;
 789   GList *clients;
 790   GstUDPClient *client;
 791   GError *err = NULL;
 792
 793   sink = GST_MULTIUDPSINK (bsink);
 794
 795   if (sink->socket == NULL) {
 796     GST_DEBUG_OBJECT (sink, "creating sockets");
 797     /* create sender socket try IP6, fall back to IP4 */
 798     sink->family = G_SOCKET_FAMILY_IPV6;
 799     if ((sink->used_socket =
 800             g_socket_new (G_SOCKET_FAMILY_IPV6,
 801                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
 802       sink->family = G_SOCKET_FAMILY_IPV4;
 803       if ((sink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
 804                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 805         goto no_socket;
 806     }
 807
 808     GST_DEBUG_OBJECT (sink, "have socket");
 809     sink->external_socket = FALSE;
 810   } else {
 811     GST_DEBUG_OBJECT (sink, "using configured socket");
 812     /* we use the configured socket */
 813     sink->used_socket = G_SOCKET (g_object_ref (sink->socket));
 814     sink->family = g_socket_get_family (sink->used_socket);
 815     sink->external_socket = TRUE;
 816   }
 817
 818 #ifdef SO_SNDBUF
 819
 820   {
 821     socklen_t len;
 822     gint sndsize, ret;
 823
 824     len = sizeof (sndsize);
 825     if (sink->buffer_size != 0) {
 826       sndsize = sink->buffer_size;
 827
 828       GST_DEBUG_OBJECT (sink, "setting udp buffer of %d bytes", sndsize);
 829       /* set buffer size, Note that on Linux this is typically limited to a
 830        * maximum of around 100K. Also a minimum of 128 bytes is required on
 831        * Linux. */
 832       ret =
 833           setsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
 834           SO_SNDBUF, (void *) &sndsize, len);
 835       if (ret != 0) {
 836         GST_ELEMENT_WARNING (sink, RESOURCE, SETTINGS, (NULL),
 837             ("Could not create a buffer of requested %d bytes, %d: %s",
 838                 sndsize, ret, g_strerror (errno)));
 839       }
 840     }
 841
 842     /* read the value of the receive buffer. Note that on linux this returns 2x the
 843      * value we set because the kernel allocates extra memory for metadata.
 844      * The default on Linux is about 100K (which is about 50K without metadata) */
 845     ret =
 846         getsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET, SO_SNDBUF,
 847         (void *) &sndsize, &len);
 848     if (ret == 0)
 849       GST_DEBUG_OBJECT (sink, "have udp buffer of %d bytes", sndsize);
 850     else
 851       GST_DEBUG_OBJECT (sink, "could not get udp buffer size");
 852   }
 853 #endif
 854
 855   g_socket_set_broadcast (sink->used_socket, TRUE);
 856
 857   sink->bytes_to_serve = 0;
 858   sink->bytes_served = 0;
 859
 860   gst_multiudpsink_setup_qos_dscp (sink);
 861
 862   /* look for multicast clients and join multicast groups appropriately
 863      set also ttl and multicast loopback delivery appropriately  */
 864   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
 865     client = (GstUDPClient *) clients->data;
 866
 867     if (!gst_multiudpsink_configure_client (sink, client))
 868       return FALSE;
 869   }
 870   return TRUE;
 871
 872   /* ERRORS */
 873 no_socket:
 874   {
 875     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
 876         ("Could not create socket: %s", err->message));
 877     g_clear_error (&err);
 878     return FALSE;
 879   }
 880 }
 881
 882 static gboolean
 883 gst_multiudpsink_stop (GstBaseSink * bsink)
 884 {
 885   GstMultiUDPSink *udpsink;
 886
 887   udpsink = GST_MULTIUDPSINK (bsink);
 888
 889   if (udpsink->used_socket) {
 890     if (udpsink->close_socket || !udpsink->external_socket) {
 891       GError *err = NULL;
 892
 893       if (!g_socket_close (udpsink->used_socket, &err)) {
 894         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
 895         g_clear_error (&err);
 896       }
 897     }
 898
 899     g_object_unref (udpsink->used_socket);
 900     udpsink->used_socket = NULL;
 901   }
 902
 903   return TRUE;
 904 }
 905
 906 static void
 907 gst_multiudpsink_add_internal (GstMultiUDPSink * sink, const gchar * host,
 908     gint port, gboolean lock)
 909 {
 910   GstUDPClient *client;
 911   GstUDPClient udpclient;
 912   GTimeVal now;
 913   GList *find;
 914
 915   udpclient.host = (gchar *) host;
 916   udpclient.port = port;
 917
 918   GST_DEBUG_OBJECT (sink, "adding client on host %s, port %d", host, port);
 919
 920   if (lock)
 921     g_mutex_lock (&sink->client_lock);
 922
 923   find = g_list_find_custom (sink->clients, &udpclient,
 924       (GCompareFunc) client_compare);
 925   if (find) {
 926     client = (GstUDPClient *) find->data;
 927
 928     GST_DEBUG_OBJECT (sink, "found %d existing clients with host %s, port %d",
 929         client->refcount, host, port);
 930     client->refcount++;
 931   } else {
 932     client = create_client (sink, host, port);
 933     if (!client)
 934       goto error;
 935
 936     g_get_current_time (&now);
 937     client->connect_time = GST_TIMEVAL_TO_TIME (now);
 938
 939     if (sink->used_socket)
 940       gst_multiudpsink_configure_client (sink, client);
 941
 942     GST_DEBUG_OBJECT (sink, "add client with host %s, port %d", host, port);
 943     sink->clients = g_list_prepend (sink->clients, client);
 944   }
 945
 946   if (lock)
 947     g_mutex_unlock (&sink->client_lock);
 948
 949   g_signal_emit (G_OBJECT (sink),
 950       gst_multiudpsink_signals[SIGNAL_CLIENT_ADDED], 0, host, port);
 951
 952   GST_DEBUG_OBJECT (sink, "added client on host %s, port %d", host, port);
 953   return;
 954
 955   /* ERRORS */
 956 error:
 957   {
 958     GST_DEBUG_OBJECT (sink, "did not add client on host %s, port %d", host,
 959         port);
 960     if (lock)
 961       g_mutex_unlock (&sink->client_lock);
 962     return;
 963   }
 964 }
 965
 966 void
 967 gst_multiudpsink_add (GstMultiUDPSink * sink, const gchar * host, gint port)
 968 {
 969   gst_multiudpsink_add_internal (sink, host, port, TRUE);
 970 }
 971
 972 void
 973 gst_multiudpsink_remove (GstMultiUDPSink * sink, const gchar * host, gint port)
 974 {
 975   GList *find;
 976   GstUDPClient udpclient;
 977   GstUDPClient *client;
 978   GTimeVal now;
 979
 980   udpclient.host = (gchar *) host;
 981   udpclient.port = port;
 982
 983   g_mutex_lock (&sink->client_lock);
 984   find = g_list_find_custom (sink->clients, &udpclient,
 985       (GCompareFunc) client_compare);
 986   if (!find)
 987     goto not_found;
 988
 989   client = (GstUDPClient *) find->data;
 990
 991   GST_DEBUG_OBJECT (sink, "found %d clients with host %s, port %d",
 992       client->refcount, host, port);
 993
 994   client->refcount--;
 995   if (client->refcount == 0) {
 996     GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
 997     GInetAddress *addr = g_inet_socket_address_get_address (saddr);
 998
 999     GST_DEBUG_OBJECT (sink, "remove client with host %s, port %d", host, port);
1000
1001     g_get_current_time (&now);
1002     client->disconnect_time = GST_TIMEVAL_TO_TIME (now);
1003
1004     if (sink->used_socket && sink->auto_multicast
1005         && g_inet_address_get_is_multicast (addr)) {
1006       GError *err = NULL;
1007
1008       if (!g_socket_leave_multicast_group (sink->used_socket, addr, FALSE, NULL,
1009               &err)) {
1010         GST_DEBUG_OBJECT (sink, "Failed to leave multicast group: %s",
1011             err->message);
1012         g_clear_error (&err);
1013       }
1014     }
1015
1016     /* Unlock to emit signal before we delete the actual client */
1017     g_mutex_unlock (&sink->client_lock);
1018     g_signal_emit (G_OBJECT (sink),
1019         gst_multiudpsink_signals[SIGNAL_CLIENT_REMOVED], 0, host, port);
1020     g_mutex_lock (&sink->client_lock);
1021
1022     sink->clients = g_list_delete_link (sink->clients, find);
1023
1024     free_client (client);
1025   }
1026   g_mutex_unlock (&sink->client_lock);
1027
1028   return;
1029
1030   /* ERRORS */
1031 not_found:
1032   {
1033     g_mutex_unlock (&sink->client_lock);
1034     GST_WARNING_OBJECT (sink, "client at host %s, port %d not found",
1035         host, port);
1036     return;
1037   }
1038 }
1039
1040 static void
1041 gst_multiudpsink_clear_internal (GstMultiUDPSink * sink, gboolean lock)
1042 {
1043   GST_DEBUG_OBJECT (sink, "clearing");
1044   /* we only need to remove the client structure, there is no additional
1045    * socket or anything to free for UDP */
1046   if (lock)
1047     g_mutex_lock (&sink->client_lock);
1048   g_list_foreach (sink->clients, (GFunc) free_client, sink);
1049   g_list_free (sink->clients);
1050   sink->clients = NULL;
1051   if (lock)
1052     g_mutex_unlock (&sink->client_lock);
1053 }
1054
1055 void
1056 gst_multiudpsink_clear (GstMultiUDPSink * sink)
1057 {
1058   gst_multiudpsink_clear_internal (sink, TRUE);
1059 }
1060
1061 GstStructure *
1062 gst_multiudpsink_get_stats (GstMultiUDPSink * sink, const gchar * host,
1063     gint port)
1064 {
1065   GstUDPClient *client;
1066   GstStructure *result = NULL;
1067   GstUDPClient udpclient;
1068   GList *find;
1069
1070   udpclient.host = (gchar *) host;
1071   udpclient.port = port;
1072
1073   g_mutex_lock (&sink->client_lock);
1074
1075   find = g_list_find_custom (sink->clients, &udpclient,
1076       (GCompareFunc) client_compare);
1077   if (!find)
1078     goto not_found;
1079
1080   GST_DEBUG_OBJECT (sink, "stats for client with host %s, port %d", host, port);
1081
1082   client = (GstUDPClient *) find->data;
1083
1084   result = gst_structure_new_empty ("multiudpsink-stats");
1085
1086   gst_structure_set (result,
1087       "bytes-sent", G_TYPE_UINT64, client->bytes_sent,
1088       "packets-sent", G_TYPE_UINT64, client->packets_sent,
1089       "connect-time", G_TYPE_UINT64, client->connect_time,
1090       "disconnect-time", G_TYPE_UINT64, client->disconnect_time, NULL);
1091
1092   g_mutex_unlock (&sink->client_lock);
1093
1094   return result;
1095
1096   /* ERRORS */
1097 not_found:
1098   {
1099     g_mutex_unlock (&sink->client_lock);
1100     GST_WARNING_OBJECT (sink, "client with host %s, port %d not found",
1101         host, port);
1102     /* Apparently (see comment in gstmultifdsink.c) returning NULL from here may
1103      * confuse/break python bindings */
1104     return gst_structure_new_empty ("multiudpsink-stats");
1105   }
1106 }
1107
1108 static gboolean
1109 gst_multiudpsink_unlock (GstBaseSink * bsink)
1110 {
1111   GstMultiUDPSink *sink;
1112
1113   sink = GST_MULTIUDPSINK (bsink);
1114
1115   g_cancellable_cancel (sink->cancellable);
1116
1117   return TRUE;
1118 }
1119
1120 static gboolean
1121 gst_multiudpsink_unlock_stop (GstBaseSink * bsink)
1122 {
1123   GstMultiUDPSink *sink;
1124
1125   sink = GST_MULTIUDPSINK (bsink);
1126
1127   g_cancellable_reset (sink->cancellable);
1128
1129   return TRUE;
1130 }