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