1 /* GIO - GLib Input, Output and Streaming Library
3 * Copyright (C) 2006-2007 Red Hat, Inc.
4 * Copyright (C) 2007 Jürg Billeter
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
16 * You should have received a copy of the GNU Lesser General
17 * Public License along with this library; if not, write to the
18 * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
19 * Boston, MA 02111-1307, USA.
21 * Author: Christian Kellner <gicmo@gnome.org>
25 #include "gbufferedinputstream.h"
26 #include "ginputstream.h"
27 #include "gcancellable.h"
28 #include "gasyncresult.h"
29 #include "gsimpleasyncresult.h"
30 #include "gseekable.h"
37 * SECTION:gbufferedinputstream
38 * @short_description: Buffered Input Stream
40 * @see_also: #GFilterInputStream, #GInputStream
42 * Buffered input stream implements #GFilterInputStream and provides
45 * By default, #GBufferedInputStream's buffer size is set at 4 kilobytes.
47 * To create a buffered input stream, use g_buffered_input_stream_new(),
48 * or g_buffered_input_stream_new_sized() to specify the buffer's size at
51 * To get the size of a buffer within a buffered input stream, use
52 * g_buffered_input_stream_get_buffer_size(). To change the size of a
53 * buffered input stream's buffer, use
54 * g_buffered_input_stream_set_buffer_size(). Note that the buffer's size
55 * cannot be reduced below the size of the data within the buffer.
59 #define DEFAULT_BUFFER_SIZE 4096
61 struct _GBufferedInputStreamPrivate {
66 GAsyncReadyCallback outstanding_callback;
74 static void g_buffered_input_stream_set_property (GObject *object,
79 static void g_buffered_input_stream_get_property (GObject *object,
83 static void g_buffered_input_stream_finalize (GObject *object);
86 static gssize g_buffered_input_stream_skip (GInputStream *stream,
88 GCancellable *cancellable,
90 static void g_buffered_input_stream_skip_async (GInputStream *stream,
93 GCancellable *cancellable,
94 GAsyncReadyCallback callback,
96 static gssize g_buffered_input_stream_skip_finish (GInputStream *stream,
99 static gssize g_buffered_input_stream_read (GInputStream *stream,
102 GCancellable *cancellable,
104 static gssize g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
106 GCancellable *cancellable,
108 static void g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
111 GCancellable *cancellable,
112 GAsyncReadyCallback callback,
114 static gssize g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
115 GAsyncResult *result,
118 static void g_buffered_input_stream_seekable_iface_init (GSeekableIface *iface);
119 static goffset g_buffered_input_stream_tell (GSeekable *seekable);
120 static gboolean g_buffered_input_stream_can_seek (GSeekable *seekable);
121 static gboolean g_buffered_input_stream_seek (GSeekable *seekable,
124 GCancellable *cancellable,
126 static gboolean g_buffered_input_stream_can_truncate (GSeekable *seekable);
127 static gboolean g_buffered_input_stream_truncate (GSeekable *seekable,
129 GCancellable *cancellable,
132 static void g_buffered_input_stream_finalize (GObject *object);
134 static void compact_buffer (GBufferedInputStream *stream);
136 G_DEFINE_TYPE_WITH_CODE (GBufferedInputStream,
137 g_buffered_input_stream,
138 G_TYPE_FILTER_INPUT_STREAM,
139 G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE,
140 g_buffered_input_stream_seekable_iface_init))
143 g_buffered_input_stream_class_init (GBufferedInputStreamClass *klass)
145 GObjectClass *object_class;
146 GInputStreamClass *istream_class;
147 GBufferedInputStreamClass *bstream_class;
149 g_type_class_add_private (klass, sizeof (GBufferedInputStreamPrivate));
151 object_class = G_OBJECT_CLASS (klass);
152 object_class->get_property = g_buffered_input_stream_get_property;
153 object_class->set_property = g_buffered_input_stream_set_property;
154 object_class->finalize = g_buffered_input_stream_finalize;
156 istream_class = G_INPUT_STREAM_CLASS (klass);
157 istream_class->skip = g_buffered_input_stream_skip;
158 istream_class->skip_async = g_buffered_input_stream_skip_async;
159 istream_class->skip_finish = g_buffered_input_stream_skip_finish;
160 istream_class->read_fn = g_buffered_input_stream_read;
162 bstream_class = G_BUFFERED_INPUT_STREAM_CLASS (klass);
163 bstream_class->fill = g_buffered_input_stream_real_fill;
164 bstream_class->fill_async = g_buffered_input_stream_real_fill_async;
165 bstream_class->fill_finish = g_buffered_input_stream_real_fill_finish;
167 g_object_class_install_property (object_class,
169 g_param_spec_uint ("buffer-size",
171 P_("The size of the backend buffer"),
175 G_PARAM_READWRITE | G_PARAM_CONSTRUCT |
176 G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
182 * g_buffered_input_stream_get_buffer_size:
183 * @stream: a #GBufferedInputStream
185 * Gets the size of the input buffer.
187 * Returns: the current buffer size.
190 g_buffered_input_stream_get_buffer_size (GBufferedInputStream *stream)
192 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), 0);
194 return stream->priv->len;
198 * g_buffered_input_stream_set_buffer_size:
199 * @stream: a #GBufferedInputStream
202 * Sets the size of the internal buffer of @stream to @size, or to the
203 * size of the contents of the buffer. The buffer can never be resized
204 * smaller than its current contents.
207 g_buffered_input_stream_set_buffer_size (GBufferedInputStream *stream,
210 GBufferedInputStreamPrivate *priv;
214 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
218 if (priv->len == size)
223 in_buffer = priv->end - priv->pos;
225 /* Never resize smaller than current buffer contents */
226 size = MAX (size, in_buffer);
228 buffer = g_malloc (size);
229 memcpy (buffer, priv->buffer + priv->pos, in_buffer);
232 priv->end = in_buffer;
233 g_free (priv->buffer);
234 priv->buffer = buffer;
241 priv->buffer = g_malloc (size);
244 g_object_notify (G_OBJECT (stream), "buffer-size");
248 g_buffered_input_stream_set_property (GObject *object,
253 GBufferedInputStream *bstream;
255 bstream = G_BUFFERED_INPUT_STREAM (object);
260 g_buffered_input_stream_set_buffer_size (bstream, g_value_get_uint (value));
264 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
270 g_buffered_input_stream_get_property (GObject *object,
275 GBufferedInputStreamPrivate *priv;
276 GBufferedInputStream *bstream;
278 bstream = G_BUFFERED_INPUT_STREAM (object);
279 priv = bstream->priv;
284 g_value_set_uint (value, priv->len);
288 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
294 g_buffered_input_stream_finalize (GObject *object)
296 GBufferedInputStreamPrivate *priv;
297 GBufferedInputStream *stream;
299 stream = G_BUFFERED_INPUT_STREAM (object);
302 g_free (priv->buffer);
304 G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize (object);
308 g_buffered_input_stream_seekable_iface_init (GSeekableIface *iface)
310 iface->tell = g_buffered_input_stream_tell;
311 iface->can_seek = g_buffered_input_stream_can_seek;
312 iface->seek = g_buffered_input_stream_seek;
313 iface->can_truncate = g_buffered_input_stream_can_truncate;
314 iface->truncate_fn = g_buffered_input_stream_truncate;
318 g_buffered_input_stream_init (GBufferedInputStream *stream)
320 stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
321 G_TYPE_BUFFERED_INPUT_STREAM,
322 GBufferedInputStreamPrivate);
327 * g_buffered_input_stream_new:
328 * @base_stream: a #GInputStream
330 * Creates a new #GInputStream from the given @base_stream, with
331 * a buffer set to the default size (4 kilobytes).
333 * Returns: a #GInputStream for the given @base_stream.
336 g_buffered_input_stream_new (GInputStream *base_stream)
338 GInputStream *stream;
340 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
342 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
343 "base-stream", base_stream,
350 * g_buffered_input_stream_new_sized:
351 * @base_stream: a #GInputStream
354 * Creates a new #GBufferedInputStream from the given @base_stream,
355 * with a buffer set to @size.
357 * Returns: a #GInputStream.
360 g_buffered_input_stream_new_sized (GInputStream *base_stream,
363 GInputStream *stream;
365 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
367 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
368 "base-stream", base_stream,
369 "buffer-size", (guint)size,
376 * g_buffered_input_stream_fill:
377 * @stream: a #GBufferedInputStream
378 * @count: the number of bytes that will be read from the stream
379 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
380 * @error: location to store the error occurring, or %NULL to ignore
382 * Tries to read @count bytes from the stream into the buffer.
383 * Will block during this read.
385 * If @count is zero, returns zero and does nothing. A value of @count
386 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
388 * On success, the number of bytes read into the buffer is returned.
389 * It is not an error if this is not the same as the requested size, as it
390 * can happen e.g. near the end of a file. Zero is returned on end of file
391 * (or if @count is zero), but never otherwise.
393 * If @count is -1 then the attempted read size is equal to the number of
394 * bytes that are required to fill the buffer.
396 * If @cancellable is not %NULL, then the operation can be cancelled by
397 * triggering the cancellable object from another thread. If the operation
398 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
399 * operation was partially finished when the operation was cancelled the
400 * partial result will be returned, without an error.
402 * On error -1 is returned and @error is set accordingly.
404 * For the asynchronous, non-blocking, version of this function, see
405 * g_buffered_input_stream_fill_async().
407 * Returns: the number of bytes read into @stream's buffer, up to @count,
411 g_buffered_input_stream_fill (GBufferedInputStream *stream,
413 GCancellable *cancellable,
416 GBufferedInputStreamClass *class;
417 GInputStream *input_stream;
420 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
422 input_stream = G_INPUT_STREAM (stream);
426 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
427 _("Too large count value passed to %s"), G_STRFUNC);
431 if (!g_input_stream_set_pending (input_stream, error))
435 g_cancellable_push_current (cancellable);
437 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
438 res = class->fill (stream, count, cancellable, error);
441 g_cancellable_pop_current (cancellable);
443 g_input_stream_clear_pending (input_stream);
449 async_fill_callback_wrapper (GObject *source_object,
453 GBufferedInputStream *stream = G_BUFFERED_INPUT_STREAM (source_object);
455 g_input_stream_clear_pending (G_INPUT_STREAM (stream));
456 (*stream->priv->outstanding_callback) (source_object, res, user_data);
457 g_object_unref (stream);
461 * g_buffered_input_stream_fill_async:
462 * @stream: a #GBufferedInputStream
463 * @count: the number of bytes that will be read from the stream
464 * @io_priority: the <link linkend="io-priority">I/O priority</link>
466 * @cancellable: (allow-none): optional #GCancellable object
467 * @callback: (scope async): a #GAsyncReadyCallback
468 * @user_data: (closure): a #gpointer
470 * Reads data into @stream's buffer asynchronously, up to @count size.
471 * @io_priority can be used to prioritize reads. For the synchronous
472 * version of this function, see g_buffered_input_stream_fill().
474 * If @count is -1 then the attempted read size is equal to the number
475 * of bytes that are required to fill the buffer.
478 g_buffered_input_stream_fill_async (GBufferedInputStream *stream,
481 GCancellable *cancellable,
482 GAsyncReadyCallback callback,
485 GBufferedInputStreamClass *class;
486 GSimpleAsyncResult *simple;
487 GError *error = NULL;
489 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
493 simple = g_simple_async_result_new (G_OBJECT (stream),
496 g_buffered_input_stream_fill_async);
497 g_simple_async_result_complete_in_idle (simple);
498 g_object_unref (simple);
504 g_simple_async_report_error_in_idle (G_OBJECT (stream),
507 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
508 _("Too large count value passed to %s"),
513 if (!g_input_stream_set_pending (G_INPUT_STREAM (stream), &error))
515 g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
522 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
524 stream->priv->outstanding_callback = callback;
525 g_object_ref (stream);
526 class->fill_async (stream, count, io_priority, cancellable,
527 async_fill_callback_wrapper, user_data);
531 * g_buffered_input_stream_fill_finish:
532 * @stream: a #GBufferedInputStream
533 * @result: a #GAsyncResult
536 * Finishes an asynchronous read.
538 * Returns: a #gssize of the read stream, or %-1 on an error.
541 g_buffered_input_stream_fill_finish (GBufferedInputStream *stream,
542 GAsyncResult *result,
545 GSimpleAsyncResult *simple;
546 GBufferedInputStreamClass *class;
548 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
549 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
551 if (g_async_result_legacy_propagate_error (result, error))
554 if (G_IS_SIMPLE_ASYNC_RESULT (result))
556 simple = G_SIMPLE_ASYNC_RESULT (result);
558 /* Special case read of 0 bytes */
559 if (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_fill_async)
563 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
564 return class->fill_finish (stream, result, error);
568 * g_buffered_input_stream_get_available:
569 * @stream: #GBufferedInputStream
571 * Gets the size of the available data within the stream.
573 * Returns: size of the available stream.
576 g_buffered_input_stream_get_available (GBufferedInputStream *stream)
578 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
580 return stream->priv->end - stream->priv->pos;
584 * g_buffered_input_stream_peek:
585 * @stream: a #GBufferedInputStream
586 * @buffer: (array length=count) (element-type guint8): a pointer to
587 * an allocated chunk of memory
591 * Peeks in the buffer, copying data of size @count into @buffer,
592 * offset @offset bytes.
594 * Returns: a #gsize of the number of bytes peeked, or -1 on error.
597 g_buffered_input_stream_peek (GBufferedInputStream *stream,
605 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
606 g_return_val_if_fail (buffer != NULL, -1);
608 available = g_buffered_input_stream_get_available (stream);
610 if (offset > available)
613 end = MIN (offset + count, available);
614 count = end - offset;
616 memcpy (buffer, stream->priv->buffer + stream->priv->pos + offset, count);
621 * g_buffered_input_stream_peek_buffer:
622 * @stream: a #GBufferedInputStream
623 * @count: (out): a #gsize to get the number of bytes available in the buffer
625 * Returns the buffer with the currently available bytes. The returned
626 * buffer must not be modified and will become invalid when reading from
627 * the stream or filling the buffer.
629 * Returns: (array length=count) (element-type guint8) (transfer none):
633 g_buffered_input_stream_peek_buffer (GBufferedInputStream *stream,
636 GBufferedInputStreamPrivate *priv;
638 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), NULL);
643 *count = priv->end - priv->pos;
645 return priv->buffer + priv->pos;
649 compact_buffer (GBufferedInputStream *stream)
651 GBufferedInputStreamPrivate *priv;
656 current_size = priv->end - priv->pos;
658 g_memmove (priv->buffer, priv->buffer + priv->pos, current_size);
661 priv->end = current_size;
665 g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
667 GCancellable *cancellable,
670 GBufferedInputStreamPrivate *priv;
671 GInputStream *base_stream;
680 in_buffer = priv->end - priv->pos;
682 /* Never fill more than can fit in the buffer */
683 count = MIN (count, priv->len - in_buffer);
685 /* If requested length does not fit at end, compact */
686 if (priv->len - priv->end < count)
687 compact_buffer (stream);
689 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
690 nread = g_input_stream_read (base_stream,
691 priv->buffer + priv->end,
703 g_buffered_input_stream_skip (GInputStream *stream,
705 GCancellable *cancellable,
708 GBufferedInputStream *bstream;
709 GBufferedInputStreamPrivate *priv;
710 GBufferedInputStreamClass *class;
711 GInputStream *base_stream;
712 gsize available, bytes_skipped;
715 bstream = G_BUFFERED_INPUT_STREAM (stream);
716 priv = bstream->priv;
718 available = priv->end - priv->pos;
720 if (count <= available)
726 /* Full request not available, skip all currently available and
727 * request refill for more
732 bytes_skipped = available;
735 if (bytes_skipped > 0)
736 error = NULL; /* Ignore further errors if we already read some data */
738 if (count > priv->len)
740 /* Large request, shortcut buffer */
742 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
744 nread = g_input_stream_skip (base_stream,
749 if (nread < 0 && bytes_skipped == 0)
753 bytes_skipped += nread;
755 return bytes_skipped;
758 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
759 nread = class->fill (bstream, priv->len, cancellable, error);
763 if (bytes_skipped == 0)
766 return bytes_skipped;
769 available = priv->end - priv->pos;
770 count = MIN (count, available);
772 bytes_skipped += count;
775 return bytes_skipped;
779 g_buffered_input_stream_read (GInputStream *stream,
782 GCancellable *cancellable,
785 GBufferedInputStream *bstream;
786 GBufferedInputStreamPrivate *priv;
787 GBufferedInputStreamClass *class;
788 GInputStream *base_stream;
789 gsize available, bytes_read;
792 bstream = G_BUFFERED_INPUT_STREAM (stream);
793 priv = bstream->priv;
795 available = priv->end - priv->pos;
797 if (count <= available)
799 memcpy (buffer, priv->buffer + priv->pos, count);
804 /* Full request not available, read all currently available and
805 * request refill for more
808 memcpy (buffer, priv->buffer + priv->pos, available);
811 bytes_read = available;
815 error = NULL; /* Ignore further errors if we already read some data */
817 if (count > priv->len)
819 /* Large request, shortcut buffer */
821 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
823 nread = g_input_stream_read (base_stream,
824 (char *)buffer + bytes_read,
829 if (nread < 0 && bytes_read == 0)
838 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
839 nread = class->fill (bstream, priv->len, cancellable, error);
848 available = priv->end - priv->pos;
849 count = MIN (count, available);
851 memcpy ((char *)buffer + bytes_read, (char *)priv->buffer + priv->pos, count);
859 g_buffered_input_stream_tell (GSeekable *seekable)
861 GBufferedInputStream *bstream;
862 GBufferedInputStreamPrivate *priv;
863 GInputStream *base_stream;
864 GSeekable *base_stream_seekable;
868 bstream = G_BUFFERED_INPUT_STREAM (seekable);
869 priv = bstream->priv;
871 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
872 if (!G_IS_SEEKABLE (base_stream))
874 base_stream_seekable = G_SEEKABLE (base_stream);
876 available = priv->end - priv->pos;
877 base_offset = g_seekable_tell (base_stream_seekable);
879 return base_offset - available;
883 g_buffered_input_stream_can_seek (GSeekable *seekable)
885 GInputStream *base_stream;
887 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
888 return G_IS_SEEKABLE (base_stream) && g_seekable_can_seek (G_SEEKABLE (base_stream));
892 g_buffered_input_stream_seek (GSeekable *seekable,
895 GCancellable *cancellable,
898 GBufferedInputStream *bstream;
899 GBufferedInputStreamPrivate *priv;
900 GInputStream *base_stream;
901 GSeekable *base_stream_seekable;
903 bstream = G_BUFFERED_INPUT_STREAM (seekable);
904 priv = bstream->priv;
906 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
907 if (!G_IS_SEEKABLE (base_stream))
909 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
910 _("Seek not supported on base stream"));
914 base_stream_seekable = G_SEEKABLE (base_stream);
916 if (type == G_SEEK_CUR)
918 if (offset <= priv->end - priv->pos && offset >= -priv->pos)
925 offset -= priv->end - priv->pos;
929 if (g_seekable_seek (base_stream_seekable, offset, type, cancellable, error))
942 g_buffered_input_stream_can_truncate (GSeekable *seekable)
948 g_buffered_input_stream_truncate (GSeekable *seekable,
950 GCancellable *cancellable,
953 g_set_error_literal (error,
955 G_IO_ERROR_NOT_SUPPORTED,
956 _("Cannot truncate GBufferedInputStream"));
961 * g_buffered_input_stream_read_byte:
962 * @stream: a #GBufferedInputStream
963 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
964 * @error: location to store the error occurring, or %NULL to ignore
966 * Tries to read a single byte from the stream or the buffer. Will block
969 * On success, the byte read from the stream is returned. On end of stream
970 * -1 is returned but it's not an exceptional error and @error is not set.
972 * If @cancellable is not %NULL, then the operation can be cancelled by
973 * triggering the cancellable object from another thread. If the operation
974 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
975 * operation was partially finished when the operation was cancelled the
976 * partial result will be returned, without an error.
978 * On error -1 is returned and @error is set accordingly.
980 * Returns: the byte read from the @stream, or -1 on end of stream or error.
983 g_buffered_input_stream_read_byte (GBufferedInputStream *stream,
984 GCancellable *cancellable,
987 GBufferedInputStreamPrivate *priv;
988 GBufferedInputStreamClass *class;
989 GInputStream *input_stream;
993 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
996 input_stream = G_INPUT_STREAM (stream);
998 if (g_input_stream_is_closed (input_stream))
1000 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
1001 _("Stream is already closed"));
1005 if (!g_input_stream_set_pending (input_stream, error))
1008 available = priv->end - priv->pos;
1012 g_input_stream_clear_pending (input_stream);
1013 return priv->buffer[priv->pos++];
1016 /* Byte not available, request refill for more */
1019 g_cancellable_push_current (cancellable);
1024 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1025 nread = class->fill (stream, priv->len, cancellable, error);
1028 g_cancellable_pop_current (cancellable);
1030 g_input_stream_clear_pending (input_stream);
1033 return -1; /* error or end of stream */
1035 return priv->buffer[priv->pos++];
1038 /* ************************** */
1039 /* Async stuff implementation */
1040 /* ************************** */
1043 fill_async_callback (GObject *source_object,
1044 GAsyncResult *result,
1049 GSimpleAsyncResult *simple;
1054 res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
1057 g_simple_async_result_set_op_res_gssize (simple, res);
1060 g_simple_async_result_take_error (simple, error);
1064 GBufferedInputStreamPrivate *priv;
1067 object = g_async_result_get_source_object (G_ASYNC_RESULT (simple));
1068 priv = G_BUFFERED_INPUT_STREAM (object)->priv;
1070 g_assert_cmpint (priv->end + res, <=, priv->len);
1073 g_object_unref (object);
1076 /* Complete immediately, not in idle, since we're already
1077 * in a mainloop callout
1079 g_simple_async_result_complete (simple);
1080 g_object_unref (simple);
1084 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
1087 GCancellable *cancellable,
1088 GAsyncReadyCallback callback,
1091 GBufferedInputStreamPrivate *priv;
1092 GInputStream *base_stream;
1093 GSimpleAsyncResult *simple;
1096 priv = stream->priv;
1101 in_buffer = priv->end - priv->pos;
1103 /* Never fill more than can fit in the buffer */
1104 count = MIN (count, priv->len - in_buffer);
1106 /* If requested length does not fit at end, compact */
1107 if (priv->len - priv->end < count)
1108 compact_buffer (stream);
1110 simple = g_simple_async_result_new (G_OBJECT (stream),
1111 callback, user_data,
1112 g_buffered_input_stream_real_fill_async);
1114 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1115 g_input_stream_read_async (base_stream,
1116 priv->buffer + priv->end,
1120 fill_async_callback,
1125 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1126 GAsyncResult *result,
1129 GSimpleAsyncResult *simple;
1132 simple = G_SIMPLE_ASYNC_RESULT (result);
1133 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_real_fill_async);
1135 if (g_simple_async_result_propagate_error (simple, error))
1138 nread = g_simple_async_result_get_op_res_gssize (simple);
1144 gssize bytes_skipped;
1149 free_skip_async_data (gpointer _data)
1151 SkipAsyncData *data = _data;
1152 g_slice_free (SkipAsyncData, data);
1156 large_skip_callback (GObject *source_object,
1157 GAsyncResult *result,
1160 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1161 SkipAsyncData *data;
1165 data = g_simple_async_result_get_op_res_gpointer (simple);
1168 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1171 /* Only report the error if we've not already read some data */
1172 if (nread < 0 && data->bytes_skipped == 0)
1173 g_simple_async_result_take_error (simple, error);
1175 g_error_free (error);
1178 data->bytes_skipped += nread;
1180 /* Complete immediately, not in idle, since we're already
1181 * in a mainloop callout
1183 g_simple_async_result_complete (simple);
1184 g_object_unref (simple);
1188 skip_fill_buffer_callback (GObject *source_object,
1189 GAsyncResult *result,
1192 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1193 GBufferedInputStream *bstream;
1194 GBufferedInputStreamPrivate *priv;
1195 SkipAsyncData *data;
1200 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1201 priv = bstream->priv;
1203 data = g_simple_async_result_get_op_res_gpointer (simple);
1206 nread = g_buffered_input_stream_fill_finish (bstream,
1209 if (nread < 0 && data->bytes_skipped == 0)
1210 g_simple_async_result_take_error (simple, error);
1212 g_error_free (error);
1216 available = priv->end - priv->pos;
1217 data->count = MIN (data->count, available);
1219 data->bytes_skipped += data->count;
1220 priv->pos += data->count;
1223 /* Complete immediately, not in idle, since we're already
1224 * in a mainloop callout
1226 g_simple_async_result_complete (simple);
1227 g_object_unref (simple);
1231 g_buffered_input_stream_skip_async (GInputStream *stream,
1234 GCancellable *cancellable,
1235 GAsyncReadyCallback callback,
1238 GBufferedInputStream *bstream;
1239 GBufferedInputStreamPrivate *priv;
1240 GBufferedInputStreamClass *class;
1241 GInputStream *base_stream;
1243 GSimpleAsyncResult *simple;
1244 SkipAsyncData *data;
1246 bstream = G_BUFFERED_INPUT_STREAM (stream);
1247 priv = bstream->priv;
1249 data = g_slice_new (SkipAsyncData);
1250 data->bytes_skipped = 0;
1251 simple = g_simple_async_result_new (G_OBJECT (stream),
1252 callback, user_data,
1253 g_buffered_input_stream_skip_async);
1254 g_simple_async_result_set_op_res_gpointer (simple, data, free_skip_async_data);
1256 available = priv->end - priv->pos;
1258 if (count <= available)
1261 data->bytes_skipped = count;
1263 g_simple_async_result_complete_in_idle (simple);
1264 g_object_unref (simple);
1268 /* Full request not available, skip all currently available
1269 * and request refill for more
1277 data->bytes_skipped = available;
1278 data->count = count;
1280 if (count > priv->len)
1282 /* Large request, shortcut buffer */
1284 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1286 g_input_stream_skip_async (base_stream,
1288 io_priority, cancellable,
1289 large_skip_callback,
1294 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1295 class->fill_async (bstream, priv->len, io_priority, cancellable,
1296 skip_fill_buffer_callback, simple);
1301 g_buffered_input_stream_skip_finish (GInputStream *stream,
1302 GAsyncResult *result,
1305 GSimpleAsyncResult *simple;
1306 SkipAsyncData *data;
1308 simple = G_SIMPLE_ASYNC_RESULT (result);
1310 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_skip_async);
1312 data = g_simple_async_result_get_op_res_gpointer (simple);
1314 return data->bytes_skipped;