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>
25 #include "gbufferedoutputstream.h"
26 #include "goutputstream.h"
27 #include "gsimpleasyncresult.h"
32 #define DEFAULT_BUFFER_SIZE 4096
34 struct _GBufferedOutputStreamPrivate {
46 static void g_buffered_output_stream_set_property (GObject *object,
51 static void g_buffered_output_stream_get_property (GObject *object,
55 static void g_buffered_output_stream_finalize (GObject *object);
58 static gssize g_buffered_output_stream_write (GOutputStream *stream,
61 GCancellable *cancellable,
63 static gboolean g_buffered_output_stream_flush (GOutputStream *stream,
64 GCancellable *cancellable,
66 static gboolean g_buffered_output_stream_close (GOutputStream *stream,
67 GCancellable *cancellable,
70 static void g_buffered_output_stream_write_async (GOutputStream *stream,
74 GCancellable *cancellable,
75 GAsyncReadyCallback callback,
77 static gssize g_buffered_output_stream_write_finish (GOutputStream *stream,
80 static void g_buffered_output_stream_flush_async (GOutputStream *stream,
82 GCancellable *cancellable,
83 GAsyncReadyCallback callback,
85 static gboolean g_buffered_output_stream_flush_finish (GOutputStream *stream,
88 static void g_buffered_output_stream_close_async (GOutputStream *stream,
90 GCancellable *cancellable,
91 GAsyncReadyCallback callback,
93 static gboolean g_buffered_output_stream_close_finish (GOutputStream *stream,
97 G_DEFINE_TYPE (GBufferedOutputStream,
98 g_buffered_output_stream,
99 G_TYPE_FILTER_OUTPUT_STREAM)
103 g_buffered_output_stream_class_init (GBufferedOutputStreamClass *klass)
105 GObjectClass *object_class;
106 GOutputStreamClass *ostream_class;
108 g_type_class_add_private (klass, sizeof (GBufferedOutputStreamPrivate));
110 object_class = G_OBJECT_CLASS (klass);
111 object_class->get_property = g_buffered_output_stream_get_property;
112 object_class->set_property = g_buffered_output_stream_set_property;
113 object_class->finalize = g_buffered_output_stream_finalize;
115 ostream_class = G_OUTPUT_STREAM_CLASS (klass);
116 ostream_class->write = g_buffered_output_stream_write;
117 ostream_class->flush = g_buffered_output_stream_flush;
118 ostream_class->close = g_buffered_output_stream_close;
119 ostream_class->write_async = g_buffered_output_stream_write_async;
120 ostream_class->write_finish = g_buffered_output_stream_write_finish;
121 ostream_class->flush_async = g_buffered_output_stream_flush_async;
122 ostream_class->flush_finish = g_buffered_output_stream_flush_finish;
123 ostream_class->close_async = g_buffered_output_stream_close_async;
124 ostream_class->close_finish = g_buffered_output_stream_close_finish;
126 g_object_class_install_property (object_class,
128 g_param_spec_uint ("buffer-size",
130 P_("The size of the backend buffer"),
134 G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY |
135 G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
140 * g_buffered_output_stream_get_buffer_size:
141 * @stream: a #GBufferedOutputStream.
143 * Returns: the current size of the buffer.
146 g_buffered_output_stream_get_buffer_size (GBufferedOutputStream *stream)
148 g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream), -1);
150 return stream->priv->len;
154 * g_buffered_output_stream_set_buffer_size:
155 * @stream: a #GBufferedOutputStream.
158 * Sets the size of the internal buffer to @size.
162 g_buffered_output_stream_set_buffer_size (GBufferedOutputStream *stream,
165 GBufferedOutputStreamPrivate *priv;
168 g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream));
174 size = MAX (size, priv->pos);
176 buffer = g_malloc (size);
177 memcpy (buffer, priv->buffer, priv->pos);
178 g_free (priv->buffer);
179 priv->buffer = buffer;
185 priv->buffer = g_malloc (size);
192 * g_buffered_output_stream_get_auto_grow:
193 * @stream: a #GBufferedOutputStream.
195 * Returns: %TRUE if the @stream's buffer automatically grows,
199 g_buffered_output_stream_get_auto_grow (GBufferedOutputStream *stream)
201 g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream), FALSE);
203 return stream->priv->auto_grow;
207 * g_buffered_output_stream_set_auto_grow:
208 * @stream: a #GBufferedOutputStream.
209 * @auto_grow: a boolean.
211 * Sets whether or not the @stream's buffer should automatically grow.
215 g_buffered_output_stream_set_auto_grow (GBufferedOutputStream *stream,
218 g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream));
220 stream->priv->auto_grow = auto_grow;
224 g_buffered_output_stream_set_property (GObject *object,
229 GBufferedOutputStream *buffered_stream;
230 GBufferedOutputStreamPrivate *priv;
232 buffered_stream = G_BUFFERED_OUTPUT_STREAM (object);
233 priv = buffered_stream->priv;
239 g_buffered_output_stream_set_buffer_size (buffered_stream, g_value_get_uint (value));
243 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
250 g_buffered_output_stream_get_property (GObject *object,
255 GBufferedOutputStream *buffered_stream;
256 GBufferedOutputStreamPrivate *priv;
258 buffered_stream = G_BUFFERED_OUTPUT_STREAM (object);
259 priv = buffered_stream->priv;
265 g_value_set_uint (value, priv->len);
269 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
276 g_buffered_output_stream_finalize (GObject *object)
278 GBufferedOutputStream *stream;
279 GBufferedOutputStreamPrivate *priv;
281 stream = G_BUFFERED_OUTPUT_STREAM (object);
284 g_free (priv->buffer);
286 if (G_OBJECT_CLASS (g_buffered_output_stream_parent_class)->finalize)
287 (*G_OBJECT_CLASS (g_buffered_output_stream_parent_class)->finalize) (object);
291 g_buffered_output_stream_init (GBufferedOutputStream *stream)
293 stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
294 G_TYPE_BUFFERED_OUTPUT_STREAM,
295 GBufferedOutputStreamPrivate);
300 * g_buffered_output_stream_new:
301 * @base_stream: a #GOutputStream.
303 * Returns: a #GOutputStream for the given @base_stream.
306 g_buffered_output_stream_new (GOutputStream *base_stream)
308 GOutputStream *stream;
310 g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), NULL);
312 stream = g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM,
313 "base-stream", base_stream,
320 * g_buffered_output_stream_new_sized:
321 * @base_stream: a #GOutputStream.
324 * Returns: a #GOutputStream with an internal buffer set to @size.
327 g_buffered_output_stream_new_sized (GOutputStream *base_stream,
330 GOutputStream *stream;
332 g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), NULL);
334 stream = g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM,
335 "base-stream", base_stream,
343 flush_buffer (GBufferedOutputStream *stream,
344 GCancellable *cancellable,
347 GBufferedOutputStreamPrivate *priv;
348 GOutputStream *base_stream;
355 base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
357 g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), FALSE);
359 res = g_output_stream_write_all (base_stream,
366 count = priv->pos - bytes_written;
369 g_memmove (priv->buffer, priv->buffer + bytes_written, count);
371 priv->pos -= bytes_written;
377 g_buffered_output_stream_write (GOutputStream *stream,
380 GCancellable *cancellable,
383 GBufferedOutputStream *bstream;
384 GBufferedOutputStreamPrivate *priv;
389 bstream = G_BUFFERED_OUTPUT_STREAM (stream);
390 priv = bstream->priv;
392 n = priv->len - priv->pos;
394 if (priv->auto_grow && n < count)
396 new_size = MAX (priv->len * 2, priv->len + count);
397 g_buffered_output_stream_set_buffer_size (bstream, new_size);
401 res = flush_buffer (bstream, cancellable, error);
407 n = priv->len - priv->pos;
409 count = MIN (count, n);
410 memcpy (priv->buffer + priv->pos, buffer, count);
417 g_buffered_output_stream_flush (GOutputStream *stream,
418 GCancellable *cancellable,
421 GBufferedOutputStream *bstream;
422 GBufferedOutputStreamPrivate *priv;
423 GOutputStream *base_stream;
426 bstream = G_BUFFERED_OUTPUT_STREAM (stream);
427 priv = bstream->priv;
428 base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
430 res = flush_buffer (bstream, cancellable, error);
436 res = g_output_stream_flush (base_stream,
443 g_buffered_output_stream_close (GOutputStream *stream,
444 GCancellable *cancellable,
447 GBufferedOutputStream *bstream;
448 GBufferedOutputStreamPrivate *priv;
449 GOutputStream *base_stream;
452 bstream = G_BUFFERED_OUTPUT_STREAM (stream);
453 priv = bstream->priv;
454 base_stream = G_FILTER_OUTPUT_STREAM (bstream)->base_stream;
456 res = flush_buffer (bstream, cancellable, error);
458 /* report the first error but still close the stream */
461 res = g_output_stream_close (base_stream,
467 g_output_stream_close (base_stream,
475 /* ************************** */
476 /* Async stuff implementation */
477 /* ************************** */
479 /* TODO: This should be using the base class async ops, not threads */
483 guint flush_stream : 1;
484 guint close_stream : 1;
489 free_flush_data (gpointer data)
491 g_slice_free (FlushData, data);
494 /* This function is used by all three (i.e.
495 * _write, _flush, _close) functions since
496 * all of them will need to flush the buffer
497 * and so closing and writing is just a special
498 * case of flushing + some addition stuff */
500 flush_buffer_thread (GSimpleAsyncResult *result,
502 GCancellable *cancellable)
504 GBufferedOutputStream *stream;
505 GOutputStream *base_stream;
508 GError *error = NULL;
510 stream = G_BUFFERED_OUTPUT_STREAM (object);
511 fdata = g_simple_async_result_get_op_res_gpointer (result);
512 base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
514 res = flush_buffer (stream, cancellable, &error);
516 /* if flushing the buffer didn't work don't even bother
517 * to flush the stream but just report that error */
518 if (res && fdata->flush_stream)
520 res = g_output_stream_flush (base_stream,
525 if (fdata->close_stream)
528 /* if flushing the buffer or the stream returned
529 * an error report that first error but still try
530 * close the stream */
533 g_output_stream_close (base_stream,
539 res = g_output_stream_close (base_stream,
548 g_simple_async_result_set_from_error (result, error);
549 g_error_free (error);
563 free_write_data (gpointer data)
565 g_slice_free (WriteData, data);
569 g_buffered_output_stream_write_async (GOutputStream *stream,
573 GCancellable *cancellable,
574 GAsyncReadyCallback callback,
577 GBufferedOutputStream *buffered_stream;
578 GBufferedOutputStreamPrivate *priv;
579 GSimpleAsyncResult *res;
582 buffered_stream = G_BUFFERED_OUTPUT_STREAM (stream);
583 priv = buffered_stream->priv;
585 wdata = g_slice_new (WriteData);
586 wdata->count = count;
587 wdata->buffer = buffer;
589 res = g_simple_async_result_new (G_OBJECT (stream),
592 g_buffered_output_stream_write_async);
594 g_simple_async_result_set_op_res_gpointer (res, wdata, free_write_data);
596 /* if we have space left directly call the
597 * callback (from idle) otherwise schedule a buffer
598 * flush in the thread. In both cases the actual
599 * copying of the data to the buffer will be done in
600 * the write_finish () func since that should
602 if (priv->len - priv->pos > 0)
604 g_simple_async_result_complete_in_idle (res);
608 wdata->fdata.flush_stream = FALSE;
609 wdata->fdata.close_stream = FALSE;
610 g_simple_async_result_run_in_thread (res,
614 g_object_unref (res);
619 g_buffered_output_stream_write_finish (GOutputStream *stream,
620 GAsyncResult *result,
623 GBufferedOutputStreamPrivate *priv;
624 GBufferedOutputStream *buffered_stream;
625 GSimpleAsyncResult *simple;
629 simple = G_SIMPLE_ASYNC_RESULT (result);
630 buffered_stream = G_BUFFERED_OUTPUT_STREAM (stream);
631 priv = buffered_stream->priv;
633 g_assert (g_simple_async_result_get_source_tag (simple) ==
634 g_buffered_output_stream_write_async);
636 wdata = g_simple_async_result_get_op_res_gpointer (simple);
638 /* Now do the real copying of data to the buffer */
639 count = priv->len - priv->pos;
640 count = MIN (wdata->count, count);
642 memcpy (priv->buffer + priv->pos, wdata->buffer, count);
650 g_buffered_output_stream_flush_async (GOutputStream *stream,
652 GCancellable *cancellable,
653 GAsyncReadyCallback callback,
656 GSimpleAsyncResult *res;
659 fdata = g_slice_new (FlushData);
660 fdata->flush_stream = TRUE;
661 fdata->close_stream = FALSE;
663 res = g_simple_async_result_new (G_OBJECT (stream),
666 g_buffered_output_stream_flush_async);
668 g_simple_async_result_set_op_res_gpointer (res, fdata, free_flush_data);
670 g_simple_async_result_run_in_thread (res,
674 g_object_unref (res);
678 g_buffered_output_stream_flush_finish (GOutputStream *stream,
679 GAsyncResult *result,
682 GSimpleAsyncResult *simple;
684 simple = G_SIMPLE_ASYNC_RESULT (result);
686 g_assert (g_simple_async_result_get_source_tag (simple) ==
687 g_buffered_output_stream_flush_async);
693 g_buffered_output_stream_close_async (GOutputStream *stream,
695 GCancellable *cancellable,
696 GAsyncReadyCallback callback,
699 GSimpleAsyncResult *res;
702 fdata = g_slice_new (FlushData);
703 fdata->close_stream = TRUE;
705 res = g_simple_async_result_new (G_OBJECT (stream),
708 g_buffered_output_stream_close_async);
710 g_simple_async_result_set_op_res_gpointer (res, fdata, free_flush_data);
712 g_simple_async_result_run_in_thread (res,
716 g_object_unref (res);
720 g_buffered_output_stream_close_finish (GOutputStream *stream,
721 GAsyncResult *result,
724 GSimpleAsyncResult *simple;
726 simple = G_SIMPLE_ASYNC_RESULT (result);
728 g_assert (g_simple_async_result_get_source_tag (simple) ==
729 g_buffered_output_stream_flush_async);
734 /* vim: ts=2 sw=2 et */