mirrors/gstreamer
mirrors/gstreamer
1
1
Fork 0
mirror of https://gitlab.freedesktop.org/gstreamer/gstreamer.git synced 2024-11-27 12:11:13 +00:00
Code Issues 1 Projects Releases Wiki Activity

gst/dccp/: CodeStyle improvements and minor fixes to the DCCP plugin.

Browse source 
Original commit message from CVS:
patch by: Leandro Melo de Sales <leandroal@gmail.com>
* gst/dccp/gstdccp.c:
* gst/dccp/gstdccp.h:
* gst/dccp/gstdccpclientsink.c:
* gst/dccp/gstdccpclientsink.h:
* gst/dccp/gstdccpclientsrc.c:
* gst/dccp/gstdccpserversink.c:
* gst/dccp/gstdccpserversrc.c:
* gst/dccp/gstdccpserversrc.h:
CodeStyle improvements and minor fixes to the DCCP plugin.
Fixes #553823.
This commit is contained in:
Leandro Melo de Sales 2008-11-20 16:01:27 +00:00 committed by Stefan Kost
parent 0803391ab1
commit df25221378
9 changed files with 266 additions and 147 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
ChangeLog
View file

@ -1,3 +1,18 @@
2008-11-20 Stefan Kost <ensonic@users.sf.net>
patch by: Leandro Melo de Sales <leandroal@gmail.com>
* gst/dccp/gstdccp.c:
* gst/dccp/gstdccp.h:
* gst/dccp/gstdccpclientsink.c:
* gst/dccp/gstdccpclientsink.h:
* gst/dccp/gstdccpclientsrc.c:
* gst/dccp/gstdccpserversink.c:
* gst/dccp/gstdccpserversrc.c:
* gst/dccp/gstdccpserversrc.h:
CodeStyle improvements and minor fixes to the DCCP plugin.
Fixes #553823.
2008-11-20 Sebastian Dröge <slomo@circular-chaos.org> 2008-11-20 Sebastian Dröge <slomo@circular-chaos.org>
* gst/rtpmanager/gstrtpjitterbuffer.c: * gst/rtpmanager/gstrtpjitterbuffer.c:

229
gst/dccp/gstdccp.c
View file

@ -36,14 +36,16 @@
#endif #endif
/* Prototypes and definitions for private functions and not exported via gstdccp.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, GstFlowReturn gst_dccp_socket_write (GstElement * element, int socket,
int packet_size); 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, struct sockaddr_in gst_dccp_create_sockaddr (GstElement * element, gchar * ip,
int port); int port);
/* Resolves host to IP address /*
* @return a gchar pointer containing the ip address or NULL * Resolves host to IP address
* @param element - the element
* @return a gchar pointer containing the ip address or NULL if it
* couldn't resolve the host to a IP adress
*/ */
gchar * gchar *
gst_dccp_host_to_ip (GstElement * element, const gchar * host) gst_dccp_host_to_ip (GstElement * element, const gchar * host)
@ -83,10 +85,15 @@ gst_dccp_host_to_ip (GstElement * element, const gchar * host)
return ip; return ip;
} }
/* Read a buffer from the given socket /*
* @returns: * Read a buffer from the given socket
* a GstBuffer from which data should be read *
* or NULL, indicating a connection close or an error. Handle it with EOS. * @param this - the element that has the socket that will be read
* @param socket - the socket fd that will be read
* @param buf - the buffer with the data read from the socket
* @return GST_FLOW_OK if the read operation was successful
* or GST_FLOW_ERROR indicating a connection close or an error.
* Handle it with EOS.
*/ */
GstFlowReturn GstFlowReturn
gst_dccp_read_buffer (GstElement * this, int socket, GstBuffer ** buf) gst_dccp_read_buffer (GstElement * this, int socket, GstBuffer ** buf)
@ -96,6 +103,8 @@ gst_dccp_read_buffer (GstElement * this, int socket, GstBuffer ** buf)
int ret; int ret;
ssize_t bytes_read; ssize_t bytes_read;
int readsize; int readsize;
struct msghdr mh;
struct iovec iov;
*buf = NULL; *buf = NULL;
@ -124,7 +133,23 @@ gst_dccp_read_buffer (GstElement * this, int socket, GstBuffer ** buf)
} }
*buf = gst_buffer_new_and_alloc (readsize); *buf = gst_buffer_new_and_alloc (readsize);
bytes_read = read (socket, GST_BUFFER_DATA (*buf), readsize);
memset (&mh, 0, sizeof (mh));
mh.msg_name = NULL;
mh.msg_namelen = 0;
iov.iov_base = GST_BUFFER_DATA (*buf);
iov.iov_len = readsize;
mh.msg_iov = &iov;
mh.msg_iovlen = 1;
mh.msg_control = NULL;
mh.msg_controllen = 0;
bytes_read = recvmsg (socket, &mh, 0);
if (bytes_read != readsize) {
GST_DEBUG_OBJECT (this, ("Error while reading data"));
return GST_FLOW_ERROR;
}
GST_LOG_OBJECT (this, "bytes read %" G_GSIZE_FORMAT, bytes_read); GST_LOG_OBJECT (this, "bytes read %" G_GSIZE_FORMAT, bytes_read);
GST_LOG_OBJECT (this, "returning buffer of size %d", GST_BUFFER_SIZE (*buf)); GST_LOG_OBJECT (this, "returning buffer of size %d", GST_BUFFER_SIZE (*buf));
@ -132,7 +157,9 @@ gst_dccp_read_buffer (GstElement * this, int socket, GstBuffer ** buf)
return GST_FLOW_OK; return GST_FLOW_OK;
} }
/* Create a new socket /* Create a new DCCP socket
*
* @param element - the element
* @return the socket file descriptor * @return the socket file descriptor
*/ */
gint gint
@ -147,7 +174,10 @@ gst_dccp_create_new_socket (GstElement * element)
} }
/* Connect to a server /* Connect to a server
* @return true in case of successfull connection, false otherwise * @param element - the element
* @param server_sin - a struct with a server address and port
* @param sock_fd - the socket to connect
* @return TRUE in case of successful connection, FALSE otherwise
*/ */
gboolean gboolean
gst_dccp_connect_to_server (GstElement * element, struct sockaddr_in server_sin, gst_dccp_connect_to_server (GstElement * element, struct sockaddr_in server_sin,
@ -158,11 +188,15 @@ gst_dccp_connect_to_server (GstElement * element, struct sockaddr_in server_sin,
if (connect (sock_fd, (struct sockaddr *) &server_sin, sizeof (server_sin))) { if (connect (sock_fd, (struct sockaddr *) &server_sin, sizeof (server_sin))) {
switch (errno) { switch (errno) {
case ECONNREFUSED: case ECONNREFUSED:
GST_ERROR_OBJECT (element, "Connection refused."); GST_ELEMENT_ERROR (element, RESOURCE, OPEN_WRITE,
("Connection to %s:%d refused.", inet_ntoa (server_sin.sin_addr),
ntohs (server_sin.sin_port)), (NULL));
return FALSE; return FALSE;
break; break;
default: default:
GST_ERROR_OBJECT (element, "Connection failed."); GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL),
("Connect to %s:%d failed: %s", inet_ntoa (server_sin.sin_addr),
ntohs (server_sin.sin_port), g_strerror (errno)));
return FALSE; return FALSE;
break; break;
} }
@ -173,6 +207,9 @@ gst_dccp_connect_to_server (GstElement * element, struct sockaddr_in server_sin,
/* FIXME support only one client */ /* FIXME support only one client */
/* /*
* Accept connection on the server socket. * Accept connection on the server socket.
*
* @param element - the element
* @param server_sock_fd - the server socket file descriptor
* @return the socket of the client connected to the server. * @return the socket of the client connected to the server.
*/ */
gint gint
@ -183,29 +220,30 @@ gst_dccp_server_wait_connections (GstElement * element, int server_sock_fd)
struct sockaddr_in client_address; struct sockaddr_in client_address;
unsigned int client_address_len; 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)); memset (&client_address, 0, sizeof (client_address));
client_address_len = 0; client_address_len = 0;
if ((client_sock_fd = if ((client_sock_fd =
accept (server_sock_fd, (struct sockaddr *) &client_address, accept (server_sock_fd, (struct sockaddr *) &client_address,
&client_address_len)) == -1) { &client_address_len)) == -1) {
GST_ELEMENT_ERROR (element, RESOURCE, OPEN_WRITE, (NULL),
("Could not accept client on server socket %d: %s (%d)",
server_sock_fd, g_strerror (errno), errno));
return -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", GST_DEBUG_OBJECT (element, "Added new client ip %s with fd %d.",
inet_ntoa (client_address.sin_addr), client_sock_fd); inet_ntoa (client_address.sin_addr), client_sock_fd);
/* return the thread object, instead of the fd */
return client_sock_fd; return client_sock_fd;
} }
/* /*
* Bind a server address. * Bind a server address.
*
* @param element - the element
* @param server_sock_fd - the server socket fd
* @param server_sin - the address and the port to bind the server on
* @return true in success, false otherwise. * @return true in success, false otherwise.
*/ */
gboolean gboolean
@ -214,14 +252,15 @@ gst_dccp_bind_server_socket (GstElement * element, int server_sock_fd,
{ {
int ret; int ret;
GST_DEBUG_OBJECT (element, "binding server socket to address"); GST_DEBUG_OBJECT (element, "Binding server socket to address.");
ret = bind (server_sock_fd, (struct sockaddr *) &server_sin, ret = bind (server_sock_fd, (struct sockaddr *) &server_sin,
sizeof (server_sin)); sizeof (server_sin));
if (ret) { if (ret) {
switch (errno) { switch (errno) {
default: default:
GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL), GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL),
("bind on port %d failed: %s", server_sin.sin_port, ("Bind on port %d failed: %s", ntohs (server_sin.sin_port),
g_strerror (errno))); g_strerror (errno)));
return FALSE; return FALSE;
break; break;
@ -230,45 +269,65 @@ gst_dccp_bind_server_socket (GstElement * element, int server_sock_fd,
return TRUE; return TRUE;
} }
/*
* Listen on server socket.
*
* @param element - the element
* @param server_sock_fd - the server socket fd
* @return true in success, false otherwise.
*/
gboolean gboolean
gst_dccp_listen_server_socket (GstElement * element, int server_sock_fd) gst_dccp_listen_server_socket (GstElement * element, int server_sock_fd)
{ {
GST_DEBUG_OBJECT (element, "listening on server socket %d with queue of %d", GST_DEBUG_OBJECT (element, "Listening on server socket %d with queue of %d",
server_sock_fd, DCCP_BACKLOG); server_sock_fd, DCCP_BACKLOG);
if (listen (server_sock_fd, DCCP_BACKLOG) == -1) { if (listen (server_sock_fd, DCCP_BACKLOG) == -1) {
GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL), GST_ELEMENT_ERROR (element, RESOURCE, OPEN_READ, (NULL),
("Could not listen on server socket: %s", g_strerror (errno))); ("Could not listen on server socket: %s", g_strerror (errno)));
return FALSE; return FALSE;
} }
GST_DEBUG_OBJECT (element, GST_DEBUG_OBJECT (element,
"listened on server socket %d, returning from connection setup", "Listened on server socket %d, returning from connection setup",
server_sock_fd); server_sock_fd);
return TRUE; return TRUE;
} }
/* FIXME */
gboolean
gst_dccp_socket_connected (GstElement * element, int server_sock_fd)
{
return FALSE;
}
/* Write buffer to given socket incrementally. /* Write buffer to given socket incrementally.
* Returns number of bytes written. *
* @param element - the element
* @param socket - the socket
* @param buf - the buffer that will be written
* @param size - the number of bytes of the buffer
* @param packet_size - the MTU
* @return the number of bytes written.
*/ */
gint GstFlowReturn
gst_dccp_socket_write (int socket, const void *buf, size_t size, gst_dccp_socket_write (GstElement * element, int socket, const void *buf,
int packet_size) size_t size, int packet_size)
{ {
size_t bytes_written = 0; size_t bytes_written = 0;
ssize_t wrote; ssize_t wrote;
struct iovec iov;
struct msghdr mh;
memset (&mh, 0, sizeof (mh));
while (bytes_written < size) { while (bytes_written < size) {
do { do {
wrote = write (socket, (const char *) buf + bytes_written, mh.msg_name = NULL;
MIN (packet_size, size - bytes_written)); mh.msg_namelen = 0;
iov.iov_base = (char *) buf + bytes_written;
iov.iov_len = MIN (packet_size, size - bytes_written);
mh.msg_iov = &iov;
mh.msg_iovlen = 1;
mh.msg_control = NULL;
mh.msg_controllen = 0;
wrote = sendmsg (socket, &mh, 0);
} while (wrote == -1 && errno == EAGAIN); } while (wrote == -1 && errno == EAGAIN);
/* TODO print the send error */ /* TODO print the send error */
@ -276,17 +335,35 @@ gst_dccp_socket_write (int socket, const void *buf, size_t size,
} }
if (bytes_written < 0) if (bytes_written < 0)
GST_WARNING ("error while writing"); GST_WARNING ("Error while writing.");
else else
GST_LOG ("wrote %" G_GSIZE_FORMAT " bytes succesfully", bytes_written); GST_LOG_OBJECT (element, "Wrote %" G_GSIZE_FORMAT " bytes succesfully.",
return bytes_written; bytes_written);
if (bytes_written != size) {
GST_ELEMENT_ERROR (element, RESOURCE, WRITE,
("Error while sending data to socket %d.", socket),
("Only %" G_GSIZE_FORMAT " of %u bytes written: %s",
bytes_written, size, g_strerror (errno)));
return GST_FLOW_ERROR;
}
return GST_FLOW_OK;
} }
/* Write buffer to given socket.
*
* @param this - the element
* @param buf - the buffer that will be written
* @param client_sock_fd - the client socket
* @param packet_size - the MTU
* @return GST_FLOW_OK if the send operation was successful, GST_FLOW_ERROR otherwise.
*/
GstFlowReturn GstFlowReturn
gst_dccp_send_buffer (GstElement * this, GstBuffer * buffer, int client_sock_fd, gst_dccp_send_buffer (GstElement * this, GstBuffer * buffer, int client_sock_fd,
int packet_size) int packet_size)
{ {
size_t wrote; // size_t wrote;
gint size = 0; gint size = 0;
guint8 *data; guint8 *data;
@ -296,21 +373,17 @@ gst_dccp_send_buffer (GstElement * this, GstBuffer * buffer, int client_sock_fd,
GST_LOG_OBJECT (this, "writing %d bytes", size); GST_LOG_OBJECT (this, "writing %d bytes", size);
if (packet_size < 0) { if (packet_size < 0) {
GST_LOG_OBJECT (this, "error getting MTU");
return GST_FLOW_ERROR; return GST_FLOW_ERROR;
} }
wrote = gst_dccp_socket_write (client_sock_fd, data, size, packet_size); return gst_dccp_socket_write (this, 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. * Create socket address.
* @param element - the element
* @param ip - the ip address
* @param port - the port
* @return sockaddr_in. * @return sockaddr_in.
*/ */
struct sockaddr_in struct sockaddr_in
@ -326,6 +399,12 @@ gst_dccp_create_sockaddr (GstElement * element, gchar * ip, int port)
return sin; return sin;
} }
/*
* Make address reusable.
* @param element - the element
* @param sock_fd - the socket
* @return TRUE if the operation was successful, FALSE otherwise.
*/
gboolean gboolean
gst_dccp_make_address_reusable (GstElement * element, int sock_fd) gst_dccp_make_address_reusable (GstElement * element, int sock_fd)
{ {
@ -340,7 +419,13 @@ gst_dccp_make_address_reusable (GstElement * element, int sock_fd)
return TRUE; return TRUE;
} }
/* DCCP socket specific stuffs */ /*
* Set DCCP congestion control.
* @param element - the element
* @param sock_fd - the socket
* @param ccid - the ccid number
* @return TRUE if the operation was successful, FALSE otherwise.
*/
gboolean gboolean
gst_dccp_set_ccid (GstElement * element, int sock_fd, uint8_t ccid) gst_dccp_set_ccid (GstElement * element, int sock_fd, uint8_t ccid)
{ {
@ -408,6 +493,12 @@ gst_dccp_get_ccid (GstElement * element, int sock_fd, int tx_or_rx)
return ccid; return ccid;
} }
/*
* Get the socket MTU.
* @param element - the element
* @param sock - the socket
* @return the MTU if the operation was successful, -1 otherwise.
*/
gint gint
gst_dccp_get_max_packet_size (GstElement * element, int sock) gst_dccp_get_max_packet_size (GstElement * element, int sock)
{ {
@ -423,36 +514,12 @@ gst_dccp_get_max_packet_size (GstElement * element, int sock)
return size; return size;
} }
/* Still not used and need to be FIXED */ void
gboolean gst_dccp_socket_close (GstElement * element, int *socket)
gst_dccp_set_sock_windowsize (GstElement * element, int sock, int winSize,
gboolean inSend)
{ {
#ifdef SO_SNDBUF if (socket >= 0) {
int rc; GST_DEBUG_OBJECT (element, "closing socket");
close (*socket);
if (!inSend) { *socket = -1;
/* 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", 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", 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;
} }

3
gst/dccp/gstdccp.h
View file

@ -90,7 +90,6 @@ GstFlowReturn gst_dccp_send_buffer (GstElement * element, GstBuffer * buffer,
int client_sock_fd, int packet_size); int client_sock_fd, int packet_size);
gboolean gst_dccp_make_address_reusable (GstElement * element, int sock_fd); gboolean gst_dccp_make_address_reusable (GstElement * element, int sock_fd);
gboolean gst_dccp_set_sock_windowsize(GstElement * element, int sock, int winSize, void gst_dccp_socket_close (GstElement * element, int * socket);
gboolean inSend);
#endif /* __GST_DCCP_H__ */ #endif /* __GST_DCCP_H__ */

25
gst/dccp/gstdccpclientsink.c
View file

@ -96,6 +96,11 @@ GST_BOILERPLATE (GstDCCPClientSink, gst_dccp_client_sink, GstBaseSink,
static guint gst_dccp_client_sink_signals[LAST_SIGNAL] = { 0 }; static guint gst_dccp_client_sink_signals[LAST_SIGNAL] = { 0 };
/*
* Write buffer to client socket.
*
* @return GST_FLOW_OK if the send operation was successful, GST_FLOW_ERROR otherwise.
*/
static GstFlowReturn static GstFlowReturn
gst_dccp_client_sink_render (GstBaseSink * bsink, GstBuffer * buf) gst_dccp_client_sink_render (GstBaseSink * bsink, GstBuffer * buf)
{ {
@ -172,6 +177,13 @@ gst_dccp_client_sink_get_property (GObject * object, guint prop_id,
} }
} }
/*
* Starts the element. If the sockfd property was not the default, this method
* will create a new socket and connect to the server.
*
* @param bsink - the element
* @return TRUE if the send operation was successful, FALSE otherwise.
*/
static gboolean static gboolean
gst_dccp_client_sink_start (GstBaseSink * bsink) gst_dccp_client_sink_start (GstBaseSink * bsink)
{ {
@ -250,10 +262,8 @@ gst_dccp_client_sink_stop (GstBaseSink * bsink)
sink = GST_DCCP_CLIENT_SINK (bsink); sink = GST_DCCP_CLIENT_SINK (bsink);
if (sink->sock_fd != -1 && sink->closed) { if (sink->sock_fd != DCCP_DEFAULT_SOCK_FD && sink->closed) {
GST_DEBUG_OBJECT (sink, "closing socket"); gst_dccp_socket_close (GST_ELEMENT (sink), &(sink->sock_fd));
close (sink->sock_fd);
sink->sock_fd = -1;
} }
return TRUE; return TRUE;
@ -276,10 +286,9 @@ gst_dccp_client_sink_class_init (GstDCCPClientSinkClass * klass)
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT, g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
g_param_spec_int ("port", "Port", g_param_spec_int ("port", "Port",
"The port to receive the packets from, 0=allocate", 0, G_MAXUINT16, "The port to send the packets to", 0, G_MAXUINT16,
DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); DCCP_DEFAULT_PORT, G_PARAM_READWRITE));
/* FIXME property server_ip */
g_object_class_install_property (gobject_class, PROP_HOST, g_object_class_install_property (gobject_class, PROP_HOST,
g_param_spec_string ("host", "Host", g_param_spec_string ("host", "Host",
"The host IP address to send packets to", DCCP_DEFAULT_HOST, "The host IP address to send packets to", DCCP_DEFAULT_HOST,
@ -303,8 +312,8 @@ gst_dccp_client_sink_class_init (GstDCCPClientSinkClass * klass)
/* signals */ /* signals */
/** /**
* GstDccpClientSink::connected: * GstDccpClientSink::connected:
* @sink: the gstdccpclientsink instance * @sink: the gstdccpclientsink element that emitted this signal
* @fd: the connected socket fd * @fd: the connected socket file descriptor
* *
* Sign that the element has connected, return the fd of the socket. * Sign that the element has connected, return the fd of the socket.
*/ */

5
gst/dccp/gstdccpclientsink.h
View file

@ -47,6 +47,11 @@ typedef struct _GstDCCPClientSink GstDCCPClientSink;
typedef struct _GstDCCPClientSinkClass GstDCCPClientSinkClass; typedef struct _GstDCCPClientSinkClass GstDCCPClientSinkClass;
/**
* GstDCCPClientSink:
*
* dccpclientsink object structure.
*/
struct _GstDCCPClientSink struct _GstDCCPClientSink
{ {
GstBaseSink element; GstBaseSink element;

24
gst/dccp/gstdccpclientsrc.c
View file

@ -102,6 +102,11 @@ GST_BOILERPLATE (GstDCCPClientSrc, gst_dccp_client_src, GstPushSrc,
static guint gst_dccp_client_src_signals[LAST_SIGNAL] = { 0 }; static guint gst_dccp_client_src_signals[LAST_SIGNAL] = { 0 };
/*
* Read a buffer from the client socket
*
* @return GST_FLOW_OK if the send operation was successful, GST_FLOW_ERROR otherwise.
*/
static GstFlowReturn static GstFlowReturn
gst_dccp_client_src_create (GstPushSrc * psrc, GstBuffer ** outbuf) gst_dccp_client_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
{ {
@ -220,6 +225,13 @@ gst_dccp_client_src_get_property (GObject * object, guint prop_id,
} }
} }
/*
* Starts the element. If the sockfd property was not the default, this method
* will create a new socket and connect to the server.
*
* @param bsrc - the element
* @return TRUE if the send operation was successful, FALSE otherwise.
*/
static gboolean static gboolean
gst_dccp_client_src_start (GstBaseSrc * bsrc) gst_dccp_client_src_start (GstBaseSrc * bsrc)
{ {
@ -323,10 +335,8 @@ gst_dccp_client_src_stop (GstBaseSrc * bsrc)
src = GST_DCCP_CLIENT_SRC (bsrc); src = GST_DCCP_CLIENT_SRC (bsrc);
if (src->sock_fd != -1 && src->closed) { if (src->sock_fd != DCCP_DEFAULT_SOCK_FD && src->closed) {
GST_DEBUG_OBJECT (src, "closing socket"); gst_dccp_socket_close (GST_ELEMENT (src), &(src->sock_fd));
close (src->sock_fd);
src->sock_fd = -1;
} }
return TRUE; return TRUE;
@ -353,7 +363,7 @@ gst_dccp_client_src_class_init (GstDCCPClientSrcClass * klass)
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT, g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
g_param_spec_int ("port", "Port", g_param_spec_int ("port", "Port",
"The port to receive the packets from, 0=allocate", 0, G_MAXUINT16, "The port to receive packets from", 0, G_MAXUINT16,
DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); DCCP_DEFAULT_PORT, G_PARAM_READWRITE));
g_object_class_install_property (gobject_class, PROP_HOST, g_object_class_install_property (gobject_class, PROP_HOST,
@ -383,8 +393,8 @@ gst_dccp_client_src_class_init (GstDCCPClientSrcClass * klass)
/* signals */ /* signals */
/** /**
* GstDccpClientSrc::connected: * GstDccpClientSrc::connected:
* @src: the gstdccpclientsrc instance * @src: the gstdccpclientsrc element that emitted this signal
* @fd: the connected socket fd * @fd: the connected socket file descriptor
* *
* Reports that the element has connected, giving the fd of the socket * Reports that the element has connected, giving the fd of the socket
*/ */

59
gst/dccp/gstdccpserversink.c
View file

@ -98,6 +98,13 @@ GST_BOILERPLATE (GstDCCPServerSink, gst_dccp_server_sink, GstBaseSink,
static guint gst_dccp_server_sink_signals[LAST_SIGNAL] = { 0 }; static guint gst_dccp_server_sink_signals[LAST_SIGNAL] = { 0 };
/*
* Create a new client with the socket and the MTU
*
* @param element - the gstdccpserversink instance
* @param socket - the client socket
* @return the client
*/
static Client * static Client *
gst_dccp_server_create_client (GstElement * element, int socket) gst_dccp_server_create_client (GstElement * element, int socket)
{ {
@ -106,6 +113,9 @@ gst_dccp_server_create_client (GstElement * element, int socket)
client->pksize = gst_dccp_get_max_packet_size (element, client->socket); client->pksize = gst_dccp_get_max_packet_size (element, client->socket);
client->flow_status = GST_FLOW_OK; client->flow_status = GST_FLOW_OK;
GST_DEBUG_OBJECT (element, "Creating a new client with fd %d and MTU %d.",
client->socket, client->pksize);
/* the socket is connected */ /* the socket is connected */
g_signal_emit (element, gst_dccp_server_sink_signals[SIGNAL_CONNECTED], 0, g_signal_emit (element, gst_dccp_server_sink_signals[SIGNAL_CONNECTED], 0,
socket); socket);
@ -113,6 +123,11 @@ gst_dccp_server_create_client (GstElement * element, int socket)
return client; return client;
} }
/*
* Wait connections of new clients
*
* @param arg - the gstdccpserversink instance
*/
static void * static void *
gst_dccp_server_accept_new_clients (void *arg) gst_dccp_server_accept_new_clients (void *arg)
{ {
@ -132,6 +147,11 @@ gst_dccp_server_accept_new_clients (void *arg)
} }
} }
/*
* Send the buffer to a client
*
* @param arg - the client
*/
static void * static void *
gst_dccp_server_send_buffer (void *arg) gst_dccp_server_send_buffer (void *arg)
{ {
@ -148,7 +168,10 @@ gst_dccp_server_send_buffer (void *arg)
return NULL; return NULL;
} }
/* Remove clients with problems to send */ /* Remove clients with problems to send.
*
* @param arg - the gstdccpserversink instance
*/
static void * static void *
gst_dccp_server_delete_dead_clients (void *arg) gst_dccp_server_delete_dead_clients (void *arg)
{ {
@ -185,23 +208,25 @@ gst_dccp_server_sink_init (GstDCCPServerSink * this,
this->clients = NULL; this->clients = NULL;
} }
/*
* Starts the element. If the sockfd property was not the default, this method
* will wait for a client connection. If wait-connections property is true, it
* creates a thread to wait for new client connections.
*
* @param bsink - the element
* @return TRUE if the send operation was successful, FALSE otherwise.
*/
static gboolean static gboolean
gst_dccp_server_sink_start (GstBaseSink * bsink) gst_dccp_server_sink_start (GstBaseSink * bsink)
{ {
GstDCCPServerSink *sink = GST_DCCP_SERVER_SINK (bsink); GstDCCPServerSink *sink = GST_DCCP_SERVER_SINK (bsink);
int ret = 1;
Client *client; Client *client;
/* create socket */
if ((sink->sock_fd = gst_dccp_create_new_socket (GST_ELEMENT (sink))) < 0) { if ((sink->sock_fd = gst_dccp_create_new_socket (GST_ELEMENT (sink))) < 0) {
return FALSE; return FALSE;
} }
/* make address reusable */ if (!gst_dccp_make_address_reusable (GST_ELEMENT (sink), sink->sock_fd)) {
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; return FALSE;
} }
@ -290,16 +315,14 @@ gst_dccp_server_sink_stop (GstBaseSink * bsink)
pthread_cancel (accept_thread_id); pthread_cancel (accept_thread_id);
} }
if (sink->sock_fd != -1 && sink->closed == TRUE) { gst_dccp_socket_close (GST_ELEMENT (sink), &(sink->sock_fd));
GST_DEBUG_OBJECT (sink, "closing socket");
close (sink->sock_fd);
sink->sock_fd = -1;
}
pthread_mutex_lock (&lock); pthread_mutex_lock (&lock);
for (i = 0; i < g_list_length (sink->clients); i++) { for (i = 0; i < g_list_length (sink->clients); i++) {
Client *client = (Client *) g_list_nth_data (sink->clients, i); Client *client = (Client *) g_list_nth_data (sink->clients, i);
close (client->socket); if (client->socket != DCCP_DEFAULT_CLIENT_SOCK_FD && sink->closed == TRUE) {
gst_dccp_socket_close (GST_ELEMENT (sink), &(client->socket));
}
g_free (client); g_free (client);
} }
pthread_mutex_unlock (&lock); pthread_mutex_unlock (&lock);
@ -391,7 +414,7 @@ gst_dccp_server_sink_class_init (GstDCCPServerSinkClass * klass)
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT, g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
g_param_spec_int ("port", "Port", g_param_spec_int ("port", "Port",
"The port to receive the packets from, 0=allocate", 0, G_MAXUINT16, "The port to listen to", 0, G_MAXUINT16,
DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); DCCP_DEFAULT_PORT, G_PARAM_READWRITE));
g_object_class_install_property (gobject_class, PROP_CLIENT_SOCK_FD, g_object_class_install_property (gobject_class, PROP_CLIENT_SOCK_FD,
@ -401,7 +424,7 @@ gst_dccp_server_sink_class_init (GstDCCPServerSinkClass * klass)
g_object_class_install_property (gobject_class, PROP_CLOSED, g_object_class_install_property (gobject_class, PROP_CLOSED,
g_param_spec_boolean ("close-socket", "Close", g_param_spec_boolean ("close-socket", "Close",
"Close socket at end of stream", "Close the client sockets at end of stream",
DCCP_DEFAULT_CLOSED, G_PARAM_READWRITE)); DCCP_DEFAULT_CLOSED, G_PARAM_READWRITE));
g_object_class_install_property (gobject_class, PROP_CCID, g_object_class_install_property (gobject_class, PROP_CCID,
@ -418,8 +441,8 @@ gst_dccp_server_sink_class_init (GstDCCPServerSinkClass * klass)
/* signals */ /* signals */
/** /**
* GstDccpServerSink::connected: * GstDccpServerSink::connected:
* @src: the gstdccpserversink instance * @sink: the gstdccpserversink element that emitted this signal
* @fd: the connected socket fd * @fd: the connected socket file descriptor
* *
* Reports that the element has connected, giving the fd of the socket * Reports that the element has connected, giving the fd of the socket
*/ */

52
gst/dccp/gstdccpserversrc.c
View file

@ -56,7 +56,6 @@
#include <fcntl.h> #include <fcntl.h>
#define DCCP_DEFAULT_CAPS NULL #define DCCP_DEFAULT_CAPS NULL
#define DCCP_DEFAULT_LISTEN_HOST NULL /* listen on all interfaces */
/* signals */ /* signals */
enum enum
@ -70,7 +69,6 @@ enum
{ {
PROP_0, PROP_0,
PROP_PORT, PROP_PORT,
PROP_HOST,
PROP_CLIENT_SOCK_FD, PROP_CLIENT_SOCK_FD,
PROP_CLOSED, PROP_CLOSED,
PROP_CCID, PROP_CCID,
@ -97,6 +95,11 @@ GST_BOILERPLATE (GstDCCPServerSrc, gst_dccp_server_src, GstPushSrc,
static guint gst_dccp_server_src_signals[LAST_SIGNAL] = { 0 }; static guint gst_dccp_server_src_signals[LAST_SIGNAL] = { 0 };
/*
* Read a buffer from the server socket
*
* @return GST_FLOW_OK if the send operation was successful, GST_FLOW_ERROR otherwise.
*/
static GstFlowReturn static GstFlowReturn
gst_dccp_server_src_create (GstPushSrc * psrc, GstBuffer ** outbuf) gst_dccp_server_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
{ {
@ -137,14 +140,6 @@ gst_dccp_server_src_set_property (GObject * object, guint prop_id,
GstDCCPServerSrc *src = GST_DCCP_SERVER_SRC (object); GstDCCPServerSrc *src = GST_DCCP_SERVER_SRC (object);
switch (prop_id) { 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: case PROP_PORT:
src->port = g_value_get_int (value); src->port = g_value_get_int (value);
break; break;
@ -193,9 +188,6 @@ gst_dccp_server_src_get_property (GObject * object, guint prop_id,
GstDCCPServerSrc *src = GST_DCCP_SERVER_SRC (object); GstDCCPServerSrc *src = GST_DCCP_SERVER_SRC (object);
switch (prop_id) { switch (prop_id) {
case PROP_HOST:
g_value_set_string (value, src->host);
break;
case PROP_PORT: case PROP_PORT:
g_value_set_int (value, src->port); g_value_set_int (value, src->port);
break; break;
@ -217,7 +209,13 @@ gst_dccp_server_src_get_property (GObject * object, guint prop_id,
} }
} }
/*
* Starts the element. If the sockfd property was not the default, this method
* will create a new server socket and wait for a client connection.
*
* @param bsrc - the element
* @return TRUE if the send operation was successful, FALSE otherwise.
*/
static gboolean static gboolean
gst_dccp_server_src_start (GstBaseSrc * bsrc) gst_dccp_server_src_start (GstBaseSrc * bsrc)
{ {
@ -229,7 +227,9 @@ gst_dccp_server_src_start (GstBaseSrc * bsrc)
return FALSE; return FALSE;
} }
gst_dccp_make_address_reusable (GST_ELEMENT (src), src->sock_fd); if (!gst_dccp_make_address_reusable (GST_ELEMENT (src), src->sock_fd)) {
return FALSE;
}
/* name the server socket */ /* name the server socket */
memset (&src->server_sin, 0, sizeof (src->server_sin)); memset (&src->server_sin, 0, sizeof (src->server_sin));
@ -283,7 +283,6 @@ gst_dccp_server_src_init (GstDCCPServerSrc * this,
GstDCCPServerSrcClass * g_class) GstDCCPServerSrcClass * g_class)
{ {
this->port = DCCP_DEFAULT_PORT; this->port = DCCP_DEFAULT_PORT;
this->host = g_strdup (DCCP_DEFAULT_HOST);
this->sock_fd = DCCP_DEFAULT_SOCK_FD; this->sock_fd = DCCP_DEFAULT_SOCK_FD;
this->client_sock_fd = DCCP_DEFAULT_CLIENT_SOCK_FD; this->client_sock_fd = DCCP_DEFAULT_CLIENT_SOCK_FD;
this->closed = DCCP_DEFAULT_CLOSED; this->closed = DCCP_DEFAULT_CLOSED;
@ -314,8 +313,6 @@ gst_dccp_server_src_finalize (GObject * gobject)
this->caps = NULL; this->caps = NULL;
} }
g_free (this->host);
G_OBJECT_CLASS (parent_class)->finalize (gobject); G_OBJECT_CLASS (parent_class)->finalize (gobject);
} }
@ -327,10 +324,9 @@ gst_dccp_server_src_stop (GstBaseSrc * bsrc)
src = GST_DCCP_SERVER_SRC (bsrc); src = GST_DCCP_SERVER_SRC (bsrc);
if (src->sock_fd != -1 && src->closed == TRUE) { gst_dccp_socket_close (GST_ELEMENT (src), &(src->sock_fd));
GST_DEBUG_OBJECT (src, "closing socket"); if (src->client_sock_fd != DCCP_DEFAULT_CLIENT_SOCK_FD && src->closed == TRUE) {
close (src->sock_fd); gst_dccp_socket_close (GST_ELEMENT (src), &(src->client_sock_fd));
src->sock_fd = -1;
} }
return TRUE; return TRUE;
@ -354,13 +350,9 @@ gst_dccp_server_src_class_init (GstDCCPServerSrcClass * klass)
g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT, g_object_class_install_property (G_OBJECT_CLASS (klass), PROP_PORT,
g_param_spec_int ("port", "Port", g_param_spec_int ("port", "Port",
"The port to receive the packets from, 0=allocate", 0, G_MAXUINT16, "The port to listen to", 0, G_MAXUINT16,
DCCP_DEFAULT_PORT, G_PARAM_READWRITE)); 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_object_class_install_property (gobject_class, PROP_CLIENT_SOCK_FD,
g_param_spec_int ("sockfd", "Socket fd", g_param_spec_int ("sockfd", "Socket fd",
"The client socket file descriptor", -1, G_MAXINT, "The client socket file descriptor", -1, G_MAXINT,
@ -368,7 +360,7 @@ gst_dccp_server_src_class_init (GstDCCPServerSrcClass * klass)
g_object_class_install_property (gobject_class, PROP_CLOSED, g_object_class_install_property (gobject_class, PROP_CLOSED,
g_param_spec_boolean ("close-socket", "Close socket", g_param_spec_boolean ("close-socket", "Close socket",
"Close socket at the end of stream", DCCP_DEFAULT_CLOSED, "Close client socket at the end of stream", DCCP_DEFAULT_CLOSED,
G_PARAM_READWRITE)); G_PARAM_READWRITE));
g_object_class_install_property (gobject_class, PROP_CCID, g_object_class_install_property (gobject_class, PROP_CCID,
@ -383,8 +375,8 @@ gst_dccp_server_src_class_init (GstDCCPServerSrcClass * klass)
/* signals */ /* signals */
/** /**
* GstDccpServerSrc::connected: * GstDccpServerSrc::connected:
* @src: the gstdccpserversrc instance * @src: the gstdccpserversrc element that emitted this signal
* @fd: the connected socket fd * @fd: the connected socket file descriptor
* *
* Reports that the element has connected, giving the fd of the socket * Reports that the element has connected, giving the fd of the socket
*/ */

1
gst/dccp/gstdccpserversrc.h
View file

@ -53,7 +53,6 @@ struct _GstDCCPServerSrc
/* server information */ /* server information */
int port; int port;
gchar *host;
struct sockaddr_in server_sin; struct sockaddr_in server_sin;
/* socket */ /* socket */