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