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