mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2025-01-02 05:28:48 +00:00
Code Issues 1 Projects Releases Wiki Activity
gstreamer/gst/udp/gstudpsrc.c
Tim-Philipp Müller f54110fd3e udp: don't use soon-to-be-deprecated g_cancellable_reset() 
From the API documentation: "Note that it is generally not
a good idea to reuse an existing cancellable for more
operations after it has been cancelled once, as this
function might tempt you to do. The recommended practice
is to drop the reference to a cancellable after cancelling
it, and let it die with the outstanding async operations.
You should create a fresh cancellable for further async
operations."

https://bugzilla.gnome.org/show_bug.cgi?id=739132
2015-05-19 19:00:20 +01:00

1276 lines
38 KiB
C
Raw Blame History

/* GStreamer
* Copyright (C) <2005> Wim Taymans <wim@fluendo.com>
* Copyright (C) <2005> Nokia Corporation <kai.vehmanen@nokia.com>
* Copyright (C) <2012> Collabora Ltd.
* Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
* Copyright (C) 2014 Tim-Philipp Müller <tim@centricular.com>
* Copyright (C) 2014 Centricular Ltd
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public
* License along with this library; if not, write to the
* Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
/**
* SECTION:element-udpsrc
* @see_also: udpsink, multifdsink
*
* udpsrc is a network source that reads UDP packets from the network.
* It can be combined with RTP depayloaders to implement RTP streaming.
*
* The udpsrc element supports automatic port allocation by setting the
* #GstUDPSrc:port property to 0. After setting the udpsrc to PAUSED, the
* allocated port can be obtained by reading the port property.
*
* udpsrc can read from multicast groups by setting the #GstUDPSrc:multicast-group
* property to the IP address of the multicast group.
*
* Alternatively one can provide a custom socket to udpsrc with the #GstUDPSrc:socket
* property, udpsrc will then not allocate a socket itself but use the provided
* one.
*
* The #GstUDPSrc:caps property is mainly used to give a type to the UDP packet
* so that they can be autoplugged in GStreamer pipelines. This is very usefull
* for RTP implementations where the contents of the UDP packets is transfered
* out-of-bounds using SDP or other means.
*
* The #GstUDPSrc:buffer-size property is used to change the default kernel
* buffersizes used for receiving packets. The buffer size may be increased for
* high-volume connections, or may be decreased to limit the possible backlog of
* incoming data. The system places an absolute limit on these values, on Linux,
* for example, the default buffer size is typically 50K and can be increased to
* maximally 100K.
*
* The #GstUDPSrc:skip-first-bytes property is used to strip off an arbitrary
* number of bytes from the start of the raw udp packet and can be used to strip
* off proprietary header, for example.
*
* The udpsrc is always a live source. It does however not provide a #GstClock,
* this is left for upstream elements such as an RTP session manager or demuxer
* (such as an MPEG demuxer). As with all live sources, the captured buffers
* will have their timestamp set to the current running time of the pipeline.
*
* udpsrc implements a #GstURIHandler interface that handles udp://host:port
* type URIs.
*
* If the #GstUDPSrc:timeout property is set to a value bigger than 0, udpsrc
* will generate an element message named
* <classname>&quot;GstUDPSrcTimeout&quot;</classname>
* if no data was recieved in the given timeout.
* The message's structure contains one field:
* <itemizedlist>
* <listitem>
* <para>
* #guint64
* <classname>&quot;timeout&quot;</classname>: the timeout in microseconds that
* expired when waiting for data.
* </para>
* </listitem>
* </itemizedlist>
* The message is typically used to detect that no UDP arrives in the receiver
* because it is blocked by a firewall.
*
* A custom file descriptor can be configured with the
* #GstUDPSrc:socket property. The socket will be closed when setting
* the element to READY by default. This behaviour can be overriden
* with the #GstUDPSrc:close-socket property, in which case the
* application is responsible for closing the file descriptor.
*
* <refsect2>
* <title>Examples</title>
* |[
* gst-launch-1.0 -v udpsrc ! fakesink dump=1
* ]| A pipeline to read from the default port and dump the udp packets.
* To actually generate udp packets on the default port one can use the
* udpsink element. When running the following pipeline in another terminal, the
* above mentioned pipeline should dump data packets to the console.
* |[
* gst-launch-1.0 -v audiotestsrc ! udpsink
* ]|
* |[
* gst-launch-1.0 -v udpsrc port=0 ! fakesink
* ]| read udp packets from a free port.
* </refsect2>
*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif
#include <string.h>
#include "gstudpsrc.h"
#include <gst/net/gstnetaddressmeta.h>
#if GLIB_CHECK_VERSION (2, 35, 7)
#include <gio/gnetworking.h>
#else
/* nicked from gnetworking.h */
#ifdef G_OS_WIN32
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0501
#endif
#include <winsock2.h>
#undef interface
#include <ws2tcpip.h> /* for socklen_t */
#endif /* G_OS_WIN32 */
#ifdef HAVE_SYS_SOCKET_H
#include <sys/socket.h>
#endif
#endif
/* not 100% correct, but a good upper bound for memory allocation purposes */
#define MAX_IPV4_UDP_PACKET_SIZE (65536 - 8)
GST_DEBUG_CATEGORY_STATIC (udpsrc_debug);
#define GST_CAT_DEFAULT (udpsrc_debug)
static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
GST_PAD_SRC,
GST_PAD_ALWAYS,
GST_STATIC_CAPS_ANY);
#define UDP_DEFAULT_PORT 5004
#define UDP_DEFAULT_MULTICAST_GROUP "0.0.0.0"
#define UDP_DEFAULT_MULTICAST_IFACE NULL
#define UDP_DEFAULT_URI "udp://"UDP_DEFAULT_MULTICAST_GROUP":"G_STRINGIFY(UDP_DEFAULT_PORT)
#define UDP_DEFAULT_CAPS NULL
#define UDP_DEFAULT_SOCKET NULL
#define UDP_DEFAULT_BUFFER_SIZE 0
#define UDP_DEFAULT_TIMEOUT 0
#define UDP_DEFAULT_SKIP_FIRST_BYTES 0
#define UDP_DEFAULT_CLOSE_SOCKET TRUE
#define UDP_DEFAULT_USED_SOCKET NULL
#define UDP_DEFAULT_AUTO_MULTICAST TRUE
#define UDP_DEFAULT_REUSE TRUE
enum
{
PROP_0,
PROP_PORT,
PROP_MULTICAST_GROUP,
PROP_MULTICAST_IFACE,
PROP_URI,
PROP_CAPS,
PROP_SOCKET,
PROP_BUFFER_SIZE,
PROP_TIMEOUT,
PROP_SKIP_FIRST_BYTES,
PROP_CLOSE_SOCKET,
PROP_USED_SOCKET,
PROP_AUTO_MULTICAST,
PROP_REUSE,
PROP_ADDRESS
};
static void gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data);
static GstCaps *gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter);
static GstFlowReturn gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf);
static gboolean gst_udpsrc_close (GstUDPSrc * src);
static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc);
static gboolean gst_udpsrc_unlock_stop (GstBaseSrc * bsrc);
static gboolean gst_udpsrc_negotiate (GstBaseSrc * basesrc);
static void gst_udpsrc_finalize (GObject * object);
static void gst_udpsrc_set_property (GObject * object, guint prop_id,
const GValue * value, GParamSpec * pspec);
static void gst_udpsrc_get_property (GObject * object, guint prop_id,
GValue * value, GParamSpec * pspec);
static GstStateChangeReturn gst_udpsrc_change_state (GstElement * element,
GstStateChange transition);
#define gst_udpsrc_parent_class parent_class
G_DEFINE_TYPE_WITH_CODE (GstUDPSrc, gst_udpsrc, GST_TYPE_PUSH_SRC,
G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_udpsrc_uri_handler_init));
static void
gst_udpsrc_class_init (GstUDPSrcClass * klass)
{
GObjectClass *gobject_class;
GstElementClass *gstelement_class;
GstBaseSrcClass *gstbasesrc_class;
GstPushSrcClass *gstpushsrc_class;
gobject_class = (GObjectClass *) klass;
gstelement_class = (GstElementClass *) klass;
gstbasesrc_class = (GstBaseSrcClass *) klass;
gstpushsrc_class = (GstPushSrcClass *) klass;
GST_DEBUG_CATEGORY_INIT (udpsrc_debug, "udpsrc", 0, "UDP src");
gobject_class->set_property = gst_udpsrc_set_property;
gobject_class->get_property = gst_udpsrc_get_property;
gobject_class->finalize = gst_udpsrc_finalize;
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
g_param_spec_int ("port", "Port",
"The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
UDP_DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
/* FIXME 2.0: Remove multicast-group property */
#ifndef GST_REMOVE_DEPRECATED
g_object_class_install_property (gobject_class, PROP_MULTICAST_GROUP,
g_param_spec_string ("multicast-group", "Multicast Group",
"The Address of multicast group to join. (DEPRECATED: "
"Use address property instead)", UDP_DEFAULT_MULTICAST_GROUP,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS | G_PARAM_DEPRECATED));
#endif
g_object_class_install_property (gobject_class, PROP_MULTICAST_IFACE,
g_param_spec_string ("multicast-iface", "Multicast Interface",
"The network interface on which to join the multicast group",
UDP_DEFAULT_MULTICAST_IFACE,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_URI,
g_param_spec_string ("uri", "URI",
"URI in the form of udp://multicast_group:port", UDP_DEFAULT_URI,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_CAPS,
g_param_spec_boxed ("caps", "Caps",
"The caps of the source pad", GST_TYPE_CAPS,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_SOCKET,
g_param_spec_object ("socket", "Socket",
"Socket to use for UDP reception. (NULL == allocate)",
G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_BUFFER_SIZE,
g_param_spec_int ("buffer-size", "Buffer Size",
"Size of the kernel receive buffer in bytes, 0=default", 0, G_MAXINT,
UDP_DEFAULT_BUFFER_SIZE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_TIMEOUT,
g_param_spec_uint64 ("timeout", "Timeout",
"Post a message after timeout nanoseconds (0 = disabled)", 0,
G_MAXUINT64, UDP_DEFAULT_TIMEOUT,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (G_OBJECT_CLASS (klass),
PROP_SKIP_FIRST_BYTES, g_param_spec_int ("skip-first-bytes",
"Skip first bytes", "number of bytes to skip for each udp packet", 0,
G_MAXINT, UDP_DEFAULT_SKIP_FIRST_BYTES,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
g_param_spec_boolean ("close-socket", "Close socket",
"Close socket if passed as property on state change",
UDP_DEFAULT_CLOSE_SOCKET,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_USED_SOCKET,
g_param_spec_object ("used-socket", "Socket Handle",
"Socket currently in use for UDP reception. (NULL = no socket)",
G_TYPE_SOCKET, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_AUTO_MULTICAST,
g_param_spec_boolean ("auto-multicast", "Auto Multicast",
"Automatically join/leave multicast groups",
UDP_DEFAULT_AUTO_MULTICAST,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_REUSE,
g_param_spec_boolean ("reuse", "Reuse", "Enable reuse of the port",
UDP_DEFAULT_REUSE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
g_object_class_install_property (gobject_class, PROP_ADDRESS,
g_param_spec_string ("address", "Address",
"Address to receive packets for. This is equivalent to the "
"multicast-group property for now", UDP_DEFAULT_MULTICAST_GROUP,
G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
gst_element_class_add_pad_template (gstelement_class,
gst_static_pad_template_get (&src_template));
gst_element_class_set_static_metadata (gstelement_class,
"UDP packet receiver", "Source/Network",
"Receive data over the network via UDP",
"Wim Taymans <wim@fluendo.com>, "
"Thijs Vermeir <thijs.vermeir@barco.com>");
gstelement_class->change_state = gst_udpsrc_change_state;
gstbasesrc_class->unlock = gst_udpsrc_unlock;
gstbasesrc_class->unlock_stop = gst_udpsrc_unlock_stop;
gstbasesrc_class->get_caps = gst_udpsrc_getcaps;
gstbasesrc_class->negotiate = gst_udpsrc_negotiate;
gstpushsrc_class->create = gst_udpsrc_create;
}
static void
gst_udpsrc_init (GstUDPSrc * udpsrc)
{
udpsrc->uri =
g_strdup_printf ("udp://%s:%u", UDP_DEFAULT_MULTICAST_GROUP,
UDP_DEFAULT_PORT);
udpsrc->address = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
udpsrc->port = UDP_DEFAULT_PORT;
udpsrc->socket = UDP_DEFAULT_SOCKET;
udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
udpsrc->buffer_size = UDP_DEFAULT_BUFFER_SIZE;
udpsrc->timeout = UDP_DEFAULT_TIMEOUT;
udpsrc->skip_first_bytes = UDP_DEFAULT_SKIP_FIRST_BYTES;
udpsrc->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
udpsrc->external_socket = (udpsrc->socket != NULL);
udpsrc->auto_multicast = UDP_DEFAULT_AUTO_MULTICAST;
udpsrc->used_socket = UDP_DEFAULT_USED_SOCKET;
udpsrc->reuse = UDP_DEFAULT_REUSE;
udpsrc->cancellable = g_cancellable_new ();
/* configure basesrc to be a live source */
gst_base_src_set_live (GST_BASE_SRC (udpsrc), TRUE);
/* make basesrc output a segment in time */
gst_base_src_set_format (GST_BASE_SRC (udpsrc), GST_FORMAT_TIME);
/* make basesrc set timestamps on outgoing buffers based on the running_time
* when they were captured */
gst_base_src_set_do_timestamp (GST_BASE_SRC (udpsrc), TRUE);
}
static void
gst_udpsrc_finalize (GObject * object)
{
GstUDPSrc *udpsrc;
udpsrc = GST_UDPSRC (object);
if (udpsrc->caps)
gst_caps_unref (udpsrc->caps);
udpsrc->caps = NULL;
g_free (udpsrc->multi_iface);
udpsrc->multi_iface = NULL;
g_free (udpsrc->uri);
udpsrc->uri = NULL;
g_free (udpsrc->address);
udpsrc->address = NULL;
if (udpsrc->socket)
g_object_unref (udpsrc->socket);
udpsrc->socket = NULL;
if (udpsrc->used_socket)
g_object_unref (udpsrc->used_socket);
udpsrc->used_socket = NULL;
if (udpsrc->cancellable)
g_object_unref (udpsrc->cancellable);
udpsrc->cancellable = NULL;
G_OBJECT_CLASS (parent_class)->finalize (object);
}
static GstCaps *
gst_udpsrc_getcaps (GstBaseSrc * src, GstCaps * filter)
{
GstUDPSrc *udpsrc;
GstCaps *caps, *result;
udpsrc = GST_UDPSRC (src);
GST_OBJECT_LOCK (src);
if ((caps = udpsrc->caps))
gst_caps_ref (caps);
GST_OBJECT_UNLOCK (src);
if (caps) {
if (filter) {
result = gst_caps_intersect_full (filter, caps, GST_CAPS_INTERSECT_FIRST);
gst_caps_unref (caps);
} else {
result = caps;
}
} else {
result = (filter) ? gst_caps_ref (filter) : gst_caps_new_any ();
}
return result;
}
static void
gst_udpsrc_reset_memory_allocator (GstUDPSrc * src)
{
if (src->mem != NULL) {
gst_memory_unmap (src->mem, &src->map);
gst_memory_unref (src->mem);
src->mem = NULL;
}
if (src->mem_max != NULL) {
gst_memory_unmap (src->mem_max, &src->map_max);
gst_memory_unref (src->mem_max);
src->mem_max = NULL;
}
src->vec[0].buffer = NULL;
src->vec[0].size = 0;
src->vec[1].buffer = NULL;
src->vec[1].size = 0;
if (src->allocator != NULL) {
gst_object_unref (src->allocator);
src->allocator = NULL;
}
}
static gboolean
gst_udpsrc_negotiate (GstBaseSrc * basesrc)
{
GstUDPSrc *src = GST_UDPSRC_CAST (basesrc);
gboolean ret;
/* just chain up to the default implementation, we just want to
* retrieve the allocator at the end of it (if there is one) */
ret = GST_BASE_SRC_CLASS (parent_class)->negotiate (basesrc);
if (ret) {
GstAllocationParams new_params;
GstAllocator *new_allocator = NULL;
/* retrieve new allocator */
gst_base_src_get_allocator (basesrc, &new_allocator, &new_params);
if (src->allocator != new_allocator ||
memcmp (&src->params, &new_params, sizeof (GstAllocationParams)) != 0) {
/* drop old allocator and throw away any memory allocated with it */
gst_udpsrc_reset_memory_allocator (src);
/* and save the new allocator and/or new allocation parameters */
src->allocator = new_allocator;
src->params = new_params;
GST_INFO_OBJECT (src, "new allocator: %" GST_PTR_FORMAT, new_allocator);
}
}
return ret;
}
static gboolean
gst_udpsrc_alloc_mem (GstUDPSrc * src, GstMemory ** p_mem, GstMapInfo * map,
gsize size)
{
GstMemory *mem;
mem = gst_allocator_alloc (src->allocator, size, &src->params);
if (!gst_memory_map (mem, map, GST_MAP_WRITE)) {
gst_memory_unref (mem);
memset (map, 0, sizeof (GstMapInfo));
return FALSE;
}
*p_mem = mem;
return TRUE;
}
static gboolean
gst_udpsrc_ensure_mem (GstUDPSrc * src)
{
if (src->mem == NULL) {
gsize mem_size = 1500; /* typical max. MTU */
/* if packets are likely to be smaller, just use that size, otherwise
* default to assuming incoming packets are around MTU size */
if (src->max_size > 0 && src->max_size < mem_size)
mem_size = src->max_size;
if (!gst_udpsrc_alloc_mem (src, &src->mem, &src->map, mem_size))
return FALSE;
src->vec[0].buffer = src->map.data;
src->vec[0].size = src->map.size;
}
if (src->mem_max == NULL) {
gsize max_size = MAX_IPV4_UDP_PACKET_SIZE;
if (!gst_udpsrc_alloc_mem (src, &src->mem_max, &src->map_max, max_size))
return FALSE;
src->vec[1].buffer = src->map_max.data;
src->vec[1].size = src->map_max.size;
}
return TRUE;
}
static GstFlowReturn
gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf)
{
GstUDPSrc *udpsrc;
GstBuffer *outbuf = NULL;
GSocketAddress *saddr = NULL;
gint flags = G_SOCKET_MSG_NONE;
gboolean try_again;
GError *err = NULL;
gssize res;
gsize offset;
udpsrc = GST_UDPSRC_CAST (psrc);
if (!gst_udpsrc_ensure_mem (udpsrc))
goto memory_alloc_error;
retry:
do {
gint64 timeout;
try_again = FALSE;
if (udpsrc->timeout)
timeout = udpsrc->timeout / 1000;
else
timeout = -1;
GST_LOG_OBJECT (udpsrc, "doing select, timeout %" G_GINT64_FORMAT, timeout);
if (!g_socket_condition_timed_wait (udpsrc->used_socket, G_IO_IN | G_IO_PRI,
timeout, udpsrc->cancellable, &err)) {
if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY)
|| g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
goto stopped;
} else if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_TIMED_OUT)) {
g_clear_error (&err);
/* timeout, post element message */
gst_element_post_message (GST_ELEMENT_CAST (udpsrc),
gst_message_new_element (GST_OBJECT_CAST (udpsrc),
gst_structure_new ("GstUDPSrcTimeout",
"timeout", G_TYPE_UINT64, udpsrc->timeout, NULL)));
} else {
goto select_error;
}
try_again = TRUE;
}
} while (G_UNLIKELY (try_again));
if (saddr != NULL) {
g_object_unref (saddr);
saddr = NULL;
}
res =
g_socket_receive_message (udpsrc->used_socket, &saddr, udpsrc->vec, 2,
NULL, NULL, &flags, udpsrc->cancellable, &err);
if (G_UNLIKELY (res < 0)) {
/* EHOSTUNREACH for a UDP socket means that a packet sent with udpsink
* generated a "port unreachable" ICMP response. We ignore that and try
* again. */
if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_HOST_UNREACHABLE)) {
g_clear_error (&err);
goto retry;
}
goto receive_error;
}
/* remember maximum packet size */
if (res > udpsrc->max_size)
udpsrc->max_size = res;
outbuf = gst_buffer_new ();
/* append first memory chunk to buffer */
gst_buffer_append_memory (outbuf, udpsrc->mem);
/* if the packet didn't fit into the first chunk, add second one as well */
if (res > udpsrc->map.size) {
gst_buffer_append_memory (outbuf, udpsrc->mem_max);
gst_memory_unmap (udpsrc->mem_max, &udpsrc->map_max);
udpsrc->vec[1].buffer = NULL;
udpsrc->vec[1].size = 0;
udpsrc->mem_max = NULL;
}
/* make sure we allocate a new chunk next time (we do this only here because
* we look at map.size to see if the second memory chunk is needed above) */
gst_memory_unmap (udpsrc->mem, &udpsrc->map);
udpsrc->vec[0].buffer = NULL;
udpsrc->vec[0].size = 0;
udpsrc->mem = NULL;
offset = udpsrc->skip_first_bytes;
if (G_UNLIKELY (offset > 0 && res < offset))
goto skip_error;
gst_buffer_resize (outbuf, offset, res - offset);
/* use buffer metadata so receivers can also track the address */
if (saddr) {
gst_buffer_add_net_address_meta (outbuf, saddr);
g_object_unref (saddr);
saddr = NULL;
}
GST_LOG_OBJECT (udpsrc, "read packet of %d bytes", (int) res);
*buf = GST_BUFFER_CAST (outbuf);
return GST_FLOW_OK;
/* ERRORS */
memory_alloc_error:
{
GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
("Failed to allocate or map memory"));
return GST_FLOW_ERROR;
}
select_error:
{
GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
("select error: %s", err->message));
g_clear_error (&err);
return GST_FLOW_ERROR;
}
stopped:
{
GST_DEBUG ("stop called");
g_clear_error (&err);
return GST_FLOW_FLUSHING;
}
receive_error:
{
if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_BUSY) ||
g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
g_clear_error (&err);
return GST_FLOW_FLUSHING;
} else {
GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL),
("receive error %" G_GSSIZE_FORMAT ": %s", res, err->message));
g_clear_error (&err);
return GST_FLOW_ERROR;
}
}
skip_error:
{
gst_buffer_unref (outbuf);
GST_ELEMENT_ERROR (udpsrc, STREAM, DECODE, (NULL),
("UDP buffer to small to skip header"));
return GST_FLOW_ERROR;
}
}
static gboolean
gst_udpsrc_set_uri (GstUDPSrc * src, const gchar * uri, GError ** error)
{
gchar *address;
guint16 port;
if (!gst_udp_parse_uri (uri, &address, &port))
goto wrong_uri;
if (port == (guint16) - 1)
port = UDP_DEFAULT_PORT;
g_free (src->address);
src->address = address;
src->port = port;
g_free (src->uri);
src->uri = g_strdup (uri);
return TRUE;
/* ERRORS */
wrong_uri:
{
GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
("error parsing uri %s", uri));
g_set_error_literal (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
"Could not parse UDP URI");
return FALSE;
}
}
static void
gst_udpsrc_set_property (GObject * object, guint prop_id, const GValue * value,
GParamSpec * pspec)
{
GstUDPSrc *udpsrc = GST_UDPSRC (object);
switch (prop_id) {
case PROP_BUFFER_SIZE:
udpsrc->buffer_size = g_value_get_int (value);
break;
case PROP_PORT:
udpsrc->port = g_value_get_int (value);
g_free (udpsrc->uri);
udpsrc->uri =
g_strdup_printf ("udp://%s:%u", udpsrc->address, udpsrc->port);
break;
case PROP_MULTICAST_GROUP:
case PROP_ADDRESS:
{
const gchar *group;
g_free (udpsrc->address);
if ((group = g_value_get_string (value)))
udpsrc->address = g_strdup (group);
else
udpsrc->address = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
g_free (udpsrc->uri);
udpsrc->uri =
g_strdup_printf ("udp://%s:%u", udpsrc->address, udpsrc->port);
break;
}
case PROP_MULTICAST_IFACE:
g_free (udpsrc->multi_iface);
if (g_value_get_string (value) == NULL)
udpsrc->multi_iface = g_strdup (UDP_DEFAULT_MULTICAST_IFACE);
else
udpsrc->multi_iface = g_value_dup_string (value);
break;
case PROP_URI:
gst_udpsrc_set_uri (udpsrc, g_value_get_string (value), NULL);
break;
case PROP_CAPS:
{
const GstCaps *new_caps_val = gst_value_get_caps (value);
GstCaps *new_caps;
GstCaps *old_caps;
if (new_caps_val == NULL) {
new_caps = gst_caps_new_any ();
} else {
new_caps = gst_caps_copy (new_caps_val);
}
GST_OBJECT_LOCK (udpsrc);
old_caps = udpsrc->caps;
udpsrc->caps = new_caps;
GST_OBJECT_UNLOCK (udpsrc);
if (old_caps)
gst_caps_unref (old_caps);
gst_pad_mark_reconfigure (GST_BASE_SRC_PAD (udpsrc));
break;
}
case PROP_SOCKET:
if (udpsrc->socket != NULL && udpsrc->socket != udpsrc->used_socket &&
udpsrc->close_socket) {
GError *err = NULL;
if (!g_socket_close (udpsrc->socket, &err)) {
GST_ERROR ("failed to close socket %p: %s", udpsrc->socket,
err->message);
g_clear_error (&err);
}
}
if (udpsrc->socket)
g_object_unref (udpsrc->socket);
udpsrc->socket = g_value_dup_object (value);
GST_DEBUG ("setting socket to %p", udpsrc->socket);
break;
case PROP_TIMEOUT:
udpsrc->timeout = g_value_get_uint64 (value);
break;
case PROP_SKIP_FIRST_BYTES:
udpsrc->skip_first_bytes = g_value_get_int (value);
break;
case PROP_CLOSE_SOCKET:
udpsrc->close_socket = g_value_get_boolean (value);
break;
case PROP_AUTO_MULTICAST:
udpsrc->auto_multicast = g_value_get_boolean (value);
break;
case PROP_REUSE:
udpsrc->reuse = g_value_get_boolean (value);
break;
default:
break;
}
}
static void
gst_udpsrc_get_property (GObject * object, guint prop_id, GValue * value,
GParamSpec * pspec)
{
GstUDPSrc *udpsrc = GST_UDPSRC (object);
switch (prop_id) {
case PROP_BUFFER_SIZE:
g_value_set_int (value, udpsrc->buffer_size);
break;
case PROP_PORT:
g_value_set_int (value, udpsrc->port);
break;
case PROP_MULTICAST_GROUP:
case PROP_ADDRESS:
g_value_set_string (value, udpsrc->address);
break;
case PROP_MULTICAST_IFACE:
g_value_set_string (value, udpsrc->multi_iface);
break;
case PROP_URI:
g_value_set_string (value, udpsrc->uri);
break;
case PROP_CAPS:
GST_OBJECT_LOCK (udpsrc);
gst_value_set_caps (value, udpsrc->caps);
GST_OBJECT_UNLOCK (udpsrc);
break;
case PROP_SOCKET:
g_value_set_object (value, udpsrc->socket);
break;
case PROP_TIMEOUT:
g_value_set_uint64 (value, udpsrc->timeout);
break;
case PROP_SKIP_FIRST_BYTES:
g_value_set_int (value, udpsrc->skip_first_bytes);
break;
case PROP_CLOSE_SOCKET:
g_value_set_boolean (value, udpsrc->close_socket);
break;
case PROP_USED_SOCKET:
g_value_set_object (value, udpsrc->used_socket);
break;
case PROP_AUTO_MULTICAST:
g_value_set_boolean (value, udpsrc->auto_multicast);
break;
case PROP_REUSE:
g_value_set_boolean (value, udpsrc->reuse);
break;
default:
G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
break;
}
}
static GInetAddress *
gst_udpsrc_resolve (GstUDPSrc * src, const gchar * address)
{
GInetAddress *addr;
GError *err = NULL;
GResolver *resolver;
addr = g_inet_address_new_from_string (address);
if (!addr) {
GList *results;
GST_DEBUG_OBJECT (src, "resolving IP address for host %s", address);
resolver = g_resolver_get_default ();
results =
g_resolver_lookup_by_name (resolver, address, src->cancellable, &err);
if (!results)
goto name_resolve;
addr = G_INET_ADDRESS (g_object_ref (results->data));
g_resolver_free_addresses (results);
g_object_unref (resolver);
}
#ifndef GST_DISABLE_GST_DEBUG
{
gchar *ip = g_inet_address_to_string (addr);
GST_DEBUG_OBJECT (src, "IP address for host %s is %s", address, ip);
g_free (ip);
}
#endif
return addr;
name_resolve:
{
GST_WARNING_OBJECT (src, "Failed to resolve %s: %s", address, err->message);
g_clear_error (&err);
g_object_unref (resolver);
return NULL;
}
}
/* create a socket for sending to remote machine */
static gboolean
gst_udpsrc_open (GstUDPSrc * src)
{
GInetAddress *addr, *bind_addr;
GSocketAddress *bind_saddr;
GError *err = NULL;
if (src->socket == NULL) {
/* need to allocate a socket */
GST_DEBUG_OBJECT (src, "allocating socket for %s:%d", src->address,
src->port);
addr = gst_udpsrc_resolve (src, src->address);
if (!addr)
goto name_resolve;
if ((src->used_socket =
g_socket_new (g_inet_address_get_family (addr),
G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
goto no_socket;
src->external_socket = FALSE;
GST_DEBUG_OBJECT (src, "got socket %p", src->used_socket);
if (src->addr)
g_object_unref (src->addr);
src->addr =
G_INET_SOCKET_ADDRESS (g_inet_socket_address_new (addr, src->port));
GST_DEBUG_OBJECT (src, "binding on port %d", src->port);
/* On Windows it's not possible to bind to a multicast address
* but the OS will make sure to filter out all packets that
* arrive not for the multicast address the socket joined.
*
* On Linux and others it is necessary to bind to a multicast
* address to let the OS filter out all packets that are received
* on the same port but for different addresses than the multicast
* address
*/
#ifdef G_OS_WIN32
if (g_inet_address_get_is_multicast (addr))
bind_addr = g_inet_address_new_any (g_inet_address_get_family (addr));
else
#endif
bind_addr = G_INET_ADDRESS (g_object_ref (addr));
g_object_unref (addr);
bind_saddr = g_inet_socket_address_new (bind_addr, src->port);
g_object_unref (bind_addr);
if (!g_socket_bind (src->used_socket, bind_saddr, src->reuse, &err))
goto bind_error;
g_object_unref (bind_saddr);
} else {
GST_DEBUG_OBJECT (src, "using provided socket %p", src->socket);
/* we use the configured socket, try to get some info about it */
src->used_socket = G_SOCKET (g_object_ref (src->socket));
src->external_socket = TRUE;
if (src->addr)
g_object_unref (src->addr);
src->addr =
G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
&err));
if (!src->addr)
goto getsockname_error;
}
#if GLIB_CHECK_VERSION (2, 35, 7)
{
gint val = 0;
if (src->buffer_size != 0) {
GError *opt_err = NULL;
GST_INFO_OBJECT (src, "setting udp buffer of %d bytes", src->buffer_size);
/* set buffer size, Note that on Linux this is typically limited to a
* maximum of around 100K. Also a minimum of 128 bytes is required on
* Linux. */
if (!g_socket_set_option (src->used_socket, SOL_SOCKET, SO_RCVBUF,
src->buffer_size, &opt_err)) {
GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
("Could not create a buffer of requested %d bytes: %s",
src->buffer_size, opt_err->message));
g_error_free (opt_err);
opt_err = NULL;
}
}
/* read the value of the receive buffer. Note that on linux this returns
* 2x the value we set because the kernel allocates extra memory for
* metadata. The default on Linux is about 100K (which is about 50K
* without metadata) */
if (g_socket_get_option (src->used_socket, SOL_SOCKET, SO_RCVBUF, &val,
NULL)) {
GST_INFO_OBJECT (src, "have udp buffer of %d bytes", val);
} else {
GST_DEBUG_OBJECT (src, "could not get udp buffer size");
}
}
#elif defined (SO_RCVBUF)
{
gint rcvsize, ret;
socklen_t len;
len = sizeof (rcvsize);
if (src->buffer_size != 0) {
rcvsize = src->buffer_size;
GST_DEBUG_OBJECT (src, "setting udp buffer of %d bytes", rcvsize);
/* set buffer size, Note that on Linux this is typically limited to a
* maximum of around 100K. Also a minimum of 128 bytes is required on
* Linux. */
ret =
setsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
(void *) &rcvsize, len);
if (ret != 0) {
GST_ELEMENT_WARNING (src, RESOURCE, SETTINGS, (NULL),
("Could not create a buffer of requested %d bytes, %d: %s (%d)",
rcvsize, ret, g_strerror (errno), errno));
}
}
/* read the value of the receive buffer. Note that on linux this returns 2x the
* value we set because the kernel allocates extra memory for metadata.
* The default on Linux is about 100K (which is about 50K without metadata) */
ret =
getsockopt (g_socket_get_fd (src->used_socket), SOL_SOCKET, SO_RCVBUF,
(void *) &rcvsize, &len);
if (ret == 0)
GST_DEBUG_OBJECT (src, "have udp buffer of %d bytes", rcvsize);
else
GST_DEBUG_OBJECT (src, "could not get udp buffer size");
}
#else
if (src->buffer_size != 0) {
GST_WARNING_OBJECT (src, "don't know how to set udp buffer size on this "
"OS. Consider upgrading your GLib to >= 2.35.7 and re-compiling the "
"GStreamer udp plugin");
}
#endif
g_socket_set_broadcast (src->used_socket, TRUE);
if (src->auto_multicast
&&
g_inet_address_get_is_multicast (g_inet_socket_address_get_address
(src->addr))) {
GST_DEBUG_OBJECT (src, "joining multicast group %s", src->address);
if (!g_socket_join_multicast_group (src->used_socket,
g_inet_socket_address_get_address (src->addr),
FALSE, src->multi_iface, &err))
goto membership;
}
/* NOTE: sockaddr_in.sin_port works for ipv4 and ipv6 because sin_port
* follows ss_family on both */
{
GInetSocketAddress *addr;
guint16 port;
addr =
G_INET_SOCKET_ADDRESS (g_socket_get_local_address (src->used_socket,
&err));
if (!addr)
goto getsockname_error;
port = g_inet_socket_address_get_port (addr);
GST_DEBUG_OBJECT (src, "bound, on port %d", port);
if (port != src->port) {
src->port = port;
GST_DEBUG_OBJECT (src, "notifying port %d", port);
g_object_notify (G_OBJECT (src), "port");
}
g_object_unref (addr);
}
src->allocator = NULL;
gst_allocation_params_init (&src->params);
src->max_size = 0;
return TRUE;
/* ERRORS */
name_resolve:
{
return FALSE;
}
no_socket:
{
GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
("no socket error: %s", err->message));
g_clear_error (&err);
g_object_unref (addr);
return FALSE;
}
bind_error:
{
GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
("bind failed: %s", err->message));
g_clear_error (&err);
g_object_unref (bind_saddr);
gst_udpsrc_close (src);
return FALSE;
}
membership:
{
GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
("could add membership: %s", err->message));
g_clear_error (&err);
gst_udpsrc_close (src);
return FALSE;
}
getsockname_error:
{
GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL),
("getsockname failed: %s", err->message));
g_clear_error (&err);
gst_udpsrc_close (src);
return FALSE;
}
}
static gboolean
gst_udpsrc_unlock (GstBaseSrc * bsrc)
{
GstUDPSrc *src;
src = GST_UDPSRC (bsrc);
GST_LOG_OBJECT (src, "Flushing");
g_cancellable_cancel (src->cancellable);
return TRUE;
}
static gboolean
gst_udpsrc_unlock_stop (GstBaseSrc * bsrc)
{
GstUDPSrc *src;
src = GST_UDPSRC (bsrc);
GST_LOG_OBJECT (src, "No longer flushing");
g_object_unref (src->cancellable);
src->cancellable = g_cancellable_new ();
return TRUE;
}
static gboolean
gst_udpsrc_close (GstUDPSrc * src)
{
GST_DEBUG ("closing sockets");
if (src->used_socket) {
if (src->auto_multicast
&&
g_inet_address_get_is_multicast (g_inet_socket_address_get_address
(src->addr))) {
GError *err = NULL;
GST_DEBUG_OBJECT (src, "leaving multicast group %s", src->address);
if (!g_socket_leave_multicast_group (src->used_socket,
g_inet_socket_address_get_address (src->addr), FALSE,
src->multi_iface, &err)) {
GST_ERROR_OBJECT (src, "Failed to leave multicast group: %s",
err->message);
g_clear_error (&err);
}
}
if (src->close_socket || !src->external_socket) {
GError *err = NULL;
if (!g_socket_close (src->used_socket, &err)) {
GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
g_clear_error (&err);
}
}
g_object_unref (src->used_socket);
src->used_socket = NULL;
g_object_unref (src->addr);
src->addr = NULL;
}
gst_udpsrc_reset_memory_allocator (src);
return TRUE;
}
static GstStateChangeReturn
gst_udpsrc_change_state (GstElement * element, GstStateChange transition)
{
GstUDPSrc *src;
GstStateChangeReturn result;
src = GST_UDPSRC (element);
switch (transition) {
case GST_STATE_CHANGE_NULL_TO_READY:
if (!gst_udpsrc_open (src))
goto open_failed;
break;
default:
break;
}
if ((result =
GST_ELEMENT_CLASS (parent_class)->change_state (element,
transition)) == GST_STATE_CHANGE_FAILURE)
goto failure;
switch (transition) {
case GST_STATE_CHANGE_READY_TO_NULL:
gst_udpsrc_close (src);
break;
default:
break;
}
return result;
/* ERRORS */
open_failed:
{
GST_DEBUG_OBJECT (src, "failed to open socket");
return GST_STATE_CHANGE_FAILURE;
}
failure:
{
GST_DEBUG_OBJECT (src, "parent failed state change");
return result;
}
}
/*** GSTURIHANDLER INTERFACE *************************************************/
static GstURIType
gst_udpsrc_uri_get_type (GType type)
{
return GST_URI_SRC;
}
static const gchar *const *
gst_udpsrc_uri_get_protocols (GType type)
{
static const gchar *protocols[] = { "udp", NULL };
return protocols;
}
static gchar *
gst_udpsrc_uri_get_uri (GstURIHandler * handler)
{
GstUDPSrc *src = GST_UDPSRC (handler);
return g_strdup (src->uri);
}
static gboolean
gst_udpsrc_uri_set_uri (GstURIHandler * handler, const gchar * uri,
GError ** error)
{
return gst_udpsrc_set_uri (GST_UDPSRC (handler), uri, error);
}
static void
gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data)
{
GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
iface->get_type = gst_udpsrc_uri_get_type;
iface->get_protocols = gst_udpsrc_uri_get_protocols;
iface->get_uri = gst_udpsrc_uri_get_uri;
iface->set_uri = gst_udpsrc_uri_set_uri;
}
Reference in a new issue View git blame Copy permalink