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, G_TYPE_VALUE_ARRAY, 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 send 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   guint n_mem, i;
 444   gpointer bdata;
 445   gsize bsize, 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
 458   size = 0;
 459   for (i = 0; i < n_mem; i++) {
 460     mem = gst_buffer_peek_memory (buffer, i, GST_MAP_READ);
 461     bdata = gst_memory_map (mem, &bsize, NULL, GST_MAP_READ);
 462
 463     if (bsize > UDP_MAX_SIZE) {
 464       GST_WARNING ("Attempting to send a UDP packet larger than maximum "
 465           "size (%" G_GSIZE_FORMAT " > %d)", bsize, UDP_MAX_SIZE);
 466     }
 467
 468     vec[i].buffer = bdata;
 469     vec[i].size = bsize;
 470
 471     size += bsize;
 472   }
 473
 474   sink->bytes_to_serve += size;
 475
 476   /* grab lock while iterating and sending to clients, this should be
 477    * fast as UDP never blocks */
 478   g_mutex_lock (&sink->client_lock);
 479   GST_LOG_OBJECT (bsink, "about to send %d bytes", size);
 480
 481   no_clients = 0;
 482   num = 0;
 483   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
 484     GstUDPClient *client;
 485     gint count;
 486
 487     client = (GstUDPClient *) clients->data;
 488     no_clients++;
 489     GST_LOG_OBJECT (sink, "sending %d bytes to client %p", size, client);
 490
 491     count = sink->send_duplicates ? client->refcount : 1;
 492
 493     while (count--) {
 494       gssize ret;
 495
 496       ret =
 497           g_socket_send_message (sink->used_socket, client->addr, vec, n_mem,
 498           NULL, 0, 0, sink->cancellable, &err);
 499
 500       if (ret < 0)
 501         goto send_error;
 502
 503       num++;
 504       client->bytes_sent += ret;
 505       client->packets_sent++;
 506       sink->bytes_served += ret;
 507     }
 508   }
 509   g_mutex_unlock (&sink->client_lock);
 510
 511   /* unmap all memory again */
 512   for (i = 0; i < n_mem; i++) {
 513     mem = gst_buffer_peek_memory (buffer, i, GST_MAP_READ);
 514     gst_memory_unmap (mem);
 515   }
 516   g_free (vec);
 517
 518   GST_LOG_OBJECT (sink, "sent %d bytes to %d (of %d) clients", size, num,
 519       no_clients);
 520
 521   return GST_FLOW_OK;
 522
 523 no_data:
 524   {
 525     return GST_FLOW_OK;
 526   }
 527 send_error:
 528   {
 529     GST_DEBUG ("got send error %s", err->message);
 530     g_clear_error (&err);
 531     return GST_FLOW_ERROR;
 532   }
 533 }
 534
 535 static void
 536 gst_multiudpsink_set_clients_string (GstMultiUDPSink * sink,
 537     const gchar * string)
 538 {
 539   gchar **clients;
 540   gint i;
 541
 542   clients = g_strsplit (string, ",", 0);
 543
 544   g_mutex_lock (&sink->client_lock);
 545   /* clear all existing clients */
 546   gst_multiudpsink_clear_internal (sink, FALSE);
 547   for (i = 0; clients[i]; i++) {
 548     gchar *host, *p;
 549     gint64 port = 0;
 550
 551     host = clients[i];
 552     p = strstr (clients[i], ":");
 553     if (p != NULL) {
 554       *p = '\0';
 555       port = g_ascii_strtoll (p + 1, NULL, 10);
 556     }
 557     if (port != 0)
 558       gst_multiudpsink_add_internal (sink, host, port, FALSE);
 559   }
 560   g_mutex_unlock (&sink->client_lock);
 561
 562   g_strfreev (clients);
 563 }
 564
 565 static gchar *
 566 gst_multiudpsink_get_clients_string (GstMultiUDPSink * sink)
 567 {
 568   GString *str;
 569   GList *clients;
 570
 571   str = g_string_new ("");
 572
 573   g_mutex_lock (&sink->client_lock);
 574   clients = sink->clients;
 575   while (clients) {
 576     GstUDPClient *client;
 577     gint count;
 578
 579     client = (GstUDPClient *) clients->data;
 580
 581     clients = g_list_next (clients);
 582
 583     count = client->refcount;
 584     while (count--) {
 585       g_string_append_printf (str, "%s:%d%s", client->host, client->port,
 586           (clients || count > 1 ? "," : ""));
 587     }
 588   }
 589   g_mutex_unlock (&sink->client_lock);
 590
 591   return g_string_free (str, FALSE);
 592 }
 593
 594 static void
 595 gst_multiudpsink_setup_qos_dscp (GstMultiUDPSink * sink)
 596 {
 597   /* don't touch on -1 */
 598   if (sink->qos_dscp < 0)
 599     return;
 600
 601   if (sink->used_socket == NULL)
 602     return;
 603
 604 #ifdef IP_TOS
 605   {
 606     gint tos;
 607     gint fd;
 608
 609     fd = g_socket_get_fd (sink->used_socket);
 610
 611     GST_DEBUG_OBJECT (sink, "setting TOS to %d", sink->qos_dscp);
 612
 613     /* Extract and shift 6 bits of DSFIELD */
 614     tos = (sink->qos_dscp & 0x3f) << 2;
 615
 616     if (setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos)) < 0) {
 617       GST_ERROR_OBJECT (sink, "could not set TOS: %s", g_strerror (errno));
 618     }
 619 #ifdef IPV6_TCLASS
 620     if (setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos)) < 0) {
 621       GST_ERROR_OBJECT (sink, "could not set TCLASS: %s", g_strerror (errno));
 622     }
 623   }
 624 #endif
 625 #endif
 626 }
 627
 628 static void
 629 gst_multiudpsink_set_property (GObject * object, guint prop_id,
 630     const GValue * value, GParamSpec * pspec)
 631 {
 632   GstMultiUDPSink *udpsink;
 633
 634   udpsink = GST_MULTIUDPSINK (object);
 635
 636   switch (prop_id) {
 637     case PROP_SOCKET:
 638       if (udpsink->socket != NULL && udpsink->socket != udpsink->used_socket &&
 639           udpsink->close_socket) {
 640         GError *err = NULL;
 641
 642         if (!g_socket_close (udpsink->socket, &err)) {
 643           GST_ERROR ("failed to close socket %p: %s", udpsink->socket,
 644               err->message);
 645           g_clear_error (&err);
 646         }
 647       }
 648       if (udpsink->socket)
 649         g_object_unref (udpsink->socket);
 650       udpsink->socket = g_value_dup_object (value);
 651       GST_DEBUG_OBJECT (udpsink, "setting socket to %p", udpsink->socket);
 652       break;
 653     case PROP_CLOSE_SOCKET:
 654       udpsink->close_socket = g_value_get_boolean (value);
 655       break;
 656     case PROP_CLIENTS:
 657       gst_multiudpsink_set_clients_string (udpsink, g_value_get_string (value));
 658       break;
 659     case PROP_AUTO_MULTICAST:
 660       udpsink->auto_multicast = g_value_get_boolean (value);
 661       break;
 662     case PROP_TTL:
 663       udpsink->ttl = g_value_get_int (value);
 664       break;
 665     case PROP_TTL_MC:
 666       udpsink->ttl_mc = g_value_get_int (value);
 667       break;
 668     case PROP_LOOP:
 669       udpsink->loop = g_value_get_boolean (value);
 670       break;
 671     case PROP_QOS_DSCP:
 672       udpsink->qos_dscp = g_value_get_int (value);
 673       gst_multiudpsink_setup_qos_dscp (udpsink);
 674       break;
 675     case PROP_SEND_DUPLICATES:
 676       udpsink->send_duplicates = g_value_get_boolean (value);
 677       break;
 678     case PROP_BUFFER_SIZE:
 679       udpsink->buffer_size = g_value_get_int (value);
 680       break;
 681     default:
 682       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 683       break;
 684   }
 685 }
 686
 687 static void
 688 gst_multiudpsink_get_property (GObject * object, guint prop_id, GValue * value,
 689     GParamSpec * pspec)
 690 {
 691   GstMultiUDPSink *udpsink;
 692
 693   udpsink = GST_MULTIUDPSINK (object);
 694
 695   switch (prop_id) {
 696     case PROP_BYTES_TO_SERVE:
 697       g_value_set_uint64 (value, udpsink->bytes_to_serve);
 698       break;
 699     case PROP_BYTES_SERVED:
 700       g_value_set_uint64 (value, udpsink->bytes_served);
 701       break;
 702     case PROP_SOCKET:
 703       g_value_set_object (value, udpsink->socket);
 704       break;
 705     case PROP_CLOSE_SOCKET:
 706       g_value_set_boolean (value, udpsink->close_socket);
 707       break;
 708     case PROP_USED_SOCKET:
 709       g_value_set_object (value, udpsink->used_socket);
 710       break;
 711     case PROP_CLIENTS:
 712       g_value_take_string (value,
 713           gst_multiudpsink_get_clients_string (udpsink));
 714       break;
 715     case PROP_AUTO_MULTICAST:
 716       g_value_set_boolean (value, udpsink->auto_multicast);
 717       break;
 718     case PROP_TTL:
 719       g_value_set_int (value, udpsink->ttl);
 720       break;
 721     case PROP_TTL_MC:
 722       g_value_set_int (value, udpsink->ttl_mc);
 723       break;
 724     case PROP_LOOP:
 725       g_value_set_boolean (value, udpsink->loop);
 726       break;
 727     case PROP_QOS_DSCP:
 728       g_value_set_int (value, udpsink->qos_dscp);
 729       break;
 730     case PROP_SEND_DUPLICATES:
 731       g_value_set_boolean (value, udpsink->send_duplicates);
 732       break;
 733     case PROP_BUFFER_SIZE:
 734       g_value_set_int (value, udpsink->buffer_size);
 735       break;
 736     default:
 737       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 738       break;
 739   }
 740 }
 741
 742 static gboolean
 743 gst_multiudpsink_configure_client (GstMultiUDPSink * sink,
 744     GstUDPClient * client)
 745 {
 746   GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
 747   GInetAddress *addr = g_inet_socket_address_get_address (saddr);
 748   GError *err = NULL;
 749
 750   GST_DEBUG_OBJECT (sink, "configuring client %p", client);
 751
 752   if (g_inet_address_get_is_multicast (addr)) {
 753     GST_DEBUG_OBJECT (sink, "we have a multicast client %p", client);
 754     if (sink->auto_multicast) {
 755       GST_DEBUG_OBJECT (sink, "autojoining group");
 756       if (!g_socket_join_multicast_group (sink->used_socket, addr, FALSE, NULL,
 757               &err))
 758         goto join_group_failed;
 759     }
 760     GST_DEBUG_OBJECT (sink, "setting loop to %d", sink->loop);
 761     g_socket_set_multicast_loopback (sink->used_socket, sink->loop);
 762     GST_DEBUG_OBJECT (sink, "setting ttl to %d", sink->ttl_mc);
 763     g_socket_set_multicast_ttl (sink->used_socket, sink->ttl_mc);
 764   } else {
 765     GST_DEBUG_OBJECT (sink, "setting unicast ttl to %d", sink->ttl);
 766     g_socket_set_ttl (sink->used_socket, sink->ttl);
 767   }
 768   return TRUE;
 769
 770   /* ERRORS */
 771 join_group_failed:
 772   {
 773     gst_multiudpsink_stop (GST_BASE_SINK (sink));
 774     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
 775         ("Could not join multicast group: %s", err->message));
 776     g_clear_error (&err);
 777     return FALSE;
 778   }
 779 }
 780
 781 /* create a socket for sending to remote machine */
 782 static gboolean
 783 gst_multiudpsink_start (GstBaseSink * bsink)
 784 {
 785   GstMultiUDPSink *sink;
 786   GList *clients;
 787   GstUDPClient *client;
 788   GError *err = NULL;
 789
 790   sink = GST_MULTIUDPSINK (bsink);
 791
 792   if (sink->socket == NULL) {
 793     GST_DEBUG_OBJECT (sink, "creating sockets");
 794     /* create sender socket try IP6, fall back to IP4 */
 795     sink->family = G_SOCKET_FAMILY_IPV6;
 796     if ((sink->used_socket =
 797             g_socket_new (G_SOCKET_FAMILY_IPV6,
 798                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
 799       sink->family = G_SOCKET_FAMILY_IPV4;
 800       if ((sink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
 801                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 802         goto no_socket;
 803     }
 804
 805     GST_DEBUG_OBJECT (sink, "have socket");
 806     sink->external_socket = FALSE;
 807   } else {
 808     GST_DEBUG_OBJECT (sink, "using configured socket");
 809     /* we use the configured socket */
 810     sink->used_socket = G_SOCKET (g_object_ref (sink->socket));
 811     sink->family = g_socket_get_family (sink->used_socket);
 812     sink->external_socket = TRUE;
 813   }
 814
 815 #ifdef SO_SNDBUF
 816
 817   {
 818     socklen_t len;
 819     gint sndsize, ret;
 820
 821     len = sizeof (sndsize);
 822     if (sink->buffer_size != 0) {
 823       sndsize = sink->buffer_size;
 824
 825       GST_DEBUG_OBJECT (sink, "setting udp buffer of %d bytes", sndsize);
 826       /* set buffer size, Note that on Linux this is typically limited to a
 827        * maximum of around 100K. Also a minimum of 128 bytes is required on
 828        * Linux. */
 829       ret =
 830           setsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET,
 831           SO_SNDBUF, (void *) &sndsize, len);
 832       if (ret != 0) {
 833         GST_ELEMENT_WARNING (sink, RESOURCE, SETTINGS, (NULL),
 834             ("Could not create a buffer of requested %d bytes, %d: %s",
 835                 sndsize, ret, g_strerror (errno)));
 836       }
 837     }
 838
 839     /* read the value of the receive buffer. Note that on linux this returns 2x the
 840      * value we set because the kernel allocates extra memory for metadata.
 841      * The default on Linux is about 100K (which is about 50K without metadata) */
 842     ret =
 843         getsockopt (g_socket_get_fd (sink->used_socket), SOL_SOCKET, SO_SNDBUF,
 844         (void *) &sndsize, &len);
 845     if (ret == 0)
 846       GST_DEBUG_OBJECT (sink, "have udp buffer of %d bytes", sndsize);
 847     else
 848       GST_DEBUG_OBJECT (sink, "could not get udp buffer size");
 849   }
 850 #endif
 851
 852   g_socket_set_broadcast (sink->used_socket, TRUE);
 853
 854   sink->bytes_to_serve = 0;
 855   sink->bytes_served = 0;
 856
 857   gst_multiudpsink_setup_qos_dscp (sink);
 858
 859   /* look for multicast clients and join multicast groups appropriately
 860      set also ttl and multicast loopback delivery appropriately  */
 861   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
 862     client = (GstUDPClient *) clients->data;
 863
 864     if (!gst_multiudpsink_configure_client (sink, client))
 865       return FALSE;
 866   }
 867   return TRUE;
 868
 869   /* ERRORS */
 870 no_socket:
 871   {
 872     GST_ELEMENT_ERROR (sink, RESOURCE, FAILED, (NULL),
 873         ("Could not create socket: %s", err->message));
 874     g_clear_error (&err);
 875     return FALSE;
 876   }
 877 }
 878
 879 static gboolean
 880 gst_multiudpsink_stop (GstBaseSink * bsink)
 881 {
 882   GstMultiUDPSink *udpsink;
 883
 884   udpsink = GST_MULTIUDPSINK (bsink);
 885
 886   if (udpsink->used_socket) {
 887     if (udpsink->close_socket || !udpsink->external_socket) {
 888       GError *err = NULL;
 889
 890       if (!g_socket_close (udpsink->used_socket, &err)) {
 891         GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
 892         g_clear_error (&err);
 893       }
 894     }
 895
 896     g_object_unref (udpsink->used_socket);
 897     udpsink->used_socket = NULL;
 898   }
 899
 900   return TRUE;
 901 }
 902
 903 static void
 904 gst_multiudpsink_add_internal (GstMultiUDPSink * sink, const gchar * host,
 905     gint port, gboolean lock)
 906 {
 907   GstUDPClient *client;
 908   GstUDPClient udpclient;
 909   GTimeVal now;
 910   GList *find;
 911
 912   udpclient.host = (gchar *) host;
 913   udpclient.port = port;
 914
 915   GST_DEBUG_OBJECT (sink, "adding client on host %s, port %d", host, port);
 916
 917   if (lock)
 918     g_mutex_lock (&sink->client_lock);
 919
 920   find = g_list_find_custom (sink->clients, &udpclient,
 921       (GCompareFunc) client_compare);
 922   if (find) {
 923     client = (GstUDPClient *) find->data;
 924
 925     GST_DEBUG_OBJECT (sink, "found %d existing clients with host %s, port %d",
 926         client->refcount, host, port);
 927     client->refcount++;
 928   } else {
 929     client = create_client (sink, host, port);
 930     if (!client)
 931       goto error;
 932
 933     g_get_current_time (&now);
 934     client->connect_time = GST_TIMEVAL_TO_TIME (now);
 935
 936     if (sink->used_socket)
 937       gst_multiudpsink_configure_client (sink, client);
 938
 939     GST_DEBUG_OBJECT (sink, "add client with host %s, port %d", host, port);
 940     sink->clients = g_list_prepend (sink->clients, client);
 941   }
 942
 943   if (lock)
 944     g_mutex_unlock (&sink->client_lock);
 945
 946   g_signal_emit (G_OBJECT (sink),
 947       gst_multiudpsink_signals[SIGNAL_CLIENT_ADDED], 0, host, port);
 948
 949   GST_DEBUG_OBJECT (sink, "added client on host %s, port %d", host, port);
 950   return;
 951
 952   /* ERRORS */
 953 error:
 954   {
 955     GST_DEBUG_OBJECT (sink, "did not add client on host %s, port %d", host,
 956         port);
 957     if (lock)
 958       g_mutex_unlock (&sink->client_lock);
 959     return;
 960   }
 961 }
 962
 963 void
 964 gst_multiudpsink_add (GstMultiUDPSink * sink, const gchar * host, gint port)
 965 {
 966   gst_multiudpsink_add_internal (sink, host, port, TRUE);
 967 }
 968
 969 void
 970 gst_multiudpsink_remove (GstMultiUDPSink * sink, const gchar * host, gint port)
 971 {
 972   GList *find;
 973   GstUDPClient udpclient;
 974   GstUDPClient *client;
 975   GTimeVal now;
 976
 977   udpclient.host = (gchar *) host;
 978   udpclient.port = port;
 979
 980   g_mutex_lock (&sink->client_lock);
 981   find = g_list_find_custom (sink->clients, &udpclient,
 982       (GCompareFunc) client_compare);
 983   if (!find)
 984     goto not_found;
 985
 986   client = (GstUDPClient *) find->data;
 987
 988   GST_DEBUG_OBJECT (sink, "found %d clients with host %s, port %d",
 989       client->refcount, host, port);
 990
 991   client->refcount--;
 992   if (client->refcount == 0) {
 993     GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
 994     GInetAddress *addr = g_inet_socket_address_get_address (saddr);
 995
 996     GST_DEBUG_OBJECT (sink, "remove client with host %s, port %d", host, port);
 997
 998     g_get_current_time (&now);
 999     client->disconnect_time = GST_TIMEVAL_TO_TIME (now);
1000
1001     if (sink->used_socket && sink->auto_multicast
1002         && g_inet_address_get_is_multicast (addr)) {
1003       GError *err = NULL;
1004
1005       if (!g_socket_leave_multicast_group (sink->used_socket, addr, FALSE, NULL,
1006               &err)) {
1007         GST_DEBUG_OBJECT (sink, "Failed to leave multicast group: %s",
1008             err->message);
1009         g_clear_error (&err);
1010       }
1011     }
1012
1013     /* Unlock to emit signal before we delete the actual client */
1014     g_mutex_unlock (&sink->client_lock);
1015     g_signal_emit (G_OBJECT (sink),
1016         gst_multiudpsink_signals[SIGNAL_CLIENT_REMOVED], 0, host, port);
1017     g_mutex_lock (&sink->client_lock);
1018
1019     sink->clients = g_list_delete_link (sink->clients, find);
1020
1021     free_client (client);
1022   }
1023   g_mutex_unlock (&sink->client_lock);
1024
1025   return;
1026
1027   /* ERRORS */
1028 not_found:
1029   {
1030     g_mutex_unlock (&sink->client_lock);
1031     GST_WARNING_OBJECT (sink, "client at host %s, port %d not found",
1032         host, port);
1033     return;
1034   }
1035 }
1036
1037 static void
1038 gst_multiudpsink_clear_internal (GstMultiUDPSink * sink, gboolean lock)
1039 {
1040   GST_DEBUG_OBJECT (sink, "clearing");
1041   /* we only need to remove the client structure, there is no additional
1042    * socket or anything to free for UDP */
1043   if (lock)
1044     g_mutex_lock (&sink->client_lock);
1045   g_list_foreach (sink->clients, (GFunc) free_client, sink);
1046   g_list_free (sink->clients);
1047   sink->clients = NULL;
1048   if (lock)
1049     g_mutex_unlock (&sink->client_lock);
1050 }
1051
1052 void
1053 gst_multiudpsink_clear (GstMultiUDPSink * sink)
1054 {
1055   gst_multiudpsink_clear_internal (sink, TRUE);
1056 }
1057
1058 GstStructure *
1059 gst_multiudpsink_get_stats (GstMultiUDPSink * sink, const gchar * host,
1060     gint port)
1061 {
1062   GstUDPClient *client;
1063   GstStructure *result = NULL;
1064   GstUDPClient udpclient;
1065   GList *find;
1066
1067   udpclient.host = (gchar *) host;
1068   udpclient.port = port;
1069
1070   g_mutex_lock (&sink->client_lock);
1071
1072   find = g_list_find_custom (sink->clients, &udpclient,
1073       (GCompareFunc) client_compare);
1074   if (!find)
1075     goto not_found;
1076
1077   GST_DEBUG_OBJECT (sink, "stats for client with host %s, port %d", host, port);
1078
1079   client = (GstUDPClient *) find->data;
1080
1081   result = gst_structure_new_empty ("multiudpsink-stats");
1082
1083   gst_structure_set (result,
1084       "bytes-sent", G_TYPE_UINT64, client->bytes_sent,
1085       "packets-sent", G_TYPE_UINT64, client->packets_sent,
1086       "connect-time", G_TYPE_UINT64, client->connect_time,
1087       "disconnect-time", G_TYPE_UINT64, client->disconnect_time, NULL);
1088
1089   g_mutex_unlock (&sink->client_lock);
1090
1091   return result;
1092
1093   /* ERRORS */
1094 not_found:
1095   {
1096     g_mutex_unlock (&sink->client_lock);
1097     GST_WARNING_OBJECT (sink, "client with host %s, port %d not found",
1098         host, port);
1099     /* Apparently (see comment in gstmultifdsink.c) returning NULL from here may
1100      * confuse/break python bindings */
1101     return gst_structure_new_empty ("multiudpsink-stats");
1102   }
1103 }
1104
1105 static gboolean
1106 gst_multiudpsink_unlock (GstBaseSink * bsink)
1107 {
1108   GstMultiUDPSink *sink;
1109
1110   sink = GST_MULTIUDPSINK (bsink);
1111
1112   g_cancellable_cancel (sink->cancellable);
1113
1114   return TRUE;
1115 }
1116
1117 static gboolean
1118 gst_multiudpsink_unlock_stop (GstBaseSink * bsink)
1119 {
1120   GstMultiUDPSink *sink;
1121
1122   sink = GST_MULTIUDPSINK (bsink);
1123
1124   g_cancellable_reset (sink->cancellable);
1125
1126   return TRUE;
1127 }