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