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 compact_buffer (GBufferedInputStream
*stream
);
134 G_DEFINE_TYPE_WITH_CODE (GBufferedInputStream
,
135 g_buffered_input_stream
,
136 G_TYPE_FILTER_INPUT_STREAM
,
137 G_ADD_PRIVATE (GBufferedInputStream
)
138 G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE
,
139 g_buffered_input_stream_seekable_iface_init
))
142 g_buffered_input_stream_class_init (GBufferedInputStreamClass
*klass
)
144 GObjectClass
*object_class
;
145 GInputStreamClass
*istream_class
;
146 GBufferedInputStreamClass
*bstream_class
;
148 object_class
= G_OBJECT_CLASS (klass
);
149 object_class
->get_property
= g_buffered_input_stream_get_property
;
150 object_class
->set_property
= g_buffered_input_stream_set_property
;
151 object_class
->finalize
= g_buffered_input_stream_finalize
;
153 istream_class
= G_INPUT_STREAM_CLASS (klass
);
154 istream_class
->skip
= g_buffered_input_stream_skip
;
155 istream_class
->skip_async
= g_buffered_input_stream_skip_async
;
156 istream_class
->skip_finish
= g_buffered_input_stream_skip_finish
;
157 istream_class
->read_fn
= g_buffered_input_stream_read
;
159 bstream_class
= G_BUFFERED_INPUT_STREAM_CLASS (klass
);
160 bstream_class
->fill
= g_buffered_input_stream_real_fill
;
161 bstream_class
->fill_async
= g_buffered_input_stream_real_fill_async
;
162 bstream_class
->fill_finish
= g_buffered_input_stream_real_fill_finish
;
164 g_object_class_install_property (object_class
,
166 g_param_spec_uint ("buffer-size",
168 P_("The size of the backend buffer"),
172 G_PARAM_READWRITE
| G_PARAM_CONSTRUCT
|
173 G_PARAM_STATIC_NAME
|G_PARAM_STATIC_NICK
|G_PARAM_STATIC_BLURB
));
179 * g_buffered_input_stream_get_buffer_size:
180 * @stream: a #GBufferedInputStream
182 * Gets the size of the input buffer.
184 * Returns: the current buffer size.
187 g_buffered_input_stream_get_buffer_size (GBufferedInputStream
*stream
)
189 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), 0);
191 return stream
->priv
->len
;
195 * g_buffered_input_stream_set_buffer_size:
196 * @stream: a #GBufferedInputStream
199 * Sets the size of the internal buffer of @stream to @size, or to the
200 * size of the contents of the buffer. The buffer can never be resized
201 * smaller than its current contents.
204 g_buffered_input_stream_set_buffer_size (GBufferedInputStream
*stream
,
207 GBufferedInputStreamPrivate
*priv
;
211 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
));
215 if (priv
->len
== size
)
220 in_buffer
= priv
->end
- priv
->pos
;
222 /* Never resize smaller than current buffer contents */
223 size
= MAX (size
, in_buffer
);
225 buffer
= g_malloc (size
);
226 memcpy (buffer
, priv
->buffer
+ priv
->pos
, in_buffer
);
229 priv
->end
= in_buffer
;
230 g_free (priv
->buffer
);
231 priv
->buffer
= buffer
;
238 priv
->buffer
= g_malloc (size
);
241 g_object_notify (G_OBJECT (stream
), "buffer-size");
245 g_buffered_input_stream_set_property (GObject
*object
,
250 GBufferedInputStream
*bstream
;
252 bstream
= G_BUFFERED_INPUT_STREAM (object
);
257 g_buffered_input_stream_set_buffer_size (bstream
, g_value_get_uint (value
));
261 G_OBJECT_WARN_INVALID_PROPERTY_ID (object
, prop_id
, pspec
);
267 g_buffered_input_stream_get_property (GObject
*object
,
272 GBufferedInputStreamPrivate
*priv
;
273 GBufferedInputStream
*bstream
;
275 bstream
= G_BUFFERED_INPUT_STREAM (object
);
276 priv
= bstream
->priv
;
281 g_value_set_uint (value
, priv
->len
);
285 G_OBJECT_WARN_INVALID_PROPERTY_ID (object
, prop_id
, pspec
);
291 g_buffered_input_stream_finalize (GObject
*object
)
293 GBufferedInputStreamPrivate
*priv
;
294 GBufferedInputStream
*stream
;
296 stream
= G_BUFFERED_INPUT_STREAM (object
);
299 g_free (priv
->buffer
);
301 G_OBJECT_CLASS (g_buffered_input_stream_parent_class
)->finalize (object
);
305 g_buffered_input_stream_seekable_iface_init (GSeekableIface
*iface
)
307 iface
->tell
= g_buffered_input_stream_tell
;
308 iface
->can_seek
= g_buffered_input_stream_can_seek
;
309 iface
->seek
= g_buffered_input_stream_seek
;
310 iface
->can_truncate
= g_buffered_input_stream_can_truncate
;
311 iface
->truncate_fn
= g_buffered_input_stream_truncate
;
315 g_buffered_input_stream_init (GBufferedInputStream
*stream
)
317 stream
->priv
= g_buffered_input_stream_get_instance_private (stream
);
322 * g_buffered_input_stream_new:
323 * @base_stream: a #GInputStream
325 * Creates a new #GInputStream from the given @base_stream, with
326 * a buffer set to the default size (4 kilobytes).
328 * Returns: a #GInputStream for the given @base_stream.
331 g_buffered_input_stream_new (GInputStream
*base_stream
)
333 GInputStream
*stream
;
335 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream
), NULL
);
337 stream
= g_object_new (G_TYPE_BUFFERED_INPUT_STREAM
,
338 "base-stream", base_stream
,
345 * g_buffered_input_stream_new_sized:
346 * @base_stream: a #GInputStream
349 * Creates a new #GBufferedInputStream from the given @base_stream,
350 * with a buffer set to @size.
352 * Returns: a #GInputStream.
355 g_buffered_input_stream_new_sized (GInputStream
*base_stream
,
358 GInputStream
*stream
;
360 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream
), NULL
);
362 stream
= g_object_new (G_TYPE_BUFFERED_INPUT_STREAM
,
363 "base-stream", base_stream
,
364 "buffer-size", (guint
)size
,
371 * g_buffered_input_stream_fill:
372 * @stream: a #GBufferedInputStream
373 * @count: the number of bytes that will be read from the stream
374 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
375 * @error: location to store the error occurring, or %NULL to ignore
377 * Tries to read @count bytes from the stream into the buffer.
378 * Will block during this read.
380 * If @count is zero, returns zero and does nothing. A value of @count
381 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
383 * On success, the number of bytes read into the buffer is returned.
384 * It is not an error if this is not the same as the requested size, as it
385 * can happen e.g. near the end of a file. Zero is returned on end of file
386 * (or if @count is zero), but never otherwise.
388 * If @count is -1 then the attempted read size is equal to the number of
389 * bytes that are required to fill the buffer.
391 * If @cancellable is not %NULL, then the operation can be cancelled by
392 * triggering the cancellable object from another thread. If the operation
393 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
394 * operation was partially finished when the operation was cancelled the
395 * partial result will be returned, without an error.
397 * On error -1 is returned and @error is set accordingly.
399 * For the asynchronous, non-blocking, version of this function, see
400 * g_buffered_input_stream_fill_async().
402 * Returns: the number of bytes read into @stream's buffer, up to @count,
406 g_buffered_input_stream_fill (GBufferedInputStream
*stream
,
408 GCancellable
*cancellable
,
411 GBufferedInputStreamClass
*class;
412 GInputStream
*input_stream
;
415 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), -1);
417 input_stream
= G_INPUT_STREAM (stream
);
421 g_set_error (error
, G_IO_ERROR
, G_IO_ERROR_INVALID_ARGUMENT
,
422 _("Too large count value passed to %s"), G_STRFUNC
);
426 if (!g_input_stream_set_pending (input_stream
, error
))
430 g_cancellable_push_current (cancellable
);
432 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
433 res
= class->fill (stream
, count
, cancellable
, error
);
436 g_cancellable_pop_current (cancellable
);
438 g_input_stream_clear_pending (input_stream
);
444 async_fill_callback_wrapper (GObject
*source_object
,
448 GBufferedInputStream
*stream
= G_BUFFERED_INPUT_STREAM (source_object
);
450 g_input_stream_clear_pending (G_INPUT_STREAM (stream
));
451 (*stream
->priv
->outstanding_callback
) (source_object
, res
, user_data
);
452 g_object_unref (stream
);
456 * g_buffered_input_stream_fill_async:
457 * @stream: a #GBufferedInputStream
458 * @count: the number of bytes that will be read from the stream
459 * @io_priority: the <link linkend="io-priority">I/O priority</link>
461 * @cancellable: (allow-none): optional #GCancellable object
462 * @callback: (scope async): a #GAsyncReadyCallback
463 * @user_data: (closure): a #gpointer
465 * Reads data into @stream's buffer asynchronously, up to @count size.
466 * @io_priority can be used to prioritize reads. For the synchronous
467 * version of this function, see g_buffered_input_stream_fill().
469 * If @count is -1 then the attempted read size is equal to the number
470 * of bytes that are required to fill the buffer.
473 g_buffered_input_stream_fill_async (GBufferedInputStream
*stream
,
476 GCancellable
*cancellable
,
477 GAsyncReadyCallback callback
,
480 GBufferedInputStreamClass
*class;
481 GError
*error
= NULL
;
483 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
));
489 task
= g_task_new (stream
, cancellable
, callback
, user_data
);
490 g_task_set_source_tag (task
, g_buffered_input_stream_fill_async
);
491 g_task_return_int (task
, 0);
492 g_object_unref (task
);
498 g_task_report_new_error (stream
, callback
, user_data
,
499 g_buffered_input_stream_fill_async
,
500 G_IO_ERROR
, G_IO_ERROR_INVALID_ARGUMENT
,
501 _("Too large count value passed to %s"),
506 if (!g_input_stream_set_pending (G_INPUT_STREAM (stream
), &error
))
508 g_task_report_error (stream
, callback
, user_data
,
509 g_buffered_input_stream_fill_async
,
514 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
516 stream
->priv
->outstanding_callback
= callback
;
517 g_object_ref (stream
);
518 class->fill_async (stream
, count
, io_priority
, cancellable
,
519 async_fill_callback_wrapper
, user_data
);
523 * g_buffered_input_stream_fill_finish:
524 * @stream: a #GBufferedInputStream
525 * @result: a #GAsyncResult
528 * Finishes an asynchronous read.
530 * Returns: a #gssize of the read stream, or %-1 on an error.
533 g_buffered_input_stream_fill_finish (GBufferedInputStream
*stream
,
534 GAsyncResult
*result
,
537 GBufferedInputStreamClass
*class;
539 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), -1);
540 g_return_val_if_fail (G_IS_ASYNC_RESULT (result
), -1);
542 if (g_async_result_legacy_propagate_error (result
, error
))
544 else if (g_async_result_is_tagged (result
, g_buffered_input_stream_fill_async
))
545 return g_task_propagate_int (G_TASK (result
), error
);
547 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
548 return class->fill_finish (stream
, result
, error
);
552 * g_buffered_input_stream_get_available:
553 * @stream: #GBufferedInputStream
555 * Gets the size of the available data within the stream.
557 * Returns: size of the available stream.
560 g_buffered_input_stream_get_available (GBufferedInputStream
*stream
)
562 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), -1);
564 return stream
->priv
->end
- stream
->priv
->pos
;
568 * g_buffered_input_stream_peek:
569 * @stream: a #GBufferedInputStream
570 * @buffer: (array length=count) (element-type guint8): a pointer to
571 * an allocated chunk of memory
575 * Peeks in the buffer, copying data of size @count into @buffer,
576 * offset @offset bytes.
578 * Returns: a #gsize of the number of bytes peeked, or -1 on error.
581 g_buffered_input_stream_peek (GBufferedInputStream
*stream
,
589 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), -1);
590 g_return_val_if_fail (buffer
!= NULL
, -1);
592 available
= g_buffered_input_stream_get_available (stream
);
594 if (offset
> available
)
597 end
= MIN (offset
+ count
, available
);
598 count
= end
- offset
;
600 memcpy (buffer
, stream
->priv
->buffer
+ stream
->priv
->pos
+ offset
, count
);
605 * g_buffered_input_stream_peek_buffer:
606 * @stream: a #GBufferedInputStream
607 * @count: (out): a #gsize to get the number of bytes available in the buffer
609 * Returns the buffer with the currently available bytes. The returned
610 * buffer must not be modified and will become invalid when reading from
611 * the stream or filling the buffer.
613 * Returns: (array length=count) (element-type guint8) (transfer none):
617 g_buffered_input_stream_peek_buffer (GBufferedInputStream
*stream
,
620 GBufferedInputStreamPrivate
*priv
;
622 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), NULL
);
627 *count
= priv
->end
- priv
->pos
;
629 return priv
->buffer
+ priv
->pos
;
633 compact_buffer (GBufferedInputStream
*stream
)
635 GBufferedInputStreamPrivate
*priv
;
640 current_size
= priv
->end
- priv
->pos
;
642 g_memmove (priv
->buffer
, priv
->buffer
+ priv
->pos
, current_size
);
645 priv
->end
= current_size
;
649 g_buffered_input_stream_real_fill (GBufferedInputStream
*stream
,
651 GCancellable
*cancellable
,
654 GBufferedInputStreamPrivate
*priv
;
655 GInputStream
*base_stream
;
664 in_buffer
= priv
->end
- priv
->pos
;
666 /* Never fill more than can fit in the buffer */
667 count
= MIN (count
, priv
->len
- in_buffer
);
669 /* If requested length does not fit at end, compact */
670 if (priv
->len
- priv
->end
< count
)
671 compact_buffer (stream
);
673 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
674 nread
= g_input_stream_read (base_stream
,
675 priv
->buffer
+ priv
->end
,
687 g_buffered_input_stream_skip (GInputStream
*stream
,
689 GCancellable
*cancellable
,
692 GBufferedInputStream
*bstream
;
693 GBufferedInputStreamPrivate
*priv
;
694 GBufferedInputStreamClass
*class;
695 GInputStream
*base_stream
;
696 gsize available
, bytes_skipped
;
699 bstream
= G_BUFFERED_INPUT_STREAM (stream
);
700 priv
= bstream
->priv
;
702 available
= priv
->end
- priv
->pos
;
704 if (count
<= available
)
710 /* Full request not available, skip all currently available and
711 * request refill for more
716 bytes_skipped
= available
;
719 if (bytes_skipped
> 0)
720 error
= NULL
; /* Ignore further errors if we already read some data */
722 if (count
> priv
->len
)
724 /* Large request, shortcut buffer */
726 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
728 nread
= g_input_stream_skip (base_stream
,
733 if (nread
< 0 && bytes_skipped
== 0)
737 bytes_skipped
+= nread
;
739 return bytes_skipped
;
742 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
743 nread
= class->fill (bstream
, priv
->len
, cancellable
, error
);
747 if (bytes_skipped
== 0)
750 return bytes_skipped
;
753 available
= priv
->end
- priv
->pos
;
754 count
= MIN (count
, available
);
756 bytes_skipped
+= count
;
759 return bytes_skipped
;
763 g_buffered_input_stream_read (GInputStream
*stream
,
766 GCancellable
*cancellable
,
769 GBufferedInputStream
*bstream
;
770 GBufferedInputStreamPrivate
*priv
;
771 GBufferedInputStreamClass
*class;
772 GInputStream
*base_stream
;
773 gsize available
, bytes_read
;
776 bstream
= G_BUFFERED_INPUT_STREAM (stream
);
777 priv
= bstream
->priv
;
779 available
= priv
->end
- priv
->pos
;
781 if (count
<= available
)
783 memcpy (buffer
, priv
->buffer
+ priv
->pos
, count
);
788 /* Full request not available, read all currently available and
789 * request refill for more
792 memcpy (buffer
, priv
->buffer
+ priv
->pos
, available
);
795 bytes_read
= available
;
799 error
= NULL
; /* Ignore further errors if we already read some data */
801 if (count
> priv
->len
)
803 /* Large request, shortcut buffer */
805 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
807 nread
= g_input_stream_read (base_stream
,
808 (char *)buffer
+ bytes_read
,
813 if (nread
< 0 && bytes_read
== 0)
822 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
823 nread
= class->fill (bstream
, priv
->len
, cancellable
, error
);
832 available
= priv
->end
- priv
->pos
;
833 count
= MIN (count
, available
);
835 memcpy ((char *)buffer
+ bytes_read
, (char *)priv
->buffer
+ priv
->pos
, count
);
843 g_buffered_input_stream_tell (GSeekable
*seekable
)
845 GBufferedInputStream
*bstream
;
846 GBufferedInputStreamPrivate
*priv
;
847 GInputStream
*base_stream
;
848 GSeekable
*base_stream_seekable
;
852 bstream
= G_BUFFERED_INPUT_STREAM (seekable
);
853 priv
= bstream
->priv
;
855 base_stream
= G_FILTER_INPUT_STREAM (seekable
)->base_stream
;
856 if (!G_IS_SEEKABLE (base_stream
))
858 base_stream_seekable
= G_SEEKABLE (base_stream
);
860 available
= priv
->end
- priv
->pos
;
861 base_offset
= g_seekable_tell (base_stream_seekable
);
863 return base_offset
- available
;
867 g_buffered_input_stream_can_seek (GSeekable
*seekable
)
869 GInputStream
*base_stream
;
871 base_stream
= G_FILTER_INPUT_STREAM (seekable
)->base_stream
;
872 return G_IS_SEEKABLE (base_stream
) && g_seekable_can_seek (G_SEEKABLE (base_stream
));
876 g_buffered_input_stream_seek (GSeekable
*seekable
,
879 GCancellable
*cancellable
,
882 GBufferedInputStream
*bstream
;
883 GBufferedInputStreamPrivate
*priv
;
884 GInputStream
*base_stream
;
885 GSeekable
*base_stream_seekable
;
887 bstream
= G_BUFFERED_INPUT_STREAM (seekable
);
888 priv
= bstream
->priv
;
890 base_stream
= G_FILTER_INPUT_STREAM (seekable
)->base_stream
;
891 if (!G_IS_SEEKABLE (base_stream
))
893 g_set_error_literal (error
, G_IO_ERROR
, G_IO_ERROR_NOT_SUPPORTED
,
894 _("Seek not supported on base stream"));
898 base_stream_seekable
= G_SEEKABLE (base_stream
);
900 if (type
== G_SEEK_CUR
)
902 if (offset
<= priv
->end
- priv
->pos
&& offset
>= -priv
->pos
)
909 offset
-= priv
->end
- priv
->pos
;
913 if (g_seekable_seek (base_stream_seekable
, offset
, type
, cancellable
, error
))
926 g_buffered_input_stream_can_truncate (GSeekable
*seekable
)
932 g_buffered_input_stream_truncate (GSeekable
*seekable
,
934 GCancellable
*cancellable
,
937 g_set_error_literal (error
,
939 G_IO_ERROR_NOT_SUPPORTED
,
940 _("Cannot truncate GBufferedInputStream"));
945 * g_buffered_input_stream_read_byte:
946 * @stream: a #GBufferedInputStream
947 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
948 * @error: location to store the error occurring, or %NULL to ignore
950 * Tries to read a single byte from the stream or the buffer. Will block
953 * On success, the byte read from the stream is returned. On end of stream
954 * -1 is returned but it's not an exceptional error and @error is not set.
956 * If @cancellable is not %NULL, then the operation can be cancelled by
957 * triggering the cancellable object from another thread. If the operation
958 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
959 * operation was partially finished when the operation was cancelled the
960 * partial result will be returned, without an error.
962 * On error -1 is returned and @error is set accordingly.
964 * Returns: the byte read from the @stream, or -1 on end of stream or error.
967 g_buffered_input_stream_read_byte (GBufferedInputStream
*stream
,
968 GCancellable
*cancellable
,
971 GBufferedInputStreamPrivate
*priv
;
972 GBufferedInputStreamClass
*class;
973 GInputStream
*input_stream
;
977 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), -1);
980 input_stream
= G_INPUT_STREAM (stream
);
982 if (g_input_stream_is_closed (input_stream
))
984 g_set_error_literal (error
, G_IO_ERROR
, G_IO_ERROR_CLOSED
,
985 _("Stream is already closed"));
989 if (!g_input_stream_set_pending (input_stream
, error
))
992 available
= priv
->end
- priv
->pos
;
996 g_input_stream_clear_pending (input_stream
);
997 return priv
->buffer
[priv
->pos
++];
1000 /* Byte not available, request refill for more */
1003 g_cancellable_push_current (cancellable
);
1008 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
1009 nread
= class->fill (stream
, priv
->len
, cancellable
, error
);
1012 g_cancellable_pop_current (cancellable
);
1014 g_input_stream_clear_pending (input_stream
);
1017 return -1; /* error or end of stream */
1019 return priv
->buffer
[priv
->pos
++];
1022 /* ************************** */
1023 /* Async stuff implementation */
1024 /* ************************** */
1027 fill_async_callback (GObject
*source_object
,
1028 GAsyncResult
*result
,
1033 GTask
*task
= user_data
;
1036 res
= g_input_stream_read_finish (G_INPUT_STREAM (source_object
),
1039 g_task_return_error (task
, error
);
1042 GBufferedInputStream
*stream
;
1043 GBufferedInputStreamPrivate
*priv
;
1045 stream
= g_task_get_source_object (task
);
1046 priv
= G_BUFFERED_INPUT_STREAM (stream
)->priv
;
1048 g_assert_cmpint (priv
->end
+ res
, <=, priv
->len
);
1051 g_task_return_int (task
, res
);
1054 g_object_unref (task
);
1058 g_buffered_input_stream_real_fill_async (GBufferedInputStream
*stream
,
1061 GCancellable
*cancellable
,
1062 GAsyncReadyCallback callback
,
1065 GBufferedInputStreamPrivate
*priv
;
1066 GInputStream
*base_stream
;
1070 priv
= stream
->priv
;
1075 in_buffer
= priv
->end
- priv
->pos
;
1077 /* Never fill more than can fit in the buffer */
1078 count
= MIN (count
, priv
->len
- in_buffer
);
1080 /* If requested length does not fit at end, compact */
1081 if (priv
->len
- priv
->end
< count
)
1082 compact_buffer (stream
);
1084 task
= g_task_new (stream
, cancellable
, callback
, user_data
);
1086 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
1087 g_input_stream_read_async (base_stream
,
1088 priv
->buffer
+ priv
->end
,
1092 fill_async_callback
,
1097 g_buffered_input_stream_real_fill_finish (GBufferedInputStream
*stream
,
1098 GAsyncResult
*result
,
1101 g_return_val_if_fail (g_task_is_valid (result
, stream
), -1);
1103 return g_task_propagate_int (G_TASK (result
), error
);
1108 gssize bytes_skipped
;
1113 free_skip_async_data (gpointer _data
)
1115 SkipAsyncData
*data
= _data
;
1116 g_slice_free (SkipAsyncData
, data
);
1120 large_skip_callback (GObject
*source_object
,
1121 GAsyncResult
*result
,
1124 GTask
*task
= G_TASK (user_data
);
1125 SkipAsyncData
*data
;
1129 data
= g_task_get_task_data (task
);
1132 nread
= g_input_stream_skip_finish (G_INPUT_STREAM (source_object
),
1135 /* Only report the error if we've not already read some data */
1136 if (nread
< 0 && data
->bytes_skipped
== 0)
1137 g_task_return_error (task
, error
);
1141 g_error_free (error
);
1144 data
->bytes_skipped
+= nread
;
1146 g_task_return_int (task
, data
->bytes_skipped
);
1149 g_object_unref (task
);
1153 skip_fill_buffer_callback (GObject
*source_object
,
1154 GAsyncResult
*result
,
1157 GTask
*task
= G_TASK (user_data
);
1158 GBufferedInputStream
*bstream
;
1159 GBufferedInputStreamPrivate
*priv
;
1160 SkipAsyncData
*data
;
1165 bstream
= G_BUFFERED_INPUT_STREAM (source_object
);
1166 priv
= bstream
->priv
;
1168 data
= g_task_get_task_data (task
);
1171 nread
= g_buffered_input_stream_fill_finish (bstream
,
1174 if (nread
< 0 && data
->bytes_skipped
== 0)
1175 g_task_return_error (task
, error
);
1179 g_error_free (error
);
1183 available
= priv
->end
- priv
->pos
;
1184 data
->count
= MIN (data
->count
, available
);
1186 data
->bytes_skipped
+= data
->count
;
1187 priv
->pos
+= data
->count
;
1190 g_task_return_int (task
, data
->bytes_skipped
);
1193 g_object_unref (task
);
1197 g_buffered_input_stream_skip_async (GInputStream
*stream
,
1200 GCancellable
*cancellable
,
1201 GAsyncReadyCallback callback
,
1204 GBufferedInputStream
*bstream
;
1205 GBufferedInputStreamPrivate
*priv
;
1206 GBufferedInputStreamClass
*class;
1207 GInputStream
*base_stream
;
1210 SkipAsyncData
*data
;
1212 bstream
= G_BUFFERED_INPUT_STREAM (stream
);
1213 priv
= bstream
->priv
;
1215 data
= g_slice_new (SkipAsyncData
);
1216 data
->bytes_skipped
= 0;
1217 task
= g_task_new (stream
, cancellable
, callback
, user_data
);
1218 g_task_set_task_data (task
, data
, free_skip_async_data
);
1220 available
= priv
->end
- priv
->pos
;
1222 if (count
<= available
)
1226 g_task_return_int (task
, count
);
1227 g_object_unref (task
);
1231 /* Full request not available, skip all currently available
1232 * and request refill for more
1240 data
->bytes_skipped
= available
;
1241 data
->count
= count
;
1243 if (count
> priv
->len
)
1245 /* Large request, shortcut buffer */
1247 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
1249 g_input_stream_skip_async (base_stream
,
1251 io_priority
, cancellable
,
1252 large_skip_callback
,
1257 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
1258 class->fill_async (bstream
, priv
->len
, io_priority
, cancellable
,
1259 skip_fill_buffer_callback
, task
);
1264 g_buffered_input_stream_skip_finish (GInputStream
*stream
,
1265 GAsyncResult
*result
,
1268 g_return_val_if_fail (g_task_is_valid (result
, stream
), -1);
1270 return g_task_propagate_int (G_TASK (result
), error
);