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"
36 * SECTION:gbufferedinputstream
37 * @short_description: Buffered Input Stream
39 * @see_also: #GFilterInputStream, #GInputStream
41 * Buffered input stream implements #GFilterInputStream and provides
44 * By default, #GBufferedInputStream's buffer size is set at 4 kilobytes.
46 * To create a buffered input stream, use g_buffered_input_stream_new(),
47 * or g_buffered_input_stream_new_sized() to specify the buffer's size at
50 * To get the size of a buffer within a buffered input stream, use
51 * g_buffered_input_stream_get_buffer_size(). To change the size of a
52 * buffered input stream's buffer, use
53 * g_buffered_input_stream_set_buffer_size(). Note that the buffer's size
54 * cannot be reduced below the size of the data within the buffer.
58 #define DEFAULT_BUFFER_SIZE 4096
60 struct _GBufferedInputStreamPrivate {
65 GAsyncReadyCallback outstanding_callback;
73 static void g_buffered_input_stream_set_property (GObject *object,
78 static void g_buffered_input_stream_get_property (GObject *object,
82 static void g_buffered_input_stream_finalize (GObject *object);
85 static gssize g_buffered_input_stream_skip (GInputStream *stream,
87 GCancellable *cancellable,
89 static void g_buffered_input_stream_skip_async (GInputStream *stream,
92 GCancellable *cancellable,
93 GAsyncReadyCallback callback,
95 static gssize g_buffered_input_stream_skip_finish (GInputStream *stream,
98 static gssize g_buffered_input_stream_read (GInputStream *stream,
101 GCancellable *cancellable,
103 static void g_buffered_input_stream_read_async (GInputStream *stream,
107 GCancellable *cancellable,
108 GAsyncReadyCallback callback,
110 static gssize g_buffered_input_stream_read_finish (GInputStream *stream,
111 GAsyncResult *result,
113 static gssize g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
115 GCancellable *cancellable,
117 static void g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
120 GCancellable *cancellable,
121 GAsyncReadyCallback callback,
123 static gssize g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
124 GAsyncResult *result,
127 static void compact_buffer (GBufferedInputStream *stream);
129 G_DEFINE_TYPE (GBufferedInputStream,
130 g_buffered_input_stream,
131 G_TYPE_FILTER_INPUT_STREAM)
135 g_buffered_input_stream_class_init (GBufferedInputStreamClass *klass)
137 GObjectClass *object_class;
138 GInputStreamClass *istream_class;
139 GBufferedInputStreamClass *bstream_class;
141 g_type_class_add_private (klass, sizeof (GBufferedInputStreamPrivate));
143 object_class = G_OBJECT_CLASS (klass);
144 object_class->get_property = g_buffered_input_stream_get_property;
145 object_class->set_property = g_buffered_input_stream_set_property;
146 object_class->finalize = g_buffered_input_stream_finalize;
148 istream_class = G_INPUT_STREAM_CLASS (klass);
149 istream_class->skip = g_buffered_input_stream_skip;
150 istream_class->skip_async = g_buffered_input_stream_skip_async;
151 istream_class->skip_finish = g_buffered_input_stream_skip_finish;
152 istream_class->read_fn = g_buffered_input_stream_read;
153 istream_class->read_async = g_buffered_input_stream_read_async;
154 istream_class->read_finish = g_buffered_input_stream_read_finish;
156 bstream_class = G_BUFFERED_INPUT_STREAM_CLASS (klass);
157 bstream_class->fill = g_buffered_input_stream_real_fill;
158 bstream_class->fill_async = g_buffered_input_stream_real_fill_async;
159 bstream_class->fill_finish = g_buffered_input_stream_real_fill_finish;
161 g_object_class_install_property (object_class,
163 g_param_spec_uint ("buffer-size",
165 P_("The size of the backend buffer"),
169 G_PARAM_READWRITE | G_PARAM_CONSTRUCT |
170 G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
176 * g_buffered_input_stream_get_buffer_size:
177 * @stream: a #GBufferedInputStream
179 * Gets the size of the input buffer.
181 * Returns: the current buffer size.
184 g_buffered_input_stream_get_buffer_size (GBufferedInputStream *stream)
186 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), 0);
188 return stream->priv->len;
192 * g_buffered_input_stream_set_buffer_size:
193 * @stream: a #GBufferedInputStream
196 * Sets the size of the internal buffer of @stream to @size, or to the
197 * size of the contents of the buffer. The buffer can never be resized
198 * smaller than its current contents.
201 g_buffered_input_stream_set_buffer_size (GBufferedInputStream *stream,
204 GBufferedInputStreamPrivate *priv;
208 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
212 if (priv->len == size)
217 in_buffer = priv->end - priv->pos;
219 /* Never resize smaller than current buffer contents */
220 size = MAX (size, in_buffer);
222 buffer = g_malloc (size);
223 memcpy (buffer, priv->buffer + priv->pos, in_buffer);
226 priv->end = in_buffer;
227 g_free (priv->buffer);
228 priv->buffer = buffer;
235 priv->buffer = g_malloc (size);
238 g_object_notify (G_OBJECT (stream), "buffer-size");
242 g_buffered_input_stream_set_property (GObject *object,
247 GBufferedInputStream *bstream;
249 bstream = G_BUFFERED_INPUT_STREAM (object);
254 g_buffered_input_stream_set_buffer_size (bstream, g_value_get_uint (value));
258 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
264 g_buffered_input_stream_get_property (GObject *object,
269 GBufferedInputStreamPrivate *priv;
270 GBufferedInputStream *bstream;
272 bstream = G_BUFFERED_INPUT_STREAM (object);
273 priv = bstream->priv;
278 g_value_set_uint (value, priv->len);
282 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
288 g_buffered_input_stream_finalize (GObject *object)
290 GBufferedInputStreamPrivate *priv;
291 GBufferedInputStream *stream;
293 stream = G_BUFFERED_INPUT_STREAM (object);
296 g_free (priv->buffer);
298 G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize (object);
302 g_buffered_input_stream_init (GBufferedInputStream *stream)
304 stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
305 G_TYPE_BUFFERED_INPUT_STREAM,
306 GBufferedInputStreamPrivate);
311 * g_buffered_input_stream_new:
312 * @base_stream: a #GInputStream
314 * Creates a new #GInputStream from the given @base_stream, with
315 * a buffer set to the default size (4 kilobytes).
317 * Returns: a #GInputStream for the given @base_stream.
320 g_buffered_input_stream_new (GInputStream *base_stream)
322 GInputStream *stream;
324 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
326 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
327 "base-stream", base_stream,
334 * g_buffered_input_stream_new_sized:
335 * @base_stream: a #GInputStream
338 * Creates a new #GBufferedInputStream from the given @base_stream,
339 * with a buffer set to @size.
341 * Returns: a #GInputStream.
344 g_buffered_input_stream_new_sized (GInputStream *base_stream,
347 GInputStream *stream;
349 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
351 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
352 "base-stream", base_stream,
353 "buffer-size", (guint)size,
360 * g_buffered_input_stream_fill:
361 * @stream: a #GBufferedInputStream
362 * @count: the number of bytes that will be read from the stream
363 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
364 * @error: location to store the error occurring, or %NULL to ignore
366 * Tries to read @count bytes from the stream into the buffer.
367 * Will block during this read.
369 * If @count is zero, returns zero and does nothing. A value of @count
370 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
372 * On success, the number of bytes read into the buffer is returned.
373 * It is not an error if this is not the same as the requested size, as it
374 * can happen e.g. near the end of a file. Zero is returned on end of file
375 * (or if @count is zero), but never otherwise.
377 * If @count is -1 then the attempted read size is equal to the number of
378 * bytes that are required to fill the buffer.
380 * If @cancellable is not %NULL, then the operation can be cancelled by
381 * triggering the cancellable object from another thread. If the operation
382 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
383 * operation was partially finished when the operation was cancelled the
384 * partial result will be returned, without an error.
386 * On error -1 is returned and @error is set accordingly.
388 * For the asynchronous, non-blocking, version of this function, see
389 * g_buffered_input_stream_fill_async().
391 * Returns: the number of bytes read into @stream's buffer, up to @count,
395 g_buffered_input_stream_fill (GBufferedInputStream *stream,
397 GCancellable *cancellable,
400 GBufferedInputStreamClass *class;
401 GInputStream *input_stream;
404 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
406 input_stream = G_INPUT_STREAM (stream);
410 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
411 _("Too large count value passed to %s"), G_STRFUNC);
415 if (!g_input_stream_set_pending (input_stream, error))
419 g_cancellable_push_current (cancellable);
421 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
422 res = class->fill (stream, count, cancellable, error);
425 g_cancellable_pop_current (cancellable);
427 g_input_stream_clear_pending (input_stream);
433 async_fill_callback_wrapper (GObject *source_object,
437 GBufferedInputStream *stream = G_BUFFERED_INPUT_STREAM (source_object);
439 g_input_stream_clear_pending (G_INPUT_STREAM (stream));
440 (*stream->priv->outstanding_callback) (source_object, res, user_data);
441 g_object_unref (stream);
445 * g_buffered_input_stream_fill_async:
446 * @stream: a #GBufferedInputStream
447 * @count: the number of bytes that will be read from the stream
448 * @io_priority: the <link linkend="io-priority">I/O priority</link>
450 * @cancellable: (allow-none): optional #GCancellable object
451 * @callback: (scope async): a #GAsyncReadyCallback
452 * @user_data: (closure): a #gpointer
454 * Reads data into @stream's buffer asynchronously, up to @count size.
455 * @io_priority can be used to prioritize reads. For the synchronous
456 * version of this function, see g_buffered_input_stream_fill().
458 * If @count is -1 then the attempted read size is equal to the number
459 * of bytes that are required to fill the buffer.
462 g_buffered_input_stream_fill_async (GBufferedInputStream *stream,
465 GCancellable *cancellable,
466 GAsyncReadyCallback callback,
469 GBufferedInputStreamClass *class;
470 GSimpleAsyncResult *simple;
471 GError *error = NULL;
473 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
477 simple = g_simple_async_result_new (G_OBJECT (stream),
480 g_buffered_input_stream_fill_async);
481 g_simple_async_result_complete_in_idle (simple);
482 g_object_unref (simple);
488 g_simple_async_report_error_in_idle (G_OBJECT (stream),
491 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
492 _("Too large count value passed to %s"),
497 if (!g_input_stream_set_pending (G_INPUT_STREAM (stream), &error))
499 g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
506 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
508 stream->priv->outstanding_callback = callback;
509 g_object_ref (stream);
510 class->fill_async (stream, count, io_priority, cancellable,
511 async_fill_callback_wrapper, user_data);
515 * g_buffered_input_stream_fill_finish:
516 * @stream: a #GBufferedInputStream
517 * @result: a #GAsyncResult
520 * Finishes an asynchronous read.
522 * Returns: a #gssize of the read stream, or %-1 on an error.
525 g_buffered_input_stream_fill_finish (GBufferedInputStream *stream,
526 GAsyncResult *result,
529 GSimpleAsyncResult *simple;
530 GBufferedInputStreamClass *class;
532 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
533 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
535 if (G_IS_SIMPLE_ASYNC_RESULT (result))
537 simple = G_SIMPLE_ASYNC_RESULT (result);
538 if (g_simple_async_result_propagate_error (simple, error))
541 /* Special case read of 0 bytes */
542 if (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_fill_async)
546 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
547 return class->fill_finish (stream, result, error);
551 * g_buffered_input_stream_get_available:
552 * @stream: #GBufferedInputStream
554 * Gets the size of the available data within the stream.
556 * Returns: size of the available stream.
559 g_buffered_input_stream_get_available (GBufferedInputStream *stream)
561 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
563 return stream->priv->end - stream->priv->pos;
567 * g_buffered_input_stream_peek:
568 * @stream: a #GBufferedInputStream
569 * @buffer: a pointer to an allocated chunk of memory
573 * Peeks in the buffer, copying data of size @count into @buffer,
574 * offset @offset bytes.
576 * Returns: a #gsize of the number of bytes peeked, or -1 on error.
579 g_buffered_input_stream_peek (GBufferedInputStream *stream,
587 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
588 g_return_val_if_fail (buffer != NULL, -1);
590 available = g_buffered_input_stream_get_available (stream);
592 if (offset > available)
595 end = MIN (offset + count, available);
596 count = end - offset;
598 memcpy (buffer, stream->priv->buffer + stream->priv->pos + offset, count);
603 * g_buffered_input_stream_peek_buffer:
604 * @stream: a #GBufferedInputStream
605 * @count: (out): a #gsize to get the number of bytes available in the buffer
607 * Returns the buffer with the currently available bytes. The returned
608 * buffer must not be modified and will become invalid when reading from
609 * the stream or filling the buffer.
611 * Returns: (array length=count) (element-type guint8) (transfer none):
615 g_buffered_input_stream_peek_buffer (GBufferedInputStream *stream,
618 GBufferedInputStreamPrivate *priv;
620 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), NULL);
625 *count = priv->end - priv->pos;
627 return priv->buffer + priv->pos;
631 compact_buffer (GBufferedInputStream *stream)
633 GBufferedInputStreamPrivate *priv;
638 current_size = priv->end - priv->pos;
640 g_memmove (priv->buffer, priv->buffer + priv->pos, current_size);
643 priv->end = current_size;
647 g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
649 GCancellable *cancellable,
652 GBufferedInputStreamPrivate *priv;
653 GInputStream *base_stream;
662 in_buffer = priv->end - priv->pos;
664 /* Never fill more than can fit in the buffer */
665 count = MIN (count, priv->len - in_buffer);
667 /* If requested length does not fit at end, compact */
668 if (priv->len - priv->end < count)
669 compact_buffer (stream);
671 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
672 nread = g_input_stream_read (base_stream,
673 priv->buffer + priv->end,
685 g_buffered_input_stream_skip (GInputStream *stream,
687 GCancellable *cancellable,
690 GBufferedInputStream *bstream;
691 GBufferedInputStreamPrivate *priv;
692 GBufferedInputStreamClass *class;
693 GInputStream *base_stream;
694 gsize available, bytes_skipped;
697 bstream = G_BUFFERED_INPUT_STREAM (stream);
698 priv = bstream->priv;
700 available = priv->end - priv->pos;
702 if (count <= available)
708 /* Full request not available, skip all currently available and
709 * request refill for more
714 bytes_skipped = available;
717 if (bytes_skipped > 0)
718 error = NULL; /* Ignore further errors if we already read some data */
720 if (count > priv->len)
722 /* Large request, shortcut buffer */
724 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
726 nread = g_input_stream_skip (base_stream,
731 if (nread < 0 && bytes_skipped == 0)
735 bytes_skipped += nread;
737 return bytes_skipped;
740 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
741 nread = class->fill (bstream, priv->len, cancellable, error);
745 if (bytes_skipped == 0)
748 return bytes_skipped;
751 available = priv->end - priv->pos;
752 count = MIN (count, available);
754 bytes_skipped += count;
757 return bytes_skipped;
761 g_buffered_input_stream_read (GInputStream *stream,
764 GCancellable *cancellable,
767 GBufferedInputStream *bstream;
768 GBufferedInputStreamPrivate *priv;
769 GBufferedInputStreamClass *class;
770 GInputStream *base_stream;
771 gsize available, bytes_read;
774 bstream = G_BUFFERED_INPUT_STREAM (stream);
775 priv = bstream->priv;
777 available = priv->end - priv->pos;
779 if (count <= available)
781 memcpy (buffer, priv->buffer + priv->pos, count);
786 /* Full request not available, read all currently available and
787 * request refill for more
790 memcpy (buffer, priv->buffer + priv->pos, available);
793 bytes_read = available;
797 error = NULL; /* Ignore further errors if we already read some data */
799 if (count > priv->len)
801 /* Large request, shortcut buffer */
803 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
805 nread = g_input_stream_read (base_stream,
806 (char *)buffer + bytes_read,
811 if (nread < 0 && bytes_read == 0)
820 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
821 nread = class->fill (bstream, priv->len, cancellable, error);
830 available = priv->end - priv->pos;
831 count = MIN (count, available);
833 memcpy ((char *)buffer + bytes_read, (char *)priv->buffer + priv->pos, count);
841 * g_buffered_input_stream_read_byte:
842 * @stream: a #GBufferedInputStream
843 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
844 * @error: location to store the error occurring, or %NULL to ignore
846 * Tries to read a single byte from the stream or the buffer. Will block
849 * On success, the byte read from the stream is returned. On end of stream
850 * -1 is returned but it's not an exceptional error and @error is not set.
852 * If @cancellable is not %NULL, then the operation can be cancelled by
853 * triggering the cancellable object from another thread. If the operation
854 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
855 * operation was partially finished when the operation was cancelled the
856 * partial result will be returned, without an error.
858 * On error -1 is returned and @error is set accordingly.
860 * Returns: the byte read from the @stream, or -1 on end of stream or error.
863 g_buffered_input_stream_read_byte (GBufferedInputStream *stream,
864 GCancellable *cancellable,
867 GBufferedInputStreamPrivate *priv;
868 GBufferedInputStreamClass *class;
869 GInputStream *input_stream;
873 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
876 input_stream = G_INPUT_STREAM (stream);
878 if (g_input_stream_is_closed (input_stream))
880 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
881 _("Stream is already closed"));
885 if (!g_input_stream_set_pending (input_stream, error))
888 available = priv->end - priv->pos;
892 g_input_stream_clear_pending (input_stream);
893 return priv->buffer[priv->pos++];
896 /* Byte not available, request refill for more */
899 g_cancellable_push_current (cancellable);
904 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
905 nread = class->fill (stream, priv->len, cancellable, error);
908 g_cancellable_pop_current (cancellable);
910 g_input_stream_clear_pending (input_stream);
913 return -1; /* error or end of stream */
915 return priv->buffer[priv->pos++];
918 /* ************************** */
919 /* Async stuff implementation */
920 /* ************************** */
923 fill_async_callback (GObject *source_object,
924 GAsyncResult *result,
929 GSimpleAsyncResult *simple;
934 res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
937 g_simple_async_result_set_op_res_gssize (simple, res);
940 g_simple_async_result_take_error (simple, error);
944 GBufferedInputStreamPrivate *priv;
947 object = g_async_result_get_source_object (G_ASYNC_RESULT (simple));
948 priv = G_BUFFERED_INPUT_STREAM (object)->priv;
950 g_assert_cmpint (priv->end + res, <=, priv->len);
953 g_object_unref (object);
956 /* Complete immediately, not in idle, since we're already
957 * in a mainloop callout
959 g_simple_async_result_complete (simple);
960 g_object_unref (simple);
964 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
967 GCancellable *cancellable,
968 GAsyncReadyCallback callback,
971 GBufferedInputStreamPrivate *priv;
972 GInputStream *base_stream;
973 GSimpleAsyncResult *simple;
981 in_buffer = priv->end - priv->pos;
983 /* Never fill more than can fit in the buffer */
984 count = MIN (count, priv->len - in_buffer);
986 /* If requested length does not fit at end, compact */
987 if (priv->len - priv->end < count)
988 compact_buffer (stream);
990 simple = g_simple_async_result_new (G_OBJECT (stream),
992 g_buffered_input_stream_real_fill_async);
994 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
995 g_input_stream_read_async (base_stream,
996 priv->buffer + priv->end,
1000 fill_async_callback,
1005 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1006 GAsyncResult *result,
1009 GSimpleAsyncResult *simple;
1012 simple = G_SIMPLE_ASYNC_RESULT (result);
1013 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_real_fill_async);
1015 nread = g_simple_async_result_get_op_res_gssize (simple);
1027 free_read_async_data (gpointer _data)
1029 ReadAsyncData *data = _data;
1030 g_slice_free (ReadAsyncData, data);
1034 large_read_callback (GObject *source_object,
1035 GAsyncResult *result,
1038 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1039 ReadAsyncData *data;
1043 data = g_simple_async_result_get_op_res_gpointer (simple);
1046 nread = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
1049 /* Only report the error if we've not already read some data */
1050 if (nread < 0 && data->bytes_read == 0)
1051 g_simple_async_result_take_error (simple, error);
1053 g_error_free (error);
1056 data->bytes_read += nread;
1058 /* Complete immediately, not in idle, since we're already
1059 * in a mainloop callout
1061 g_simple_async_result_complete (simple);
1062 g_object_unref (simple);
1066 read_fill_buffer_callback (GObject *source_object,
1067 GAsyncResult *result,
1070 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1071 GBufferedInputStream *bstream;
1072 GBufferedInputStreamPrivate *priv;
1073 ReadAsyncData *data;
1078 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1079 priv = bstream->priv;
1081 data = g_simple_async_result_get_op_res_gpointer (simple);
1084 nread = g_buffered_input_stream_fill_finish (bstream,
1087 if (nread < 0 && data->bytes_read == 0)
1088 g_simple_async_result_take_error (simple, error);
1090 g_error_free (error);
1094 available = priv->end - priv->pos;
1095 data->count = MIN (data->count, available);
1097 memcpy ((char *)data->buffer + data->bytes_read, (char *)priv->buffer + priv->pos, data->count);
1098 data->bytes_read += data->count;
1099 priv->pos += data->count;
1102 /* Complete immediately, not in idle, since we're already
1103 * in a mainloop callout
1105 g_simple_async_result_complete (simple);
1106 g_object_unref (simple);
1110 g_buffered_input_stream_read_async (GInputStream *stream,
1114 GCancellable *cancellable,
1115 GAsyncReadyCallback callback,
1118 GBufferedInputStream *bstream;
1119 GBufferedInputStreamPrivate *priv;
1120 GBufferedInputStreamClass *class;
1121 GInputStream *base_stream;
1123 GSimpleAsyncResult *simple;
1124 ReadAsyncData *data;
1126 bstream = G_BUFFERED_INPUT_STREAM (stream);
1127 priv = bstream->priv;
1129 data = g_slice_new (ReadAsyncData);
1130 data->buffer = buffer;
1131 data->bytes_read = 0;
1132 simple = g_simple_async_result_new (G_OBJECT (stream),
1133 callback, user_data,
1134 g_buffered_input_stream_read_async);
1135 g_simple_async_result_set_op_res_gpointer (simple, data, free_read_async_data);
1137 available = priv->end - priv->pos;
1139 if (count <= available)
1141 memcpy (buffer, priv->buffer + priv->pos, count);
1143 data->bytes_read = count;
1145 g_simple_async_result_complete_in_idle (simple);
1146 g_object_unref (simple);
1151 /* Full request not available, read all currently available
1152 * and request refill for more
1155 memcpy (buffer, priv->buffer + priv->pos, available);
1161 data->bytes_read = available;
1162 data->count = count;
1164 if (count > priv->len)
1166 /* Large request, shortcut buffer */
1168 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1170 g_input_stream_read_async (base_stream,
1171 (char *)buffer + data->bytes_read,
1173 io_priority, cancellable,
1174 large_read_callback,
1179 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1180 class->fill_async (bstream, priv->len, io_priority, cancellable,
1181 read_fill_buffer_callback, simple);
1186 g_buffered_input_stream_read_finish (GInputStream *stream,
1187 GAsyncResult *result,
1190 GSimpleAsyncResult *simple;
1191 ReadAsyncData *data;
1193 simple = G_SIMPLE_ASYNC_RESULT (result);
1195 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_read_async);
1197 data = g_simple_async_result_get_op_res_gpointer (simple);
1199 return data->bytes_read;
1204 gssize bytes_skipped;
1209 free_skip_async_data (gpointer _data)
1211 SkipAsyncData *data = _data;
1212 g_slice_free (SkipAsyncData, data);
1216 large_skip_callback (GObject *source_object,
1217 GAsyncResult *result,
1220 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1221 SkipAsyncData *data;
1225 data = g_simple_async_result_get_op_res_gpointer (simple);
1228 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1231 /* Only report the error if we've not already read some data */
1232 if (nread < 0 && data->bytes_skipped == 0)
1233 g_simple_async_result_take_error (simple, error);
1235 g_error_free (error);
1238 data->bytes_skipped += nread;
1240 /* Complete immediately, not in idle, since we're already
1241 * in a mainloop callout
1243 g_simple_async_result_complete (simple);
1244 g_object_unref (simple);
1248 skip_fill_buffer_callback (GObject *source_object,
1249 GAsyncResult *result,
1252 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1253 GBufferedInputStream *bstream;
1254 GBufferedInputStreamPrivate *priv;
1255 SkipAsyncData *data;
1260 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1261 priv = bstream->priv;
1263 data = g_simple_async_result_get_op_res_gpointer (simple);
1266 nread = g_buffered_input_stream_fill_finish (bstream,
1269 if (nread < 0 && data->bytes_skipped == 0)
1270 g_simple_async_result_take_error (simple, error);
1272 g_error_free (error);
1276 available = priv->end - priv->pos;
1277 data->count = MIN (data->count, available);
1279 data->bytes_skipped += data->count;
1280 priv->pos += data->count;
1283 /* Complete immediately, not in idle, since we're already
1284 * in a mainloop callout
1286 g_simple_async_result_complete (simple);
1287 g_object_unref (simple);
1291 g_buffered_input_stream_skip_async (GInputStream *stream,
1294 GCancellable *cancellable,
1295 GAsyncReadyCallback callback,
1298 GBufferedInputStream *bstream;
1299 GBufferedInputStreamPrivate *priv;
1300 GBufferedInputStreamClass *class;
1301 GInputStream *base_stream;
1303 GSimpleAsyncResult *simple;
1304 SkipAsyncData *data;
1306 bstream = G_BUFFERED_INPUT_STREAM (stream);
1307 priv = bstream->priv;
1309 data = g_slice_new (SkipAsyncData);
1310 data->bytes_skipped = 0;
1311 simple = g_simple_async_result_new (G_OBJECT (stream),
1312 callback, user_data,
1313 g_buffered_input_stream_skip_async);
1314 g_simple_async_result_set_op_res_gpointer (simple, data, free_skip_async_data);
1316 available = priv->end - priv->pos;
1318 if (count <= available)
1321 data->bytes_skipped = count;
1323 g_simple_async_result_complete_in_idle (simple);
1324 g_object_unref (simple);
1328 /* Full request not available, skip all currently available
1329 * and request refill for more
1337 data->bytes_skipped = available;
1338 data->count = count;
1340 if (count > priv->len)
1342 /* Large request, shortcut buffer */
1344 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1346 g_input_stream_skip_async (base_stream,
1348 io_priority, cancellable,
1349 large_skip_callback,
1354 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1355 class->fill_async (bstream, priv->len, io_priority, cancellable,
1356 skip_fill_buffer_callback, simple);
1361 g_buffered_input_stream_skip_finish (GInputStream *stream,
1362 GAsyncResult *result,
1365 GSimpleAsyncResult *simple;
1366 SkipAsyncData *data;
1368 simple = G_SIMPLE_ASYNC_RESULT (result);
1370 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_skip_async);
1372 data = g_simple_async_result_get_op_res_gpointer (simple);
1374 return data->bytes_skipped;