Rework GstSegment handling
[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 (GstPad * pad, 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_details_simple (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
149   g_object_class_install_property (gobject_class, ARG_FD,
150       g_param_spec_int ("fd", "fd", "An open file descriptor to write to",
151           0, G_MAXINT, 1, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
152 }
153
154 static void
155 gst_fd_sink_init (GstFdSink * fdsink)
156 {
157   GstPad *pad;
158
159   pad = GST_BASE_SINK_PAD (fdsink);
160   gst_pad_set_query_function (pad, GST_DEBUG_FUNCPTR (gst_fd_sink_query));
161
162   fdsink->fd = 1;
163   fdsink->uri = g_strdup_printf ("fd://%d", fdsink->fd);
164   fdsink->bytes_written = 0;
165   fdsink->current_pos = 0;
166
167   gst_base_sink_set_sync (GST_BASE_SINK (fdsink), FALSE);
168 }
169
170 static void
171 gst_fd_sink_dispose (GObject * obj)
172 {
173   GstFdSink *fdsink = GST_FD_SINK (obj);
174
175   g_free (fdsink->uri);
176   fdsink->uri = NULL;
177
178   G_OBJECT_CLASS (parent_class)->dispose (obj);
179 }
180
181 static gboolean
182 gst_fd_sink_query (GstPad * pad, GstQuery ** query)
183 {
184   GstFdSink *fdsink;
185   GstFormat format;
186
187   fdsink = GST_FD_SINK (GST_PAD_PARENT (pad));
188
189   switch (GST_QUERY_TYPE (*query)) {
190     case GST_QUERY_POSITION:
191       gst_query_parse_position (*query, &format, NULL);
192       switch (format) {
193         case GST_FORMAT_DEFAULT:
194         case GST_FORMAT_BYTES:
195           gst_query_set_position (*query, GST_FORMAT_BYTES,
196               fdsink->current_pos);
197           return TRUE;
198         default:
199           return FALSE;
200       }
201
202     case GST_QUERY_FORMATS:
203       gst_query_set_formats (*query, 2, GST_FORMAT_DEFAULT, GST_FORMAT_BYTES);
204       return TRUE;
205
206     case GST_QUERY_URI:
207       gst_query_set_uri (*query, fdsink->uri);
208       return TRUE;
209
210     default:
211       return gst_pad_query_default (pad, query);
212   }
213 }
214
215 static GstFlowReturn
216 gst_fd_sink_render (GstBaseSink * sink, GstBuffer * buffer)
217 {
218   GstFdSink *fdsink;
219   guint8 *data, *ptr;
220   gsize size, left;
221   gint written;
222
223 #ifndef HAVE_WIN32
224   gint retval;
225 #endif
226
227   fdsink = GST_FD_SINK (sink);
228
229   g_return_val_if_fail (fdsink->fd >= 0, GST_FLOW_ERROR);
230
231   data = gst_buffer_map (buffer, &size, NULL, GST_MAP_READ);
232
233   ptr = data;
234   left = size;
235
236 again:
237 #ifndef HAVE_WIN32
238   do {
239     GST_DEBUG_OBJECT (fdsink, "going into select, have %d bytes to write",
240         size);
241     retval = gst_poll_wait (fdsink->fdset, GST_CLOCK_TIME_NONE);
242   } while (retval == -1 && (errno == EINTR || errno == EAGAIN));
243
244   if (retval == -1) {
245     if (errno == EBUSY)
246       goto stopped;
247     else
248       goto select_error;
249   }
250 #endif
251
252   GST_DEBUG_OBJECT (fdsink, "writing %d bytes to file descriptor %d", size,
253       fdsink->fd);
254
255   written = write (fdsink->fd, ptr, left);
256
257   /* check for errors */
258   if (G_UNLIKELY (written < 0)) {
259     /* try to write again on non-fatal errors */
260     if (errno == EAGAIN || errno == EINTR)
261       goto again;
262
263     /* else go to our error handler */
264     goto write_error;
265   }
266
267   /* all is fine when we get here */
268   left -= written;
269   ptr += written;
270   fdsink->bytes_written += written;
271   fdsink->current_pos += written;
272
273   GST_DEBUG_OBJECT (fdsink, "wrote %d bytes, %" G_GSIZE_FORMAT " left", written,
274       left);
275
276   /* short write, select and try to write the remainder */
277   if (G_UNLIKELY (left > 0))
278     goto again;
279
280   gst_buffer_unmap (buffer, data, size);
281
282   return GST_FLOW_OK;
283
284 #ifndef HAVE_WIN32
285 select_error:
286   {
287     GST_ELEMENT_ERROR (fdsink, RESOURCE, READ, (NULL),
288         ("select on file descriptor: %s.", g_strerror (errno)));
289     GST_DEBUG_OBJECT (fdsink, "Error during select");
290     gst_buffer_unmap (buffer, data, size);
291     return GST_FLOW_ERROR;
292   }
293 stopped:
294   {
295     GST_DEBUG_OBJECT (fdsink, "Select stopped");
296     gst_buffer_unmap (buffer, data, size);
297     return GST_FLOW_WRONG_STATE;
298   }
299 #endif
300
301 write_error:
302   {
303     switch (errno) {
304       case ENOSPC:
305         GST_ELEMENT_ERROR (fdsink, RESOURCE, NO_SPACE_LEFT, (NULL), (NULL));
306         break;
307       default:{
308         GST_ELEMENT_ERROR (fdsink, RESOURCE, WRITE, (NULL),
309             ("Error while writing to file descriptor %d: %s",
310                 fdsink->fd, g_strerror (errno)));
311       }
312     }
313     gst_buffer_unmap (buffer, data, size);
314     return GST_FLOW_ERROR;
315   }
316 }
317
318 static gboolean
319 gst_fd_sink_check_fd (GstFdSink * fdsink, int fd)
320 {
321   struct stat stat_results;
322   off_t result;
323
324   /* see that it is a valid file descriptor */
325   if (fstat (fd, &stat_results) < 0)
326     goto invalid;
327
328   if (!S_ISREG (stat_results.st_mode))
329     goto not_seekable;
330
331   /* see if it is a seekable stream */
332   result = lseek (fd, 0, SEEK_CUR);
333   if (result == -1) {
334     switch (errno) {
335       case EINVAL:
336       case EBADF:
337         goto invalid;
338
339       case ESPIPE:
340         goto not_seekable;
341     }
342   } else
343     GST_DEBUG_OBJECT (fdsink, "File descriptor %d is seekable", fd);
344
345   return TRUE;
346
347 invalid:
348   {
349     GST_ELEMENT_ERROR (fdsink, RESOURCE, WRITE, (NULL),
350         ("File descriptor %d is not valid: %s", fd, g_strerror (errno)));
351     return FALSE;
352   }
353 not_seekable:
354   {
355     GST_DEBUG_OBJECT (fdsink, "File descriptor %d is a pipe", fd);
356     return TRUE;
357   }
358 }
359
360 static gboolean
361 gst_fd_sink_start (GstBaseSink * basesink)
362 {
363   GstFdSink *fdsink;
364   GstPollFD fd = GST_POLL_FD_INIT;
365
366   fdsink = GST_FD_SINK (basesink);
367   if (!gst_fd_sink_check_fd (fdsink, fdsink->fd))
368     return FALSE;
369
370   if ((fdsink->fdset = gst_poll_new (TRUE)) == NULL)
371     goto socket_pair;
372
373   fd.fd = fdsink->fd;
374   gst_poll_add_fd (fdsink->fdset, &fd);
375   gst_poll_fd_ctl_write (fdsink->fdset, &fd, TRUE);
376
377   fdsink->bytes_written = 0;
378   fdsink->current_pos = 0;
379
380   return TRUE;
381
382   /* ERRORS */
383 socket_pair:
384   {
385     GST_ELEMENT_ERROR (fdsink, RESOURCE, OPEN_READ_WRITE, (NULL),
386         GST_ERROR_SYSTEM);
387     return FALSE;
388   }
389 }
390
391 static gboolean
392 gst_fd_sink_stop (GstBaseSink * basesink)
393 {
394   GstFdSink *fdsink = GST_FD_SINK (basesink);
395
396   if (fdsink->fdset) {
397     gst_poll_free (fdsink->fdset);
398     fdsink->fdset = NULL;
399   }
400
401   return TRUE;
402 }
403
404 static gboolean
405 gst_fd_sink_unlock (GstBaseSink * basesink)
406 {
407   GstFdSink *fdsink = GST_FD_SINK (basesink);
408
409   GST_LOG_OBJECT (fdsink, "Flushing");
410   GST_OBJECT_LOCK (fdsink);
411   gst_poll_set_flushing (fdsink->fdset, TRUE);
412   GST_OBJECT_UNLOCK (fdsink);
413
414   return TRUE;
415 }
416
417 static gboolean
418 gst_fd_sink_unlock_stop (GstBaseSink * basesink)
419 {
420   GstFdSink *fdsink = GST_FD_SINK (basesink);
421
422   GST_LOG_OBJECT (fdsink, "No longer flushing");
423   GST_OBJECT_LOCK (fdsink);
424   gst_poll_set_flushing (fdsink->fdset, FALSE);
425   GST_OBJECT_UNLOCK (fdsink);
426
427   return TRUE;
428 }
429
430 static gboolean
431 gst_fd_sink_update_fd (GstFdSink * fdsink, int new_fd)
432 {
433   if (new_fd < 0)
434     return FALSE;
435
436   if (!gst_fd_sink_check_fd (fdsink, new_fd))
437     goto invalid;
438
439   /* assign the fd */
440   GST_OBJECT_LOCK (fdsink);
441   if (fdsink->fdset) {
442     GstPollFD fd = GST_POLL_FD_INIT;
443
444     fd.fd = fdsink->fd;
445     gst_poll_remove_fd (fdsink->fdset, &fd);
446
447     fd.fd = new_fd;
448     gst_poll_add_fd (fdsink->fdset, &fd);
449     gst_poll_fd_ctl_write (fdsink->fdset, &fd, TRUE);
450   }
451   fdsink->fd = new_fd;
452   g_free (fdsink->uri);
453   fdsink->uri = g_strdup_printf ("fd://%d", fdsink->fd);
454
455   GST_OBJECT_UNLOCK (fdsink);
456
457   return TRUE;
458
459 invalid:
460   {
461     return FALSE;
462   }
463 }
464
465 static void
466 gst_fd_sink_set_property (GObject * object, guint prop_id,
467     const GValue * value, GParamSpec * pspec)
468 {
469   GstFdSink *fdsink;
470
471   fdsink = GST_FD_SINK (object);
472
473   switch (prop_id) {
474     case ARG_FD:{
475       int fd;
476
477       fd = g_value_get_int (value);
478       gst_fd_sink_update_fd (fdsink, fd);
479       break;
480     }
481     default:
482       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
483       break;
484   }
485 }
486
487 static void
488 gst_fd_sink_get_property (GObject * object, guint prop_id, GValue * value,
489     GParamSpec * pspec)
490 {
491   GstFdSink *fdsink;
492
493   fdsink = GST_FD_SINK (object);
494
495   switch (prop_id) {
496     case ARG_FD:
497       g_value_set_int (value, fdsink->fd);
498       break;
499     default:
500       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
501       break;
502   }
503 }
504
505 static gboolean
506 gst_fd_sink_do_seek (GstFdSink * fdsink, guint64 new_offset)
507 {
508   off_t result;
509
510   result = lseek (fdsink->fd, new_offset, SEEK_SET);
511
512   if (result == -1)
513     goto seek_failed;
514
515   fdsink->current_pos = new_offset;
516
517   GST_DEBUG_OBJECT (fdsink, "File descriptor %d to seek to position "
518       "%" G_GUINT64_FORMAT, fdsink->fd, fdsink->current_pos);
519
520   return TRUE;
521
522   /* ERRORS */
523 seek_failed:
524   {
525     GST_DEBUG_OBJECT (fdsink, "File descriptor %d failed to seek to position "
526         "%" G_GUINT64_FORMAT, fdsink->fd, new_offset);
527     return FALSE;
528   }
529 }
530
531 static gboolean
532 gst_fd_sink_event (GstBaseSink * sink, GstEvent * event)
533 {
534   GstEventType type;
535   GstFdSink *fdsink;
536
537   fdsink = GST_FD_SINK (sink);
538
539   type = GST_EVENT_TYPE (event);
540
541   switch (type) {
542     case GST_EVENT_SEGMENT:
543     {
544       GstSegment segment;
545
546       gst_event_parse_segment (event, &segment);
547
548       if (segment.format == GST_FORMAT_BYTES) {
549         /* only try to seek and fail when we are going to a different
550          * position */
551         if (fdsink->current_pos != segment.start) {
552           /* FIXME, the seek should be performed on the pos field, start/stop are
553            * just boundaries for valid bytes offsets. We should also fill the file
554            * with zeroes if the new position extends the current EOF (sparse streams
555            * and segment accumulation). */
556           if (!gst_fd_sink_do_seek (fdsink, (guint64) segment.start))
557             goto seek_failed;
558         }
559       } else {
560         GST_DEBUG_OBJECT (fdsink,
561             "Ignored SEGMENT event of format %u (%s)", (guint) segment.format,
562             gst_format_get_name (segment.format));
563       }
564       break;
565     }
566     default:
567       break;
568   }
569
570   return TRUE;
571
572 seek_failed:
573   {
574     GST_ELEMENT_ERROR (fdsink, RESOURCE, SEEK, (NULL),
575         ("Error while seeking on file descriptor %d: %s",
576             fdsink->fd, g_strerror (errno)));
577     return FALSE;
578   }
579
580 }
581
582 /*** GSTURIHANDLER INTERFACE *************************************************/
583
584 static GstURIType
585 gst_fd_sink_uri_get_type (void)
586 {
587   return GST_URI_SINK;
588 }
589
590 static gchar **
591 gst_fd_sink_uri_get_protocols (void)
592 {
593   static gchar *protocols[] = { (char *) "fd", NULL };
594
595   return protocols;
596 }
597
598 static const gchar *
599 gst_fd_sink_uri_get_uri (GstURIHandler * handler)
600 {
601   GstFdSink *sink = GST_FD_SINK (handler);
602
603   return sink->uri;
604 }
605
606 static gboolean
607 gst_fd_sink_uri_set_uri (GstURIHandler * handler, const gchar * uri)
608 {
609   gchar *protocol;
610   GstFdSink *sink = GST_FD_SINK (handler);
611   gint fd;
612
613   protocol = gst_uri_get_protocol (uri);
614   if (strcmp (protocol, "fd") != 0) {
615     g_free (protocol);
616     return FALSE;
617   }
618   g_free (protocol);
619
620   if (sscanf (uri, "fd://%d", &fd) != 1)
621     return FALSE;
622
623   return gst_fd_sink_update_fd (sink, fd);
624 }
625
626 static void
627 gst_fd_sink_uri_handler_init (gpointer g_iface, gpointer iface_data)
628 {
629   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
630
631   iface->get_type = gst_fd_sink_uri_get_type;
632   iface->get_protocols = gst_fd_sink_uri_get_protocols;
633   iface->get_uri = gst_fd_sink_uri_get_uri;
634   iface->set_uri = gst_fd_sink_uri_set_uri;
635 }