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