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