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