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 #include <sys/socket.h>
  44
  45 #include "gst/glib-compat-private.h"
  46
  47 GST_DEBUG_CATEGORY_STATIC (multiudpsink_debug);
  48 #define GST_CAT_DEFAULT (multiudpsink_debug)
  49
  50 #define UDP_MAX_SIZE 65507
  51
  52 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
  53     GST_PAD_SINK,
  54     GST_PAD_ALWAYS,
  55     GST_STATIC_CAPS_ANY);
  56
  57 /* MultiUDPSink signals and args */
  58 enum
  59 {
  60   /* methods */
  61   SIGNAL_ADD,
  62   SIGNAL_REMOVE,
  63   SIGNAL_CLEAR,
  64   SIGNAL_GET_STATS,
  65
  66   /* signals */
  67   SIGNAL_CLIENT_ADDED,
  68   SIGNAL_CLIENT_REMOVED,
  69
  70   /* FILL ME */
  71   LAST_SIGNAL
  72 };
  73
  74 #define DEFAULT_SOCKET             NULL
  75 #define DEFAULT_CLOSE_SOCKET       TRUE
  76 #define DEFAULT_USED_SOCKET        NULL
  77 #define DEFAULT_CLIENTS            NULL
  78 #define DEFAULT_FAMILY             G_SOCKET_FAMILY_IPV6
  79 /* FIXME, this should be disabled by default, we don't need to join a multicast
  80  * group for sending, if this socket is also used for receiving, it should
  81  * be configured in the element that does the receive. */
  82 #define DEFAULT_AUTO_MULTICAST     TRUE
  83 #define DEFAULT_TTL                64
  84 #define DEFAULT_TTL_MC             1
  85 #define DEFAULT_LOOP               TRUE
  86 #define DEFAULT_QOS_DSCP           -1
  87 #define DEFAULT_SEND_DUPLICATES    TRUE
  88 #define DEFAULT_BUFFER_SIZE        0
  89
  90 enum
  91 {
  92   PROP_0,
  93   PROP_BYTES_TO_SERVE,
  94   PROP_BYTES_SERVED,
  95   PROP_SOCKET,
  96   PROP_CLOSE_SOCKET,
  97   PROP_USED_SOCKET,
  98   PROP_CLIENTS,
  99   PROP_AUTO_MULTICAST,
 100   PROP_TTL,
 101   PROP_TTL_MC,
 102   PROP_LOOP,
 103   PROP_QOS_DSCP,
 104   PROP_SEND_DUPLICATES,
 105   PROP_BUFFER_SIZE,
 106   PROP_LAST
 107 };
 108
 109 static void gst_multiudpsink_finalize (GObject * object);
 110
 111 static GstFlowReturn gst_multiudpsink_render (GstBaseSink * sink,
 112     GstBuffer * buffer);
 113
 114 static gboolean gst_multiudpsink_start (GstBaseSink * bsink);
 115 static gboolean gst_multiudpsink_stop (GstBaseSink * bsink);
 116 static gboolean gst_multiudpsink_unlock (GstBaseSink * bsink);
 117 static gboolean gst_multiudpsink_unlock_stop (GstBaseSink * bsink);
 118
 119 static void gst_multiudpsink_set_property (GObject * object, guint prop_id,
 120     const GValue * value, GParamSpec * pspec);
 121 static void gst_multiudpsink_get_property (GObject * object, guint prop_id,
 122     GValue * value, GParamSpec * pspec);
 123
 124 static void gst_multiudpsink_add_internal (GstMultiUDPSink * sink,
 125     const gchar * host, gint port, gboolean lock);
 126 static void gst_multiudpsink_clear_internal (GstMultiUDPSink * sink,
 127     gboolean lock);
 128
 129 static guint gst_multiudpsink_signals[LAST_SIGNAL] = { 0 };
 130
 131 #define gst_multiudpsink_parent_class parent_class
 132 G_DEFINE_TYPE (GstMultiUDPSink, gst_multiudpsink, GST_TYPE_BASE_SINK);
 133
 134 static void
 135 gst_multiudpsink_class_init (GstMultiUDPSinkClass * klass)
 136 {
 137   GObjectClass *gobject_class;
 138   GstElementClass *gstelement_class;
 139   GstBaseSinkClass *gstbasesink_class;
 140
 141   gobject_class = (GObjectClass *) klass;
 142   gstelement_class = (GstElementClass *) klass;
 143   gstbasesink_class = (GstBaseSinkClass *) klass;
 144
 145   gobject_class->set_property = gst_multiudpsink_set_property;
 146   gobject_class->get_property = gst_multiudpsink_get_property;
 147   gobject_class->finalize = gst_multiudpsink_finalize;
 148
 149   /**
 150    * GstMultiUDPSink::add:
 151    * @gstmultiudpsink: the sink on which the signal is emitted
 152    * @host: the hostname/IP address of the client to add
 153    * @port: the port of the client to add
 154    *
 155    * Add a client with destination @host and @port to the list of
 156    * clients. When the same host/port pair is added multiple times, the
 157    * send-duplicates property defines if the packets are sent multiple times to
 158    * the same host/port pair or not.
 159    *
 160    * When a host/port pair is added multiple times, an equal amount of remove
 161    * calls must be performed to actually remove the host/port pair from the list
 162    * of destinations.
 163    */
 164   gst_multiudpsink_signals[SIGNAL_ADD] =
 165       g_signal_new ("add", G_TYPE_FROM_CLASS (klass),
 166       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 167       G_STRUCT_OFFSET (GstMultiUDPSinkClass, add),
 168       NULL, NULL, gst_udp_marshal_VOID__STRING_INT, G_TYPE_NONE, 2,
 169       G_TYPE_STRING, G_TYPE_INT);
 170   /**
 171    * GstMultiUDPSink::remove:
 172    * @gstmultiudpsink: the sink on which the signal is emitted
 173    * @host: the hostname/IP address of the client to remove
 174    * @port: the port of the client to remove
 175    *
 176    * Remove the client with destination @host and @port from the list of
 177    * clients.
 178    */
 179   gst_multiudpsink_signals[SIGNAL_REMOVE] =
 180       g_signal_new ("remove", G_TYPE_FROM_CLASS (klass),
 181       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 182       G_STRUCT_OFFSET (GstMultiUDPSinkClass, remove),
 183       NULL, NULL, gst_udp_marshal_VOID__STRING_INT, G_TYPE_NONE, 2,
 184       G_TYPE_STRING, G_TYPE_INT);
 185   /**
 186    * GstMultiUDPSink::clear:
 187    * @gstmultiudpsink: the sink on which the signal is emitted
 188    *
 189    * Clear the list of clients.
 190    */
 191   gst_multiudpsink_signals[SIGNAL_CLEAR] =
 192       g_signal_new ("clear", G_TYPE_FROM_CLASS (klass),
 193       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 194       G_STRUCT_OFFSET (GstMultiUDPSinkClass, clear),
 195       NULL, NULL, g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 196   /**
 197    * GstMultiUDPSink::get-stats:
 198    * @gstmultiudpsink: the sink on which the signal is emitted
 199    * @host: the hostname/IP address of the client to get stats on
 200    * @port: the port of the client to get stats on
 201    *
 202    * Get the statistics of the client with destination @host and @port.
 203    *
 204    * Returns: a GstStructure: bytes_sent, packets_sent,
 205    *           connect_time (in epoch seconds), disconnect_time (in epoch seconds)
 206    */
 207   gst_multiudpsink_signals[SIGNAL_GET_STATS] =
 208       g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
 209       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 210       G_STRUCT_OFFSET (GstMultiUDPSinkClass, get_stats),
 211       NULL, NULL, gst_udp_marshal_BOXED__STRING_INT, GST_TYPE_STRUCTURE, 2,
 212       G_TYPE_STRING, G_TYPE_INT);
 213   /**
 214    * GstMultiUDPSink::client-added:
 215    * @gstmultiudpsink: the sink emitting the signal
 216    * @host: the hostname/IP address of the added client
 217    * @port: the port of the added client
 218    *
 219    * Signal emited when a new client is added to the list of
 220    * clients.
 221    */
 222   gst_multiudpsink_signals[SIGNAL_CLIENT_ADDED] =
 223       g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
 224       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiUDPSinkClass, client_added),
 225       NULL, NULL, gst_udp_marshal_VOID__STRING_INT, G_TYPE_NONE, 2,
 226       G_TYPE_STRING, G_TYPE_INT);
 227   /**
 228    * GstMultiUDPSink::client-removed:
 229    * @gstmultiudpsink: the sink emitting the signal
 230    * @host: the hostname/IP address of the removed client
 231    * @port: the port of the removed client
 232    *
 233    * Signal emited when a client is removed from the list of
 234    * clients.
 235    */
 236   gst_multiudpsink_signals[SIGNAL_CLIENT_REMOVED] =
 237       g_signal_new ("client-removed", G_TYPE_FROM_CLASS (klass),
 238       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiUDPSinkClass,
 239           client_removed), NULL, NULL, gst_udp_marshal_VOID__STRING_INT,
 240       G_TYPE_NONE, 2, G_TYPE_STRING, G_TYPE_INT);
 241
 242   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BYTES_TO_SERVE,
 243       g_param_spec_uint64 ("bytes-to-serve", "Bytes to serve",
 244           "Number of bytes received to serve to clients", 0, G_MAXUINT64, 0,
 245           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 246   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BYTES_SERVED,
 247       g_param_spec_uint64 ("bytes-served", "Bytes served",
 248           "Total number of bytes sent to all clients", 0, G_MAXUINT64, 0,
 249           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 250   g_object_class_install_property (gobject_class, PROP_SOCKET,
 251       g_param_spec_object ("socket", "Socket Handle",
 252           "Socket to use for UDP sending. (NULL == allocate)",
 253           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 254   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 255       g_param_spec_boolean ("close-socket", "Close socket",
 256           "Close socket if passed as property on state change",
 257           DEFAULT_CLOSE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 258   g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
 259       g_param_spec_object ("used-socket", "Used Socket Handle",
 260           "Socket currently in use for UDP sending. (NULL == no socket)",
 261           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 262   g_object_class_install_property (gobject_class, PROP_CLIENTS,
 263       g_param_spec_string ("clients", "Clients",
 264           "A comma separated list of host:port pairs with destinations",
 265           DEFAULT_CLIENTS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 266   g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
 267       g_param_spec_boolean ("auto-multicast",
 268           "Automatically join/leave multicast groups",
 269           "Automatically join/leave the multicast groups, FALSE means user"
 270           " has to do it himself", DEFAULT_AUTO_MULTICAST,
 271           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 272   g_object_class_install_property (gobject_class, PROP_TTL,
 273       g_param_spec_int ("ttl", "Unicast TTL",
 274           "Used for setting the unicast TTL parameter",
 275           0, 255, DEFAULT_TTL, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 276   g_object_class_install_property (gobject_class, PROP_TTL_MC,
 277       g_param_spec_int ("ttl-mc", "Multicast TTL",
 278           "Used for setting the multicast TTL parameter",
 279           0, 255, DEFAULT_TTL_MC, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 280   g_object_class_install_property (gobject_class, PROP_LOOP,
 281       g_param_spec_boolean ("loop", "Multicast Loopback",
 282           "Used for setting the multicast loop parameter. TRUE = enable,"
 283           " FALSE = disable", DEFAULT_LOOP,
 284           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 285   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_QOS_DSCP,
 286       g_param_spec_int ("qos-dscp", "QoS diff srv code point",
 287           "Quality of Service, differentiated services code point (-1 default)",
 288           -1, 63, DEFAULT_QOS_DSCP,
 289           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 290   /**
 291    * GstMultiUDPSink::send-duplicates
 292    *
 293    * When a host/port pair is added mutliple times, send the packet to the host
 294    * multiple times as well.
 295    *
 296    * Since: 0.10.26
 297    */
 298   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_SEND_DUPLICATES,
 299       g_param_spec_boolean ("send-duplicates", "Send Duplicates",
 300           "When a distination/port pair is added multiple times, send packets "
 301           "multiple times as well", DEFAULT_SEND_DUPLICATES,
 302           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 303
 304   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
 305       g_param_spec_int ("buffer-size", "Buffer Size",
 306           "Size of the kernel send buffer in bytes, 0=default", 0, G_MAXINT,
 307           DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 308
 309   gst_element_class_add_pad_template (gstelement_class,
 310       gst_static_pad_template_get (&sink_template));
 311
 312   gst_element_class_set_details_simple (gstelement_class, "UDP packet sender",
 313       "Sink/Network",
 314       "Send data over the network via UDP",
 315       "Wim Taymans <wim.taymans@gmail.com>");
 316
 317   gstbasesink_class->render = gst_multiudpsink_render;
 318   gstbasesink_class->start = gst_multiudpsink_start;
 319   gstbasesink_class->stop = gst_multiudpsink_stop;
 320   gstbasesink_class->unlock = gst_multiudpsink_unlock;
 321   gstbasesink_class->unlock_stop = gst_multiudpsink_unlock_stop;
 322   klass->add = gst_multiudpsink_add;
 323   klass->remove = gst_multiudpsink_remove;
 324   klass->clear = gst_multiudpsink_clear;
 325   klass->get_stats = gst_multiudpsink_get_stats;
 326
 327   GST_DEBUG_CATEGORY_INIT (multiudpsink_debug, "multiudpsink", 0, "UDP sink");
 328 }
 329
 330
 331 static void
 332 gst_multiudpsink_init (GstMultiUDPSink * sink)
 333 {
 334   g_mutex_init (&sink->client_lock);
 335   sink->socket = DEFAULT_SOCKET;
 336   sink->used_socket = DEFAULT_USED_SOCKET;
 337   sink->close_socket = DEFAULT_CLOSE_SOCKET;
 338   sink->external_socket = (sink->socket != NULL);
 339   sink->auto_multicast = DEFAULT_AUTO_MULTICAST;
 340   sink->ttl = DEFAULT_TTL;
 341   sink->ttl_mc = DEFAULT_TTL_MC;
 342   sink->loop = DEFAULT_LOOP;
 343   sink->qos_dscp = DEFAULT_QOS_DSCP;
 344   sink->family = DEFAULT_FAMILY;
 345   sink->send_duplicates = DEFAULT_SEND_DUPLICATES;
 346
 347   sink->cancellable = g_cancellable_new ();
 348 }
 349
 350 static GstUDPClient *
 351 create_client (GstMultiUDPSink * sink, const gchar * host, gint port)
 352 {
 353   GstUDPClient *client;
 354   GInetAddress *addr;
 355   GResolver *resolver;
 356   GError *err = NULL;
 357
 358   addr = g_inet_address_new_from_string (host);
 359   if (!addr) {
 360     GList *results;
 361
 362     resolver = g_resolver_get_default ();
 363     results =
 364         g_resolver_lookup_by_name (resolver, host, sink->cancellable, &err);
 365     if (!results)
 366       goto name_resolve;
 367     addr = G_INET_ADDRESS (g_object_ref (results->data));
 368
 369     g_resolver_free_addresses (results);
 370     g_object_unref (resolver);
 371   }
 372 #ifndef GST_DISABLE_GST_DEBUG
 373   {
 374     gchar *ip = g_inet_address_to_string (addr);
 375
 376     GST_DEBUG_OBJECT (sink, "IP address for host %s is %s", host, ip);
 377     g_free (ip);
 378   }
 379 #endif
 380
 381   client = g_slice_new0 (GstUDPClient);
 382   client->refcount = 1;
 383   client->host = g_strdup (host);
 384   client->port = port;
 385   client->addr = g_inet_socket_address_new (addr, port);
 386   g_object_unref (addr);
 387
 388   return client;
 389
 390 name_resolve:
 391   {
 392     g_object_unref (resolver);
 393
 394     return NULL;
 395   }
 396 }
 397
 398 static void
 399 free_client (GstUDPClient * client)
 400 {
 401   g_object_unref (client->addr);
 402   g_free (client->host);
 403   g_slice_free (GstUDPClient, client);
 404 }
 405
 406 static gint
 407 client_compare (GstUDPClient * a, GstUDPClient * b)
 408 {
 409   if ((a->port == b->port) && (strcmp (a->host, b->host) == 0))
 410     return 0;
 411
 412   return 1;
 413 }
 414
 415 static void
 416 gst_multiudpsink_finalize (GObject * object)
 417 {
 418   GstMultiUDPSink *sink;
 419
 420   sink = GST_MULTIUDPSINK (object);
 421
 422   g_list_foreach (sink->clients, (GFunc) free_client, NULL);
 423   g_list_free (sink->clients);
 424
 425   if (sink->socket)
 426     g_object_unref (sink->socket);
 427   sink->socket = NULL;
 428
 429   if (sink->used_socket)
 430     g_object_unref (sink->used_socket);
 431   sink->used_socket = NULL;
 432
 433   if (sink->cancellable)
 434     g_object_unref (sink->cancellable);
 435   sink->cancellable = NULL;
 436
 437   g_mutex_clear (&sink->client_lock);
 438
 439   G_OBJECT_CLASS (parent_class)->finalize (object);
 440 }
 441
 442 static GstFlowReturn
 443 gst_multiudpsink_render (GstBaseSink * bsink, GstBuffer * buffer)
 444 {
 445   GstMultiUDPSink *sink;
 446   GList *clients;
 447   GOutputVector *vec;
 448   GstMapInfo *map;
 449   guint n_mem, i;
 450   gsize size;
 451   GstMemory *mem;
 452   gint num, no_clients;
 453   GError *err = NULL;
 454
 455   sink = GST_MULTIUDPSINK (bsink);
 456
 457   n_mem = gst_buffer_n_memory (buffer);
 458   if (n_mem == 0)
 459     goto no_data;
 460
 461   vec = g_new (GOutputVector, n_mem);
 462   map = g_new (GstMapInfo, n_mem);
 463
 464   size = 0;
 465   for (i = 0; i < n_mem; i++) {
 466     mem = gst_buffer_get_memory (buffer, i);
 467     gst_memory_map (mem, &map[i], GST_MAP_READ);
 468
 469     if (map[i].size > UDP_MAX_SIZE) {
 470       GST_WARNING ("Attempting to send a UDP packet larger than maximum "
 471           "size (%" G_GSIZE_FORMAT " > %d)", map[i].size, UDP_MAX_SIZE);
 472     }
 473
 474     vec[i].buffer = map[i].data;
 475     vec[i].size = map[i].size;
 476
 477     size += map[i].size;
 478   }
 479
 480   sink->bytes_to_serve += size;
 481
 482   /* grab lock while iterating and sending to clients, this should be
 483    * fast as UDP never blocks */
 484   g_mutex_lock (&sink->client_lock);
 485   GST_LOG_OBJECT (bsink, "about to send %" G_GSIZE_FORMAT " bytes", size);
 486
 487   no_clients = 0;
 488   num = 0;
 489   for (clients = sink->clients; clients; clients = g_list_next (clients)) {
 490     GstUDPClient *client;
 491     gint count;
 492
 493     client = (GstUDPClient *) clients->data;
 494     no_clients++;
 495     GST_LOG_OBJECT (sink, "sending %" G_GSIZE_FORMAT " bytes to client %p",
 496         size, client);
 497
 498     count = sink->send_duplicates ? client->refcount : 1;
 499
 500     while (count--) {
 501       gssize ret;
 502
 503       ret =
 504           g_socket_send_message (sink->used_socket, client->addr, vec, n_mem,
 505           NULL, 0, 0, sink->cancellable, &err);
 506
 507       if (ret < 0)
 508         goto send_error;
 509
 510       num++;
 511       client->bytes_sent += ret;
 512       client->packets_sent++;
 513       sink->bytes_served += ret;
 514     }
 515   }
 516   g_mutex_unlock (&sink->client_lock);
 517
 518   /* unmap all memory again */
 519   for (i = 0; i < n_mem; i++) {
 520     gst_memory_unmap (map[i].memory, &map[i]);
 521     gst_memory_unref (map[i].memory);
 522   }
 523
 524   g_free (vec);
 525   g_free (map);
 526
 527   GST_LOG_OBJECT (sink, "sent %" G_GSIZE_FORMAT " bytes to %d (of %d) clients",
 528       size, num, no_clients);
 529
 530   return GST_FLOW_OK;
 531
 532 no_data:
 533   {
 534     return GST_FLOW_OK;
 535   }
 536 send_error:
 537   {
 538     GST_DEBUG ("got send error %s", err->message);
 539     g_clear_error (&err);
 540     return GST_FLOW_ERROR;
 541   }
 542 }
 543
 544 static void
 545 gst_multiudpsink_set_clients_string (GstMultiUDPSink * sink,
 546     const gchar * string)
 547 {
 548   gchar **clients;
 549   gint i;
 550
 551   clients = g_strsplit (string, ",", 0);
 552
 553   g_mutex_lock (&sink->client_lock);
 554   /* clear all existing clients */
 555   gst_multiudpsink_clear_internal (sink, FALSE);
 556   for (i = 0; clients[i]; i++) {
 557     gchar *host, *p;
 558     gint64 port = 0;
 559
 560     host = clients[i];
 561     p = strstr (clients[i], ":");
 562     if (p != NULL) {
 563       *p = '\0';
 564       port = g_ascii_strtoll (p + 1, NULL, 10);
 565     }
 566     if (port != 0)
 567       gst_multiudpsink_add_internal (sink, host, port, FALSE);
 568   }
 569   g_mutex_unlock (&sink->client_lock);
 570
 571   g_strfreev (clients);
 572 }
 573
 574 static gchar *
 575 gst_multiudpsink_get_clients_string (GstMultiUDPSink * sink)
 576 {
 577   GString *str;
 578   GList *clients;
 579
 580   str = g_string_new ("");
 581
 582   g_mutex_lock (&sink->client_lock);
 583   clients = sink->clients;
 584   while (clients) {
 585     GstUDPClient *client;
 586     gint count;
 587
 588     client = (GstUDPClient *) clients->data;
 589
 590     clients = g_list_next (clients);
 591
 592     count = client->refcount;
 593     while (count--) {
 594       g_string_append_printf (str, "%s:%d%s", client->host, client->port,
 595           (clients || count > 1 ? "," : ""));
 596     }
 597   }
 598   g_mutex_unlock (&sink->client_lock);
 599
 600   return g_string_free (str, FALSE);
 601 }
 602
 603 static void
 604 gst_multiudpsink_setup_qos_dscp (GstMultiUDPSink * sink)
 605 {
 606   /* don't touch on -1 */
 607   if (sink->qos_dscp < 0)
 608     return;
 609
 610   if (sink->used_socket == NULL)
 611     return;
 612
 613 #ifdef IP_TOS
 614   {
 615     gint tos;
 616     gint fd;
 617
 618     fd = g_socket_get_fd (sink->used_socket);
 619
 620     GST_DEBUG_OBJECT (sink, "setting TOS to %d", sink->qos_dscp);
 621
 622     /* Extract and shift 6 bits of DSFIELD */
 623     tos = (sink->qos_dscp & 0x3f) << 2;
 624
 625     if (setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos)) < 0) {
 626       GST_ERROR_OBJECT (sink, "could not set TOS: %s", g_strerror (errno));
 627     }
 628 #ifdef IPV6_TCLASS
 629     if (setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos)) < 0) {
 630       GST_ERROR_OBJECT (sink, "could not set TCLASS: %s", g_strerror (errno));
 631     }
 632   }
 633 #endif
 634 #endif
 635 }
 636
 637 static void
 638 gst_multiudpsink_set_property (GObject * object, guint prop_id,
 639     const GValue * value, GParamSpec * pspec)
 640 {
 641   GstMultiUDPSink *udpsink;
 642
 643   udpsink = GST_MULTIUDPSINK (object);
 644
 645   switch (prop_id) {
 646     case PROP_SOCKET:
 647       if (udpsink->socket != NULL && udpsink->socket != udpsink->used_socket &&
 648           udpsink->close_socket) {
 649         GError *err = NULL;
 650
 651         if (!g_socket_close (udpsink->socket, &err)) {
 652           GST_ERROR ("failed to close socket %p: %s", udpsink->socket,
 653               err->message);
 654           g_clear_error (&err);
 655         }
 656       }
 657       if (udpsink->socket)
 658         g_object_unref (udpsink->socket);
 659       udpsink->socket = g_value_dup_object (value);
 660       GST_DEBUG_OBJECT (udpsink, "setting socket to %p", udpsink->socket);
 661       break;
 662     case PROP_CLOSE_SOCKET:
 663       udpsink->close_socket = g_value_get_boolean (value);
 664       break;
 665     case PROP_CLIENTS:
 666       gst_multiudpsink_set_clients_string (udpsink, g_value_get_string (value));
 667       break;
 668     case PROP_AUTO_MULTICAST:
 669       udpsink->auto_multicast = g_value_get_boolean (value);
 670       break;
 671     case PROP_TTL:
 672       udpsink->ttl = g_value_get_int (value);
 673       break;
 674     case PROP_TTL_MC:
 675       udpsink->ttl_mc = g_value_get_int (value);
 676       break;
 677     case PROP_LOOP:
 678       udpsink->loop = g_value_get_boolean (value);
 679       break;
 680     case PROP_QOS_DSCP:
 681       udpsink->qos_dscp = g_value_get_int (value);
 682       gst_multiudpsink_setup_qos_dscp (udpsink);
 683       break;
 684     case PROP_SEND_DUPLICATES:
 685       udpsink->send_duplicates = g_value_get_boolean (value);
 686       break;
 687     case PROP_BUFFER_SIZE:
 688       udpsink->buffer_size = g_value_get_int (value);
 689       break;
 690     default:
 691       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 692       break;
 693   }
 694 }
 695
 696 static void
 697 gst_multiudpsink_get_property (GObject * object, guint prop_id, GValue * value,
 698     GParamSpec * pspec)
 699 {
 700   GstMultiUDPSink *udpsink;
 701
 702   udpsink = GST_MULTIUDPSINK (object);
 703
 704   switch (prop_id) {
 705     case PROP_BYTES_TO_SERVE:
 706       g_value_set_uint64 (value, udpsink->bytes_to_serve);
 707       break;
 708     case PROP_BYTES_SERVED:
 709       g_value_set_uint64 (value, udpsink->bytes_served);
 710       break;
 711     case PROP_SOCKET:
 712       g_value_set_object (value, udpsink->socket);
 713       break;
 714     case PROP_CLOSE_SOCKET:
 715       g_value_set_boolean (value, udpsink->close_socket);
 716       break;
 717     case PROP_USED_SOCKET:
 718       g_value_set_object (value, udpsink->used_socket);
 719       break;
 720     case PROP_CLIENTS:
 721       g_value_take_string (value,
 722           gst_multiudpsink_get_clients_string (udpsink));
 723       break;
 724     case PROP_AUTO_MULTICAST:
 725       g_value_set_boolean (value, udpsink->auto_multicast);
 726       break;
 727     case PROP_TTL:
 728       g_value_set_int (value, udpsink->ttl);
 729       break;
 730     case PROP_TTL_MC:
 731       g_value_set_int (value, udpsink->ttl_mc);
 732       break;
 733     case PROP_LOOP:
 734       g_value_set_boolean (value, udpsink->loop);
 735       break;
 736     case PROP_QOS_DSCP:
 737       g_value_set_int (value, udpsink->qos_dscp);
 738       break;
 739     case PROP_SEND_DUPLICATES:
 740       g_value_set_boolean (value, udpsink->send_duplicates);
 741       break;
 742     case PROP_BUFFER_SIZE:
 743       g_value_set_int (value, udpsink->buffer_size);
 744       break;
 745     default:
 746       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 747       break;
 748   }
 749 }
 750
 751 static gboolean
 752 gst_multiudpsink_configure_client (GstMultiUDPSink * sink,
 753     GstUDPClient * client)
 754 {
 755   GInetSocketAddress *saddr = G_INET_SOCKET_ADDRESS (client->addr);
 756   GInetAddress *addr = g_inet_socket_address_get_address (saddr);
 757   GError *err = NULL;
 758
 759   GST_DEBUG_OBJECT (sink, "configuring client %p", client);
 760
 761   if (g_inet_address_get_is_multicast (addr)) {
 762     GST_DEBUG_OBJECT (sink, "we have a multicast client %p", client);
 763     if (sink->auto_multicast) {
 764       GST_DEBUG_OBJECT (sink, "autojoining group");
 765       if (!g_socket_join_multicast_group (sink->used_socket, addr, FALSE, NULL,
 766               &err))
 767         goto join_group_failed;
 768     }
 769     GST_DEBUG_OBJECT (sink, "setting loop to %d", sink->loop);
 770     g_socket_set_multicast_loopback (sink->used_socket, sink->loop);
 771     GST_DEBUG_OBJECT (sink, "setting ttl to %d", sink->ttl_mc);
 772     g_socket_set_multicast_ttl (sink->used_socket, sink->ttl_mc);
 773   } else {
 774     GST_DEBUG_OBJECT (sink, "setting unicast ttl to %d", sink->ttl);
 775     g_socket_set_ttl (sink->used_socket, sink->ttl);
 776   }
 777   return TRUE;
 778
 779   /* ERRORS */
 780 join_group_failed:
 781   {
 782     gst_multiudpsink_stop (GST_BASE_SINK (sink));
 783     GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL),
 784         ("Could not join multicast group: %s", err->message));
 785     g_clear_error (&err);
 786     return FALSE;
 787   }
 788 }
 789
 790 /* create a socket for sending to remote machine */
 791 static gboolean
 792 gst_multiudpsink_start (GstBaseSink * bsink)
 793 {
 794   GstMultiUDPSink *sink;
 795   GList *clients;
 796   GstUDPClient *client;
 797   GError *err = NULL;
 798
 799   sink = GST_MULTIUDPSINK (bsink);
 800
 801   if (sink->socket == NULL) {
 802     GST_DEBUG_OBJECT (sink, "creating sockets");
 803     /* create sender socket try IP6, fall back to IP4 */
 804     sink->family = G_SOCKET_FAMILY_IPV6;
 805     if ((sink->used_socket =
 806             g_socket_new (G_SOCKET_FAMILY_IPV6,
 807                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
 808       sink->family = G_SOCKET_FAMILY_IPV4;
 809       if ((sink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
 810                   G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 811         goto no_socket;
 812     }
 813
 814     GST_DEBUG_OBJECT (sink, "have socket");
 815     sink->external_socket = FALSE;
 816   } else {
 817     GST_DEBUG_OBJECT (sink, "using configured socket");
 818     /* we use the configured socket */
 819     sink->used_socket = G_SOCKET (g_object_ref (sink->socket));
 820     sink->family = g_socket_get_family (sink->used_socket);
 821     sink->external_socket = TRUE;
 822   }
 823
 824 #ifdef SO_SNDBUF
 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 }