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"
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_ADD_PRIVATE (GBufferedInputStream)
140 G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE,
141 g_buffered_input_stream_seekable_iface_init))
144 g_buffered_input_stream_class_init (GBufferedInputStreamClass *klass)
146 GObjectClass *object_class;
147 GInputStreamClass *istream_class;
148 GBufferedInputStreamClass *bstream_class;
150 object_class = G_OBJECT_CLASS (klass);
151 object_class->get_property = g_buffered_input_stream_get_property;
152 object_class->set_property = g_buffered_input_stream_set_property;
153 object_class->finalize = g_buffered_input_stream_finalize;
155 istream_class = G_INPUT_STREAM_CLASS (klass);
156 istream_class->skip = g_buffered_input_stream_skip;
157 istream_class->skip_async = g_buffered_input_stream_skip_async;
158 istream_class->skip_finish = g_buffered_input_stream_skip_finish;
159 istream_class->read_fn = g_buffered_input_stream_read;
161 bstream_class = G_BUFFERED_INPUT_STREAM_CLASS (klass);
162 bstream_class->fill = g_buffered_input_stream_real_fill;
163 bstream_class->fill_async = g_buffered_input_stream_real_fill_async;
164 bstream_class->fill_finish = g_buffered_input_stream_real_fill_finish;
166 g_object_class_install_property (object_class,
168 g_param_spec_uint ("buffer-size",
170 P_("The size of the backend buffer"),
174 G_PARAM_READWRITE | G_PARAM_CONSTRUCT |
175 G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
181 * g_buffered_input_stream_get_buffer_size:
182 * @stream: a #GBufferedInputStream
184 * Gets the size of the input buffer.
186 * Returns: the current buffer size.
189 g_buffered_input_stream_get_buffer_size (GBufferedInputStream *stream)
191 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), 0);
193 return stream->priv->len;
197 * g_buffered_input_stream_set_buffer_size:
198 * @stream: a #GBufferedInputStream
201 * Sets the size of the internal buffer of @stream to @size, or to the
202 * size of the contents of the buffer. The buffer can never be resized
203 * smaller than its current contents.
206 g_buffered_input_stream_set_buffer_size (GBufferedInputStream *stream,
209 GBufferedInputStreamPrivate *priv;
213 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
217 if (priv->len == size)
222 in_buffer = priv->end - priv->pos;
224 /* Never resize smaller than current buffer contents */
225 size = MAX (size, in_buffer);
227 buffer = g_malloc (size);
228 memcpy (buffer, priv->buffer + priv->pos, in_buffer);
231 priv->end = in_buffer;
232 g_free (priv->buffer);
233 priv->buffer = buffer;
240 priv->buffer = g_malloc (size);
243 g_object_notify (G_OBJECT (stream), "buffer-size");
247 g_buffered_input_stream_set_property (GObject *object,
252 GBufferedInputStream *bstream;
254 bstream = G_BUFFERED_INPUT_STREAM (object);
259 g_buffered_input_stream_set_buffer_size (bstream, g_value_get_uint (value));
263 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
269 g_buffered_input_stream_get_property (GObject *object,
274 GBufferedInputStreamPrivate *priv;
275 GBufferedInputStream *bstream;
277 bstream = G_BUFFERED_INPUT_STREAM (object);
278 priv = bstream->priv;
283 g_value_set_uint (value, priv->len);
287 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
293 g_buffered_input_stream_finalize (GObject *object)
295 GBufferedInputStreamPrivate *priv;
296 GBufferedInputStream *stream;
298 stream = G_BUFFERED_INPUT_STREAM (object);
301 g_free (priv->buffer);
303 G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize (object);
307 g_buffered_input_stream_seekable_iface_init (GSeekableIface *iface)
309 iface->tell = g_buffered_input_stream_tell;
310 iface->can_seek = g_buffered_input_stream_can_seek;
311 iface->seek = g_buffered_input_stream_seek;
312 iface->can_truncate = g_buffered_input_stream_can_truncate;
313 iface->truncate_fn = g_buffered_input_stream_truncate;
317 g_buffered_input_stream_init (GBufferedInputStream *stream)
319 stream->priv = g_buffered_input_stream_get_instance_private (stream);
324 * g_buffered_input_stream_new:
325 * @base_stream: a #GInputStream
327 * Creates a new #GInputStream from the given @base_stream, with
328 * a buffer set to the default size (4 kilobytes).
330 * Returns: a #GInputStream for the given @base_stream.
333 g_buffered_input_stream_new (GInputStream *base_stream)
335 GInputStream *stream;
337 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
339 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
340 "base-stream", base_stream,
347 * g_buffered_input_stream_new_sized:
348 * @base_stream: a #GInputStream
351 * Creates a new #GBufferedInputStream from the given @base_stream,
352 * with a buffer set to @size.
354 * Returns: a #GInputStream.
357 g_buffered_input_stream_new_sized (GInputStream *base_stream,
360 GInputStream *stream;
362 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
364 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
365 "base-stream", base_stream,
366 "buffer-size", (guint)size,
373 * g_buffered_input_stream_fill:
374 * @stream: a #GBufferedInputStream
375 * @count: the number of bytes that will be read from the stream
376 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
377 * @error: location to store the error occurring, or %NULL to ignore
379 * Tries to read @count bytes from the stream into the buffer.
380 * Will block during this read.
382 * If @count is zero, returns zero and does nothing. A value of @count
383 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
385 * On success, the number of bytes read into the buffer is returned.
386 * It is not an error if this is not the same as the requested size, as it
387 * can happen e.g. near the end of a file. Zero is returned on end of file
388 * (or if @count is zero), but never otherwise.
390 * If @count is -1 then the attempted read size is equal to the number of
391 * bytes that are required to fill the buffer.
393 * If @cancellable is not %NULL, then the operation can be cancelled by
394 * triggering the cancellable object from another thread. If the operation
395 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
396 * operation was partially finished when the operation was cancelled the
397 * partial result will be returned, without an error.
399 * On error -1 is returned and @error is set accordingly.
401 * For the asynchronous, non-blocking, version of this function, see
402 * g_buffered_input_stream_fill_async().
404 * Returns: the number of bytes read into @stream's buffer, up to @count,
408 g_buffered_input_stream_fill (GBufferedInputStream *stream,
410 GCancellable *cancellable,
413 GBufferedInputStreamClass *class;
414 GInputStream *input_stream;
417 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
419 input_stream = G_INPUT_STREAM (stream);
423 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
424 _("Too large count value passed to %s"), G_STRFUNC);
428 if (!g_input_stream_set_pending (input_stream, error))
432 g_cancellable_push_current (cancellable);
434 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
435 res = class->fill (stream, count, cancellable, error);
438 g_cancellable_pop_current (cancellable);
440 g_input_stream_clear_pending (input_stream);
446 async_fill_callback_wrapper (GObject *source_object,
450 GBufferedInputStream *stream = G_BUFFERED_INPUT_STREAM (source_object);
452 g_input_stream_clear_pending (G_INPUT_STREAM (stream));
453 (*stream->priv->outstanding_callback) (source_object, res, user_data);
454 g_object_unref (stream);
458 * g_buffered_input_stream_fill_async:
459 * @stream: a #GBufferedInputStream
460 * @count: the number of bytes that will be read from the stream
461 * @io_priority: the <link linkend="io-priority">I/O priority</link>
463 * @cancellable: (allow-none): optional #GCancellable object
464 * @callback: (scope async): a #GAsyncReadyCallback
465 * @user_data: (closure): a #gpointer
467 * Reads data into @stream's buffer asynchronously, up to @count size.
468 * @io_priority can be used to prioritize reads. For the synchronous
469 * version of this function, see g_buffered_input_stream_fill().
471 * If @count is -1 then the attempted read size is equal to the number
472 * of bytes that are required to fill the buffer.
475 g_buffered_input_stream_fill_async (GBufferedInputStream *stream,
478 GCancellable *cancellable,
479 GAsyncReadyCallback callback,
482 GBufferedInputStreamClass *class;
483 GError *error = NULL;
485 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
491 task = g_task_new (stream, cancellable, callback, user_data);
492 g_task_set_source_tag (task, g_buffered_input_stream_fill_async);
493 g_task_return_int (task, 0);
494 g_object_unref (task);
500 g_task_report_new_error (stream, callback, user_data,
501 g_buffered_input_stream_fill_async,
502 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
503 _("Too large count value passed to %s"),
508 if (!g_input_stream_set_pending (G_INPUT_STREAM (stream), &error))
510 g_task_report_error (stream, callback, user_data,
511 g_buffered_input_stream_fill_async,
516 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
518 stream->priv->outstanding_callback = callback;
519 g_object_ref (stream);
520 class->fill_async (stream, count, io_priority, cancellable,
521 async_fill_callback_wrapper, user_data);
525 * g_buffered_input_stream_fill_finish:
526 * @stream: a #GBufferedInputStream
527 * @result: a #GAsyncResult
530 * Finishes an asynchronous read.
532 * Returns: a #gssize of the read stream, or %-1 on an error.
535 g_buffered_input_stream_fill_finish (GBufferedInputStream *stream,
536 GAsyncResult *result,
539 GBufferedInputStreamClass *class;
541 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
542 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
544 if (g_async_result_legacy_propagate_error (result, error))
546 else if (g_async_result_is_tagged (result, g_buffered_input_stream_fill_async))
547 return g_task_propagate_int (G_TASK (result), error);
549 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
550 return class->fill_finish (stream, result, error);
554 * g_buffered_input_stream_get_available:
555 * @stream: #GBufferedInputStream
557 * Gets the size of the available data within the stream.
559 * Returns: size of the available stream.
562 g_buffered_input_stream_get_available (GBufferedInputStream *stream)
564 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
566 return stream->priv->end - stream->priv->pos;
570 * g_buffered_input_stream_peek:
571 * @stream: a #GBufferedInputStream
572 * @buffer: (array length=count) (element-type guint8): a pointer to
573 * an allocated chunk of memory
577 * Peeks in the buffer, copying data of size @count into @buffer,
578 * offset @offset bytes.
580 * Returns: a #gsize of the number of bytes peeked, or -1 on error.
583 g_buffered_input_stream_peek (GBufferedInputStream *stream,
591 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
592 g_return_val_if_fail (buffer != NULL, -1);
594 available = g_buffered_input_stream_get_available (stream);
596 if (offset > available)
599 end = MIN (offset + count, available);
600 count = end - offset;
602 memcpy (buffer, stream->priv->buffer + stream->priv->pos + offset, count);
607 * g_buffered_input_stream_peek_buffer:
608 * @stream: a #GBufferedInputStream
609 * @count: (out): a #gsize to get the number of bytes available in the buffer
611 * Returns the buffer with the currently available bytes. The returned
612 * buffer must not be modified and will become invalid when reading from
613 * the stream or filling the buffer.
615 * Returns: (array length=count) (element-type guint8) (transfer none):
619 g_buffered_input_stream_peek_buffer (GBufferedInputStream *stream,
622 GBufferedInputStreamPrivate *priv;
624 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), NULL);
629 *count = priv->end - priv->pos;
631 return priv->buffer + priv->pos;
635 compact_buffer (GBufferedInputStream *stream)
637 GBufferedInputStreamPrivate *priv;
642 current_size = priv->end - priv->pos;
644 g_memmove (priv->buffer, priv->buffer + priv->pos, current_size);
647 priv->end = current_size;
651 g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
653 GCancellable *cancellable,
656 GBufferedInputStreamPrivate *priv;
657 GInputStream *base_stream;
666 in_buffer = priv->end - priv->pos;
668 /* Never fill more than can fit in the buffer */
669 count = MIN (count, priv->len - in_buffer);
671 /* If requested length does not fit at end, compact */
672 if (priv->len - priv->end < count)
673 compact_buffer (stream);
675 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
676 nread = g_input_stream_read (base_stream,
677 priv->buffer + priv->end,
689 g_buffered_input_stream_skip (GInputStream *stream,
691 GCancellable *cancellable,
694 GBufferedInputStream *bstream;
695 GBufferedInputStreamPrivate *priv;
696 GBufferedInputStreamClass *class;
697 GInputStream *base_stream;
698 gsize available, bytes_skipped;
701 bstream = G_BUFFERED_INPUT_STREAM (stream);
702 priv = bstream->priv;
704 available = priv->end - priv->pos;
706 if (count <= available)
712 /* Full request not available, skip all currently available and
713 * request refill for more
718 bytes_skipped = available;
721 if (bytes_skipped > 0)
722 error = NULL; /* Ignore further errors if we already read some data */
724 if (count > priv->len)
726 /* Large request, shortcut buffer */
728 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
730 nread = g_input_stream_skip (base_stream,
735 if (nread < 0 && bytes_skipped == 0)
739 bytes_skipped += nread;
741 return bytes_skipped;
744 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
745 nread = class->fill (bstream, priv->len, cancellable, error);
749 if (bytes_skipped == 0)
752 return bytes_skipped;
755 available = priv->end - priv->pos;
756 count = MIN (count, available);
758 bytes_skipped += count;
761 return bytes_skipped;
765 g_buffered_input_stream_read (GInputStream *stream,
768 GCancellable *cancellable,
771 GBufferedInputStream *bstream;
772 GBufferedInputStreamPrivate *priv;
773 GBufferedInputStreamClass *class;
774 GInputStream *base_stream;
775 gsize available, bytes_read;
778 bstream = G_BUFFERED_INPUT_STREAM (stream);
779 priv = bstream->priv;
781 available = priv->end - priv->pos;
783 if (count <= available)
785 memcpy (buffer, priv->buffer + priv->pos, count);
790 /* Full request not available, read all currently available and
791 * request refill for more
794 memcpy (buffer, priv->buffer + priv->pos, available);
797 bytes_read = available;
801 error = NULL; /* Ignore further errors if we already read some data */
803 if (count > priv->len)
805 /* Large request, shortcut buffer */
807 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
809 nread = g_input_stream_read (base_stream,
810 (char *)buffer + bytes_read,
815 if (nread < 0 && bytes_read == 0)
824 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
825 nread = class->fill (bstream, priv->len, cancellable, error);
834 available = priv->end - priv->pos;
835 count = MIN (count, available);
837 memcpy ((char *)buffer + bytes_read, (char *)priv->buffer + priv->pos, count);
845 g_buffered_input_stream_tell (GSeekable *seekable)
847 GBufferedInputStream *bstream;
848 GBufferedInputStreamPrivate *priv;
849 GInputStream *base_stream;
850 GSeekable *base_stream_seekable;
854 bstream = G_BUFFERED_INPUT_STREAM (seekable);
855 priv = bstream->priv;
857 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
858 if (!G_IS_SEEKABLE (base_stream))
860 base_stream_seekable = G_SEEKABLE (base_stream);
862 available = priv->end - priv->pos;
863 base_offset = g_seekable_tell (base_stream_seekable);
865 return base_offset - available;
869 g_buffered_input_stream_can_seek (GSeekable *seekable)
871 GInputStream *base_stream;
873 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
874 return G_IS_SEEKABLE (base_stream) && g_seekable_can_seek (G_SEEKABLE (base_stream));
878 g_buffered_input_stream_seek (GSeekable *seekable,
881 GCancellable *cancellable,
884 GBufferedInputStream *bstream;
885 GBufferedInputStreamPrivate *priv;
886 GInputStream *base_stream;
887 GSeekable *base_stream_seekable;
889 bstream = G_BUFFERED_INPUT_STREAM (seekable);
890 priv = bstream->priv;
892 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
893 if (!G_IS_SEEKABLE (base_stream))
895 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
896 _("Seek not supported on base stream"));
900 base_stream_seekable = G_SEEKABLE (base_stream);
902 if (type == G_SEEK_CUR)
904 if (offset <= priv->end - priv->pos && offset >= -priv->pos)
911 offset -= priv->end - priv->pos;
915 if (g_seekable_seek (base_stream_seekable, offset, type, cancellable, error))
928 g_buffered_input_stream_can_truncate (GSeekable *seekable)
934 g_buffered_input_stream_truncate (GSeekable *seekable,
936 GCancellable *cancellable,
939 g_set_error_literal (error,
941 G_IO_ERROR_NOT_SUPPORTED,
942 _("Cannot truncate GBufferedInputStream"));
947 * g_buffered_input_stream_read_byte:
948 * @stream: a #GBufferedInputStream
949 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
950 * @error: location to store the error occurring, or %NULL to ignore
952 * Tries to read a single byte from the stream or the buffer. Will block
955 * On success, the byte read from the stream is returned. On end of stream
956 * -1 is returned but it's not an exceptional error and @error is not set.
958 * If @cancellable is not %NULL, then the operation can be cancelled by
959 * triggering the cancellable object from another thread. If the operation
960 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
961 * operation was partially finished when the operation was cancelled the
962 * partial result will be returned, without an error.
964 * On error -1 is returned and @error is set accordingly.
966 * Returns: the byte read from the @stream, or -1 on end of stream or error.
969 g_buffered_input_stream_read_byte (GBufferedInputStream *stream,
970 GCancellable *cancellable,
973 GBufferedInputStreamPrivate *priv;
974 GBufferedInputStreamClass *class;
975 GInputStream *input_stream;
979 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
982 input_stream = G_INPUT_STREAM (stream);
984 if (g_input_stream_is_closed (input_stream))
986 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
987 _("Stream is already closed"));
991 if (!g_input_stream_set_pending (input_stream, error))
994 available = priv->end - priv->pos;
998 g_input_stream_clear_pending (input_stream);
999 return priv->buffer[priv->pos++];
1002 /* Byte not available, request refill for more */
1005 g_cancellable_push_current (cancellable);
1010 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1011 nread = class->fill (stream, priv->len, cancellable, error);
1014 g_cancellable_pop_current (cancellable);
1016 g_input_stream_clear_pending (input_stream);
1019 return -1; /* error or end of stream */
1021 return priv->buffer[priv->pos++];
1024 /* ************************** */
1025 /* Async stuff implementation */
1026 /* ************************** */
1029 fill_async_callback (GObject *source_object,
1030 GAsyncResult *result,
1035 GTask *task = user_data;
1038 res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
1041 g_task_return_error (task, error);
1044 GBufferedInputStream *stream;
1045 GBufferedInputStreamPrivate *priv;
1047 stream = g_task_get_source_object (task);
1048 priv = G_BUFFERED_INPUT_STREAM (stream)->priv;
1050 g_assert_cmpint (priv->end + res, <=, priv->len);
1053 g_task_return_int (task, res);
1056 g_object_unref (task);
1060 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
1063 GCancellable *cancellable,
1064 GAsyncReadyCallback callback,
1067 GBufferedInputStreamPrivate *priv;
1068 GInputStream *base_stream;
1072 priv = stream->priv;
1077 in_buffer = priv->end - priv->pos;
1079 /* Never fill more than can fit in the buffer */
1080 count = MIN (count, priv->len - in_buffer);
1082 /* If requested length does not fit at end, compact */
1083 if (priv->len - priv->end < count)
1084 compact_buffer (stream);
1086 task = g_task_new (stream, cancellable, callback, user_data);
1088 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1089 g_input_stream_read_async (base_stream,
1090 priv->buffer + priv->end,
1094 fill_async_callback,
1099 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1100 GAsyncResult *result,
1103 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1105 return g_task_propagate_int (G_TASK (result), error);
1110 gssize bytes_skipped;
1115 free_skip_async_data (gpointer _data)
1117 SkipAsyncData *data = _data;
1118 g_slice_free (SkipAsyncData, data);
1122 large_skip_callback (GObject *source_object,
1123 GAsyncResult *result,
1126 GTask *task = G_TASK (user_data);
1127 SkipAsyncData *data;
1131 data = g_task_get_task_data (task);
1134 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1137 /* Only report the error if we've not already read some data */
1138 if (nread < 0 && data->bytes_skipped == 0)
1139 g_task_return_error (task, error);
1143 g_error_free (error);
1146 data->bytes_skipped += nread;
1148 g_task_return_int (task, data->bytes_skipped);
1151 g_object_unref (task);
1155 skip_fill_buffer_callback (GObject *source_object,
1156 GAsyncResult *result,
1159 GTask *task = G_TASK (user_data);
1160 GBufferedInputStream *bstream;
1161 GBufferedInputStreamPrivate *priv;
1162 SkipAsyncData *data;
1167 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1168 priv = bstream->priv;
1170 data = g_task_get_task_data (task);
1173 nread = g_buffered_input_stream_fill_finish (bstream,
1176 if (nread < 0 && data->bytes_skipped == 0)
1177 g_task_return_error (task, error);
1181 g_error_free (error);
1185 available = priv->end - priv->pos;
1186 data->count = MIN (data->count, available);
1188 data->bytes_skipped += data->count;
1189 priv->pos += data->count;
1192 g_task_return_int (task, data->bytes_skipped);
1195 g_object_unref (task);
1199 g_buffered_input_stream_skip_async (GInputStream *stream,
1202 GCancellable *cancellable,
1203 GAsyncReadyCallback callback,
1206 GBufferedInputStream *bstream;
1207 GBufferedInputStreamPrivate *priv;
1208 GBufferedInputStreamClass *class;
1209 GInputStream *base_stream;
1212 SkipAsyncData *data;
1214 bstream = G_BUFFERED_INPUT_STREAM (stream);
1215 priv = bstream->priv;
1217 data = g_slice_new (SkipAsyncData);
1218 data->bytes_skipped = 0;
1219 task = g_task_new (stream, cancellable, callback, user_data);
1220 g_task_set_task_data (task, data, free_skip_async_data);
1222 available = priv->end - priv->pos;
1224 if (count <= available)
1228 g_task_return_int (task, count);
1229 g_object_unref (task);
1233 /* Full request not available, skip all currently available
1234 * and request refill for more
1242 data->bytes_skipped = available;
1243 data->count = count;
1245 if (count > priv->len)
1247 /* Large request, shortcut buffer */
1249 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1251 g_input_stream_skip_async (base_stream,
1253 io_priority, cancellable,
1254 large_skip_callback,
1259 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1260 class->fill_async (bstream, priv->len, io_priority, cancellable,
1261 skip_fill_buffer_callback, task);
1266 g_buffered_input_stream_skip_finish (GInputStream *stream,
1267 GAsyncResult *result,
1270 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1272 return g_task_propagate_int (G_TASK (result), error);