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 "gsimpleasyncresult.h"
34 * SECTION:gbufferedinputstream
35 * @short_description: Buffered Input Stream
36 * @see_also: #GFilterInputStream, #GInputStream
38 * Buffered input stream implements #GFilterInputStream and provides
41 * By default, #GBufferedInputStream's buffer size is set at 4 kilobytes.
43 * To create a buffered input stream, use g_buffered_input_stream_new(),
44 * or g_buffered_input_stream_new_sized() to specify the buffer's size at
47 * To get the size of a buffer within a buffered input stream, use
48 * g_buffered_input_stream_get_buffer_size(). To change the size of a
49 * buffered input stream's buffer, use
50 * g_buffered_input_stream_set_buffer_size(). Note that the buffer's size
51 * cannot be reduced below the size of the data within the buffer.
57 #define DEFAULT_BUFFER_SIZE 4096
59 struct _GBufferedInputStreamPrivate {
64 GAsyncReadyCallback outstanding_callback;
72 static void g_buffered_input_stream_set_property (GObject *object,
77 static void g_buffered_input_stream_get_property (GObject *object,
81 static void g_buffered_input_stream_finalize (GObject *object);
84 static gssize g_buffered_input_stream_skip (GInputStream *stream,
86 GCancellable *cancellable,
88 static void g_buffered_input_stream_skip_async (GInputStream *stream,
91 GCancellable *cancellable,
92 GAsyncReadyCallback callback,
94 static gssize g_buffered_input_stream_skip_finish (GInputStream *stream,
97 static gssize g_buffered_input_stream_read (GInputStream *stream,
100 GCancellable *cancellable,
102 static void g_buffered_input_stream_read_async (GInputStream *stream,
106 GCancellable *cancellable,
107 GAsyncReadyCallback callback,
109 static gssize g_buffered_input_stream_read_finish (GInputStream *stream,
110 GAsyncResult *result,
112 static gssize g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
114 GCancellable *cancellable,
116 static void g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
119 GCancellable *cancellable,
120 GAsyncReadyCallback callback,
122 static gssize g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
123 GAsyncResult *result,
126 static void compact_buffer (GBufferedInputStream *stream);
128 G_DEFINE_TYPE (GBufferedInputStream,
129 g_buffered_input_stream,
130 G_TYPE_FILTER_INPUT_STREAM)
134 g_buffered_input_stream_class_init (GBufferedInputStreamClass *klass)
136 GObjectClass *object_class;
137 GInputStreamClass *istream_class;
138 GBufferedInputStreamClass *bstream_class;
140 g_type_class_add_private (klass, sizeof (GBufferedInputStreamPrivate));
142 object_class = G_OBJECT_CLASS (klass);
143 object_class->get_property = g_buffered_input_stream_get_property;
144 object_class->set_property = g_buffered_input_stream_set_property;
145 object_class->finalize = g_buffered_input_stream_finalize;
147 istream_class = G_INPUT_STREAM_CLASS (klass);
148 istream_class->skip = g_buffered_input_stream_skip;
149 istream_class->skip_async = g_buffered_input_stream_skip_async;
150 istream_class->skip_finish = g_buffered_input_stream_skip_finish;
151 istream_class->read = g_buffered_input_stream_read;
152 istream_class->read_async = g_buffered_input_stream_read_async;
153 istream_class->read_finish = g_buffered_input_stream_read_finish;
155 bstream_class = G_BUFFERED_INPUT_STREAM_CLASS (klass);
156 bstream_class->fill = g_buffered_input_stream_real_fill;
157 bstream_class->fill_async = g_buffered_input_stream_real_fill_async;
158 bstream_class->fill_finish = g_buffered_input_stream_real_fill_finish;
160 g_object_class_install_property (object_class,
162 g_param_spec_uint ("buffer-size",
164 P_("The size of the backend buffer"),
168 G_PARAM_READWRITE | G_PARAM_CONSTRUCT |
169 G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
175 * g_buffered_input_stream_get_buffer_size:
176 * @stream: #GBufferedInputStream.
178 * Gets the size of the input buffer.
180 * Returns: the current buffer size, or %-1 on error.
183 g_buffered_input_stream_get_buffer_size (GBufferedInputStream *stream)
185 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
187 return stream->priv->len;
191 * g_buffered_input_stream_set_buffer_size:
192 * @stream: #GBufferedInputStream.
195 * Sets the size of the internal buffer of @stream to @size, or to the
196 * size of the contents of the buffer. The buffer can never be resized
197 * smaller than its current contents.
200 g_buffered_input_stream_set_buffer_size (GBufferedInputStream *stream,
203 GBufferedInputStreamPrivate *priv;
207 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
211 if (priv->len == size)
216 in_buffer = priv->end - priv->pos;
218 /* Never resize smaller than current buffer contents */
219 size = MAX (size, in_buffer);
221 buffer = g_malloc (size);
222 memcpy (buffer, priv->buffer + priv->pos, in_buffer);
225 priv->end = in_buffer;
226 g_free (priv->buffer);
227 priv->buffer = buffer;
234 priv->buffer = g_malloc (size);
237 g_object_notify (G_OBJECT (stream), "buffer-size");
241 g_buffered_input_stream_set_property (GObject *object,
246 GBufferedInputStreamPrivate *priv;
247 GBufferedInputStream *bstream;
249 bstream = G_BUFFERED_INPUT_STREAM (object);
250 priv = bstream->priv;
255 g_buffered_input_stream_set_buffer_size (bstream, g_value_get_uint (value));
259 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
266 g_buffered_input_stream_get_property (GObject *object,
271 GBufferedInputStreamPrivate *priv;
272 GBufferedInputStream *bstream;
274 bstream = G_BUFFERED_INPUT_STREAM (object);
275 priv = bstream->priv;
280 g_value_set_uint (value, priv->len);
284 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
290 g_buffered_input_stream_finalize (GObject *object)
292 GBufferedInputStreamPrivate *priv;
293 GBufferedInputStream *stream;
295 stream = G_BUFFERED_INPUT_STREAM (object);
298 g_free (priv->buffer);
300 if (G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize)
301 (*G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize) (object);
305 g_buffered_input_stream_init (GBufferedInputStream *stream)
307 stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
308 G_TYPE_BUFFERED_INPUT_STREAM,
309 GBufferedInputStreamPrivate);
314 * g_buffered_input_stream_new:
315 * @base_stream: a #GInputStream.
317 * Creates a new #GInputStream from the given @base_stream, with
318 * a buffer set to the default size (4 kilobytes).
320 * Returns: a #GInputStream for the given @base_stream.
323 g_buffered_input_stream_new (GInputStream *base_stream)
325 GInputStream *stream;
327 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
329 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
330 "base-stream", base_stream,
337 * g_buffered_input_stream_new_sized:
338 * @base_stream: a #GOutputStream.
341 * Creates a new #GBufferedInputStream from the given @base_stream,
342 * with a buffer set to @size.
344 * Returns: a #GInputStream.
347 g_buffered_input_stream_new_sized (GInputStream *base_stream,
350 GInputStream *stream;
352 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
354 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
355 "base-stream", base_stream,
356 "buffer-size", (guint)size,
363 * g_buffered_input_stream_fill:
364 * @stream: #GBufferedInputStream.
365 * @count: the number of bytes that will be read from the stream.
366 * @cancellable: optional #GCancellable object, %NULL to ignore.
367 * @error: location to store the error occuring, or %NULL to ignore.
369 * Tries to read @count bytes from the stream into the buffer.
370 * Will block during this read.
372 * If @count is zero, returns zero and does nothing. A value of @count
373 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
375 * On success, the number of bytes read into the buffer is returned.
376 * It is not an error if this is not the same as the requested size, as it
377 * can happen e.g. near the end of a file. Zero is returned on end of file
378 * (or if @count is zero), but never otherwise.
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);
408 if (g_input_stream_is_closed (input_stream))
410 g_set_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
411 _("Stream is already closed"));
415 if (g_input_stream_has_pending (input_stream))
417 g_set_error (error, G_IO_ERROR, G_IO_ERROR_PENDING,
418 _("Stream has outstanding operation"));
422 g_input_stream_set_pending (input_stream, TRUE);
425 g_push_current_cancellable (cancellable);
427 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
428 res = class->fill (stream, count, cancellable, error);
431 g_pop_current_cancellable (cancellable);
433 g_input_stream_set_pending (input_stream, FALSE);
439 async_fill_callback_wrapper (GObject *source_object,
443 GBufferedInputStream *stream = G_BUFFERED_INPUT_STREAM (source_object);
445 g_input_stream_set_pending (G_INPUT_STREAM (stream), FALSE);
446 (*stream->priv->outstanding_callback) (source_object, res, user_data);
447 g_object_unref (stream);
451 * g_buffered_input_stream_fill_async:
452 * @stream: #GBufferedInputStream.
454 * @io_priority: the <link linkend="io-priority">I/O priority</link>
456 * @cancellable: optional #GCancellable object
457 * @callback: a #GAsyncReadyCallback.
458 * @user_data: a #gpointer.
460 * Reads data into @stream's buffer asynchronously, up to @count size.
461 * @io_priority can be used to prioritize reads. For the synchronous
462 * version of this function, see g_buffered_input_stream_fill().
465 g_buffered_input_stream_fill_async (GBufferedInputStream *stream,
468 GCancellable *cancellable,
469 GAsyncReadyCallback callback,
472 GBufferedInputStreamClass *class;
473 GSimpleAsyncResult *simple;
475 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
479 simple = g_simple_async_result_new (G_OBJECT (stream),
482 g_buffered_input_stream_fill_async);
483 g_simple_async_result_complete_in_idle (simple);
484 g_object_unref (simple);
488 if (((gssize) count) < 0)
490 g_simple_async_report_error_in_idle (G_OBJECT (stream),
493 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
494 _("Too large count value passed to g_input_stream_read_async"));
498 if (g_input_stream_is_closed (G_INPUT_STREAM (stream)))
500 g_simple_async_report_error_in_idle (G_OBJECT (stream),
503 G_IO_ERROR, G_IO_ERROR_CLOSED,
504 _("Stream is already closed"));
508 if (g_input_stream_has_pending (G_INPUT_STREAM (stream)))
510 g_simple_async_report_error_in_idle (G_OBJECT (stream),
513 G_IO_ERROR, G_IO_ERROR_PENDING,
514 _("Stream has outstanding operation"));
518 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
520 g_input_stream_set_pending (G_INPUT_STREAM (stream), TRUE);
521 stream->priv->outstanding_callback = callback;
522 g_object_ref (stream);
523 class->fill_async (stream, count, io_priority, cancellable,
524 async_fill_callback_wrapper, user_data);
528 * g_buffered_input_stream_fill_finish:
529 * @stream: a #GBufferedInputStream.
530 * @result: a #GAsyncResult.
533 * Finishes an asynchronous read.
535 * Returns: a #gssize of the read stream, or %-1 on an error.
538 g_buffered_input_stream_fill_finish (GBufferedInputStream *stream,
539 GAsyncResult *result,
542 GSimpleAsyncResult *simple;
543 GBufferedInputStreamClass *class;
545 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
546 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
548 if (G_IS_SIMPLE_ASYNC_RESULT (result))
550 simple = G_SIMPLE_ASYNC_RESULT (result);
551 if (g_simple_async_result_propagate_error (simple, error))
554 /* Special case read of 0 bytes */
555 if (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_fill_async)
559 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
560 return class->fill_finish (stream, result, error);
564 * g_buffered_input_stream_get_available:
565 * @stream: #GBufferedInputStream.
567 * Gets the size of the available data within the stream.
569 * Returns: size of the available stream.
572 g_buffered_input_stream_get_available (GBufferedInputStream *stream)
574 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
576 return stream->priv->end - stream->priv->pos;
580 * g_buffered_input_stream_peek:
581 * @stream: a #GBufferedInputStream.
582 * @buffer: a pointer to an allocated chunk of memory.
586 * Peeks in the buffer, copying data of size @count into @buffer,
587 * offset @offset bytes.
589 * Returns: a #gsize of the number of bytes peeked, or %-1 on error.
592 g_buffered_input_stream_peek (GBufferedInputStream *stream,
600 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
601 g_return_val_if_fail (buffer != NULL, -1);
603 available = g_buffered_input_stream_get_available (stream);
605 if (offset > available)
608 end = MIN (offset + count, available);
609 count = end - offset;
611 memcpy (buffer, stream->priv->buffer + stream->priv->pos + offset, count);
616 * g_buffered_input_stream_peek_buffer:
617 * @stream: a #GBufferedInputStream.
618 * @count: a #gsize to get the number of bytes available in the buffer.
620 * Returns the buffer with the currently available bytes. The returned
621 * buffer must not be modified and will become invalid when reading from
622 * the stream or filling the buffer.
624 * Returns: read-only buffer
627 g_buffered_input_stream_peek_buffer (GBufferedInputStream *stream,
630 GBufferedInputStreamPrivate *priv;
632 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), NULL);
637 *count = priv->end - priv->pos;
639 return priv->buffer + priv->pos;
643 compact_buffer (GBufferedInputStream *stream)
645 GBufferedInputStreamPrivate *priv;
650 current_size = priv->end - priv->pos;
652 g_memmove (priv->buffer, priv->buffer + priv->pos, current_size);
655 priv->end = current_size;
659 g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
661 GCancellable *cancellable,
664 GBufferedInputStreamPrivate *priv;
665 GInputStream *base_stream;
674 in_buffer = priv->end - priv->pos;
676 /* Never fill more than can fit in the buffer */
677 count = MIN (count, priv->len - in_buffer);
679 /* If requested length does not fit at end, compact */
680 if (priv->len - priv->end < count)
681 compact_buffer (stream);
683 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
684 nread = g_input_stream_read (base_stream,
685 priv->buffer + priv->end,
697 g_buffered_input_stream_skip (GInputStream *stream,
699 GCancellable *cancellable,
702 GBufferedInputStream *bstream;
703 GBufferedInputStreamPrivate *priv;
704 GInputStream *base_stream;
705 gsize available, bytes_skipped;
708 bstream = G_BUFFERED_INPUT_STREAM (stream);
709 priv = bstream->priv;
711 available = priv->end - priv->pos;
713 if (count <= available)
719 /* Full request not available, skip all currently available and
720 * request refill for more
725 bytes_skipped = available;
728 if (bytes_skipped > 0)
729 error = NULL; /* Ignore further errors if we already read some data */
731 if (count > priv->len)
733 /* Large request, shortcut buffer */
735 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
737 nread = g_input_stream_skip (base_stream,
742 if (nread < 0 && bytes_skipped == 0)
746 bytes_skipped += nread;
748 return bytes_skipped;
751 g_input_stream_set_pending (stream, FALSE); /* to avoid already pending error */
752 nread = g_buffered_input_stream_fill (bstream, priv->len, cancellable, error);
753 g_input_stream_set_pending (stream, TRUE); /* enable again */
757 if (bytes_skipped == 0)
760 return bytes_skipped;
763 available = priv->end - priv->pos;
764 count = MIN (count, available);
766 bytes_skipped += count;
769 return bytes_skipped;
773 g_buffered_input_stream_read (GInputStream *stream,
776 GCancellable *cancellable,
779 GBufferedInputStream *bstream;
780 GBufferedInputStreamPrivate *priv;
781 GInputStream *base_stream;
782 gsize available, bytes_read;
785 bstream = G_BUFFERED_INPUT_STREAM (stream);
786 priv = bstream->priv;
788 available = priv->end - priv->pos;
790 if (count <= available)
792 memcpy (buffer, priv->buffer + priv->pos, count);
797 /* Full request not available, read all currently availbile and request refill for more */
799 memcpy (buffer, priv->buffer + priv->pos, available);
802 bytes_read = available;
806 error = NULL; /* Ignore further errors if we already read some data */
808 if (count > priv->len)
810 /* Large request, shortcut buffer */
812 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
814 nread = g_input_stream_read (base_stream,
815 (char *)buffer + bytes_read,
820 if (nread < 0 && bytes_read == 0)
829 g_input_stream_set_pending (stream, FALSE); /* to avoid already pending error */
830 nread = g_buffered_input_stream_fill (bstream, priv->len, cancellable, error);
831 g_input_stream_set_pending (stream, TRUE); /* enable again */
840 available = priv->end - priv->pos;
841 count = MIN (count, available);
843 memcpy ((char *)buffer + bytes_read, (char *)priv->buffer + priv->pos, count);
851 * g_buffered_input_stream_read_byte:
852 * @stream: #GBufferedInputStream.
853 * @cancellable: optional #GCancellable object, %NULL to ignore.
854 * @error: location to store the error occuring, or %NULL to ignore.
856 * Tries to read a single byte from the stream or the buffer. Will block
859 * On success, the byte read from the stream is returned. On end of stream
860 * -1 is returned but it's not an exceptional error and @error is not set.
862 * If @cancellable is not %NULL, then the operation can be cancelled by
863 * triggering the cancellable object from another thread. If the operation
864 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
865 * operation was partially finished when the operation was cancelled the
866 * partial result will be returned, without an error.
868 * On error -1 is returned and @error is set accordingly.
870 * Returns: the byte read from the @stream, or -1 on end of stream or error.
873 g_buffered_input_stream_read_byte (GBufferedInputStream *stream,
874 GCancellable *cancellable,
877 GBufferedInputStreamPrivate *priv;
878 GInputStream *input_stream;
882 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
885 input_stream = G_INPUT_STREAM (stream);
887 if (g_input_stream_is_closed (input_stream))
889 g_set_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
890 _("Stream is already closed"));
894 if (g_input_stream_has_pending (input_stream))
896 g_set_error (error, G_IO_ERROR, G_IO_ERROR_PENDING,
897 _("Stream has outstanding operation"));
901 available = priv->end - priv->pos;
904 return priv->buffer[priv->pos++];
906 /* Byte not available, request refill for more */
909 g_push_current_cancellable (cancellable);
914 nread = g_buffered_input_stream_fill (stream, priv->len, cancellable, error);
917 g_pop_current_cancellable (cancellable);
920 return -1; /* error or end of stream */
922 return priv->buffer[priv->pos++];
925 /* ************************** */
926 /* Async stuff implementation */
927 /* ************************** */
930 fill_async_callback (GObject *source_object,
931 GAsyncResult *result,
936 GSimpleAsyncResult *simple;
941 res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
944 g_simple_async_result_set_op_res_gssize (simple, res);
947 g_simple_async_result_set_from_error (simple, error);
948 g_error_free (error);
951 /* Complete immediately, not in idle, since we're already in a mainloop callout */
952 g_simple_async_result_complete (simple);
953 g_object_unref (simple);
957 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
960 GCancellable *cancellable,
961 GAsyncReadyCallback callback,
964 GBufferedInputStreamPrivate *priv;
965 GInputStream *base_stream;
966 GSimpleAsyncResult *simple;
974 in_buffer = priv->end - priv->pos;
976 /* Never fill more than can fit in the buffer */
977 count = MIN (count, priv->len - in_buffer);
979 /* If requested length does not fit at end, compact */
980 if (priv->len - priv->end < count)
981 compact_buffer (stream);
983 simple = g_simple_async_result_new (G_OBJECT (stream),
985 g_buffered_input_stream_real_fill_async);
987 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
988 g_input_stream_read_async (base_stream,
989 priv->buffer + priv->end,
998 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
999 GAsyncResult *result,
1002 GSimpleAsyncResult *simple;
1005 simple = G_SIMPLE_ASYNC_RESULT (result);
1006 g_assert (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_real_fill_async);
1008 nread = g_simple_async_result_get_op_res_gssize (simple);
1019 free_read_async_data (gpointer _data)
1021 ReadAsyncData *data = _data;
1022 g_slice_free (ReadAsyncData, data);
1026 large_read_callback (GObject *source_object,
1027 GAsyncResult *result,
1030 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1031 ReadAsyncData *data;
1035 data = g_simple_async_result_get_op_res_gpointer (simple);
1038 nread = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
1041 /* Only report the error if we've not already read some data */
1042 if (nread < 0 && data->bytes_read == 0)
1043 g_simple_async_result_set_from_error (simple, error);
1046 data->bytes_read += nread;
1049 g_error_free (error);
1051 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1052 g_simple_async_result_complete (simple);
1053 g_object_unref (simple);
1057 read_fill_buffer_callback (GObject *source_object,
1058 GAsyncResult *result,
1061 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1062 GBufferedInputStream *bstream;
1063 GBufferedInputStreamPrivate *priv;
1064 ReadAsyncData *data;
1069 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1070 priv = bstream->priv;
1072 g_input_stream_set_pending (G_INPUT_STREAM (bstream), TRUE); /* enable again */
1074 data = g_simple_async_result_get_op_res_gpointer (simple);
1077 nread = g_buffered_input_stream_fill_finish (bstream,
1080 if (nread < 0 && data->bytes_read == 0)
1081 g_simple_async_result_set_from_error (simple, error);
1086 available = priv->end - priv->pos;
1087 data->count = MIN (data->count, available);
1089 memcpy ((char *)data->buffer + data->bytes_read, (char *)priv->buffer + priv->pos, data->count);
1090 data->bytes_read += data->count;
1091 priv->pos += data->count;
1095 g_error_free (error);
1097 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1098 g_simple_async_result_complete (simple);
1099 g_object_unref (simple);
1103 g_buffered_input_stream_read_async (GInputStream *stream,
1107 GCancellable *cancellable,
1108 GAsyncReadyCallback callback,
1111 GBufferedInputStream *bstream;
1112 GBufferedInputStreamPrivate *priv;
1113 GInputStream *base_stream;
1115 GSimpleAsyncResult *simple;
1116 ReadAsyncData *data;
1118 bstream = G_BUFFERED_INPUT_STREAM (stream);
1119 priv = bstream->priv;
1121 data = g_slice_new (ReadAsyncData);
1122 data->buffer = buffer;
1123 data->bytes_read = 0;
1124 simple = g_simple_async_result_new (G_OBJECT (stream),
1125 callback, user_data,
1126 g_buffered_input_stream_read_async);
1127 g_simple_async_result_set_op_res_gpointer (simple, data, free_read_async_data);
1129 available = priv->end - priv->pos;
1131 if (count <= available)
1133 memcpy (buffer, priv->buffer + priv->pos, count);
1135 data->bytes_read = count;
1137 g_simple_async_result_complete_in_idle (simple);
1138 g_object_unref (simple);
1143 /* Full request not available, read all currently availbile and request refill for more */
1145 memcpy (buffer, priv->buffer + priv->pos, available);
1151 data->bytes_read = available;
1152 data->count = count;
1154 if (count > priv->len)
1156 /* Large request, shortcut buffer */
1158 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1160 g_input_stream_read_async (base_stream,
1161 (char *)buffer + data->bytes_read,
1163 io_priority, cancellable,
1164 large_read_callback,
1169 g_input_stream_set_pending (stream, FALSE); /* to avoid already pending error */
1170 g_buffered_input_stream_fill_async (bstream, priv->len,
1171 io_priority, cancellable,
1172 read_fill_buffer_callback, simple);
1177 g_buffered_input_stream_read_finish (GInputStream *stream,
1178 GAsyncResult *result,
1181 GSimpleAsyncResult *simple;
1182 ReadAsyncData *data;
1184 simple = G_SIMPLE_ASYNC_RESULT (result);
1186 g_assert (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_read_async);
1188 data = g_simple_async_result_get_op_res_gpointer (simple);
1190 return data->bytes_read;
1194 gssize bytes_skipped;
1199 free_skip_async_data (gpointer _data)
1201 SkipAsyncData *data = _data;
1202 g_slice_free (SkipAsyncData, data);
1206 large_skip_callback (GObject *source_object,
1207 GAsyncResult *result,
1210 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1211 SkipAsyncData *data;
1215 data = g_simple_async_result_get_op_res_gpointer (simple);
1218 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1221 /* Only report the error if we've not already read some data */
1222 if (nread < 0 && data->bytes_skipped == 0)
1223 g_simple_async_result_set_from_error (simple, error);
1226 data->bytes_skipped += nread;
1229 g_error_free (error);
1231 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1232 g_simple_async_result_complete (simple);
1233 g_object_unref (simple);
1237 skip_fill_buffer_callback (GObject *source_object,
1238 GAsyncResult *result,
1241 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1242 GBufferedInputStream *bstream;
1243 GBufferedInputStreamPrivate *priv;
1244 SkipAsyncData *data;
1249 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1250 priv = bstream->priv;
1252 g_input_stream_set_pending (G_INPUT_STREAM (bstream), TRUE); /* enable again */
1254 data = g_simple_async_result_get_op_res_gpointer (simple);
1257 nread = g_buffered_input_stream_fill_finish (bstream,
1260 if (nread < 0 && data->bytes_skipped == 0)
1261 g_simple_async_result_set_from_error (simple, error);
1266 available = priv->end - priv->pos;
1267 data->count = MIN (data->count, available);
1269 data->bytes_skipped += data->count;
1270 priv->pos += data->count;
1274 g_error_free (error);
1276 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1277 g_simple_async_result_complete (simple);
1278 g_object_unref (simple);
1282 g_buffered_input_stream_skip_async (GInputStream *stream,
1285 GCancellable *cancellable,
1286 GAsyncReadyCallback callback,
1289 GBufferedInputStream *bstream;
1290 GBufferedInputStreamPrivate *priv;
1291 GInputStream *base_stream;
1293 GSimpleAsyncResult *simple;
1294 SkipAsyncData *data;
1296 bstream = G_BUFFERED_INPUT_STREAM (stream);
1297 priv = bstream->priv;
1299 data = g_slice_new (SkipAsyncData);
1300 data->bytes_skipped = 0;
1301 simple = g_simple_async_result_new (G_OBJECT (stream),
1302 callback, user_data,
1303 g_buffered_input_stream_skip_async);
1304 g_simple_async_result_set_op_res_gpointer (simple, data, free_skip_async_data);
1306 available = priv->end - priv->pos;
1308 if (count <= available)
1311 data->bytes_skipped = count;
1313 g_simple_async_result_complete_in_idle (simple);
1314 g_object_unref (simple);
1319 /* Full request not available, skip all currently availbile and request refill for more */
1326 data->bytes_skipped = available;
1327 data->count = count;
1329 if (count > priv->len)
1331 /* Large request, shortcut buffer */
1333 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1335 g_input_stream_skip_async (base_stream,
1337 io_priority, cancellable,
1338 large_skip_callback,
1343 g_input_stream_set_pending (stream, FALSE); /* to avoid already pending error */
1344 g_buffered_input_stream_fill_async (bstream, priv->len,
1345 io_priority, cancellable,
1346 skip_fill_buffer_callback, simple);
1351 g_buffered_input_stream_skip_finish (GInputStream *stream,
1352 GAsyncResult *result,
1355 GSimpleAsyncResult *simple;
1356 SkipAsyncData *data;
1358 simple = G_SIMPLE_ASYNC_RESULT (result);
1360 g_assert (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_skip_async);
1362 data = g_simple_async_result_get_op_res_gpointer (simple);
1364 return data->bytes_skipped;
1368 #define __G_BUFFERED_INPUT_STREAM_C__
1369 #include "gioaliasdef.c"