diff options
Diffstat (limited to 'gst')
-rw-r--r-- | gst/dccp/Makefile.am | 36 | ||||
-rw-r--r-- | gst/dccp/gstdccp.c | 450 | ||||
-rw-r--r-- | gst/dccp/gstdccp.h | 96 | ||||
-rw-r--r-- | gst/dccp/gstdccpclientsink.c | 322 | ||||
-rw-r--r-- | gst/dccp/gstdccpclientsink.h | 81 | ||||
-rw-r--r-- | gst/dccp/gstdccpclientsrc.c | 398 | ||||
-rw-r--r-- | gst/dccp/gstdccpclientsrc.h | 77 | ||||
-rw-r--r-- | gst/dccp/gstdccpplugin.c | 60 | ||||
-rw-r--r-- | gst/dccp/gstdccpserversink.c | 435 | ||||
-rw-r--r-- | gst/dccp/gstdccpserversink.h | 96 | ||||
-rw-r--r-- | gst/dccp/gstdccpserversrc.c | 403 | ||||
-rw-r--r-- | gst/dccp/gstdccpserversrc.h | 94 |
12 files changed, 2548 insertions, 0 deletions
diff --git a/gst/dccp/Makefile.am b/gst/dccp/Makefile.am new file mode 100644 index 000000000..647d2500a --- /dev/null +++ b/gst/dccp/Makefile.am @@ -0,0 +1,36 @@ +# plugindir is set in configure + +############################################################################## +# change libgstplugin.la to something more suitable, e.g. libmysomething.la # +############################################################################## +plugin_LTLIBRARIES = libgstdccp.la + +############################################################################## +# for the next set of variables, rename the prefix if you renamed the .la, # +# e.g. libgstplugin_la_SOURCES => libmysomething_la_SOURCES # +# libgstplugin_la_CFLAGS => libmysomething_la_CFLAGS # +# libgstplugin_la_LIBADD => libmysomething_la_LIBADD # +# libgstplugin_la_LDFLAGS => libmysomething_la_LDFLAGS # +############################################################################## + +# sources used to compile this plug-in +libgstdccp_la_SOURCES = gstdccpplugin.c \ + gstdccp.c \ + gstdccpserversink.c \ + gstdccpserversrc.c \ + gstdccpclientsink.c \ + gstdccpclientsrc.c + +# flags used to compile this plugin +# add other _CFLAGS and _LIBS as needed +libgstdccp_la_CFLAGS = $(GST_CFLAGS) +libgstdccp_la_LIBADD = $(GST_BASE_LIBS) $(GST_LIBS) +libgstdccp_la_LDFLAGS = $(GST_PLUGIN_LDFLAGS) + +# headers we need but don't want installed +noinst_HEADERS = gstdccp.h \ + gstdccpserversink.h \ + gstdccpserversrc.h \ + gstdccpclientsink.h \ + gstdccpclientsrc.h + diff --git a/gst/dccp/gstdccp.c b/gst/dccp/gstdccp.c new file mode 100644 index 000000000..a53af7dc8 --- /dev/null +++ b/gst/dccp/gstdccp.c @@ -0,0 +1,450 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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. + */ + +#include "gstdccp.h" +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> +#include <arpa/inet.h> +#include <netdb.h> +#include <unistd.h> +#include <sys/ioctl.h> +#include <string.h> + +/* Prototypes and definitions for private functions and not exported via gstdccp.h */ +gint gst_dccp_socket_write (int socket, const void *buf, size_t count, + int packet_size); +gboolean gst_dccp_socket_connected (GstElement * element, int server_sock_fd); +struct sockaddr_in gst_dccp_create_sockaddr (GstElement * element, gchar * ip, + int port); + +/* Resolves host to IP address + * @return a gchar pointer containing the ip address or NULL + */ +gchar * +gst_dccp_host_to_ip (GstElement * element, const gchar * host) +{ + struct hostent *hostinfo; + char **addrs; + gchar *ip; + struct in_addr addr; + + GST_DEBUG_OBJECT (element, "resolving host %s", host); + + /* first check if it already is an IP address */ + if (inet_aton (host, &addr)) { + ip = g_strdup (host); + GST_DEBUG_OBJECT (element, "resolved to IP %s", ip); + return ip; + } + + /* perform a name lookup */ + if (!(hostinfo = gethostbyname (host))) { + GST_ELEMENT_ERROR (element, RESOURCE, NOT_FOUND, (NULL), + ("Could not find IP address for host \"%s\".", host)); + return NULL; + } + + if (hostinfo->h_addrtype != AF_INET) { + GST_ELEMENT_ERROR (element, RESOURCE, NOT_FOUND, (NULL), + ("host \"%s\" is not an IP host", host)); + return NULL; + } + + addrs = hostinfo->h_addr_list; + + /* There could be more than one IP address, but we just return the first */ + ip = g_strdup (inet_ntoa (*(struct in_addr *) *addrs)); + + return ip; +} + +/* Read a buffer from the given socket + * @returns: + * a GstBuffer from which data should be read + * or NULL, indicating a connection close or an error. Handle it with EOS. + */ +GstFlowReturn +gst_dccp_read_buffer (GstElement * this, int socket, GstBuffer ** buf) +{ + fd_set testfds; + int maxfdp1; + int ret; + ssize_t bytes_read; + int readsize; + + *buf = NULL; + + /* do a blocking select on the socket */ + FD_ZERO (&testfds); + FD_SET (socket, &testfds); + maxfdp1 = socket + 1; + + /* no action (0) is also an error in our case */ + if ((ret = select (maxfdp1, &testfds, NULL, NULL, 0)) <= 0) { + GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), + ("select failed: %s", g_strerror (errno))); + return GST_FLOW_ERROR; + } + + /* ask how much is available for reading on the socket */ + if ((ret = ioctl (socket, FIONREAD, &readsize)) < 0) { + GST_ELEMENT_ERROR (this, RESOURCE, READ, (NULL), + ("read FIONREAD value failed: %s", g_strerror (errno))); + return GST_FLOW_ERROR; + } + + if (readsize == 0) { + GST_DEBUG_OBJECT (this, "Got EOS on socket stream"); + return GST_FLOW_UNEXPECTED; + } + + *buf = gst_buffer_new_and_alloc (readsize); + bytes_read = read (socket, GST_BUFFER_DATA (*buf), readsize); + + GST_LOG_OBJECT (this, "bytes read %d\n", bytes_read); + GST_LOG_OBJECT (this, "returning buffer of size %d", GST_BUFFER_SIZE (*buf)); + + return GST_FLOW_OK; +} + +/* Create a new socket + * @return the socket file descriptor + */ +gint +gst_dccp_create_new_socket (GstElement * element) +{ + int sock_fd; + if ((sock_fd = socket (AF_INET, SOCK_DCCP, IPPROTO_DCCP)) < 0) { + GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL), GST_ERROR_SYSTEM); + } + + return sock_fd; +} + +/* Connect to a server + * @return true in case of successfull connection, false otherwise + */ +gboolean +gst_dccp_connect_to_server (GstElement * element, struct sockaddr_in server_sin, + int sock_fd) +{ + GST_DEBUG_OBJECT (element, "connecting to server"); + + if (connect (sock_fd, (struct sockaddr *) &server_sin, sizeof (server_sin))) { + switch (errno) { + case ECONNREFUSED: + GST_ERROR_OBJECT (element, "Connection refused."); + return FALSE; + break; + default: + GST_ERROR_OBJECT (element, "Connection failed."); + return FALSE; + break; + } + } + return TRUE; +} + +/* FIXME support only one client */ +/* + * Accept connection on the server socket. + * @return the socket of the client connected to the server. + */ +gint +gst_dccp_server_wait_connections (GstElement * element, int server_sock_fd) +{ + /* new client */ + int client_sock_fd; + struct sockaddr_in client_address; + unsigned int client_address_len; + + /* For some stupid reason, client_address and client_address_len has to be + * zeroed */ + memset (&client_address, 0, sizeof (client_address)); + client_address_len = 0; + + if ((client_sock_fd = + accept (server_sock_fd, (struct sockaddr *) &client_address, + &client_address_len)) == -1) { + return -1; + } + /* to support multiple connection, fork here a new thread passing the + * client_sock_fd returned by accept function. + */ + + GST_DEBUG_OBJECT (element, "added new client ip %s with fd %d", + inet_ntoa (client_address.sin_addr), client_sock_fd); + + /* return the thread object, instead of the fd */ + return client_sock_fd; +} + +/* + * Bind a server address. + * @return true in success, false otherwise. + */ +gboolean +gst_dccp_bind_server_socket (GstElement * element, int server_sock_fd, + struct sockaddr_in server_sin) +{ + int ret; + + GST_DEBUG_OBJECT (element, "binding server socket to address"); + ret = bind (server_sock_fd, (struct sockaddr *) &server_sin, + sizeof (server_sin)); + if (ret) { + switch (errno) { + default: + GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL), + ("bind on port %d failed: %s", server_sin.sin_port, + g_strerror (errno))); + return FALSE; + break; + } + } + return TRUE; +} + +gboolean +gst_dccp_listen_server_socket (GstElement * element, int server_sock_fd) +{ + + GST_DEBUG_OBJECT (element, "listening on server socket %d with queue of %d", + server_sock_fd, DCCP_BACKLOG); + if (listen (server_sock_fd, DCCP_BACKLOG) == -1) { + GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL), + ("Could not listen on server socket: %s", g_strerror (errno))); + return FALSE; + } + GST_DEBUG_OBJECT (element, + "listened on server socket %d, returning from connection setup", + server_sock_fd); + + return TRUE; +} + +/* FIXME */ +gboolean +gst_dccp_socket_connected (GstElement * element, int server_sock_fd) +{ + return FALSE; +} + +/* Write buffer to given socket incrementally. + * Returns number of bytes written. + */ +gint +gst_dccp_socket_write (int socket, const void *buf, size_t size, + int packet_size) +{ + size_t bytes_written = 0; + ssize_t wrote; + + while (bytes_written < size) { + do { + wrote = write (socket, (const char *) buf + bytes_written, + MIN (packet_size, size - bytes_written)); + } while (wrote == -1 && errno == EAGAIN); + + /* TODO print the send error */ + bytes_written += wrote; + } + + if (bytes_written < 0) + GST_WARNING ("error while writing"); + else + GST_LOG ("wrote %" G_GSIZE_FORMAT " bytes succesfully", bytes_written); + return bytes_written; +} + +GstFlowReturn +gst_dccp_send_buffer (GstElement * this, GstBuffer * buffer, int client_sock_fd, + int packet_size) +{ + size_t wrote; + gint size = 0; + guint8 *data; + + size = GST_BUFFER_SIZE (buffer); + data = GST_BUFFER_DATA (buffer); + + GST_LOG_OBJECT (this, "writing %d bytes\n", size); + + if (packet_size < 0) { + GST_LOG_OBJECT (this, "error getting MTU\n"); + return GST_FLOW_ERROR; + } + + wrote = gst_dccp_socket_write (client_sock_fd, data, size, packet_size); + + if (wrote != size) { + GST_DEBUG_OBJECT (this, ("Error while sending data")); + return GST_FLOW_ERROR; + } + return GST_FLOW_OK; +} + +/* + * Create socket address. + * @return sockaddr_in. + */ +struct sockaddr_in +gst_dccp_create_sockaddr (GstElement * element, gchar * ip, int port) +{ + struct sockaddr_in sin; + + memset (&sin, 0, sizeof (sin)); + sin.sin_family = AF_INET; /* network socket */ + sin.sin_port = htons (port); /* on port */ + sin.sin_addr.s_addr = inet_addr (ip); /* on host ip */ + + return sin; +} + +gboolean +gst_dccp_make_address_reusable (GstElement * element, int sock_fd) +{ + int ret = 1; + /* make address reusable */ + if (setsockopt (sock_fd, SOL_SOCKET, SO_REUSEADDR, + (void *) &ret, sizeof (ret)) < 0) { + GST_ELEMENT_ERROR (element, RESOURCE, SETTINGS, (NULL), + ("Could not setsockopt: %s", g_strerror (errno))); + return FALSE; + } + return TRUE; +} + +/* DCCP socket specific stuffs */ +gboolean +gst_dccp_set_ccid (GstElement * element, int sock_fd, uint8_t ccid) +{ + uint8_t ccids[4]; /* for getting the available CCIDs, should be large enough */ + socklen_t len = sizeof (ccids); + int i, ret; + gboolean ccid_supported = FALSE; + + /* + * Determine which CCIDs are available on the host + */ + ret = getsockopt (sock_fd, SOL_DCCP, DCCP_SOCKOPT_AVAILABLE_CCIDS, &ccids, + &len); + if (ret < 0) { + GST_ERROR_OBJECT (element, "Can not determine available CCIDs"); + return FALSE; + } + + for (i = 0; i < sizeof (ccids); i++) { + if (ccid == ccids[i]) { + ccid_supported = TRUE; + } + } + + if (!ccid_supported) { + GST_ERROR_OBJECT (element, "CCID specified is not supported"); + return FALSE; + } + + if (setsockopt (sock_fd, SOL_DCCP, DCCP_SOCKOPT_CCID, &ccid, + sizeof (ccid)) < 0) { + GST_ERROR_OBJECT (element, "Can not set CCID"); + return FALSE; + } + + return TRUE; +} + +/* + * Get the current ccid of TX or RX half-connection. tx_or_rx parameter must be + * DCCP_SOCKOPT_TX_CCID or DCCP_SOCKOPT_RX_CCID. + * @return ccid or -1 on error or tx_or_rx not the correct option + */ +uint8_t +gst_dccp_get_ccid (GstElement * element, int sock_fd, int tx_or_rx) +{ + uint8_t ccid; + socklen_t ccidlen; + int ret; + + switch (tx_or_rx) { + case DCCP_SOCKOPT_TX_CCID: + case DCCP_SOCKOPT_RX_CCID: + break; + default: + return -1; + } + + ccidlen = sizeof (ccid); + ret = getsockopt (sock_fd, SOL_DCCP, tx_or_rx, &ccid, &ccidlen); + if (ret < 0) { + GST_ERROR_OBJECT (element, "Can not determine available CCIDs"); + return -1; + } + return ccid; +} + +gint +gst_dccp_get_max_packet_size (GstElement * element, int sock) +{ + int size; + socklen_t sizelen = sizeof (size); + if (getsockopt (sock, SOL_DCCP, DCCP_SOCKOPT_GET_CUR_MPS, + &size, &sizelen) < 0) { + GST_ELEMENT_ERROR (element, RESOURCE, SETTINGS, (NULL), + ("Could not get current MTU %d: %s", errno, g_strerror (errno))); + return -1; + } + GST_DEBUG_OBJECT (element, "MTU: %d", size); + return size; +} + +/* Still not used and need to be FIXED */ +gboolean +gst_dccp_set_sock_windowsize (GstElement * element, int sock, int winSize, + gboolean inSend) +{ +#ifdef SO_SNDBUF + int rc; + + if (!inSend) { + /* receive buffer -- set + * note: results are verified after connect() or listen(), + * since some OS's don't show the corrected value until then. */ + rc = setsockopt (sock, SOL_DCCP, SO_RCVBUF, + (char *) &winSize, sizeof (winSize)); + GST_DEBUG_OBJECT (element, "set rcv sockbuf: %d\n", winSize); + } else { + /* send buffer -- set + * note: results are verified after connect() or listen(), + * since some OS's don't show the corrected value until then. */ + rc = setsockopt (sock, SOL_DCCP, SO_SNDBUF, + (char *) &winSize, sizeof (winSize)); + GST_DEBUG_OBJECT (element, "set snd sockbuf: %d\n", winSize); + } + + if (rc < 0) { + GST_ELEMENT_ERROR (element, RESOURCE, SETTINGS, (NULL), + ("Could not set window size %d: %s", errno, g_strerror (errno))); + return FALSE; + } +#endif /* SO_SNDBUF */ + + return TRUE; +} diff --git a/gst/dccp/gstdccp.h b/gst/dccp/gstdccp.h new file mode 100644 index 000000000..117b0df50 --- /dev/null +++ b/gst/dccp/gstdccp.h @@ -0,0 +1,96 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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. + */ + +#ifndef __GST_DCCP_H__ +#define __GST_DCCP_H__ + +#include <gst/gst.h> +#include <gst/base/gstadapter.h> +#include <sys/socket.h> +#include <netinet/in.h> /* sockaddr_in */ + +/* DCCP socket general options */ +#define DCCP_BACKLOG 5 +#ifndef SOCK_DCCP + #define SOCK_DCCP 6 +#endif + +#ifndef IPPROTO_DCCP + #define IPPROTO_DCCP 33 +#endif + +#ifndef SOL_DCCP + #define SOL_DCCP 269 +#endif + +/* dccp socket specific options */ +#define DCCP_SOCKOPT_PACKET_SIZE 1 /* XXX deprecated, without effect */ +#define DCCP_SOCKOPT_SERVICE 2 +#define DCCP_SOCKOPT_CHANGE_L 3 +#define DCCP_SOCKOPT_CHANGE_R 4 +#define DCCP_SOCKOPT_GET_CUR_MPS 5 +#define DCCP_SOCKOPT_SERVER_TIMEWAIT 6 +#define DCCP_SOCKOPT_SEND_CSCOV 10 +#define DCCP_SOCKOPT_RECV_CSCOV 11 +#define DCCP_SOCKOPT_AVAILABLE_CCIDS 12 +#define DCCP_SOCKOPT_CCID 13 +#define DCCP_SOCKOPT_TX_CCID 14 +#define DCCP_SOCKOPT_RX_CCID 15 +#define DCCP_SOCKOPT_CCID_RX_INFO 128 +#define DCCP_SOCKOPT_CCID_TX_INFO 192 + +/* Default parameters for the gst dccp element property */ +#define DCCP_DEFAULT_PORT 5001 +#define DCCP_DEFAULT_SOCK_FD -1 +#define DCCP_DEFAULT_CLIENT_SOCK_FD -1 +#define DCCP_DEFAULT_CLOSED TRUE +#define DCCP_DEFAULT_WAIT_CONNECTIONS FALSE +#define DCCP_DEFAULT_HOST "127.0.0.1" +#define DCCP_DEFAULT_CCID 2 + +#define DCCP_DELTA 100 + +gchar *gst_dccp_host_to_ip (GstElement * element, const gchar * host); + +GstFlowReturn gst_dccp_read_buffer (GstElement * this, int socket, + GstBuffer ** buf); + +gint gst_dccp_create_new_socket (GstElement * element); +gboolean gst_dccp_connect_to_server (GstElement * element, + struct sockaddr_in server_sin, + int sock_fd); + +gint gst_dccp_server_wait_connections (GstElement * element, int server_sock_fd); + +gboolean gst_dccp_bind_server_socket (GstElement * element, int server_sock_fd, + struct sockaddr_in server_sin); + +gboolean gst_dccp_listen_server_socket (GstElement * element, int server_sock_fd); +gboolean gst_dccp_set_ccid (GstElement * element, int sock_fd, uint8_t ccid); + +gint gst_dccp_get_max_packet_size(GstElement * element, int sock); + +GstFlowReturn gst_dccp_send_buffer (GstElement * element, GstBuffer * buffer, + int client_sock_fd, int packet_size); + +gboolean gst_dccp_make_address_reusable (GstElement * element, int sock_fd); +gboolean gst_dccp_set_sock_windowsize(GstElement * element, int sock, int winSize, + gboolean inSend); + +#endif /* __GST_DCCP_H__ */ diff --git a/gst/dccp/gstdccpclientsink.c b/gst/dccp/gstdccpclientsink.c new file mode 100644 index 000000000..d8fca1fbd --- /dev/null +++ b/gst/dccp/gstdccpclientsink.c @@ -0,0 +1,322 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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-dccpclientsink + * @see_also: dccpserversrc, dccpclientsrc, dccpserversink + * + * This element connect to a DCCP server and send data to it. + * <ulink url="http://www.linuxfoundation.org/en/Net:DCCP">DCCP</ulink> (Datagram + * Congestion Control Protocol) is a Transport Layer protocol like + * TCP and UDP. + * + * <refsect2> + * <title>Example pipeline</title> + * <para> + * |[ + * gst-launch -v filesrc location=music.mp3 ! mp3parse ! dccpclientsink host=localhost port=9011 ccid=2 + * ]| Client + * |[ + * gst-launch -v dccpserversrc port=9011 ccid=2 ! decodebin ! alsasink + * ]| Server + * + * This example pipeline will send a MP3 stream to the server using DCCP. + * The server will decode the MP3 and play it. + * Run the server pipeline first than the client pipeline. + * </para> + * </refsect2> + */ + + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "gstdccpclientsink.h" +#include "gstdccp.h" +#include <string.h> /* memset */ +#include <unistd.h> +#include <arpa/inet.h> +#include <fcntl.h> + +/* signals */ +enum +{ + SIGNAL_CONNECTED, + LAST_SIGNAL +}; + +/* properties */ +enum +{ + PROP_0, + PROP_PORT, + PROP_HOST, + PROP_SOCK_FD, + PROP_CCID, + PROP_CLOSE_FD +}; + +static gboolean gst_dccp_client_sink_stop (GstBaseSink * bsink); +static gboolean gst_dccp_client_sink_start (GstBaseSink * bsink); +static GstFlowReturn gst_dccp_client_sink_render (GstBaseSink * bsink, + GstBuffer * buf); + +GST_DEBUG_CATEGORY_STATIC (dccpclientsink_debug); + +static const GstElementDetails gst_dccp_client_sink_details = +GST_ELEMENT_DETAILS ("DCCP client sink", + "Sink/Network", + "Send data as a client over the network via DCCP", + "E-Phone Team at Federal University of Campina Grande <leandroal@gmail.com>"); + +static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink", + GST_PAD_SINK, + GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY); + +GST_BOILERPLATE (GstDCCPClientSink, gst_dccp_client_sink, GstBaseSink, + GST_TYPE_BASE_SINK); + +static guint gst_dccp_client_sink_signals[LAST_SIGNAL] = { 0 }; + +static GstFlowReturn +gst_dccp_client_sink_render (GstBaseSink * bsink, GstBuffer * buf) +{ + GstDCCPClientSink *sink = GST_DCCP_CLIENT_SINK (bsink); + + return gst_dccp_send_buffer (GST_ELEMENT (sink), buf, sink->sock_fd, + sink->pksize); +} + +/* + * Set the value of a property for the client sink. + */ +static void +gst_dccp_client_sink_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstDCCPClientSink *sink = GST_DCCP_CLIENT_SINK (object); + + switch (prop_id) { + case PROP_PORT: + sink->port = g_value_get_int (value); + break; + case PROP_SOCK_FD: + sink->sock_fd = g_value_get_int (value); + break; + case PROP_HOST: + if (!g_value_get_string (value)) { + g_warning ("host property cannot be NULL"); + break; + } + g_free (sink->host); + sink->host = g_strdup (g_value_get_string (value)); + break; + case PROP_CLOSE_FD: + sink->closed = g_value_get_boolean (value); + break; + case PROP_CCID: + sink->ccid = g_value_get_int (value); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +/* + * Get a given property value for the client sink. + */ +static void +gst_dccp_client_sink_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstDCCPClientSink *sink = GST_DCCP_CLIENT_SINK (object); + + switch (prop_id) { + case PROP_PORT: + g_value_set_int (value, sink->port); + break; + case PROP_SOCK_FD: + g_value_set_int (value, sink->sock_fd); + break; + case PROP_HOST: + g_value_set_string (value, sink->host); + break; + case PROP_CLOSE_FD: + g_value_set_boolean (value, sink->closed); + break; + case PROP_CCID: + g_value_set_int (value, sink->ccid); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static gboolean +gst_dccp_client_sink_start (GstBaseSink * bsink) +{ + GstDCCPClientSink *sink = GST_DCCP_CLIENT_SINK (bsink); + + if (sink->sock_fd == DCCP_DEFAULT_SOCK_FD) { + gchar *ip = NULL; + + /* look up name if we need to */ + if (!(ip = gst_dccp_host_to_ip (GST_ELEMENT (sink), sink->host))) { + GST_ERROR_OBJECT (sink, "cannot resolve hostname"); + gst_dccp_client_sink_stop (GST_BASE_SINK (sink)); + return FALSE; + } + + /* name the server socket */ + memset (&sink->server_sin, 0, sizeof (sink->server_sin)); + sink->server_sin.sin_family = AF_INET; /* network socket */ + sink->server_sin.sin_port = htons (sink->port); /* on port */ + sink->server_sin.sin_addr.s_addr = inet_addr (ip); /* on host ip */ + g_free (ip); + + /* create socket */ + if ((sink->sock_fd = gst_dccp_create_new_socket (GST_ELEMENT (sink))) < 0) { + return FALSE; + } + + if (!gst_dccp_set_ccid (GST_ELEMENT (sink), sink->sock_fd, sink->ccid)) { + gst_dccp_client_sink_stop (GST_BASE_SINK (sink)); + return FALSE; + } + + if (!gst_dccp_connect_to_server (GST_ELEMENT (sink), sink->server_sin, + sink->sock_fd)) { + gst_dccp_client_sink_stop (GST_BASE_SINK (sink)); + return FALSE; + } + + /* the socket is connected */ + g_signal_emit (sink, gst_dccp_client_sink_signals[SIGNAL_CONNECTED], 0, + sink->sock_fd); + } + + sink->pksize = + gst_dccp_get_max_packet_size (GST_ELEMENT (sink), sink->sock_fd); + + return TRUE; +} + +static void +gst_dccp_client_sink_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 (&sinktemplate)); + + gst_element_class_set_details (element_class, &gst_dccp_client_sink_details); +} + +static void +gst_dccp_client_sink_init (GstDCCPClientSink * this, + GstDCCPClientSinkClass * g_class) +{ + this->port = DCCP_DEFAULT_PORT; + this->host = g_strdup (DCCP_DEFAULT_HOST); + this->sock_fd = DCCP_DEFAULT_SOCK_FD; + this->closed = DCCP_DEFAULT_CLOSED; + this->ccid = DCCP_DEFAULT_CCID; +} + +static gboolean +gst_dccp_client_sink_stop (GstBaseSink * bsink) +{ + GstDCCPClientSink *sink; + + sink = GST_DCCP_CLIENT_SINK (bsink); + + if (sink->sock_fd != -1 && sink->closed) { + GST_DEBUG_OBJECT (sink, "closing socket"); + close (sink->sock_fd); + sink->sock_fd = -1; + } + + return TRUE; +} + +/* + * Define the gst class, callbacks, etc. + */ +static void +gst_dccp_client_sink_class_init (GstDCCPClientSinkClass * klass) +{ + GObjectClass *gobject_class; + GstBaseSinkClass *gstbasesink_class; + + gobject_class = (GObjectClass *) klass; + gstbasesink_class = (GstBaseSinkClass *) klass; + + gobject_class->set_property = gst_dccp_client_sink_set_property; + gobject_class->get_property = gst_dccp_client_sink_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, + DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); + +/* FIXME property server_ip */ + g_object_class_install_property (gobject_class, PROP_HOST, + g_param_spec_string ("host", "Host", + "The host IP address to send packets to", DCCP_DEFAULT_HOST, + G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_SOCK_FD, + g_param_spec_int ("sockfd", "Socket fd", + "The socket file descriptor", -1, G_MAXINT, + DCCP_DEFAULT_SOCK_FD, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CLOSE_FD, + g_param_spec_boolean ("close-socket", "Close", + "Close socket at end of stream", + DCCP_DEFAULT_CLOSED, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CCID, + g_param_spec_int ("ccid", "CCID", + "The Congestion Control IDentified to be used", 2, G_MAXINT, + DCCP_DEFAULT_CCID, G_PARAM_READWRITE)); + + /* signals */ + /** + * GstDccpClientSink::connected: + * @sink: the gstdccpclientsink instance + * @fd: the connected socket fd + * + * Sign that the element has connected, return the fd of the socket. + */ + gst_dccp_client_sink_signals[SIGNAL_CONNECTED] = + g_signal_new ("connected", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST, + G_STRUCT_OFFSET (GstDCCPClientSinkClass, connected), NULL, NULL, + gst_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT); + + gstbasesink_class->start = gst_dccp_client_sink_start; + gstbasesink_class->stop = gst_dccp_client_sink_stop; + gstbasesink_class->render = gst_dccp_client_sink_render; + + GST_DEBUG_CATEGORY_INIT (dccpclientsink_debug, "dccpclientsink", 0, + "DCCP Client Sink"); +} diff --git a/gst/dccp/gstdccpclientsink.h b/gst/dccp/gstdccpclientsink.h new file mode 100644 index 000000000..6af03b5f9 --- /dev/null +++ b/gst/dccp/gstdccpclientsink.h @@ -0,0 +1,81 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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. + */ + +#ifndef __GST_DCCP_CLIENT_SINK_H__ +#define __GST_DCCP_CLIENT_SINK_H__ + +#include <gst/gst.h> +#include <gst/base/gstbasesink.h> +#include <gst/base/gstadapter.h> + +G_BEGIN_DECLS + +#include <netdb.h> /* sockaddr_in */ +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> /* sockaddr_in */ +#include <unistd.h> + +#define GST_TYPE_DCCP_CLIENT_SINK \ + (gst_dccp_client_sink_get_type()) +#define GST_DCCP_CLIENT_SINK(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_DCCP_CLIENT_SINK,GstDCCPClientSink)) +#define GST_DCCP_CLIENT_SINK_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_DCCP_CLIENT_SINK,GstDCCPClientSinkClass)) +#define GST_IS_DCCP_CLIENT_SINK(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_DCCP_CLIENT_SINK)) +#define GST_IS_DCCP_CLIENT_SINK_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_DCCP_CLIENT_SINK)) + +typedef struct _GstDCCPClientSink GstDCCPClientSink; +typedef struct _GstDCCPClientSinkClass GstDCCPClientSinkClass; + + +struct _GstDCCPClientSink +{ + GstBaseSink element; + + /* server information */ + int port; + gchar *host; + struct sockaddr_in server_sin; + + /* socket */ + int sock_fd; + gboolean closed; + + int pksize; + + GstCaps *caps; + uint8_t ccid; +}; + +struct _GstDCCPClientSinkClass +{ + GstBaseSinkClass parent_class; + + /* signals */ + void (*connected) (GstElement *sink, gint fd); +}; + +GType gst_dccp_client_sink_get_type (void); + +G_END_DECLS + +#endif /* __GST_DCCP_CLIENT_SRC_H__ */ diff --git a/gst/dccp/gstdccpclientsrc.c b/gst/dccp/gstdccpclientsrc.c new file mode 100644 index 000000000..bd42bf66e --- /dev/null +++ b/gst/dccp/gstdccpclientsrc.c @@ -0,0 +1,398 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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-dccpclientsrc + * @see_also: dccpclientsink, dccpserversink, dccpserversrc + * + * This element connect to a DCCP server and send data to it. + * <ulink url="http://www.linuxfoundation.org/en/Net:DCCP">DCCP</ulink> (Datagram + * Congestion Control Protocol) is a Transport Layer protocol like + * TCP and UDP. + * + * <refsect2> + * <title>Example pipeline</title> + * <para> + * |[ + * gst-launch -v dccpclientsrc host=localhost port=9011 ccid=2 ! decodebin ! alsasink + * ]| Client + * |[ + * gst-launch -v filesrc location=music.mp3 ! mp3parse ! dccpserversink port=9011 ccid=2 + * ]| Server + * + * This example pipeline will send a MP3 stream to the client using DCCP. + * The client will decode the MP3 and play it. Run the server pipeline + * first than the client pipeline. If you want, you can run more than one dccpclientsrc + * to connect to the same server (see wait-connections property at dccpserversink). + * </para> + * </refsect2> + */ + + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "gstdccpclientsrc.h" +#include "gstdccp.h" +#include <string.h> /* memset */ +#include <unistd.h> +#include <arpa/inet.h> +#include <fcntl.h> + +#define DCCP_DEFAULT_CAPS NULL + +/* signals */ +enum +{ + SIGNAL_CONNECTED, + LAST_SIGNAL +}; + +/* properties */ +enum +{ + PROP_0, + PROP_PORT, + PROP_HOST, + PROP_SOCK_FD, + PROP_CLOSED, + PROP_CCID, + PROP_CAPS +}; + +static gboolean gst_dccp_client_src_stop (GstBaseSrc * bsrc); + +GST_DEBUG_CATEGORY_STATIC (dccpclientsrc_debug); + +static const GstElementDetails gst_dccp_client_src_details = +GST_ELEMENT_DETAILS ("DCCP client source", + "Source/Network", + "Receive data as a client over the network via DCCP", + "E-Phone Team at Federal University of Campina Grande <leandroal@gmail.com>"); + +static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src", + GST_PAD_SRC, + GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY); + +GST_BOILERPLATE (GstDCCPClientSrc, gst_dccp_client_src, GstPushSrc, + GST_TYPE_PUSH_SRC); + +static guint gst_dccp_client_src_signals[LAST_SIGNAL] = { 0 }; + +static GstFlowReturn +gst_dccp_client_src_create (GstPushSrc * psrc, GstBuffer ** outbuf) +{ + GstDCCPClientSrc *src; + GstFlowReturn ret = GST_FLOW_OK; + + src = GST_DCCP_CLIENT_SRC (psrc); + + GST_LOG_OBJECT (src, "reading a buffer"); + ret = gst_dccp_read_buffer (GST_ELEMENT (src), src->sock_fd, outbuf); + + if (ret == GST_FLOW_OK) { + GST_LOG_OBJECT (src, + "Returning buffer from _get of size %d, ts %" + GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT + ", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT, + GST_BUFFER_SIZE (*outbuf), + GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (*outbuf)), + GST_TIME_ARGS (GST_BUFFER_DURATION (*outbuf)), + GST_BUFFER_OFFSET (*outbuf), GST_BUFFER_OFFSET_END (*outbuf)); + + if (!gst_caps_is_equal (src->caps, GST_CAPS_ANY)) { + gst_buffer_set_caps (*outbuf, src->caps); + } + } + + return ret; +} + +/* + * Set the value of a property for the client src. + */ +static void +gst_dccp_client_src_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstDCCPClientSrc *src = GST_DCCP_CLIENT_SRC (object); + + switch (prop_id) { + case PROP_PORT: + src->port = g_value_get_int (value); + break; + case PROP_HOST: + if (!g_value_get_string (value)) { + g_warning ("host property cannot be NULL"); + break; + } + g_free (src->host); + src->host = g_strdup (g_value_get_string (value)); + break; + case PROP_SOCK_FD: + src->sock_fd = g_value_get_int (value); + break; + case PROP_CLOSED: + src->closed = g_value_get_boolean (value); + break; + case PROP_CCID: + src->ccid = g_value_get_int (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 = src->caps; + src->caps = new_caps; + if (old_caps) + gst_caps_unref (old_caps); + gst_pad_set_caps (GST_BASE_SRC (src)->srcpad, new_caps); + break; + } + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +/* + * Get a given property value for the client src. + */ +static void +gst_dccp_client_src_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstDCCPClientSrc *src = GST_DCCP_CLIENT_SRC (object); + + switch (prop_id) { + case PROP_PORT: + g_value_set_int (value, src->port); + break; + case PROP_HOST: + g_value_set_string (value, src->host); + break; + case PROP_SOCK_FD: + g_value_set_int (value, src->sock_fd); + break; + case PROP_CLOSED: + g_value_set_boolean (value, src->closed); + break; + case PROP_CCID: + g_value_set_int (value, src->ccid); + break; + case PROP_CAPS: + gst_value_set_caps (value, src->caps); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +static gboolean +gst_dccp_client_src_start (GstBaseSrc * bsrc) +{ + GstDCCPClientSrc *src = GST_DCCP_CLIENT_SRC (bsrc); + + if (src->sock_fd == DCCP_DEFAULT_SOCK_FD) { + gchar *ip = NULL; + + /* look up name if we need to */ + if (!(ip = gst_dccp_host_to_ip (GST_ELEMENT (src), src->host))) { + GST_ERROR_OBJECT (src, "cannot resolve hostname"); + gst_dccp_client_src_stop (GST_BASE_SRC (src)); + return FALSE; + } + + /* name the server socket */ + memset (&src->server_sin, 0, sizeof (src->server_sin)); + src->server_sin.sin_family = AF_INET; /* network socket */ + src->server_sin.sin_port = htons (src->port); /* on port */ + src->server_sin.sin_addr.s_addr = inet_addr (ip); /* on host ip */ + g_free (ip); + + /* create socket */ + if ((src->sock_fd = gst_dccp_create_new_socket (GST_ELEMENT (src))) < 0) { + return FALSE; + } + + if (!gst_dccp_set_ccid (GST_ELEMENT (src), src->sock_fd, src->ccid)) { + gst_dccp_client_src_stop (GST_BASE_SRC (src)); + return FALSE; + } + + if (!gst_dccp_connect_to_server (GST_ELEMENT (src), src->server_sin, + src->sock_fd)) { + gst_dccp_client_src_stop (GST_BASE_SRC (src)); + return FALSE; + } + + /* the socket is connected */ + g_signal_emit (src, gst_dccp_client_src_signals[SIGNAL_CONNECTED], 0, + src->sock_fd); + } + + return TRUE; +} + +static void +gst_dccp_client_src_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 (&srctemplate)); + + gst_element_class_set_details (element_class, &gst_dccp_client_src_details); +} + +static void +gst_dccp_client_src_init (GstDCCPClientSrc * this, + GstDCCPClientSrcClass * g_class) +{ + this->port = DCCP_DEFAULT_PORT; + this->host = g_strdup (DCCP_DEFAULT_HOST); + this->sock_fd = DCCP_DEFAULT_SOCK_FD; + this->closed = DCCP_DEFAULT_CLOSED; + this->ccid = DCCP_DEFAULT_CCID; + this->caps = NULL; + + gst_base_src_set_format (GST_BASE_SRC (this), GST_FORMAT_TIME); + + /* Checking if the version of the gstreamer is bigger that 0.10.15 */ +#if ((GST_VERSION_MAJOR > 0) || \ + (GST_VERSION_MAJOR == 0 && GST_VERSION_MINOR > 10) || \ + (GST_VERSION_MAJOR == 0 && GST_VERSION_MINOR == 10 && GST_VERSION_MICRO >= 15)) + gst_base_src_set_do_timestamp (GST_BASE_SRC (this), TRUE); +#endif + + /* FIXME is this correct? */ + gst_base_src_set_live (GST_BASE_SRC (this), TRUE); +} + +static void +gst_dccp_client_src_finalize (GObject * gobject) +{ + GstDCCPClientSrc *this = GST_DCCP_CLIENT_SRC (gobject); + + if (this->caps) { + gst_caps_unref (this->caps); + this->caps = NULL; + } + + g_free (this->host); + + G_OBJECT_CLASS (parent_class)->finalize (gobject); +} + +static gboolean +gst_dccp_client_src_stop (GstBaseSrc * bsrc) +{ + GstDCCPClientSrc *src; + + src = GST_DCCP_CLIENT_SRC (bsrc); + + if (src->sock_fd != -1 && src->closed) { + GST_DEBUG_OBJECT (src, "closing socket"); + close (src->sock_fd); + src->sock_fd = -1; + } + + return TRUE; +} + +/* + * Define the gst class, callbacks, etc. + */ +static void +gst_dccp_client_src_class_init (GstDCCPClientSrcClass * klass) +{ + GObjectClass *gobject_class; + GstBaseSrcClass *gstbasesrc_class; + GstPushSrcClass *gstpush_src_class; + + gobject_class = (GObjectClass *) klass; + gstbasesrc_class = (GstBaseSrcClass *) klass; + gstpush_src_class = (GstPushSrcClass *) klass; + + gobject_class->set_property = gst_dccp_client_src_set_property; + gobject_class->get_property = gst_dccp_client_src_get_property; + + gobject_class->finalize = gst_dccp_client_src_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, + DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_HOST, + g_param_spec_string ("host", "Host", + "The host IP address to receive packets from", DCCP_DEFAULT_HOST, + G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_SOCK_FD, + g_param_spec_int ("sockfd", "Socket fd", + "The socket file descriptor", -1, G_MAXINT, DCCP_DEFAULT_SOCK_FD, + G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CLOSED, + g_param_spec_boolean ("close-socket", "Close socket", + "Close socket at the end of stream", DCCP_DEFAULT_CLOSED, + 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_CCID, + g_param_spec_int ("ccid", "CCID", + "The Congestion Control IDentified to be used", 2, G_MAXINT, + DCCP_DEFAULT_CCID, G_PARAM_READWRITE)); + + /* signals */ + /** + * GstDccpClientSrc::connected: + * @src: the gstdccpclientsrc instance + * @fd: the connected socket fd + * + * Reports that the element has connected, giving the fd of the socket + */ + gst_dccp_client_src_signals[SIGNAL_CONNECTED] = + g_signal_new ("connected", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST, + G_STRUCT_OFFSET (GstDCCPClientSrcClass, connected), NULL, NULL, + gst_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT); + + gstbasesrc_class->start = gst_dccp_client_src_start; + gstbasesrc_class->stop = gst_dccp_client_src_stop; + + gstpush_src_class->create = gst_dccp_client_src_create; + + GST_DEBUG_CATEGORY_INIT (dccpclientsrc_debug, "dccpclientsrc", 0, + "DCCP Client Source"); +} diff --git a/gst/dccp/gstdccpclientsrc.h b/gst/dccp/gstdccpclientsrc.h new file mode 100644 index 000000000..9be42b2df --- /dev/null +++ b/gst/dccp/gstdccpclientsrc.h @@ -0,0 +1,77 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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. + */ + +#ifndef __GST_DCCP_CLIENT_SRC_H__ +#define __GST_DCCP_CLIENT_SRC_H__ + +#include <gst/gst.h> +#include <gst/base/gstpushsrc.h> +#include <gst/base/gstbasesrc.h> + +G_BEGIN_DECLS + +#include <netdb.h> /* sockaddr_in */ +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> /* sockaddr_in */ +#include <unistd.h> + + +#define GST_TYPE_DCCP_CLIENT_SRC \ + (gst_dccp_client_src_get_type()) +#define GST_DCCP_CLIENT_SRC(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_DCCP_CLIENT_SRC,GstDCCPClientSrc)) +#define GST_DCCP_CLIENT_SRC_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_DCCP_CLIENT_SRC,GstDCCPClientSrcClass)) +#define GST_IS_DCCP_CLIENT_SRC(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_DCCP_CLIENT_SRC)) +#define GST_IS_DCCP_CLIENT_SRC_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_DCCP_CLIENT_SRC)) + +typedef struct _GstDCCPClientSrc GstDCCPClientSrc; +typedef struct _GstDCCPClientSrcClass GstDCCPClientSrcClass; + +struct _GstDCCPClientSrc { + GstPushSrc element; + + /* server information */ + int port; + gchar *host; + struct sockaddr_in server_sin; + + /* socket */ + int sock_fd; + gboolean closed; + + GstCaps *caps; + uint8_t ccid; +}; + +struct _GstDCCPClientSrcClass { + GstPushSrcClass parent_class; + + /* signals */ + void (*connected) (GstElement *src, gint fd); +}; + +GType gst_dccp_client_src_get_type (void); + +G_END_DECLS + +#endif /* __GST_DCCP_CLIENT_SRC_H__ */ diff --git a/gst/dccp/gstdccpplugin.c b/gst/dccp/gstdccpplugin.c new file mode 100644 index 000000000..05c48cd57 --- /dev/null +++ b/gst/dccp/gstdccpplugin.c @@ -0,0 +1,60 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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. + */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "gstdccpclientsrc.h" +#include "gstdccpserversink.h" +#include "gstdccpclientsink.h" +#include "gstdccpserversrc.h" + +GST_DEBUG_CATEGORY (dccp_debug); + +static gboolean +plugin_init (GstPlugin * plugin) +{ + if (!gst_element_register (plugin, "dccpclientsrc", GST_RANK_NONE, + GST_TYPE_DCCP_CLIENT_SRC)) + return FALSE; + + if (!gst_element_register (plugin, "dccpserversink", GST_RANK_NONE, + GST_TYPE_DCCP_SERVER_SINK)) + return FALSE; + + if (!gst_element_register (plugin, "dccpclientsink", GST_RANK_NONE, + GST_TYPE_DCCP_CLIENT_SINK)) + return FALSE; + + if (!gst_element_register (plugin, "dccpserversrc", GST_RANK_NONE, + GST_TYPE_DCCP_SERVER_SRC)) + return FALSE; + + GST_DEBUG_CATEGORY_INIT (dccp_debug, "dccp", 0, "DCCP calls"); + + return TRUE; +} + +GST_PLUGIN_DEFINE (GST_VERSION_MAJOR, + GST_VERSION_MINOR, + "dccp", + "transfer data over the network via DCCP.", + plugin_init, VERSION, GST_LICENSE, "DCCP", + "http://garage.maemo.org/projects/ephone") diff --git a/gst/dccp/gstdccpserversink.c b/gst/dccp/gstdccpserversink.c new file mode 100644 index 000000000..72ffa16c9 --- /dev/null +++ b/gst/dccp/gstdccpserversink.c @@ -0,0 +1,435 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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-dccpserversink + * @see_also: dccpclientsink, dccpclientsrc, dccpserversrc + * + * This element wait for connections from clients and send data to them. + * <ulink url="http://www.linuxfoundation.org/en/Net:DCCP">DCCP</ulink> (Datagram + * Congestion Control Protocol) is a Transport Layer protocol like + * TCP and UDP. + * + * <refsect2> + * <title>Example pipeline</title> + * <para> + * |[ + * gst-launch -v dccpclientsrc host=localhost port=9011 ccid=2 ! decodebin ! alsasink + * ]| Client + * |[ + * gst-launch -v filesrc location=music.mp3 ! mp3parse ! dccpserversink port=9011 ccid=2 + * ]| Server + * + * This example pipeline will send a MP3 stream to the client using DCCP. + * The client will decode the MP3 and play it. Run the server pipeline + * first than the client pipeline. If you want, you can run more than one dccpclientsrc + * to connect to the same server (see wait-connections property at dccpserversink). + * </para> + * </refsect2> + */ + + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "gstdccpserversink.h" +#include "gstdccp.h" +#include <string.h> /* memset */ +#include <unistd.h> +#include <arpa/inet.h> +#include <fcntl.h> + +/* signals */ +enum +{ + SIGNAL_CONNECTED, + LAST_SIGNAL +}; + +/* properties */ +enum +{ + PROP_0, + PROP_PORT, + PROP_CLIENT_SOCK_FD, + PROP_CCID, + PROP_CLOSED, + PROP_WAIT_CONNECTIONS +}; + +pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; +static gboolean gst_dccp_server_sink_stop (GstBaseSink * bsink); + +GST_DEBUG_CATEGORY_STATIC (dccpserversink_debug); + +static const GstElementDetails gst_dccp_server_sink_details = +GST_ELEMENT_DETAILS ("DCCP server sink", + "Sink/Network", + "Send data as a server over the network via DCCP", + "E-Phone Team at Federal University of Campina Grande <leandroal@gmail.com>"); + + +static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink", + GST_PAD_SINK, + GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY); + +GST_BOILERPLATE (GstDCCPServerSink, gst_dccp_server_sink, GstBaseSink, + GST_TYPE_BASE_SINK); + +static guint gst_dccp_server_sink_signals[LAST_SIGNAL] = { 0 }; + +static Client * +gst_dccp_server_create_client (GstElement * element, int socket) +{ + Client *client = (Client *) g_malloc (sizeof (Client)); + client->socket = socket; + client->pksize = gst_dccp_get_max_packet_size (element, client->socket); + client->flow_status = GST_FLOW_OK; + + /* the socket is connected */ + g_signal_emit (element, gst_dccp_server_sink_signals[SIGNAL_CONNECTED], 0, + socket); + + return client; +} + +static void * +gst_dccp_server_accept_new_clients (void *arg) +{ + GstDCCPServerSink *sink = (GstDCCPServerSink *) arg; + int newsockfd; + Client *client; + + while (1) { + newsockfd = + gst_dccp_server_wait_connections (GST_ELEMENT (sink), sink->sock_fd); + + client = gst_dccp_server_create_client (GST_ELEMENT (sink), newsockfd); + + pthread_mutex_lock (&lock); + sink->clients = g_list_append (sink->clients, client); + pthread_mutex_unlock (&lock); + } +} + +static void * +gst_dccp_server_send_buffer (void *arg) +{ + Client *client = (Client *) arg; + GstDCCPServerSink *sink = client->server; + GstBuffer *buf = client->buf; + int client_sock_fd = client->socket; + int pksize = client->pksize; + + if (gst_dccp_send_buffer (GST_ELEMENT (sink), buf, client_sock_fd, + pksize) == GST_FLOW_ERROR) { + client->flow_status = GST_FLOW_ERROR; + } + return NULL; +} + +/* Remove clients with problems to send */ +static void * +gst_dccp_server_delete_dead_clients (void *arg) +{ + GstDCCPServerSink *sink = (GstDCCPServerSink *) arg; + int i; + GList *tmp = NULL; + + pthread_mutex_lock (&lock); + for (i = 0; i < g_list_length (sink->clients); i++) { + Client *client = (Client *) g_list_nth_data (sink->clients, i); + if (client->flow_status == GST_FLOW_OK) { + tmp = g_list_append (tmp, client); + } else { + close (client->socket); + g_free (client); + } + } + g_list_free (sink->clients); + sink->clients = tmp; + pthread_mutex_unlock (&lock); + return 0; +} + +static void +gst_dccp_server_sink_init (GstDCCPServerSink * this, + GstDCCPServerSinkClass * g_class) +{ + this->port = DCCP_DEFAULT_PORT; + this->sock_fd = DCCP_DEFAULT_SOCK_FD; + this->client_sock_fd = DCCP_DEFAULT_CLIENT_SOCK_FD; + this->closed = DCCP_DEFAULT_CLOSED; + this->ccid = DCCP_DEFAULT_CCID; + this->wait_connections = DCCP_DEFAULT_WAIT_CONNECTIONS; + this->clients = NULL; +} + +static gboolean +gst_dccp_server_sink_start (GstBaseSink * bsink) +{ + GstDCCPServerSink *sink = GST_DCCP_SERVER_SINK (bsink); + int ret = 1; + Client *client; + + /* create socket */ + if ((sink->sock_fd = gst_dccp_create_new_socket (GST_ELEMENT (sink))) < 0) { + return FALSE; + } + + /* make address reusable */ + if (setsockopt (sink->sock_fd, SOL_SOCKET, SO_REUSEADDR, + (void *) &ret, sizeof (ret)) < 0) { + GST_ELEMENT_ERROR (sink, RESOURCE, SETTINGS, (NULL), + ("Could not setsockopt: %s", g_strerror (errno))); + return FALSE; + } + + /* name the server socket */ + memset (&sink->server_sin, 0, sizeof (sink->server_sin)); + sink->server_sin.sin_family = AF_INET; /* network socket */ + sink->server_sin.sin_port = htons (sink->port); /* on port */ + sink->server_sin.sin_addr.s_addr = htonl (INADDR_ANY); /* for hosts */ + + if (!gst_dccp_bind_server_socket (GST_ELEMENT (sink), sink->sock_fd, + sink->server_sin)) { + return FALSE; + } + + if (!gst_dccp_set_ccid (GST_ELEMENT (sink), sink->sock_fd, sink->ccid)) { + return FALSE; + } + + if (!gst_dccp_listen_server_socket (GST_ELEMENT (sink), sink->sock_fd)) { + return FALSE; + } + + + if (sink->client_sock_fd == DCCP_DEFAULT_CLIENT_SOCK_FD) { + sink->client_sock_fd = + gst_dccp_server_wait_connections (GST_ELEMENT (sink), sink->sock_fd); + } + + if (sink->client_sock_fd == -1) { + return FALSE; + } + + client = + gst_dccp_server_create_client (GST_ELEMENT (sink), sink->client_sock_fd); + sink->clients = g_list_append (sink->clients, client); + + pthread_mutex_init (&lock, NULL); + + if (sink->wait_connections == TRUE) { + pthread_create (&accept_thread_id, NULL, gst_dccp_server_accept_new_clients, + sink); + pthread_detach (accept_thread_id); + } + + return TRUE; +} + +static GstFlowReturn +gst_dccp_server_sink_render (GstBaseSink * bsink, GstBuffer * buf) +{ + GstDCCPServerSink *sink = GST_DCCP_SERVER_SINK (bsink); + + pthread_t thread_id; + int i; + + pthread_mutex_lock (&lock); + + for (i = 0; i < g_list_length (sink->clients); i++) { + Client *client = (Client *) g_list_nth_data (sink->clients, i); + client->buf = buf; + client->server = sink; + + if (client->flow_status == GST_FLOW_OK) { + pthread_create (&thread_id, NULL, gst_dccp_server_send_buffer, + (void *) client); + pthread_detach (thread_id); + } else { + pthread_create (&thread_id, NULL, gst_dccp_server_delete_dead_clients, + (void *) sink); + pthread_detach (thread_id); + } + } + + pthread_mutex_unlock (&lock); + return GST_FLOW_OK; +} + +static gboolean +gst_dccp_server_sink_stop (GstBaseSink * bsink) +{ + GstDCCPServerSink *sink; + int i; + sink = GST_DCCP_SERVER_SINK (bsink); + + if (sink->wait_connections == TRUE) { + pthread_cancel (accept_thread_id); + } + + if (sink->sock_fd != -1 && sink->closed == TRUE) { + GST_DEBUG_OBJECT (sink, "closing socket"); + close (sink->sock_fd); + sink->sock_fd = -1; + } + + pthread_mutex_lock (&lock); + for (i = 0; i < g_list_length (sink->clients); i++) { + Client *client = (Client *) g_list_nth_data (sink->clients, i); + close (client->socket); + g_free (client); + } + pthread_mutex_unlock (&lock); + + return TRUE; +} + +static void +gst_dccp_server_sink_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 (&sinktemplate)); + + gst_element_class_set_details (element_class, &gst_dccp_server_sink_details); +} + +/* + * Set the value of a property for the server sink. + */ +static void +gst_dccp_server_sink_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstDCCPServerSink *sink = GST_DCCP_SERVER_SINK (object); + + switch (prop_id) { + case PROP_PORT: + sink->port = g_value_get_int (value); + break; + case PROP_CLIENT_SOCK_FD: + sink->client_sock_fd = g_value_get_int (value); + break; + case PROP_CLOSED: + sink->closed = g_value_get_boolean (value); + break; + case PROP_WAIT_CONNECTIONS: + sink->wait_connections = g_value_get_boolean (value); + break; + case PROP_CCID: + sink->ccid = g_value_get_int (value); + break; + default: + break; + } +} + +static void +gst_dccp_server_sink_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstDCCPServerSink *sink = GST_DCCP_SERVER_SINK (object); + + switch (prop_id) { + case PROP_PORT: + g_value_set_int (value, sink->port); + break; + case PROP_CLIENT_SOCK_FD: + g_value_set_int (value, sink->client_sock_fd); + break; + case PROP_CLOSED: + g_value_set_boolean (value, sink->closed); + break; + case PROP_WAIT_CONNECTIONS: + g_value_set_boolean (value, sink->wait_connections); + break; + case PROP_CCID: + g_value_set_int (value, sink->ccid); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + + +static void +gst_dccp_server_sink_class_init (GstDCCPServerSinkClass * klass) +{ + GObjectClass *gobject_class; + GstBaseSinkClass *gstbasesink_class; + + gobject_class = (GObjectClass *) klass; + gstbasesink_class = (GstBaseSinkClass *) klass; + + gobject_class->set_property = gst_dccp_server_sink_set_property; + gobject_class->get_property = gst_dccp_server_sink_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, + DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CLIENT_SOCK_FD, + g_param_spec_int ("sockfd", "Socket fd", + "The client socket file descriptor", -1, G_MAXINT, + DCCP_DEFAULT_CLIENT_SOCK_FD, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CLOSED, + g_param_spec_boolean ("close-socket", "Close", + "Close socket at end of stream", + DCCP_DEFAULT_CLOSED, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CCID, + g_param_spec_int ("ccid", "CCID", + "The Congestion Control IDentified to be used", 2, G_MAXINT, + DCCP_DEFAULT_CCID, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_WAIT_CONNECTIONS, + g_param_spec_boolean ("wait-connections", "Wait connections", + "Wait for many client connections", + DCCP_DEFAULT_WAIT_CONNECTIONS, G_PARAM_READWRITE)); + + + /* signals */ + /** + * GstDccpServerSink::connected: + * @src: the gstdccpserversink instance + * @fd: the connected socket fd + * + * Reports that the element has connected, giving the fd of the socket + */ + gst_dccp_server_sink_signals[SIGNAL_CONNECTED] = + g_signal_new ("connected", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST, + G_STRUCT_OFFSET (GstDCCPServerSinkClass, connected), NULL, NULL, + gst_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT); + + gstbasesink_class->start = gst_dccp_server_sink_start; + gstbasesink_class->stop = gst_dccp_server_sink_stop; + gstbasesink_class->render = gst_dccp_server_sink_render; + + GST_DEBUG_CATEGORY_INIT (dccpserversink_debug, "dccpserversink", 0, + "DCCP Server Sink"); +} diff --git a/gst/dccp/gstdccpserversink.h b/gst/dccp/gstdccpserversink.h new file mode 100644 index 000000000..5fead4749 --- /dev/null +++ b/gst/dccp/gstdccpserversink.h @@ -0,0 +1,96 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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. + */ + +#ifndef __GST_DCCP_SERVER_SINK_H__ +#define __GST_DCCP_SERVER_SINK_H__ + +#include <gst/gst.h> +#include <gst/base/gstbasesink.h> +#include <gst/base/gstadapter.h> + +G_BEGIN_DECLS + +#include <netdb.h> /* sockaddr_in */ +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> /* sockaddr_in */ +#include <unistd.h> +#include <pthread.h> + +#define GST_TYPE_DCCP_SERVER_SINK \ + (gst_dccp_server_sink_get_type()) +#define GST_DCCP_SERVER_SINK(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_DCCP_SERVER_SINK,GstDCCPServerSink)) +#define GST_DCCP_SERVER_SINK_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_DCCP_SERVER_SINK,GstDCCPServerSinkClass)) +#define GST_IS_DCCP_SERVER_SINK(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_DCCP_SERVER_SINK)) +#define GST_IS_DCCP_SERVER_SINK_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_DCCP_SERVER_SINK)) + +typedef struct _GstDCCPServerSink GstDCCPServerSink; +typedef struct _GstDCCPServerSinkClass GstDCCPServerSinkClass; + +typedef struct _Client Client; + +struct _Client +{ + GstDCCPServerSink *server; + GstBuffer * buf; + int socket; + int pksize; + GstFlowReturn flow_status; +}; + +pthread_t accept_thread_id; + +struct _GstDCCPServerSink +{ + GstBaseSink element; + + /* server information */ + int port; + struct sockaddr_in server_sin; + + /* socket */ + int sock_fd; + + /* multiple clients */ + GList *clients; + + /* properties */ + int client_sock_fd; + uint8_t ccid; + gboolean wait_connections; + gboolean closed; +}; + +struct _GstDCCPServerSinkClass +{ + GstBaseSinkClass parent_class; + + /* signals */ + void (*connected) (GstElement *sink, gint fd); +}; + +GType gst_dccp_server_sink_get_type (void); + +G_END_DECLS + +#endif /* __GST_DCCP_SERVER_SINK_H__ */ diff --git a/gst/dccp/gstdccpserversrc.c b/gst/dccp/gstdccpserversrc.c new file mode 100644 index 000000000..d3d46b0da --- /dev/null +++ b/gst/dccp/gstdccpserversrc.c @@ -0,0 +1,403 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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-dccpserversrc + * @see_also: dccpclientsink, dccpclientsrc, dccpserversink + * + * This element wait for connection from a client and receive data. + * <ulink url="http://www.linuxfoundation.org/en/Net:DCCP">DCCP</ulink> (Datagram + * Congestion Control Protocol) is a Transport Layer protocol like + * TCP and UDP. + * + * <refsect2> + * <title>Example pipeline</title> + * <para> + * |[ + * gst-launch -v filesrc location=music.mp3 ! mp3parse ! dccpclientsink host=localhost port=9011 ccid=2 + * ]| Client + * |[ + * gst-launch -v dccpserversrc port=9011 ccid=2 ! decodebin ! alsasink + * ]| Server + * + * This example pipeline will send a MP3 stream to the server using DCCP. + * The server will decode the MP3 and play it. + * Run the server pipeline first than the client pipeline. + * </para> + * </refsect2> + */ + + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include "gstdccpserversrc.h" +#include "gstdccp.h" +#include <string.h> /* memset */ +#include <unistd.h> +#include <arpa/inet.h> +#include <fcntl.h> + +#define DCCP_DEFAULT_CAPS NULL +#define DCCP_DEFAULT_LISTEN_HOST NULL /* listen on all interfaces */ + +/* signals */ +enum +{ + SIGNAL_CONNECTED, + LAST_SIGNAL +}; + +/* properties */ +enum +{ + PROP_0, + PROP_PORT, + PROP_HOST, + PROP_CLIENT_SOCK_FD, + PROP_CLOSED, + PROP_CCID, + PROP_CAPS +}; + +static gboolean gst_dccp_server_src_stop (GstBaseSrc * bsrc); + +GST_DEBUG_CATEGORY_STATIC (dccpserversrc_debug); + +static const GstElementDetails gst_dccp_server_src_details = +GST_ELEMENT_DETAILS ("DCCP server source", + "Source/Network", + "Receive data as a server over the network via DCCP", + "E-Phone Team at Federal University of Campina Grande <leandroal@gmail.com>"); + +static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src", + GST_PAD_SRC, + GST_PAD_ALWAYS, + GST_STATIC_CAPS_ANY); + +GST_BOILERPLATE (GstDCCPServerSrc, gst_dccp_server_src, GstPushSrc, + GST_TYPE_PUSH_SRC); + +static guint gst_dccp_server_src_signals[LAST_SIGNAL] = { 0 }; + +static GstFlowReturn +gst_dccp_server_src_create (GstPushSrc * psrc, GstBuffer ** outbuf) +{ + GstDCCPServerSrc *src; + GstFlowReturn ret = GST_FLOW_OK; + + src = GST_DCCP_SERVER_SRC (psrc); + + GST_LOG_OBJECT (src, "reading a buffer"); + + ret = gst_dccp_read_buffer (GST_ELEMENT (src), src->client_sock_fd, outbuf); + + if (ret == GST_FLOW_OK) { + GST_LOG_OBJECT (src, + "Returning buffer from _get of size %d, ts %" + GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT + ", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT, + GST_BUFFER_SIZE (*outbuf), + GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (*outbuf)), + GST_TIME_ARGS (GST_BUFFER_DURATION (*outbuf)), + GST_BUFFER_OFFSET (*outbuf), GST_BUFFER_OFFSET_END (*outbuf)); + + if (!gst_caps_is_equal (src->caps, GST_CAPS_ANY)) { + gst_buffer_set_caps (*outbuf, src->caps); + } + } + + return ret; +} + +/* + * Set the value of a property for the server src. + */ +static void +gst_dccp_server_src_set_property (GObject * object, guint prop_id, + const GValue * value, GParamSpec * pspec) +{ + GstDCCPServerSrc *src = GST_DCCP_SERVER_SRC (object); + + switch (prop_id) { + case PROP_HOST: + if (!g_value_get_string (value)) { + g_warning ("host property cannot be NULL"); + break; + } + g_free (src->host); + src->host = g_strdup (g_value_get_string (value)); + break; + case PROP_PORT: + src->port = g_value_get_int (value); + break; + case PROP_CLIENT_SOCK_FD: + src->client_sock_fd = g_value_get_int (value); + break; + case PROP_CLOSED: + src->closed = g_value_get_boolean (value); + break; + case PROP_CCID: + src->ccid = g_value_get_int (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 = src->caps; + src->caps = new_caps; + if (old_caps) { + gst_caps_unref (old_caps); + } + gst_pad_set_caps (GST_BASE_SRC (src)->srcpad, new_caps); + break; + } + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + +/* + * Get a given property value for the server src. + */ +static void +gst_dccp_server_src_get_property (GObject * object, guint prop_id, + GValue * value, GParamSpec * pspec) +{ + GstDCCPServerSrc *src = GST_DCCP_SERVER_SRC (object); + + switch (prop_id) { + case PROP_HOST: + g_value_set_string (value, src->host); + break; + case PROP_PORT: + g_value_set_int (value, src->port); + break; + case PROP_CLIENT_SOCK_FD: + g_value_set_int (value, src->client_sock_fd); + break; + case PROP_CLOSED: + g_value_set_boolean (value, src->closed); + break; + case PROP_CAPS: + gst_value_set_caps (value, src->caps); + break; + case PROP_CCID: + g_value_set_int (value, src->ccid); + break; + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); + break; + } +} + + +static gboolean +gst_dccp_server_src_start (GstBaseSrc * bsrc) +{ + GstDCCPServerSrc *src = GST_DCCP_SERVER_SRC (bsrc); + + if (src->client_sock_fd == DCCP_DEFAULT_CLIENT_SOCK_FD) { + /* create socket */ + if ((src->sock_fd = gst_dccp_create_new_socket (GST_ELEMENT (src))) < 0) { + return FALSE; + } + + gst_dccp_make_address_reusable (GST_ELEMENT (src), src->sock_fd); + + /* name the server socket */ + memset (&src->server_sin, 0, sizeof (src->server_sin)); + src->server_sin.sin_family = AF_INET; /* network socket */ + src->server_sin.sin_port = htons (src->port); /* on port */ + src->server_sin.sin_addr.s_addr = htonl (INADDR_ANY); /* for hosts */ + + + if (!gst_dccp_bind_server_socket (GST_ELEMENT (src), src->sock_fd, + src->server_sin)) { + return FALSE; + } + + if (!gst_dccp_set_ccid (GST_ELEMENT (src), src->sock_fd, src->ccid)) { + return FALSE; + } + + if (!gst_dccp_listen_server_socket (GST_ELEMENT (src), src->sock_fd)) { + return FALSE; + } + + src->client_sock_fd = gst_dccp_server_wait_connections (GST_ELEMENT (src), + src->sock_fd); + if (src->client_sock_fd == -1) { + return FALSE; + } + + /* the socket is connected */ + g_signal_emit (src, gst_dccp_server_src_signals[SIGNAL_CONNECTED], 0, + src->client_sock_fd); + } + + return TRUE; +} + + +static void +gst_dccp_server_src_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 (&srctemplate)); + + gst_element_class_set_details (element_class, &gst_dccp_server_src_details); +} + + +static void +gst_dccp_server_src_init (GstDCCPServerSrc * this, + GstDCCPServerSrcClass * g_class) +{ + this->port = DCCP_DEFAULT_PORT; + this->host = g_strdup (DCCP_DEFAULT_HOST); + this->sock_fd = DCCP_DEFAULT_SOCK_FD; + this->client_sock_fd = DCCP_DEFAULT_CLIENT_SOCK_FD; + this->closed = DCCP_DEFAULT_CLOSED; + this->ccid = DCCP_DEFAULT_CCID; + this->caps = DCCP_DEFAULT_CAPS; + + gst_base_src_set_format (GST_BASE_SRC (this), GST_FORMAT_TIME); + + /* Checking if the version of the gstreamer is bigger that 0.10.15 */ +#if ((GST_VERSION_MAJOR > 0) || \ + (GST_VERSION_MAJOR == 0 && GST_VERSION_MINOR > 10) || \ + (GST_VERSION_MAJOR == 0 && GST_VERSION_MINOR == 10 && GST_VERSION_MICRO >= 15)) + gst_base_src_set_do_timestamp (GST_BASE_SRC (this), TRUE); +#endif + + /* FIXME is this correct? */ + gst_base_src_set_live (GST_BASE_SRC (this), TRUE); +} + + +static void +gst_dccp_server_src_finalize (GObject * gobject) +{ + GstDCCPServerSrc *this = GST_DCCP_SERVER_SRC (gobject); + + if (this->caps) { + gst_caps_unref (this->caps); + this->caps = NULL; + } + + g_free (this->host); + + G_OBJECT_CLASS (parent_class)->finalize (gobject); +} + + +static gboolean +gst_dccp_server_src_stop (GstBaseSrc * bsrc) +{ + GstDCCPServerSrc *src; + + src = GST_DCCP_SERVER_SRC (bsrc); + + if (src->sock_fd != -1 && src->closed == TRUE) { + GST_DEBUG_OBJECT (src, "closing socket"); + close (src->sock_fd); + src->sock_fd = -1; + } + + return TRUE; +} + +static void +gst_dccp_server_src_class_init (GstDCCPServerSrcClass * klass) +{ + GObjectClass *gobject_class; + GstBaseSrcClass *gstbasesrc_class; + GstPushSrcClass *gstpush_src_class; + + gobject_class = (GObjectClass *) klass; + gstbasesrc_class = (GstBaseSrcClass *) klass; + gstpush_src_class = (GstPushSrcClass *) klass; + + gobject_class->set_property = gst_dccp_server_src_set_property; + gobject_class->get_property = gst_dccp_server_src_get_property; + + gobject_class->finalize = gst_dccp_server_src_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, + DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_HOST, + g_param_spec_string ("host", "Host", "The hostname to listen as", + DCCP_DEFAULT_LISTEN_HOST, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CLIENT_SOCK_FD, + g_param_spec_int ("sockfd", "Socket fd", + "The client socket file descriptor", -1, G_MAXINT, + DCCP_DEFAULT_CLIENT_SOCK_FD, G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CLOSED, + g_param_spec_boolean ("close-socket", "Close socket", + "Close socket at the end of stream", DCCP_DEFAULT_CLOSED, + G_PARAM_READWRITE)); + + g_object_class_install_property (gobject_class, PROP_CCID, + g_param_spec_int ("ccid", "CCID", + "The Congestion Control IDentified to be used", 2, G_MAXINT, + DCCP_DEFAULT_CCID, 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)); + + /* signals */ + /** + * GstDccpServerSrc::connected: + * @src: the gstdccpserversrc instance + * @fd: the connected socket fd + * + * Reports that the element has connected, giving the fd of the socket + */ + gst_dccp_server_src_signals[SIGNAL_CONNECTED] = + g_signal_new ("connected", G_TYPE_FROM_CLASS (klass), G_SIGNAL_RUN_FIRST, + G_STRUCT_OFFSET (GstDCCPServerSrcClass, connected), NULL, NULL, + gst_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT); + + gstbasesrc_class->start = gst_dccp_server_src_start; + gstbasesrc_class->stop = gst_dccp_server_src_stop; + + gstpush_src_class->create = gst_dccp_server_src_create; + + GST_DEBUG_CATEGORY_INIT (dccpserversrc_debug, "dccpserversrc", 0, + "DCCP Server Source"); +} diff --git a/gst/dccp/gstdccpserversrc.h b/gst/dccp/gstdccpserversrc.h new file mode 100644 index 000000000..87ef19a54 --- /dev/null +++ b/gst/dccp/gstdccpserversrc.h @@ -0,0 +1,94 @@ +/* GStreamer + * Copyright (C) <2007> Leandro Melo de Sales <leandroal@gmail.com> + * + * 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. + */ + +#ifndef __GST_DCCP_SERVER_SRC_H__ +#define __GST_DCCP_SERVER_SRC_H__ + +#include <gst/gst.h> +#include <gst/base/gstpushsrc.h> +#include <gst/base/gstbasesrc.h> + +G_BEGIN_DECLS + +#include <netdb.h> /* sockaddr_in */ +#include <sys/types.h> +#include <sys/socket.h> +#include <netinet/in.h> /* sockaddr_in */ +#include <unistd.h> + + +#define GST_TYPE_DCCP_SERVER_SRC \ + (gst_dccp_server_src_get_type()) +#define GST_DCCP_SERVER_SRC(obj) \ + (G_TYPE_CHECK_INSTANCE_CAST((obj),GST_TYPE_DCCP_SERVER_SRC,GstDCCPServerSrc)) +#define GST_DCCP_SERVER_SRC_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_CAST((klass),GST_TYPE_DCCP_SERVER_SRC,GstDCCPServerSrcClass)) +#define GST_IS_DCCP_SERVER_SRC(obj) \ + (G_TYPE_CHECK_INSTANCE_TYPE((obj),GST_TYPE_DCCP_SERVER_SRC)) +#define GST_IS_DCCP_SERVER_SRC_CLASS(klass) \ + (G_TYPE_CHECK_CLASS_TYPE((klass),GST_TYPE_DCCP_SERVER_SRC)) + +typedef struct _GstDCCPServerSrc GstDCCPServerSrc; +typedef struct _GstDCCPServerSrcClass GstDCCPServerSrcClass; + +struct _GstDCCPServerSrc +{ + GstPushSrc element; + + /* server information */ + int port; + gchar *host; + struct sockaddr_in server_sin; + + /* socket */ + int sock_fd; + gboolean closed; + + GstCaps *caps; + uint8_t ccid; + + /* single client */ + int client_sock_fd; +}; + +struct _GstDCCPServerSrcClass +{ + GstPushSrcClass parent_class; + + /* signals */ + void (*connected) (GstElement *src, gint fd); +}; + +GType gst_dccp_server_src_get_type (void); + +G_END_DECLS + +#endif /* __GST_DCCP_SERVER_SRC_H__ */ + + + + + + + + + + + + |