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