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