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