Revert "GIOScheduler: Avoid constant iteration over pending job list"
[platform/upstream/glib.git] / gio / gbufferedinputstream.c
1 /* GIO - GLib Input, Output and Streaming Library
2  *
3  * Copyright (C) 2006-2007 Red Hat, Inc.
4  * Copyright (C) 2007 Jürg Billeter
5  *
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.
10  *
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.
15  *
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.
20  *
21  * Author: Christian Kellner <gicmo@gnome.org>
22  */
23
24 #include "config.h"
25 #include "gbufferedinputstream.h"
26 #include "ginputstream.h"
27 #include "gcancellable.h"
28 #include "gasyncresult.h"
29 #include "gsimpleasyncresult.h"
30 #include "gseekable.h"
31 #include "gioerror.h"
32 #include <string.h>
33 #include "glibintl.h"
34
35
36 /**
37  * SECTION:gbufferedinputstream
38  * @short_description: Buffered Input Stream
39  * @include: gio/gio.h
40  * @see_also: #GFilterInputStream, #GInputStream
41  *
42  * Buffered input stream implements #GFilterInputStream and provides
43  * for buffered reads.
44  *
45  * By default, #GBufferedInputStream's buffer size is set at 4 kilobytes.
46  *
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
49  * construction.
50  *
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.
56  */
57
58
59 #define DEFAULT_BUFFER_SIZE 4096
60
61 struct _GBufferedInputStreamPrivate {
62   guint8 *buffer;
63   gsize   len;
64   gsize   pos;
65   gsize   end;
66   GAsyncReadyCallback outstanding_callback;
67 };
68
69 enum {
70   PROP_0,
71   PROP_BUFSIZE
72 };
73
74 static void g_buffered_input_stream_set_property  (GObject      *object,
75                                                    guint         prop_id,
76                                                    const GValue *value,
77                                                    GParamSpec   *pspec);
78
79 static void g_buffered_input_stream_get_property  (GObject      *object,
80                                                    guint         prop_id,
81                                                    GValue       *value,
82                                                    GParamSpec   *pspec);
83 static void g_buffered_input_stream_finalize      (GObject *object);
84
85
86 static gssize g_buffered_input_stream_skip             (GInputStream          *stream,
87                                                         gsize                  count,
88                                                         GCancellable          *cancellable,
89                                                         GError               **error);
90 static void   g_buffered_input_stream_skip_async       (GInputStream          *stream,
91                                                         gsize                  count,
92                                                         int                    io_priority,
93                                                         GCancellable          *cancellable,
94                                                         GAsyncReadyCallback    callback,
95                                                         gpointer               user_data);
96 static gssize g_buffered_input_stream_skip_finish      (GInputStream          *stream,
97                                                         GAsyncResult          *result,
98                                                         GError               **error);
99 static gssize g_buffered_input_stream_read             (GInputStream          *stream,
100                                                         void                  *buffer,
101                                                         gsize                  count,
102                                                         GCancellable          *cancellable,
103                                                         GError               **error);
104 static gssize g_buffered_input_stream_real_fill        (GBufferedInputStream  *stream,
105                                                         gssize                 count,
106                                                         GCancellable          *cancellable,
107                                                         GError               **error);
108 static void   g_buffered_input_stream_real_fill_async  (GBufferedInputStream  *stream,
109                                                         gssize                 count,
110                                                         int                    io_priority,
111                                                         GCancellable          *cancellable,
112                                                         GAsyncReadyCallback    callback,
113                                                         gpointer               user_data);
114 static gssize g_buffered_input_stream_real_fill_finish (GBufferedInputStream  *stream,
115                                                         GAsyncResult          *result,
116                                                         GError               **error);
117
118 static void     g_buffered_input_stream_seekable_iface_init (GSeekableIface  *iface);
119 static goffset  g_buffered_input_stream_tell                (GSeekable       *seekable);
120 static gboolean g_buffered_input_stream_can_seek            (GSeekable       *seekable);
121 static gboolean g_buffered_input_stream_seek                (GSeekable       *seekable,
122                                                              goffset          offset,
123                                                              GSeekType        type,
124                                                              GCancellable    *cancellable,
125                                                              GError         **error);
126 static gboolean g_buffered_input_stream_can_truncate        (GSeekable       *seekable);
127 static gboolean g_buffered_input_stream_truncate            (GSeekable       *seekable,
128                                                              goffset          offset,
129                                                              GCancellable    *cancellable,
130                                                              GError         **error);
131
132 static void     g_buffered_input_stream_finalize            (GObject         *object);
133
134 static void compact_buffer (GBufferedInputStream *stream);
135
136 G_DEFINE_TYPE_WITH_CODE (GBufferedInputStream,
137                          g_buffered_input_stream,
138                          G_TYPE_FILTER_INPUT_STREAM,
139                          G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE,
140                                                 g_buffered_input_stream_seekable_iface_init))
141
142 static void
143 g_buffered_input_stream_class_init (GBufferedInputStreamClass *klass)
144 {
145   GObjectClass *object_class;
146   GInputStreamClass *istream_class;
147   GBufferedInputStreamClass *bstream_class;
148
149   g_type_class_add_private (klass, sizeof (GBufferedInputStreamPrivate));
150
151   object_class = G_OBJECT_CLASS (klass);
152   object_class->get_property = g_buffered_input_stream_get_property;
153   object_class->set_property = g_buffered_input_stream_set_property;
154   object_class->finalize     = g_buffered_input_stream_finalize;
155
156   istream_class = G_INPUT_STREAM_CLASS (klass);
157   istream_class->skip = g_buffered_input_stream_skip;
158   istream_class->skip_async  = g_buffered_input_stream_skip_async;
159   istream_class->skip_finish = g_buffered_input_stream_skip_finish;
160   istream_class->read_fn = g_buffered_input_stream_read;
161
162   bstream_class = G_BUFFERED_INPUT_STREAM_CLASS (klass);
163   bstream_class->fill = g_buffered_input_stream_real_fill;
164   bstream_class->fill_async = g_buffered_input_stream_real_fill_async;
165   bstream_class->fill_finish = g_buffered_input_stream_real_fill_finish;
166
167   g_object_class_install_property (object_class,
168                                    PROP_BUFSIZE,
169                                    g_param_spec_uint ("buffer-size",
170                                                       P_("Buffer Size"),
171                                                       P_("The size of the backend buffer"),
172                                                       1,
173                                                       G_MAXUINT,
174                                                       DEFAULT_BUFFER_SIZE,
175                                                       G_PARAM_READWRITE | G_PARAM_CONSTRUCT |
176                                                       G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
177
178
179 }
180
181 /**
182  * g_buffered_input_stream_get_buffer_size:
183  * @stream: a #GBufferedInputStream
184  *
185  * Gets the size of the input buffer.
186  *
187  * Returns: the current buffer size.
188  */
189 gsize
190 g_buffered_input_stream_get_buffer_size (GBufferedInputStream  *stream)
191 {
192   g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), 0);
193
194   return stream->priv->len;
195 }
196
197 /**
198  * g_buffered_input_stream_set_buffer_size:
199  * @stream: a #GBufferedInputStream
200  * @size: a #gsize
201  *
202  * Sets the size of the internal buffer of @stream to @size, or to the
203  * size of the contents of the buffer. The buffer can never be resized
204  * smaller than its current contents.
205  */
206 void
207 g_buffered_input_stream_set_buffer_size (GBufferedInputStream *stream,
208                                          gsize                 size)
209 {
210   GBufferedInputStreamPrivate *priv;
211   gsize in_buffer;
212   guint8 *buffer;
213
214   g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
215
216   priv = stream->priv;
217
218   if (priv->len == size)
219     return;
220
221   if (priv->buffer)
222     {
223       in_buffer = priv->end - priv->pos;
224
225       /* Never resize smaller than current buffer contents */
226       size = MAX (size, in_buffer);
227
228       buffer = g_malloc (size);
229       memcpy (buffer, priv->buffer + priv->pos, in_buffer);
230       priv->len = size;
231       priv->pos = 0;
232       priv->end = in_buffer;
233       g_free (priv->buffer);
234       priv->buffer = buffer;
235     }
236   else
237     {
238       priv->len = size;
239       priv->pos = 0;
240       priv->end = 0;
241       priv->buffer = g_malloc (size);
242     }
243
244   g_object_notify (G_OBJECT (stream), "buffer-size");
245 }
246
247 static void
248 g_buffered_input_stream_set_property (GObject      *object,
249                                       guint         prop_id,
250                                       const GValue *value,
251                                       GParamSpec   *pspec)
252 {
253   GBufferedInputStream        *bstream;
254
255   bstream = G_BUFFERED_INPUT_STREAM (object);
256
257   switch (prop_id)
258     {
259     case PROP_BUFSIZE:
260       g_buffered_input_stream_set_buffer_size (bstream, g_value_get_uint (value));
261       break;
262
263     default:
264       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
265       break;
266     }
267 }
268
269 static void
270 g_buffered_input_stream_get_property (GObject    *object,
271                                       guint       prop_id,
272                                       GValue     *value,
273                                       GParamSpec *pspec)
274 {
275   GBufferedInputStreamPrivate *priv;
276   GBufferedInputStream        *bstream;
277
278   bstream = G_BUFFERED_INPUT_STREAM (object);
279   priv = bstream->priv;
280
281   switch (prop_id)
282     {
283     case PROP_BUFSIZE:
284       g_value_set_uint (value, priv->len);
285       break;
286
287     default:
288       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
289       break;
290     }
291 }
292
293 static void
294 g_buffered_input_stream_finalize (GObject *object)
295 {
296   GBufferedInputStreamPrivate *priv;
297   GBufferedInputStream        *stream;
298
299   stream = G_BUFFERED_INPUT_STREAM (object);
300   priv = stream->priv;
301
302   g_free (priv->buffer);
303
304   G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize (object);
305 }
306
307 static void
308 g_buffered_input_stream_seekable_iface_init (GSeekableIface *iface)
309 {
310   iface->tell         = g_buffered_input_stream_tell;
311   iface->can_seek     = g_buffered_input_stream_can_seek;
312   iface->seek         = g_buffered_input_stream_seek;
313   iface->can_truncate = g_buffered_input_stream_can_truncate;
314   iface->truncate_fn  = g_buffered_input_stream_truncate;
315 }
316
317 static void
318 g_buffered_input_stream_init (GBufferedInputStream *stream)
319 {
320   stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
321                                               G_TYPE_BUFFERED_INPUT_STREAM,
322                                               GBufferedInputStreamPrivate);
323 }
324
325
326 /**
327  * g_buffered_input_stream_new:
328  * @base_stream: a #GInputStream
329  *
330  * Creates a new #GInputStream from the given @base_stream, with
331  * a buffer set to the default size (4 kilobytes).
332  *
333  * Returns: a #GInputStream for the given @base_stream.
334  */
335 GInputStream *
336 g_buffered_input_stream_new (GInputStream *base_stream)
337 {
338   GInputStream *stream;
339
340   g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
341
342   stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
343                          "base-stream", base_stream,
344                          NULL);
345
346   return stream;
347 }
348
349 /**
350  * g_buffered_input_stream_new_sized:
351  * @base_stream: a #GInputStream
352  * @size: a #gsize
353  *
354  * Creates a new #GBufferedInputStream from the given @base_stream,
355  * with a buffer set to @size.
356  *
357  * Returns: a #GInputStream.
358  */
359 GInputStream *
360 g_buffered_input_stream_new_sized (GInputStream *base_stream,
361                                    gsize         size)
362 {
363   GInputStream *stream;
364
365   g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
366
367   stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
368                          "base-stream", base_stream,
369                          "buffer-size", (guint)size,
370                          NULL);
371
372   return stream;
373 }
374
375 /**
376  * g_buffered_input_stream_fill:
377  * @stream: a #GBufferedInputStream
378  * @count: the number of bytes that will be read from the stream
379  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
380  * @error: location to store the error occurring, or %NULL to ignore
381  *
382  * Tries to read @count bytes from the stream into the buffer.
383  * Will block during this read.
384  *
385  * If @count is zero, returns zero and does nothing. A value of @count
386  * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
387  *
388  * On success, the number of bytes read into the buffer is returned.
389  * It is not an error if this is not the same as the requested size, as it
390  * can happen e.g. near the end of a file. Zero is returned on end of file
391  * (or if @count is zero),  but never otherwise.
392  *
393  * If @count is -1 then the attempted read size is equal to the number of
394  * bytes that are required to fill the buffer.
395  *
396  * If @cancellable is not %NULL, then the operation can be cancelled by
397  * triggering the cancellable object from another thread. If the operation
398  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
399  * operation was partially finished when the operation was cancelled the
400  * partial result will be returned, without an error.
401  *
402  * On error -1 is returned and @error is set accordingly.
403  *
404  * For the asynchronous, non-blocking, version of this function, see
405  * g_buffered_input_stream_fill_async().
406  *
407  * Returns: the number of bytes read into @stream's buffer, up to @count,
408  *     or -1 on error.
409  */
410 gssize
411 g_buffered_input_stream_fill (GBufferedInputStream  *stream,
412                               gssize                 count,
413                               GCancellable          *cancellable,
414                               GError               **error)
415 {
416   GBufferedInputStreamClass *class;
417   GInputStream *input_stream;
418   gssize res;
419
420   g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
421
422   input_stream = G_INPUT_STREAM (stream);
423
424   if (count < -1)
425     {
426       g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
427                    _("Too large count value passed to %s"), G_STRFUNC);
428       return -1;
429     }
430
431   if (!g_input_stream_set_pending (input_stream, error))
432     return -1;
433
434   if (cancellable)
435     g_cancellable_push_current (cancellable);
436
437   class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
438   res = class->fill (stream, count, cancellable, error);
439
440   if (cancellable)
441     g_cancellable_pop_current (cancellable);
442
443   g_input_stream_clear_pending (input_stream);
444
445   return res;
446 }
447
448 static void
449 async_fill_callback_wrapper (GObject      *source_object,
450                              GAsyncResult *res,
451                              gpointer      user_data)
452 {
453   GBufferedInputStream *stream = G_BUFFERED_INPUT_STREAM (source_object);
454
455   g_input_stream_clear_pending (G_INPUT_STREAM (stream));
456   (*stream->priv->outstanding_callback) (source_object, res, user_data);
457   g_object_unref (stream);
458 }
459
460 /**
461  * g_buffered_input_stream_fill_async:
462  * @stream: a #GBufferedInputStream
463  * @count: the number of bytes that will be read from the stream
464  * @io_priority: the <link linkend="io-priority">I/O priority</link>
465  *     of the request
466  * @cancellable: (allow-none): optional #GCancellable object
467  * @callback: (scope async): a #GAsyncReadyCallback
468  * @user_data: (closure): a #gpointer
469  *
470  * Reads data into @stream's buffer asynchronously, up to @count size.
471  * @io_priority can be used to prioritize reads. For the synchronous
472  * version of this function, see g_buffered_input_stream_fill().
473  *
474  * If @count is -1 then the attempted read size is equal to the number
475  * of bytes that are required to fill the buffer.
476  */
477 void
478 g_buffered_input_stream_fill_async (GBufferedInputStream *stream,
479                                     gssize                count,
480                                     int                   io_priority,
481                                     GCancellable         *cancellable,
482                                     GAsyncReadyCallback   callback,
483                                     gpointer              user_data)
484 {
485   GBufferedInputStreamClass *class;
486   GSimpleAsyncResult *simple;
487   GError *error = NULL;
488
489   g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
490
491   if (count == 0)
492     {
493       simple = g_simple_async_result_new (G_OBJECT (stream),
494                                           callback,
495                                           user_data,
496                                           g_buffered_input_stream_fill_async);
497       g_simple_async_result_complete_in_idle (simple);
498       g_object_unref (simple);
499       return;
500     }
501
502   if (count < -1)
503     {
504       g_simple_async_report_error_in_idle (G_OBJECT (stream),
505                                            callback,
506                                            user_data,
507                                            G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
508                                            _("Too large count value passed to %s"),
509                                            G_STRFUNC);
510       return;
511     }
512
513   if (!g_input_stream_set_pending (G_INPUT_STREAM (stream), &error))
514     {
515       g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
516                                             callback,
517                                             user_data,
518                                             error);
519       return;
520     }
521
522   class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
523
524   stream->priv->outstanding_callback = callback;
525   g_object_ref (stream);
526   class->fill_async (stream, count, io_priority, cancellable,
527                      async_fill_callback_wrapper, user_data);
528 }
529
530 /**
531  * g_buffered_input_stream_fill_finish:
532  * @stream: a #GBufferedInputStream
533  * @result: a #GAsyncResult
534  * @error: a #GError
535  *
536  * Finishes an asynchronous read.
537  *
538  * Returns: a #gssize of the read stream, or %-1 on an error.
539  */
540 gssize
541 g_buffered_input_stream_fill_finish (GBufferedInputStream  *stream,
542                                      GAsyncResult          *result,
543                                      GError               **error)
544 {
545   GSimpleAsyncResult *simple;
546   GBufferedInputStreamClass *class;
547
548   g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
549   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
550
551   if (G_IS_SIMPLE_ASYNC_RESULT (result))
552     {
553       simple = G_SIMPLE_ASYNC_RESULT (result);
554       if (g_simple_async_result_propagate_error (simple, error))
555         return -1;
556
557       /* Special case read of 0 bytes */
558       if (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_fill_async)
559         return 0;
560     }
561
562   class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
563   return class->fill_finish (stream, result, error);
564 }
565
566 /**
567  * g_buffered_input_stream_get_available:
568  * @stream: #GBufferedInputStream
569  *
570  * Gets the size of the available data within the stream.
571  *
572  * Returns: size of the available stream.
573  */
574 gsize
575 g_buffered_input_stream_get_available (GBufferedInputStream *stream)
576 {
577   g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
578
579   return stream->priv->end - stream->priv->pos;
580 }
581
582 /**
583  * g_buffered_input_stream_peek:
584  * @stream: a #GBufferedInputStream
585  * @buffer: (array length=count) (element-type guint8): a pointer to
586  *   an allocated chunk of memory
587  * @offset: a #gsize
588  * @count: a #gsize
589  *
590  * Peeks in the buffer, copying data of size @count into @buffer,
591  * offset @offset bytes.
592  *
593  * Returns: a #gsize of the number of bytes peeked, or -1 on error.
594  */
595 gsize
596 g_buffered_input_stream_peek (GBufferedInputStream *stream,
597                               void                 *buffer,
598                               gsize                 offset,
599                               gsize                 count)
600 {
601   gsize available;
602   gsize end;
603
604   g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
605   g_return_val_if_fail (buffer != NULL, -1);
606
607   available = g_buffered_input_stream_get_available (stream);
608
609   if (offset > available)
610     return 0;
611
612   end = MIN (offset + count, available);
613   count = end - offset;
614
615   memcpy (buffer, stream->priv->buffer + stream->priv->pos + offset, count);
616   return count;
617 }
618
619 /**
620  * g_buffered_input_stream_peek_buffer:
621  * @stream: a #GBufferedInputStream
622  * @count: (out): a #gsize to get the number of bytes available in the buffer
623  *
624  * Returns the buffer with the currently available bytes. The returned
625  * buffer must not be modified and will become invalid when reading from
626  * the stream or filling the buffer.
627  *
628  * Returns: (array length=count) (element-type guint8) (transfer none):
629  *          read-only buffer
630  */
631 const void*
632 g_buffered_input_stream_peek_buffer (GBufferedInputStream *stream,
633                                      gsize                *count)
634 {
635   GBufferedInputStreamPrivate *priv;
636
637   g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), NULL);
638
639   priv = stream->priv;
640
641   if (count)
642     *count = priv->end - priv->pos;
643
644   return priv->buffer + priv->pos;
645 }
646
647 static void
648 compact_buffer (GBufferedInputStream *stream)
649 {
650   GBufferedInputStreamPrivate *priv;
651   gsize current_size;
652
653   priv = stream->priv;
654
655   current_size = priv->end - priv->pos;
656
657   g_memmove (priv->buffer, priv->buffer + priv->pos, current_size);
658
659   priv->pos = 0;
660   priv->end = current_size;
661 }
662
663 static gssize
664 g_buffered_input_stream_real_fill (GBufferedInputStream  *stream,
665                                    gssize                 count,
666                                    GCancellable          *cancellable,
667                                    GError               **error)
668 {
669   GBufferedInputStreamPrivate *priv;
670   GInputStream *base_stream;
671   gssize nread;
672   gsize in_buffer;
673
674   priv = stream->priv;
675
676   if (count == -1)
677     count = priv->len;
678
679   in_buffer = priv->end - priv->pos;
680
681   /* Never fill more than can fit in the buffer */
682   count = MIN (count, priv->len - in_buffer);
683
684   /* If requested length does not fit at end, compact */
685   if (priv->len - priv->end < count)
686     compact_buffer (stream);
687
688   base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
689   nread = g_input_stream_read (base_stream,
690                                priv->buffer + priv->end,
691                                count,
692                                cancellable,
693                                error);
694
695   if (nread > 0)
696     priv->end += nread;
697
698   return nread;
699 }
700
701 static gssize
702 g_buffered_input_stream_skip (GInputStream  *stream,
703                               gsize          count,
704                               GCancellable  *cancellable,
705                               GError       **error)
706 {
707   GBufferedInputStream        *bstream;
708   GBufferedInputStreamPrivate *priv;
709   GBufferedInputStreamClass *class;
710   GInputStream *base_stream;
711   gsize available, bytes_skipped;
712   gssize nread;
713
714   bstream = G_BUFFERED_INPUT_STREAM (stream);
715   priv = bstream->priv;
716
717   available = priv->end - priv->pos;
718
719   if (count <= available)
720     {
721       priv->pos += count;
722       return count;
723     }
724
725   /* Full request not available, skip all currently available and
726    * request refill for more
727    */
728
729   priv->pos = 0;
730   priv->end = 0;
731   bytes_skipped = available;
732   count -= available;
733
734   if (bytes_skipped > 0)
735     error = NULL; /* Ignore further errors if we already read some data */
736
737   if (count > priv->len)
738     {
739       /* Large request, shortcut buffer */
740
741       base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
742
743       nread = g_input_stream_skip (base_stream,
744                                    count,
745                                    cancellable,
746                                    error);
747
748       if (nread < 0 && bytes_skipped == 0)
749         return -1;
750
751       if (nread > 0)
752         bytes_skipped += nread;
753
754       return bytes_skipped;
755     }
756
757   class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
758   nread = class->fill (bstream, priv->len, cancellable, error);
759
760   if (nread < 0)
761     {
762       if (bytes_skipped == 0)
763         return -1;
764       else
765         return bytes_skipped;
766     }
767
768   available = priv->end - priv->pos;
769   count = MIN (count, available);
770
771   bytes_skipped += count;
772   priv->pos += count;
773
774   return bytes_skipped;
775 }
776
777 static gssize
778 g_buffered_input_stream_read (GInputStream *stream,
779                               void         *buffer,
780                               gsize         count,
781                               GCancellable *cancellable,
782                               GError      **error)
783 {
784   GBufferedInputStream        *bstream;
785   GBufferedInputStreamPrivate *priv;
786   GBufferedInputStreamClass *class;
787   GInputStream *base_stream;
788   gsize available, bytes_read;
789   gssize nread;
790
791   bstream = G_BUFFERED_INPUT_STREAM (stream);
792   priv = bstream->priv;
793
794   available = priv->end - priv->pos;
795
796   if (count <= available)
797     {
798       memcpy (buffer, priv->buffer + priv->pos, count);
799       priv->pos += count;
800       return count;
801     }
802
803   /* Full request not available, read all currently available and
804    * request refill for more
805    */
806
807   memcpy (buffer, priv->buffer + priv->pos, available);
808   priv->pos = 0;
809   priv->end = 0;
810   bytes_read = available;
811   count -= available;
812
813   if (bytes_read > 0)
814     error = NULL; /* Ignore further errors if we already read some data */
815
816   if (count > priv->len)
817     {
818       /* Large request, shortcut buffer */
819
820       base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
821
822       nread = g_input_stream_read (base_stream,
823                                    (char *)buffer + bytes_read,
824                                    count,
825                                    cancellable,
826                                    error);
827
828       if (nread < 0 && bytes_read == 0)
829         return -1;
830
831       if (nread > 0)
832         bytes_read += nread;
833
834       return bytes_read;
835     }
836
837   class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
838   nread = class->fill (bstream, priv->len, cancellable, error);
839   if (nread < 0)
840     {
841       if (bytes_read == 0)
842         return -1;
843       else
844         return bytes_read;
845     }
846
847   available = priv->end - priv->pos;
848   count = MIN (count, available);
849
850   memcpy ((char *)buffer + bytes_read, (char *)priv->buffer + priv->pos, count);
851   bytes_read += count;
852   priv->pos += count;
853
854   return bytes_read;
855 }
856
857 static goffset
858 g_buffered_input_stream_tell (GSeekable *seekable)
859 {
860   GBufferedInputStream        *bstream;
861   GBufferedInputStreamPrivate *priv;
862   GInputStream *base_stream;
863   GSeekable    *base_stream_seekable;
864   gsize available;
865   goffset base_offset;
866   
867   bstream = G_BUFFERED_INPUT_STREAM (seekable);
868   priv = bstream->priv;
869
870   base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
871   if (!G_IS_SEEKABLE (base_stream))
872     return 0;
873   base_stream_seekable = G_SEEKABLE (base_stream);
874   
875   available = priv->end - priv->pos;
876   base_offset = g_seekable_tell (base_stream_seekable);
877
878   return base_offset - available;
879 }
880
881 static gboolean
882 g_buffered_input_stream_can_seek (GSeekable *seekable)
883 {
884   GInputStream *base_stream;
885   
886   base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
887   return G_IS_SEEKABLE (base_stream) && g_seekable_can_seek (G_SEEKABLE (base_stream));
888 }
889
890 static gboolean
891 g_buffered_input_stream_seek (GSeekable     *seekable,
892                               goffset        offset,
893                               GSeekType      type,
894                               GCancellable  *cancellable,
895                               GError       **error)
896 {
897   GBufferedInputStream        *bstream;
898   GBufferedInputStreamPrivate *priv;
899   GInputStream *base_stream;
900   GSeekable *base_stream_seekable;
901
902   bstream = G_BUFFERED_INPUT_STREAM (seekable);
903   priv = bstream->priv;
904
905   base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
906   if (!G_IS_SEEKABLE (base_stream))
907     {
908       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
909                            _("Seek not supported on base stream"));
910       return FALSE;
911     }
912
913   base_stream_seekable = G_SEEKABLE (base_stream);
914   
915   if (type == G_SEEK_CUR)
916     {
917       if (offset <= priv->end - priv->pos && offset >= -priv->pos)
918         {
919           priv->pos += offset;
920           return TRUE;
921         }
922       else
923         {
924           offset -= priv->end - priv->pos;
925         }
926     }
927
928   if (g_seekable_seek (base_stream_seekable, offset, type, cancellable, error))
929     {
930       priv->pos = 0;
931       priv->end = 0;
932       return TRUE;
933     }
934   else
935     {
936       return FALSE;
937     }
938 }
939
940 static gboolean
941 g_buffered_input_stream_can_truncate (GSeekable *seekable)
942 {
943   return FALSE;
944 }
945
946 static gboolean
947 g_buffered_input_stream_truncate (GSeekable     *seekable,
948                                   goffset        offset,
949                                   GCancellable  *cancellable,
950                                   GError       **error)
951 {
952   g_set_error_literal (error,
953                        G_IO_ERROR,
954                        G_IO_ERROR_NOT_SUPPORTED,
955                        _("Cannot truncate GBufferedInputStream"));
956   return FALSE;
957 }
958
959 /**
960  * g_buffered_input_stream_read_byte:
961  * @stream: a #GBufferedInputStream
962  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
963  * @error: location to store the error occurring, or %NULL to ignore
964  *
965  * Tries to read a single byte from the stream or the buffer. Will block
966  * during this read.
967  *
968  * On success, the byte read from the stream is returned. On end of stream
969  * -1 is returned but it's not an exceptional error and @error is not set.
970  *
971  * If @cancellable is not %NULL, then the operation can be cancelled by
972  * triggering the cancellable object from another thread. If the operation
973  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
974  * operation was partially finished when the operation was cancelled the
975  * partial result will be returned, without an error.
976  *
977  * On error -1 is returned and @error is set accordingly.
978  *
979  * Returns: the byte read from the @stream, or -1 on end of stream or error.
980  */
981 int
982 g_buffered_input_stream_read_byte (GBufferedInputStream  *stream,
983                                    GCancellable          *cancellable,
984                                    GError               **error)
985 {
986   GBufferedInputStreamPrivate *priv;
987   GBufferedInputStreamClass *class;
988   GInputStream *input_stream;
989   gsize available;
990   gssize nread;
991
992   g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
993
994   priv = stream->priv;
995   input_stream = G_INPUT_STREAM (stream);
996
997   if (g_input_stream_is_closed (input_stream))
998     {
999       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
1000                            _("Stream is already closed"));
1001       return -1;
1002     }
1003
1004   if (!g_input_stream_set_pending (input_stream, error))
1005     return -1;
1006
1007   available = priv->end - priv->pos;
1008
1009   if (available != 0)
1010     {
1011       g_input_stream_clear_pending (input_stream);
1012       return priv->buffer[priv->pos++];
1013     }
1014
1015   /* Byte not available, request refill for more */
1016
1017   if (cancellable)
1018     g_cancellable_push_current (cancellable);
1019
1020   priv->pos = 0;
1021   priv->end = 0;
1022
1023   class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1024   nread = class->fill (stream, priv->len, cancellable, error);
1025
1026   if (cancellable)
1027     g_cancellable_pop_current (cancellable);
1028
1029   g_input_stream_clear_pending (input_stream);
1030
1031   if (nread <= 0)
1032     return -1; /* error or end of stream */
1033
1034   return priv->buffer[priv->pos++];
1035 }
1036
1037 /* ************************** */
1038 /* Async stuff implementation */
1039 /* ************************** */
1040
1041 static void
1042 fill_async_callback (GObject      *source_object,
1043                      GAsyncResult *result,
1044                      gpointer      user_data)
1045 {
1046   GError *error;
1047   gssize res;
1048   GSimpleAsyncResult *simple;
1049
1050   simple = user_data;
1051
1052   error = NULL;
1053   res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
1054                                     result, &error);
1055
1056   g_simple_async_result_set_op_res_gssize (simple, res);
1057   if (res == -1)
1058     {
1059       g_simple_async_result_take_error (simple, error);
1060     }
1061   else
1062     {
1063       GBufferedInputStreamPrivate *priv;
1064       GObject *object;
1065
1066       object = g_async_result_get_source_object (G_ASYNC_RESULT (simple));
1067       priv = G_BUFFERED_INPUT_STREAM (object)->priv;
1068
1069       g_assert_cmpint (priv->end + res, <=, priv->len);
1070       priv->end += res;
1071
1072       g_object_unref (object);
1073     }
1074
1075   /* Complete immediately, not in idle, since we're already
1076    * in a mainloop callout
1077    */
1078   g_simple_async_result_complete (simple);
1079   g_object_unref (simple);
1080 }
1081
1082 static void
1083 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
1084                                          gssize                count,
1085                                          int                   io_priority,
1086                                          GCancellable         *cancellable,
1087                                          GAsyncReadyCallback   callback,
1088                                          gpointer              user_data)
1089 {
1090   GBufferedInputStreamPrivate *priv;
1091   GInputStream *base_stream;
1092   GSimpleAsyncResult *simple;
1093   gsize in_buffer;
1094
1095   priv = stream->priv;
1096
1097   if (count == -1)
1098     count = priv->len;
1099
1100   in_buffer = priv->end - priv->pos;
1101
1102   /* Never fill more than can fit in the buffer */
1103   count = MIN (count, priv->len - in_buffer);
1104
1105   /* If requested length does not fit at end, compact */
1106   if (priv->len - priv->end < count)
1107     compact_buffer (stream);
1108
1109   simple = g_simple_async_result_new (G_OBJECT (stream),
1110                                       callback, user_data,
1111                                       g_buffered_input_stream_real_fill_async);
1112
1113   base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1114   g_input_stream_read_async (base_stream,
1115                              priv->buffer + priv->end,
1116                              count,
1117                              io_priority,
1118                              cancellable,
1119                              fill_async_callback,
1120                              simple);
1121 }
1122
1123 static gssize
1124 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1125                                           GAsyncResult         *result,
1126                                           GError              **error)
1127 {
1128   GSimpleAsyncResult *simple;
1129   gssize nread;
1130
1131   simple = G_SIMPLE_ASYNC_RESULT (result);
1132   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_real_fill_async);
1133
1134   nread = g_simple_async_result_get_op_res_gssize (simple);
1135   return nread;
1136 }
1137
1138 typedef struct
1139 {
1140   gssize bytes_skipped;
1141   gssize count;
1142 } SkipAsyncData;
1143
1144 static void
1145 free_skip_async_data (gpointer _data)
1146 {
1147   SkipAsyncData *data = _data;
1148   g_slice_free (SkipAsyncData, data);
1149 }
1150
1151 static void
1152 large_skip_callback (GObject      *source_object,
1153                      GAsyncResult *result,
1154                      gpointer      user_data)
1155 {
1156   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1157   SkipAsyncData *data;
1158   GError *error;
1159   gssize nread;
1160
1161   data = g_simple_async_result_get_op_res_gpointer (simple);
1162
1163   error = NULL;
1164   nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1165                                       result, &error);
1166
1167   /* Only report the error if we've not already read some data */
1168   if (nread < 0 && data->bytes_skipped == 0)
1169     g_simple_async_result_take_error (simple, error);
1170   else if (error)
1171     g_error_free (error);
1172
1173   if (nread > 0)
1174     data->bytes_skipped += nread;
1175
1176   /* Complete immediately, not in idle, since we're already
1177    * in a mainloop callout
1178    */
1179   g_simple_async_result_complete (simple);
1180   g_object_unref (simple);
1181 }
1182
1183 static void
1184 skip_fill_buffer_callback (GObject      *source_object,
1185                            GAsyncResult *result,
1186                            gpointer      user_data)
1187 {
1188   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (user_data);
1189   GBufferedInputStream *bstream;
1190   GBufferedInputStreamPrivate *priv;
1191   SkipAsyncData *data;
1192   GError *error;
1193   gssize nread;
1194   gsize available;
1195
1196   bstream = G_BUFFERED_INPUT_STREAM (source_object);
1197   priv = bstream->priv;
1198
1199   data = g_simple_async_result_get_op_res_gpointer (simple);
1200
1201   error = NULL;
1202   nread = g_buffered_input_stream_fill_finish (bstream,
1203                                                result, &error);
1204
1205   if (nread < 0 && data->bytes_skipped == 0)
1206     g_simple_async_result_take_error (simple, error);
1207   else if (error)
1208     g_error_free (error);
1209
1210   if (nread > 0)
1211     {
1212       available = priv->end - priv->pos;
1213       data->count = MIN (data->count, available);
1214
1215       data->bytes_skipped += data->count;
1216       priv->pos += data->count;
1217     }
1218
1219   /* Complete immediately, not in idle, since we're already
1220    * in a mainloop callout
1221    */
1222   g_simple_async_result_complete (simple);
1223   g_object_unref (simple);
1224 }
1225
1226 static void
1227 g_buffered_input_stream_skip_async (GInputStream        *stream,
1228                                     gsize                count,
1229                                     int                  io_priority,
1230                                     GCancellable        *cancellable,
1231                                     GAsyncReadyCallback  callback,
1232                                     gpointer             user_data)
1233 {
1234   GBufferedInputStream *bstream;
1235   GBufferedInputStreamPrivate *priv;
1236   GBufferedInputStreamClass *class;
1237   GInputStream *base_stream;
1238   gsize available;
1239   GSimpleAsyncResult *simple;
1240   SkipAsyncData *data;
1241
1242   bstream = G_BUFFERED_INPUT_STREAM (stream);
1243   priv = bstream->priv;
1244
1245   data = g_slice_new (SkipAsyncData);
1246   data->bytes_skipped = 0;
1247   simple = g_simple_async_result_new (G_OBJECT (stream),
1248                                       callback, user_data,
1249                                       g_buffered_input_stream_skip_async);
1250   g_simple_async_result_set_op_res_gpointer (simple, data, free_skip_async_data);
1251
1252   available = priv->end - priv->pos;
1253
1254   if (count <= available)
1255     {
1256       priv->pos += count;
1257       data->bytes_skipped = count;
1258
1259       g_simple_async_result_complete_in_idle (simple);
1260       g_object_unref (simple);
1261       return;
1262     }
1263
1264   /* Full request not available, skip all currently available
1265    * and request refill for more
1266    */
1267
1268   priv->pos = 0;
1269   priv->end = 0;
1270
1271   count -= available;
1272
1273   data->bytes_skipped = available;
1274   data->count = count;
1275
1276   if (count > priv->len)
1277     {
1278       /* Large request, shortcut buffer */
1279
1280       base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1281
1282       g_input_stream_skip_async (base_stream,
1283                                  count,
1284                                  io_priority, cancellable,
1285                                  large_skip_callback,
1286                                  simple);
1287     }
1288   else
1289     {
1290       class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1291       class->fill_async (bstream, priv->len, io_priority, cancellable,
1292                          skip_fill_buffer_callback, simple);
1293     }
1294 }
1295
1296 static gssize
1297 g_buffered_input_stream_skip_finish (GInputStream   *stream,
1298                                      GAsyncResult   *result,
1299                                      GError        **error)
1300 {
1301   GSimpleAsyncResult *simple;
1302   SkipAsyncData *data;
1303
1304   simple = G_SIMPLE_ASYNC_RESULT (result);
1305
1306   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_buffered_input_stream_skip_async);
1307
1308   data = g_simple_async_result_get_op_res_gpointer (simple);
1309
1310   return data->bytes_skipped;
1311 }