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"
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.
61 #define DEFAULT_BUFFER_SIZE 4096
63 struct _GBufferedInputStreamPrivate {
68 GAsyncReadyCallback outstanding_callback;
76 static void g_buffered_input_stream_set_property (GObject *object,
81 static void g_buffered_input_stream_get_property (GObject *object,
85 static void g_buffered_input_stream_finalize (GObject *object);
88 static gssize g_buffered_input_stream_skip (GInputStream *stream,
90 GCancellable *cancellable,
92 static void g_buffered_input_stream_skip_async (GInputStream *stream,
95 GCancellable *cancellable,
96 GAsyncReadyCallback callback,
98 static gssize g_buffered_input_stream_skip_finish (GInputStream *stream,
101 static gssize g_buffered_input_stream_read (GInputStream *stream,
104 GCancellable *cancellable,
106 static void g_buffered_input_stream_read_async (GInputStream *stream,
110 GCancellable *cancellable,
111 GAsyncReadyCallback callback,
113 static gssize g_buffered_input_stream_read_finish (GInputStream *stream,
114 GAsyncResult *result,
116 static gssize g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
118 GCancellable *cancellable,
120 static void g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
123 GCancellable *cancellable,
124 GAsyncReadyCallback callback,
126 static gssize g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
127 GAsyncResult *result,
130 static void compact_buffer (GBufferedInputStream *stream);
132 G_DEFINE_TYPE (GBufferedInputStream,
133 g_buffered_input_stream,
134 G_TYPE_FILTER_INPUT_STREAM)
138 g_buffered_input_stream_class_init (GBufferedInputStreamClass *klass)
140 GObjectClass *object_class;
141 GInputStreamClass *istream_class;
142 GBufferedInputStreamClass *bstream_class;
144 g_type_class_add_private (klass, sizeof (GBufferedInputStreamPrivate));
146 object_class = G_OBJECT_CLASS (klass);
147 object_class->get_property = g_buffered_input_stream_get_property;
148 object_class->set_property = g_buffered_input_stream_set_property;
149 object_class->finalize = g_buffered_input_stream_finalize;
151 istream_class = G_INPUT_STREAM_CLASS (klass);
152 istream_class->skip = g_buffered_input_stream_skip;
153 istream_class->skip_async = g_buffered_input_stream_skip_async;
154 istream_class->skip_finish = g_buffered_input_stream_skip_finish;
155 istream_class->read_fn = g_buffered_input_stream_read;
156 istream_class->read_async = g_buffered_input_stream_read_async;
157 istream_class->read_finish = g_buffered_input_stream_read_finish;
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: #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: #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);
268 g_buffered_input_stream_get_property (GObject *object,
273 GBufferedInputStreamPrivate *priv;
274 GBufferedInputStream *bstream;
276 bstream = G_BUFFERED_INPUT_STREAM (object);
277 priv = bstream->priv;
282 g_value_set_uint (value, priv->len);
286 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
292 g_buffered_input_stream_finalize (GObject *object)
294 GBufferedInputStreamPrivate *priv;
295 GBufferedInputStream *stream;
297 stream = G_BUFFERED_INPUT_STREAM (object);
300 g_free (priv->buffer);
302 G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize (object);
306 g_buffered_input_stream_init (GBufferedInputStream *stream)
308 stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
309 G_TYPE_BUFFERED_INPUT_STREAM,
310 GBufferedInputStreamPrivate);
315 * g_buffered_input_stream_new:
316 * @base_stream: a #GInputStream.
318 * Creates a new #GInputStream from the given @base_stream, with
319 * a buffer set to the default size (4 kilobytes).
321 * Returns: a #GInputStream for the given @base_stream.
324 g_buffered_input_stream_new (GInputStream *base_stream)
326 GInputStream *stream;
328 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
330 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
331 "base-stream", base_stream,
338 * g_buffered_input_stream_new_sized:
339 * @base_stream: a #GInputStream.
342 * Creates a new #GBufferedInputStream from the given @base_stream,
343 * with a buffer set to @size.
345 * Returns: a #GInputStream.
348 g_buffered_input_stream_new_sized (GInputStream *base_stream,
351 GInputStream *stream;
353 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
355 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
356 "base-stream", base_stream,
357 "buffer-size", (guint)size,
364 * g_buffered_input_stream_fill:
365 * @stream: #GBufferedInputStream.
366 * @count: the number of bytes that will be read from the stream.
367 * @cancellable: optional #GCancellable object, %NULL to ignore.
368 * @error: location to store the error occuring, or %NULL to ignore.
370 * Tries to read @count bytes from the stream into the buffer.
371 * Will block during this read.
373 * If @count is zero, returns zero and does nothing. A value of @count
374 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
376 * On success, the number of bytes read into the buffer is returned.
377 * It is not an error if this is not the same as the requested size, as it
378 * can happen e.g. near the end of a file. Zero is returned on end of file
379 * (or if @count is zero), but never otherwise.
381 * If @count is -1 then the attempted read size is equal to the number of
382 * bytes that are required to fill the buffer.
384 * If @cancellable is not %NULL, then the operation can be cancelled by
385 * triggering the cancellable object from another thread. If the operation
386 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
387 * operation was partially finished when the operation was cancelled the
388 * partial result will be returned, without an error.
390 * On error -1 is returned and @error is set accordingly.
392 * For the asynchronous, non-blocking, version of this function, see
393 * g_buffered_input_stream_fill_async().
395 * Returns: the number of bytes read into @stream's buffer, up to @count,
399 g_buffered_input_stream_fill (GBufferedInputStream *stream,
401 GCancellable *cancellable,
404 GBufferedInputStreamClass *class;
405 GInputStream *input_stream;
408 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
410 input_stream = G_INPUT_STREAM (stream);
414 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
415 _("Too large count value passed to %s"), G_STRFUNC);
419 if (!g_input_stream_set_pending (input_stream, error))
423 g_cancellable_push_current (cancellable);
425 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
426 res = class->fill (stream, count, cancellable, error);
429 g_cancellable_pop_current (cancellable);
431 g_input_stream_clear_pending (input_stream);
437 async_fill_callback_wrapper (GObject *source_object,
441 GBufferedInputStream *stream = G_BUFFERED_INPUT_STREAM (source_object);
443 g_input_stream_clear_pending (G_INPUT_STREAM (stream));
444 (*stream->priv->outstanding_callback) (source_object, res, user_data);
445 g_object_unref (stream);
449 * g_buffered_input_stream_fill_async:
450 * @stream: #GBufferedInputStream.
451 * @count: the number of bytes that will be read from the stream.
452 * @io_priority: the <link linkend="io-priority">I/O priority</link>
454 * @cancellable: optional #GCancellable object
455 * @callback: a #GAsyncReadyCallback.
456 * @user_data: a #gpointer.
458 * Reads data into @stream's buffer asynchronously, up to @count size.
459 * @io_priority can be used to prioritize reads. For the synchronous
460 * version of this function, see g_buffered_input_stream_fill().
462 * If @count is -1 then the attempted read size is equal to the number
463 * of bytes that are required to fill the buffer.
466 g_buffered_input_stream_fill_async (GBufferedInputStream *stream,
469 GCancellable *cancellable,
470 GAsyncReadyCallback callback,
473 GBufferedInputStreamClass *class;
474 GSimpleAsyncResult *simple;
475 GError *error = NULL;
477 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
481 simple = g_simple_async_result_new (G_OBJECT (stream),
484 g_buffered_input_stream_fill_async);
485 g_simple_async_result_complete_in_idle (simple);
486 g_object_unref (simple);
492 g_simple_async_report_error_in_idle (G_OBJECT (stream),
495 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
496 _("Too large count value passed to %s"),
501 if (!g_input_stream_set_pending (G_INPUT_STREAM (stream), &error))
503 g_simple_async_report_gerror_in_idle (G_OBJECT (stream),
507 g_error_free (error);
511 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
513 stream->priv->outstanding_callback = callback;
514 g_object_ref (stream);
515 class->fill_async (stream, count, io_priority, cancellable,
516 async_fill_callback_wrapper, user_data);
520 * g_buffered_input_stream_fill_finish:
521 * @stream: a #GBufferedInputStream.
522 * @result: a #GAsyncResult.
525 * Finishes an asynchronous read.
527 * Returns: a #gssize of the read stream, or %-1 on an error.
530 g_buffered_input_stream_fill_finish (GBufferedInputStream *stream,
531 GAsyncResult *result,
534 GSimpleAsyncResult *simple;
535 GBufferedInputStreamClass *class;
537 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
538 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
540 if (G_IS_SIMPLE_ASYNC_RESULT (result))
542 simple = G_SIMPLE_ASYNC_RESULT (result);
543 if (g_simple_async_result_propagate_error (simple, error))
546 /* Special case read of 0 bytes */
547 if (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_fill_async)
551 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
552 return class->fill_finish (stream, result, error);
556 * g_buffered_input_stream_get_available:
557 * @stream: #GBufferedInputStream.
559 * Gets the size of the available data within the stream.
561 * Returns: size of the available stream.
564 g_buffered_input_stream_get_available (GBufferedInputStream *stream)
566 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
568 return stream->priv->end - stream->priv->pos;
572 * g_buffered_input_stream_peek:
573 * @stream: a #GBufferedInputStream.
574 * @buffer: a pointer to an allocated chunk of memory.
578 * Peeks in the buffer, copying data of size @count into @buffer,
579 * offset @offset bytes.
581 * Returns: a #gsize of the number of bytes peeked, or %-1 on error.
584 g_buffered_input_stream_peek (GBufferedInputStream *stream,
592 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
593 g_return_val_if_fail (buffer != NULL, -1);
595 available = g_buffered_input_stream_get_available (stream);
597 if (offset > available)
600 end = MIN (offset + count, available);
601 count = end - offset;
603 memcpy (buffer, stream->priv->buffer + stream->priv->pos + offset, count);
608 * g_buffered_input_stream_peek_buffer:
609 * @stream: a #GBufferedInputStream.
610 * @count: a #gsize to get the number of bytes available in the buffer.
612 * Returns the buffer with the currently available bytes. The returned
613 * buffer must not be modified and will become invalid when reading from
614 * the stream or filling the buffer.
616 * Returns: read-only buffer
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 availbile and 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_read_byte:
844 * @stream: #GBufferedInputStream.
845 * @cancellable: optional #GCancellable object, %NULL to ignore.
846 * @error: location to store the error occuring, or %NULL to ignore.
848 * Tries to read a single byte from the stream or the buffer. Will block
851 * On success, the byte read from the stream is returned. On end of stream
852 * -1 is returned but it's not an exceptional error and @error is not set.
854 * If @cancellable is not %NULL, then the operation can be cancelled by
855 * triggering the cancellable object from another thread. If the operation
856 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
857 * operation was partially finished when the operation was cancelled the
858 * partial result will be returned, without an error.
860 * On error -1 is returned and @error is set accordingly.
862 * Returns: the byte read from the @stream, or -1 on end of stream or error.
865 g_buffered_input_stream_read_byte (GBufferedInputStream *stream,
866 GCancellable *cancellable,
869 GBufferedInputStreamPrivate *priv;
870 GBufferedInputStreamClass *class;
871 GInputStream *input_stream;
875 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
878 input_stream = G_INPUT_STREAM (stream);
880 if (g_input_stream_is_closed (input_stream))
882 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
883 _("Stream is already closed"));
887 if (!g_input_stream_set_pending (input_stream, error))
890 available = priv->end - priv->pos;
894 g_input_stream_clear_pending (input_stream);
895 return priv->buffer[priv->pos++];
898 /* Byte not available, request refill for more */
901 g_cancellable_push_current (cancellable);
906 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
907 nread = class->fill (stream, priv->len, cancellable, error);
910 g_cancellable_pop_current (cancellable);
912 g_input_stream_clear_pending (input_stream);
915 return -1; /* error or end of stream */
917 return priv->buffer[priv->pos++];
920 /* ************************** */
921 /* Async stuff implementation */
922 /* ************************** */
925 fill_async_callback (GObject *source_object,
926 GAsyncResult *result,
931 GSimpleAsyncResult *simple;
936 res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
939 g_simple_async_result_set_op_res_gssize (simple, res);
942 g_simple_async_result_set_from_error (simple, error);
943 g_error_free (error);
947 GBufferedInputStreamPrivate *priv;
950 object = g_async_result_get_source_object (G_ASYNC_RESULT (simple));
951 priv = G_BUFFERED_INPUT_STREAM (object)->priv;
953 g_assert_cmpint (priv->end + res, <=, priv->len);
956 g_object_unref (object);
959 /* Complete immediately, not in idle, since we're already in a mainloop callout */
960 g_simple_async_result_complete (simple);
961 g_object_unref (simple);
965 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
968 GCancellable *cancellable,
969 GAsyncReadyCallback callback,
972 GBufferedInputStreamPrivate *priv;
973 GInputStream *base_stream;
974 GSimpleAsyncResult *simple;
982 in_buffer = priv->end - priv->pos;
984 /* Never fill more than can fit in the buffer */
985 count = MIN (count, priv->len - in_buffer);
987 /* If requested length does not fit at end, compact */
988 if (priv->len - priv->end < count)
989 compact_buffer (stream);
991 simple = g_simple_async_result_new (G_OBJECT (stream),
993 g_buffered_input_stream_real_fill_async);
995 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
996 g_input_stream_read_async (base_stream,
997 priv->buffer + priv->end,
1001 fill_async_callback,
1006 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1007 GAsyncResult *result,
1010 GSimpleAsyncResult *simple;
1013 simple = G_SIMPLE_ASYNC_RESULT (result);
1014 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_real_fill_async);
1016 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_set_from_error (simple, error);
1054 data->bytes_read += nread;
1057 g_error_free (error);
1059 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1060 g_simple_async_result_complete (simple);
1061 g_object_unref (simple);
1065 read_fill_buffer_callback (GObject *source_object,
1066 GAsyncResult *result,
1069 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1070 GBufferedInputStream *bstream;
1071 GBufferedInputStreamPrivate *priv;
1072 ReadAsyncData *data;
1077 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1078 priv = bstream->priv;
1080 data = g_simple_async_result_get_op_res_gpointer (simple);
1083 nread = g_buffered_input_stream_fill_finish (bstream,
1086 if (nread < 0 && data->bytes_read == 0)
1087 g_simple_async_result_set_from_error (simple, error);
1092 available = priv->end - priv->pos;
1093 data->count = MIN (data->count, available);
1095 memcpy ((char *)data->buffer + data->bytes_read, (char *)priv->buffer + priv->pos, data->count);
1096 data->bytes_read += data->count;
1097 priv->pos += data->count;
1101 g_error_free (error);
1103 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1104 g_simple_async_result_complete (simple);
1105 g_object_unref (simple);
1109 g_buffered_input_stream_read_async (GInputStream *stream,
1113 GCancellable *cancellable,
1114 GAsyncReadyCallback callback,
1117 GBufferedInputStream *bstream;
1118 GBufferedInputStreamPrivate *priv;
1119 GBufferedInputStreamClass *class;
1120 GInputStream *base_stream;
1122 GSimpleAsyncResult *simple;
1123 ReadAsyncData *data;
1125 bstream = G_BUFFERED_INPUT_STREAM (stream);
1126 priv = bstream->priv;
1128 data = g_slice_new (ReadAsyncData);
1129 data->buffer = buffer;
1130 data->bytes_read = 0;
1131 simple = g_simple_async_result_new (G_OBJECT (stream),
1132 callback, user_data,
1133 g_buffered_input_stream_read_async);
1134 g_simple_async_result_set_op_res_gpointer (simple, data, free_read_async_data);
1136 available = priv->end - priv->pos;
1138 if (count <= available)
1140 memcpy (buffer, priv->buffer + priv->pos, count);
1142 data->bytes_read = count;
1144 g_simple_async_result_complete_in_idle (simple);
1145 g_object_unref (simple);
1150 /* Full request not available, read all currently availbile and request refill for more */
1152 memcpy (buffer, priv->buffer + priv->pos, available);
1158 data->bytes_read = available;
1159 data->count = count;
1161 if (count > priv->len)
1163 /* Large request, shortcut buffer */
1165 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1167 g_input_stream_read_async (base_stream,
1168 (char *)buffer + data->bytes_read,
1170 io_priority, cancellable,
1171 large_read_callback,
1176 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1177 class->fill_async (bstream, priv->len, io_priority, cancellable,
1178 read_fill_buffer_callback, simple);
1183 g_buffered_input_stream_read_finish (GInputStream *stream,
1184 GAsyncResult *result,
1187 GSimpleAsyncResult *simple;
1188 ReadAsyncData *data;
1190 simple = G_SIMPLE_ASYNC_RESULT (result);
1192 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_read_async);
1194 data = g_simple_async_result_get_op_res_gpointer (simple);
1196 return data->bytes_read;
1200 gssize bytes_skipped;
1205 free_skip_async_data (gpointer _data)
1207 SkipAsyncData *data = _data;
1208 g_slice_free (SkipAsyncData, data);
1212 large_skip_callback (GObject *source_object,
1213 GAsyncResult *result,
1216 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1217 SkipAsyncData *data;
1221 data = g_simple_async_result_get_op_res_gpointer (simple);
1224 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1227 /* Only report the error if we've not already read some data */
1228 if (nread < 0 && data->bytes_skipped == 0)
1229 g_simple_async_result_set_from_error (simple, error);
1232 data->bytes_skipped += nread;
1235 g_error_free (error);
1237 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1238 g_simple_async_result_complete (simple);
1239 g_object_unref (simple);
1243 skip_fill_buffer_callback (GObject *source_object,
1244 GAsyncResult *result,
1247 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1248 GBufferedInputStream *bstream;
1249 GBufferedInputStreamPrivate *priv;
1250 SkipAsyncData *data;
1255 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1256 priv = bstream->priv;
1258 data = g_simple_async_result_get_op_res_gpointer (simple);
1261 nread = g_buffered_input_stream_fill_finish (bstream,
1264 if (nread < 0 && data->bytes_skipped == 0)
1265 g_simple_async_result_set_from_error (simple, error);
1270 available = priv->end - priv->pos;
1271 data->count = MIN (data->count, available);
1273 data->bytes_skipped += data->count;
1274 priv->pos += data->count;
1278 g_error_free (error);
1280 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1281 g_simple_async_result_complete (simple);
1282 g_object_unref (simple);
1286 g_buffered_input_stream_skip_async (GInputStream *stream,
1289 GCancellable *cancellable,
1290 GAsyncReadyCallback callback,
1293 GBufferedInputStream *bstream;
1294 GBufferedInputStreamPrivate *priv;
1295 GBufferedInputStreamClass *class;
1296 GInputStream *base_stream;
1298 GSimpleAsyncResult *simple;
1299 SkipAsyncData *data;
1301 bstream = G_BUFFERED_INPUT_STREAM (stream);
1302 priv = bstream->priv;
1304 data = g_slice_new (SkipAsyncData);
1305 data->bytes_skipped = 0;
1306 simple = g_simple_async_result_new (G_OBJECT (stream),
1307 callback, user_data,
1308 g_buffered_input_stream_skip_async);
1309 g_simple_async_result_set_op_res_gpointer (simple, data, free_skip_async_data);
1311 available = priv->end - priv->pos;
1313 if (count <= available)
1316 data->bytes_skipped = count;
1318 g_simple_async_result_complete_in_idle (simple);
1319 g_object_unref (simple);
1324 /* Full request not available, skip all currently availbile and request refill for more */
1331 data->bytes_skipped = available;
1332 data->count = count;
1334 if (count > priv->len)
1336 /* Large request, shortcut buffer */
1338 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1340 g_input_stream_skip_async (base_stream,
1342 io_priority, cancellable,
1343 large_skip_callback,
1348 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1349 class->fill_async (bstream, priv->len, io_priority, cancellable,
1350 skip_fill_buffer_callback, simple);
1355 g_buffered_input_stream_skip_finish (GInputStream *stream,
1356 GAsyncResult *result,
1359 GSimpleAsyncResult *simple;
1360 SkipAsyncData *data;
1362 simple = G_SIMPLE_ASYNC_RESULT (result);
1364 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_skip_async);
1366 data = g_simple_async_result_get_op_res_gpointer (simple);
1368 return data->bytes_skipped;
1372 #define __G_BUFFERED_INPUT_STREAM_C__
1373 #include "gioaliasdef.c"