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