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 usefull
  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 #include <string.h>
 111 #include "gstudpsrc.h"
 112
 113 #include <gst/net/gstnetaddressmeta.h>
 114
 115 #if GLIB_CHECK_VERSION (2, 35, 7)
 116 #include <gio/gnetworking.h>
 117 #else
 118
 119 /* nicked from gnetworking.h */
 120 #ifdef G_OS_WIN32
 121 #ifndef _WIN32_WINNT
 122 #define _WIN32_WINNT 0x0501
 123 #endif
 124 #include <winsock2.h>
 125 #undef interface
 126 #include <ws2tcpip.h>           /* for socklen_t */
 127 #endif /* G_OS_WIN32 */
 128
 129 #ifdef HAVE_SYS_SOCKET_H
 130 #include <sys/socket.h>
 131 #endif
 132 #endif
 133
 134 /* not 100% correct, but a good upper bound for memory allocation purposes */
 135 #define MAX_IPV4_UDP_PACKET_SIZE (65536 - 8)
 136
 137 GST_DEBUG_CATEGORY_STATIC (udpsrc_debug);
 138 #define GST_CAT_DEFAULT (udpsrc_debug)
 139
 140 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
 141     GST_PAD_SRC,
 142     GST_PAD_ALWAYS,
 143     GST_STATIC_CAPS_ANY);
 144
 145 #define UDP_DEFAULT_PORT                5004
 146 #define UDP_DEFAULT_MULTICAST_GROUP     "0.0.0.0"
 147 #define UDP_DEFAULT_MULTICAST_IFACE     NULL
 148 #define UDP_DEFAULT_URI                 "udp://"UDP_DEFAULT_MULTICAST_GROUP":"G_STRINGIFY(UDP_DEFAULT_PORT)
 149 #define UDP_DEFAULT_CAPS                NULL
 150 #define UDP_DEFAULT_SOCKET              NULL
 151 #define UDP_DEFAULT_BUFFER_SIZE         0
 152 #define UDP_DEFAULT_TIMEOUT             0
 153 #define UDP_DEFAULT_SKIP_FIRST_BYTES    0
 154 #define UDP_DEFAULT_CLOSE_SOCKET       TRUE
 155 #define UDP_DEFAULT_USED_SOCKET        NULL
 156 #define UDP_DEFAULT_AUTO_MULTICAST     TRUE
 157 #define UDP_DEFAULT_REUSE              TRUE
 158
 159 enum
 160 {
 161   PROP_0,
 162
 163   PROP_PORT,
 164   PROP_MULTICAST_GROUP,
 165   PROP_MULTICAST_IFACE,
 166   PROP_URI,
 167   PROP_CAPS,
 168   PROP_SOCKET,
 169   PROP_BUFFER_SIZE,
 170   PROP_TIMEOUT,
 171   PROP_SKIP_FIRST_BYTES,
 172   PROP_CLOSE_SOCKET,
 173   PROP_USED_SOCKET,
 174   PROP_AUTO_MULTICAST,
 175   PROP_REUSE,
 176   PROP_ADDRESS,
 177
 178   PROP_LAST
 179 };
 180
 181 static void gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data);
 182
 183 static GstCaps *gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter);
 184 static GstFlowReturn gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf);
 185 static gboolean gst_udpsrc_close (GstUDPSrc * src);
 186 static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc);
 187 static gboolean gst_udpsrc_unlock_stop (GstBaseSrc * bsrc);
 188 static gboolean gst_udpsrc_negotiate (GstBaseSrc * basesrc);
 189
 190 static void gst_udpsrc_finalize (GObject * object);
 191
 192 static void gst_udpsrc_set_property (GObject * object, guint prop_id,
 193     const GValue * value, GParamSpec * pspec);
 194 static void gst_udpsrc_get_property (GObject * object, guint prop_id,
 195     GValue * value, GParamSpec * pspec);
 196
 197 static GstStateChangeReturn gst_udpsrc_change_state (GstElement * element,
 198     GstStateChange transition);
 199
 200 #define gst_udpsrc_parent_class parent_class
 201 G_DEFINE_TYPE_WITH_CODE (GstUDPSrc, gst_udpsrc, GST_TYPE_PUSH_SRC,
 202     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_udpsrc_uri_handler_init));
 203
 204 static void
 205 gst_udpsrc_class_init (GstUDPSrcClass * klass)
 206 {
 207   GObjectClass *gobject_class;
 208   GstElementClass *gstelement_class;
 209   GstBaseSrcClass *gstbasesrc_class;
 210   GstPushSrcClass *gstpushsrc_class;
 211
 212   gobject_class = (GObjectClass *) klass;
 213   gstelement_class = (GstElementClass *) klass;
 214   gstbasesrc_class = (GstBaseSrcClass *) klass;
 215   gstpushsrc_class = (GstPushSrcClass *) klass;
 216
 217   GST_DEBUG_CATEGORY_INIT (udpsrc_debug, "udpsrc", 0, "UDP src");
 218
 219   gobject_class->set_property = gst_udpsrc_set_property;
 220   gobject_class->get_property = gst_udpsrc_get_property;
 221   gobject_class->finalize = gst_udpsrc_finalize;
 222
 223   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
 224       g_param_spec_int ("port", "Port",
 225           "The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
 226           UDP_DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 227   /* FIXME 2.0: Remove multicast-group property */
 228 #ifndef GST_REMOVE_DEPRECATED
 229   g_object_class_install_property (gobject_class, PROP_MULTICAST_GROUP,
 230       g_param_spec_string ("multicast-group", "Multicast Group",
 231           "The Address of multicast group to join. (DEPRECATED: "
 232           "Use address property instead)", UDP_DEFAULT_MULTICAST_GROUP,
 233           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_DEPRECATED));
 234 #endif
 235   g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
 236       g_param_spec_string ("multicast-iface", "Multicast Interface",
 237           "The network interface on which to join the multicast group",
 238           UDP_DEFAULT_MULTICAST_IFACE,
 239           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 240   g_object_class_install_property (gobject_class, PROP_URI,
 241       g_param_spec_string ("uri", "URI",
 242           "URI in the form of udp://multicast_group:port", UDP_DEFAULT_URI,
 243           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 244   g_object_class_install_property (gobject_class, PROP_CAPS,
 245       g_param_spec_boxed ("caps", "Caps",
 246           "The caps of the source pad", GST_TYPE_CAPS,
 247           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 248   g_object_class_install_property (gobject_class, PROP_SOCKET,
 249       g_param_spec_object ("socket", "Socket",
 250           "Socket to use for UDP reception. (NULL == allocate)",
 251           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 252   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
 253       g_param_spec_int ("buffer-size", "Buffer Size",
 254           "Size of the kernel receive buffer in bytes, 0=default", 0, G_MAXINT,
 255           UDP_DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 256   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_TIMEOUT,
 257       g_param_spec_uint64 ("timeout", "Timeout",
 258           "Post a message after timeout nanoseconds (0 = disabled)", 0,
 259           G_MAXUINT64, UDP_DEFAULT_TIMEOUT,
 260           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 261   g_object_class_install_property (G_OBJECT_CLASS (klass),
 262       PROP_SKIP_FIRST_BYTES, g_param_spec_int ("skip-first-bytes",
 263           "Skip first bytes", "number of bytes to skip for each udp packet", 0,
 264           G_MAXINT, UDP_DEFAULT_SKIP_FIRST_BYTES,
 265           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 266   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 267       g_param_spec_boolean ("close-socket", "Close socket",
 268           "Close socket if passed as property on state change",
 269           UDP_DEFAULT_CLOSE_SOCKET,
 270           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 271   g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
 272       g_param_spec_object ("used-socket", "Socket Handle",
 273           "Socket currently in use for UDP reception. (NULL = no socket)",
 274           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 275   g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
 276       g_param_spec_boolean ("auto-multicast", "Auto Multicast",
 277           "Automatically join/leave multicast groups",
 278           UDP_DEFAULT_AUTO_MULTICAST,
 279           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 280   g_object_class_install_property (gobject_class, PROP_REUSE,
 281       g_param_spec_boolean ("reuse", "Reuse", "Enable reuse of the port",
 282           UDP_DEFAULT_REUSE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 283   g_object_class_install_property (gobject_class, PROP_ADDRESS,
 284       g_param_spec_string ("address", "Address",
 285           "Address to receive packets for. This is equivalent to the "
 286           "multicast-group property for now", UDP_DEFAULT_MULTICAST_GROUP,
 287           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 288
 289   gst_element_class_add_pad_template (gstelement_class,
 290       gst_static_pad_template_get (&src_template));
 291
 292   gst_element_class_set_static_metadata (gstelement_class,
 293       "UDP packet receiver", "Source/Network",
 294       "Receive data over the network via UDP",
 295       "Wim Taymans <wim@fluendo.com>, "
 296       "Thijs Vermeir <thijs.vermeir@barco.com>");
 297
 298   gstelement_class->change_state = gst_udpsrc_change_state;
 299
 300   gstbasesrc_class->unlock = gst_udpsrc_unlock;
 301   gstbasesrc_class->unlock_stop = gst_udpsrc_unlock_stop;
 302   gstbasesrc_class->get_caps = gst_udpsrc_getcaps;
 303   gstbasesrc_class->negotiate = gst_udpsrc_negotiate;
 304
 305   gstpushsrc_class->create = gst_udpsrc_create;
 306 }
 307
 308 static void
 309 gst_udpsrc_init (GstUDPSrc * udpsrc)
 310 {
 311   udpsrc->uri =
 312       g_strdup_printf ("udp://%s:%u", UDP_DEFAULT_MULTICAST_GROUP,
 313       UDP_DEFAULT_PORT);
 314
 315   udpsrc->address = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 316   udpsrc->port = UDP_DEFAULT_PORT;
 317   udpsrc->socket = UDP_DEFAULT_SOCKET;
 318   udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 319   udpsrc->buffer_size = UDP_DEFAULT_BUFFER_SIZE;
 320   udpsrc->timeout = UDP_DEFAULT_TIMEOUT;
 321   udpsrc->skip_first_bytes = UDP_DEFAULT_SKIP_FIRST_BYTES;
 322   udpsrc->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
 323   udpsrc->external_socket = (udpsrc->socket != NULL);
 324   udpsrc->auto_multicast = UDP_DEFAULT_AUTO_MULTICAST;
 325   udpsrc->used_socket = UDP_DEFAULT_USED_SOCKET;
 326   udpsrc->reuse = UDP_DEFAULT_REUSE;
 327
 328   udpsrc->cancellable = g_cancellable_new ();
 329
 330   /* configure basesrc to be a live source */
 331   gst_base_src_set_live (GST_BASE_SRC (udpsrc), TRUE);
 332   /* make basesrc output a segment in time */
 333   gst_base_src_set_format (GST_BASE_SRC (udpsrc), GST_FORMAT_TIME);
 334   /* make basesrc set timestamps on outgoing buffers based on the running_time
 335    * when they were captured */
 336   gst_base_src_set_do_timestamp (GST_BASE_SRC (udpsrc), TRUE);
 337 }
 338
 339 static void
 340 gst_udpsrc_finalize (GObject * object)
 341 {
 342   GstUDPSrc *udpsrc;
 343
 344   udpsrc = GST_UDPSRC (object);
 345
 346   if (udpsrc->caps)
 347     gst_caps_unref (udpsrc->caps);
 348   udpsrc->caps = NULL;
 349
 350   g_free (udpsrc->multi_iface);
 351   udpsrc->multi_iface = NULL;
 352
 353   g_free (udpsrc->uri);
 354   udpsrc->uri = NULL;
 355
 356   g_free (udpsrc->address);
 357   udpsrc->address = NULL;
 358
 359   if (udpsrc->socket)
 360     g_object_unref (udpsrc->socket);
 361   udpsrc->socket = NULL;
 362
 363   if (udpsrc->used_socket)
 364     g_object_unref (udpsrc->used_socket);
 365   udpsrc->used_socket = NULL;
 366
 367   if (udpsrc->cancellable)
 368     g_object_unref (udpsrc->cancellable);
 369   udpsrc->cancellable = NULL;
 370
 371   G_OBJECT_CLASS (parent_class)->finalize (object);
 372 }
 373
 374 static GstCaps *
 375 gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter)
 376 {
 377   GstUDPSrc *udpsrc;
 378   GstCaps *caps, *result;
 379
 380   udpsrc = GST_UDPSRC (src);
 381
 382   GST_OBJECT_LOCK (src);
 383   if ((caps = udpsrc->caps))
 384     gst_caps_ref (caps);
 385   GST_OBJECT_UNLOCK (src);
 386
 387   if (caps) {
 388     if (filter) {
 389       result = gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
 390       gst_caps_unref (caps);
 391     } else {
 392       result = caps;
 393     }
 394   } else {
 395     result = (filter) ? gst_caps_ref (filter) : gst_caps_new_any ();
 396   }
 397   return result;
 398 }
 399
 400 static void
 401 gst_udpsrc_reset_memory_allocator (GstUDPSrc * src)
 402 {
 403   if (src->mem != NULL) {
 404     gst_memory_unmap (src->mem, &src->map);
 405     gst_memory_unref (src->mem);
 406     src->mem = NULL;
 407   }
 408   if (src->mem_max != NULL) {
 409     gst_memory_unmap (src->mem_max, &src->map_max);
 410     gst_memory_unref (src->mem_max);
 411     src->mem_max = NULL;
 412   }
 413
 414   src->vec[0].buffer = NULL;
 415   src->vec[0].size = 0;
 416   src->vec[1].buffer = NULL;
 417   src->vec[1].size = 0;
 418
 419   if (src->allocator != NULL) {
 420     gst_object_unref (src->allocator);
 421     src->allocator = NULL;
 422   }
 423 }
 424
 425 static gboolean
 426 gst_udpsrc_negotiate (GstBaseSrc * basesrc)
 427 {
 428   GstUDPSrc *src = GST_UDPSRC_CAST (basesrc);
 429   gboolean ret;
 430
 431   /* just chain up to the default implementation, we just want to
 432    * retrieve the allocator at the end of it (if there is one) */
 433   ret = GST_BASE_SRC_CLASS (parent_class)->negotiate (basesrc);
 434
 435   if (ret) {
 436     GstAllocationParams new_params;
 437     GstAllocator *new_allocator = NULL;
 438
 439     /* retrieve new allocator */
 440     gst_base_src_get_allocator (basesrc, &new_allocator, &new_params);
 441
 442     if (src->allocator != new_allocator ||
 443         memcmp (&src->params, &new_params, sizeof (GstAllocationParams)) != 0) {
 444       /* drop old allocator and throw away any memory allocated with it */
 445       gst_udpsrc_reset_memory_allocator (src);
 446
 447       /* and save the new allocator and/or new allocation parameters */
 448       src->allocator = new_allocator;
 449       src->params = new_params;
 450
 451       GST_INFO_OBJECT (src, "new allocator: %" GST_PTR_FORMAT, new_allocator);
 452     }
 453   }
 454
 455   return ret;
 456 }
 457
 458 static gboolean
 459 gst_udpsrc_alloc_mem (GstUDPSrc * src, GstMemory ** p_mem, GstMapInfo * map,
 460     gsize size)
 461 {
 462   GstMemory *mem;
 463
 464   mem = gst_allocator_alloc (src->allocator, size, &src->params);
 465
 466   if (!gst_memory_map (mem, map, GST_MAP_WRITE)) {
 467     gst_memory_unref (mem);
 468     memset (map, 0, sizeof (GstMapInfo));
 469     return FALSE;
 470   }
 471   *p_mem = mem;
 472   return TRUE;
 473 }
 474
 475 static gboolean
 476 gst_udpsrc_ensure_mem (GstUDPSrc * src)
 477 {
 478   if (src->mem == NULL) {
 479     gsize mem_size = 1500;      /* typical max. MTU */
 480
 481     /* if packets are likely to be smaller, just use that size, otherwise
 482      * default to assuming incoming packets are around MTU size */
 483     if (src->max_size > 0 && src->max_size < mem_size)
 484       mem_size = src->max_size;
 485
 486     if (!gst_udpsrc_alloc_mem (src, &src->mem, &src->map, mem_size))
 487       return FALSE;
 488
 489     src->vec[0].buffer = src->map.data;
 490     src->vec[0].size = src->map.size;
 491   }
 492
 493   if (src->mem_max == NULL) {
 494     gsize max_size = MAX_IPV4_UDP_PACKET_SIZE;
 495
 496     if (!gst_udpsrc_alloc_mem (src, &src->mem_max, &src->map_max, max_size))
 497       return FALSE;
 498
 499     src->vec[1].buffer = src->map_max.data;
 500     src->vec[1].size = src->map_max.size;
 501   }
 502
 503   return TRUE;
 504 }
 505
 506 static GstFlowReturn
 507 gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf)
 508 {
 509   GstUDPSrc *udpsrc;
 510   GstBuffer *outbuf = NULL;
 511   GSocketAddress *saddr = NULL;
 512   gint flags = G_SOCKET_MSG_NONE;
 513   gboolean try_again;
 514   GError *err = NULL;
 515   gssize res;
 516   gsize offset;
 517
 518   udpsrc = GST_UDPSRC_CAST (psrc);
 519
 520   if (!gst_udpsrc_ensure_mem (udpsrc))
 521     goto memory_alloc_error;
 522
 523 retry:
 524
 525   do {
 526     gint64 timeout;
 527
 528     try_again = FALSE;
 529
 530     if (udpsrc->timeout)
 531       timeout = udpsrc->timeout / 1000;
 532     else
 533       timeout = -1;
 534
 535     GST_LOG_OBJECT (udpsrc, "doing select, timeout %" G_GINT64_FORMAT, timeout);
 536
 537     if (!g_socket_condition_timed_wait (udpsrc->used_socket, G_IO_IN | G_IO_PRI,
 538             timeout, udpsrc->cancellable, &err)) {
 539       if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY)
 540           || g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 541         goto stopped;
 542       } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
 543         g_clear_error (&err);
 544         /* timeout, post element message */
 545         gst_element_post_message (GST_ELEMENT_CAST (udpsrc),
 546             gst_message_new_element (GST_OBJECT_CAST (udpsrc),
 547                 gst_structure_new ("GstUDPSrcTimeout",
 548                     "timeout", G_TYPE_UINT64, udpsrc->timeout, NULL)));
 549       } else {
 550         goto select_error;
 551       }
 552
 553       try_again = TRUE;
 554     }
 555   } while (G_UNLIKELY (try_again));
 556
 557   if (saddr != NULL) {
 558     g_object_unref (saddr);
 559     saddr = NULL;
 560   }
 561
 562   res =
 563       g_socket_receive_message (udpsrc->used_socket, &saddr, udpsrc->vec, 2,
 564       NULL, NULL, &flags, udpsrc->cancellable, &err);
 565
 566   if (G_UNLIKELY (res < 0)) {
 567     /* EHOSTUNREACH for a UDP socket means that a packet sent with udpsink
 568      * generated a "port unreachable" ICMP response. We ignore that and try
 569      * again. */
 570     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_HOST_UNREACHABLE)) {
 571       g_clear_error (&err);
 572       goto retry;
 573     }
 574     goto receive_error;
 575   }
 576
 577   /* remember maximum packet size */
 578   if (res > udpsrc->max_size)
 579     udpsrc->max_size = res;
 580
 581   outbuf = gst_buffer_new ();
 582
 583   /* append first memory chunk to buffer */
 584   gst_buffer_append_memory (outbuf, udpsrc->mem);
 585
 586   /* if the packet didn't fit into the first chunk, add second one as well */
 587   if (res > udpsrc->map.size) {
 588     gst_buffer_append_memory (outbuf, udpsrc->mem_max);
 589     gst_memory_unmap (udpsrc->mem_max, &udpsrc->map_max);
 590     udpsrc->vec[1].buffer = NULL;
 591     udpsrc->vec[1].size = 0;
 592     udpsrc->mem_max = NULL;
 593   }
 594
 595   /* make sure we allocate a new chunk next time (we do this only here because
 596    * we look at map.size to see if the second memory chunk is needed above) */
 597   gst_memory_unmap (udpsrc->mem, &udpsrc->map);
 598   udpsrc->vec[0].buffer = NULL;
 599   udpsrc->vec[0].size = 0;
 600   udpsrc->mem = NULL;
 601
 602   offset = udpsrc->skip_first_bytes;
 603
 604   if (G_UNLIKELY (offset > 0 && res < offset))
 605     goto skip_error;
 606
 607   gst_buffer_resize (outbuf, offset, res - offset);
 608
 609   /* use buffer metadata so receivers can also track the address */
 610   if (saddr) {
 611     gst_buffer_add_net_address_meta (outbuf, saddr);
 612     g_object_unref (saddr);
 613     saddr = NULL;
 614   }
 615
 616   GST_LOG_OBJECT (udpsrc, "read packet of %d bytes", (int) res);
 617
 618   *buf = GST_BUFFER_CAST (outbuf);
 619
 620   return GST_FLOW_OK;
 621
 622   /* ERRORS */
 623 memory_alloc_error:
 624   {
 625     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 626         ("Failed to allocate or map memory"));
 627     return GST_FLOW_ERROR;
 628   }
 629 select_error:
 630   {
 631     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 632         ("select error: %s", err->message));
 633     g_clear_error (&err);
 634     return GST_FLOW_ERROR;
 635   }
 636 stopped:
 637   {
 638     GST_DEBUG ("stop called");
 639     g_clear_error (&err);
 640     return GST_FLOW_FLUSHING;
 641   }
 642 receive_error:
 643   {
 644     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY) ||
 645         g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 646       g_clear_error (&err);
 647       return GST_FLOW_FLUSHING;
 648     } else {
 649       GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 650           ("receive error %" G_GSSIZE_FORMAT ": %s", res, err->message));
 651       g_clear_error (&err);
 652       return GST_FLOW_ERROR;
 653     }
 654   }
 655 skip_error:
 656   {
 657     gst_buffer_unref (outbuf);
 658
 659     GST_ELEMENT_ERROR (udpsrc, STREAM, DECODE, (NULL),
 660         ("UDP buffer to small to skip header"));
 661     return GST_FLOW_ERROR;
 662   }
 663 }
 664
 665 static gboolean
 666 gst_udpsrc_set_uri (GstUDPSrc * src, const gchar * uri, GError ** error)
 667 {
 668   gchar *address;
 669   guint16 port;
 670
 671   if (!gst_udp_parse_uri (uri, &address, &port))
 672     goto wrong_uri;
 673
 674   if (port == (guint16) - 1)
 675     port = UDP_DEFAULT_PORT;
 676
 677   g_free (src->address);
 678   src->address = address;
 679   src->port = port;
 680
 681   g_free (src->uri);
 682   src->uri = g_strdup (uri);
 683
 684   return TRUE;
 685
 686   /* ERRORS */
 687 wrong_uri:
 688   {
 689     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 690         ("error parsing uri %s", uri));
 691     g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
 692         "Could not parse UDP URI");
 693     return FALSE;
 694   }
 695 }
 696
 697 static void
 698 gst_udpsrc_set_property (GObject * object, guint prop_id, const GValue * value,
 699     GParamSpec * pspec)
 700 {
 701   GstUDPSrc *udpsrc = GST_UDPSRC (object);
 702
 703   switch (prop_id) {
 704     case PROP_BUFFER_SIZE:
 705       udpsrc->buffer_size = g_value_get_int (value);
 706       break;
 707     case PROP_PORT:
 708       udpsrc->port = g_value_get_int (value);
 709       g_free (udpsrc->uri);
 710       udpsrc->uri =
 711           g_strdup_printf ("udp://%s:%u", udpsrc->address, udpsrc->port);
 712       break;
 713     case PROP_MULTICAST_GROUP:
 714     case PROP_ADDRESS:
 715     {
 716       const gchar *group;
 717
 718       g_free (udpsrc->address);
 719       if ((group = g_value_get_string (value)))
 720         udpsrc->address = g_strdup (group);
 721       else
 722         udpsrc->address = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 723
 724       g_free (udpsrc->uri);
 725       udpsrc->uri =
 726           g_strdup_printf ("udp://%s:%u", udpsrc->address, udpsrc->port);
 727       break;
 728     }
 729     case PROP_MULTICAST_IFACE:
 730       g_free (udpsrc->multi_iface);
 731
 732       if (g_value_get_string (value) == NULL)
 733         udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 734       else
 735         udpsrc->multi_iface = g_value_dup_string (value);
 736       break;
 737     case PROP_URI:
 738       gst_udpsrc_set_uri (udpsrc, g_value_get_string (value), NULL);
 739       break;
 740     case PROP_CAPS:
 741     {
 742       const GstCaps *new_caps_val = gst_value_get_caps (value);
 743       GstCaps *new_caps;
 744       GstCaps *old_caps;
 745
 746       if (new_caps_val == NULL) {
 747         new_caps = gst_caps_new_any ();
 748       } else {
 749         new_caps = gst_caps_copy (new_caps_val);
 750       }
 751
 752       GST_OBJECT_LOCK (udpsrc);
 753       old_caps = udpsrc->caps;
 754       udpsrc->caps = new_caps;
 755       GST_OBJECT_UNLOCK (udpsrc);
 756       if (old_caps)
 757         gst_caps_unref (old_caps);
 758
 759       gst_pad_mark_reconfigure (GST_BASE_SRC_PAD (udpsrc));
 760       break;
 761     }
 762     case PROP_SOCKET:
 763       if (udpsrc->socket != NULL && udpsrc->socket != udpsrc->used_socket &&
 764           udpsrc->close_socket) {
 765         GError *err = NULL;
 766
 767         if (!g_socket_close (udpsrc->socket, &err)) {
 768           GST_ERROR ("failed to close socket %p: %s", udpsrc->socket,
 769               err->message);
 770           g_clear_error (&err);
 771         }
 772       }
 773       if (udpsrc->socket)
 774         g_object_unref (udpsrc->socket);
 775       udpsrc->socket = g_value_dup_object (value);
 776       GST_DEBUG ("setting socket to %p", udpsrc->socket);
 777       break;
 778     case PROP_TIMEOUT:
 779       udpsrc->timeout = g_value_get_uint64 (value);
 780       break;
 781     case PROP_SKIP_FIRST_BYTES:
 782       udpsrc->skip_first_bytes = g_value_get_int (value);
 783       break;
 784     case PROP_CLOSE_SOCKET:
 785       udpsrc->close_socket = g_value_get_boolean (value);
 786       break;
 787     case PROP_AUTO_MULTICAST:
 788       udpsrc->auto_multicast = g_value_get_boolean (value);
 789       break;
 790     case PROP_REUSE:
 791       udpsrc->reuse = g_value_get_boolean (value);
 792       break;
 793     default:
 794       break;
 795   }
 796 }
 797
 798 static void
 799 gst_udpsrc_get_property (GObject * object, guint prop_id, GValue * value,
 800     GParamSpec * pspec)
 801 {
 802   GstUDPSrc *udpsrc = GST_UDPSRC (object);
 803
 804   switch (prop_id) {
 805     case PROP_BUFFER_SIZE:
 806       g_value_set_int (value, udpsrc->buffer_size);
 807       break;
 808     case PROP_PORT:
 809       g_value_set_int (value, udpsrc->port);
 810       break;
 811     case PROP_MULTICAST_GROUP:
 812     case PROP_ADDRESS:
 813       g_value_set_string (value, udpsrc->address);
 814       break;
 815     case PROP_MULTICAST_IFACE:
 816       g_value_set_string (value, udpsrc->multi_iface);
 817       break;
 818     case PROP_URI:
 819       g_value_set_string (value, udpsrc->uri);
 820       break;
 821     case PROP_CAPS:
 822       GST_OBJECT_LOCK (udpsrc);
 823       gst_value_set_caps (value, udpsrc->caps);
 824       GST_OBJECT_UNLOCK (udpsrc);
 825       break;
 826     case PROP_SOCKET:
 827       g_value_set_object (value, udpsrc->socket);
 828       break;
 829     case PROP_TIMEOUT:
 830       g_value_set_uint64 (value, udpsrc->timeout);
 831       break;
 832     case PROP_SKIP_FIRST_BYTES:
 833       g_value_set_int (value, udpsrc->skip_first_bytes);
 834       break;
 835     case PROP_CLOSE_SOCKET:
 836       g_value_set_boolean (value, udpsrc->close_socket);
 837       break;
 838     case PROP_USED_SOCKET:
 839       g_value_set_object (value, udpsrc->used_socket);
 840       break;
 841     case PROP_AUTO_MULTICAST:
 842       g_value_set_boolean (value, udpsrc->auto_multicast);
 843       break;
 844     case PROP_REUSE:
 845       g_value_set_boolean (value, udpsrc->reuse);
 846       break;
 847     default:
 848       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 849       break;
 850   }
 851 }
 852
 853 static GInetAddress *
 854 gst_udpsrc_resolve (GstUDPSrc * src, const gchar * address)
 855 {
 856   GInetAddress *addr;
 857   GError *err = NULL;
 858   GResolver *resolver;
 859
 860   addr = g_inet_address_new_from_string (address);
 861   if (!addr) {
 862     GList *results;
 863
 864     GST_DEBUG_OBJECT (src, "resolving IP address for host %s", address);
 865     resolver = g_resolver_get_default ();
 866     results =
 867         g_resolver_lookup_by_name (resolver, address, src->cancellable, &err);
 868     if (!results)
 869       goto name_resolve;
 870     addr = G_INET_ADDRESS (g_object_ref (results->data));
 871
 872     g_resolver_free_addresses (results);
 873     g_object_unref (resolver);
 874   }
 875 #ifndef GST_DISABLE_GST_DEBUG
 876   {
 877     gchar *ip = g_inet_address_to_string (addr);
 878
 879     GST_DEBUG_OBJECT (src, "IP address for host %s is %s", address, ip);
 880     g_free (ip);
 881   }
 882 #endif
 883
 884   return addr;
 885
 886 name_resolve:
 887   {
 888     GST_WARNING_OBJECT (src, "Failed to resolve %s: %s", address, err->message);
 889     g_clear_error (&err);
 890     g_object_unref (resolver);
 891     return NULL;
 892   }
 893 }
 894
 895 /* create a socket for sending to remote machine */
 896 static gboolean
 897 gst_udpsrc_open (GstUDPSrc * src)
 898 {
 899   GInetAddress *addr, *bind_addr;
 900   GSocketAddress *bind_saddr;
 901   GError *err = NULL;
 902
 903   if (src->socket == NULL) {
 904     /* need to allocate a socket */
 905     GST_DEBUG_OBJECT (src, "allocating socket for %s:%d", src->address,
 906         src->port);
 907
 908     addr = gst_udpsrc_resolve (src, src->address);
 909     if (!addr)
 910       goto name_resolve;
 911
 912     if ((src->used_socket =
 913             g_socket_new (g_inet_address_get_family (addr),
 914                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 915       goto no_socket;
 916
 917     src->external_socket = FALSE;
 918
 919     GST_DEBUG_OBJECT (src, "got socket %p", src->used_socket);
 920
 921     if (src->addr)
 922       g_object_unref (src->addr);
 923     src->addr =
 924         G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
 925
 926     GST_DEBUG_OBJECT (src, "binding on port %d", src->port);
 927
 928     /* On Windows it's not possible to bind to a multicast address
 929      * but the OS will make sure to filter out all packets that
 930      * arrive not for the multicast address the socket joined.
 931      *
 932      * On Linux and others it is necessary to bind to a multicast
 933      * address to let the OS filter out all packets that are received
 934      * on the same port but for different addresses than the multicast
 935      * address
 936      */
 937 #ifdef G_OS_WIN32
 938     if (g_inet_address_get_is_multicast (addr))
 939       bind_addr = g_inet_address_new_any (g_inet_address_get_family (addr));
 940     else
 941 #endif
 942       bind_addr = G_INET_ADDRESS (g_object_ref (addr));
 943
 944     g_object_unref (addr);
 945
 946     bind_saddr = g_inet_socket_address_new (bind_addr, src->port);
 947     g_object_unref (bind_addr);
 948     if (!g_socket_bind (src->used_socket, bind_saddr, src->reuse, &err))
 949       goto bind_error;
 950
 951     g_object_unref (bind_saddr);
 952   } else {
 953     GST_DEBUG_OBJECT (src, "using provided socket %p", src->socket);
 954     /* we use the configured socket, try to get some info about it */
 955     src->used_socket = G_SOCKET (g_object_ref (src->socket));
 956     src->external_socket = TRUE;
 957
 958     if (src->addr)
 959       g_object_unref (src->addr);
 960     src->addr =
 961         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
 962             &err));
 963     if (!src->addr)
 964       goto getsockname_error;
 965   }
 966
 967 #if GLIB_CHECK_VERSION (2, 35, 7)
 968   {
 969     gint val = 0;
 970
 971     if (src->buffer_size != 0) {
 972       GError *opt_err = NULL;
 973
 974       GST_INFO_OBJECT (src, "setting udp buffer of %d bytes", src->buffer_size);
 975       /* set buffer size, Note that on Linux this is typically limited to a
 976        * maximum of around 100K. Also a minimum of 128 bytes is required on
 977        * Linux. */
 978       if (!g_socket_set_option (src->used_socket, SOL_SOCKET, SO_RCVBUF,
 979               src->buffer_size, &opt_err)) {
 980         GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
 981             ("Could not create a buffer of requested %d bytes: %s",
 982                 src->buffer_size, opt_err->message));
 983         g_error_free (opt_err);
 984         opt_err = NULL;
 985       }
 986     }
 987
 988     /* read the value of the receive buffer. Note that on linux this returns
 989      * 2x the value we set because the kernel allocates extra memory for
 990      * metadata. The default on Linux is about 100K (which is about 50K
 991      * without metadata) */
 992     if (g_socket_get_option (src->used_socket, SOL_SOCKET, SO_RCVBUF, &val,
 993             NULL)) {
 994       GST_INFO_OBJECT (src, "have udp buffer of %d bytes", val);
 995     } else {
 996       GST_DEBUG_OBJECT (src, "could not get udp buffer size");
 997     }
 998   }
 999 #elif defined (SO_RCVBUF)
1000   {
1001     gint rcvsize, ret;
1002     socklen_t len;
1003
1004     len = sizeof (rcvsize);
1005     if (src->buffer_size != 0) {
1006       rcvsize = src->buffer_size;
1007
1008       GST_DEBUG_OBJECT (src, "setting udp buffer of %d bytes", rcvsize);
1009       /* set buffer size, Note that on Linux this is typically limited to a
1010        * maximum of around 100K. Also a minimum of 128 bytes is required on
1011        * Linux. */
1012       ret =
1013           setsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
1014           (void *) &rcvsize, len);
1015       if (ret != 0) {
1016         GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
1017             ("Could not create a buffer of requested %d bytes, %d: %s (%d)",
1018                 rcvsize, ret, g_strerror (errno), errno));
1019       }
1020     }
1021
1022     /* read the value of the receive buffer. Note that on linux this returns 2x the
1023      * value we set because the kernel allocates extra memory for metadata.
1024      * The default on Linux is about 100K (which is about 50K without metadata) */
1025     ret =
1026         getsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
1027         (void *) &rcvsize, &len);
1028     if (ret == 0)
1029       GST_DEBUG_OBJECT (src, "have udp buffer of %d bytes", rcvsize);
1030     else
1031       GST_DEBUG_OBJECT (src, "could not get udp buffer size");
1032   }
1033 #else
1034   if (src->buffer_size != 0) {
1035     GST_WARNING_OBJECT (src, "don't know how to set udp buffer size on this "
1036         "OS. Consider upgrading your GLib to >= 2.35.7 and re-compiling the "
1037         "GStreamer udp plugin");
1038   }
1039 #endif
1040
1041   g_socket_set_broadcast (src->used_socket, TRUE);
1042
1043   if (src->auto_multicast
1044       &&
1045       g_inet_address_get_is_multicast (g_inet_socket_address_get_address
1046           (src->addr))) {
1047     GST_DEBUG_OBJECT (src, "joining multicast group %s", src->address);
1048     if (!g_socket_join_multicast_group (src->used_socket,
1049             g_inet_socket_address_get_address (src->addr),
1050             FALSE, src->multi_iface, &err))
1051       goto membership;
1052   }
1053
1054   /* NOTE: sockaddr_in.sin_port works for ipv4 and ipv6 because sin_port
1055    * follows ss_family on both */
1056   {
1057     GInetSocketAddress *addr;
1058     guint16 port;
1059
1060     addr =
1061         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
1062             &err));
1063     if (!addr)
1064       goto getsockname_error;
1065
1066     port = g_inet_socket_address_get_port (addr);
1067     GST_DEBUG_OBJECT (src, "bound, on port %d", port);
1068     if (port != src->port) {
1069       src->port = port;
1070       GST_DEBUG_OBJECT (src, "notifying port %d", port);
1071       g_object_notify (G_OBJECT (src), "port");
1072     }
1073     g_object_unref (addr);
1074   }
1075
1076   src->allocator = NULL;
1077   gst_allocation_params_init (&src->params);
1078
1079   src->max_size = 0;
1080
1081   return TRUE;
1082
1083   /* ERRORS */
1084 name_resolve:
1085   {
1086     return FALSE;
1087   }
1088 no_socket:
1089   {
1090     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
1091         ("no socket error: %s", err->message));
1092     g_clear_error (&err);
1093     g_object_unref (addr);
1094     return FALSE;
1095   }
1096 bind_error:
1097   {
1098     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1099         ("bind failed: %s", err->message));
1100     g_clear_error (&err);
1101     g_object_unref (bind_saddr);
1102     gst_udpsrc_close (src);
1103     return FALSE;
1104   }
1105 membership:
1106   {
1107     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1108         ("could add membership: %s", err->message));
1109     g_clear_error (&err);
1110     gst_udpsrc_close (src);
1111     return FALSE;
1112   }
1113 getsockname_error:
1114   {
1115     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1116         ("getsockname failed: %s", err->message));
1117     g_clear_error (&err);
1118     gst_udpsrc_close (src);
1119     return FALSE;
1120   }
1121 }
1122
1123 static gboolean
1124 gst_udpsrc_unlock (GstBaseSrc * bsrc)
1125 {
1126   GstUDPSrc *src;
1127
1128   src = GST_UDPSRC (bsrc);
1129
1130   GST_LOG_OBJECT (src, "Flushing");
1131   g_cancellable_cancel (src->cancellable);
1132
1133   return TRUE;
1134 }
1135
1136 static gboolean
1137 gst_udpsrc_unlock_stop (GstBaseSrc * bsrc)
1138 {
1139   GstUDPSrc *src;
1140
1141   src = GST_UDPSRC (bsrc);
1142
1143   GST_LOG_OBJECT (src, "No longer flushing");
1144   g_cancellable_reset (src->cancellable);
1145
1146   return TRUE;
1147 }
1148
1149 static gboolean
1150 gst_udpsrc_close (GstUDPSrc * src)
1151 {
1152   GST_DEBUG ("closing sockets");
1153
1154   if (src->used_socket) {
1155     if (src->auto_multicast
1156         &&
1157         g_inet_address_get_is_multicast (g_inet_socket_address_get_address
1158             (src->addr))) {
1159       GError *err = NULL;
1160
1161       GST_DEBUG_OBJECT (src, "leaving multicast group %s", src->address);
1162
1163       if (!g_socket_leave_multicast_group (src->used_socket,
1164               g_inet_socket_address_get_address (src->addr), FALSE,
1165               src->multi_iface, &err)) {
1166         GST_ERROR_OBJECT (src, "Failed to leave multicast group: %s",
1167             err->message);
1168         g_clear_error (&err);
1169       }
1170     }
1171
1172     if (src->close_socket || !src->external_socket) {
1173       GError *err = NULL;
1174       if (!g_socket_close (src->used_socket, &err)) {
1175         GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
1176         g_clear_error (&err);
1177       }
1178     }
1179
1180     g_object_unref (src->used_socket);
1181     src->used_socket = NULL;
1182     g_object_unref (src->addr);
1183     src->addr = NULL;
1184   }
1185
1186   gst_udpsrc_reset_memory_allocator (src);
1187
1188   return TRUE;
1189 }
1190
1191
1192 static GstStateChangeReturn
1193 gst_udpsrc_change_state (GstElement * element, GstStateChange transition)
1194 {
1195   GstUDPSrc *src;
1196   GstStateChangeReturn result;
1197
1198   src = GST_UDPSRC (element);
1199
1200   switch (transition) {
1201     case GST_STATE_CHANGE_NULL_TO_READY:
1202       if (!gst_udpsrc_open (src))
1203         goto open_failed;
1204       break;
1205     default:
1206       break;
1207   }
1208   if ((result =
1209           GST_ELEMENT_CLASS (parent_class)->change_state (element,
1210               transition)) == GST_STATE_CHANGE_FAILURE)
1211     goto failure;
1212
1213   switch (transition) {
1214     case GST_STATE_CHANGE_READY_TO_NULL:
1215       gst_udpsrc_close (src);
1216       break;
1217     default:
1218       break;
1219   }
1220   return result;
1221   /* ERRORS */
1222 open_failed:
1223   {
1224     GST_DEBUG_OBJECT (src, "failed to open socket");
1225     return GST_STATE_CHANGE_FAILURE;
1226   }
1227 failure:
1228   {
1229     GST_DEBUG_OBJECT (src, "parent failed state change");
1230     return result;
1231   }
1232 }
1233
1234
1235
1236
1237 /*** GSTURIHANDLER INTERFACE *************************************************/
1238
1239 static GstURIType
1240 gst_udpsrc_uri_get_type (GType type)
1241 {
1242   return GST_URI_SRC;
1243 }
1244
1245 static const gchar *const *
1246 gst_udpsrc_uri_get_protocols (GType type)
1247 {
1248   static const gchar *protocols[] = { "udp", NULL };
1249
1250   return protocols;
1251 }
1252
1253 static gchar *
1254 gst_udpsrc_uri_get_uri (GstURIHandler * handler)
1255 {
1256   GstUDPSrc *src = GST_UDPSRC (handler);
1257
1258   return g_strdup (src->uri);
1259 }
1260
1261 static gboolean
1262 gst_udpsrc_uri_set_uri (GstURIHandler * handler, const gchar * uri,
1263     GError ** error)
1264 {
1265   return gst_udpsrc_set_uri (GST_UDPSRC (handler), uri, error);
1266 }
1267
1268 static void
1269 gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data)
1270 {
1271   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
1272
1273   iface->get_type = gst_udpsrc_uri_get_type;
1274   iface->get_protocols = gst_udpsrc_uri_get_protocols;
1275   iface->get_uri = gst_udpsrc_uri_get_uri;
1276   iface->set_uri = gst_udpsrc_uri_set_uri;
1277 }