1 /* GIO - GLib Input, Output and Streaming Library
3 * Copyright (C) 2006-2007 Red Hat, Inc.
5 * This library is free software; you can redistribute it and/or
6 * modify it under the terms of the GNU Lesser General Public
7 * License as published by the Free Software Foundation; either
8 * version 2 of the License, or (at your option) any later version.
10 * This library is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 * Lesser General Public License for more details.
15 * You should have received a copy of the GNU Lesser General
16 * Public License along with this library; if not, write to the
17 * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
18 * Boston, MA 02111-1307, USA.
20 * Author: Christian Kellner <gicmo@gnome.org>
24 #include "gbufferedoutputstream.h"
25 #include "goutputstream.h"
26 #include "gseekable.h"
33 * SECTION:gbufferedoutputstream
34 * @short_description: Buffered Output Stream
36 * @see_also: #GFilterOutputStream, #GOutputStream
38 * Buffered output stream implements #GFilterOutputStream and provides
39 * for buffered writes.
41 * By default, #GBufferedOutputStream's buffer size is set at 4 kilobytes.
43 * To create a buffered output stream, use g_buffered_output_stream_new(),
44 * or g_buffered_output_stream_new_sized() to specify the buffer's size
47 * To get the size of a buffer within a buffered input stream, use
48 * g_buffered_output_stream_get_buffer_size(). To change the size of a
49 * buffered output stream's buffer, use
50 * g_buffered_output_stream_set_buffer_size(). Note that the buffer's
51 * size cannot be reduced below the size of the data within the buffer.
54 #define DEFAULT_BUFFER_SIZE 4096
56 struct _GBufferedOutputStreamPrivate
{
69 static void g_buffered_output_stream_set_property (GObject
*object
,
74 static void g_buffered_output_stream_get_property (GObject
*object
,
78 static void g_buffered_output_stream_finalize (GObject
*object
);
81 static gssize
g_buffered_output_stream_write (GOutputStream
*stream
,
84 GCancellable
*cancellable
,
86 static gboolean
g_buffered_output_stream_flush (GOutputStream
*stream
,
87 GCancellable
*cancellable
,
89 static gboolean
g_buffered_output_stream_close (GOutputStream
*stream
,
90 GCancellable
*cancellable
,
93 static void g_buffered_output_stream_flush_async (GOutputStream
*stream
,
95 GCancellable
*cancellable
,
96 GAsyncReadyCallback callback
,
98 static gboolean
g_buffered_output_stream_flush_finish (GOutputStream
*stream
,
101 static void g_buffered_output_stream_close_async (GOutputStream
*stream
,
103 GCancellable
*cancellable
,
104 GAsyncReadyCallback callback
,
106 static gboolean
g_buffered_output_stream_close_finish (GOutputStream
*stream
,
107 GAsyncResult
*result
,
110 static void g_buffered_output_stream_seekable_iface_init (GSeekableIface
*iface
);
111 static goffset
g_buffered_output_stream_tell (GSeekable
*seekable
);
112 static gboolean
g_buffered_output_stream_can_seek (GSeekable
*seekable
);
113 static gboolean
g_buffered_output_stream_seek (GSeekable
*seekable
,
116 GCancellable
*cancellable
,
118 static gboolean
g_buffered_output_stream_can_truncate (GSeekable
*seekable
);
119 static gboolean
g_buffered_output_stream_truncate (GSeekable
*seekable
,
121 GCancellable
*cancellable
,
124 G_DEFINE_TYPE_WITH_CODE (GBufferedOutputStream
,
125 g_buffered_output_stream
,
126 G_TYPE_FILTER_OUTPUT_STREAM
,
127 G_ADD_PRIVATE (GBufferedOutputStream
)
128 G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE
,
129 g_buffered_output_stream_seekable_iface_init
))
133 g_buffered_output_stream_class_init (GBufferedOutputStreamClass
*klass
)
135 GObjectClass
*object_class
;
136 GOutputStreamClass
*ostream_class
;
138 object_class
= G_OBJECT_CLASS (klass
);
139 object_class
->get_property
= g_buffered_output_stream_get_property
;
140 object_class
->set_property
= g_buffered_output_stream_set_property
;
141 object_class
->finalize
= g_buffered_output_stream_finalize
;
143 ostream_class
= G_OUTPUT_STREAM_CLASS (klass
);
144 ostream_class
->write_fn
= g_buffered_output_stream_write
;
145 ostream_class
->flush
= g_buffered_output_stream_flush
;
146 ostream_class
->close_fn
= g_buffered_output_stream_close
;
147 ostream_class
->flush_async
= g_buffered_output_stream_flush_async
;
148 ostream_class
->flush_finish
= g_buffered_output_stream_flush_finish
;
149 ostream_class
->close_async
= g_buffered_output_stream_close_async
;
150 ostream_class
->close_finish
= g_buffered_output_stream_close_finish
;
152 g_object_class_install_property (object_class
,
154 g_param_spec_uint ("buffer-size",
156 P_("The size of the backend buffer"),
160 G_PARAM_READWRITE
|G_PARAM_CONSTRUCT
|
161 G_PARAM_STATIC_NAME
|G_PARAM_STATIC_NICK
|G_PARAM_STATIC_BLURB
));
163 g_object_class_install_property (object_class
,
165 g_param_spec_boolean ("auto-grow",
167 P_("Whether the buffer should automatically grow"),
170 G_PARAM_STATIC_NAME
|G_PARAM_STATIC_NICK
|G_PARAM_STATIC_BLURB
));
175 * g_buffered_output_stream_get_buffer_size:
176 * @stream: a #GBufferedOutputStream.
178 * Gets the size of the buffer in the @stream.
180 * Returns: the current size of the buffer.
183 g_buffered_output_stream_get_buffer_size (GBufferedOutputStream
*stream
)
185 g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream
), -1);
187 return stream
->priv
->len
;
191 * g_buffered_output_stream_set_buffer_size:
192 * @stream: a #GBufferedOutputStream.
195 * Sets the size of the internal buffer to @size.
198 g_buffered_output_stream_set_buffer_size (GBufferedOutputStream
*stream
,
201 GBufferedOutputStreamPrivate
*priv
;
204 g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream
));
208 if (size
== priv
->len
)
213 size
= MAX (size
, priv
->pos
);
215 buffer
= g_malloc (size
);
216 memcpy (buffer
, priv
->buffer
, priv
->pos
);
217 g_free (priv
->buffer
);
218 priv
->buffer
= buffer
;
224 priv
->buffer
= g_malloc (size
);
229 g_object_notify (G_OBJECT (stream
), "buffer-size");
233 * g_buffered_output_stream_get_auto_grow:
234 * @stream: a #GBufferedOutputStream.
236 * Checks if the buffer automatically grows as data is added.
238 * Returns: %TRUE if the @stream's buffer automatically grows,
242 g_buffered_output_stream_get_auto_grow (GBufferedOutputStream
*stream
)
244 g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream
), FALSE
);
246 return stream
->priv
->auto_grow
;
250 * g_buffered_output_stream_set_auto_grow:
251 * @stream: a #GBufferedOutputStream.
252 * @auto_grow: a #gboolean.
254 * Sets whether or not the @stream's buffer should automatically grow.
255 * If @auto_grow is true, then each write will just make the buffer
256 * larger, and you must manually flush the buffer to actually write out
257 * the data to the underlying stream.
260 g_buffered_output_stream_set_auto_grow (GBufferedOutputStream
*stream
,
263 GBufferedOutputStreamPrivate
*priv
;
264 g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream
));
266 auto_grow
= auto_grow
!= FALSE
;
267 if (priv
->auto_grow
!= auto_grow
)
269 priv
->auto_grow
= auto_grow
;
270 g_object_notify (G_OBJECT (stream
), "auto-grow");
275 g_buffered_output_stream_set_property (GObject
*object
,
280 GBufferedOutputStream
*stream
;
282 stream
= G_BUFFERED_OUTPUT_STREAM (object
);
287 g_buffered_output_stream_set_buffer_size (stream
, g_value_get_uint (value
));
291 g_buffered_output_stream_set_auto_grow (stream
, g_value_get_boolean (value
));
295 G_OBJECT_WARN_INVALID_PROPERTY_ID (object
, prop_id
, pspec
);
302 g_buffered_output_stream_get_property (GObject
*object
,
307 GBufferedOutputStream
*buffered_stream
;
308 GBufferedOutputStreamPrivate
*priv
;
310 buffered_stream
= G_BUFFERED_OUTPUT_STREAM (object
);
311 priv
= buffered_stream
->priv
;
316 g_value_set_uint (value
, priv
->len
);
320 g_value_set_boolean (value
, priv
->auto_grow
);
324 G_OBJECT_WARN_INVALID_PROPERTY_ID (object
, prop_id
, pspec
);
331 g_buffered_output_stream_finalize (GObject
*object
)
333 GBufferedOutputStream
*stream
;
334 GBufferedOutputStreamPrivate
*priv
;
336 stream
= G_BUFFERED_OUTPUT_STREAM (object
);
339 g_free (priv
->buffer
);
341 G_OBJECT_CLASS (g_buffered_output_stream_parent_class
)->finalize (object
);
345 g_buffered_output_stream_init (GBufferedOutputStream
*stream
)
347 stream
->priv
= g_buffered_output_stream_get_instance_private (stream
);
351 g_buffered_output_stream_seekable_iface_init (GSeekableIface
*iface
)
353 iface
->tell
= g_buffered_output_stream_tell
;
354 iface
->can_seek
= g_buffered_output_stream_can_seek
;
355 iface
->seek
= g_buffered_output_stream_seek
;
356 iface
->can_truncate
= g_buffered_output_stream_can_truncate
;
357 iface
->truncate_fn
= g_buffered_output_stream_truncate
;
361 * g_buffered_output_stream_new:
362 * @base_stream: a #GOutputStream.
364 * Creates a new buffered output stream for a base stream.
366 * Returns: a #GOutputStream for the given @base_stream.
369 g_buffered_output_stream_new (GOutputStream
*base_stream
)
371 GOutputStream
*stream
;
373 g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream
), NULL
);
375 stream
= g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM
,
376 "base-stream", base_stream
,
383 * g_buffered_output_stream_new_sized:
384 * @base_stream: a #GOutputStream.
387 * Creates a new buffered output stream with a given buffer size.
389 * Returns: a #GOutputStream with an internal buffer set to @size.
392 g_buffered_output_stream_new_sized (GOutputStream
*base_stream
,
395 GOutputStream
*stream
;
397 g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream
), NULL
);
399 stream
= g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM
,
400 "base-stream", base_stream
,
408 flush_buffer (GBufferedOutputStream
*stream
,
409 GCancellable
*cancellable
,
412 GBufferedOutputStreamPrivate
*priv
;
413 GOutputStream
*base_stream
;
420 base_stream
= G_FILTER_OUTPUT_STREAM (stream
)->base_stream
;
422 g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream
), FALSE
);
424 res
= g_output_stream_write_all (base_stream
,
431 count
= priv
->pos
- bytes_written
;
434 g_memmove (priv
->buffer
, priv
->buffer
+ bytes_written
, count
);
436 priv
->pos
-= bytes_written
;
442 g_buffered_output_stream_write (GOutputStream
*stream
,
445 GCancellable
*cancellable
,
448 GBufferedOutputStream
*bstream
;
449 GBufferedOutputStreamPrivate
*priv
;
454 bstream
= G_BUFFERED_OUTPUT_STREAM (stream
);
455 priv
= bstream
->priv
;
457 n
= priv
->len
- priv
->pos
;
459 if (priv
->auto_grow
&& n
< count
)
461 new_size
= MAX (priv
->len
* 2, priv
->len
+ count
);
462 g_buffered_output_stream_set_buffer_size (bstream
, new_size
);
466 res
= flush_buffer (bstream
, cancellable
, error
);
472 n
= priv
->len
- priv
->pos
;
474 count
= MIN (count
, n
);
475 memcpy (priv
->buffer
+ priv
->pos
, buffer
, count
);
482 g_buffered_output_stream_flush (GOutputStream
*stream
,
483 GCancellable
*cancellable
,
486 GBufferedOutputStream
*bstream
;
487 GOutputStream
*base_stream
;
490 bstream
= G_BUFFERED_OUTPUT_STREAM (stream
);
491 base_stream
= G_FILTER_OUTPUT_STREAM (stream
)->base_stream
;
493 res
= flush_buffer (bstream
, cancellable
, error
);
498 res
= g_output_stream_flush (base_stream
, cancellable
, error
);
504 g_buffered_output_stream_close (GOutputStream
*stream
,
505 GCancellable
*cancellable
,
508 GBufferedOutputStream
*bstream
;
509 GOutputStream
*base_stream
;
512 bstream
= G_BUFFERED_OUTPUT_STREAM (stream
);
513 base_stream
= G_FILTER_OUTPUT_STREAM (bstream
)->base_stream
;
514 res
= flush_buffer (bstream
, cancellable
, error
);
516 if (g_filter_output_stream_get_close_base_stream (G_FILTER_OUTPUT_STREAM (stream
)))
518 /* report the first error but still close the stream */
520 res
= g_output_stream_close (base_stream
, cancellable
, error
);
522 g_output_stream_close (base_stream
, cancellable
, NULL
);
529 g_buffered_output_stream_tell (GSeekable
*seekable
)
531 GBufferedOutputStream
*bstream
;
532 GBufferedOutputStreamPrivate
*priv
;
533 GOutputStream
*base_stream
;
534 GSeekable
*base_stream_seekable
;
537 bstream
= G_BUFFERED_OUTPUT_STREAM (seekable
);
538 priv
= bstream
->priv
;
540 base_stream
= G_FILTER_OUTPUT_STREAM (seekable
)->base_stream
;
541 if (!G_IS_SEEKABLE (base_stream
))
544 base_stream_seekable
= G_SEEKABLE (base_stream
);
546 base_offset
= g_seekable_tell (base_stream_seekable
);
547 return base_offset
+ priv
->pos
;
551 g_buffered_output_stream_can_seek (GSeekable
*seekable
)
553 GOutputStream
*base_stream
;
555 base_stream
= G_FILTER_OUTPUT_STREAM (seekable
)->base_stream
;
556 return G_IS_SEEKABLE (base_stream
) && g_seekable_can_seek (G_SEEKABLE (base_stream
));
560 g_buffered_output_stream_seek (GSeekable
*seekable
,
563 GCancellable
*cancellable
,
566 GBufferedOutputStream
*bstream
;
567 GOutputStream
*base_stream
;
568 GSeekable
*base_stream_seekable
;
571 bstream
= G_BUFFERED_OUTPUT_STREAM (seekable
);
573 base_stream
= G_FILTER_OUTPUT_STREAM (seekable
)->base_stream
;
574 if (!G_IS_SEEKABLE (base_stream
))
576 g_set_error_literal (error
, G_IO_ERROR
, G_IO_ERROR_NOT_SUPPORTED
,
577 _("Seek not supported on base stream"));
581 base_stream_seekable
= G_SEEKABLE (base_stream
);
582 flushed
= flush_buffer (bstream
, cancellable
, error
);
586 return g_seekable_seek (base_stream_seekable
, offset
, type
, cancellable
, error
);
590 g_buffered_output_stream_can_truncate (GSeekable
*seekable
)
592 GOutputStream
*base_stream
;
594 base_stream
= G_FILTER_OUTPUT_STREAM (seekable
)->base_stream
;
595 return G_IS_SEEKABLE (base_stream
) && g_seekable_can_truncate (G_SEEKABLE (base_stream
));
599 g_buffered_output_stream_truncate (GSeekable
*seekable
,
601 GCancellable
*cancellable
,
604 GBufferedOutputStream
*bstream
;
605 GOutputStream
*base_stream
;
606 GSeekable
*base_stream_seekable
;
609 bstream
= G_BUFFERED_OUTPUT_STREAM (seekable
);
610 base_stream
= G_FILTER_OUTPUT_STREAM (seekable
)->base_stream
;
611 if (!G_IS_SEEKABLE (base_stream
))
613 g_set_error_literal (error
, G_IO_ERROR
, G_IO_ERROR_NOT_SUPPORTED
,
614 _("Truncate not supported on base stream"));
618 base_stream_seekable
= G_SEEKABLE (base_stream
);
620 flushed
= flush_buffer (bstream
, cancellable
, error
);
623 return g_seekable_truncate (base_stream_seekable
, offset
, cancellable
, error
);
626 /* ************************** */
627 /* Async stuff implementation */
628 /* ************************** */
630 /* TODO: This should be using the base class async ops, not threads */
634 guint flush_stream
: 1;
635 guint close_stream
: 1;
640 free_flush_data (gpointer data
)
642 g_slice_free (FlushData
, data
);
645 /* This function is used by all three (i.e.
646 * _write, _flush, _close) functions since
647 * all of them will need to flush the buffer
648 * and so closing and writing is just a special
649 * case of flushing + some addition stuff */
651 flush_buffer_thread (GTask
*task
,
654 GCancellable
*cancellable
)
656 GBufferedOutputStream
*stream
;
657 GOutputStream
*base_stream
;
660 GError
*error
= NULL
;
662 stream
= G_BUFFERED_OUTPUT_STREAM (object
);
664 base_stream
= G_FILTER_OUTPUT_STREAM (stream
)->base_stream
;
666 res
= flush_buffer (stream
, cancellable
, &error
);
668 /* if flushing the buffer didn't work don't even bother
669 * to flush the stream but just report that error */
670 if (res
&& fdata
->flush_stream
)
671 res
= g_output_stream_flush (base_stream
, cancellable
, &error
);
673 if (fdata
->close_stream
)
676 /* if flushing the buffer or the stream returned
677 * an error report that first error but still try
678 * close the stream */
679 if (g_filter_output_stream_get_close_base_stream (G_FILTER_OUTPUT_STREAM (stream
)))
682 g_output_stream_close (base_stream
, cancellable
, NULL
);
684 res
= g_output_stream_close (base_stream
, cancellable
, &error
);
689 g_task_return_error (task
, error
);
691 g_task_return_boolean (task
, TRUE
);
695 g_buffered_output_stream_flush_async (GOutputStream
*stream
,
697 GCancellable
*cancellable
,
698 GAsyncReadyCallback callback
,
704 fdata
= g_slice_new (FlushData
);
705 fdata
->flush_stream
= TRUE
;
706 fdata
->close_stream
= FALSE
;
708 task
= g_task_new (stream
, cancellable
, callback
, data
);
709 g_task_set_task_data (task
, fdata
, free_flush_data
);
710 g_task_set_priority (task
, io_priority
);
712 g_task_run_in_thread (task
, flush_buffer_thread
);
713 g_object_unref (task
);
717 g_buffered_output_stream_flush_finish (GOutputStream
*stream
,
718 GAsyncResult
*result
,
721 g_return_val_if_fail (g_task_is_valid (result
, stream
), FALSE
);
723 return g_task_propagate_boolean (G_TASK (result
), error
);
727 g_buffered_output_stream_close_async (GOutputStream
*stream
,
729 GCancellable
*cancellable
,
730 GAsyncReadyCallback callback
,
736 fdata
= g_slice_new (FlushData
);
737 fdata
->close_stream
= TRUE
;
739 task
= g_task_new (stream
, cancellable
, callback
, data
);
740 g_task_set_task_data (task
, fdata
, free_flush_data
);
741 g_task_set_priority (task
, io_priority
);
743 g_task_run_in_thread (task
, flush_buffer_thread
);
744 g_object_unref (task
);
748 g_buffered_output_stream_close_finish (GOutputStream
*stream
,
749 GAsyncResult
*result
,
752 g_return_val_if_fail (g_task_is_valid (result
, stream
), FALSE
);
754 return g_task_propagate_boolean (G_TASK (result
), error
);