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