gst/udp/gstudpsrc.c

   1 /* GStreamer
   2  * Copyright (C) <2005> Wim Taymans <wim@fluendo.com>
   3  * Copyright (C) <2005> Nokia Corporation <kai.vehmanen@nokia.com>
   4  * Copyright (C) <2012> Collabora Ltd.
   5  *   Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
   6  * Copyright (C) 2014 Tim-Philipp Müller <tim@centricular.com>
   7  * Copyright (C) 2014 Centricular Ltd
   8  *
   9  * This library is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Library General Public
  11  * License as published by the Free Software Foundation; either
  12  * version 2 of the License, or (at your option) any later version.
  13  *
  14  * This library is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17  * Library General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU Library General Public
  20  * License along with this library; if not, write to the
  21  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  22  * Boston, MA 02110-1301, USA.
  23  */
  24
  25 /**
  26  * SECTION:element-udpsrc
  27  * @see_also: udpsink, multifdsink
  28  *
  29  * udpsrc is a network source that reads UDP packets from the network.
  30  * It can be combined with RTP depayloaders to implement RTP streaming.
  31  *
  32  * The udpsrc element supports automatic port allocation by setting the
  33  * #GstUDPSrc:port property to 0. After setting the udpsrc to PAUSED, the
  34  * allocated port can be obtained by reading the port property.
  35  *
  36  * udpsrc can read from multicast groups by setting the #GstUDPSrc:multicast-group
  37  * property to the IP address of the multicast group.
  38  *
  39  * Alternatively one can provide a custom socket to udpsrc with the #GstUDPSrc:socket
  40  * property, udpsrc will then not allocate a socket itself but use the provided
  41  * one.
  42  *
  43  * The #GstUDPSrc:caps property is mainly used to give a type to the UDP packet
  44  * so that they can be autoplugged in GStreamer pipelines. This is very useful
  45  * for RTP implementations where the contents of the UDP packets is transfered
  46  * out-of-bounds using SDP or other means.
  47  *
  48  * The #GstUDPSrc:buffer-size property is used to change the default kernel
  49  * buffersizes used for receiving packets. The buffer size may be increased for
  50  * high-volume connections, or may be decreased to limit the possible backlog of
  51  * incoming data. The system places an absolute limit on these values, on Linux,
  52  * for example, the default buffer size is typically 50K and can be increased to
  53  * maximally 100K.
  54  *
  55  * The #GstUDPSrc:skip-first-bytes property is used to strip off an arbitrary
  56  * number of bytes from the start of the raw udp packet and can be used to strip
  57  * off proprietary header, for example.
  58  *
  59  * The udpsrc is always a live source. It does however not provide a #GstClock,
  60  * this is left for upstream elements such as an RTP session manager or demuxer
  61  * (such as an MPEG demuxer). As with all live sources, the captured buffers
  62  * will have their timestamp set to the current running time of the pipeline.
  63  *
  64  * udpsrc implements a #GstURIHandler interface that handles udp://host:port
  65  * type URIs.
  66  *
  67  * If the #GstUDPSrc:timeout property is set to a value bigger than 0, udpsrc
  68  * will generate an element message named
  69  * <classname>&quot;GstUDPSrcTimeout&quot;</classname>
  70  * if no data was recieved in the given timeout.
  71  * The message's structure contains one field:
  72  * <itemizedlist>
  73  * <listitem>
  74  *   <para>
  75  *   #guint64
  76  *   <classname>&quot;timeout&quot;</classname>: the timeout in microseconds that
  77  *   expired when waiting for data.
  78  *   </para>
  79  * </listitem>
  80  * </itemizedlist>
  81  * The message is typically used to detect that no UDP arrives in the receiver
  82  * because it is blocked by a firewall.
  83  *
  84  * A custom file descriptor can be configured with the
  85  * #GstUDPSrc:socket property. The socket will be closed when setting
  86  * the element to READY by default. This behaviour can be overriden
  87  * with the #GstUDPSrc:close-socket property, in which case the
  88  * application is responsible for closing the file descriptor.
  89  *
  90  * <refsect2>
  91  * <title>Examples</title>
  92  * |[
  93  * gst-launch-1.0 -v udpsrc ! fakesink dump=1
  94  * ]| A pipeline to read from the default port and dump the udp packets.
  95  * To actually generate udp packets on the default port one can use the
  96  * udpsink element. When running the following pipeline in another terminal, the
  97  * above mentioned pipeline should dump data packets to the console.
  98  * |[
  99  * gst-launch-1.0 -v audiotestsrc ! udpsink
 100  * ]|
 101  * |[
 102  * gst-launch-1.0 -v udpsrc port=0 ! fakesink
 103  * ]| read udp packets from a free port.
 104  * </refsect2>
 105  */
 106 #ifdef HAVE_CONFIG_H
 107 #include "config.h"
 108 #endif
 109
 110 /* Needed to get struct in6_pktinfo */
 111 #define _GNU_SOURCE
 112 #include <sys/types.h>
 113 #ifdef HAVE_SYS_SOCKET_H
 114 #include <sys/socket.h>
 115 #endif
 116 #ifndef G_OS_WIN32
 117 #include <netinet/in.h>
 118 #endif
 119
 120 #include <string.h>
 121 #include "gstudpsrc.h"
 122
 123 #include <gst/net/gstnetaddressmeta.h>
 124
 125 #include <gio/gnetworking.h>
 126
 127 /* Control messages for getting the destination address */
 128 #ifdef IP_PKTINFO
 129 GType gst_ip_pktinfo_message_get_type (void);
 130
 131 #define GST_TYPE_IP_PKTINFO_MESSAGE         (gst_ip_pktinfo_message_get_type ())
 132 #define GST_IP_PKTINFO_MESSAGE(o)           (G_TYPE_CHECK_INSTANCE_CAST ((o), GST_TYPE_IP_PKTINFO_MESSAGE, GstIPPktinfoMessage))
 133 #define GST_IP_PKTINFO_MESSAGE_CLASS(c)     (G_TYPE_CHECK_CLASS_CAST ((c), GST_TYPE_IP_PKTINFO_MESSAGE, GstIPPktinfoMessageClass))
 134 #define GST_IS_IP_PKTINFO_MESSAGE(o)        (G_TYPE_CHECK_INSTANCE_TYPE ((o), GST_TYPE_IP_PKTINFO_MESSAGE))
 135 #define GST_IS_IP_PKTINFO_MESSAGE_CLASS(c)  (G_TYPE_CHECK_CLASS_TYPE ((c), GST_TYPE_IP_PKTINFO_MESSAGE))
 136 #define GST_IP_PKTINFO_MESSAGE_GET_CLASS(o) (G_TYPE_INSTANCE_GET_CLASS ((o), GST_TYPE_IP_PKTINFO_MESSAGE, GstIPPktinfoMessageClass))
 137
 138 typedef struct _GstIPPktinfoMessage GstIPPktinfoMessage;
 139 typedef struct _GstIPPktinfoMessageClass GstIPPktinfoMessageClass;
 140
 141 struct _GstIPPktinfoMessageClass
 142 {
 143   GSocketControlMessageClass parent_class;
 144
 145 };
 146
 147 struct _GstIPPktinfoMessage
 148 {
 149   GSocketControlMessage parent;
 150
 151   guint ifindex;
 152   struct in_addr spec_dst, addr;
 153 };
 154
 155 G_DEFINE_TYPE (GstIPPktinfoMessage, gst_ip_pktinfo_message,
 156     G_TYPE_SOCKET_CONTROL_MESSAGE);
 157
 158 static gsize
 159 gst_ip_pktinfo_message_get_size (GSocketControlMessage * message)
 160 {
 161   return sizeof (struct in_pktinfo);
 162 }
 163
 164 static int
 165 gst_ip_pktinfo_message_get_level (GSocketControlMessage * message)
 166 {
 167   return IPPROTO_IP;
 168 }
 169
 170 static int
 171 gst_ip_pktinfo_message_get_msg_type (GSocketControlMessage * message)
 172 {
 173   return IP_PKTINFO;
 174 }
 175
 176 static GSocketControlMessage *
 177 gst_ip_pktinfo_message_deserialize (gint level,
 178     gint type, gsize size, gpointer data)
 179 {
 180   struct in_pktinfo *pktinfo;
 181   GstIPPktinfoMessage *message;
 182
 183   if (level != IPPROTO_IP || type != IP_PKTINFO)
 184     return NULL;
 185
 186   if (size < sizeof (struct in_pktinfo))
 187     return NULL;
 188
 189   pktinfo = data;
 190
 191   message = g_object_new (GST_TYPE_IP_PKTINFO_MESSAGE, NULL);
 192   message->ifindex = pktinfo->ipi_ifindex;
 193   message->spec_dst = pktinfo->ipi_spec_dst;
 194   message->addr = pktinfo->ipi_addr;
 195
 196   return G_SOCKET_CONTROL_MESSAGE (message);
 197 }
 198
 199 static void
 200 gst_ip_pktinfo_message_init (GstIPPktinfoMessage * message)
 201 {
 202 }
 203
 204 static void
 205 gst_ip_pktinfo_message_class_init (GstIPPktinfoMessageClass * class)
 206 {
 207   GSocketControlMessageClass *scm_class;
 208
 209   scm_class = G_SOCKET_CONTROL_MESSAGE_CLASS (class);
 210   scm_class->get_size = gst_ip_pktinfo_message_get_size;
 211   scm_class->get_level = gst_ip_pktinfo_message_get_level;
 212   scm_class->get_type = gst_ip_pktinfo_message_get_msg_type;
 213   scm_class->deserialize = gst_ip_pktinfo_message_deserialize;
 214 }
 215 #endif
 216
 217 #ifdef IPV6_PKTINFO
 218 GType gst_ipv6_pktinfo_message_get_type (void);
 219
 220 #define GST_TYPE_IPV6_PKTINFO_MESSAGE         (gst_ipv6_pktinfo_message_get_type ())
 221 #define GST_IPV6_PKTINFO_MESSAGE(o)           (G_TYPE_CHECK_INSTANCE_CAST ((o), GST_TYPE_IPV6_PKTINFO_MESSAGE, GstIPV6PktinfoMessage))
 222 #define GST_IPV6_PKTINFO_MESSAGE_CLASS(c)     (G_TYPE_CHECK_CLASS_CAST ((c), GST_TYPE_IPV6_PKTINFO_MESSAGE, GstIPV6PktinfoMessageClass))
 223 #define GST_IS_IPV6_PKTINFO_MESSAGE(o)        (G_TYPE_CHECK_INSTANCE_TYPE ((o), GST_TYPE_IPV6_PKTINFO_MESSAGE))
 224 #define GST_IS_IPV6_PKTINFO_MESSAGE_CLASS(c)  (G_TYPE_CHECK_CLASS_TYPE ((c), GST_TYPE_IPV6_PKTINFO_MESSAGE))
 225 #define GST_IPV6_PKTINFO_MESSAGE_GET_CLASS(o) (G_TYPE_INSTANCE_GET_CLASS ((o), GST_TYPE_IPV6_PKTINFO_MESSAGE, GstIPV6PktinfoMessageClass))
 226
 227 typedef struct _GstIPV6PktinfoMessage GstIPV6PktinfoMessage;
 228 typedef struct _GstIPV6PktinfoMessageClass GstIPV6PktinfoMessageClass;
 229
 230 struct _GstIPV6PktinfoMessageClass
 231 {
 232   GSocketControlMessageClass parent_class;
 233
 234 };
 235
 236 struct _GstIPV6PktinfoMessage
 237 {
 238   GSocketControlMessage parent;
 239
 240   guint ifindex;
 241   struct in6_addr addr;
 242 };
 243
 244 G_DEFINE_TYPE (GstIPV6PktinfoMessage, gst_ipv6_pktinfo_message,
 245     G_TYPE_SOCKET_CONTROL_MESSAGE);
 246
 247 static gsize
 248 gst_ipv6_pktinfo_message_get_size (GSocketControlMessage * message)
 249 {
 250   return sizeof (struct in6_pktinfo);
 251 }
 252
 253 static int
 254 gst_ipv6_pktinfo_message_get_level (GSocketControlMessage * message)
 255 {
 256   return IPPROTO_IPV6;
 257 }
 258
 259 static int
 260 gst_ipv6_pktinfo_message_get_msg_type (GSocketControlMessage * message)
 261 {
 262   return IPV6_PKTINFO;
 263 }
 264
 265 static GSocketControlMessage *
 266 gst_ipv6_pktinfo_message_deserialize (gint level,
 267     gint type, gsize size, gpointer data)
 268 {
 269   struct in6_pktinfo *pktinfo;
 270   GstIPV6PktinfoMessage *message;
 271
 272   if (level != IPPROTO_IPV6 || type != IPV6_PKTINFO)
 273     return NULL;
 274
 275   if (size < sizeof (struct in_pktinfo))
 276     return NULL;
 277
 278   pktinfo = data;
 279
 280   message = g_object_new (GST_TYPE_IPV6_PKTINFO_MESSAGE, NULL);
 281   message->ifindex = pktinfo->ipi6_ifindex;
 282   message->addr = pktinfo->ipi6_addr;
 283
 284   return G_SOCKET_CONTROL_MESSAGE (message);
 285 }
 286
 287 static void
 288 gst_ipv6_pktinfo_message_init (GstIPV6PktinfoMessage * message)
 289 {
 290 }
 291
 292 static void
 293 gst_ipv6_pktinfo_message_class_init (GstIPV6PktinfoMessageClass * class)
 294 {
 295   GSocketControlMessageClass *scm_class;
 296
 297   scm_class = G_SOCKET_CONTROL_MESSAGE_CLASS (class);
 298   scm_class->get_size = gst_ipv6_pktinfo_message_get_size;
 299   scm_class->get_level = gst_ipv6_pktinfo_message_get_level;
 300   scm_class->get_type = gst_ipv6_pktinfo_message_get_msg_type;
 301   scm_class->deserialize = gst_ipv6_pktinfo_message_deserialize;
 302 }
 303
 304 #endif
 305
 306 #ifdef IP_RECVDSTADDR
 307 GType gst_ip_recvdstaddr_message_get_type (void);
 308
 309 #define GST_TYPE_IP_RECVDSTADDR_MESSAGE         (gst_ip_recvdstaddr_message_get_type ())
 310 #define GST_IP_RECVDSTADDR_MESSAGE(o)           (G_TYPE_CHECK_INSTANCE_CAST ((o), GST_TYPE_IP_RECVDSTADDR_MESSAGE, GstIPRecvdstaddrMessage))
 311 #define GST_IP_RECVDSTADDR_MESSAGE_CLASS(c)     (G_TYPE_CHECK_CLASS_CAST ((c), GST_TYPE_IP_RECVDSTADDR_MESSAGE, GstIPRecvdstaddrMessageClass))
 312 #define GST_IS_IP_RECVDSTADDR_MESSAGE(o)        (G_TYPE_CHECK_INSTANCE_TYPE ((o), GST_TYPE_IP_RECVDSTADDR_MESSAGE))
 313 #define GST_IS_IP_RECVDSTADDR_MESSAGE_CLASS(c)  (G_TYPE_CHECK_CLASS_TYPE ((c), GST_TYPE_IP_RECVDSTADDR_MESSAGE))
 314 #define GST_IP_RECVDSTADDR_MESSAGE_GET_CLASS(o) (G_TYPE_INSTANCE_GET_CLASS ((o), GST_TYPE_IP_RECVDSTADDR_MESSAGE, GstIPRecvdstaddrMessageClass))
 315
 316 typedef struct _GstIPRecvdstaddrMessage GstIPRecvdstaddrMessage;
 317 typedef struct _GstIPRecvdstaddrMessageClass GstIPRecvdstaddrMessageClass;
 318
 319 struct _GstIPRecvdstaddrMessageClass
 320 {
 321   GSocketControlMessageClass parent_class;
 322
 323 };
 324
 325 struct _GstIPRecvdstaddrMessage
 326 {
 327   GSocketControlMessage parent;
 328
 329   guint ifindex;
 330   struct in_addr addr;
 331 };
 332
 333 G_DEFINE_TYPE (GstIPRecvdstaddrMessage, gst_ip_recvdstaddr_message,
 334     G_TYPE_SOCKET_CONTROL_MESSAGE);
 335
 336 static gsize
 337 gst_ip_recvdstaddr_message_get_size (GSocketControlMessage * message)
 338 {
 339   return sizeof (struct in_addr);
 340 }
 341
 342 static int
 343 gst_ip_recvdstaddr_message_get_level (GSocketControlMessage * message)
 344 {
 345   return IPPROTO_IP;
 346 }
 347
 348 static int
 349 gst_ip_recvdstaddr_message_get_msg_type (GSocketControlMessage * message)
 350 {
 351   return IP_RECVDSTADDR;
 352 }
 353
 354 static GSocketControlMessage *
 355 gst_ip_recvdstaddr_message_deserialize (gint level,
 356     gint type, gsize size, gpointer data)
 357 {
 358   struct in_addr *addr;
 359   GstIPRecvdstaddrMessage *message;
 360
 361   if (level != IPPROTO_IP || type != IP_RECVDSTADDR)
 362     return NULL;
 363
 364   if (size < sizeof (struct in_addr))
 365     return NULL;
 366
 367   addr = data;
 368
 369   message = g_object_new (GST_TYPE_IP_RECVDSTADDR_MESSAGE, NULL);
 370   message->addr = *addr;
 371
 372   return G_SOCKET_CONTROL_MESSAGE (message);
 373 }
 374
 375 static void
 376 gst_ip_recvdstaddr_message_init (GstIPRecvdstaddrMessage * message)
 377 {
 378 }
 379
 380 static void
 381 gst_ip_recvdstaddr_message_class_init (GstIPRecvdstaddrMessageClass * class)
 382 {
 383   GSocketControlMessageClass *scm_class;
 384
 385   scm_class = G_SOCKET_CONTROL_MESSAGE_CLASS (class);
 386   scm_class->get_size = gst_ip_recvdstaddr_message_get_size;
 387   scm_class->get_level = gst_ip_recvdstaddr_message_get_level;
 388   scm_class->get_type = gst_ip_recvdstaddr_message_get_msg_type;
 389   scm_class->deserialize = gst_ip_recvdstaddr_message_deserialize;
 390 }
 391 #endif
 392
 393 /* not 100% correct, but a good upper bound for memory allocation purposes */
 394 #define MAX_IPV4_UDP_PACKET_SIZE (65536 - 8)
 395
 396 GST_DEBUG_CATEGORY_STATIC (udpsrc_debug);
 397 #define GST_CAT_DEFAULT (udpsrc_debug)
 398
 399 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
 400     GST_PAD_SRC,
 401     GST_PAD_ALWAYS,
 402     GST_STATIC_CAPS_ANY);
 403
 404 #define UDP_DEFAULT_PORT                5004
 405 #define UDP_DEFAULT_MULTICAST_GROUP     "0.0.0.0"
 406 #define UDP_DEFAULT_MULTICAST_IFACE     NULL
 407 #define UDP_DEFAULT_URI                 "udp://"UDP_DEFAULT_MULTICAST_GROUP":"G_STRINGIFY(UDP_DEFAULT_PORT)
 408 #define UDP_DEFAULT_CAPS                NULL
 409 #define UDP_DEFAULT_SOCKET              NULL
 410 #define UDP_DEFAULT_BUFFER_SIZE         0
 411 #define UDP_DEFAULT_TIMEOUT             0
 412 #define UDP_DEFAULT_SKIP_FIRST_BYTES    0
 413 #define UDP_DEFAULT_CLOSE_SOCKET       TRUE
 414 #define UDP_DEFAULT_USED_SOCKET        NULL
 415 #define UDP_DEFAULT_AUTO_MULTICAST     TRUE
 416 #define UDP_DEFAULT_REUSE              TRUE
 417 #define UDP_DEFAULT_LOOP               TRUE
 418 #define UDP_DEFAULT_RETRIEVE_SENDER_ADDRESS TRUE
 419
 420 enum
 421 {
 422   PROP_0,
 423
 424   PROP_PORT,
 425   PROP_MULTICAST_GROUP,
 426   PROP_MULTICAST_IFACE,
 427   PROP_URI,
 428   PROP_CAPS,
 429   PROP_SOCKET,
 430   PROP_BUFFER_SIZE,
 431   PROP_TIMEOUT,
 432   PROP_SKIP_FIRST_BYTES,
 433   PROP_CLOSE_SOCKET,
 434   PROP_USED_SOCKET,
 435   PROP_AUTO_MULTICAST,
 436   PROP_REUSE,
 437   PROP_ADDRESS,
 438   PROP_LOOP,
 439   PROP_RETRIEVE_SENDER_ADDRESS
 440 };
 441
 442 static void gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data);
 443
 444 static GstCaps *gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter);
 445 static GstFlowReturn gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf);
 446 static gboolean gst_udpsrc_close (GstUDPSrc * src);
 447 static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc);
 448 static gboolean gst_udpsrc_unlock_stop (GstBaseSrc * bsrc);
 449 static gboolean gst_udpsrc_negotiate (GstBaseSrc * basesrc);
 450
 451 static void gst_udpsrc_finalize (GObject * object);
 452
 453 static void gst_udpsrc_set_property (GObject * object, guint prop_id,
 454     const GValue * value, GParamSpec * pspec);
 455 static void gst_udpsrc_get_property (GObject * object, guint prop_id,
 456     GValue * value, GParamSpec * pspec);
 457
 458 static GstStateChangeReturn gst_udpsrc_change_state (GstElement * element,
 459     GstStateChange transition);
 460
 461 #define gst_udpsrc_parent_class parent_class
 462 G_DEFINE_TYPE_WITH_CODE (GstUDPSrc, gst_udpsrc, GST_TYPE_PUSH_SRC,
 463     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_udpsrc_uri_handler_init));
 464
 465 static void
 466 gst_udpsrc_class_init (GstUDPSrcClass * klass)
 467 {
 468   GObjectClass *gobject_class;
 469   GstElementClass *gstelement_class;
 470   GstBaseSrcClass *gstbasesrc_class;
 471   GstPushSrcClass *gstpushsrc_class;
 472
 473   gobject_class = (GObjectClass *) klass;
 474   gstelement_class = (GstElementClass *) klass;
 475   gstbasesrc_class = (GstBaseSrcClass *) klass;
 476   gstpushsrc_class = (GstPushSrcClass *) klass;
 477
 478   GST_DEBUG_CATEGORY_INIT (udpsrc_debug, "udpsrc", 0, "UDP src");
 479
 480 #ifdef IP_PKTINFO
 481   GST_TYPE_IP_PKTINFO_MESSAGE;
 482 #endif
 483 #ifdef IPV6_PKTINFO
 484   GST_TYPE_IPV6_PKTINFO_MESSAGE;
 485 #endif
 486 #ifdef IP_RECVDSTADDR
 487   GST_TYPE_IP_RECVDSTADDR_MESSAGE;
 488 #endif
 489
 490   gobject_class->set_property = gst_udpsrc_set_property;
 491   gobject_class->get_property = gst_udpsrc_get_property;
 492   gobject_class->finalize = gst_udpsrc_finalize;
 493
 494   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
 495       g_param_spec_int ("port", "Port",
 496           "The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
 497           UDP_DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 498   /* FIXME 2.0: Remove multicast-group property */
 499 #ifndef GST_REMOVE_DEPRECATED
 500   g_object_class_install_property (gobject_class, PROP_MULTICAST_GROUP,
 501       g_param_spec_string ("multicast-group", "Multicast Group",
 502           "The Address of multicast group to join. (DEPRECATED: "
 503           "Use address property instead)", UDP_DEFAULT_MULTICAST_GROUP,
 504           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_DEPRECATED));
 505 #endif
 506   g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
 507       g_param_spec_string ("multicast-iface", "Multicast Interface",
 508           "The network interface on which to join the multicast group",
 509           UDP_DEFAULT_MULTICAST_IFACE,
 510           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 511   g_object_class_install_property (gobject_class, PROP_URI,
 512       g_param_spec_string ("uri", "URI",
 513           "URI in the form of udp://multicast_group:port", UDP_DEFAULT_URI,
 514           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 515   g_object_class_install_property (gobject_class, PROP_CAPS,
 516       g_param_spec_boxed ("caps", "Caps",
 517           "The caps of the source pad", GST_TYPE_CAPS,
 518           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 519   g_object_class_install_property (gobject_class, PROP_SOCKET,
 520       g_param_spec_object ("socket", "Socket",
 521           "Socket to use for UDP reception. (NULL == allocate)",
 522           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 523   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
 524       g_param_spec_int ("buffer-size", "Buffer Size",
 525           "Size of the kernel receive buffer in bytes, 0=default", 0, G_MAXINT,
 526           UDP_DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 527   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_TIMEOUT,
 528       g_param_spec_uint64 ("timeout", "Timeout",
 529           "Post a message after timeout nanoseconds (0 = disabled)", 0,
 530           G_MAXUINT64, UDP_DEFAULT_TIMEOUT,
 531           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 532   g_object_class_install_property (G_OBJECT_CLASS (klass),
 533       PROP_SKIP_FIRST_BYTES, g_param_spec_int ("skip-first-bytes",
 534           "Skip first bytes", "number of bytes to skip for each udp packet", 0,
 535           G_MAXINT, UDP_DEFAULT_SKIP_FIRST_BYTES,
 536           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 537   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 538       g_param_spec_boolean ("close-socket", "Close socket",
 539           "Close socket if passed as property on state change",
 540           UDP_DEFAULT_CLOSE_SOCKET,
 541           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 542   g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
 543       g_param_spec_object ("used-socket", "Socket Handle",
 544           "Socket currently in use for UDP reception. (NULL = no socket)",
 545           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 546   g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
 547       g_param_spec_boolean ("auto-multicast", "Auto Multicast",
 548           "Automatically join/leave multicast groups",
 549           UDP_DEFAULT_AUTO_MULTICAST,
 550           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 551   g_object_class_install_property (gobject_class, PROP_REUSE,
 552       g_param_spec_boolean ("reuse", "Reuse", "Enable reuse of the port",
 553           UDP_DEFAULT_REUSE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 554   g_object_class_install_property (gobject_class, PROP_ADDRESS,
 555       g_param_spec_string ("address", "Address",
 556           "Address to receive packets for. This is equivalent to the "
 557           "multicast-group property for now", UDP_DEFAULT_MULTICAST_GROUP,
 558           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 559   /**
 560    * GstUDPSrc::loop:
 561    *
 562    * Can be used to disable multicast loopback.
 563    *
 564    * Since: 1.8
 565    */
 566   g_object_class_install_property (gobject_class, PROP_LOOP,
 567       g_param_spec_boolean ("loop", "Multicast Loopback",
 568           "Used for setting the multicast loop parameter. TRUE = enable,"
 569           " FALSE = disable", UDP_DEFAULT_LOOP,
 570           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 571   /**
 572    * GstUDPSrc::retrieve-sender-address:
 573    *
 574    * Whether to retrieve the sender address and add it to the buffers as
 575    * meta. Disabling this might result in minor performance improvements
 576    * in certain scenarios.
 577    *
 578    * Since: 1.10
 579    */
 580   g_object_class_install_property (gobject_class, PROP_RETRIEVE_SENDER_ADDRESS,
 581       g_param_spec_boolean ("retrieve-sender-address",
 582           "Retrieve Sender Address",
 583           "Whether to retrieve the sender address and add it to buffers as "
 584           "meta. Disabling this might result in minor performance improvements "
 585           "in certain scenarios", UDP_DEFAULT_RETRIEVE_SENDER_ADDRESS,
 586           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 587
 588   gst_element_class_add_static_pad_template (gstelement_class, &src_template);
 589
 590   gst_element_class_set_static_metadata (gstelement_class,
 591       "UDP packet receiver", "Source/Network",
 592       "Receive data over the network via UDP",
 593       "Wim Taymans <wim@fluendo.com>, "
 594       "Thijs Vermeir <thijs.vermeir@barco.com>");
 595
 596   gstelement_class->change_state = gst_udpsrc_change_state;
 597
 598   gstbasesrc_class->unlock = gst_udpsrc_unlock;
 599   gstbasesrc_class->unlock_stop = gst_udpsrc_unlock_stop;
 600   gstbasesrc_class->get_caps = gst_udpsrc_getcaps;
 601   gstbasesrc_class->negotiate = gst_udpsrc_negotiate;
 602
 603   gstpushsrc_class->create = gst_udpsrc_create;
 604 }
 605
 606 static void
 607 gst_udpsrc_init (GstUDPSrc * udpsrc)
 608 {
 609   udpsrc->uri =
 610       g_strdup_printf ("udp://%s:%u", UDP_DEFAULT_MULTICAST_GROUP,
 611       UDP_DEFAULT_PORT);
 612
 613   udpsrc->address = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 614   udpsrc->port = UDP_DEFAULT_PORT;
 615   udpsrc->socket = UDP_DEFAULT_SOCKET;
 616   udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 617   udpsrc->buffer_size = UDP_DEFAULT_BUFFER_SIZE;
 618   udpsrc->timeout = UDP_DEFAULT_TIMEOUT;
 619   udpsrc->skip_first_bytes = UDP_DEFAULT_SKIP_FIRST_BYTES;
 620   udpsrc->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
 621   udpsrc->external_socket = (udpsrc->socket != NULL);
 622   udpsrc->auto_multicast = UDP_DEFAULT_AUTO_MULTICAST;
 623   udpsrc->used_socket = UDP_DEFAULT_USED_SOCKET;
 624   udpsrc->reuse = UDP_DEFAULT_REUSE;
 625   udpsrc->loop = UDP_DEFAULT_LOOP;
 626   udpsrc->retrieve_sender_address = UDP_DEFAULT_RETRIEVE_SENDER_ADDRESS;
 627
 628   /* configure basesrc to be a live source */
 629   gst_base_src_set_live (GST_BASE_SRC (udpsrc), TRUE);
 630   /* make basesrc output a segment in time */
 631   gst_base_src_set_format (GST_BASE_SRC (udpsrc), GST_FORMAT_TIME);
 632   /* make basesrc set timestamps on outgoing buffers based on the running_time
 633    * when they were captured */
 634   gst_base_src_set_do_timestamp (GST_BASE_SRC (udpsrc), TRUE);
 635 }
 636
 637 static void
 638 gst_udpsrc_finalize (GObject * object)
 639 {
 640   GstUDPSrc *udpsrc;
 641
 642   udpsrc = GST_UDPSRC (object);
 643
 644   if (udpsrc->caps)
 645     gst_caps_unref (udpsrc->caps);
 646   udpsrc->caps = NULL;
 647
 648   g_free (udpsrc->multi_iface);
 649   udpsrc->multi_iface = NULL;
 650
 651   g_free (udpsrc->uri);
 652   udpsrc->uri = NULL;
 653
 654   g_free (udpsrc->address);
 655   udpsrc->address = NULL;
 656
 657   if (udpsrc->socket)
 658     g_object_unref (udpsrc->socket);
 659   udpsrc->socket = NULL;
 660
 661   if (udpsrc->used_socket)
 662     g_object_unref (udpsrc->used_socket);
 663   udpsrc->used_socket = NULL;
 664
 665   G_OBJECT_CLASS (parent_class)->finalize (object);
 666 }
 667
 668 static GstCaps *
 669 gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter)
 670 {
 671   GstUDPSrc *udpsrc;
 672   GstCaps *caps, *result;
 673
 674   udpsrc = GST_UDPSRC (src);
 675
 676   GST_OBJECT_LOCK (src);
 677   if ((caps = udpsrc->caps))
 678     gst_caps_ref (caps);
 679   GST_OBJECT_UNLOCK (src);
 680
 681   if (caps) {
 682     if (filter) {
 683       result = gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
 684       gst_caps_unref (caps);
 685     } else {
 686       result = caps;
 687     }
 688   } else {
 689     result = (filter) ? gst_caps_ref (filter) : gst_caps_new_any ();
 690   }
 691   return result;
 692 }
 693
 694 static void
 695 gst_udpsrc_reset_memory_allocator (GstUDPSrc * src)
 696 {
 697   if (src->mem != NULL) {
 698     gst_memory_unmap (src->mem, &src->map);
 699     gst_memory_unref (src->mem);
 700     src->mem = NULL;
 701   }
 702   if (src->mem_max != NULL) {
 703     gst_memory_unmap (src->mem_max, &src->map_max);
 704     gst_memory_unref (src->mem_max);
 705     src->mem_max = NULL;
 706   }
 707
 708   src->vec[0].buffer = NULL;
 709   src->vec[0].size = 0;
 710   src->vec[1].buffer = NULL;
 711   src->vec[1].size = 0;
 712
 713   if (src->allocator != NULL) {
 714     gst_object_unref (src->allocator);
 715     src->allocator = NULL;
 716   }
 717 }
 718
 719 static gboolean
 720 gst_udpsrc_negotiate (GstBaseSrc * basesrc)
 721 {
 722   GstUDPSrc *src = GST_UDPSRC_CAST (basesrc);
 723   gboolean ret;
 724
 725   /* just chain up to the default implementation, we just want to
 726    * retrieve the allocator at the end of it (if there is one) */
 727   ret = GST_BASE_SRC_CLASS (parent_class)->negotiate (basesrc);
 728
 729   if (ret) {
 730     GstAllocationParams new_params;
 731     GstAllocator *new_allocator = NULL;
 732
 733     /* retrieve new allocator */
 734     gst_base_src_get_allocator (basesrc, &new_allocator, &new_params);
 735
 736     if (src->allocator != new_allocator ||
 737         memcmp (&src->params, &new_params, sizeof (GstAllocationParams)) != 0) {
 738       /* drop old allocator and throw away any memory allocated with it */
 739       gst_udpsrc_reset_memory_allocator (src);
 740
 741       /* and save the new allocator and/or new allocation parameters */
 742       src->allocator = new_allocator;
 743       src->params = new_params;
 744
 745       GST_INFO_OBJECT (src, "new allocator: %" GST_PTR_FORMAT, new_allocator);
 746     }
 747   }
 748
 749   return ret;
 750 }
 751
 752 static gboolean
 753 gst_udpsrc_alloc_mem (GstUDPSrc * src, GstMemory ** p_mem, GstMapInfo * map,
 754     gsize size)
 755 {
 756   GstMemory *mem;
 757
 758   mem = gst_allocator_alloc (src->allocator, size, &src->params);
 759
 760   if (!gst_memory_map (mem, map, GST_MAP_WRITE)) {
 761     gst_memory_unref (mem);
 762     memset (map, 0, sizeof (GstMapInfo));
 763     return FALSE;
 764   }
 765   *p_mem = mem;
 766   return TRUE;
 767 }
 768
 769 static gboolean
 770 gst_udpsrc_ensure_mem (GstUDPSrc * src)
 771 {
 772   if (src->mem == NULL) {
 773     gsize mem_size = 1500;      /* typical max. MTU */
 774
 775     /* if packets are likely to be smaller, just use that size, otherwise
 776      * default to assuming incoming packets are around MTU size */
 777     if (src->max_size > 0 && src->max_size < mem_size)
 778       mem_size = src->max_size;
 779
 780     if (!gst_udpsrc_alloc_mem (src, &src->mem, &src->map, mem_size))
 781       return FALSE;
 782
 783     src->vec[0].buffer = src->map.data;
 784     src->vec[0].size = src->map.size;
 785   }
 786
 787   if (src->mem_max == NULL) {
 788     gsize max_size = MAX_IPV4_UDP_PACKET_SIZE;
 789
 790     if (!gst_udpsrc_alloc_mem (src, &src->mem_max, &src->map_max, max_size))
 791       return FALSE;
 792
 793     src->vec[1].buffer = src->map_max.data;
 794     src->vec[1].size = src->map_max.size;
 795   }
 796
 797   return TRUE;
 798 }
 799
 800 static void
 801 gst_udpsrc_create_cancellable (GstUDPSrc * src)
 802 {
 803   GPollFD pollfd;
 804
 805   src->cancellable = g_cancellable_new ();
 806   src->made_cancel_fd = g_cancellable_make_pollfd (src->cancellable, &pollfd);
 807 }
 808
 809 static void
 810 gst_udpsrc_free_cancellable (GstUDPSrc * src)
 811 {
 812   if (src->made_cancel_fd) {
 813     g_cancellable_release_fd (src->cancellable);
 814     src->made_cancel_fd = FALSE;
 815   }
 816   g_object_unref (src->cancellable);
 817   src->cancellable = NULL;
 818 }
 819
 820 static GstFlowReturn
 821 gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf)
 822 {
 823   GstUDPSrc *udpsrc;
 824   GstBuffer *outbuf = NULL;
 825   GSocketAddress *saddr = NULL;
 826   GSocketAddress **p_saddr;
 827   gint flags = G_SOCKET_MSG_NONE;
 828   gboolean try_again;
 829   GError *err = NULL;
 830   gssize res;
 831   gsize offset;
 832   GSocketControlMessage **msgs = NULL;
 833   gint n_msgs = 0, i;
 834
 835   udpsrc = GST_UDPSRC_CAST (psrc);
 836
 837   if (!gst_udpsrc_ensure_mem (udpsrc))
 838     goto memory_alloc_error;
 839
 840   /* Retrieve sender address unless we've been configured not to do so */
 841   p_saddr = (udpsrc->retrieve_sender_address) ? &saddr : NULL;
 842
 843 retry:
 844
 845   do {
 846     gint64 timeout;
 847
 848     try_again = FALSE;
 849
 850     if (udpsrc->timeout)
 851       timeout = udpsrc->timeout / 1000;
 852     else
 853       timeout = -1;
 854
 855     GST_LOG_OBJECT (udpsrc, "doing select, timeout %" G_GINT64_FORMAT, timeout);
 856
 857     if (!g_socket_condition_timed_wait (udpsrc->used_socket, G_IO_IN | G_IO_PRI,
 858             timeout, udpsrc->cancellable, &err)) {
 859       if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY)
 860           || g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 861         goto stopped;
 862       } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
 863         g_clear_error (&err);
 864         /* timeout, post element message */
 865         gst_element_post_message (GST_ELEMENT_CAST (udpsrc),
 866             gst_message_new_element (GST_OBJECT_CAST (udpsrc),
 867                 gst_structure_new ("GstUDPSrcTimeout",
 868                     "timeout", G_TYPE_UINT64, udpsrc->timeout, NULL)));
 869       } else {
 870         goto select_error;
 871       }
 872
 873       try_again = TRUE;
 874     }
 875   } while (G_UNLIKELY (try_again));
 876
 877   if (saddr != NULL) {
 878     g_object_unref (saddr);
 879     saddr = NULL;
 880   }
 881
 882   res =
 883       g_socket_receive_message (udpsrc->used_socket, p_saddr, udpsrc->vec, 2,
 884       &msgs, &n_msgs, &flags, udpsrc->cancellable, &err);
 885
 886   if (G_UNLIKELY (res < 0)) {
 887     /* G_IO_ERROR_HOST_UNREACHABLE for a UDP socket means that a packet sent
 888      * with udpsink generated a "port unreachable" ICMP response. We ignore
 889      * that and try again.
 890      * On Windows we get G_IO_ERROR_CONNECTION_CLOSED instead */
 891 #if GLIB_CHECK_VERSION(2,44,0)
 892     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_HOST_UNREACHABLE) ||
 893         g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CONNECTION_CLOSED)) {
 894 #else
 895     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_HOST_UNREACHABLE)) {
 896 #endif
 897       g_clear_error (&err);
 898       goto retry;
 899     }
 900     goto receive_error;
 901   }
 902
 903   /* remember maximum packet size */
 904   if (res > udpsrc->max_size)
 905     udpsrc->max_size = res;
 906
 907   /* Retry if multicast and the destination address is not ours. We don't want
 908    * to receive arbitrary packets */
 909   {
 910     GInetAddress *iaddr = g_inet_socket_address_get_address (udpsrc->addr);
 911     gboolean skip_packet = FALSE;
 912     gsize iaddr_size = g_inet_address_get_native_size (iaddr);
 913     const guint8 *iaddr_bytes = g_inet_address_to_bytes (iaddr);
 914
 915     if (g_inet_address_get_is_multicast (iaddr)) {
 916
 917       for (i = 0; i < n_msgs && !skip_packet; i++) {
 918 #ifdef IP_PKTINFO
 919         if (GST_IS_IP_PKTINFO_MESSAGE (msgs[i])) {
 920           GstIPPktinfoMessage *msg = GST_IP_PKTINFO_MESSAGE (msgs[i]);
 921
 922           if (sizeof (msg->addr) == iaddr_size
 923               && memcmp (iaddr_bytes, &msg->addr, sizeof (msg->addr)))
 924             skip_packet = TRUE;
 925         }
 926 #endif
 927 #ifdef IPV6_PKTINFO
 928         if (GST_IS_IPV6_PKTINFO_MESSAGE (msgs[i])) {
 929           GstIPV6PktinfoMessage *msg = GST_IPV6_PKTINFO_MESSAGE (msgs[i]);
 930
 931           if (sizeof (msg->addr) == iaddr_size
 932               && memcmp (iaddr_bytes, &msg->addr, sizeof (msg->addr)))
 933             skip_packet = TRUE;
 934         }
 935 #endif
 936 #ifdef IP_RECVDSTADDR
 937         if (GST_IS_IP_RECVDSTADDR_MESSAGE (msgs[i])) {
 938           GstIPRecvdstaddrMessage *msg = GST_IP_RECVDSTADDR_MESSAGE (msgs[i]);
 939
 940           if (sizeof (msg->addr) == iaddr_size
 941               && memcmp (iaddr_bytes, &msg->addr, sizeof (msg->addr)))
 942             skip_packet = TRUE;
 943         }
 944 #endif
 945       }
 946
 947     }
 948
 949     for (i = 0; i < n_msgs; i++) {
 950       g_object_unref (msgs[i]);
 951     }
 952     g_free (msgs);
 953
 954     if (skip_packet) {
 955       GST_DEBUG_OBJECT (udpsrc,
 956           "Dropping packet for a different multicast address");
 957
 958       if (saddr != NULL) {
 959         g_object_unref (saddr);
 960         saddr = NULL;
 961       }
 962
 963       goto retry;
 964     }
 965   }
 966
 967   outbuf = gst_buffer_new ();
 968
 969   /* append first memory chunk to buffer */
 970   gst_buffer_append_memory (outbuf, udpsrc->mem);
 971
 972   /* if the packet didn't fit into the first chunk, add second one as well */
 973   if (res > udpsrc->map.size) {
 974     gst_buffer_append_memory (outbuf, udpsrc->mem_max);
 975     gst_memory_unmap (udpsrc->mem_max, &udpsrc->map_max);
 976     udpsrc->vec[1].buffer = NULL;
 977     udpsrc->vec[1].size = 0;
 978     udpsrc->mem_max = NULL;
 979   }
 980
 981   /* make sure we allocate a new chunk next time (we do this only here because
 982    * we look at map.size to see if the second memory chunk is needed above) */
 983   gst_memory_unmap (udpsrc->mem, &udpsrc->map);
 984   udpsrc->vec[0].buffer = NULL;
 985   udpsrc->vec[0].size = 0;
 986   udpsrc->mem = NULL;
 987
 988   offset = udpsrc->skip_first_bytes;
 989
 990   if (G_UNLIKELY (offset > 0 && res < offset))
 991     goto skip_error;
 992
 993   gst_buffer_resize (outbuf, offset, res - offset);
 994
 995   /* use buffer metadata so receivers can also track the address */
 996   if (saddr) {
 997     gst_buffer_add_net_address_meta (outbuf, saddr);
 998     g_object_unref (saddr);
 999     saddr = NULL;
1000   }
1001
1002   GST_LOG_OBJECT (udpsrc, "read packet of %d bytes", (int) res);
1003
1004   *buf = GST_BUFFER_CAST (outbuf);
1005
1006   return GST_FLOW_OK;
1007
1008   /* ERRORS */
1009 memory_alloc_error:
1010   {
1011     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
1012         ("Failed to allocate or map memory"));
1013     return GST_FLOW_ERROR;
1014   }
1015 select_error:
1016   {
1017     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
1018         ("select error: %s", err->message));
1019     g_clear_error (&err);
1020     return GST_FLOW_ERROR;
1021   }
1022 stopped:
1023   {
1024     GST_DEBUG ("stop called");
1025     g_clear_error (&err);
1026     return GST_FLOW_FLUSHING;
1027   }
1028 receive_error:
1029   {
1030     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY) ||
1031         g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
1032       g_clear_error (&err);
1033       return GST_FLOW_FLUSHING;
1034     } else {
1035       GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
1036           ("receive error %" G_GSSIZE_FORMAT ": %s", res, err->message));
1037       g_clear_error (&err);
1038       return GST_FLOW_ERROR;
1039     }
1040   }
1041 skip_error:
1042   {
1043     gst_buffer_unref (outbuf);
1044
1045     GST_ELEMENT_ERROR (udpsrc, STREAM, DECODE, (NULL),
1046         ("UDP buffer to small to skip header"));
1047     return GST_FLOW_ERROR;
1048   }
1049 }
1050
1051 static gboolean
1052 gst_udpsrc_set_uri (GstUDPSrc * src, const gchar * uri, GError ** error)
1053 {
1054   gchar *address;
1055   guint16 port;
1056
1057   if (!gst_udp_parse_uri (uri, &address, &port))
1058     goto wrong_uri;
1059
1060   if (port == (guint16) - 1)
1061     port = UDP_DEFAULT_PORT;
1062
1063   g_free (src->address);
1064   src->address = address;
1065   src->port = port;
1066
1067   g_free (src->uri);
1068   src->uri = g_strdup (uri);
1069
1070   return TRUE;
1071
1072   /* ERRORS */
1073 wrong_uri:
1074   {
1075     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
1076         ("error parsing uri %s", uri));
1077     g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
1078         "Could not parse UDP URI");
1079     return FALSE;
1080   }
1081 }
1082
1083 static void
1084 gst_udpsrc_set_property (GObject * object, guint prop_id, const GValue * value,
1085     GParamSpec * pspec)
1086 {
1087   GstUDPSrc *udpsrc = GST_UDPSRC (object);
1088
1089   switch (prop_id) {
1090     case PROP_BUFFER_SIZE:
1091       udpsrc->buffer_size = g_value_get_int (value);
1092       break;
1093     case PROP_PORT:
1094       udpsrc->port = g_value_get_int (value);
1095       g_free (udpsrc->uri);
1096       udpsrc->uri =
1097           g_strdup_printf ("udp://%s:%u", udpsrc->address, udpsrc->port);
1098       break;
1099     case PROP_MULTICAST_GROUP:
1100     case PROP_ADDRESS:
1101     {
1102       const gchar *group;
1103
1104       g_free (udpsrc->address);
1105       if ((group = g_value_get_string (value)))
1106         udpsrc->address = g_strdup (group);
1107       else
1108         udpsrc->address = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
1109
1110       g_free (udpsrc->uri);
1111       udpsrc->uri =
1112           g_strdup_printf ("udp://%s:%u", udpsrc->address, udpsrc->port);
1113       break;
1114     }
1115     case PROP_MULTICAST_IFACE:
1116       g_free (udpsrc->multi_iface);
1117
1118       if (g_value_get_string (value) == NULL)
1119         udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
1120       else
1121         udpsrc->multi_iface = g_value_dup_string (value);
1122       break;
1123     case PROP_URI:
1124       gst_udpsrc_set_uri (udpsrc, g_value_get_string (value), NULL);
1125       break;
1126     case PROP_CAPS:
1127     {
1128       const GstCaps *new_caps_val = gst_value_get_caps (value);
1129       GstCaps *new_caps;
1130       GstCaps *old_caps;
1131
1132       if (new_caps_val == NULL) {
1133         new_caps = gst_caps_new_any ();
1134       } else {
1135         new_caps = gst_caps_copy (new_caps_val);
1136       }
1137
1138       GST_OBJECT_LOCK (udpsrc);
1139       old_caps = udpsrc->caps;
1140       udpsrc->caps = new_caps;
1141       GST_OBJECT_UNLOCK (udpsrc);
1142       if (old_caps)
1143         gst_caps_unref (old_caps);
1144
1145       gst_pad_mark_reconfigure (GST_BASE_SRC_PAD (udpsrc));
1146       break;
1147     }
1148     case PROP_SOCKET:
1149       if (udpsrc->socket != NULL && udpsrc->socket != udpsrc->used_socket &&
1150           udpsrc->close_socket) {
1151         GError *err = NULL;
1152
1153         if (!g_socket_close (udpsrc->socket, &err)) {
1154           GST_ERROR ("failed to close socket %p: %s", udpsrc->socket,
1155               err->message);
1156           g_clear_error (&err);
1157         }
1158       }
1159       if (udpsrc->socket)
1160         g_object_unref (udpsrc->socket);
1161       udpsrc->socket = g_value_dup_object (value);
1162       GST_DEBUG ("setting socket to %p", udpsrc->socket);
1163       break;
1164     case PROP_TIMEOUT:
1165       udpsrc->timeout = g_value_get_uint64 (value);
1166       break;
1167     case PROP_SKIP_FIRST_BYTES:
1168       udpsrc->skip_first_bytes = g_value_get_int (value);
1169       break;
1170     case PROP_CLOSE_SOCKET:
1171       udpsrc->close_socket = g_value_get_boolean (value);
1172       break;
1173     case PROP_AUTO_MULTICAST:
1174       udpsrc->auto_multicast = g_value_get_boolean (value);
1175       break;
1176     case PROP_REUSE:
1177       udpsrc->reuse = g_value_get_boolean (value);
1178       break;
1179     case PROP_LOOP:
1180       udpsrc->loop = g_value_get_boolean (value);
1181       break;
1182     case PROP_RETRIEVE_SENDER_ADDRESS:
1183       udpsrc->retrieve_sender_address = g_value_get_boolean (value);
1184       break;
1185     default:
1186       break;
1187   }
1188 }
1189
1190 static void
1191 gst_udpsrc_get_property (GObject * object, guint prop_id, GValue * value,
1192     GParamSpec * pspec)
1193 {
1194   GstUDPSrc *udpsrc = GST_UDPSRC (object);
1195
1196   switch (prop_id) {
1197     case PROP_BUFFER_SIZE:
1198       g_value_set_int (value, udpsrc->buffer_size);
1199       break;
1200     case PROP_PORT:
1201       g_value_set_int (value, udpsrc->port);
1202       break;
1203     case PROP_MULTICAST_GROUP:
1204     case PROP_ADDRESS:
1205       g_value_set_string (value, udpsrc->address);
1206       break;
1207     case PROP_MULTICAST_IFACE:
1208       g_value_set_string (value, udpsrc->multi_iface);
1209       break;
1210     case PROP_URI:
1211       g_value_set_string (value, udpsrc->uri);
1212       break;
1213     case PROP_CAPS:
1214       GST_OBJECT_LOCK (udpsrc);
1215       gst_value_set_caps (value, udpsrc->caps);
1216       GST_OBJECT_UNLOCK (udpsrc);
1217       break;
1218     case PROP_SOCKET:
1219       g_value_set_object (value, udpsrc->socket);
1220       break;
1221     case PROP_TIMEOUT:
1222       g_value_set_uint64 (value, udpsrc->timeout);
1223       break;
1224     case PROP_SKIP_FIRST_BYTES:
1225       g_value_set_int (value, udpsrc->skip_first_bytes);
1226       break;
1227     case PROP_CLOSE_SOCKET:
1228       g_value_set_boolean (value, udpsrc->close_socket);
1229       break;
1230     case PROP_USED_SOCKET:
1231       g_value_set_object (value, udpsrc->used_socket);
1232       break;
1233     case PROP_AUTO_MULTICAST:
1234       g_value_set_boolean (value, udpsrc->auto_multicast);
1235       break;
1236     case PROP_REUSE:
1237       g_value_set_boolean (value, udpsrc->reuse);
1238       break;
1239     case PROP_LOOP:
1240       g_value_set_boolean (value, udpsrc->loop);
1241       break;
1242     case PROP_RETRIEVE_SENDER_ADDRESS:
1243       g_value_set_boolean (value, udpsrc->retrieve_sender_address);
1244       break;
1245     default:
1246       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1247       break;
1248   }
1249 }
1250
1251 static GInetAddress *
1252 gst_udpsrc_resolve (GstUDPSrc * src, const gchar * address)
1253 {
1254   GInetAddress *addr;
1255   GError *err = NULL;
1256   GResolver *resolver;
1257
1258   addr = g_inet_address_new_from_string (address);
1259   if (!addr) {
1260     GList *results;
1261
1262     GST_DEBUG_OBJECT (src, "resolving IP address for host %s", address);
1263     resolver = g_resolver_get_default ();
1264     results =
1265         g_resolver_lookup_by_name (resolver, address, src->cancellable, &err);
1266     if (!results)
1267       goto name_resolve;
1268     addr = G_INET_ADDRESS (g_object_ref (results->data));
1269
1270     g_resolver_free_addresses (results);
1271     g_object_unref (resolver);
1272   }
1273 #ifndef GST_DISABLE_GST_DEBUG
1274   {
1275     gchar *ip = g_inet_address_to_string (addr);
1276
1277     GST_DEBUG_OBJECT (src, "IP address for host %s is %s", address, ip);
1278     g_free (ip);
1279   }
1280 #endif
1281
1282   return addr;
1283
1284 name_resolve:
1285   {
1286     GST_WARNING_OBJECT (src, "Failed to resolve %s: %s", address, err->message);
1287     g_clear_error (&err);
1288     g_object_unref (resolver);
1289     return NULL;
1290   }
1291 }
1292
1293 /* create a socket for sending to remote machine */
1294 static gboolean
1295 gst_udpsrc_open (GstUDPSrc * src)
1296 {
1297   GInetAddress *addr, *bind_addr;
1298   GSocketAddress *bind_saddr;
1299   GError *err = NULL;
1300
1301   gst_udpsrc_create_cancellable (src);
1302
1303   if (src->socket == NULL) {
1304     /* need to allocate a socket */
1305     GST_DEBUG_OBJECT (src, "allocating socket for %s:%d", src->address,
1306         src->port);
1307
1308     addr = gst_udpsrc_resolve (src, src->address);
1309     if (!addr)
1310       goto name_resolve;
1311
1312     if ((src->used_socket =
1313             g_socket_new (g_inet_address_get_family (addr),
1314                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
1315       goto no_socket;
1316
1317     src->external_socket = FALSE;
1318
1319     GST_DEBUG_OBJECT (src, "got socket %p", src->used_socket);
1320
1321     if (src->addr)
1322       g_object_unref (src->addr);
1323     src->addr =
1324         G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
1325
1326     GST_DEBUG_OBJECT (src, "binding on port %d", src->port);
1327
1328     /* For multicast, bind to ANY and join the multicast group later */
1329     if (g_inet_address_get_is_multicast (addr))
1330       bind_addr = g_inet_address_new_any (g_inet_address_get_family (addr));
1331     else
1332       bind_addr = G_INET_ADDRESS (g_object_ref (addr));
1333
1334     g_object_unref (addr);
1335
1336     bind_saddr = g_inet_socket_address_new (bind_addr, src->port);
1337     g_object_unref (bind_addr);
1338     if (!g_socket_bind (src->used_socket, bind_saddr, src->reuse, &err))
1339       goto bind_error;
1340
1341     g_object_unref (bind_saddr);
1342     g_socket_set_multicast_loopback (src->used_socket, src->loop);
1343   } else {
1344     GInetSocketAddress *local_addr;
1345
1346     GST_DEBUG_OBJECT (src, "using provided socket %p", src->socket);
1347     /* we use the configured socket, try to get some info about it */
1348     src->used_socket = G_SOCKET (g_object_ref (src->socket));
1349     src->external_socket = TRUE;
1350
1351     local_addr =
1352         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
1353             &err));
1354     if (!local_addr)
1355       goto getsockname_error;
1356
1357     addr = gst_udpsrc_resolve (src, src->address);
1358     if (!addr)
1359       goto name_resolve;
1360
1361     /* If bound to ANY and address points to a multicast address, make
1362      * sure that address is not overridden with ANY but we have the
1363      * opportunity later to join the multicast address. This ensures that we
1364      * have the same behaviour as for sockets created by udpsrc */
1365     if (!src->auto_multicast ||
1366         !g_inet_address_get_is_any (g_inet_socket_address_get_address
1367             (local_addr))
1368         || !g_inet_address_get_is_multicast (addr)) {
1369       g_object_unref (addr);
1370       if (src->addr)
1371         g_object_unref (src->addr);
1372       src->addr = local_addr;
1373     } else {
1374       g_object_unref (local_addr);
1375       if (src->addr)
1376         g_object_unref (src->addr);
1377       src->addr =
1378           G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
1379       g_object_unref (addr);
1380     }
1381   }
1382
1383   {
1384     gint val = 0;
1385
1386     if (src->buffer_size != 0) {
1387       GError *opt_err = NULL;
1388
1389       GST_INFO_OBJECT (src, "setting udp buffer of %d bytes", src->buffer_size);
1390       /* set buffer size, Note that on Linux this is typically limited to a
1391        * maximum of around 100K. Also a minimum of 128 bytes is required on
1392        * Linux. */
1393       if (!g_socket_set_option (src->used_socket, SOL_SOCKET, SO_RCVBUF,
1394               src->buffer_size, &opt_err)) {
1395         GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
1396             ("Could not create a buffer of requested %d bytes: %s",
1397                 src->buffer_size, opt_err->message));
1398         g_error_free (opt_err);
1399         opt_err = NULL;
1400       }
1401     }
1402
1403     /* read the value of the receive buffer. Note that on linux this returns
1404      * 2x the value we set because the kernel allocates extra memory for
1405      * metadata. The default on Linux is about 100K (which is about 50K
1406      * without metadata) */
1407     if (g_socket_get_option (src->used_socket, SOL_SOCKET, SO_RCVBUF, &val,
1408             NULL)) {
1409       GST_INFO_OBJECT (src, "have udp buffer of %d bytes", val);
1410     } else {
1411       GST_DEBUG_OBJECT (src, "could not get udp buffer size");
1412     }
1413   }
1414
1415   g_socket_set_broadcast (src->used_socket, TRUE);
1416
1417   if (src->auto_multicast
1418       &&
1419       g_inet_address_get_is_multicast (g_inet_socket_address_get_address
1420           (src->addr))) {
1421     GST_DEBUG_OBJECT (src, "joining multicast group %s", src->address);
1422     if (!g_socket_join_multicast_group (src->used_socket,
1423             g_inet_socket_address_get_address (src->addr),
1424             FALSE, src->multi_iface, &err))
1425       goto membership;
1426
1427     if (g_inet_address_get_family (g_inet_socket_address_get_address
1428             (src->addr)) == G_SOCKET_FAMILY_IPV4) {
1429 #if defined(IP_PKTINFO)
1430       if (!g_socket_set_option (src->used_socket, IPPROTO_IP, IP_PKTINFO, TRUE,
1431               &err)) {
1432         GST_WARNING_OBJECT (src, "Failed to enable IP_PKTINFO: %s",
1433             err->message);
1434         g_clear_error (&err);
1435       }
1436 #elif defined(IP_RECVDSTADDR)
1437       if (!g_socket_set_option (src->used_socket, IPPROTO_IP, IP_RECVDSTADDR,
1438               TRUE, &err)) {
1439         GST_WARNING_OBJECT (src, "Failed to enable IP_RECVDSTADDR: %s",
1440             err->message);
1441         g_clear_error (&err);
1442       }
1443 #else
1444 #pragma message("No API available for getting IPv4 destination address")
1445       GST_WARNING_OBJECT (src, "No API available for getting IPv4 destination "
1446           "address, will receive packets for every destination to our port");
1447 #endif
1448     } else
1449         if (g_inet_address_get_family (g_inet_socket_address_get_address
1450             (src->addr)) == G_SOCKET_FAMILY_IPV6) {
1451 #ifdef IPV6_PKTINFO
1452 #ifdef IPV6_RECVPKTINFO
1453       if (!g_socket_set_option (src->used_socket, IPPROTO_IPV6,
1454               IPV6_RECVPKTINFO, TRUE, &err)) {
1455 #else
1456       if (!g_socket_set_option (src->used_socket, IPPROTO_IPV6, IPV6_PKTINFO,
1457               TRUE, &err)) {
1458 #endif
1459         GST_WARNING_OBJECT (src, "Failed to enable IPV6_PKTINFO: %s",
1460             err->message);
1461         g_clear_error (&err);
1462       }
1463 #else
1464 #pragma message("No API available for getting IPv6 destination address")
1465       GST_WARNING_OBJECT (src, "No API available for getting IPv6 destination "
1466           "address, will receive packets for every destination to our port");
1467 #endif
1468     }
1469   }
1470
1471   /* NOTE: sockaddr_in.sin_port works for ipv4 and ipv6 because sin_port
1472    * follows ss_family on both */
1473   {
1474     GInetSocketAddress *addr;
1475     guint16 port;
1476
1477     addr =
1478         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
1479             &err));
1480     if (!addr)
1481       goto getsockname_error;
1482
1483     port = g_inet_socket_address_get_port (addr);
1484     GST_DEBUG_OBJECT (src, "bound, on port %d", port);
1485     if (port != src->port) {
1486       src->port = port;
1487       GST_DEBUG_OBJECT (src, "notifying port %d", port);
1488       g_object_notify (G_OBJECT (src), "port");
1489     }
1490     g_object_unref (addr);
1491   }
1492
1493   src->allocator = NULL;
1494   gst_allocation_params_init (&src->params);
1495
1496   src->max_size = 0;
1497
1498   return TRUE;
1499
1500   /* ERRORS */
1501 name_resolve:
1502   {
1503     return FALSE;
1504   }
1505 no_socket:
1506   {
1507     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
1508         ("no socket error: %s", err->message));
1509     g_clear_error (&err);
1510     g_object_unref (addr);
1511     return FALSE;
1512   }
1513 bind_error:
1514   {
1515     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1516         ("bind failed: %s", err->message));
1517     g_clear_error (&err);
1518     g_object_unref (bind_saddr);
1519     gst_udpsrc_close (src);
1520     return FALSE;
1521   }
1522 membership:
1523   {
1524     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1525         ("could add membership: %s", err->message));
1526     g_clear_error (&err);
1527     gst_udpsrc_close (src);
1528     return FALSE;
1529   }
1530 getsockname_error:
1531   {
1532     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1533         ("getsockname failed: %s", err->message));
1534     g_clear_error (&err);
1535     gst_udpsrc_close (src);
1536     return FALSE;
1537   }
1538 }
1539
1540 static gboolean
1541 gst_udpsrc_unlock (GstBaseSrc * bsrc)
1542 {
1543   GstUDPSrc *src;
1544
1545   src = GST_UDPSRC (bsrc);
1546
1547   GST_LOG_OBJECT (src, "Flushing");
1548   g_cancellable_cancel (src->cancellable);
1549
1550   return TRUE;
1551 }
1552
1553 static gboolean
1554 gst_udpsrc_unlock_stop (GstBaseSrc * bsrc)
1555 {
1556   GstUDPSrc *src;
1557
1558   src = GST_UDPSRC (bsrc);
1559
1560   GST_LOG_OBJECT (src, "No longer flushing");
1561
1562   gst_udpsrc_free_cancellable (src);
1563   gst_udpsrc_create_cancellable (src);
1564
1565   return TRUE;
1566 }
1567
1568 static gboolean
1569 gst_udpsrc_close (GstUDPSrc * src)
1570 {
1571   GST_DEBUG ("closing sockets");
1572
1573   if (src->used_socket) {
1574     if (src->auto_multicast
1575         &&
1576         g_inet_address_get_is_multicast (g_inet_socket_address_get_address
1577             (src->addr))) {
1578       GError *err = NULL;
1579
1580       GST_DEBUG_OBJECT (src, "leaving multicast group %s", src->address);
1581
1582       if (!g_socket_leave_multicast_group (src->used_socket,
1583               g_inet_socket_address_get_address (src->addr), FALSE,
1584               src->multi_iface, &err)) {
1585         GST_ERROR_OBJECT (src, "Failed to leave multicast group: %s",
1586             err->message);
1587         g_clear_error (&err);
1588       }
1589     }
1590
1591     if (src->close_socket || !src->external_socket) {
1592       GError *err = NULL;
1593       if (!g_socket_close (src->used_socket, &err)) {
1594         GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
1595         g_clear_error (&err);
1596       }
1597     }
1598
1599     g_object_unref (src->used_socket);
1600     src->used_socket = NULL;
1601     g_object_unref (src->addr);
1602     src->addr = NULL;
1603   }
1604
1605   gst_udpsrc_reset_memory_allocator (src);
1606
1607   gst_udpsrc_free_cancellable (src);
1608
1609   return TRUE;
1610 }
1611
1612
1613 static GstStateChangeReturn
1614 gst_udpsrc_change_state (GstElement * element, GstStateChange transition)
1615 {
1616   GstUDPSrc *src;
1617   GstStateChangeReturn result;
1618
1619   src = GST_UDPSRC (element);
1620
1621   switch (transition) {
1622     case GST_STATE_CHANGE_NULL_TO_READY:
1623       if (!gst_udpsrc_open (src))
1624         goto open_failed;
1625       break;
1626     default:
1627       break;
1628   }
1629   if ((result =
1630           GST_ELEMENT_CLASS (parent_class)->change_state (element,
1631               transition)) == GST_STATE_CHANGE_FAILURE)
1632     goto failure;
1633
1634   switch (transition) {
1635     case GST_STATE_CHANGE_READY_TO_NULL:
1636       gst_udpsrc_close (src);
1637       break;
1638     default:
1639       break;
1640   }
1641   return result;
1642   /* ERRORS */
1643 open_failed:
1644   {
1645     GST_DEBUG_OBJECT (src, "failed to open socket");
1646     return GST_STATE_CHANGE_FAILURE;
1647   }
1648 failure:
1649   {
1650     GST_DEBUG_OBJECT (src, "parent failed state change");
1651     return result;
1652   }
1653 }
1654
1655
1656
1657
1658 /*** GSTURIHANDLER INTERFACE *************************************************/
1659
1660 static GstURIType
1661 gst_udpsrc_uri_get_type (GType type)
1662 {
1663   return GST_URI_SRC;
1664 }
1665
1666 static const gchar *const *
1667 gst_udpsrc_uri_get_protocols (GType type)
1668 {
1669   static const gchar *protocols[] = { "udp", NULL };
1670
1671   return protocols;
1672 }
1673
1674 static gchar *
1675 gst_udpsrc_uri_get_uri (GstURIHandler * handler)
1676 {
1677   GstUDPSrc *src = GST_UDPSRC (handler);
1678
1679   return g_strdup (src->uri);
1680 }
1681
1682 static gboolean
1683 gst_udpsrc_uri_set_uri (GstURIHandler * handler, const gchar * uri,
1684     GError ** error)
1685 {
1686   return gst_udpsrc_set_uri (GST_UDPSRC (handler), uri, error);
1687 }
1688
1689 static void
1690 gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data)
1691 {
1692   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
1693
1694   iface->get_type = gst_udpsrc_uri_get_type;
1695   iface->get_protocols = gst_udpsrc_uri_get_protocols;
1696   iface->get_uri = gst_udpsrc_uri_get_uri;
1697   iface->set_uri = gst_udpsrc_uri_set_uri;
1698 }