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  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Library General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Library General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Library General Public
  18  * License along with this library; if not, write to the
  19  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  20  * Boston, MA 02110-1301, USA.
  21  */
  22
  23 /**
  24  * SECTION:element-udpsrc
  25  * @see_also: udpsink, multifdsink
  26  *
  27  * udpsrc is a network source that reads UDP packets from the network.
  28  * It can be combined with RTP depayloaders to implement RTP streaming.
  29  *
  30  * The udpsrc element supports automatic port allocation by setting the
  31  * #GstUDPSrc:port property to 0. After setting the udpsrc to PAUSED, the
  32  * allocated port can be obtained by reading the port property.
  33  *
  34  * udpsrc can read from multicast groups by setting the #GstUDPSrc:multicast-group
  35  * property to the IP address of the multicast group.
  36  *
  37  * Alternatively one can provide a custom socket to udpsrc with the #GstUDPSrc:sockfd
  38  * property, udpsrc will then not allocate a socket itself but use the provided
  39  * one.
  40  *
  41  * The #GstUDPSrc:caps property is mainly used to give a type to the UDP packet
  42  * so that they can be autoplugged in GStreamer pipelines. This is very usefull
  43  * for RTP implementations where the contents of the UDP packets is transfered
  44  * out-of-bounds using SDP or other means.
  45  *
  46  * The #GstUDPSrc:buffer-size property is used to change the default kernel
  47  * buffersizes used for receiving packets. The buffer size may be increased for
  48  * high-volume connections, or may be decreased to limit the possible backlog of
  49  * incoming data. The system places an absolute limit on these values, on Linux,
  50  * for example, the default buffer size is typically 50K and can be increased to
  51  * maximally 100K.
  52  *
  53  * The #GstUDPSrc:skip-first-bytes property is used to strip off an arbitrary
  54  * number of bytes from the start of the raw udp packet and can be used to strip
  55  * off proprietary header, for example.
  56  *
  57  * The udpsrc is always a live source. It does however not provide a #GstClock,
  58  * this is left for upstream elements such as an RTP session manager or demuxer
  59  * (such as an MPEG demuxer). As with all live sources, the captured buffers
  60  * will have their timestamp set to the current running time of the pipeline.
  61  *
  62  * udpsrc implements a #GstURIHandler interface that handles udp://host:port
  63  * type URIs.
  64  *
  65  * If the #GstUDPSrc:timeout property is set to a value bigger than 0, udpsrc
  66  * will generate an element message named
  67  * <classname>&quot;GstUDPSrcTimeout&quot;</classname>
  68  * if no data was recieved in the given timeout.
  69  * The message's structure contains one field:
  70  * <itemizedlist>
  71  * <listitem>
  72  *   <para>
  73  *   #guint64
  74  *   <classname>&quot;timeout&quot;</classname>: the timeout in microseconds that
  75  *   expired when waiting for data.
  76  *   </para>
  77  * </listitem>
  78  * </itemizedlist>
  79  * The message is typically used to detect that no UDP arrives in the receiver
  80  * because it is blocked by a firewall.
  81  * </para>
  82  * <para>
  83  * A custom file descriptor can be configured with the
  84  * #GstUDPSrc:sockfd property. The socket will be closed when setting the
  85  * element to READY by default. This behaviour can be
  86  * overriden with the #GstUDPSrc:closefd property, in which case the application
  87  * is responsible for closing the file descriptor.
  88  *
  89  * <refsect2>
  90  * <title>Examples</title>
  91  * |[
  92  * gst-launch-1.0 -v udpsrc ! fakesink dump=1
  93  * ]| A pipeline to read from the default port and dump the udp packets.
  94  * To actually generate udp packets on the default port one can use the
  95  * udpsink element. When running the following pipeline in another terminal, the
  96  * above mentioned pipeline should dump data packets to the console.
  97  * |[
  98  * gst-launch-1.0 -v audiotestsrc ! udpsink
  99  * ]|
 100  * |[
 101  * gst-launch-1.0 -v udpsrc port=0 ! fakesink
 102  * ]| read udp packets from a free port.
 103  * </refsect2>
 104  *
 105  * Last reviewed on 2007-09-20 (0.10.7)
 106  */
 107 #ifdef HAVE_CONFIG_H
 108 #include "config.h"
 109 #endif
 110
 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_BIND_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
 185 static GstFlowReturn gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf);
 186
 187 static gboolean gst_udpsrc_start (GstBaseSrc * bsrc);
 188
 189 static gboolean gst_udpsrc_stop (GstBaseSrc * bsrc);
 190
 191 static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc);
 192
 193 static gboolean gst_udpsrc_unlock_stop (GstBaseSrc * bsrc);
 194
 195 static void gst_udpsrc_finalize (GObject * object);
 196
 197 static void gst_udpsrc_set_property (GObject * object, guint prop_id,
 198     const GValue * value, GParamSpec * pspec);
 199 static void gst_udpsrc_get_property (GObject * object, guint prop_id,
 200     GValue * value, GParamSpec * pspec);
 201
 202 #define gst_udpsrc_parent_class parent_class
 203 G_DEFINE_TYPE_WITH_CODE (GstUDPSrc, gst_udpsrc, GST_TYPE_PUSH_SRC,
 204     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_udpsrc_uri_handler_init));
 205
 206 static void
 207 gst_udpsrc_class_init (GstUDPSrcClass * klass)
 208 {
 209   GObjectClass *gobject_class;
 210   GstElementClass *gstelement_class;
 211   GstBaseSrcClass *gstbasesrc_class;
 212   GstPushSrcClass *gstpushsrc_class;
 213
 214   gobject_class = (GObjectClass *) klass;
 215   gstelement_class = (GstElementClass *) klass;
 216   gstbasesrc_class = (GstBaseSrcClass *) klass;
 217   gstpushsrc_class = (GstPushSrcClass *) klass;
 218
 219   GST_DEBUG_CATEGORY_INIT (udpsrc_debug, "udpsrc", 0, "UDP src");
 220
 221   gobject_class->set_property = gst_udpsrc_set_property;
 222   gobject_class->get_property = gst_udpsrc_get_property;
 223   gobject_class->finalize = gst_udpsrc_finalize;
 224
 225   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
 226       g_param_spec_int ("port", "Port",
 227           "The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
 228           UDP_DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 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", UDP_DEFAULT_MULTICAST_GROUP,
 232           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 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
 282   /* FIXME 2.0: multicast-group and bind-address should
 283    * be separated, the former only being the multicast group and
 284    * the latter always being the address the socket is bound too,
 285    * even if a multicast group is given.
 286    */
 287   g_object_class_install_property (gobject_class, PROP_BIND_ADDRESS,
 288       g_param_spec_string ("bind-address", "Bind Address",
 289           "Address to bind the socket to. This is equivalent to the "
 290           "multicast-group property", UDP_DEFAULT_MULTICAST_GROUP,
 291           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 292
 293   gst_element_class_add_pad_template (gstelement_class,
 294       gst_static_pad_template_get (&src_template));
 295
 296   gst_element_class_set_static_metadata (gstelement_class,
 297       "UDP packet receiver", "Source/Network",
 298       "Receive data over the network via UDP",
 299       "Wim Taymans <wim@fluendo.com>, "
 300       "Thijs Vermeir <thijs.vermeir@barco.com>");
 301
 302   gstbasesrc_class->start = gst_udpsrc_start;
 303   gstbasesrc_class->stop = gst_udpsrc_stop;
 304   gstbasesrc_class->unlock = gst_udpsrc_unlock;
 305   gstbasesrc_class->unlock_stop = gst_udpsrc_unlock_stop;
 306   gstbasesrc_class->get_caps = gst_udpsrc_getcaps;
 307
 308   gstpushsrc_class->create = gst_udpsrc_create;
 309 }
 310
 311 static void
 312 gst_udpsrc_init (GstUDPSrc * udpsrc)
 313 {
 314   udpsrc->uri =
 315       g_strdup_printf ("udp://%s:%u", UDP_DEFAULT_MULTICAST_GROUP,
 316       UDP_DEFAULT_PORT);
 317
 318   udpsrc->multi_group = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 319   udpsrc->port = UDP_DEFAULT_PORT;
 320   udpsrc->socket = UDP_DEFAULT_SOCKET;
 321   udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 322   udpsrc->buffer_size = UDP_DEFAULT_BUFFER_SIZE;
 323   udpsrc->timeout = UDP_DEFAULT_TIMEOUT;
 324   udpsrc->skip_first_bytes = UDP_DEFAULT_SKIP_FIRST_BYTES;
 325   udpsrc->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
 326   udpsrc->external_socket = (udpsrc->socket != NULL);
 327   udpsrc->auto_multicast = UDP_DEFAULT_AUTO_MULTICAST;
 328   udpsrc->used_socket = UDP_DEFAULT_USED_SOCKET;
 329   udpsrc->reuse = UDP_DEFAULT_REUSE;
 330
 331   udpsrc->cancellable = g_cancellable_new ();
 332
 333   /* configure basesrc to be a live source */
 334   gst_base_src_set_live (GST_BASE_SRC (udpsrc), TRUE);
 335   /* make basesrc output a segment in time */
 336   gst_base_src_set_format (GST_BASE_SRC (udpsrc), GST_FORMAT_TIME);
 337   /* make basesrc set timestamps on outgoing buffers based on the running_time
 338    * when they were captured */
 339   gst_base_src_set_do_timestamp (GST_BASE_SRC (udpsrc), TRUE);
 340 }
 341
 342 static void
 343 gst_udpsrc_finalize (GObject * object)
 344 {
 345   GstUDPSrc *udpsrc;
 346
 347   udpsrc = GST_UDPSRC (object);
 348
 349   if (udpsrc->caps)
 350     gst_caps_unref (udpsrc->caps);
 351   udpsrc->caps = NULL;
 352
 353   g_free (udpsrc->multi_iface);
 354   udpsrc->multi_iface = NULL;
 355
 356   g_free (udpsrc->uri);
 357   udpsrc->uri = NULL;
 358
 359   g_free (udpsrc->multi_group);
 360   udpsrc->multi_group = NULL;
 361
 362   if (udpsrc->socket)
 363     g_object_unref (udpsrc->socket);
 364   udpsrc->socket = NULL;
 365
 366   if (udpsrc->used_socket)
 367     g_object_unref (udpsrc->used_socket);
 368   udpsrc->used_socket = NULL;
 369
 370   if (udpsrc->cancellable)
 371     g_object_unref (udpsrc->cancellable);
 372   udpsrc->cancellable = NULL;
 373
 374   G_OBJECT_CLASS (parent_class)->finalize (object);
 375 }
 376
 377 static GstCaps *
 378 gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter)
 379 {
 380   GstUDPSrc *udpsrc;
 381
 382   udpsrc = GST_UDPSRC (src);
 383
 384   if (udpsrc->caps) {
 385     return (filter) ? gst_caps_intersect_full (filter, udpsrc->caps,
 386         GST_CAPS_INTERSECT_FIRST) : gst_caps_ref (udpsrc->caps);
 387   } else {
 388     return (filter) ? gst_caps_ref (filter) : gst_caps_new_any ();
 389   }
 390 }
 391
 392 static GstFlowReturn
 393 gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf)
 394 {
 395   GstFlowReturn ret;
 396   GstUDPSrc *udpsrc;
 397   GstBuffer *outbuf;
 398   GstMapInfo info;
 399   GSocketAddress *saddr = NULL;
 400   gsize offset;
 401   gssize readsize;
 402   gssize res;
 403   gboolean try_again;
 404   GError *err = NULL;
 405
 406   udpsrc = GST_UDPSRC_CAST (psrc);
 407
 408 retry:
 409   /* quick check, avoid going in select when we already have data */
 410   readsize = g_socket_get_available_bytes (udpsrc->used_socket);
 411   if (readsize > 0)
 412     goto no_select;
 413
 414   do {
 415     gint64 timeout;
 416
 417     try_again = FALSE;
 418
 419     if (udpsrc->timeout)
 420       timeout = udpsrc->timeout / 1000;
 421     else
 422       timeout = -1;
 423
 424     GST_LOG_OBJECT (udpsrc, "doing select, timeout %" G_GUINT64_FORMAT,
 425         timeout);
 426
 427     if (!g_socket_condition_timed_wait (udpsrc->used_socket, G_IO_IN | G_IO_PRI,
 428             timeout, udpsrc->cancellable, &err)) {
 429       if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY)
 430           || g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 431         goto stopped;
 432       } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
 433         g_clear_error (&err);
 434         /* timeout, post element message */
 435         gst_element_post_message (GST_ELEMENT_CAST (udpsrc),
 436             gst_message_new_element (GST_OBJECT_CAST (udpsrc),
 437                 gst_structure_new ("GstUDPSrcTimeout",
 438                     "timeout", G_TYPE_UINT64, udpsrc->timeout, NULL)));
 439       } else {
 440         goto select_error;
 441       }
 442
 443       try_again = TRUE;
 444     }
 445   } while (G_UNLIKELY (try_again));
 446
 447   /* ask how much is available for reading on the socket, this should be exactly
 448    * one UDP packet. We will check the return value, though, because in some
 449    * case it can return 0 and we don't want a 0 sized buffer. */
 450   readsize = g_socket_get_available_bytes (udpsrc->used_socket);
 451   if (G_UNLIKELY (readsize < 0))
 452     goto get_available_error;
 453
 454   /* If we get here and the readsize is zero, then either select was woken up
 455    * by activity that is not a read, or a poll error occurred, or a UDP packet
 456    * was received that has no data. Since we cannot identify which case it is,
 457    * we handle all of them. This could possibly lead to a UDP packet getting
 458    * lost, but since UDP is not reliable, we can accept this. */
 459   if (G_UNLIKELY (!readsize)) {
 460     /* try to read a packet (and it will be ignored),
 461      * in case a packet with no data arrived */
 462     res =
 463         g_socket_receive_from (udpsrc->used_socket, NULL, NULL,
 464         0, udpsrc->cancellable, &err);
 465     if (G_UNLIKELY (res < 0))
 466       goto receive_error;
 467
 468     /* poll again */
 469     goto retry;
 470   }
 471
 472 no_select:
 473   GST_LOG_OBJECT (udpsrc, "ioctl says %d bytes available", (int) readsize);
 474
 475   /* sanity check value from _get_available_bytes(), which might be as
 476    * large as the kernel-side buffer on some operating systems */
 477   if (g_socket_get_family (udpsrc->used_socket) == G_SOCKET_FAMILY_IPV4)
 478     readsize = MIN (MAX_IPV4_UDP_PACKET_SIZE, readsize);
 479
 480   ret = GST_BASE_SRC_CLASS (parent_class)->alloc (GST_BASE_SRC_CAST (udpsrc),
 481       -1, readsize, &outbuf);
 482   if (ret != GST_FLOW_OK)
 483     goto alloc_failed;
 484
 485   gst_buffer_map (outbuf, &info, GST_MAP_WRITE);
 486   offset = 0;
 487
 488   if (saddr)
 489     g_object_unref (saddr);
 490   saddr = NULL;
 491
 492   res =
 493       g_socket_receive_from (udpsrc->used_socket, &saddr, (gchar *) info.data,
 494       info.size, udpsrc->cancellable, &err);
 495
 496   if (G_UNLIKELY (res < 0)) {
 497     /* EHOSTUNREACH for a UDP socket means that a packet sent with udpsink
 498      * generated a "port unreachable" ICMP response. We ignore that and try
 499      * again. */
 500     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_HOST_UNREACHABLE)) {
 501       gst_buffer_unmap (outbuf, &info);
 502       gst_buffer_unref (outbuf);
 503       outbuf = NULL;
 504       g_clear_error (&err);
 505       goto retry;
 506     }
 507     goto receive_error;
 508   }
 509
 510   /* patch offset and size when stripping off the headers */
 511   if (G_UNLIKELY (udpsrc->skip_first_bytes != 0)) {
 512     if (G_UNLIKELY (readsize < udpsrc->skip_first_bytes))
 513       goto skip_error;
 514
 515     offset += udpsrc->skip_first_bytes;
 516     res -= udpsrc->skip_first_bytes;
 517   }
 518
 519   gst_buffer_unmap (outbuf, &info);
 520   gst_buffer_resize (outbuf, offset, res);
 521
 522   /* use buffer metadata so receivers can also track the address */
 523   if (saddr) {
 524     gst_buffer_add_net_address_meta (outbuf, saddr);
 525     g_object_unref (saddr);
 526   }
 527   saddr = NULL;
 528
 529   GST_LOG_OBJECT (udpsrc, "read %d bytes", (int) readsize);
 530
 531   *buf = GST_BUFFER_CAST (outbuf);
 532
 533   return ret;
 534
 535   /* ERRORS */
 536 select_error:
 537   {
 538     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 539         ("select error: %s", err->message));
 540     g_clear_error (&err);
 541     return GST_FLOW_ERROR;
 542   }
 543 stopped:
 544   {
 545     GST_DEBUG ("stop called");
 546     g_clear_error (&err);
 547     return GST_FLOW_FLUSHING;
 548   }
 549 get_available_error:
 550   {
 551     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 552         ("get available bytes failed"));
 553     return GST_FLOW_ERROR;
 554   }
 555 alloc_failed:
 556   {
 557     GST_DEBUG ("Allocation failed");
 558     return ret;
 559   }
 560 receive_error:
 561   {
 562     gst_buffer_unmap (outbuf, &info);
 563     gst_buffer_unref (outbuf);
 564
 565     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY) ||
 566         g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 567       g_clear_error (&err);
 568       return GST_FLOW_FLUSHING;
 569     } else {
 570       GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 571           ("receive error %" G_GSSIZE_FORMAT ": %s", res, err->message));
 572       g_clear_error (&err);
 573       return GST_FLOW_ERROR;
 574     }
 575   }
 576 skip_error:
 577   {
 578     gst_buffer_unmap (outbuf, &info);
 579     gst_buffer_unref (outbuf);
 580
 581     GST_ELEMENT_ERROR (udpsrc, STREAM, DECODE, (NULL),
 582         ("UDP buffer to small to skip header"));
 583     return GST_FLOW_ERROR;
 584   }
 585 }
 586
 587 static gboolean
 588 gst_udpsrc_set_uri (GstUDPSrc * src, const gchar * uri, GError ** error)
 589 {
 590   gchar *multi_group;
 591   guint16 port;
 592
 593   if (!gst_udp_parse_uri (uri, &multi_group, &port))
 594     goto wrong_uri;
 595
 596   if (port == (guint16) - 1)
 597     port = UDP_DEFAULT_PORT;
 598
 599   g_free (src->multi_group);
 600   src->multi_group = multi_group;
 601   src->port = port;
 602
 603   g_free (src->uri);
 604   src->uri = g_strdup (uri);
 605
 606   return TRUE;
 607
 608   /* ERRORS */
 609 wrong_uri:
 610   {
 611     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 612         ("error parsing uri %s", uri));
 613     g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
 614         "Could not parse UDP URI");
 615     return FALSE;
 616   }
 617 }
 618
 619 static void
 620 gst_udpsrc_set_property (GObject * object, guint prop_id, const GValue * value,
 621     GParamSpec * pspec)
 622 {
 623   GstUDPSrc *udpsrc = GST_UDPSRC (object);
 624
 625   switch (prop_id) {
 626     case PROP_BUFFER_SIZE:
 627       udpsrc->buffer_size = g_value_get_int (value);
 628       break;
 629     case PROP_PORT:
 630       udpsrc->port = g_value_get_int (value);
 631       g_free (udpsrc->uri);
 632       udpsrc->uri =
 633           g_strdup_printf ("udp://%s:%u", udpsrc->multi_group, udpsrc->port);
 634       break;
 635     case PROP_MULTICAST_GROUP:
 636     case PROP_BIND_ADDRESS:
 637     {
 638       const gchar *group;
 639
 640       g_free (udpsrc->multi_group);
 641       if ((group = g_value_get_string (value)))
 642         udpsrc->multi_group = g_strdup (group);
 643       else
 644         udpsrc->multi_group = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 645
 646       g_free (udpsrc->uri);
 647       udpsrc->uri =
 648           g_strdup_printf ("udp://%s:%u", udpsrc->multi_group, udpsrc->port);
 649       break;
 650     }
 651     case PROP_MULTICAST_IFACE:
 652       g_free (udpsrc->multi_iface);
 653
 654       if (g_value_get_string (value) == NULL)
 655         udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 656       else
 657         udpsrc->multi_iface = g_value_dup_string (value);
 658       break;
 659     case PROP_URI:
 660       gst_udpsrc_set_uri (udpsrc, g_value_get_string (value), NULL);
 661       break;
 662     case PROP_CAPS:
 663     {
 664       const GstCaps *new_caps_val = gst_value_get_caps (value);
 665
 666       GstCaps *new_caps;
 667
 668       GstCaps *old_caps;
 669
 670       if (new_caps_val == NULL) {
 671         new_caps = gst_caps_new_any ();
 672       } else {
 673         new_caps = gst_caps_copy (new_caps_val);
 674       }
 675
 676       old_caps = udpsrc->caps;
 677       udpsrc->caps = new_caps;
 678       if (old_caps)
 679         gst_caps_unref (old_caps);
 680       gst_pad_set_caps (GST_BASE_SRC (udpsrc)->srcpad, new_caps);
 681       break;
 682     }
 683     case PROP_SOCKET:
 684       if (udpsrc->socket != NULL && udpsrc->socket != udpsrc->used_socket &&
 685           udpsrc->close_socket) {
 686         GError *err = NULL;
 687
 688         if (!g_socket_close (udpsrc->socket, &err)) {
 689           GST_ERROR ("failed to close socket %p: %s", udpsrc->socket,
 690               err->message);
 691           g_clear_error (&err);
 692         }
 693       }
 694       if (udpsrc->socket)
 695         g_object_unref (udpsrc->socket);
 696       udpsrc->socket = g_value_dup_object (value);
 697       GST_DEBUG ("setting socket to %p", udpsrc->socket);
 698       break;
 699     case PROP_TIMEOUT:
 700       udpsrc->timeout = g_value_get_uint64 (value);
 701       break;
 702     case PROP_SKIP_FIRST_BYTES:
 703       udpsrc->skip_first_bytes = g_value_get_int (value);
 704       break;
 705     case PROP_CLOSE_SOCKET:
 706       udpsrc->close_socket = g_value_get_boolean (value);
 707       break;
 708     case PROP_AUTO_MULTICAST:
 709       udpsrc->auto_multicast = g_value_get_boolean (value);
 710       break;
 711     case PROP_REUSE:
 712       udpsrc->reuse = g_value_get_boolean (value);
 713       break;
 714     default:
 715       break;
 716   }
 717 }
 718
 719 static void
 720 gst_udpsrc_get_property (GObject * object, guint prop_id, GValue * value,
 721     GParamSpec * pspec)
 722 {
 723   GstUDPSrc *udpsrc = GST_UDPSRC (object);
 724
 725   switch (prop_id) {
 726     case PROP_BUFFER_SIZE:
 727       g_value_set_int (value, udpsrc->buffer_size);
 728       break;
 729     case PROP_PORT:
 730       g_value_set_int (value, udpsrc->port);
 731       break;
 732     case PROP_MULTICAST_GROUP:
 733     case PROP_BIND_ADDRESS:
 734       g_value_set_string (value, udpsrc->multi_group);
 735       break;
 736     case PROP_MULTICAST_IFACE:
 737       g_value_set_string (value, udpsrc->multi_iface);
 738       break;
 739     case PROP_URI:
 740       g_value_set_string (value, udpsrc->uri);
 741       break;
 742     case PROP_CAPS:
 743       gst_value_set_caps (value, udpsrc->caps);
 744       break;
 745     case PROP_SOCKET:
 746       g_value_set_object (value, udpsrc->socket);
 747       break;
 748     case PROP_TIMEOUT:
 749       g_value_set_uint64 (value, udpsrc->timeout);
 750       break;
 751     case PROP_SKIP_FIRST_BYTES:
 752       g_value_set_int (value, udpsrc->skip_first_bytes);
 753       break;
 754     case PROP_CLOSE_SOCKET:
 755       g_value_set_boolean (value, udpsrc->close_socket);
 756       break;
 757     case PROP_USED_SOCKET:
 758       g_value_set_object (value, udpsrc->used_socket);
 759       break;
 760     case PROP_AUTO_MULTICAST:
 761       g_value_set_boolean (value, udpsrc->auto_multicast);
 762       break;
 763     case PROP_REUSE:
 764       g_value_set_boolean (value, udpsrc->reuse);
 765       break;
 766     default:
 767       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 768       break;
 769   }
 770 }
 771
 772 /* create a socket for sending to remote machine */
 773 static gboolean
 774 gst_udpsrc_start (GstBaseSrc * bsrc)
 775 {
 776   GstUDPSrc *src;
 777   GInetAddress *addr, *bind_addr;
 778   GSocketAddress *bind_saddr;
 779   GResolver *resolver;
 780   GError *err = NULL;
 781
 782   src = GST_UDPSRC (bsrc);
 783
 784   if (src->socket == NULL) {
 785     /* need to allocate a socket */
 786     GST_DEBUG_OBJECT (src, "allocating socket for %s:%d", src->multi_group,
 787         src->port);
 788
 789     addr = g_inet_address_new_from_string (src->multi_group);
 790     if (!addr) {
 791       GList *results;
 792
 793       GST_DEBUG_OBJECT (src, "resolving IP address for host %s",
 794           src->multi_group);
 795       resolver = g_resolver_get_default ();
 796       results =
 797           g_resolver_lookup_by_name (resolver, src->multi_group,
 798           src->cancellable, &err);
 799       if (!results)
 800         goto name_resolve;
 801       addr = G_INET_ADDRESS (g_object_ref (results->data));
 802
 803       g_resolver_free_addresses (results);
 804       g_object_unref (resolver);
 805     }
 806 #ifndef GST_DISABLE_GST_DEBUG
 807     {
 808       gchar *ip = g_inet_address_to_string (addr);
 809
 810       GST_DEBUG_OBJECT (src, "IP address for host %s is %s", src->multi_group,
 811           ip);
 812       g_free (ip);
 813     }
 814 #endif
 815
 816     if ((src->used_socket =
 817             g_socket_new (g_inet_address_get_family (addr),
 818                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 819       goto no_socket;
 820
 821     src->external_socket = FALSE;
 822
 823     GST_DEBUG_OBJECT (src, "got socket %p", src->used_socket);
 824
 825     if (src->addr)
 826       g_object_unref (src->addr);
 827     src->addr =
 828         G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
 829
 830     GST_DEBUG_OBJECT (src, "binding on port %d", src->port);
 831
 832     if (g_inet_address_get_is_multicast (addr))
 833       bind_addr = g_inet_address_new_any (g_inet_address_get_family (addr));
 834     else
 835       bind_addr = G_INET_ADDRESS (g_object_ref (addr));
 836
 837     g_object_unref (addr);
 838
 839     bind_saddr = g_inet_socket_address_new (bind_addr, src->port);
 840     g_object_unref (bind_addr);
 841     if (!g_socket_bind (src->used_socket, bind_saddr, src->reuse, &err))
 842       goto bind_error;
 843
 844     g_object_unref (bind_saddr);
 845   } else {
 846     GST_DEBUG_OBJECT (src, "using provided socket %p", src->socket);
 847     /* we use the configured socket, try to get some info about it */
 848     src->used_socket = G_SOCKET (g_object_ref (src->socket));
 849     src->external_socket = TRUE;
 850
 851     if (src->addr)
 852       g_object_unref (src->addr);
 853     src->addr =
 854         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
 855             &err));
 856     if (!src->addr)
 857       goto getsockname_error;
 858   }
 859
 860 #if GLIB_CHECK_VERSION (2, 35, 7)
 861   {
 862     gint val = 0;
 863
 864     if (src->buffer_size != 0) {
 865       GError *opt_err = NULL;
 866
 867       GST_INFO_OBJECT (src, "setting udp buffer of %d bytes", src->buffer_size);
 868       /* set buffer size, Note that on Linux this is typically limited to a
 869        * maximum of around 100K. Also a minimum of 128 bytes is required on
 870        * Linux. */
 871       if (!g_socket_set_option (src->used_socket, SOL_SOCKET, SO_RCVBUF,
 872               src->buffer_size, &opt_err)) {
 873         GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
 874             ("Could not create a buffer of requested %d bytes: %s",
 875                 src->buffer_size, opt_err->message));
 876         g_error_free (opt_err);
 877         opt_err = NULL;
 878       }
 879     }
 880
 881     /* read the value of the receive buffer. Note that on linux this returns
 882      * 2x the value we set because the kernel allocates extra memory for
 883      * metadata. The default on Linux is about 100K (which is about 50K
 884      * without metadata) */
 885     if (g_socket_get_option (src->used_socket, SOL_SOCKET, SO_RCVBUF, &val,
 886             NULL)) {
 887       GST_INFO_OBJECT (src, "have udp buffer of %d bytes", val);
 888     } else {
 889       GST_DEBUG_OBJECT (src, "could not get udp buffer size");
 890     }
 891   }
 892 #elif defined (SO_RCVBUF)
 893   {
 894     gint rcvsize, ret;
 895     socklen_t len;
 896
 897     len = sizeof (rcvsize);
 898     if (src->buffer_size != 0) {
 899       rcvsize = src->buffer_size;
 900
 901       GST_DEBUG_OBJECT (src, "setting udp buffer of %d bytes", rcvsize);
 902       /* set buffer size, Note that on Linux this is typically limited to a
 903        * maximum of around 100K. Also a minimum of 128 bytes is required on
 904        * Linux. */
 905       ret =
 906           setsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
 907           (void *) &rcvsize, len);
 908       if (ret != 0) {
 909         GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
 910             ("Could not create a buffer of requested %d bytes, %d: %s (%d)",
 911                 rcvsize, ret, g_strerror (errno), errno));
 912       }
 913     }
 914
 915     /* read the value of the receive buffer. Note that on linux this returns 2x the
 916      * value we set because the kernel allocates extra memory for metadata.
 917      * The default on Linux is about 100K (which is about 50K without metadata) */
 918     ret =
 919         getsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
 920         (void *) &rcvsize, &len);
 921     if (ret == 0)
 922       GST_DEBUG_OBJECT (src, "have udp buffer of %d bytes", rcvsize);
 923     else
 924       GST_DEBUG_OBJECT (src, "could not get udp buffer size");
 925   }
 926 #else
 927   if (src->buffer_size != 0) {
 928     GST_WARNING_OBJECT (src, "don't know how to set udp buffer size on this "
 929         "OS. Consider upgrading your GLib to >= 2.35.7 and re-compiling the "
 930         "GStreamer udp plugin");
 931   }
 932 #endif
 933
 934   g_socket_set_broadcast (src->used_socket, TRUE);
 935
 936   if (src->auto_multicast
 937       &&
 938       g_inet_address_get_is_multicast (g_inet_socket_address_get_address
 939           (src->addr))) {
 940     GST_DEBUG_OBJECT (src, "joining multicast group %s", src->multi_group);
 941     if (!g_socket_join_multicast_group (src->used_socket,
 942             g_inet_socket_address_get_address (src->addr),
 943             FALSE, src->multi_iface, &err))
 944       goto membership;
 945   }
 946
 947   /* NOTE: sockaddr_in.sin_port works for ipv4 and ipv6 because sin_port
 948    * follows ss_family on both */
 949   {
 950     GInetSocketAddress *addr;
 951     guint16 port;
 952
 953     addr =
 954         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
 955             &err));
 956     if (!addr)
 957       goto getsockname_error;
 958
 959     port = g_inet_socket_address_get_port (addr);
 960     GST_DEBUG_OBJECT (src, "bound, on port %d", port);
 961     if (port != src->port) {
 962       src->port = port;
 963       GST_DEBUG_OBJECT (src, "notifying port %d", port);
 964       g_object_notify (G_OBJECT (src), "port");
 965     }
 966     g_object_unref (addr);
 967   }
 968
 969   return TRUE;
 970
 971   /* ERRORS */
 972 name_resolve:
 973   {
 974     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
 975         ("Name resolval failed: %s", err->message));
 976     g_clear_error (&err);
 977     g_object_unref (resolver);
 978     return FALSE;
 979   }
 980 no_socket:
 981   {
 982     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 983         ("no socket error: %s", err->message));
 984     g_clear_error (&err);
 985     g_object_unref (addr);
 986     return FALSE;
 987   }
 988 bind_error:
 989   {
 990     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
 991         ("bind failed: %s", err->message));
 992     g_clear_error (&err);
 993     g_object_unref (bind_saddr);
 994     gst_udpsrc_stop (GST_BASE_SRC (src));
 995     return FALSE;
 996   }
 997 membership:
 998   {
 999     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1000         ("could add membership: %s", err->message));
1001     g_clear_error (&err);
1002     gst_udpsrc_stop (GST_BASE_SRC (src));
1003     return FALSE;
1004   }
1005 getsockname_error:
1006   {
1007     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
1008         ("getsockname failed: %s", err->message));
1009     g_clear_error (&err);
1010     gst_udpsrc_stop (GST_BASE_SRC (src));
1011     return FALSE;
1012   }
1013 }
1014
1015 static gboolean
1016 gst_udpsrc_unlock (GstBaseSrc * bsrc)
1017 {
1018   GstUDPSrc *src;
1019
1020   src = GST_UDPSRC (bsrc);
1021
1022   GST_LOG_OBJECT (src, "Flushing");
1023   g_cancellable_cancel (src->cancellable);
1024
1025   return TRUE;
1026 }
1027
1028 static gboolean
1029 gst_udpsrc_unlock_stop (GstBaseSrc * bsrc)
1030 {
1031   GstUDPSrc *src;
1032
1033   src = GST_UDPSRC (bsrc);
1034
1035   GST_LOG_OBJECT (src, "No longer flushing");
1036   g_cancellable_reset (src->cancellable);
1037
1038   return TRUE;
1039 }
1040
1041 static gboolean
1042 gst_udpsrc_stop (GstBaseSrc * bsrc)
1043 {
1044   GstUDPSrc *src;
1045
1046   src = GST_UDPSRC (bsrc);
1047
1048   GST_DEBUG ("stopping, closing sockets");
1049
1050   if (src->used_socket) {
1051     if (src->auto_multicast
1052         &&
1053         g_inet_address_get_is_multicast (g_inet_socket_address_get_address
1054             (src->addr))) {
1055       GError *err = NULL;
1056
1057       GST_DEBUG_OBJECT (src, "leaving multicast group %s", src->multi_group);
1058
1059       if (!g_socket_leave_multicast_group (src->used_socket,
1060               g_inet_socket_address_get_address (src->addr), FALSE,
1061               src->multi_iface, &err)) {
1062         GST_ERROR_OBJECT (src, "Failed to leave multicast group: %s",
1063             err->message);
1064         g_clear_error (&err);
1065       }
1066     }
1067
1068     if (src->close_socket || !src->external_socket) {
1069       GError *err = NULL;
1070       if (!g_socket_close (src->used_socket, &err)) {
1071         GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
1072         g_clear_error (&err);
1073       }
1074     }
1075
1076     g_object_unref (src->used_socket);
1077     src->used_socket = NULL;
1078     g_object_unref (src->addr);
1079     src->addr = NULL;
1080   }
1081
1082   return TRUE;
1083 }
1084
1085 /*** GSTURIHANDLER INTERFACE *************************************************/
1086
1087 static GstURIType
1088 gst_udpsrc_uri_get_type (GType type)
1089 {
1090   return GST_URI_SRC;
1091 }
1092
1093 static const gchar *const *
1094 gst_udpsrc_uri_get_protocols (GType type)
1095 {
1096   static const gchar *protocols[] = { "udp", NULL };
1097
1098   return protocols;
1099 }
1100
1101 static gchar *
1102 gst_udpsrc_uri_get_uri (GstURIHandler * handler)
1103 {
1104   GstUDPSrc *src = GST_UDPSRC (handler);
1105
1106   return g_strdup (src->uri);
1107 }
1108
1109 static gboolean
1110 gst_udpsrc_uri_set_uri (GstURIHandler * handler, const gchar * uri,
1111     GError ** error)
1112 {
1113   return gst_udpsrc_set_uri (GST_UDPSRC (handler), uri, error);
1114 }
1115
1116 static void
1117 gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data)
1118 {
1119   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
1120
1121   iface->get_type = gst_udpsrc_uri_get_type;
1122   iface->get_protocols = gst_udpsrc_uri_get_protocols;
1123   iface->get_uri = gst_udpsrc_uri_get_uri;
1124   iface->set_uri = gst_udpsrc_uri_set_uri;
1125 }