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