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 downstream 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 static gboolean
 411 gst_udpsrc_decide_allocation (GstBaseSrc * bsrc, GstQuery * query)
 412 {
 413   GstUDPSrc *udpsrc;
 414   GstBufferPool *pool;
 415   gboolean update;
 416   GstStructure *config;
 417   GstCaps *caps = NULL;
 418
 419   udpsrc = GST_UDPSRC (bsrc);
 420
 421   if (gst_query_get_n_allocation_pools (query) > 0) {
 422     update = TRUE;
 423   } else {
 424     update = FALSE;
 425   }
 426
 427   pool = gst_buffer_pool_new ();
 428
 429   config = gst_buffer_pool_get_config (pool);
 430
 431   gst_query_parse_allocation (query, &caps, NULL);
 432
 433   gst_buffer_pool_config_set_params (config, caps, udpsrc->mtu, 0, 0);
 434
 435   gst_buffer_pool_set_config (pool, config);
 436
 437   if (update)
 438     gst_query_set_nth_allocation_pool (query, 0, pool, udpsrc->mtu, 0, 0);
 439   else
 440     gst_query_add_allocation_pool (query, pool, udpsrc->mtu, 0, 0);
 441
 442   gst_object_unref (pool);
 443
 444   return TRUE;
 445 }
 446
 447 /* not 100% correct, but a good upper bound for memory allocation purposes */
 448 #define MAX_IPV4_UDP_PACKET_SIZE (65536 - 8)
 449
 450 GST_DEBUG_CATEGORY_STATIC (udpsrc_debug);
 451 #define GST_CAT_DEFAULT (udpsrc_debug)
 452
 453 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
 454     GST_PAD_SRC,
 455     GST_PAD_ALWAYS,
 456     GST_STATIC_CAPS_ANY);
 457
 458 #define UDP_DEFAULT_PORT                5004
 459 #define UDP_DEFAULT_MULTICAST_GROUP     "0.0.0.0"
 460 #define UDP_DEFAULT_MULTICAST_IFACE     NULL
 461 #define UDP_DEFAULT_URI                 "udp://"UDP_DEFAULT_MULTICAST_GROUP":"G_STRINGIFY(UDP_DEFAULT_PORT)
 462 #define UDP_DEFAULT_CAPS                NULL
 463 #define UDP_DEFAULT_SOCKET              NULL
 464 #define UDP_DEFAULT_BUFFER_SIZE         0
 465 #define UDP_DEFAULT_TIMEOUT             0
 466 #define UDP_DEFAULT_SKIP_FIRST_BYTES    0
 467 #define UDP_DEFAULT_CLOSE_SOCKET       TRUE
 468 #define UDP_DEFAULT_USED_SOCKET        NULL
 469 #define UDP_DEFAULT_AUTO_MULTICAST     TRUE
 470 #define UDP_DEFAULT_REUSE              TRUE
 471 #define UDP_DEFAULT_LOOP               TRUE
 472 #define UDP_DEFAULT_RETRIEVE_SENDER_ADDRESS TRUE
 473 #define UDP_DEFAULT_MTU                (1492)
 474
 475 enum
 476 {
 477   PROP_0,
 478
 479   PROP_PORT,
 480   PROP_MULTICAST_GROUP,
 481   PROP_MULTICAST_IFACE,
 482   PROP_URI,
 483   PROP_CAPS,
 484   PROP_SOCKET,
 485   PROP_BUFFER_SIZE,
 486   PROP_TIMEOUT,
 487   PROP_SKIP_FIRST_BYTES,
 488   PROP_CLOSE_SOCKET,
 489   PROP_USED_SOCKET,
 490   PROP_AUTO_MULTICAST,
 491   PROP_REUSE,
 492   PROP_ADDRESS,
 493   PROP_LOOP,
 494   PROP_RETRIEVE_SENDER_ADDRESS,
 495   PROP_MTU,
 496 };
 497
 498 static void gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data);
 499
 500 static GstCaps *gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter);
 501 static gboolean gst_udpsrc_close (GstUDPSrc * src);
 502 static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc);
 503 static gboolean gst_udpsrc_unlock_stop (GstBaseSrc * bsrc);
 504 static GstFlowReturn gst_udpsrc_fill (GstPushSrc * psrc, GstBuffer * outbuf);
 505
 506 static void gst_udpsrc_finalize (GObject * object);
 507
 508 static void gst_udpsrc_set_property (GObject * object, guint prop_id,
 509     const GValue * value, GParamSpec * pspec);
 510 static void gst_udpsrc_get_property (GObject * object, guint prop_id,
 511     GValue * value, GParamSpec * pspec);
 512
 513 static GstStateChangeReturn gst_udpsrc_change_state (GstElement * element,
 514     GstStateChange transition);
 515
 516 #define gst_udpsrc_parent_class parent_class
 517 G_DEFINE_TYPE_WITH_CODE (GstUDPSrc, gst_udpsrc, GST_TYPE_PUSH_SRC,
 518     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_udpsrc_uri_handler_init));
 519
 520 static void
 521 gst_udpsrc_class_init (GstUDPSrcClass * klass)
 522 {
 523   GObjectClass *gobject_class;
 524   GstElementClass *gstelement_class;
 525   GstBaseSrcClass *gstbasesrc_class;
 526   GstPushSrcClass *gstpushsrc_class;
 527
 528   gobject_class = (GObjectClass *) klass;
 529   gstelement_class = (GstElementClass *) klass;
 530   gstbasesrc_class = (GstBaseSrcClass *) klass;
 531   gstpushsrc_class = (GstPushSrcClass *) klass;
 532
 533   GST_DEBUG_CATEGORY_INIT (udpsrc_debug, "udpsrc", 0, "UDP src");
 534
 535 #ifdef IP_PKTINFO
 536   GST_TYPE_IP_PKTINFO_MESSAGE;
 537 #endif
 538 #ifdef IPV6_PKTINFO
 539   GST_TYPE_IPV6_PKTINFO_MESSAGE;
 540 #endif
 541 #ifdef IP_RECVDSTADDR
 542   GST_TYPE_IP_RECVDSTADDR_MESSAGE;
 543 #endif
 544
 545   gobject_class->set_property = gst_udpsrc_set_property;
 546   gobject_class->get_property = gst_udpsrc_get_property;
 547   gobject_class->finalize = gst_udpsrc_finalize;
 548
 549   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
 550       g_param_spec_int ("port", "Port",
 551           "The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
 552           UDP_DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 553   /* FIXME 2.0: Remove multicast-group property */
 554 #ifndef GST_REMOVE_DEPRECATED
 555   g_object_class_install_property (gobject_class, PROP_MULTICAST_GROUP,
 556       g_param_spec_string ("multicast-group", "Multicast Group",
 557           "The Address of multicast group to join. (DEPRECATED: "
 558           "Use address property instead)", UDP_DEFAULT_MULTICAST_GROUP,
 559           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_DEPRECATED));
 560 #endif
 561   g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
 562       g_param_spec_string ("multicast-iface", "Multicast Interface",
 563           "The network interface on which to join the multicast group."
 564           "This allows multiple interfaces seperated by comma. (\"eth0,eth1\")",
 565           UDP_DEFAULT_MULTICAST_IFACE,
 566           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 567   g_object_class_install_property (gobject_class, PROP_URI,
 568       g_param_spec_string ("uri", "URI",
 569           "URI in the form of udp://multicast_group:port", UDP_DEFAULT_URI,
 570           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 571   g_object_class_install_property (gobject_class, PROP_CAPS,
 572       g_param_spec_boxed ("caps", "Caps",
 573           "The caps of the source pad", GST_TYPE_CAPS,
 574           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 575   g_object_class_install_property (gobject_class, PROP_SOCKET,
 576       g_param_spec_object ("socket", "Socket",
 577           "Socket to use for UDP reception. (NULL == allocate)",
 578           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 579   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
 580       g_param_spec_int ("buffer-size", "Buffer Size",
 581           "Size of the kernel receive buffer in bytes, 0=default", 0, G_MAXINT,
 582           UDP_DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 583   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_TIMEOUT,
 584       g_param_spec_uint64 ("timeout", "Timeout",
 585           "Post a message after timeout nanoseconds (0 = disabled)", 0,
 586           G_MAXUINT64, UDP_DEFAULT_TIMEOUT,
 587           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 588   g_object_class_install_property (G_OBJECT_CLASS (klass),
 589       PROP_SKIP_FIRST_BYTES, g_param_spec_int ("skip-first-bytes",
 590           "Skip first bytes", "number of bytes to skip for each udp packet", 0,
 591           G_MAXINT, UDP_DEFAULT_SKIP_FIRST_BYTES,
 592           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 593   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 594       g_param_spec_boolean ("close-socket", "Close socket",
 595           "Close socket if passed as property on state change",
 596           UDP_DEFAULT_CLOSE_SOCKET,
 597           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 598   g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
 599       g_param_spec_object ("used-socket", "Socket Handle",
 600           "Socket currently in use for UDP reception. (NULL = no socket)",
 601           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 602   g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
 603       g_param_spec_boolean ("auto-multicast", "Auto Multicast",
 604           "Automatically join/leave multicast groups",
 605           UDP_DEFAULT_AUTO_MULTICAST,
 606           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 607   g_object_class_install_property (gobject_class, PROP_REUSE,
 608       g_param_spec_boolean ("reuse", "Reuse", "Enable reuse of the port",
 609           UDP_DEFAULT_REUSE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 610   g_object_class_install_property (gobject_class, PROP_ADDRESS,
 611       g_param_spec_string ("address", "Address",
 612           "Address to receive packets for. This is equivalent to the "
 613           "multicast-group property for now", UDP_DEFAULT_MULTICAST_GROUP,
 614           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 615   /**
 616    * GstUDPSrc::loop:
 617    *
 618    * Can be used to disable multicast loopback.
 619    *
 620    * Since: 1.8
 621    */
 622   g_object_class_install_property (gobject_class, PROP_LOOP,
 623       g_param_spec_boolean ("loop", "Multicast Loopback",
 624           "Used for setting the multicast loop parameter. TRUE = enable,"
 625           " FALSE = disable", UDP_DEFAULT_LOOP,
 626           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 627   /**
 628    * GstUDPSrc::retrieve-sender-address:
 629    *
 630    * Whether to retrieve the sender address and add it to the buffers as
 631    * meta. Disabling this might result in minor performance improvements
 632    * in certain scenarios.
 633    *
 634    * Since: 1.10
 635    */
 636   g_object_class_install_property (gobject_class, PROP_RETRIEVE_SENDER_ADDRESS,
 637       g_param_spec_boolean ("retrieve-sender-address",
 638           "Retrieve Sender Address",
 639           "Whether to retrieve the sender address and add it to buffers as "
 640           "meta. Disabling this might result in minor performance improvements "
 641           "in certain scenarios", UDP_DEFAULT_RETRIEVE_SENDER_ADDRESS,
 642           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 643   /**
 644    * GstUDPSrc::mtu:
 645    *
 646    * Maximum expected packet size. This directly defines the allocation
 647    * size of the receive buffer pool.
 648    *
 649    * In case more data is received, a new #GstMemory is appended to the
 650    * output buffer, ensuring no data is lost, this however leads to that
 651    * buffer being freed and reallocated.
 652    *
 653    * Since: 1.14
 654    */
 655   g_object_class_install_property (gobject_class, PROP_MTU,
 656       g_param_spec_uint ("mtu", "Expected Maximum Transmission Unit",
 657           "Maximum expected packet size. This directly defines the allocation"
 658           "size of the receive buffer pool.",
 659           0, G_MAXINT, UDP_DEFAULT_MTU,
 660           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 661
 662   gst_element_class_add_static_pad_template (gstelement_class, &src_template);
 663
 664   gst_element_class_set_static_metadata (gstelement_class,
 665       "UDP packet receiver", "Source/Network",
 666       "Receive data over the network via UDP",
 667       "Wim Taymans <wim@fluendo.com>, "
 668       "Thijs Vermeir <thijs.vermeir@barco.com>");
 669
 670   gstelement_class->change_state = gst_udpsrc_change_state;
 671
 672   gstbasesrc_class->unlock = gst_udpsrc_unlock;
 673   gstbasesrc_class->unlock_stop = gst_udpsrc_unlock_stop;
 674   gstbasesrc_class->get_caps = gst_udpsrc_getcaps;
 675   gstbasesrc_class->decide_allocation = gst_udpsrc_decide_allocation;
 676
 677   gstpushsrc_class->fill = gst_udpsrc_fill;
 678 }
 679
 680 static void
 681 gst_udpsrc_init (GstUDPSrc * udpsrc)
 682 {
 683   udpsrc->uri =
 684       g_strdup_printf ("udp://%s:%u", UDP_DEFAULT_MULTICAST_GROUP,
 685       UDP_DEFAULT_PORT);
 686
 687   udpsrc->address = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 688   udpsrc->port = UDP_DEFAULT_PORT;
 689   udpsrc->socket = UDP_DEFAULT_SOCKET;
 690   udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 691   udpsrc->buffer_size = UDP_DEFAULT_BUFFER_SIZE;
 692   udpsrc->timeout = UDP_DEFAULT_TIMEOUT;
 693   udpsrc->skip_first_bytes = UDP_DEFAULT_SKIP_FIRST_BYTES;
 694   udpsrc->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
 695   udpsrc->external_socket = (udpsrc->socket != NULL);
 696   udpsrc->auto_multicast = UDP_DEFAULT_AUTO_MULTICAST;
 697   udpsrc->used_socket = UDP_DEFAULT_USED_SOCKET;
 698   udpsrc->reuse = UDP_DEFAULT_REUSE;
 699   udpsrc->loop = UDP_DEFAULT_LOOP;
 700   udpsrc->retrieve_sender_address = UDP_DEFAULT_RETRIEVE_SENDER_ADDRESS;
 701   udpsrc->mtu = UDP_DEFAULT_MTU;
 702
 703   /* configure basesrc to be a live source */
 704   gst_base_src_set_live (GST_BASE_SRC (udpsrc), TRUE);
 705   /* make basesrc output a segment in time */
 706   gst_base_src_set_format (GST_BASE_SRC (udpsrc), GST_FORMAT_TIME);
 707   /* make basesrc set timestamps on outgoing buffers based on the running_time
 708    * when they were captured */
 709   gst_base_src_set_do_timestamp (GST_BASE_SRC (udpsrc), TRUE);
 710 }
 711
 712 static void
 713 gst_udpsrc_finalize (GObject * object)
 714 {
 715   GstUDPSrc *udpsrc;
 716
 717   udpsrc = GST_UDPSRC (object);
 718
 719   if (udpsrc->caps)
 720     gst_caps_unref (udpsrc->caps);
 721   udpsrc->caps = NULL;
 722
 723   g_free (udpsrc->multi_iface);
 724   udpsrc->multi_iface = NULL;
 725
 726   g_free (udpsrc->uri);
 727   udpsrc->uri = NULL;
 728
 729   g_free (udpsrc->address);
 730   udpsrc->address = NULL;
 731
 732   if (udpsrc->socket)
 733     g_object_unref (udpsrc->socket);
 734   udpsrc->socket = NULL;
 735
 736   if (udpsrc->used_socket)
 737     g_object_unref (udpsrc->used_socket);
 738   udpsrc->used_socket = NULL;
 739
 740   if (udpsrc->extra_mem)
 741     gst_memory_unref (udpsrc->extra_mem);
 742   udpsrc->extra_mem = NULL;
 743
 744   G_OBJECT_CLASS (parent_class)->finalize (object);
 745 }
 746
 747 static GstCaps *
 748 gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter)
 749 {
 750   GstUDPSrc *udpsrc;
 751   GstCaps *caps, *result;
 752
 753   udpsrc = GST_UDPSRC (src);
 754
 755   GST_OBJECT_LOCK (src);
 756   if ((caps = udpsrc->caps))
 757     gst_caps_ref (caps);
 758   GST_OBJECT_UNLOCK (src);
 759
 760   if (caps) {
 761     if (filter) {
 762       result = gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
 763       gst_caps_unref (caps);
 764     } else {
 765       result = caps;
 766     }
 767   } else {
 768     result = (filter) ? gst_caps_ref (filter) : gst_caps_new_any ();
 769   }
 770   return result;
 771 }
 772
 773 static void
 774 gst_udpsrc_create_cancellable (GstUDPSrc * src)
 775 {
 776   GPollFD pollfd;
 777
 778   src->cancellable = g_cancellable_new ();
 779   src->made_cancel_fd = g_cancellable_make_pollfd (src->cancellable, &pollfd);
 780 }
 781
 782 static void
 783 gst_udpsrc_free_cancellable (GstUDPSrc * src)
 784 {
 785   if (src->made_cancel_fd) {
 786     g_cancellable_release_fd (src->cancellable);
 787     src->made_cancel_fd = FALSE;
 788   }
 789   g_object_unref (src->cancellable);
 790   src->cancellable = NULL;
 791 }
 792
 793 static GstFlowReturn
 794 gst_udpsrc_fill (GstPushSrc * psrc, GstBuffer * outbuf)
 795 {
 796   GstUDPSrc *udpsrc;
 797   GSocketAddress *saddr = NULL;
 798   GSocketAddress **p_saddr;
 799   gint flags = G_SOCKET_MSG_NONE;
 800   gboolean try_again;
 801   GError *err = NULL;
 802   gssize res;
 803   gsize offset;
 804   GSocketControlMessage **msgs = NULL;
 805   GSocketControlMessage ***p_msgs;
 806   gint n_msgs = 0, i;
 807   GstMapInfo info;
 808   GstMapInfo extra_info;
 809   GInputVector ivec[2];
 810
 811   udpsrc = GST_UDPSRC_CAST (psrc);
 812
 813   /* optimization: use messages only in multicast mode and
 814    * if we can't let the kernel do the filtering for us */
 815   p_msgs =
 816       (g_inet_address_get_is_multicast (g_inet_socket_address_get_address
 817           (udpsrc->addr))) ? &msgs : NULL;
 818 #ifdef IP_MULTICAST_ALL
 819   if (g_inet_address_get_family (g_inet_socket_address_get_address
 820           (udpsrc->addr)) == G_SOCKET_FAMILY_IPV4)
 821     p_msgs = NULL;
 822 #endif
 823
 824   /* Retrieve sender address unless we've been configured not to do so */
 825   p_saddr = (udpsrc->retrieve_sender_address) ? &saddr : NULL;
 826
 827   if (!gst_buffer_map (outbuf, &info, GST_MAP_READWRITE))
 828     goto buffer_map_error;
 829
 830   ivec[0].buffer = info.data;
 831   ivec[0].size = info.size;
 832
 833   /* Prepare memory in case the data size exceeds mtu */
 834   if (udpsrc->extra_mem == NULL) {
 835     GstBufferPool *pool;
 836     GstStructure *config;
 837     GstAllocator *allocator = NULL;
 838     GstAllocationParams params;
 839
 840     pool = gst_base_src_get_buffer_pool (GST_BASE_SRC_CAST (psrc));
 841     config = gst_buffer_pool_get_config (pool);
 842     gst_buffer_pool_config_get_allocator (config, &allocator, &params);
 843
 844     udpsrc->extra_mem =
 845         gst_allocator_alloc (allocator, MAX_IPV4_UDP_PACKET_SIZE, &params);
 846
 847     gst_object_unref (pool);
 848     gst_structure_free (config);
 849     if (allocator)
 850       gst_object_unref (allocator);
 851   }
 852
 853   if (!gst_memory_map (udpsrc->extra_mem, &extra_info, GST_MAP_READWRITE))
 854     goto memory_map_error;
 855
 856   ivec[1].buffer = extra_info.data;
 857   ivec[1].size = extra_info.size;
 858
 859 retry:
 860   if (saddr != NULL) {
 861     g_object_unref (saddr);
 862     saddr = NULL;
 863   }
 864
 865   do {
 866     gint64 timeout;
 867
 868     try_again = FALSE;
 869
 870     if (udpsrc->timeout)
 871       timeout = udpsrc->timeout / 1000;
 872     else
 873       timeout = -1;
 874
 875     GST_LOG_OBJECT (udpsrc, "doing select, timeout %" G_GINT64_FORMAT, timeout);
 876
 877     if (!g_socket_condition_timed_wait (udpsrc->used_socket, G_IO_IN | G_IO_PRI,
 878             timeout, udpsrc->cancellable, &err)) {
 879       if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY)
 880           || g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 881         goto stopped;
 882       } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
 883         g_clear_error (&err);
 884         /* timeout, post element message */
 885         gst_element_post_message (GST_ELEMENT_CAST (udpsrc),
 886             gst_message_new_element (GST_OBJECT_CAST (udpsrc),
 887                 gst_structure_new ("GstUDPSrcTimeout",
 888                     "timeout", G_TYPE_UINT64, udpsrc->timeout, NULL)));
 889       } else {
 890         goto select_error;
 891       }
 892
 893       try_again = TRUE;
 894     }
 895   } while (G_UNLIKELY (try_again));
 896
 897   res =
 898       g_socket_receive_message (udpsrc->used_socket, p_saddr, ivec, 2,
 899       p_msgs, &n_msgs, &flags, udpsrc->cancellable, &err);
 900
 901   if (G_UNLIKELY (res < 0)) {
 902     /* G_IO_ERROR_HOST_UNREACHABLE for a UDP socket means that a packet sent
 903      * with udpsink generated a "port unreachable" ICMP response. We ignore
 904      * that and try again.
 905      * On Windows we get G_IO_ERROR_CONNECTION_CLOSED instead */
 906 #if GLIB_CHECK_VERSION(2,44,0)
 907     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_HOST_UNREACHABLE) ||
 908         g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CONNECTION_CLOSED)) {
 909 #else
 910     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_HOST_UNREACHABLE)) {
 911 #endif
 912       g_clear_error (&err);
 913       goto retry;
 914     }
 915     goto receive_error;
 916   }
 917
 918   /* Retry if multicast and the destination address is not ours. We don't want
 919    * to receive arbitrary packets */
 920   if (p_msgs) {
 921     GInetAddress *iaddr = g_inet_socket_address_get_address (udpsrc->addr);
 922     gboolean skip_packet = FALSE;
 923     gsize iaddr_size = g_inet_address_get_native_size (iaddr);
 924     const guint8 *iaddr_bytes = g_inet_address_to_bytes (iaddr);
 925
 926     for (i = 0; i < n_msgs && !skip_packet; i++) {
 927 #ifdef IP_PKTINFO
 928       if (GST_IS_IP_PKTINFO_MESSAGE (msgs[i])) {
 929         GstIPPktinfoMessage *msg = GST_IP_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 IPV6_PKTINFO
 937       if (GST_IS_IPV6_PKTINFO_MESSAGE (msgs[i])) {
 938         GstIPV6PktinfoMessage *msg = GST_IPV6_PKTINFO_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 #ifdef IP_RECVDSTADDR
 946       if (GST_IS_IP_RECVDSTADDR_MESSAGE (msgs[i])) {
 947         GstIPRecvdstaddrMessage *msg = GST_IP_RECVDSTADDR_MESSAGE (msgs[i]);
 948
 949         if (sizeof (msg->addr) == iaddr_size
 950             && memcmp (iaddr_bytes, &msg->addr, sizeof (msg->addr)))
 951           skip_packet = TRUE;
 952       }
 953 #endif
 954     }
 955
 956     for (i = 0; i < n_msgs; i++) {
 957       g_object_unref (msgs[i]);
 958     }
 959     g_free (msgs);
 960
 961     if (skip_packet) {
 962       GST_DEBUG_OBJECT (udpsrc,
 963           "Dropping packet for a different multicast address");
 964       goto retry;
 965     }
 966   }
 967
 968   gst_buffer_unmap (outbuf, &info);
 969   gst_memory_unmap (udpsrc->extra_mem, &extra_info);
 970
 971   /* If this is the case, the buffer will be freed once unreffed,
 972    * and the buffer pool will have to reallocate a new one.
 973    */
 974   if (res > udpsrc->mtu) {
 975     gst_buffer_append_memory (outbuf, udpsrc->extra_mem);
 976     udpsrc->extra_mem = NULL;
 977   }
 978
 979   offset = udpsrc->skip_first_bytes;
 980
 981   if (G_UNLIKELY (offset > 0 && res < offset))
 982     goto skip_error;
 983
 984   gst_buffer_resize (outbuf, offset, res - offset);
 985
 986   /* use buffer metadata so receivers can also track the address */
 987   if (saddr) {
 988     gst_buffer_add_net_address_meta (outbuf, saddr);
 989     g_object_unref (saddr);
 990     saddr = NULL;
 991   }
 992
 993   GST_LOG_OBJECT (udpsrc, "read packet of %d bytes", (int) res);
 994
 995   return GST_FLOW_OK;
 996
 997   /* ERRORS */
 998 buffer_map_error:
 999   {
1000     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
1001         ("Failed to map memory"));
1002     return GST_FLOW_ERROR;
1003   }
1004 memory_map_error:
1005   {
1006     gst_buffer_unmap (outbuf, &info);
1007     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
1008         ("Failed to map memory"));
1009     return GST_FLOW_ERROR;
1010   }
1011 select_error:
1012   {
1013     gst_buffer_unmap (outbuf, &info);
1014     gst_memory_unmap (udpsrc->extra_mem, &extra_info);
1015     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
1016         ("select error: %s", err->message));
1017     g_clear_error (&err);
1018     return GST_FLOW_ERROR;
1019   }
1020 stopped:
1021   {
1022     gst_buffer_unmap (outbuf, &info);
1023     gst_memory_unmap (udpsrc->extra_mem, &extra_info);
1024     GST_DEBUG ("stop called");
1025     g_clear_error (&err);
1026     return GST_FLOW_FLUSHING;
1027   }
1028 receive_error:
1029   {
1030     gst_buffer_unmap (outbuf, &info);
1031     gst_memory_unmap (udpsrc->extra_mem, &extra_info);
1032     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY) ||
1033         g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
1034       g_clear_error (&err);
1035       return GST_FLOW_FLUSHING;
1036     } else {
1037       GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
1038           ("receive error %" G_GSSIZE_FORMAT ": %s", res, err->message));
1039       g_clear_error (&err);
1040       return GST_FLOW_ERROR;
1041     }
1042   }
1043 skip_error:
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     case PROP_MTU:
1186       udpsrc->mtu = g_value_get_uint (value);
1187       break;
1188     default:
1189       break;
1190   }
1191 }
1192
1193 static void
1194 gst_udpsrc_get_property (GObject * object, guint prop_id, GValue * value,
1195     GParamSpec * pspec)
1196 {
1197   GstUDPSrc *udpsrc = GST_UDPSRC (object);
1198
1199   switch (prop_id) {
1200     case PROP_BUFFER_SIZE:
1201       g_value_set_int (value, udpsrc->buffer_size);
1202       break;
1203     case PROP_PORT:
1204       g_value_set_int (value, udpsrc->port);
1205       break;
1206     case PROP_MULTICAST_GROUP:
1207     case PROP_ADDRESS:
1208       g_value_set_string (value, udpsrc->address);
1209       break;
1210     case PROP_MULTICAST_IFACE:
1211       g_value_set_string (value, udpsrc->multi_iface);
1212       break;
1213     case PROP_URI:
1214       g_value_set_string (value, udpsrc->uri);
1215       break;
1216     case PROP_CAPS:
1217       GST_OBJECT_LOCK (udpsrc);
1218       gst_value_set_caps (value, udpsrc->caps);
1219       GST_OBJECT_UNLOCK (udpsrc);
1220       break;
1221     case PROP_SOCKET:
1222       g_value_set_object (value, udpsrc->socket);
1223       break;
1224     case PROP_TIMEOUT:
1225       g_value_set_uint64 (value, udpsrc->timeout);
1226       break;
1227     case PROP_SKIP_FIRST_BYTES:
1228       g_value_set_int (value, udpsrc->skip_first_bytes);
1229       break;
1230     case PROP_CLOSE_SOCKET:
1231       g_value_set_boolean (value, udpsrc->close_socket);
1232       break;
1233     case PROP_USED_SOCKET:
1234       g_value_set_object (value, udpsrc->used_socket);
1235       break;
1236     case PROP_AUTO_MULTICAST:
1237       g_value_set_boolean (value, udpsrc->auto_multicast);
1238       break;
1239     case PROP_REUSE:
1240       g_value_set_boolean (value, udpsrc->reuse);
1241       break;
1242     case PROP_LOOP:
1243       g_value_set_boolean (value, udpsrc->loop);
1244       break;
1245     case PROP_RETRIEVE_SENDER_ADDRESS:
1246       g_value_set_boolean (value, udpsrc->retrieve_sender_address);
1247       break;
1248     case PROP_MTU:
1249       g_value_set_uint (value, udpsrc->mtu);
1250       break;
1251     default:
1252       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1253       break;
1254   }
1255 }
1256
1257 static GInetAddress *
1258 gst_udpsrc_resolve (GstUDPSrc * src, const gchar * address)
1259 {
1260   GInetAddress *addr;
1261   GError *err = NULL;
1262   GResolver *resolver;
1263
1264   addr = g_inet_address_new_from_string (address);
1265   if (!addr) {
1266     GList *results;
1267
1268     GST_DEBUG_OBJECT (src, "resolving IP address for host %s", address);
1269     resolver = g_resolver_get_default ();
1270     results =
1271         g_resolver_lookup_by_name (resolver, address, src->cancellable, &err);
1272     if (!results)
1273       goto name_resolve;
1274     addr = G_INET_ADDRESS (g_object_ref (results->data));
1275
1276     g_resolver_free_addresses (results);
1277     g_object_unref (resolver);
1278   }
1279 #ifndef GST_DISABLE_GST_DEBUG
1280   {
1281     gchar *ip = g_inet_address_to_string (addr);
1282
1283     GST_DEBUG_OBJECT (src, "IP address for host %s is %s", address, ip);
1284     g_free (ip);
1285   }
1286 #endif
1287
1288   return addr;
1289
1290 name_resolve:
1291   {
1292     GST_WARNING_OBJECT (src, "Failed to resolve %s: %s", address, err->message);
1293     g_clear_error (&err);
1294     g_object_unref (resolver);
1295     return NULL;
1296   }
1297 }
1298
1299 static gint
1300 gst_udpsrc_get_rcvbuf (GstUDPSrc * src)
1301 {
1302   gint val = 0;
1303
1304   /* read the value of the receive buffer. Note that on linux this returns
1305    * 2x the value we set because the kernel allocates extra memory for
1306    * metadata. The default on Linux is about 100K (which is about 50K
1307    * without metadata) */
1308   if (!g_socket_get_option (src->used_socket, SOL_SOCKET, SO_RCVBUF, &val,
1309           NULL)) {
1310     GST_DEBUG_OBJECT (src, "could not get udp buffer size");
1311     return 0;
1312   }
1313 #ifdef __linux__
1314   /* Devise by 2 so that the numbers matches when we do get/set */
1315   val /= 2;
1316 #endif
1317
1318   return val;
1319 }
1320
1321 /* create a socket for sending to remote machine */
1322 static gboolean
1323 gst_udpsrc_open (GstUDPSrc * src)
1324 {
1325   GInetAddress *addr, *bind_addr;
1326   GSocketAddress *bind_saddr;
1327   GError *err = NULL;
1328
1329   gst_udpsrc_create_cancellable (src);
1330
1331   if (src->socket == NULL) {
1332     /* need to allocate a socket */
1333     GST_DEBUG_OBJECT (src, "allocating socket for %s:%d", src->address,
1334         src->port);
1335
1336     addr = gst_udpsrc_resolve (src, src->address);
1337     if (!addr)
1338       goto name_resolve;
1339
1340     if ((src->used_socket =
1341             g_socket_new (g_inet_address_get_family (addr),
1342                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
1343       goto no_socket;
1344
1345     src->external_socket = FALSE;
1346
1347     GST_DEBUG_OBJECT (src, "got socket %p", src->used_socket);
1348
1349     if (src->addr)
1350       g_object_unref (src->addr);
1351     src->addr =
1352         G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
1353
1354     GST_DEBUG_OBJECT (src, "binding on port %d", src->port);
1355
1356     /* For multicast, bind to ANY and join the multicast group later */
1357     if (g_inet_address_get_is_multicast (addr))
1358       bind_addr = g_inet_address_new_any (g_inet_address_get_family (addr));
1359     else
1360       bind_addr = G_INET_ADDRESS (g_object_ref (addr));
1361
1362     g_object_unref (addr);
1363
1364     bind_saddr = g_inet_socket_address_new (bind_addr, src->port);
1365     g_object_unref (bind_addr);
1366     if (!g_socket_bind (src->used_socket, bind_saddr, src->reuse, &err))
1367       goto bind_error;
1368
1369     g_object_unref (bind_saddr);
1370     g_socket_set_multicast_loopback (src->used_socket, src->loop);
1371   } else {
1372     GInetSocketAddress *local_addr;
1373
1374     GST_DEBUG_OBJECT (src, "using provided socket %p", src->socket);
1375     /* we use the configured socket, try to get some info about it */
1376     src->used_socket = G_SOCKET (g_object_ref (src->socket));
1377     src->external_socket = TRUE;
1378
1379     local_addr =
1380         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
1381             &err));
1382     if (!local_addr)
1383       goto getsockname_error;
1384
1385     addr = gst_udpsrc_resolve (src, src->address);
1386     if (!addr)
1387       goto name_resolve;
1388
1389     /* If bound to ANY and address points to a multicast address, make
1390      * sure that address is not overridden with ANY but we have the
1391      * opportunity later to join the multicast address. This ensures that we
1392      * have the same behaviour as for sockets created by udpsrc */
1393     if (!src->auto_multicast ||
1394         !g_inet_address_get_is_any (g_inet_socket_address_get_address
1395             (local_addr))
1396         || !g_inet_address_get_is_multicast (addr)) {
1397       g_object_unref (addr);
1398       if (src->addr)
1399         g_object_unref (src->addr);
1400       src->addr = local_addr;
1401     } else {
1402       g_object_unref (local_addr);
1403       if (src->addr)
1404         g_object_unref (src->addr);
1405       src->addr =
1406           G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
1407       g_object_unref (addr);
1408     }
1409   }
1410
1411   {
1412     gint val;
1413     GError *opt_err = NULL;
1414     gboolean force_rcvbuf G_GNUC_UNUSED = FALSE;
1415
1416     if (src->buffer_size != 0) {
1417       GST_INFO_OBJECT (src, "setting udp buffer of %d bytes", src->buffer_size);
1418       /* set buffer size, Note that on Linux this is typically limited to a
1419        * maximum of around 100K. Also a minimum of 128 bytes is required on
1420        * Linux. */
1421       if (!g_socket_set_option (src->used_socket, SOL_SOCKET, SO_RCVBUF,
1422               src->buffer_size, &opt_err)) {
1423         GST_INFO_OBJECT (src,
1424             "Could not create a buffer of requested %d bytes (%s) try forcing",
1425             src->buffer_size, opt_err->message);
1426         g_clear_error (&opt_err);
1427         force_rcvbuf = TRUE;
1428       }
1429     }
1430 #if defined(SO_RCVBUFFORCE)
1431     val = gst_udpsrc_get_rcvbuf (src);
1432     if (val < src->buffer_size)
1433       force_rcvbuf = TRUE;
1434
1435     if (force_rcvbuf) {
1436       GST_INFO_OBJECT (src,
1437           "forcibly setting udp buffer of %d bytes", src->buffer_size);
1438
1439       /* Will only work with CAP_NET_ADMIN privilege */
1440       if (!g_socket_set_option (src->used_socket, SOL_SOCKET, SO_RCVBUFFORCE,
1441               src->buffer_size, &opt_err)) {
1442         GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
1443             ("Could not create a buffer of requested %d bytes (%s). Need net.admin privilege?",
1444                 src->buffer_size, opt_err->message));
1445         g_clear_error (&opt_err);
1446       }
1447     }
1448 #endif
1449
1450     val = gst_udpsrc_get_rcvbuf (src);
1451     if (val < src->buffer_size)
1452       GST_WARNING_OBJECT (src,
1453           "have udp buffer of %d bytes while %d were requested",
1454           val, src->buffer_size);
1455     else
1456       GST_INFO_OBJECT (src, "have udp buffer of %d bytes", val);
1457   }
1458
1459   g_socket_set_broadcast (src->used_socket, TRUE);
1460
1461   if (src->auto_multicast
1462       &&
1463       g_inet_address_get_is_multicast (g_inet_socket_address_get_address
1464           (src->addr))) {
1465
1466     if (src->multi_iface) {
1467       GStrv multi_ifaces = g_strsplit (src->multi_iface, ",", -1);
1468       gchar **ifaces = multi_ifaces;
1469       while (*ifaces) {
1470         g_strstrip (*ifaces);
1471         GST_DEBUG_OBJECT (src, "joining multicast group %s interface %s",
1472             src->address, *ifaces);
1473         if (!g_socket_join_multicast_group (src->used_socket,
1474                 g_inet_socket_address_get_address (src->addr),
1475                 FALSE, *ifaces, &err)) {
1476           g_strfreev (multi_ifaces);
1477           goto membership;
1478         }
1479
1480         ifaces++;
1481       }
1482       g_strfreev (multi_ifaces);
1483     } else {
1484       GST_DEBUG_OBJECT (src, "joining multicast group %s", src->address);
1485       if (!g_socket_join_multicast_group (src->used_socket,
1486               g_inet_socket_address_get_address (src->addr), FALSE, NULL, &err))
1487         goto membership;
1488     }
1489
1490     if (g_inet_address_get_family (g_inet_socket_address_get_address
1491             (src->addr)) == G_SOCKET_FAMILY_IPV4) {
1492 #if defined(IP_MULTICAST_ALL)
1493       if (!g_socket_set_option (src->used_socket, IPPROTO_IP, IP_MULTICAST_ALL,
1494               0, &err)) {
1495         GST_WARNING_OBJECT (src, "Failed to disable IP_MULTICAST_ALL: %s",
1496             err->message);
1497         g_clear_error (&err);
1498       }
1499 #elif defined(IP_PKTINFO)
1500       if (!g_socket_set_option (src->used_socket, IPPROTO_IP, IP_PKTINFO, TRUE,
1501               &err)) {
1502         GST_WARNING_OBJECT (src, "Failed to enable IP_PKTINFO: %s",
1503             err->message);
1504         g_clear_error (&err);
1505       }
1506 #elif defined(IP_RECVDSTADDR)
1507       if (!g_socket_set_option (src->used_socket, IPPROTO_IP, IP_RECVDSTADDR,
1508               TRUE, &err)) {
1509         GST_WARNING_OBJECT (src, "Failed to enable IP_RECVDSTADDR: %s",
1510             err->message);
1511         g_clear_error (&err);
1512       }
1513 #else
1514 #pragma message("No API available for getting IPv4 destination address")
1515       GST_WARNING_OBJECT (src, "No API available for getting IPv4 destination "
1516           "address, will receive packets for every destination to our port");
1517 #endif
1518     } else
1519         if (g_inet_address_get_family (g_inet_socket_address_get_address
1520             (src->addr)) == G_SOCKET_FAMILY_IPV6) {
1521 #ifdef IPV6_PKTINFO
1522 #ifdef IPV6_RECVPKTINFO
1523       if (!g_socket_set_option (src->used_socket, IPPROTO_IPV6,
1524               IPV6_RECVPKTINFO, TRUE, &err)) {
1525 #else
1526       if (!g_socket_set_option (src->used_socket, IPPROTO_IPV6, IPV6_PKTINFO,
1527               TRUE, &err)) {
1528 #endif
1529         GST_WARNING_OBJECT (src, "Failed to enable IPV6_PKTINFO: %s",
1530             err->message);
1531         g_clear_error (&err);
1532       }
1533 #else
1534 #pragma message("No API available for getting IPv6 destination address")
1535       GST_WARNING_OBJECT (src, "No API available for getting IPv6 destination "
1536           "address, will receive packets for every destination to our port");
1537 #endif
1538     }
1539   }
1540
1541   /* NOTE: sockaddr_in.sin_port works for ipv4 and ipv6 because sin_port
1542    * follows ss_family on both */
1543   {
1544     GInetSocketAddress *addr;
1545     guint16 port;
1546
1547     addr =
1548         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
1549             &err));
1550     if (!addr)
1551       goto getsockname_error;
1552
1553     port = g_inet_socket_address_get_port (addr);
1554     GST_DEBUG_OBJECT (src, "bound, on port %d", port);
1555     if (port != src->port) {
1556       src->port = port;
1557       GST_DEBUG_OBJECT (src, "notifying port %d", port);
1558       g_object_notify (G_OBJECT (src), "port");
1559     }
1560     g_object_unref (addr);
1561   }
1562
1563   return TRUE;
1564
1565   /* ERRORS */
1566 name_resolve:
1567   {
1568     return FALSE;
1569   }
1570 no_socket:
1571   {
1572     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
1573         ("no socket error: %s", err->message));
1574     g_clear_error (&err);
1575     g_object_unref (addr);
1576     return FALSE;
1577   }
1578 bind_error:
1579   {
1580     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1581         ("bind failed: %s", err->message));
1582     g_clear_error (&err);
1583     g_object_unref (bind_saddr);
1584     gst_udpsrc_close (src);
1585     return FALSE;
1586   }
1587 membership:
1588   {
1589     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1590         ("could not add membership: %s", err->message));
1591     g_clear_error (&err);
1592     gst_udpsrc_close (src);
1593     return FALSE;
1594   }
1595 getsockname_error:
1596   {
1597     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1598         ("getsockname failed: %s", err->message));
1599     g_clear_error (&err);
1600     gst_udpsrc_close (src);
1601     return FALSE;
1602   }
1603 }
1604
1605 static gboolean
1606 gst_udpsrc_unlock (GstBaseSrc * bsrc)
1607 {
1608   GstUDPSrc *src;
1609
1610   src = GST_UDPSRC (bsrc);
1611
1612   GST_LOG_OBJECT (src, "Flushing");
1613   g_cancellable_cancel (src->cancellable);
1614
1615   return TRUE;
1616 }
1617
1618 static gboolean
1619 gst_udpsrc_unlock_stop (GstBaseSrc * bsrc)
1620 {
1621   GstUDPSrc *src;
1622
1623   src = GST_UDPSRC (bsrc);
1624
1625   GST_LOG_OBJECT (src, "No longer flushing");
1626
1627   gst_udpsrc_free_cancellable (src);
1628   gst_udpsrc_create_cancellable (src);
1629
1630   return TRUE;
1631 }
1632
1633 static gboolean
1634 gst_udpsrc_close (GstUDPSrc * src)
1635 {
1636   GST_DEBUG ("closing sockets");
1637
1638   if (src->used_socket) {
1639     if (src->auto_multicast
1640         &&
1641         g_inet_address_get_is_multicast (g_inet_socket_address_get_address
1642             (src->addr))) {
1643       GError *err = NULL;
1644
1645       if (src->multi_iface) {
1646         GStrv multi_ifaces = g_strsplit (src->multi_iface, ",", -1);
1647         gchar **ifaces = multi_ifaces;
1648         while (*ifaces) {
1649           g_strstrip (*ifaces);
1650           GST_DEBUG_OBJECT (src, "leaving multicast group %s interface %s",
1651               src->address, *ifaces);
1652           if (!g_socket_leave_multicast_group (src->used_socket,
1653                   g_inet_socket_address_get_address (src->addr),
1654                   FALSE, *ifaces, &err)) {
1655             GST_ERROR_OBJECT (src, "Failed to leave multicast group: %s",
1656                 err->message);
1657             g_clear_error (&err);
1658           }
1659           ifaces++;
1660         }
1661         g_strfreev (multi_ifaces);
1662
1663       } else {
1664         GST_DEBUG_OBJECT (src, "leaving multicast group %s", src->address);
1665         if (!g_socket_leave_multicast_group (src->used_socket,
1666                 g_inet_socket_address_get_address (src->addr), FALSE,
1667                 NULL, &err)) {
1668           GST_ERROR_OBJECT (src, "Failed to leave multicast group: %s",
1669               err->message);
1670           g_clear_error (&err);
1671         }
1672       }
1673     }
1674
1675     if (src->close_socket || !src->external_socket) {
1676       GError *err = NULL;
1677       if (!g_socket_close (src->used_socket, &err)) {
1678         GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
1679         g_clear_error (&err);
1680       }
1681     }
1682
1683     g_object_unref (src->used_socket);
1684     src->used_socket = NULL;
1685     g_object_unref (src->addr);
1686     src->addr = NULL;
1687   }
1688
1689   gst_udpsrc_free_cancellable (src);
1690
1691   return TRUE;
1692 }
1693
1694
1695 static GstStateChangeReturn
1696 gst_udpsrc_change_state (GstElement * element, GstStateChange transition)
1697 {
1698   GstUDPSrc *src;
1699   GstStateChangeReturn result;
1700
1701   src = GST_UDPSRC (element);
1702
1703   switch (transition) {
1704     case GST_STATE_CHANGE_NULL_TO_READY:
1705       if (!gst_udpsrc_open (src))
1706         goto open_failed;
1707       break;
1708     default:
1709       break;
1710   }
1711   if ((result =
1712           GST_ELEMENT_CLASS (parent_class)->change_state (element,
1713               transition)) == GST_STATE_CHANGE_FAILURE)
1714     goto failure;
1715
1716   switch (transition) {
1717     case GST_STATE_CHANGE_READY_TO_NULL:
1718       gst_udpsrc_close (src);
1719       break;
1720     default:
1721       break;
1722   }
1723   return result;
1724   /* ERRORS */
1725 open_failed:
1726   {
1727     GST_DEBUG_OBJECT (src, "failed to open socket");
1728     return GST_STATE_CHANGE_FAILURE;
1729   }
1730 failure:
1731   {
1732     GST_DEBUG_OBJECT (src, "parent failed state change");
1733     return result;
1734   }
1735 }
1736
1737
1738
1739
1740 /*** GSTURIHANDLER INTERFACE *************************************************/
1741
1742 static GstURIType
1743 gst_udpsrc_uri_get_type (GType type)
1744 {
1745   return GST_URI_SRC;
1746 }
1747
1748 static const gchar *const *
1749 gst_udpsrc_uri_get_protocols (GType type)
1750 {
1751   static const gchar *protocols[] = { "udp", NULL };
1752
1753   return protocols;
1754 }
1755
1756 static gchar *
1757 gst_udpsrc_uri_get_uri (GstURIHandler * handler)
1758 {
1759   GstUDPSrc *src = GST_UDPSRC (handler);
1760
1761   return g_strdup (src->uri);
1762 }
1763
1764 static gboolean
1765 gst_udpsrc_uri_set_uri (GstURIHandler * handler, const gchar * uri,
1766     GError ** error)
1767 {
1768   return gst_udpsrc_set_uri (GST_UDPSRC (handler), uri, error);
1769 }
1770
1771 static void
1772 gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data)
1773 {
1774   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
1775
1776   iface->get_type = gst_udpsrc_uri_get_type;
1777   iface->get_protocols = gst_udpsrc_uri_get_protocols;
1778   iface->get_uri = gst_udpsrc_uri_get_uri;
1779   iface->set_uri = gst_udpsrc_uri_set_uri;
1780 }