Merge branch 'master' into 0.11
[platform/upstream/gstreamer.git] / gst / tcp / gsttcpclientsrc.c
1 /* GStreamer
2  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
3  * Copyright (C) <2004> Thomas Vander Stichele <thomas at apestaart dot org>
4  * Copyright (C) <2011> Collabora Ltd.
5  *     Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
6  *
7  * This library is free software; you can redistribute it and/or
8  * modify it under the terms of the GNU Library General Public
9  * License as published by the Free Software Foundation; either
10  * version 2 of the License, or (at your option) any later version.
11  *
12  * This library is distributed in the hope that it will be useful,
13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * Library General Public License for more details.
16  *
17  * You should have received a copy of the GNU Library General Public
18  * License along with this library; if not, write to the
19  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
20  * Boston, MA 02111-1307, USA.
21  */
22
23 /**
24  * SECTION:element-tcpclientsrc
25  * @see_also: #tcpclientsink
26  *
27  * <refsect2>
28  * <title>Example launch line</title>
29  * |[
30  * # server:
31  * nc -l -p 3000
32  * # client:
33  * gst-launch tcpclientsrc port=3000 ! fdsink fd=2
34  * ]| everything you type in the server is shown on the client
35  * </refsect2>
36  */
37
38 #ifdef HAVE_CONFIG_H
39 #include "config.h"
40 #endif
41
42 #include <gst/gst-i18n-plugin.h>
43 #include "gsttcpclientsrc.h"
44 #include "gsttcp.h"
45
46 GST_DEBUG_CATEGORY_STATIC (tcpclientsrc_debug);
47 #define GST_CAT_DEFAULT tcpclientsrc_debug
48
49 #define MAX_READ_SIZE                   4 * 1024
50
51
52 static GstStaticPadTemplate srctemplate = GST_STATIC_PAD_TEMPLATE ("src",
53     GST_PAD_SRC,
54     GST_PAD_ALWAYS,
55     GST_STATIC_CAPS_ANY);
56
57
58 enum
59 {
60   PROP_0,
61   PROP_HOST,
62   PROP_PORT
63 };
64
65 #define gst_tcp_client_src_parent_class parent_class
66 G_DEFINE_TYPE (GstTCPClientSrc, gst_tcp_client_src, GST_TYPE_PUSH_SRC);
67
68
69 static void gst_tcp_client_src_finalize (GObject * gobject);
70
71 static GstCaps *gst_tcp_client_src_getcaps (GstBaseSrc * psrc,
72     GstCaps * filter);
73
74 static GstFlowReturn gst_tcp_client_src_create (GstPushSrc * psrc,
75     GstBuffer ** outbuf);
76 static gboolean gst_tcp_client_src_stop (GstBaseSrc * bsrc);
77 static gboolean gst_tcp_client_src_start (GstBaseSrc * bsrc);
78 static gboolean gst_tcp_client_src_unlock (GstBaseSrc * bsrc);
79 static gboolean gst_tcp_client_src_unlock_stop (GstBaseSrc * bsrc);
80
81 static void gst_tcp_client_src_set_property (GObject * object, guint prop_id,
82     const GValue * value, GParamSpec * pspec);
83 static void gst_tcp_client_src_get_property (GObject * object, guint prop_id,
84     GValue * value, GParamSpec * pspec);
85
86 static void
87 gst_tcp_client_src_class_init (GstTCPClientSrcClass * klass)
88 {
89   GObjectClass *gobject_class;
90   GstElementClass *gstelement_class;
91   GstBaseSrcClass *gstbasesrc_class;
92   GstPushSrcClass *gstpush_src_class;
93
94   gobject_class = (GObjectClass *) klass;
95   gstelement_class = (GstElementClass *) klass;
96   gstbasesrc_class = (GstBaseSrcClass *) klass;
97   gstpush_src_class = (GstPushSrcClass *) klass;
98
99   gobject_class->set_property = gst_tcp_client_src_set_property;
100   gobject_class->get_property = gst_tcp_client_src_get_property;
101   gobject_class->finalize = gst_tcp_client_src_finalize;
102
103   g_object_class_install_property (gobject_class, PROP_HOST,
104       g_param_spec_string ("host", "Host",
105           "The host IP address to receive packets from", TCP_DEFAULT_HOST,
106           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
107   g_object_class_install_property (gobject_class, PROP_PORT,
108       g_param_spec_int ("port", "Port", "The port to receive packets from", 0,
109           TCP_HIGHEST_PORT, TCP_DEFAULT_PORT,
110           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
111
112   gst_element_class_add_pad_template (gstelement_class,
113       gst_static_pad_template_get (&srctemplate));
114
115   gst_element_class_set_details_simple (gstelement_class,
116       "TCP client source", "Source/Network",
117       "Receive data as a client over the network via TCP",
118       "Thomas Vander Stichele <thomas at apestaart dot org>");
119
120   gstbasesrc_class->get_caps = gst_tcp_client_src_getcaps;
121   gstbasesrc_class->start = gst_tcp_client_src_start;
122   gstbasesrc_class->stop = gst_tcp_client_src_stop;
123   gstbasesrc_class->unlock = gst_tcp_client_src_unlock;
124   gstbasesrc_class->unlock_stop = gst_tcp_client_src_unlock_stop;
125
126   gstpush_src_class->create = gst_tcp_client_src_create;
127
128   GST_DEBUG_CATEGORY_INIT (tcpclientsrc_debug, "tcpclientsrc", 0,
129       "TCP Client Source");
130 }
131
132 static void
133 gst_tcp_client_src_init (GstTCPClientSrc * this)
134 {
135   this->port = TCP_DEFAULT_PORT;
136   this->host = g_strdup (TCP_DEFAULT_HOST);
137   this->socket = NULL;
138   this->cancellable = g_cancellable_new ();
139
140   GST_OBJECT_FLAG_UNSET (this, GST_TCP_CLIENT_SRC_OPEN);
141 }
142
143 static void
144 gst_tcp_client_src_finalize (GObject * gobject)
145 {
146   GstTCPClientSrc *this = GST_TCP_CLIENT_SRC (gobject);
147
148   if (this->cancellable)
149     g_object_unref (this->cancellable);
150   this->cancellable = NULL;
151   if (this->socket)
152     g_object_unref (this->socket);
153   this->socket = NULL;
154   g_free (this->host);
155   this->host = NULL;
156
157   G_OBJECT_CLASS (parent_class)->finalize (gobject);
158 }
159
160 static GstCaps *
161 gst_tcp_client_src_getcaps (GstBaseSrc * bsrc, GstCaps * filter)
162 {
163   GstTCPClientSrc *src;
164   GstCaps *caps = NULL;
165
166   src = GST_TCP_CLIENT_SRC (bsrc);
167
168   caps = (filter ? gst_caps_ref (filter) : gst_caps_new_any ());
169
170   GST_DEBUG_OBJECT (src, "returning caps %" GST_PTR_FORMAT, caps);
171   g_assert (GST_IS_CAPS (caps));
172   return caps;
173 }
174
175 static GstFlowReturn
176 gst_tcp_client_src_create (GstPushSrc * psrc, GstBuffer ** outbuf)
177 {
178   GstTCPClientSrc *src;
179   GstFlowReturn ret = GST_FLOW_OK;
180   gssize rret;
181   GError *err = NULL;
182   guint8 *data;
183   gssize avail, read;
184
185   src = GST_TCP_CLIENT_SRC (psrc);
186
187   if (!GST_OBJECT_FLAG_IS_SET (src, GST_TCP_CLIENT_SRC_OPEN))
188     goto wrong_state;
189
190   GST_LOG_OBJECT (src, "asked for a buffer");
191
192   /* read the buffer header */
193   avail = g_socket_get_available_bytes (src->socket);
194   if (avail < 0) {
195     goto get_available_error;
196   } else if (avail == 0) {
197     GIOCondition condition;
198
199     if (!g_socket_condition_wait (src->socket,
200             G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP, src->cancellable, &err))
201       goto select_error;
202
203     condition =
204         g_socket_condition_check (src->socket,
205         G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP);
206
207     if ((condition & G_IO_ERR)) {
208       GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
209           ("Socket in error state"));
210       *outbuf = NULL;
211       ret = GST_FLOW_ERROR;
212       goto done;
213     } else if ((condition & G_IO_HUP)) {
214       GST_DEBUG_OBJECT (src, "Connection closed");
215       *outbuf = NULL;
216       ret = GST_FLOW_EOS;
217       goto done;
218     }
219     avail = g_socket_get_available_bytes (src->socket);
220     if (avail < 0)
221       goto get_available_error;
222   }
223
224   if (avail > 0) {
225     read = MIN (avail, MAX_READ_SIZE);
226     *outbuf = gst_buffer_new_and_alloc (read);
227     data = gst_buffer_map (*outbuf, NULL, NULL, GST_MAP_READWRITE);
228     rret =
229         g_socket_receive (src->socket, (gchar *) data, read,
230         src->cancellable, &err);
231   } else {
232     /* Connection closed */
233     *outbuf = NULL;
234     data = NULL;
235     read = 0;
236     rret = 0;
237   }
238
239   if (rret == 0) {
240     GST_DEBUG_OBJECT (src, "Connection closed");
241     ret = GST_FLOW_EOS;
242     if (*outbuf) {
243       gst_buffer_unmap (*outbuf, data, read);
244       gst_buffer_unref (*outbuf);
245     }
246     *outbuf = NULL;
247   } else if (rret < 0) {
248     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
249       ret = GST_FLOW_WRONG_STATE;
250       GST_DEBUG_OBJECT (src, "Cancelled reading from socket");
251     } else {
252       ret = GST_FLOW_ERROR;
253       GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
254           ("Failed to read from socket: %s", err->message));
255     }
256     gst_buffer_unmap (*outbuf, data, read);
257     gst_buffer_unref (*outbuf);
258     *outbuf = NULL;
259   } else {
260     ret = GST_FLOW_OK;
261     gst_buffer_unmap (*outbuf, data, rret);
262
263     GST_LOG_OBJECT (src,
264         "Returning buffer from _get of size %" G_GSIZE_FORMAT ", ts %"
265         GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT
266         ", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT,
267         gst_buffer_get_size (*outbuf),
268         GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (*outbuf)),
269         GST_TIME_ARGS (GST_BUFFER_DURATION (*outbuf)),
270         GST_BUFFER_OFFSET (*outbuf), GST_BUFFER_OFFSET_END (*outbuf));
271   }
272   g_clear_error (&err);
273
274 done:
275   return ret;
276
277 select_error:
278   {
279     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
280         ("Select failed: %s", err->message));
281     g_clear_error (&err);
282     return GST_FLOW_ERROR;
283   }
284 get_available_error:
285   {
286     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL),
287         ("Failed to get available bytes from socket"));
288     return GST_FLOW_ERROR;
289   }
290 wrong_state:
291   {
292     GST_DEBUG_OBJECT (src, "connection to closed, cannot read data");
293     return GST_FLOW_WRONG_STATE;
294   }
295 }
296
297 static void
298 gst_tcp_client_src_set_property (GObject * object, guint prop_id,
299     const GValue * value, GParamSpec * pspec)
300 {
301   GstTCPClientSrc *tcpclientsrc = GST_TCP_CLIENT_SRC (object);
302
303   switch (prop_id) {
304     case PROP_HOST:
305       if (!g_value_get_string (value)) {
306         g_warning ("host property cannot be NULL");
307         break;
308       }
309       g_free (tcpclientsrc->host);
310       tcpclientsrc->host = g_strdup (g_value_get_string (value));
311       break;
312     case PROP_PORT:
313       tcpclientsrc->port = g_value_get_int (value);
314       break;
315
316     default:
317       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
318       break;
319   }
320 }
321
322 static void
323 gst_tcp_client_src_get_property (GObject * object, guint prop_id,
324     GValue * value, GParamSpec * pspec)
325 {
326   GstTCPClientSrc *tcpclientsrc = GST_TCP_CLIENT_SRC (object);
327
328   switch (prop_id) {
329     case PROP_HOST:
330       g_value_set_string (value, tcpclientsrc->host);
331       break;
332     case PROP_PORT:
333       g_value_set_int (value, tcpclientsrc->port);
334       break;
335     default:
336       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
337       break;
338   }
339 }
340
341 /* create a socket for connecting to remote server */
342 static gboolean
343 gst_tcp_client_src_start (GstBaseSrc * bsrc)
344 {
345   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
346   GError *err = NULL;
347   GInetAddress *addr;
348   GSocketAddress *saddr;
349   GResolver *resolver;
350
351   /* look up name if we need to */
352   addr = g_inet_address_new_from_string (src->host);
353   if (!addr) {
354     GList *results;
355
356     resolver = g_resolver_get_default ();
357
358     results =
359         g_resolver_lookup_by_name (resolver, src->host, src->cancellable, &err);
360     if (!results)
361       goto name_resolve;
362     addr = G_INET_ADDRESS (g_object_ref (results->data));
363
364     g_resolver_free_addresses (results);
365     g_object_unref (resolver);
366   }
367 #ifndef GST_DISABLE_GST_DEBUG
368   {
369     gchar *ip = g_inet_address_to_string (addr);
370
371     GST_DEBUG_OBJECT (src, "IP address for host %s is %s", src->host, ip);
372     g_free (ip);
373   }
374 #endif
375
376   saddr = g_inet_socket_address_new (addr, src->port);
377   g_object_unref (addr);
378
379   /* create receiving client socket */
380   GST_DEBUG_OBJECT (src, "opening receiving client socket to %s:%d",
381       src->host, src->port);
382
383   src->socket =
384       g_socket_new (g_socket_address_get_family (saddr), G_SOCKET_TYPE_STREAM,
385       G_SOCKET_PROTOCOL_TCP, &err);
386   if (!src->socket)
387     goto no_socket;
388
389   GST_DEBUG_OBJECT (src, "opened receiving client socket");
390   GST_OBJECT_FLAG_SET (src, GST_TCP_CLIENT_SRC_OPEN);
391
392   /* connect to server */
393   if (!g_socket_connect (src->socket, saddr, src->cancellable, &err))
394     goto connect_failed;
395
396   g_object_unref (saddr);
397
398   return TRUE;
399
400 no_socket:
401   {
402     GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
403         ("Failed to create socket: %s", err->message));
404     g_clear_error (&err);
405     g_object_unref (saddr);
406     return FALSE;
407   }
408 name_resolve:
409   {
410     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
411       GST_DEBUG_OBJECT (src, "Cancelled name resolval");
412     } else {
413       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
414           ("Failed to resolve host '%s': %s", src->host, err->message));
415     }
416     g_clear_error (&err);
417     g_object_unref (resolver);
418     return FALSE;
419   }
420 connect_failed:
421   {
422     if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
423       GST_DEBUG_OBJECT (src, "Cancelled connecting");
424     } else {
425       GST_ELEMENT_ERROR (src, RESOURCE, OPEN_READ, (NULL),
426           ("Failed to connect to host '%s:%d': %s", src->host, src->port,
427               err->message));
428     }
429     g_clear_error (&err);
430     g_object_unref (saddr);
431     gst_tcp_client_src_stop (GST_BASE_SRC (src));
432     return FALSE;
433   }
434 }
435
436 /* close the socket and associated resources
437  * unset OPEN flag
438  * used both to recover from errors and go to NULL state */
439 static gboolean
440 gst_tcp_client_src_stop (GstBaseSrc * bsrc)
441 {
442   GstTCPClientSrc *src;
443   GError *err = NULL;
444
445   src = GST_TCP_CLIENT_SRC (bsrc);
446
447   if (src->socket) {
448     GST_DEBUG_OBJECT (src, "closing socket");
449
450     if (!g_socket_close (src->socket, &err)) {
451       GST_ERROR_OBJECT (src, "Failed to close socket: %s", err->message);
452       g_clear_error (&err);
453     }
454     g_object_unref (src->socket);
455     src->socket = NULL;
456   }
457
458   GST_OBJECT_FLAG_UNSET (src, GST_TCP_CLIENT_SRC_OPEN);
459
460   return TRUE;
461 }
462
463 /* will be called only between calls to start() and stop() */
464 static gboolean
465 gst_tcp_client_src_unlock (GstBaseSrc * bsrc)
466 {
467   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
468
469   GST_DEBUG_OBJECT (src, "set to flushing");
470   g_cancellable_cancel (src->cancellable);
471
472   return TRUE;
473 }
474
475 /* will be called only between calls to start() and stop() */
476 static gboolean
477 gst_tcp_client_src_unlock_stop (GstBaseSrc * bsrc)
478 {
479   GstTCPClientSrc *src = GST_TCP_CLIENT_SRC (bsrc);
480
481   GST_DEBUG_OBJECT (src, "unset flushing");
482   g_cancellable_reset (src->cancellable);
483
484   return TRUE;
485 }