/* GStreamer * Copyright (C) <2005> Wim Taymans * Copyright (C) <2005> Nokia Corporation * * 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., 59 Temple Place - Suite 330, * Boston, MA 02111-1307, 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. * * Examples * * Here is a simple pipeline to read from the default port and dump the udp packets. * * gst-launch -v udpsrc ! fakesink dump=1 * * 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 -v audiotestsrc ! udpsink * * * * The udpsrc element supports automatic port allocation by setting the * "port" property to 0. the following pipeline reads UDP from a free port. * * gst-launch -v udpsrc port=0 ! fakesink * * * * udpsrc can read from multicast groups by setting the multicast_group property * to the IP address of the multicast group. * * * Alternatively one can provide a custom socket to udpsrc with the "sockfd" property, * udpsrc will then not allocate a socket itself but use the provided one. * * * The "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 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). * * * udpsrc implements a GstURIHandler interface that handles udp://host:port type * URIs. * * */ #ifdef HAVE_CONFIG_H #include "config.h" #endif #include "gstudpsrc.h" #include #include #include #ifdef HAVE_FIONREAD_IN_SYS_FILIO #include #endif GST_DEBUG_CATEGORY (udpsrc_debug); #define GST_CAT_DEFAULT (udpsrc_debug) /* the select call is also performed on the control sockets, that way * we can send special commands to unblock or restart the select call */ #define CONTROL_RESTART 'R' /* restart the select call */ #define CONTROL_STOP 'S' /* stop the select call */ #define CONTROL_SOCKETS(src) src->control_sock #define WRITE_SOCKET(src) src->control_sock[1] #define READ_SOCKET(src) src->control_sock[0] #define SEND_COMMAND(src, command) \ G_STMT_START { \ unsigned char c; c = command; \ write (WRITE_SOCKET(src), &c, 1); \ } G_STMT_END #define READ_COMMAND(src, command, res) \ G_STMT_START { \ res = read(READ_SOCKET(src), &command, 1); \ } G_STMT_END static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src", GST_PAD_SRC, GST_PAD_ALWAYS, GST_STATIC_CAPS_ANY); static const GstElementDetails gst_udpsrc_details = GST_ELEMENT_DETAILS ("UDP packet receiver", "Source/Network", "Receive data over the network via UDP", "Wim Taymans "); #define UDP_DEFAULT_PORT 4951 #define UDP_DEFAULT_MULTICAST_GROUP "0.0.0.0" #define UDP_DEFAULT_URI "udp://"UDP_DEFAULT_MULTICAST_GROUP":"G_STRINGIFY(UDP_DEFAULT_PORT) #define UDP_DEFAULT_CAPS NULL #define UDP_DEFAULT_SOCKFD -1 enum { PROP_0, PROP_PORT, PROP_MULTICAST_GROUP, PROP_URI, PROP_CAPS, PROP_SOCKFD /* FILL ME */ }; static void gst_udpsrc_uri_handler_init (gpointer g_iface, gpointer iface_data); static GstCaps *gst_udpsrc_getcaps (GstBaseSrc * src); static GstFlowReturn gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf); static gboolean gst_udpsrc_start (GstBaseSrc * bsrc); static gboolean gst_udpsrc_stop (GstBaseSrc * bsrc); static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc); 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 void _do_init (GType type) { static const GInterfaceInfo urihandler_info = { gst_udpsrc_uri_handler_init, NULL, NULL }; g_type_add_interface_static (type, GST_TYPE_URI_HANDLER, &urihandler_info); GST_DEBUG_CATEGORY_INIT (udpsrc_debug, "udpsrc", 0, "UDP src"); } GST_BOILERPLATE_FULL (GstUDPSrc, gst_udpsrc, GstPushSrc, GST_TYPE_PUSH_SRC, _do_init); static void gst_udpsrc_base_init (gpointer g_class) { GstElementClass *element_class = GST_ELEMENT_CLASS (g_class); gst_element_class_add_pad_template (element_class, gst_static_pad_template_get (&src_template)); gst_element_class_set_details (element_class, &gst_udpsrc_details); } static void gst_udpsrc_class_init (GstUDPSrcClass * klass) { GObjectClass *gobject_class; GstBaseSrcClass *gstbasesrc_class; GstPushSrcClass *gstpushsrc_class; gobject_class = (GObjectClass *) klass; gstbasesrc_class = (GstBaseSrcClass *) klass; gstpushsrc_class = (GstPushSrcClass *) klass; gobject_class->set_property = gst_udpsrc_set_property; gobject_class->get_property = gst_udpsrc_get_property; 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_object_class_install_property (gobject_class, PROP_MULTICAST_GROUP, g_param_spec_string ("multicast_group", "multicast_group", "The Address of multicast group to join", UDP_DEFAULT_MULTICAST_GROUP, G_PARAM_READWRITE)); 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_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_object_class_install_property (gobject_class, PROP_SOCKFD, g_param_spec_int ("sockfd", "socket handle", "Socket to use for UDP reception. (-1 == allocate)", -1, G_MAXINT, UDP_DEFAULT_SOCKFD, G_PARAM_READWRITE)); gstbasesrc_class->start = gst_udpsrc_start; gstbasesrc_class->stop = gst_udpsrc_stop; gstbasesrc_class->unlock = gst_udpsrc_unlock; gstbasesrc_class->get_caps = gst_udpsrc_getcaps; gstpushsrc_class->create = gst_udpsrc_create; } static void gst_udpsrc_init (GstUDPSrc * udpsrc, GstUDPSrcClass * g_class) { gst_base_src_set_live (GST_BASE_SRC (udpsrc), TRUE); udpsrc->port = UDP_DEFAULT_PORT; udpsrc->sock = UDP_DEFAULT_SOCKFD; udpsrc->multi_group = g_strdup (UDP_DEFAULT_MULTICAST_GROUP); udpsrc->uri = g_strdup (UDP_DEFAULT_URI); } static GstCaps * gst_udpsrc_getcaps (GstBaseSrc * src) { GstUDPSrc *udpsrc; udpsrc = GST_UDPSRC (src); if (udpsrc->caps) return gst_caps_ref (udpsrc->caps); else return gst_caps_new_any (); } static GstFlowReturn gst_udpsrc_create (GstPushSrc * psrc, GstBuffer ** buf) { GstUDPSrc *udpsrc; GstNetBuffer *outbuf; struct sockaddr_in tmpaddr; socklen_t len; fd_set read_fds; guint max_sock; gchar *pktdata; gint pktsize; gint readsize; gint ret; gboolean try_again; udpsrc = GST_UDPSRC (psrc); FD_ZERO (&read_fds); FD_SET (udpsrc->sock, &read_fds); FD_SET (READ_SOCKET (udpsrc), &read_fds); max_sock = MAX (udpsrc->sock, READ_SOCKET (udpsrc)); do { gboolean stop; try_again = FALSE; stop = FALSE; GST_LOG_OBJECT (udpsrc, "doing select"); ret = select (max_sock + 1, &read_fds, NULL, NULL, NULL); GST_LOG_OBJECT (udpsrc, "select returned %d", ret); if (ret <= 0) { if (errno != EAGAIN && errno != EINTR) goto select_error; else try_again = TRUE; } else { if (FD_ISSET (READ_SOCKET (udpsrc), &read_fds)) { /* got control message */ while (TRUE) { gchar command; int res; READ_COMMAND (udpsrc, command, res); if (res < 0) { GST_LOG_OBJECT (udpsrc, "no more commands"); /* no more commands */ break; } switch (command) { case CONTROL_STOP: /* break out of the select loop */ GST_LOG_OBJECT (udpsrc, "stop"); /* stop this function */ stop = TRUE; break; default: GST_WARNING_OBJECT (udpsrc, "unkown"); g_warning ("multiudpsink: unknown control message received"); break; } } } } if (stop) goto stopped; } while (try_again); /* ask how much is available for reading on the socket */ if ((ret = ioctl (udpsrc->sock, FIONREAD, &readsize)) < 0) goto ioctl_failed; GST_LOG_OBJECT (udpsrc, "ioctl says %d bytes available", readsize); pktdata = g_malloc (readsize); pktsize = readsize; len = sizeof (struct sockaddr); while (TRUE) { ret = recvfrom (udpsrc->sock, pktdata, pktsize, 0, (struct sockaddr *) &tmpaddr, &len); if (ret < 0) { if (errno != EAGAIN && errno != EINTR) goto receive_error; } else break; } outbuf = gst_netbuffer_new (); GST_BUFFER_DATA (outbuf) = (guint8 *) pktdata; GST_BUFFER_MALLOCDATA (outbuf) = (guint8 *) pktdata; GST_BUFFER_SIZE (outbuf) = ret; gst_netaddress_set_ip4_address (&outbuf->from, tmpaddr.sin_addr.s_addr, tmpaddr.sin_port); gst_buffer_set_caps (GST_BUFFER (outbuf), udpsrc->caps); GST_LOG_OBJECT (udpsrc, "read %d bytes", readsize); *buf = GST_BUFFER (outbuf); return GST_FLOW_OK; select_error: { GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL), ("select error %d: %s (%d)", ret, g_strerror (errno), errno)); return GST_FLOW_ERROR; } stopped: { GST_DEBUG ("stop called"); return GST_FLOW_WRONG_STATE; } ioctl_failed: { GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL), ("ioctl failed %d: %s (%d)", ret, g_strerror (errno), errno)); return GST_FLOW_ERROR; } receive_error: { g_free (pktdata); GST_ELEMENT_ERROR (udpsrc, RESOURCE, READ, (NULL), ("receive error %d: %s (%d)", ret, g_strerror (errno), errno)); return GST_FLOW_ERROR; } } /* Call this function when multicastgroup and/or port are updated */ static void gst_udpsrc_update_uri (GstUDPSrc * src) { g_free (src->uri); src->uri = g_strdup_printf ("udp://%s:%d", src->multi_group, src->port); } static gboolean gst_udpsrc_set_uri (GstUDPSrc * src, const gchar * uri) { gchar *protocol; gchar *location; gchar *colptr; protocol = gst_uri_get_protocol (uri); if (strcmp (protocol, "udp") != 0) goto wrong_protocol; g_free (protocol); location = gst_uri_get_location (uri); if (!location) return FALSE; colptr = strstr (location, ":"); if (colptr != NULL) { g_free (src->multi_group); src->multi_group = g_strndup (location, colptr - location); src->port = atoi (colptr + 1); } else { g_free (src->multi_group); src->multi_group = g_strdup (location); src->port = UDP_DEFAULT_PORT; } g_free (location); gst_udpsrc_update_uri (src); return TRUE; wrong_protocol: { g_free (protocol); GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), ("error parsing uri %s: wrong protocol", 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_PORT: udpsrc->port = g_value_get_int (value); gst_udpsrc_update_uri (udpsrc); break; case PROP_MULTICAST_GROUP: g_free (udpsrc->multi_group); if (g_value_get_string (value) == NULL) udpsrc->multi_group = g_strdup (UDP_DEFAULT_MULTICAST_GROUP); else udpsrc->multi_group = g_strdup (g_value_get_string (value)); gst_udpsrc_update_uri (udpsrc); break; case PROP_URI: gst_udpsrc_set_uri (udpsrc, g_value_get_string (value)); 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); } old_caps = udpsrc->caps; udpsrc->caps = new_caps; if (old_caps) gst_caps_unref (old_caps); gst_pad_set_caps (GST_BASE_SRC (udpsrc)->srcpad, new_caps); break; } case PROP_SOCKFD: udpsrc->sock = g_value_get_int (value); GST_DEBUG ("setting SOCKFD to %d", udpsrc->sock); 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_PORT: g_value_set_int (value, udpsrc->port); break; case PROP_MULTICAST_GROUP: g_value_set_string (value, udpsrc->multi_group); break; case PROP_URI: g_value_set_string (value, udpsrc->uri); break; case PROP_CAPS: gst_value_set_caps (value, udpsrc->caps); break; case PROP_SOCKFD: g_value_set_int (value, udpsrc->sock); break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); break; } } /* create a socket for sending to remote machine */ static gboolean gst_udpsrc_start (GstBaseSrc * bsrc) { guint bc_val; gint reuse; struct sockaddr_in my_addr; guint len; int port; GstUDPSrc *src; gint ret; src = GST_UDPSRC (bsrc); GST_DEBUG_OBJECT (src, "creating socket pair"); if ((ret = socketpair (PF_UNIX, SOCK_STREAM, 0, CONTROL_SOCKETS (src))) < 0) goto no_socket_pair; fcntl (READ_SOCKET (src), F_SETFL, O_NONBLOCK); fcntl (WRITE_SOCKET (src), F_SETFL, O_NONBLOCK); if (src->sock == -1) { if ((ret = socket (AF_INET, SOCK_DGRAM, 0)) < 0) goto no_socket; src->sock = ret; reuse = 1; if ((ret = setsockopt (src->sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof (reuse))) < 0) goto setsockopt_error; memset (&src->myaddr, 0, sizeof (src->myaddr)); src->myaddr.sin_family = AF_INET; /* host byte order */ src->myaddr.sin_port = htons (src->port); /* short, network byte order */ src->myaddr.sin_addr.s_addr = INADDR_ANY; GST_DEBUG_OBJECT (src, "binding on port %d", src->port); if ((ret = bind (src->sock, (struct sockaddr *) &src->myaddr, sizeof (src->myaddr))) < 0) goto bind_error; } if (inet_aton (src->multi_group, &(src->multi_addr.imr_multiaddr))) { if (src->multi_addr.imr_multiaddr.s_addr) { src->multi_addr.imr_interface.s_addr = INADDR_ANY; if ((ret = setsockopt (src->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &src->multi_addr, sizeof (src->multi_addr))) < 0) goto membership; } } len = sizeof (my_addr); if ((ret = getsockname (src->sock, (struct sockaddr *) &my_addr, &len)) < 0) goto getsockname_error; port = ntohs (my_addr.sin_port); GST_DEBUG_OBJECT (src, "bound, on port %d", port); if (port != src->port) { src->port = port; GST_DEBUG_OBJECT (src, "notifying %d", port); g_object_notify (G_OBJECT (src), "port"); } bc_val = 1; if ((ret = setsockopt (src->sock, SOL_SOCKET, SO_BROADCAST, &bc_val, sizeof (bc_val))) < 0) goto no_broadcast; src->myaddr.sin_port = htons (src->port + 1); return TRUE; /* ERRORS */ no_socket_pair: { GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ_WRITE, (NULL), ("no socket pair %d: %s (%d)", ret, g_strerror (errno), errno)); return FALSE; } no_socket: { GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL), ("no socket error %d: %s (%d)", ret, g_strerror (errno), errno)); return FALSE; } setsockopt_error: { close (src->sock); src->sock = -1; GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL), ("setsockopt failed %d: %s (%d)", ret, g_strerror (errno), errno)); return FALSE; } bind_error: { close (src->sock); src->sock = -1; GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL), ("bind failed %d: %s (%d)", ret, g_strerror (errno), errno)); return FALSE; } membership: { close (src->sock); src->sock = -1; GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL), ("could add membership %d: %s (%d)", ret, g_strerror (errno), errno)); return FALSE; } getsockname_error: { close (src->sock); src->sock = -1; GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL), ("getsockname failed %d: %s (%d)", ret, g_strerror (errno), errno)); return FALSE; } no_broadcast: { close (src->sock); src->sock = -1; GST_ELEMENT_ERROR (src, RESOURCE, SETTINGS, (NULL), ("could not configure socket for broadcast %d: %s (%d)", ret, g_strerror (errno), errno)); return FALSE; } } static gboolean gst_udpsrc_unlock (GstBaseSrc * bsrc) { GstUDPSrc *src; src = GST_UDPSRC (bsrc); GST_DEBUG ("sending stop command"); SEND_COMMAND (src, CONTROL_STOP); return TRUE; } static gboolean gst_udpsrc_stop (GstBaseSrc * bsrc) { GstUDPSrc *src; src = GST_UDPSRC (bsrc); if (src->sock != -1) { close (src->sock); src->sock = -1; } if (src->control_sock[0] != -1) { close (src->control_sock[0]); src->control_sock[0] = -1; } if (src->control_sock[1] != -1) { close (src->control_sock[1]); src->control_sock[1] = -1; } return TRUE; } /*** GSTURIHANDLER INTERFACE *************************************************/ static guint gst_udpsrc_uri_get_type (void) { return GST_URI_SRC; } static gchar ** gst_udpsrc_uri_get_protocols (void) { static gchar *protocols[] = { "udp", NULL }; return protocols; } static const gchar * gst_udpsrc_uri_get_uri (GstURIHandler * handler) { GstUDPSrc *src = GST_UDPSRC (handler); return src->uri; } static gboolean gst_udpsrc_uri_set_uri (GstURIHandler * handler, const gchar * uri) { gboolean ret; GstUDPSrc *src = GST_UDPSRC (handler); ret = gst_udpsrc_set_uri (src, uri); return ret; } 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; }