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