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., 59 Temple Place - Suite 330,
  20  * Boston, MA 02111-1307, 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 -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 -v audiotestsrc ! udpsink
  99  * ]|
 100  * |[
 101  * gst-launch -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 #ifdef HAVE_SYS_SOCKET_H
 116 #include <sys/socket.h>
 117 #endif
 118
 119 GST_DEBUG_CATEGORY_STATIC (udpsrc_debug);
 120 #define GST_CAT_DEFAULT (udpsrc_debug)
 121
 122 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
 123     GST_PAD_SRC,
 124     GST_PAD_ALWAYS,
 125     GST_STATIC_CAPS_ANY);
 126
 127 #define UDP_DEFAULT_PORT                5004
 128 #define UDP_DEFAULT_MULTICAST_GROUP     "0.0.0.0"
 129 #define UDP_DEFAULT_MULTICAST_IFACE     NULL
 130 #define UDP_DEFAULT_URI                 "udp://"UDP_DEFAULT_MULTICAST_GROUP":"G_STRINGIFY(UDP_DEFAULT_PORT)
 131 #define UDP_DEFAULT_CAPS                NULL
 132 #define UDP_DEFAULT_SOCKET              NULL
 133 #define UDP_DEFAULT_BUFFER_SIZE         0
 134 #define UDP_DEFAULT_TIMEOUT             0
 135 #define UDP_DEFAULT_SKIP_FIRST_BYTES    0
 136 #define UDP_DEFAULT_CLOSE_SOCKET       TRUE
 137 #define UDP_DEFAULT_USED_SOCKET        NULL
 138 #define UDP_DEFAULT_AUTO_MULTICAST     TRUE
 139 #define UDP_DEFAULT_REUSE              TRUE
 140
 141 enum
 142 {
 143   PROP_0,
 144
 145   PROP_PORT,
 146   PROP_MULTICAST_GROUP,
 147   PROP_MULTICAST_IFACE,
 148   PROP_URI,
 149   PROP_CAPS,
 150   PROP_SOCKET,
 151   PROP_BUFFER_SIZE,
 152   PROP_TIMEOUT,
 153   PROP_SKIP_FIRST_BYTES,
 154   PROP_CLOSE_SOCKET,
 155   PROP_USED_SOCKET,
 156   PROP_AUTO_MULTICAST,
 157   PROP_REUSE,
 158
 159   PROP_LAST
 160 };
 161
 162 static void gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data);
 163
 164 static GstCaps *gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter);
 165
 166 static GstFlowReturn gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf);
 167
 168 static gboolean gst_udpsrc_start (GstBaseSrc * bsrc);
 169
 170 static gboolean gst_udpsrc_stop (GstBaseSrc * bsrc);
 171
 172 static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc);
 173
 174 static gboolean gst_udpsrc_unlock_stop (GstBaseSrc * bsrc);
 175
 176 static void gst_udpsrc_finalize (GObject * object);
 177
 178 static void gst_udpsrc_set_property (GObject * object, guint prop_id,
 179     const GValue * value, GParamSpec * pspec);
 180 static void gst_udpsrc_get_property (GObject * object, guint prop_id,
 181     GValue * value, GParamSpec * pspec);
 182
 183 #define gst_udpsrc_parent_class parent_class
 184 G_DEFINE_TYPE_WITH_CODE (GstUDPSrc, gst_udpsrc, GST_TYPE_PUSH_SRC,
 185     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_udpsrc_uri_handler_init));
 186
 187 static void
 188 gst_udpsrc_class_init (GstUDPSrcClass * klass)
 189 {
 190   GObjectClass *gobject_class;
 191   GstElementClass *gstelement_class;
 192   GstBaseSrcClass *gstbasesrc_class;
 193   GstPushSrcClass *gstpushsrc_class;
 194
 195   gobject_class = (GObjectClass *) klass;
 196   gstelement_class = (GstElementClass *) klass;
 197   gstbasesrc_class = (GstBaseSrcClass *) klass;
 198   gstpushsrc_class = (GstPushSrcClass *) klass;
 199
 200   GST_DEBUG_CATEGORY_INIT (udpsrc_debug, "udpsrc", 0, "UDP src");
 201
 202   gobject_class->set_property = gst_udpsrc_set_property;
 203   gobject_class->get_property = gst_udpsrc_get_property;
 204   gobject_class->finalize = gst_udpsrc_finalize;
 205
 206   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
 207       g_param_spec_int ("port", "Port",
 208           "The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
 209           UDP_DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 210   g_object_class_install_property (gobject_class, PROP_MULTICAST_GROUP,
 211       g_param_spec_string ("multicast-group", "Multicast Group",
 212           "The Address of multicast group to join", UDP_DEFAULT_MULTICAST_GROUP,
 213           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 214   g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
 215       g_param_spec_string ("multicast-iface", "Multicast Interface",
 216           "The network interface on which to join the multicast group",
 217           UDP_DEFAULT_MULTICAST_IFACE,
 218           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 219   g_object_class_install_property (gobject_class, PROP_URI,
 220       g_param_spec_string ("uri", "URI",
 221           "URI in the form of udp://multicast_group:port", UDP_DEFAULT_URI,
 222           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 223   g_object_class_install_property (gobject_class, PROP_CAPS,
 224       g_param_spec_boxed ("caps", "Caps",
 225           "The caps of the source pad", GST_TYPE_CAPS,
 226           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 227   g_object_class_install_property (gobject_class, PROP_SOCKET,
 228       g_param_spec_object ("socket", "Socket",
 229           "Socket to use for UDP reception. (NULL == allocate)",
 230           G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 231   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
 232       g_param_spec_int ("buffer-size", "Buffer Size",
 233           "Size of the kernel receive buffer in bytes, 0=default", 0, G_MAXINT,
 234           UDP_DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 235   g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_TIMEOUT,
 236       g_param_spec_uint64 ("timeout", "Timeout",
 237           "Post a message after timeout nanoseconds (0 = disabled)", 0,
 238           G_MAXUINT64, UDP_DEFAULT_TIMEOUT,
 239           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 240   g_object_class_install_property (G_OBJECT_CLASS (klass),
 241       PROP_SKIP_FIRST_BYTES, g_param_spec_int ("skip-first-bytes",
 242           "Skip first bytes", "number of bytes to skip for each udp packet", 0,
 243           G_MAXINT, UDP_DEFAULT_SKIP_FIRST_BYTES,
 244           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 245   g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
 246       g_param_spec_boolean ("close-socket", "Close socket",
 247           "Close socket if passed as property on state change",
 248           UDP_DEFAULT_CLOSE_SOCKET,
 249           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 250   g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
 251       g_param_spec_object ("used-socket", "Socket Handle",
 252           "Socket currently in use for UDP reception. (NULL = no socket)",
 253           G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 254   g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
 255       g_param_spec_boolean ("auto-multicast", "Auto Multicast",
 256           "Automatically join/leave multicast groups",
 257           UDP_DEFAULT_AUTO_MULTICAST,
 258           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 259   g_object_class_install_property (gobject_class, PROP_REUSE,
 260       g_param_spec_boolean ("reuse", "Reuse", "Enable reuse of the port",
 261           UDP_DEFAULT_REUSE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 262
 263   gst_element_class_add_pad_template (gstelement_class,
 264       gst_static_pad_template_get (&src_template));
 265
 266   gst_element_class_set_details_simple (gstelement_class, "UDP packet receiver",
 267       "Source/Network",
 268       "Receive data over the network via UDP",
 269       "Wim Taymans <wim@fluendo.com>, "
 270       "Thijs Vermeir <thijs.vermeir@barco.com>");
 271
 272   gstbasesrc_class->start = gst_udpsrc_start;
 273   gstbasesrc_class->stop = gst_udpsrc_stop;
 274   gstbasesrc_class->unlock = gst_udpsrc_unlock;
 275   gstbasesrc_class->unlock_stop = gst_udpsrc_unlock_stop;
 276   gstbasesrc_class->get_caps = gst_udpsrc_getcaps;
 277
 278   gstpushsrc_class->create = gst_udpsrc_create;
 279 }
 280
 281 static void
 282 gst_udpsrc_init (GstUDPSrc * udpsrc)
 283 {
 284   udpsrc->uri =
 285       g_strdup_printf ("udp://%s:%u", UDP_DEFAULT_MULTICAST_GROUP,
 286       UDP_DEFAULT_PORT);
 287
 288   udpsrc->host = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 289   udpsrc->port = UDP_DEFAULT_PORT;
 290   udpsrc->socket = UDP_DEFAULT_SOCKET;
 291   udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 292   udpsrc->buffer_size = UDP_DEFAULT_BUFFER_SIZE;
 293   udpsrc->timeout = UDP_DEFAULT_TIMEOUT;
 294   udpsrc->skip_first_bytes = UDP_DEFAULT_SKIP_FIRST_BYTES;
 295   udpsrc->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
 296   udpsrc->external_socket = (udpsrc->socket != NULL);
 297   udpsrc->auto_multicast = UDP_DEFAULT_AUTO_MULTICAST;
 298   udpsrc->used_socket = UDP_DEFAULT_USED_SOCKET;
 299   udpsrc->reuse = UDP_DEFAULT_REUSE;
 300
 301   udpsrc->cancellable = g_cancellable_new ();
 302
 303   /* configure basesrc to be a live source */
 304   gst_base_src_set_live (GST_BASE_SRC (udpsrc), TRUE);
 305   /* make basesrc output a segment in time */
 306   gst_base_src_set_format (GST_BASE_SRC (udpsrc), GST_FORMAT_TIME);
 307   /* make basesrc set timestamps on outgoing buffers based on the running_time
 308    * when they were captured */
 309   gst_base_src_set_do_timestamp (GST_BASE_SRC (udpsrc), TRUE);
 310 }
 311
 312 static void
 313 gst_udpsrc_finalize (GObject * object)
 314 {
 315   GstUDPSrc *udpsrc;
 316
 317   udpsrc = GST_UDPSRC (object);
 318
 319   if (udpsrc->caps)
 320     gst_caps_unref (udpsrc->caps);
 321   udpsrc->caps = NULL;
 322
 323   g_free (udpsrc->multi_iface);
 324   udpsrc->multi_iface = NULL;
 325
 326   g_free (udpsrc->uri);
 327   udpsrc->uri = NULL;
 328
 329   g_free (udpsrc->host);
 330   udpsrc->host = NULL;
 331
 332   if (udpsrc->socket)
 333     g_object_unref (udpsrc->socket);
 334   udpsrc->socket = NULL;
 335
 336   if (udpsrc->used_socket)
 337     g_object_unref (udpsrc->used_socket);
 338   udpsrc->used_socket = NULL;
 339
 340   if (udpsrc->cancellable)
 341     g_object_unref (udpsrc->cancellable);
 342   udpsrc->cancellable = NULL;
 343
 344   G_OBJECT_CLASS (parent_class)->finalize (object);
 345 }
 346
 347 static GstCaps *
 348 gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter)
 349 {
 350   GstUDPSrc *udpsrc;
 351
 352   udpsrc = GST_UDPSRC (src);
 353
 354   if (udpsrc->caps) {
 355     return (filter) ? gst_caps_intersect_full (filter, udpsrc->caps,
 356         GST_CAPS_INTERSECT_FIRST) : gst_caps_ref (udpsrc->caps);
 357   } else {
 358     return (filter) ? gst_caps_ref (filter) : gst_caps_new_any ();
 359   }
 360 }
 361
 362 static GstFlowReturn
 363 gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf)
 364 {
 365   GstUDPSrc *udpsrc;
 366   GstBuffer *outbuf;
 367   GSocketAddress *saddr = NULL;
 368   guint8 *pktdata;
 369   gint pktsize;
 370   gsize offset;
 371   gssize readsize;
 372   gssize ret;
 373   gboolean try_again;
 374   GError *err = NULL;
 375
 376   udpsrc = GST_UDPSRC_CAST (psrc);
 377
 378 retry:
 379   /* quick check, avoid going in select when we already have data */
 380   readsize = g_socket_get_available_bytes (udpsrc->used_socket);
 381   if (readsize > 0)
 382     goto no_select;
 383
 384   do {
 385     try_again = FALSE;
 386
 387     GST_LOG_OBJECT (udpsrc, "doing select, timeout %" G_GUINT64_FORMAT,
 388         udpsrc->timeout);
 389
 390     if (!g_socket_condition_wait (udpsrc->used_socket, G_IO_IN | G_IO_PRI,
 391             udpsrc->cancellable, &err)) {
 392       if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY)
 393           || g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 394         goto stopped;
 395       } else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
 396         /* timeout, post element message */
 397         gst_element_post_message (GST_ELEMENT_CAST (udpsrc),
 398             gst_message_new_element (GST_OBJECT_CAST (udpsrc),
 399                 gst_structure_new ("GstUDPSrcTimeout",
 400                     "timeout", G_TYPE_UINT64, udpsrc->timeout, NULL)));
 401       } else {
 402         goto select_error;
 403       }
 404
 405       try_again = TRUE;
 406     }
 407   } while (G_UNLIKELY (try_again));
 408
 409   /* ask how much is available for reading on the socket, this should be exactly
 410    * one UDP packet. We will check the return value, though, because in some
 411    * case it can return 0 and we don't want a 0 sized buffer. */
 412   readsize = g_socket_get_available_bytes (udpsrc->used_socket);
 413   if (G_UNLIKELY (readsize < 0))
 414     goto get_available_error;
 415
 416   /* If we get here and the readsize is zero, then either select was woken up
 417    * by activity that is not a read, or a poll error occurred, or a UDP packet
 418    * was received that has no data. Since we cannot identify which case it is,
 419    * we handle all of them. This could possibly lead to a UDP packet getting
 420    * lost, but since UDP is not reliable, we can accept this. */
 421   if (G_UNLIKELY (!readsize)) {
 422     /* try to read a packet (and it will be ignored),
 423      * in case a packet with no data arrived */
 424
 425     pktdata = NULL;
 426     pktsize = 0;
 427     ret =
 428         g_socket_receive_from (udpsrc->used_socket, NULL, (gchar *) pktdata,
 429         pktsize, udpsrc->cancellable, &err);
 430     if (G_UNLIKELY (ret < 0))
 431       goto receive_error;
 432
 433     /* poll again */
 434     goto retry;
 435   }
 436
 437 no_select:
 438   GST_LOG_OBJECT (udpsrc, "ioctl says %d bytes available", (int) readsize);
 439
 440   pktdata = g_malloc (readsize);
 441   pktsize = readsize;
 442   offset = 0;
 443
 444   if (saddr)
 445     g_object_unref (saddr);
 446   saddr = NULL;
 447
 448   ret =
 449       g_socket_receive_from (udpsrc->used_socket, &saddr, (gchar *) pktdata,
 450       pktsize, udpsrc->cancellable, &err);
 451
 452   if (G_UNLIKELY (ret < 0))
 453     goto receive_error;
 454
 455   /* patch pktdata and len when stripping off the headers */
 456   if (G_UNLIKELY (udpsrc->skip_first_bytes != 0)) {
 457     if (G_UNLIKELY (readsize < udpsrc->skip_first_bytes))
 458       goto skip_error;
 459
 460     offset += udpsrc->skip_first_bytes;
 461     ret -= udpsrc->skip_first_bytes;
 462   }
 463
 464   outbuf = gst_buffer_new ();
 465   gst_buffer_append_memory (outbuf,
 466       gst_memory_new_wrapped (0, pktdata, pktsize, offset, ret, pktdata,
 467           g_free));
 468
 469   /* use buffer metadata so receivers can also track the address */
 470   if (saddr) {
 471     gst_buffer_add_net_address_meta (outbuf, saddr);
 472     g_object_unref (saddr);
 473   }
 474   saddr = NULL;
 475
 476   GST_LOG_OBJECT (udpsrc, "read %d bytes", (int) readsize);
 477
 478   *buf = GST_BUFFER_CAST (outbuf);
 479
 480   return GST_FLOW_OK;
 481
 482   /* ERRORS */
 483 select_error:
 484   {
 485     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 486         ("select error: %s", err->message));
 487     g_clear_error (&err);
 488     return GST_FLOW_ERROR;
 489   }
 490 stopped:
 491   {
 492     GST_DEBUG ("stop called");
 493     return GST_FLOW_FLUSHING;
 494   }
 495 get_available_error:
 496   {
 497     GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 498         ("get available bytes failed"));
 499     return GST_FLOW_ERROR;
 500   }
 501 receive_error:
 502   {
 503     g_free (pktdata);
 504
 505     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY) ||
 506         g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
 507       g_clear_error (&err);
 508       return GST_FLOW_FLUSHING;
 509     } else {
 510       GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
 511           ("receive error %" G_GSSIZE_FORMAT ": %s", ret, err->message));
 512       g_clear_error (&err);
 513       return GST_FLOW_ERROR;
 514     }
 515   }
 516 skip_error:
 517   {
 518     GST_ELEMENT_ERROR (udpsrc, STREAM, DECODE, (NULL),
 519         ("UDP buffer to small to skip header"));
 520     return GST_FLOW_ERROR;
 521   }
 522 }
 523
 524 static gboolean
 525 gst_udpsrc_set_uri (GstUDPSrc * src, const gchar * uri, GError ** error)
 526 {
 527   gchar *host;
 528   guint16 port;
 529
 530   if (!gst_udp_parse_uri (uri, &host, &port))
 531     goto wrong_uri;
 532
 533   if (port == (guint16) - 1)
 534     port = UDP_DEFAULT_PORT;
 535
 536   g_free (src->host);
 537   src->host = host;
 538   src->port = port;
 539
 540   g_free (src->uri);
 541   src->uri = g_strdup (uri);
 542
 543   return TRUE;
 544
 545   /* ERRORS */
 546 wrong_uri:
 547   {
 548     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
 549         ("error parsing uri %s", uri));
 550     g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
 551         "Could not parse UDP URI");
 552     return FALSE;
 553   }
 554 }
 555
 556 static void
 557 gst_udpsrc_set_property (GObject * object, guint prop_id, const GValue * value,
 558     GParamSpec * pspec)
 559 {
 560   GstUDPSrc *udpsrc = GST_UDPSRC (object);
 561
 562   switch (prop_id) {
 563     case PROP_BUFFER_SIZE:
 564       udpsrc->buffer_size = g_value_get_int (value);
 565       break;
 566     case PROP_PORT:
 567       udpsrc->port = g_value_get_int (value);
 568       g_free (udpsrc->uri);
 569       udpsrc->uri = g_strdup_printf ("udp://%s:%u", udpsrc->host, udpsrc->port);
 570       break;
 571     case PROP_MULTICAST_GROUP:
 572     {
 573       const gchar *group;
 574
 575       g_free (udpsrc->host);
 576       if ((group = g_value_get_string (value)))
 577         udpsrc->host = g_strdup (group);
 578       else
 579         udpsrc->host = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 580
 581       g_free (udpsrc->uri);
 582       udpsrc->uri = g_strdup_printf ("udp://%s:%u", udpsrc->host, udpsrc->port);
 583       break;
 584     }
 585     case PROP_MULTICAST_IFACE:
 586       g_free (udpsrc->multi_iface);
 587
 588       if (g_value_get_string (value) == NULL)
 589         udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
 590       else
 591         udpsrc->multi_iface = g_value_dup_string (value);
 592       break;
 593     case PROP_URI:
 594       gst_udpsrc_set_uri (udpsrc, g_value_get_string (value), NULL);
 595       break;
 596     case PROP_CAPS:
 597     {
 598       const GstCaps *new_caps_val = gst_value_get_caps (value);
 599
 600       GstCaps *new_caps;
 601
 602       GstCaps *old_caps;
 603
 604       if (new_caps_val == NULL) {
 605         new_caps = gst_caps_new_any ();
 606       } else {
 607         new_caps = gst_caps_copy (new_caps_val);
 608       }
 609
 610       old_caps = udpsrc->caps;
 611       udpsrc->caps = new_caps;
 612       if (old_caps)
 613         gst_caps_unref (old_caps);
 614       gst_pad_set_caps (GST_BASE_SRC (udpsrc)->srcpad, new_caps);
 615       break;
 616     }
 617     case PROP_SOCKET:
 618       if (udpsrc->socket != NULL && udpsrc->socket != udpsrc->used_socket &&
 619           udpsrc->close_socket) {
 620         GError *err = NULL;
 621
 622         if (!g_socket_close (udpsrc->socket, &err)) {
 623           GST_ERROR ("failed to close socket %p: %s", udpsrc->socket,
 624               err->message);
 625           g_clear_error (&err);
 626         }
 627       }
 628       if (udpsrc->socket)
 629         g_object_unref (udpsrc->socket);
 630       udpsrc->socket = g_value_dup_object (value);
 631       GST_DEBUG ("setting socket to %p", udpsrc->socket);
 632       break;
 633     case PROP_TIMEOUT:
 634       udpsrc->timeout = g_value_get_uint64 (value);
 635       break;
 636     case PROP_SKIP_FIRST_BYTES:
 637       udpsrc->skip_first_bytes = g_value_get_int (value);
 638       break;
 639     case PROP_CLOSE_SOCKET:
 640       udpsrc->close_socket = g_value_get_boolean (value);
 641       break;
 642     case PROP_AUTO_MULTICAST:
 643       udpsrc->auto_multicast = g_value_get_boolean (value);
 644       break;
 645     case PROP_REUSE:
 646       udpsrc->reuse = g_value_get_boolean (value);
 647       break;
 648     default:
 649       break;
 650   }
 651 }
 652
 653 static void
 654 gst_udpsrc_get_property (GObject * object, guint prop_id, GValue * value,
 655     GParamSpec * pspec)
 656 {
 657   GstUDPSrc *udpsrc = GST_UDPSRC (object);
 658
 659   switch (prop_id) {
 660     case PROP_BUFFER_SIZE:
 661       g_value_set_int (value, udpsrc->buffer_size);
 662       break;
 663     case PROP_PORT:
 664       g_value_set_int (value, udpsrc->port);
 665       break;
 666     case PROP_MULTICAST_GROUP:
 667       g_value_set_string (value, udpsrc->host);
 668       break;
 669     case PROP_MULTICAST_IFACE:
 670       g_value_set_string (value, udpsrc->multi_iface);
 671       break;
 672     case PROP_URI:
 673       g_value_set_string (value, udpsrc->uri);
 674       break;
 675     case PROP_CAPS:
 676       gst_value_set_caps (value, udpsrc->caps);
 677       break;
 678     case PROP_SOCKET:
 679       g_value_set_object (value, udpsrc->socket);
 680       break;
 681     case PROP_TIMEOUT:
 682       g_value_set_uint64 (value, udpsrc->timeout);
 683       break;
 684     case PROP_SKIP_FIRST_BYTES:
 685       g_value_set_int (value, udpsrc->skip_first_bytes);
 686       break;
 687     case PROP_CLOSE_SOCKET:
 688       g_value_set_boolean (value, udpsrc->close_socket);
 689       break;
 690     case PROP_USED_SOCKET:
 691       g_value_set_object (value, udpsrc->used_socket);
 692       break;
 693     case PROP_AUTO_MULTICAST:
 694       g_value_set_boolean (value, udpsrc->auto_multicast);
 695       break;
 696     case PROP_REUSE:
 697       g_value_set_boolean (value, udpsrc->reuse);
 698       break;
 699     default:
 700       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 701       break;
 702   }
 703 }
 704
 705 /* create a socket for sending to remote machine */
 706 static gboolean
 707 gst_udpsrc_start (GstBaseSrc * bsrc)
 708 {
 709   GstUDPSrc *src;
 710   GInetAddress *addr, *bind_addr;
 711   GSocketAddress *bind_saddr;
 712   GResolver *resolver;
 713   GError *err = NULL;
 714
 715   src = GST_UDPSRC (bsrc);
 716
 717   if (src->socket == NULL) {
 718     /* need to allocate a socket */
 719     GST_DEBUG_OBJECT (src, "allocating socket for %s:%d", src->host, src->port);
 720
 721     addr = g_inet_address_new_from_string (src->host);
 722     if (!addr) {
 723       GList *results;
 724
 725       resolver = g_resolver_get_default ();
 726       results =
 727           g_resolver_lookup_by_name (resolver, src->host, src->cancellable,
 728           &err);
 729       if (!results)
 730         goto name_resolve;
 731       addr = G_INET_ADDRESS (g_object_ref (results->data));
 732
 733       g_resolver_free_addresses (results);
 734       g_object_unref (resolver);
 735     }
 736 #ifndef GST_DISABLE_GST_DEBUG
 737     {
 738       gchar *ip = g_inet_address_to_string (addr);
 739
 740       GST_DEBUG_OBJECT (src, "IP address for host %s is %s", src->host, ip);
 741       g_free (ip);
 742     }
 743 #endif
 744
 745     if ((src->used_socket =
 746             g_socket_new (g_inet_address_get_family (addr),
 747                 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
 748       goto no_socket;
 749
 750     src->external_socket = FALSE;
 751
 752     GST_DEBUG_OBJECT (src, "got socket %p", src->used_socket);
 753
 754     if (src->addr)
 755       g_object_unref (src->addr);
 756     src->addr =
 757         G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
 758
 759     GST_DEBUG_OBJECT (src, "binding on port %d", src->port);
 760
 761     if (g_inet_address_get_is_multicast (addr))
 762       bind_addr = g_inet_address_new_any (g_inet_address_get_family (addr));
 763     else
 764       bind_addr = G_INET_ADDRESS (g_object_ref (addr));
 765
 766     g_object_unref (addr);
 767
 768     bind_saddr = g_inet_socket_address_new (bind_addr, src->port);
 769     g_object_unref (bind_addr);
 770     if (!g_socket_bind (src->used_socket, bind_saddr, src->reuse, &err))
 771       goto bind_error;
 772
 773     g_object_unref (bind_saddr);
 774   } else {
 775     GST_DEBUG_OBJECT (src, "using provided socket %p", src->socket);
 776     /* we use the configured socket, try to get some info about it */
 777     src->used_socket = G_SOCKET (g_object_ref (src->socket));
 778     src->external_socket = TRUE;
 779
 780     if (src->addr)
 781       g_object_unref (src->addr);
 782     src->addr =
 783         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
 784             &err));
 785     if (!src->addr)
 786       goto getsockname_error;
 787   }
 788
 789   if (src->timeout)
 790     g_socket_set_timeout (src->used_socket, src->timeout / GST_SECOND);
 791
 792 #ifdef SO_RCVBUF
 793   {
 794     gint rcvsize, ret;
 795     socklen_t len;
 796
 797     len = sizeof (rcvsize);
 798     if (src->buffer_size != 0) {
 799       rcvsize = src->buffer_size;
 800
 801       GST_DEBUG_OBJECT (src, "setting udp buffer of %d bytes", rcvsize);
 802       /* set buffer size, Note that on Linux this is typically limited to a
 803        * maximum of around 100K. Also a minimum of 128 bytes is required on
 804        * Linux. */
 805       ret =
 806           setsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
 807           (void *) &rcvsize, len);
 808       if (ret != 0) {
 809         GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
 810             ("Could not create a buffer of requested %d bytes, %d: %s (%d)",
 811                 rcvsize, ret, g_strerror (errno), errno));
 812       }
 813     }
 814
 815     /* read the value of the receive buffer. Note that on linux this returns 2x the
 816      * value we set because the kernel allocates extra memory for metadata.
 817      * The default on Linux is about 100K (which is about 50K without metadata) */
 818     ret =
 819         getsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
 820         (void *) &rcvsize, &len);
 821     if (ret == 0)
 822       GST_DEBUG_OBJECT (src, "have udp buffer of %d bytes", rcvsize);
 823     else
 824       GST_DEBUG_OBJECT (src, "could not get udp buffer size");
 825   }
 826 #endif
 827
 828   g_socket_set_broadcast (src->used_socket, TRUE);
 829
 830   if (src->auto_multicast
 831       &&
 832       g_inet_address_get_is_multicast (g_inet_socket_address_get_address
 833           (src->addr))) {
 834     GST_DEBUG_OBJECT (src, "joining multicast group %s", src->host);
 835     if (!g_socket_join_multicast_group (src->used_socket,
 836             g_inet_socket_address_get_address (src->addr),
 837             FALSE, src->multi_iface, &err))
 838       goto membership;
 839   }
 840
 841   /* NOTE: sockaddr_in.sin_port works for ipv4 and ipv6 because sin_port
 842    * follows ss_family on both */
 843   {
 844     GInetSocketAddress *addr;
 845     guint16 port;
 846
 847     addr =
 848         G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
 849             &err));
 850     if (!addr)
 851       goto getsockname_error;
 852
 853     port = g_inet_socket_address_get_port (addr);
 854     GST_DEBUG_OBJECT (src, "bound, on port %d", port);
 855     if (port != src->port) {
 856       src->port = port;
 857       GST_DEBUG_OBJECT (src, "notifying port %d", port);
 858       g_object_notify (G_OBJECT (src), "port");
 859     }
 860     g_object_unref (addr);
 861   }
 862
 863   return TRUE;
 864
 865   /* ERRORS */
 866 name_resolve:
 867   {
 868     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
 869         ("Name resolval failed: %s", err->message));
 870     g_clear_error (&err);
 871     g_object_unref (resolver);
 872     return FALSE;
 873   }
 874 no_socket:
 875   {
 876     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
 877         ("no socket error: %s", err->message));
 878     g_clear_error (&err);
 879     g_object_unref (addr);
 880     return FALSE;
 881   }
 882 bind_error:
 883   {
 884     gst_udpsrc_stop (GST_BASE_SRC (src));
 885     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
 886         ("bind failed: %s", err->message));
 887     g_clear_error (&err);
 888     g_object_unref (bind_saddr);
 889     return FALSE;
 890   }
 891 membership:
 892   {
 893     gst_udpsrc_stop (GST_BASE_SRC (src));
 894     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
 895         ("could add membership: %s", err->message));
 896     g_clear_error (&err);
 897     return FALSE;
 898   }
 899 getsockname_error:
 900   {
 901     gst_udpsrc_stop (GST_BASE_SRC (src));
 902     GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
 903         ("getsockname failed: %s", err->message));
 904     g_clear_error (&err);
 905     return FALSE;
 906   }
 907 }
 908
 909 static gboolean
 910 gst_udpsrc_unlock (GstBaseSrc * bsrc)
 911 {
 912   GstUDPSrc *src;
 913
 914   src = GST_UDPSRC (bsrc);
 915
 916   GST_LOG_OBJECT (src, "Flushing");
 917   g_cancellable_cancel (src->cancellable);
 918
 919   return TRUE;
 920 }
 921
 922 static gboolean
 923 gst_udpsrc_unlock_stop (GstBaseSrc * bsrc)
 924 {
 925   GstUDPSrc *src;
 926
 927   src = GST_UDPSRC (bsrc);
 928
 929   GST_LOG_OBJECT (src, "No longer flushing");
 930   g_cancellable_reset (src->cancellable);
 931
 932   return TRUE;
 933 }
 934
 935 static gboolean
 936 gst_udpsrc_stop (GstBaseSrc * bsrc)
 937 {
 938   GstUDPSrc *src;
 939
 940   src = GST_UDPSRC (bsrc);
 941
 942   GST_DEBUG ("stopping, closing sockets");
 943
 944   if (src->used_socket) {
 945     if (src->auto_multicast
 946         &&
 947         g_inet_address_get_is_multicast (g_inet_socket_address_get_address
 948             (src->addr))) {
 949       GError *err = NULL;
 950
 951       GST_DEBUG_OBJECT (src, "leaving multicast group %s", src->host);
 952
 953       if (!g_socket_leave_multicast_group (src->used_socket,
 954               g_inet_socket_address_get_address (src->addr), FALSE,
 955               src->multi_iface, NULL)) {
 956         GST_ERROR_OBJECT (src, "Failed to leave multicast group: %s",
 957             err->message);
 958         g_clear_error (&err);
 959       }
 960     }
 961
 962     if (src->close_socket || !src->external_socket) {
 963       GError *err = NULL;
 964       if (!g_socket_close (src->used_socket, &err)) {
 965         GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
 966         g_clear_error (&err);
 967       }
 968     }
 969
 970     g_object_unref (src->used_socket);
 971     src->used_socket = NULL;
 972     g_object_unref (src->addr);
 973     src->addr = NULL;
 974   }
 975
 976   return TRUE;
 977 }
 978
 979 /*** GSTURIHANDLER INTERFACE *************************************************/
 980
 981 static GstURIType
 982 gst_udpsrc_uri_get_type (GType type)
 983 {
 984   return GST_URI_SRC;
 985 }
 986
 987 static const gchar *const *
 988 gst_udpsrc_uri_get_protocols (GType type)
 989 {
 990   static const gchar *protocols[] = { "udp", NULL };
 991
 992   return protocols;
 993 }
 994
 995 static gchar *
 996 gst_udpsrc_uri_get_uri (GstURIHandler * handler)
 997 {
 998   GstUDPSrc *src = GST_UDPSRC (handler);
 999
1000   return g_strdup (src->uri);
1001 }
1002
1003 static gboolean
1004 gst_udpsrc_uri_set_uri (GstURIHandler * handler, const gchar * uri,
1005     GError ** error)
1006 {
1007   return gst_udpsrc_set_uri (GST_UDPSRC (handler), uri, error);
1008 }
1009
1010 static void
1011 gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data)
1012 {
1013   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
1014
1015   iface->get_type = gst_udpsrc_uri_get_type;
1016   iface->get_protocols = gst_udpsrc_uri_get_protocols;
1017   iface->get_uri = gst_udpsrc_uri_get_uri;
1018   iface->set_uri = gst_udpsrc_uri_set_uri;
1019 }