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