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