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: (array length=count) (element-type guint8): a pointer to
570 * an allocated chunk of memory
574 * Peeks in the buffer, copying data of size @count into @buffer,
575 * offset @offset bytes.
577 * Returns: a #gsize of the number of bytes peeked, or -1 on error.
580 g_buffered_input_stream_peek (GBufferedInputStream
*stream
,
588 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), -1);
589 g_return_val_if_fail (buffer
!= NULL
, -1);
591 available
= g_buffered_input_stream_get_available (stream
);
593 if (offset
> available
)
596 end
= MIN (offset
+ count
, available
);
597 count
= end
- offset
;
599 memcpy (buffer
, stream
->priv
->buffer
+ stream
->priv
->pos
+ offset
, count
);
604 * g_buffered_input_stream_peek_buffer:
605 * @stream: a #GBufferedInputStream
606 * @count: (out): a #gsize to get the number of bytes available in the buffer
608 * Returns the buffer with the currently available bytes. The returned
609 * buffer must not be modified and will become invalid when reading from
610 * the stream or filling the buffer.
612 * Returns: (array length=count) (element-type guint8) (transfer none):
616 g_buffered_input_stream_peek_buffer (GBufferedInputStream
*stream
,
619 GBufferedInputStreamPrivate
*priv
;
621 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), NULL
);
626 *count
= priv
->end
- priv
->pos
;
628 return priv
->buffer
+ priv
->pos
;
632 compact_buffer (GBufferedInputStream
*stream
)
634 GBufferedInputStreamPrivate
*priv
;
639 current_size
= priv
->end
- priv
->pos
;
641 g_memmove (priv
->buffer
, priv
->buffer
+ priv
->pos
, current_size
);
644 priv
->end
= current_size
;
648 g_buffered_input_stream_real_fill (GBufferedInputStream
*stream
,
650 GCancellable
*cancellable
,
653 GBufferedInputStreamPrivate
*priv
;
654 GInputStream
*base_stream
;
663 in_buffer
= priv
->end
- priv
->pos
;
665 /* Never fill more than can fit in the buffer */
666 count
= MIN (count
, priv
->len
- in_buffer
);
668 /* If requested length does not fit at end, compact */
669 if (priv
->len
- priv
->end
< count
)
670 compact_buffer (stream
);
672 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
673 nread
= g_input_stream_read (base_stream
,
674 priv
->buffer
+ priv
->end
,
686 g_buffered_input_stream_skip (GInputStream
*stream
,
688 GCancellable
*cancellable
,
691 GBufferedInputStream
*bstream
;
692 GBufferedInputStreamPrivate
*priv
;
693 GBufferedInputStreamClass
*class;
694 GInputStream
*base_stream
;
695 gsize available
, bytes_skipped
;
698 bstream
= G_BUFFERED_INPUT_STREAM (stream
);
699 priv
= bstream
->priv
;
701 available
= priv
->end
- priv
->pos
;
703 if (count
<= available
)
709 /* Full request not available, skip all currently available and
710 * request refill for more
715 bytes_skipped
= available
;
718 if (bytes_skipped
> 0)
719 error
= NULL
; /* Ignore further errors if we already read some data */
721 if (count
> priv
->len
)
723 /* Large request, shortcut buffer */
725 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
727 nread
= g_input_stream_skip (base_stream
,
732 if (nread
< 0 && bytes_skipped
== 0)
736 bytes_skipped
+= nread
;
738 return bytes_skipped
;
741 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
742 nread
= class->fill (bstream
, priv
->len
, cancellable
, error
);
746 if (bytes_skipped
== 0)
749 return bytes_skipped
;
752 available
= priv
->end
- priv
->pos
;
753 count
= MIN (count
, available
);
755 bytes_skipped
+= count
;
758 return bytes_skipped
;
762 g_buffered_input_stream_read (GInputStream
*stream
,
765 GCancellable
*cancellable
,
768 GBufferedInputStream
*bstream
;
769 GBufferedInputStreamPrivate
*priv
;
770 GBufferedInputStreamClass
*class;
771 GInputStream
*base_stream
;
772 gsize available
, bytes_read
;
775 bstream
= G_BUFFERED_INPUT_STREAM (stream
);
776 priv
= bstream
->priv
;
778 available
= priv
->end
- priv
->pos
;
780 if (count
<= available
)
782 memcpy (buffer
, priv
->buffer
+ priv
->pos
, count
);
787 /* Full request not available, read all currently available and
788 * request refill for more
791 memcpy (buffer
, priv
->buffer
+ priv
->pos
, available
);
794 bytes_read
= available
;
798 error
= NULL
; /* Ignore further errors if we already read some data */
800 if (count
> priv
->len
)
802 /* Large request, shortcut buffer */
804 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
806 nread
= g_input_stream_read (base_stream
,
807 (char *)buffer
+ bytes_read
,
812 if (nread
< 0 && bytes_read
== 0)
821 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
822 nread
= class->fill (bstream
, priv
->len
, cancellable
, error
);
831 available
= priv
->end
- priv
->pos
;
832 count
= MIN (count
, available
);
834 memcpy ((char *)buffer
+ bytes_read
, (char *)priv
->buffer
+ priv
->pos
, count
);
842 * g_buffered_input_stream_read_byte:
843 * @stream: a #GBufferedInputStream
844 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
845 * @error: location to store the error occurring, or %NULL to ignore
847 * Tries to read a single byte from the stream or the buffer. Will block
850 * On success, the byte read from the stream is returned. On end of stream
851 * -1 is returned but it's not an exceptional error and @error is not set.
853 * If @cancellable is not %NULL, then the operation can be cancelled by
854 * triggering the cancellable object from another thread. If the operation
855 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
856 * operation was partially finished when the operation was cancelled the
857 * partial result will be returned, without an error.
859 * On error -1 is returned and @error is set accordingly.
861 * Returns: the byte read from the @stream, or -1 on end of stream or error.
864 g_buffered_input_stream_read_byte (GBufferedInputStream
*stream
,
865 GCancellable
*cancellable
,
868 GBufferedInputStreamPrivate
*priv
;
869 GBufferedInputStreamClass
*class;
870 GInputStream
*input_stream
;
874 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream
), -1);
877 input_stream
= G_INPUT_STREAM (stream
);
879 if (g_input_stream_is_closed (input_stream
))
881 g_set_error_literal (error
, G_IO_ERROR
, G_IO_ERROR_CLOSED
,
882 _("Stream is already closed"));
886 if (!g_input_stream_set_pending (input_stream
, error
))
889 available
= priv
->end
- priv
->pos
;
893 g_input_stream_clear_pending (input_stream
);
894 return priv
->buffer
[priv
->pos
++];
897 /* Byte not available, request refill for more */
900 g_cancellable_push_current (cancellable
);
905 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
906 nread
= class->fill (stream
, priv
->len
, cancellable
, error
);
909 g_cancellable_pop_current (cancellable
);
911 g_input_stream_clear_pending (input_stream
);
914 return -1; /* error or end of stream */
916 return priv
->buffer
[priv
->pos
++];
919 /* ************************** */
920 /* Async stuff implementation */
921 /* ************************** */
924 fill_async_callback (GObject
*source_object
,
925 GAsyncResult
*result
,
930 GSimpleAsyncResult
*simple
;
935 res
= g_input_stream_read_finish (G_INPUT_STREAM (source_object
),
938 g_simple_async_result_set_op_res_gssize (simple
, res
);
941 g_simple_async_result_take_error (simple
, error
);
945 GBufferedInputStreamPrivate
*priv
;
948 object
= g_async_result_get_source_object (G_ASYNC_RESULT (simple
));
949 priv
= G_BUFFERED_INPUT_STREAM (object
)->priv
;
951 g_assert_cmpint (priv
->end
+ res
, <=, priv
->len
);
954 g_object_unref (object
);
957 /* Complete immediately, not in idle, since we're already
958 * 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
);
1028 free_read_async_data (gpointer _data
)
1030 ReadAsyncData
*data
= _data
;
1031 g_slice_free (ReadAsyncData
, data
);
1035 large_read_callback (GObject
*source_object
,
1036 GAsyncResult
*result
,
1039 GSimpleAsyncResult
*simple
= G_SIMPLE_ASYNC_RESULT (user_data
);
1040 ReadAsyncData
*data
;
1044 data
= g_simple_async_result_get_op_res_gpointer (simple
);
1047 nread
= g_input_stream_read_finish (G_INPUT_STREAM (source_object
),
1050 /* Only report the error if we've not already read some data */
1051 if (nread
< 0 && data
->bytes_read
== 0)
1052 g_simple_async_result_take_error (simple
, error
);
1054 g_error_free (error
);
1057 data
->bytes_read
+= nread
;
1059 /* Complete immediately, not in idle, since we're already
1060 * in a mainloop callout
1062 g_simple_async_result_complete (simple
);
1063 g_object_unref (simple
);
1067 read_fill_buffer_callback (GObject
*source_object
,
1068 GAsyncResult
*result
,
1071 GSimpleAsyncResult
*simple
= G_SIMPLE_ASYNC_RESULT (user_data
);
1072 GBufferedInputStream
*bstream
;
1073 GBufferedInputStreamPrivate
*priv
;
1074 ReadAsyncData
*data
;
1079 bstream
= G_BUFFERED_INPUT_STREAM (source_object
);
1080 priv
= bstream
->priv
;
1082 data
= g_simple_async_result_get_op_res_gpointer (simple
);
1085 nread
= g_buffered_input_stream_fill_finish (bstream
,
1088 if (nread
< 0 && data
->bytes_read
== 0)
1089 g_simple_async_result_take_error (simple
, error
);
1091 g_error_free (error
);
1095 available
= priv
->end
- priv
->pos
;
1096 data
->count
= MIN (data
->count
, available
);
1098 memcpy ((char *)data
->buffer
+ data
->bytes_read
, (char *)priv
->buffer
+ priv
->pos
, data
->count
);
1099 data
->bytes_read
+= data
->count
;
1100 priv
->pos
+= data
->count
;
1103 /* Complete immediately, not in idle, since we're already
1104 * in a mainloop callout
1106 g_simple_async_result_complete (simple
);
1107 g_object_unref (simple
);
1111 g_buffered_input_stream_read_async (GInputStream
*stream
,
1115 GCancellable
*cancellable
,
1116 GAsyncReadyCallback callback
,
1119 GBufferedInputStream
*bstream
;
1120 GBufferedInputStreamPrivate
*priv
;
1121 GBufferedInputStreamClass
*class;
1122 GInputStream
*base_stream
;
1124 GSimpleAsyncResult
*simple
;
1125 ReadAsyncData
*data
;
1127 bstream
= G_BUFFERED_INPUT_STREAM (stream
);
1128 priv
= bstream
->priv
;
1130 data
= g_slice_new (ReadAsyncData
);
1131 data
->buffer
= buffer
;
1132 data
->bytes_read
= 0;
1133 simple
= g_simple_async_result_new (G_OBJECT (stream
),
1134 callback
, user_data
,
1135 g_buffered_input_stream_read_async
);
1136 g_simple_async_result_set_op_res_gpointer (simple
, data
, free_read_async_data
);
1138 available
= priv
->end
- priv
->pos
;
1140 if (count
<= available
)
1142 memcpy (buffer
, priv
->buffer
+ priv
->pos
, count
);
1144 data
->bytes_read
= count
;
1146 g_simple_async_result_complete_in_idle (simple
);
1147 g_object_unref (simple
);
1152 /* Full request not available, read all currently available
1153 * and request refill for more
1156 memcpy (buffer
, priv
->buffer
+ priv
->pos
, available
);
1162 data
->bytes_read
= available
;
1163 data
->count
= count
;
1165 if (count
> priv
->len
)
1167 /* Large request, shortcut buffer */
1169 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
1171 g_input_stream_read_async (base_stream
,
1172 (char *)buffer
+ data
->bytes_read
,
1174 io_priority
, cancellable
,
1175 large_read_callback
,
1180 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
1181 class->fill_async (bstream
, priv
->len
, io_priority
, cancellable
,
1182 read_fill_buffer_callback
, simple
);
1187 g_buffered_input_stream_read_finish (GInputStream
*stream
,
1188 GAsyncResult
*result
,
1191 GSimpleAsyncResult
*simple
;
1192 ReadAsyncData
*data
;
1194 simple
= G_SIMPLE_ASYNC_RESULT (result
);
1196 g_warn_if_fail (g_simple_async_result_get_source_tag (simple
) == g_buffered_input_stream_read_async
);
1198 data
= g_simple_async_result_get_op_res_gpointer (simple
);
1200 return data
->bytes_read
;
1205 gssize bytes_skipped
;
1210 free_skip_async_data (gpointer _data
)
1212 SkipAsyncData
*data
= _data
;
1213 g_slice_free (SkipAsyncData
, data
);
1217 large_skip_callback (GObject
*source_object
,
1218 GAsyncResult
*result
,
1221 GSimpleAsyncResult
*simple
= G_SIMPLE_ASYNC_RESULT (user_data
);
1222 SkipAsyncData
*data
;
1226 data
= g_simple_async_result_get_op_res_gpointer (simple
);
1229 nread
= g_input_stream_skip_finish (G_INPUT_STREAM (source_object
),
1232 /* Only report the error if we've not already read some data */
1233 if (nread
< 0 && data
->bytes_skipped
== 0)
1234 g_simple_async_result_take_error (simple
, error
);
1236 g_error_free (error
);
1239 data
->bytes_skipped
+= nread
;
1241 /* Complete immediately, not in idle, since we're already
1242 * in a mainloop callout
1244 g_simple_async_result_complete (simple
);
1245 g_object_unref (simple
);
1249 skip_fill_buffer_callback (GObject
*source_object
,
1250 GAsyncResult
*result
,
1253 GSimpleAsyncResult
*simple
= G_SIMPLE_ASYNC_RESULT (user_data
);
1254 GBufferedInputStream
*bstream
;
1255 GBufferedInputStreamPrivate
*priv
;
1256 SkipAsyncData
*data
;
1261 bstream
= G_BUFFERED_INPUT_STREAM (source_object
);
1262 priv
= bstream
->priv
;
1264 data
= g_simple_async_result_get_op_res_gpointer (simple
);
1267 nread
= g_buffered_input_stream_fill_finish (bstream
,
1270 if (nread
< 0 && data
->bytes_skipped
== 0)
1271 g_simple_async_result_take_error (simple
, error
);
1273 g_error_free (error
);
1277 available
= priv
->end
- priv
->pos
;
1278 data
->count
= MIN (data
->count
, available
);
1280 data
->bytes_skipped
+= data
->count
;
1281 priv
->pos
+= data
->count
;
1284 /* Complete immediately, not in idle, since we're already
1285 * in a mainloop callout
1287 g_simple_async_result_complete (simple
);
1288 g_object_unref (simple
);
1292 g_buffered_input_stream_skip_async (GInputStream
*stream
,
1295 GCancellable
*cancellable
,
1296 GAsyncReadyCallback callback
,
1299 GBufferedInputStream
*bstream
;
1300 GBufferedInputStreamPrivate
*priv
;
1301 GBufferedInputStreamClass
*class;
1302 GInputStream
*base_stream
;
1304 GSimpleAsyncResult
*simple
;
1305 SkipAsyncData
*data
;
1307 bstream
= G_BUFFERED_INPUT_STREAM (stream
);
1308 priv
= bstream
->priv
;
1310 data
= g_slice_new (SkipAsyncData
);
1311 data
->bytes_skipped
= 0;
1312 simple
= g_simple_async_result_new (G_OBJECT (stream
),
1313 callback
, user_data
,
1314 g_buffered_input_stream_skip_async
);
1315 g_simple_async_result_set_op_res_gpointer (simple
, data
, free_skip_async_data
);
1317 available
= priv
->end
- priv
->pos
;
1319 if (count
<= available
)
1322 data
->bytes_skipped
= count
;
1324 g_simple_async_result_complete_in_idle (simple
);
1325 g_object_unref (simple
);
1329 /* Full request not available, skip all currently available
1330 * and request refill for more
1338 data
->bytes_skipped
= available
;
1339 data
->count
= count
;
1341 if (count
> priv
->len
)
1343 /* Large request, shortcut buffer */
1345 base_stream
= G_FILTER_INPUT_STREAM (stream
)->base_stream
;
1347 g_input_stream_skip_async (base_stream
,
1349 io_priority
, cancellable
,
1350 large_skip_callback
,
1355 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream
);
1356 class->fill_async (bstream
, priv
->len
, io_priority
, cancellable
,
1357 skip_fill_buffer_callback
, simple
);
1362 g_buffered_input_stream_skip_finish (GInputStream
*stream
,
1363 GAsyncResult
*result
,
1366 GSimpleAsyncResult
*simple
;
1367 SkipAsyncData
*data
;
1369 simple
= G_SIMPLE_ASYNC_RESULT (result
);
1371 g_warn_if_fail (g_simple_async_result_get_source_tag (simple
) == g_buffered_input_stream_skip_async
);
1373 data
= g_simple_async_result_get_op_res_gpointer (simple
);
1375 return data
->bytes_skipped
;