7852561cdc6c1bc1ede586f65a59a6131af4820d
[platform/upstream/gstreamer.git] / plugins / elements / gstfdsink.c
1 /* GStreamer
2  * Copyright (C) 1999,2000 Erik Walthinsen <omega@cse.ogi.edu>
3  *                    2000 Wim Taymans <wtay@chello.be>
4  *
5  * gstfdsink.c:
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-fdsink
25  * @see_also: #GstFdSrc
26  *
27  * Write data to a unix file descriptor.
28  *
29  * This element will synchronize on the clock before writing the data on the
30  * socket. For file descriptors where this does not make sense (files, ...) the
31  * #GstBaseSink:sync property can be used to disable synchronisation.
32  *
33  * Last reviewed on 2006-04-28 (0.10.6)
34  */
35
36 #ifdef HAVE_CONFIG_H
37 #  include "config.h"
38 #endif
39
40 #include "../../gst/gst-i18n-lib.h"
41
42 #include <sys/types.h>
43
44 #ifdef G_OS_WIN32
45 #include <io.h>                 /* lseek, open, close, read */
46 #undef lseek
47 #define lseek _lseeki64
48 #undef off_t
49 #define off_t guint64
50 #endif
51
52 #include <sys/stat.h>
53 #ifdef HAVE_SYS_SOCKET_H
54 #include <sys/socket.h>
55 #endif
56 #include <fcntl.h>
57 #include <stdio.h>
58 #ifdef HAVE_UNISTD_H
59 #include <unistd.h>
60 #endif
61 #ifdef _MSC_VER
62 #undef stat
63 #define stat _stat
64 #define fstat _fstat
65 #define S_ISREG(m)      (((m)&S_IFREG)==S_IFREG)
66 #endif
67 #include <errno.h>
68 #include <string.h>
69
70 #include "gstfdsink.h"
71
72 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
73     GST_PAD_SINK,
74     GST_PAD_ALWAYS,
75     GST_STATIC_CAPS_ANY);
76
77 GST_DEBUG_CATEGORY_STATIC (gst_fd_sink__debug);
78 #define GST_CAT_DEFAULT gst_fd_sink__debug
79
80
81 /* FdSink signals and args */
82 enum
83 {
84   /* FILL ME */
85   LAST_SIGNAL
86 };
87
88 enum
89 {
90   ARG_0,
91   ARG_FD
92 };
93
94 static void gst_fd_sink_uri_handler_init (gpointer g_iface,
95     gpointer iface_data);
96
97 #define _do_init \
98   G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_fd_sink_uri_handler_init); \
99   GST_DEBUG_CATEGORY_INIT (gst_fd_sink__debug, "fdsink", 0, "fdsink element");
100 #define gst_fd_sink_parent_class parent_class
101 G_DEFINE_TYPE_WITH_CODE (GstFdSink, gst_fd_sink, GST_TYPE_BASE_SINK, _do_init);
102
103 static void gst_fd_sink_set_property (GObject * object, guint prop_id,
104     const GValue * value, GParamSpec * pspec);
105 static void gst_fd_sink_get_property (GObject * object, guint prop_id,
106     GValue * value, GParamSpec * pspec);
107 static void gst_fd_sink_dispose (GObject * obj);
108
109 static gboolean gst_fd_sink_query (GstBaseSink * bsink, GstQuery * query);
110 static GstFlowReturn gst_fd_sink_render (GstBaseSink * sink,
111     GstBuffer * buffer);
112 static gboolean gst_fd_sink_start (GstBaseSink * basesink);
113 static gboolean gst_fd_sink_stop (GstBaseSink * basesink);
114 static gboolean gst_fd_sink_unlock (GstBaseSink * basesink);
115 static gboolean gst_fd_sink_unlock_stop (GstBaseSink * basesink);
116 static gboolean gst_fd_sink_event (GstBaseSink * sink, GstEvent * event);
117
118 static gboolean gst_fd_sink_do_seek (GstFdSink * fdsink, guint64 new_offset);
119
120 static void
121 gst_fd_sink_class_init (GstFdSinkClass * klass)
122 {
123   GObjectClass *gobject_class;
124   GstElementClass *gstelement_class;
125   GstBaseSinkClass *gstbasesink_class;
126
127   gobject_class = G_OBJECT_CLASS (klass);
128   gstelement_class = GST_ELEMENT_CLASS (klass);
129   gstbasesink_class = GST_BASE_SINK_CLASS (klass);
130
131   gobject_class->set_property = gst_fd_sink_set_property;
132   gobject_class->get_property = gst_fd_sink_get_property;
133   gobject_class->dispose = gst_fd_sink_dispose;
134
135   gst_element_class_set_static_metadata (gstelement_class,
136       "Filedescriptor Sink",
137       "Sink/File",
138       "Write data to a file descriptor", "Erik Walthinsen <omega@cse.ogi.edu>");
139   gst_element_class_add_pad_template (gstelement_class,
140       gst_static_pad_template_get (&sinktemplate));
141
142   gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_fd_sink_render);
143   gstbasesink_class->start = GST_DEBUG_FUNCPTR (gst_fd_sink_start);
144   gstbasesink_class->stop = GST_DEBUG_FUNCPTR (gst_fd_sink_stop);
145   gstbasesink_class->unlock = GST_DEBUG_FUNCPTR (gst_fd_sink_unlock);
146   gstbasesink_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_fd_sink_unlock_stop);
147   gstbasesink_class->event = GST_DEBUG_FUNCPTR (gst_fd_sink_event);
148   gstbasesink_class->query = GST_DEBUG_FUNCPTR (gst_fd_sink_query);
149
150   g_object_class_install_property (gobject_class, ARG_FD,
151       g_param_spec_int ("fd", "fd", "An open file descriptor to write to",
152           0, G_MAXINT, 1, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
153 }
154
155 static void
156 gst_fd_sink_init (GstFdSink * fdsink)
157 {
158   fdsink->fd = 1;
159   fdsink->uri = g_strdup_printf ("fd://%d", fdsink->fd);
160   fdsink->bytes_written = 0;
161   fdsink->current_pos = 0;
162
163   gst_base_sink_set_sync (GST_BASE_SINK (fdsink), FALSE);
164 }
165
166 static void
167 gst_fd_sink_dispose (GObject * obj)
168 {
169   GstFdSink *fdsink = GST_FD_SINK (obj);
170
171   g_free (fdsink->uri);
172   fdsink->uri = NULL;
173
174   G_OBJECT_CLASS (parent_class)->dispose (obj);
175 }
176
177 static gboolean
178 gst_fd_sink_query (GstBaseSink * bsink, GstQuery * query)
179 {
180   gboolean res = FALSE;
181   GstFdSink *fdsink;
182
183   fdsink = GST_FD_SINK (bsink);
184
185   switch (GST_QUERY_TYPE (query)) {
186     case GST_QUERY_POSITION:
187     {
188       GstFormat format;
189
190       gst_query_parse_position (query, &format, NULL);
191
192       switch (format) {
193         case GST_FORMAT_DEFAULT:
194         case GST_FORMAT_BYTES:
195           gst_query_set_position (query, GST_FORMAT_BYTES, fdsink->current_pos);
196           res = TRUE;
197           break;
198         default:
199           break;
200       }
201       break;
202     }
203     case GST_QUERY_FORMATS:
204       gst_query_set_formats (query, 2, GST_FORMAT_DEFAULT, GST_FORMAT_BYTES);
205       res = TRUE;
206       break;
207     case GST_QUERY_URI:
208       gst_query_set_uri (query, fdsink->uri);
209       res = TRUE;
210       break;
211     case GST_QUERY_SEEKING:{
212       GstFormat format;
213
214       gst_query_parse_seeking (query, &format, NULL, NULL, NULL);
215       if (format == GST_FORMAT_BYTES || format == GST_FORMAT_DEFAULT) {
216         gst_query_set_seeking (query, GST_FORMAT_BYTES, fdsink->seekable, 0,
217             -1);
218       } else {
219         gst_query_set_seeking (query, format, FALSE, 0, -1);
220       }
221       res = TRUE;
222       break;
223     }
224     default:
225       res = GST_BASE_SINK_CLASS (parent_class)->query (bsink, query);
226       break;
227
228   }
229   return res;
230 }
231
232 static GstFlowReturn
233 gst_fd_sink_render (GstBaseSink * sink, GstBuffer * buffer)
234 {
235   GstFdSink *fdsink;
236   GstMapInfo info;
237   guint8 *ptr;
238   gsize left;
239   gint written;
240
241 #ifndef HAVE_WIN32
242   gint retval;
243 #endif
244
245   fdsink = GST_FD_SINK (sink);
246
247   g_return_val_if_fail (fdsink->fd >= 0, GST_FLOW_ERROR);
248
249   gst_buffer_map (buffer, &info, GST_MAP_READ);
250
251   ptr = info.data;
252   left = info.size;
253
254 again:
255 #ifndef HAVE_WIN32
256   do {
257     GST_DEBUG_OBJECT (fdsink, "going into select, have %" G_GSIZE_FORMAT
258         " bytes to write", info.size);
259     retval = gst_poll_wait (fdsink->fdset, GST_CLOCK_TIME_NONE);
260   } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
261
262   if (retval == -1) {
263     if (errno == EBUSY)
264       goto stopped;
265     else
266       goto select_error;
267   }
268 #endif
269
270   GST_DEBUG_OBJECT (fdsink, "writing %" G_GSIZE_FORMAT " bytes to"
271       " file descriptor %d", info.size, fdsink->fd);
272
273   written = write (fdsink->fd, ptr, left);
274
275   /* check for errors */
276   if (G_UNLIKELY (written < 0)) {
277     /* try to write again on non-fatal errors */
278     if (errno == EAGAIN || errno == EINTR)
279       goto again;
280
281     /* else go to our error handler */
282     goto write_error;
283   }
284
285   /* all is fine when we get here */
286   left -= written;
287   ptr += written;
288   fdsink->bytes_written += written;
289   fdsink->current_pos += written;
290
291   GST_DEBUG_OBJECT (fdsink, "wrote %d bytes, %" G_GSIZE_FORMAT " left", written,
292       left);
293
294   /* short write, select and try to write the remainder */
295   if (G_UNLIKELY (left > 0))
296     goto again;
297
298   gst_buffer_unmap (buffer, &info);
299
300   return GST_FLOW_OK;
301
302 #ifndef HAVE_WIN32
303 select_error:
304   {
305     GST_ELEMENT_ERROR (fdsink, RESOURCE, READ, (NULL),
306         ("select on file descriptor: %s.", g_strerror (errno)));
307     GST_DEBUG_OBJECT (fdsink, "Error during select");
308     gst_buffer_unmap (buffer, &info);
309     return GST_FLOW_ERROR;
310   }
311 stopped:
312   {
313     GST_DEBUG_OBJECT (fdsink, "Select stopped");
314     gst_buffer_unmap (buffer, &info);
315     return GST_FLOW_FLUSHING;
316   }
317 #endif
318
319 write_error:
320   {
321     switch (errno) {
322       case ENOSPC:
323         GST_ELEMENT_ERROR (fdsink, RESOURCE, NO_SPACE_LEFT, (NULL), (NULL));
324         break;
325       default:{
326         GST_ELEMENT_ERROR (fdsink, RESOURCE, WRITE, (NULL),
327             ("Error while writing to file descriptor %d: %s",
328                 fdsink->fd, g_strerror (errno)));
329       }
330     }
331     gst_buffer_unmap (buffer, &info);
332     return GST_FLOW_ERROR;
333   }
334 }
335
336 static gboolean
337 gst_fd_sink_check_fd (GstFdSink * fdsink, int fd, GError ** error)
338 {
339   struct stat stat_results;
340   off_t result;
341
342   /* see that it is a valid file descriptor */
343   if (fstat (fd, &stat_results) < 0)
344     goto invalid;
345
346   if (!S_ISREG (stat_results.st_mode))
347     goto not_seekable;
348
349   /* see if it is a seekable stream */
350   result = lseek (fd, 0, SEEK_CUR);
351   if (result == -1) {
352     switch (errno) {
353       case EINVAL:
354       case EBADF:
355         goto invalid;
356
357       case ESPIPE:
358         goto not_seekable;
359     }
360   } else
361     GST_DEBUG_OBJECT (fdsink, "File descriptor %d is seekable", fd);
362
363   return TRUE;
364
365 invalid:
366   {
367     GST_ELEMENT_ERROR (fdsink, RESOURCE, WRITE, (NULL),
368         ("File descriptor %d is not valid: %s", fd, g_strerror (errno)));
369     g_set_error (error, GST_URI_ERROR, GST_URI_ERROR_BAD_REFERENCE,
370         "File descriptor %d is not valid: %s", fd, g_strerror (errno));
371     return FALSE;
372   }
373 not_seekable:
374   {
375     GST_DEBUG_OBJECT (fdsink, "File descriptor %d is a pipe", fd);
376     return TRUE;
377   }
378 }
379
380 static gboolean
381 gst_fd_sink_start (GstBaseSink * basesink)
382 {
383   GstFdSink *fdsink;
384   GstPollFD fd = GST_POLL_FD_INIT;
385
386   fdsink = GST_FD_SINK (basesink);
387   if (!gst_fd_sink_check_fd (fdsink, fdsink->fd, NULL))
388     return FALSE;
389
390   if ((fdsink->fdset = gst_poll_new (TRUE)) == NULL)
391     goto socket_pair;
392
393   fd.fd = fdsink->fd;
394   gst_poll_add_fd (fdsink->fdset, &fd);
395   gst_poll_fd_ctl_write (fdsink->fdset, &fd, TRUE);
396
397   fdsink->bytes_written = 0;
398   fdsink->current_pos = 0;
399
400   fdsink->seekable = gst_fd_sink_do_seek (fdsink, 0);
401   GST_INFO_OBJECT (fdsink, "seeking supported: %d", fdsink->seekable);
402
403   return TRUE;
404
405   /* ERRORS */
406 socket_pair:
407   {
408     GST_ELEMENT_ERROR (fdsink, RESOURCE, OPEN_READ_WRITE, (NULL),
409         GST_ERROR_SYSTEM);
410     return FALSE;
411   }
412 }
413
414 static gboolean
415 gst_fd_sink_stop (GstBaseSink * basesink)
416 {
417   GstFdSink *fdsink = GST_FD_SINK (basesink);
418
419   if (fdsink->fdset) {
420     gst_poll_free (fdsink->fdset);
421     fdsink->fdset = NULL;
422   }
423
424   return TRUE;
425 }
426
427 static gboolean
428 gst_fd_sink_unlock (GstBaseSink * basesink)
429 {
430   GstFdSink *fdsink = GST_FD_SINK (basesink);
431
432   GST_LOG_OBJECT (fdsink, "Flushing");
433   GST_OBJECT_LOCK (fdsink);
434   gst_poll_set_flushing (fdsink->fdset, TRUE);
435   GST_OBJECT_UNLOCK (fdsink);
436
437   return TRUE;
438 }
439
440 static gboolean
441 gst_fd_sink_unlock_stop (GstBaseSink * basesink)
442 {
443   GstFdSink *fdsink = GST_FD_SINK (basesink);
444
445   GST_LOG_OBJECT (fdsink, "No longer flushing");
446   GST_OBJECT_LOCK (fdsink);
447   gst_poll_set_flushing (fdsink->fdset, FALSE);
448   GST_OBJECT_UNLOCK (fdsink);
449
450   return TRUE;
451 }
452
453 static gboolean
454 gst_fd_sink_update_fd (GstFdSink * fdsink, int new_fd, GError ** error)
455 {
456   if (new_fd < 0) {
457     g_set_error (error, GST_URI_ERROR, GST_URI_ERROR_BAD_REFERENCE,
458         "File descriptor %d is not valid", new_fd);
459     return FALSE;
460   }
461
462   if (!gst_fd_sink_check_fd (fdsink, new_fd, error))
463     goto invalid;
464
465   /* assign the fd */
466   GST_OBJECT_LOCK (fdsink);
467   if (fdsink->fdset) {
468     GstPollFD fd = GST_POLL_FD_INIT;
469
470     fd.fd = fdsink->fd;
471     gst_poll_remove_fd (fdsink->fdset, &fd);
472
473     fd.fd = new_fd;
474     gst_poll_add_fd (fdsink->fdset, &fd);
475     gst_poll_fd_ctl_write (fdsink->fdset, &fd, TRUE);
476   }
477   fdsink->fd = new_fd;
478   g_free (fdsink->uri);
479   fdsink->uri = g_strdup_printf ("fd://%d", fdsink->fd);
480
481   GST_OBJECT_UNLOCK (fdsink);
482
483   return TRUE;
484
485 invalid:
486   {
487     return FALSE;
488   }
489 }
490
491 static void
492 gst_fd_sink_set_property (GObject * object, guint prop_id,
493     const GValue * value, GParamSpec * pspec)
494 {
495   GstFdSink *fdsink;
496
497   fdsink = GST_FD_SINK (object);
498
499   switch (prop_id) {
500     case ARG_FD:{
501       int fd;
502
503       fd = g_value_get_int (value);
504       gst_fd_sink_update_fd (fdsink, fd, NULL);
505       break;
506     }
507     default:
508       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
509       break;
510   }
511 }
512
513 static void
514 gst_fd_sink_get_property (GObject * object, guint prop_id, GValue * value,
515     GParamSpec * pspec)
516 {
517   GstFdSink *fdsink;
518
519   fdsink = GST_FD_SINK (object);
520
521   switch (prop_id) {
522     case ARG_FD:
523       g_value_set_int (value, fdsink->fd);
524       break;
525     default:
526       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
527       break;
528   }
529 }
530
531 static gboolean
532 gst_fd_sink_do_seek (GstFdSink * fdsink, guint64 new_offset)
533 {
534   off_t result;
535
536   result = lseek (fdsink->fd, new_offset, SEEK_SET);
537
538   if (result == -1)
539     goto seek_failed;
540
541   fdsink->current_pos = new_offset;
542
543   GST_DEBUG_OBJECT (fdsink, "File descriptor %d to seek to position "
544       "%" G_GUINT64_FORMAT, fdsink->fd, fdsink->current_pos);
545
546   return TRUE;
547
548   /* ERRORS */
549 seek_failed:
550   {
551     GST_DEBUG_OBJECT (fdsink, "File descriptor %d failed to seek to position "
552         "%" G_GUINT64_FORMAT, fdsink->fd, new_offset);
553     return FALSE;
554   }
555 }
556
557 static gboolean
558 gst_fd_sink_event (GstBaseSink * sink, GstEvent * event)
559 {
560   GstEventType type;
561   GstFdSink *fdsink;
562
563   fdsink = GST_FD_SINK (sink);
564
565   type = GST_EVENT_TYPE (event);
566
567   switch (type) {
568     case GST_EVENT_SEGMENT:
569     {
570       const GstSegment *segment;
571
572       gst_event_parse_segment (event, &segment);
573
574       if (segment->format == GST_FORMAT_BYTES) {
575         /* only try to seek and fail when we are going to a different
576          * position */
577         if (fdsink->current_pos != segment->start) {
578           /* FIXME, the seek should be performed on the pos field, start/stop are
579            * just boundaries for valid bytes offsets. We should also fill the file
580            * with zeroes if the new position extends the current EOF (sparse streams
581            * and segment accumulation). */
582           if (!gst_fd_sink_do_seek (fdsink, (guint64) segment->start))
583             goto seek_failed;
584         }
585       } else {
586         GST_DEBUG_OBJECT (fdsink,
587             "Ignored SEGMENT event of format %u (%s)", (guint) segment->format,
588             gst_format_get_name (segment->format));
589       }
590       break;
591     }
592     default:
593       break;
594   }
595
596   return GST_BASE_SINK_CLASS (parent_class)->event (sink, event);
597
598 seek_failed:
599   {
600     GST_ELEMENT_ERROR (fdsink, RESOURCE, SEEK, (NULL),
601         ("Error while seeking on file descriptor %d: %s",
602             fdsink->fd, g_strerror (errno)));
603     gst_event_unref (event);
604     return FALSE;
605   }
606
607 }
608
609 /*** GSTURIHANDLER INTERFACE *************************************************/
610
611 static GstURIType
612 gst_fd_sink_uri_get_type (GType type)
613 {
614   return GST_URI_SINK;
615 }
616
617 static const gchar *const *
618 gst_fd_sink_uri_get_protocols (GType type)
619 {
620   static const gchar *protocols[] = { "fd", NULL };
621
622   return protocols;
623 }
624
625 static gchar *
626 gst_fd_sink_uri_get_uri (GstURIHandler * handler)
627 {
628   GstFdSink *sink = GST_FD_SINK (handler);
629
630   /* FIXME: make thread-safe */
631   return g_strdup (sink->uri);
632 }
633
634 static gboolean
635 gst_fd_sink_uri_set_uri (GstURIHandler * handler, const gchar * uri,
636     GError ** error)
637 {
638   GstFdSink *sink = GST_FD_SINK (handler);
639   gint fd;
640
641   if (sscanf (uri, "fd://%d", &fd) != 1) {
642     g_set_error (error, GST_URI_ERROR, GST_URI_ERROR_BAD_URI,
643         "File descriptor URI could not be parsed");
644     return FALSE;
645   }
646
647   return gst_fd_sink_update_fd (sink, fd, error);
648 }
649
650 static void
651 gst_fd_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
652 {
653   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
654
655   iface->get_type = gst_fd_sink_uri_get_type;
656   iface->get_protocols = gst_fd_sink_uri_get_protocols;
657   iface->get_uri = gst_fd_sink_uri_get_uri;
658   iface->set_uri = gst_fd_sink_uri_set_uri;
659 }