gio/ginputstream.c

   1 /* GIO - GLib Input, Output and Streaming Library
   2  *
   3  * Copyright (C) 2006-2007 Red Hat, Inc.
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Lesser General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Lesser General
  16  * Public License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  *
  20  * Author: Alexander Larsson <alexl@redhat.com>
  21  */
  22
  23 #include "config.h"
  24 #include <glib.h>
  25 #include "glibintl.h"
  26
  27 #include "ginputstream.h"
  28 #include "gseekable.h"
  29 #include "gcancellable.h"
  30 #include "gasyncresult.h"
  31 #include "gsimpleasyncresult.h"
  32 #include "gioerror.h"
  33 #include "gpollableinputstream.h"
  34
  35 /**
  36  * SECTION:ginputstream
  37  * @short_description: Base class for implementing streaming input
  38  * @include: gio/gio.h
  39  *
  40  * #GInputStream has functions to read from a stream (g_input_stream_read()),
  41  * to close a stream (g_input_stream_close()) and to skip some content
  42  * (g_input_stream_skip()).
  43  *
  44  * To copy the content of an input stream to an output stream without
  45  * manually handling the reads and writes, use g_output_stream_splice().
  46  *
  47  * All of these functions have async variants too.
  48  **/
  49
  50 G_DEFINE_ABSTRACT_TYPE (GInputStream, g_input_stream, G_TYPE_OBJECT);
  51
  52 struct _GInputStreamPrivate {
  53   guint closed : 1;
  54   guint pending : 1;
  55   GAsyncReadyCallback outstanding_callback;
  56 };
  57
  58 static gssize   g_input_stream_real_skip         (GInputStream         *stream,
  59                                                   gsize                 count,
  60                                                   GCancellable         *cancellable,
  61                                                   GError              **error);
  62 static void     g_input_stream_real_read_async   (GInputStream         *stream,
  63                                                   void                 *buffer,
  64                                                   gsize                 count,
  65                                                   int                   io_priority,
  66                                                   GCancellable         *cancellable,
  67                                                   GAsyncReadyCallback   callback,
  68                                                   gpointer              user_data);
  69 static gssize   g_input_stream_real_read_finish  (GInputStream         *stream,
  70                                                   GAsyncResult         *result,
  71                                                   GError              **error);
  72 static void     g_input_stream_real_skip_async   (GInputStream         *stream,
  73                                                   gsize                 count,
  74                                                   int                   io_priority,
  75                                                   GCancellable         *cancellable,
  76                                                   GAsyncReadyCallback   callback,
  77                                                   gpointer              data);
  78 static gssize   g_input_stream_real_skip_finish  (GInputStream         *stream,
  79                                                   GAsyncResult         *result,
  80                                                   GError              **error);
  81 static void     g_input_stream_real_close_async  (GInputStream         *stream,
  82                                                   int                   io_priority,
  83                                                   GCancellable         *cancellable,
  84                                                   GAsyncReadyCallback   callback,
  85                                                   gpointer              data);
  86 static gboolean g_input_stream_real_close_finish (GInputStream         *stream,
  87                                                   GAsyncResult         *result,
  88                                                   GError              **error);
  89
  90 static void
  91 g_input_stream_finalize (GObject *object)
  92 {
  93   G_OBJECT_CLASS (g_input_stream_parent_class)->finalize (object);
  94 }
  95
  96 static void
  97 g_input_stream_dispose (GObject *object)
  98 {
  99   GInputStream *stream;
 100
 101   stream = G_INPUT_STREAM (object);
 102
 103   if (!stream->priv->closed)
 104     g_input_stream_close (stream, NULL, NULL);
 105
 106   G_OBJECT_CLASS (g_input_stream_parent_class)->dispose (object);
 107 }
 108
 109
 110 static void
 111 g_input_stream_class_init (GInputStreamClass *klass)
 112 {
 113   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 114
 115   g_type_class_add_private (klass, sizeof (GInputStreamPrivate));
 116
 117   gobject_class->finalize = g_input_stream_finalize;
 118   gobject_class->dispose = g_input_stream_dispose;
 119
 120   klass->skip = g_input_stream_real_skip;
 121   klass->read_async = g_input_stream_real_read_async;
 122   klass->read_finish = g_input_stream_real_read_finish;
 123   klass->skip_async = g_input_stream_real_skip_async;
 124   klass->skip_finish = g_input_stream_real_skip_finish;
 125   klass->close_async = g_input_stream_real_close_async;
 126   klass->close_finish = g_input_stream_real_close_finish;
 127 }
 128
 129 static void
 130 g_input_stream_init (GInputStream *stream)
 131 {
 132   stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
 133                                               G_TYPE_INPUT_STREAM,
 134                                               GInputStreamPrivate);
 135 }
 136
 137 /**
 138  * g_input_stream_read:
 139  * @stream: a #GInputStream.
 140  * @buffer: a buffer to read data into (which should be at least count bytes long).
 141  * @count: the number of bytes that will be read from the stream
 142  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 143  * @error: location to store the error occurring, or %NULL to ignore
 144  *
 145  * Tries to read @count bytes from the stream into the buffer starting at
 146  * @buffer. Will block during this read.
 147  *
 148  * If count is zero returns zero and does nothing. A value of @count
 149  * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 150  *
 151  * On success, the number of bytes read into the buffer is returned.
 152  * It is not an error if this is not the same as the requested size, as it
 153  * can happen e.g. near the end of a file. Zero is returned on end of file
 154  * (or if @count is zero),  but never otherwise.
 155  *
 156  * If @cancellable is not %NULL, then the operation can be cancelled by
 157  * triggering the cancellable object from another thread. If the operation
 158  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
 159  * operation was partially finished when the operation was cancelled the
 160  * partial result will be returned, without an error.
 161  *
 162  * On error -1 is returned and @error is set accordingly.
 163  *
 164  * Return value: Number of bytes read, or -1 on error, or 0 on end of file.
 165  **/
 166 gssize
 167 g_input_stream_read  (GInputStream  *stream,
 168                       void          *buffer,
 169                       gsize          count,
 170                       GCancellable  *cancellable,
 171                       GError       **error)
 172 {
 173   GInputStreamClass *class;
 174   gssize res;
 175
 176   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 177   g_return_val_if_fail (buffer != NULL, 0);
 178
 179   if (count == 0)
 180     return 0;
 181
 182   if (((gssize) count) < 0)
 183     {
 184       g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 185                    _("Too large count value passed to %s"), G_STRFUNC);
 186       return -1;
 187     }
 188
 189   class = G_INPUT_STREAM_GET_CLASS (stream);
 190
 191   if (class->read_fn == NULL)
 192     {
 193       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
 194                            _("Input stream doesn't implement read"));
 195       return -1;
 196     }
 197
 198   if (!g_input_stream_set_pending (stream, error))
 199     return -1;
 200
 201   if (cancellable)
 202     g_cancellable_push_current (cancellable);
 203
 204   res = class->read_fn (stream, buffer, count, cancellable, error);
 205
 206   if (cancellable)
 207     g_cancellable_pop_current (cancellable);
 208
 209   g_input_stream_clear_pending (stream);
 210
 211   return res;
 212 }
 213
 214 /**
 215  * g_input_stream_read_all:
 216  * @stream: a #GInputStream.
 217  * @buffer: a buffer to read data into (which should be at least count bytes long).
 218  * @count: the number of bytes that will be read from the stream
 219  * @bytes_read: (out): location to store the number of bytes that was read from the stream
 220  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 221  * @error: location to store the error occurring, or %NULL to ignore
 222  *
 223  * Tries to read @count bytes from the stream into the buffer starting at
 224  * @buffer. Will block during this read.
 225  *
 226  * This function is similar to g_input_stream_read(), except it tries to
 227  * read as many bytes as requested, only stopping on an error or end of stream.
 228  *
 229  * On a successful read of @count bytes, or if we reached the end of the
 230  * stream,  %TRUE is returned, and @bytes_read is set to the number of bytes
 231  * read into @buffer.
 232  *
 233  * If there is an error during the operation %FALSE is returned and @error
 234  * is set to indicate the error status, @bytes_read is updated to contain
 235  * the number of bytes read into @buffer before the error occurred.
 236  *
 237  * Return value: %TRUE on success, %FALSE if there was an error
 238  **/
 239 gboolean
 240 g_input_stream_read_all (GInputStream  *stream,
 241                          void          *buffer,
 242                          gsize          count,
 243                          gsize         *bytes_read,
 244                          GCancellable  *cancellable,
 245                          GError       **error)
 246 {
 247   gsize _bytes_read;
 248   gssize res;
 249
 250   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
 251   g_return_val_if_fail (buffer != NULL, FALSE);
 252
 253   _bytes_read = 0;
 254   while (_bytes_read < count)
 255     {
 256       res = g_input_stream_read (stream, (char *)buffer + _bytes_read, count - _bytes_read,
 257                                  cancellable, error);
 258       if (res == -1)
 259         {
 260           if (bytes_read)
 261             *bytes_read = _bytes_read;
 262           return FALSE;
 263         }
 264
 265       if (res == 0)
 266         break;
 267
 268       _bytes_read += res;
 269     }
 270
 271   if (bytes_read)
 272     *bytes_read = _bytes_read;
 273   return TRUE;
 274 }
 275
 276 /**
 277  * g_input_stream_read_bytes:
 278  * @stream: a #GInputStream.
 279  * @count: maximum number of bytes that will be read from the stream. Common
 280  * values include 4096 and 8192.
 281  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 282  * @error: location to store the error occurring, or %NULL to ignore
 283  *
 284  * Like g_input_stream_read(), this tries to read @count bytes from
 285  * the stream in a blocking fashion. However, rather than reading into
 286  * a user-supplied buffer, this will create a new #GBytes containing
 287  * the data that was read. This may be easier to use from language
 288  * bindings.
 289  *
 290  * If count is zero, returns a zero-length #GBytes and does nothing. A
 291  * value of @count larger than %G_MAXSSIZE will cause a
 292  * %G_IO_ERROR_INVALID_ARGUMENT error.
 293  *
 294  * On success, a new #GBytes is returned. It is not an error if the
 295  * size of this object is not the same as the requested size, as it
 296  * can happen e.g. near the end of a file. A zero-length #GBytes is
 297  * returned on end of file (or if @count is zero), but never
 298  * otherwise.
 299  *
 300  * If @cancellable is not %NULL, then the operation can be cancelled by
 301  * triggering the cancellable object from another thread. If the operation
 302  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
 303  * operation was partially finished when the operation was cancelled the
 304  * partial result will be returned, without an error.
 305  *
 306  * On error %NULL is returned and @error is set accordingly.
 307  *
 308  * Return value: a new #GBytes, or %NULL on error
 309  **/
 310 GBytes *
 311 g_input_stream_read_bytes (GInputStream  *stream,
 312                            gsize          count,
 313                            GCancellable  *cancellable,
 314                            GError       **error)
 315 {
 316   guchar *buf;
 317   gssize nread;
 318
 319   buf = g_malloc (count);
 320   nread = g_input_stream_read (stream, buf, count, cancellable, error);
 321   if (nread == -1)
 322     {
 323       g_free (buf);
 324       return NULL;
 325     }
 326   else if (nread == 0)
 327     {
 328       g_free (buf);
 329       return g_bytes_new_static ("", 0);
 330     }
 331   else
 332     return g_bytes_new_take (buf, nread);
 333 }
 334
 335 /**
 336  * g_input_stream_skip:
 337  * @stream: a #GInputStream.
 338  * @count: the number of bytes that will be skipped from the stream
 339  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 340  * @error: location to store the error occurring, or %NULL to ignore
 341  *
 342  * Tries to skip @count bytes from the stream. Will block during the operation.
 343  *
 344  * This is identical to g_input_stream_read(), from a behaviour standpoint,
 345  * but the bytes that are skipped are not returned to the user. Some
 346  * streams have an implementation that is more efficient than reading the data.
 347  *
 348  * This function is optional for inherited classes, as the default implementation
 349  * emulates it using read.
 350  *
 351  * If @cancellable is not %NULL, then the operation can be cancelled by
 352  * triggering the cancellable object from another thread. If the operation
 353  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
 354  * operation was partially finished when the operation was cancelled the
 355  * partial result will be returned, without an error.
 356  *
 357  * Return value: Number of bytes skipped, or -1 on error
 358  **/
 359 gssize
 360 g_input_stream_skip (GInputStream  *stream,
 361                      gsize          count,
 362                      GCancellable  *cancellable,
 363                      GError       **error)
 364 {
 365   GInputStreamClass *class;
 366   gssize res;
 367
 368   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 369
 370   if (count == 0)
 371     return 0;
 372
 373   if (((gssize) count) < 0)
 374     {
 375       g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 376                    _("Too large count value passed to %s"), G_STRFUNC);
 377       return -1;
 378     }
 379
 380   class = G_INPUT_STREAM_GET_CLASS (stream);
 381
 382   if (!g_input_stream_set_pending (stream, error))
 383     return -1;
 384
 385   if (cancellable)
 386     g_cancellable_push_current (cancellable);
 387
 388   res = class->skip (stream, count, cancellable, error);
 389
 390   if (cancellable)
 391     g_cancellable_pop_current (cancellable);
 392
 393   g_input_stream_clear_pending (stream);
 394
 395   return res;
 396 }
 397
 398 static gssize
 399 g_input_stream_real_skip (GInputStream  *stream,
 400                           gsize          count,
 401                           GCancellable  *cancellable,
 402                           GError       **error)
 403 {
 404   GInputStreamClass *class;
 405   gssize ret, read_bytes;
 406   char buffer[8192];
 407   GError *my_error;
 408
 409   if (G_IS_SEEKABLE (stream) && g_seekable_can_seek (G_SEEKABLE (stream)))
 410     {
 411       if (g_seekable_seek (G_SEEKABLE (stream),
 412                            count,
 413                            G_SEEK_CUR,
 414                            cancellable,
 415                            NULL))
 416         return count;
 417     }
 418
 419   /* If not seekable, or seek failed, fall back to reading data: */
 420
 421   class = G_INPUT_STREAM_GET_CLASS (stream);
 422
 423   read_bytes = 0;
 424   while (1)
 425     {
 426       my_error = NULL;
 427
 428       ret = class->read_fn (stream, buffer, MIN (sizeof (buffer), count),
 429                             cancellable, &my_error);
 430       if (ret == -1)
 431         {
 432           if (read_bytes > 0 &&
 433               my_error->domain == G_IO_ERROR &&
 434               my_error->code == G_IO_ERROR_CANCELLED)
 435             {
 436               g_error_free (my_error);
 437               return read_bytes;
 438             }
 439
 440           g_propagate_error (error, my_error);
 441           return -1;
 442         }
 443
 444       count -= ret;
 445       read_bytes += ret;
 446
 447       if (ret == 0 || count == 0)
 448         return read_bytes;
 449     }
 450 }
 451
 452 /**
 453  * g_input_stream_close:
 454  * @stream: A #GInputStream.
 455  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 456  * @error: location to store the error occurring, or %NULL to ignore
 457  *
 458  * Closes the stream, releasing resources related to it.
 459  *
 460  * Once the stream is closed, all other operations will return %G_IO_ERROR_CLOSED.
 461  * Closing a stream multiple times will not return an error.
 462  *
 463  * Streams will be automatically closed when the last reference
 464  * is dropped, but you might want to call this function to make sure
 465  * resources are released as early as possible.
 466  *
 467  * Some streams might keep the backing store of the stream (e.g. a file descriptor)
 468  * open after the stream is closed. See the documentation for the individual
 469  * stream for details.
 470  *
 471  * On failure the first error that happened will be reported, but the close
 472  * operation will finish as much as possible. A stream that failed to
 473  * close will still return %G_IO_ERROR_CLOSED for all operations. Still, it
 474  * is important to check and report the error to the user.
 475  *
 476  * If @cancellable is not %NULL, then the operation can be cancelled by
 477  * triggering the cancellable object from another thread. If the operation
 478  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 479  * Cancelling a close will still leave the stream closed, but some streams
 480  * can use a faster close that doesn't block to e.g. check errors.
 481  *
 482  * Return value: %TRUE on success, %FALSE on failure
 483  **/
 484 gboolean
 485 g_input_stream_close (GInputStream  *stream,
 486                       GCancellable  *cancellable,
 487                       GError       **error)
 488 {
 489   GInputStreamClass *class;
 490   gboolean res;
 491
 492   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
 493
 494   class = G_INPUT_STREAM_GET_CLASS (stream);
 495
 496   if (stream->priv->closed)
 497     return TRUE;
 498
 499   res = TRUE;
 500
 501   if (!g_input_stream_set_pending (stream, error))
 502     return FALSE;
 503
 504   if (cancellable)
 505     g_cancellable_push_current (cancellable);
 506
 507   if (class->close_fn)
 508     res = class->close_fn (stream, cancellable, error);
 509
 510   if (cancellable)
 511     g_cancellable_pop_current (cancellable);
 512
 513   g_input_stream_clear_pending (stream);
 514
 515   stream->priv->closed = TRUE;
 516
 517   return res;
 518 }
 519
 520 static void
 521 async_ready_callback_wrapper (GObject      *source_object,
 522                               GAsyncResult *res,
 523                               gpointer      user_data)
 524 {
 525   GInputStream *stream = G_INPUT_STREAM (source_object);
 526
 527   g_input_stream_clear_pending (stream);
 528   if (stream->priv->outstanding_callback)
 529     (*stream->priv->outstanding_callback) (source_object, res, user_data);
 530   g_object_unref (stream);
 531 }
 532
 533 static void
 534 async_ready_close_callback_wrapper (GObject      *source_object,
 535                                     GAsyncResult *res,
 536                                     gpointer      user_data)
 537 {
 538   GInputStream *stream = G_INPUT_STREAM (source_object);
 539
 540   g_input_stream_clear_pending (stream);
 541   stream->priv->closed = TRUE;
 542   if (stream->priv->outstanding_callback)
 543     (*stream->priv->outstanding_callback) (source_object, res, user_data);
 544   g_object_unref (stream);
 545 }
 546
 547 /**
 548  * g_input_stream_read_async:
 549  * @stream: A #GInputStream.
 550  * @buffer: a buffer to read data into (which should be at least count bytes long).
 551  * @count: the number of bytes that will be read from the stream
 552  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 553  * of the request.
 554  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 555  * @callback: (scope async): callback to call when the request is satisfied
 556  * @user_data: (closure): the data to pass to callback function
 557  *
 558  * Request an asynchronous read of @count bytes from the stream into the buffer
 559  * starting at @buffer. When the operation is finished @callback will be called.
 560  * You can then call g_input_stream_read_finish() to get the result of the
 561  * operation.
 562  *
 563  * During an async request no other sync and async calls are allowed on @stream, and will
 564  * result in %G_IO_ERROR_PENDING errors.
 565  *
 566  * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 567  *
 568  * On success, the number of bytes read into the buffer will be passed to the
 569  * callback. It is not an error if this is not the same as the requested size, as it
 570  * can happen e.g. near the end of a file, but generally we try to read
 571  * as many bytes as requested. Zero is returned on end of file
 572  * (or if @count is zero),  but never otherwise.
 573  *
 574  * Any outstanding i/o request with higher priority (lower numerical value) will
 575  * be executed before an outstanding request with lower priority. Default
 576  * priority is %G_PRIORITY_DEFAULT.
 577  *
 578  * The asyncronous methods have a default fallback that uses threads to implement
 579  * asynchronicity, so they are optional for inheriting classes. However, if you
 580  * override one you must override all.
 581  **/
 582 void
 583 g_input_stream_read_async (GInputStream        *stream,
 584                            void                *buffer,
 585                            gsize                count,
 586                            int                  io_priority,
 587                            GCancellable        *cancellable,
 588                            GAsyncReadyCallback  callback,
 589                            gpointer             user_data)
 590 {
 591   GInputStreamClass *class;
 592   GSimpleAsyncResult *simple;
 593   GError *error = NULL;
 594
 595   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 596   g_return_if_fail (buffer != NULL);
 597
 598   if (count == 0)
 599     {
 600       simple = g_simple_async_result_new (G_OBJECT (stream),
 601                                           callback,
 602                                           user_data,
 603                                           g_input_stream_read_async);
 604       g_simple_async_result_complete_in_idle (simple);
 605       g_object_unref (simple);
 606       return;
 607     }
 608
 609   if (((gssize) count) < 0)
 610     {
 611       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 612                                            callback,
 613                                            user_data,
 614                                            G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 615                                            _("Too large count value passed to %s"),
 616                                            G_STRFUNC);
 617       return;
 618     }
 619
 620   if (!g_input_stream_set_pending (stream, &error))
 621     {
 622       g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
 623                                             callback,
 624                                             user_data,
 625                                             error);
 626       return;
 627     }
 628
 629   class = G_INPUT_STREAM_GET_CLASS (stream);
 630   stream->priv->outstanding_callback = callback;
 631   g_object_ref (stream);
 632   class->read_async (stream, buffer, count, io_priority, cancellable,
 633                      async_ready_callback_wrapper, user_data);
 634 }
 635
 636 /**
 637  * g_input_stream_read_finish:
 638  * @stream: a #GInputStream.
 639  * @result: a #GAsyncResult.
 640  * @error: a #GError location to store the error occurring, or %NULL to
 641  * ignore.
 642  *
 643  * Finishes an asynchronous stream read operation.
 644  *
 645  * Returns: number of bytes read in, or -1 on error, or 0 on end of file.
 646  **/
 647 gssize
 648 g_input_stream_read_finish (GInputStream  *stream,
 649                             GAsyncResult  *result,
 650                             GError       **error)
 651 {
 652   GInputStreamClass *class;
 653
 654   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 655   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 656
 657   if (g_async_result_legacy_propagate_error (result, error))
 658     return -1;
 659   else if (g_async_result_is_tagged (result, g_input_stream_read_async))
 660     {
 661       /* Special case read of 0 bytes */
 662       return 0;
 663     }
 664
 665   class = G_INPUT_STREAM_GET_CLASS (stream);
 666   return class->read_finish (stream, result, error);
 667 }
 668
 669 static void
 670 read_bytes_callback (GObject      *stream,
 671                      GAsyncResult *result,
 672                      gpointer      user_data)
 673 {
 674   GSimpleAsyncResult *simple = user_data;
 675   guchar *buf = g_simple_async_result_get_op_res_gpointer (simple);
 676   GError *error = NULL;
 677   gssize nread;
 678   GBytes *bytes = NULL;
 679
 680   nread = g_input_stream_read_finish (G_INPUT_STREAM (stream),
 681                                       result, &error);
 682   if (nread == -1)
 683     {
 684       g_free (buf);
 685       g_simple_async_result_take_error (simple, error);
 686     }
 687   else if (nread == 0)
 688     {
 689       g_free (buf);
 690       bytes = g_bytes_new_static ("", 0);
 691     }
 692   else
 693     bytes = g_bytes_new_take (buf, nread);
 694
 695   if (bytes)
 696     {
 697       g_simple_async_result_set_op_res_gpointer (simple, bytes,
 698                                                  (GDestroyNotify)g_bytes_unref);
 699     }
 700   g_simple_async_result_complete (simple);
 701   g_object_unref (simple);
 702 }
 703
 704 /**
 705  * g_input_stream_read_bytes_async:
 706  * @stream: A #GInputStream.
 707  * @count: the number of bytes that will be read from the stream
 708  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 709  *   of the request.
 710  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 711  * @callback: (scope async): callback to call when the request is satisfied
 712  * @user_data: (closure): the data to pass to callback function
 713  *
 714  * Request an asynchronous read of @count bytes from the stream into a
 715  * new #GBytes. When the operation is finished @callback will be
 716  * called. You can then call g_input_stream_read_bytes_finish() to get the
 717  * result of the operation.
 718  *
 719  * During an async request no other sync and async calls are allowed
 720  * on @stream, and will result in %G_IO_ERROR_PENDING errors.
 721  *
 722  * A value of @count larger than %G_MAXSSIZE will cause a
 723  * %G_IO_ERROR_INVALID_ARGUMENT error.
 724  *
 725  * On success, the new #GBytes will be passed to the callback. It is
 726  * not an error if this is smaller than the requested size, as it can
 727  * happen e.g. near the end of a file, but generally we try to read as
 728  * many bytes as requested. Zero is returned on end of file (or if
 729  * @count is zero), but never otherwise.
 730  *
 731  * Any outstanding I/O request with higher priority (lower numerical
 732  * value) will be executed before an outstanding request with lower
 733  * priority. Default priority is %G_PRIORITY_DEFAULT.
 734  **/
 735 void
 736 g_input_stream_read_bytes_async (GInputStream          *stream,
 737                                  gsize                  count,
 738                                  int                    io_priority,
 739                                  GCancellable          *cancellable,
 740                                  GAsyncReadyCallback    callback,
 741                                  gpointer               user_data)
 742 {
 743   GSimpleAsyncResult *simple;
 744   guchar *buf;
 745
 746   simple = g_simple_async_result_new (G_OBJECT (stream),
 747                                       callback, user_data,
 748                                       g_input_stream_read_bytes_async);
 749   buf = g_malloc (count);
 750   g_simple_async_result_set_op_res_gpointer (simple, buf, NULL);
 751
 752   g_input_stream_read_async (stream, buf, count,
 753                              io_priority, cancellable,
 754                              read_bytes_callback, simple);
 755 }
 756
 757 /**
 758  * g_input_stream_read_bytes_finish:
 759  * @stream: a #GInputStream.
 760  * @result: a #GAsyncResult.
 761  * @error: a #GError location to store the error occurring, or %NULL to
 762  *   ignore.
 763  *
 764  * Finishes an asynchronous stream read-into-#GBytes operation.
 765  *
 766  * Returns: the newly-allocated #GBytes, or %NULL on error
 767  **/
 768 GBytes *
 769 g_input_stream_read_bytes_finish (GInputStream  *stream,
 770                                   GAsyncResult  *result,
 771                                   GError       **error)
 772 {
 773   GSimpleAsyncResult *simple;
 774
 775   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), NULL);
 776   g_return_val_if_fail (g_simple_async_result_is_valid (result, G_OBJECT (stream), g_input_stream_read_bytes_async), NULL);
 777
 778   simple = G_SIMPLE_ASYNC_RESULT (result);
 779   if (g_simple_async_result_propagate_error (simple, error))
 780     return NULL;
 781   return g_bytes_ref (g_simple_async_result_get_op_res_gpointer (simple));
 782 }
 783
 784 /**
 785  * g_input_stream_skip_async:
 786  * @stream: A #GInputStream.
 787  * @count: the number of bytes that will be skipped from the stream
 788  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 789  * of the request.
 790  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 791  * @callback: (scope async): callback to call when the request is satisfied
 792  * @user_data: (closure): the data to pass to callback function
 793  *
 794  * Request an asynchronous skip of @count bytes from the stream.
 795  * When the operation is finished @callback will be called.
 796  * You can then call g_input_stream_skip_finish() to get the result
 797  * of the operation.
 798  *
 799  * During an async request no other sync and async calls are allowed,
 800  * and will result in %G_IO_ERROR_PENDING errors.
 801  *
 802  * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 803  *
 804  * On success, the number of bytes skipped will be passed to the callback.
 805  * It is not an error if this is not the same as the requested size, as it
 806  * can happen e.g. near the end of a file, but generally we try to skip
 807  * as many bytes as requested. Zero is returned on end of file
 808  * (or if @count is zero), but never otherwise.
 809  *
 810  * Any outstanding i/o request with higher priority (lower numerical value)
 811  * will be executed before an outstanding request with lower priority.
 812  * Default priority is %G_PRIORITY_DEFAULT.
 813  *
 814  * The asynchronous methods have a default fallback that uses threads to
 815  * implement asynchronicity, so they are optional for inheriting classes.
 816  * However, if you override one, you must override all.
 817  **/
 818 void
 819 g_input_stream_skip_async (GInputStream        *stream,
 820                            gsize                count,
 821                            int                  io_priority,
 822                            GCancellable        *cancellable,
 823                            GAsyncReadyCallback  callback,
 824                            gpointer             user_data)
 825 {
 826   GInputStreamClass *class;
 827   GSimpleAsyncResult *simple;
 828   GError *error = NULL;
 829
 830   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 831
 832   if (count == 0)
 833     {
 834       simple = g_simple_async_result_new (G_OBJECT (stream),
 835                                           callback,
 836                                           user_data,
 837                                           g_input_stream_skip_async);
 838
 839       g_simple_async_result_complete_in_idle (simple);
 840       g_object_unref (simple);
 841       return;
 842     }
 843
 844   if (((gssize) count) < 0)
 845     {
 846       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 847                                            callback,
 848                                            user_data,
 849                                            G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 850                                            _("Too large count value passed to %s"),
 851                                            G_STRFUNC);
 852       return;
 853     }
 854
 855   if (!g_input_stream_set_pending (stream, &error))
 856     {
 857       g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
 858                                             callback,
 859                                             user_data,
 860                                             error);
 861       return;
 862     }
 863
 864   class = G_INPUT_STREAM_GET_CLASS (stream);
 865   stream->priv->outstanding_callback = callback;
 866   g_object_ref (stream);
 867   class->skip_async (stream, count, io_priority, cancellable,
 868                      async_ready_callback_wrapper, user_data);
 869 }
 870
 871 /**
 872  * g_input_stream_skip_finish:
 873  * @stream: a #GInputStream.
 874  * @result: a #GAsyncResult.
 875  * @error: a #GError location to store the error occurring, or %NULL to
 876  * ignore.
 877  *
 878  * Finishes a stream skip operation.
 879  *
 880  * Returns: the size of the bytes skipped, or %-1 on error.
 881  **/
 882 gssize
 883 g_input_stream_skip_finish (GInputStream  *stream,
 884                             GAsyncResult  *result,
 885                             GError       **error)
 886 {
 887   GInputStreamClass *class;
 888
 889   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 890   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 891
 892   if (g_async_result_legacy_propagate_error (result, error))
 893     return -1;
 894   else if (g_async_result_is_tagged (result, g_input_stream_skip_async))
 895     {
 896       /* Special case skip of 0 bytes */
 897       return 0;
 898     }
 899
 900   class = G_INPUT_STREAM_GET_CLASS (stream);
 901   return class->skip_finish (stream, result, error);
 902 }
 903
 904 /**
 905  * g_input_stream_close_async:
 906  * @stream: A #GInputStream.
 907  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 908  * of the request.
 909  * @cancellable: (allow-none): optional cancellable object
 910  * @callback: (scope async): callback to call when the request is satisfied
 911  * @user_data: (closure): the data to pass to callback function
 912  *
 913  * Requests an asynchronous closes of the stream, releasing resources related to it.
 914  * When the operation is finished @callback will be called.
 915  * You can then call g_input_stream_close_finish() to get the result of the
 916  * operation.
 917  *
 918  * For behaviour details see g_input_stream_close().
 919  *
 920  * The asyncronous methods have a default fallback that uses threads to implement
 921  * asynchronicity, so they are optional for inheriting classes. However, if you
 922  * override one you must override all.
 923  **/
 924 void
 925 g_input_stream_close_async (GInputStream        *stream,
 926                             int                  io_priority,
 927                             GCancellable        *cancellable,
 928                             GAsyncReadyCallback  callback,
 929                             gpointer             user_data)
 930 {
 931   GInputStreamClass *class;
 932   GSimpleAsyncResult *simple;
 933   GError *error = NULL;
 934
 935   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 936
 937   if (stream->priv->closed)
 938     {
 939       simple = g_simple_async_result_new (G_OBJECT (stream),
 940                                           callback,
 941                                           user_data,
 942                                           g_input_stream_close_async);
 943
 944       g_simple_async_result_complete_in_idle (simple);
 945       g_object_unref (simple);
 946       return;
 947     }
 948
 949   if (!g_input_stream_set_pending (stream, &error))
 950     {
 951       g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
 952                                             callback,
 953                                             user_data,
 954                                             error);
 955       return;
 956     }
 957
 958   class = G_INPUT_STREAM_GET_CLASS (stream);
 959   stream->priv->outstanding_callback = callback;
 960   g_object_ref (stream);
 961   class->close_async (stream, io_priority, cancellable,
 962                       async_ready_close_callback_wrapper, user_data);
 963 }
 964
 965 /**
 966  * g_input_stream_close_finish:
 967  * @stream: a #GInputStream.
 968  * @result: a #GAsyncResult.
 969  * @error: a #GError location to store the error occurring, or %NULL to
 970  * ignore.
 971  *
 972  * Finishes closing a stream asynchronously, started from g_input_stream_close_async().
 973  *
 974  * Returns: %TRUE if the stream was closed successfully.
 975  **/
 976 gboolean
 977 g_input_stream_close_finish (GInputStream  *stream,
 978                              GAsyncResult  *result,
 979                              GError       **error)
 980 {
 981   GInputStreamClass *class;
 982
 983   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
 984   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
 985
 986   if (g_async_result_legacy_propagate_error (result, error))
 987     return FALSE;
 988   else if (g_async_result_is_tagged (result, g_input_stream_close_async))
 989     {
 990       /* Special case already closed */
 991       return TRUE;
 992     }
 993
 994   class = G_INPUT_STREAM_GET_CLASS (stream);
 995   return class->close_finish (stream, result, error);
 996 }
 997
 998 /**
 999  * g_input_stream_is_closed:
1000  * @stream: input stream.
1001  *
1002  * Checks if an input stream is closed.
1003  *
1004  * Returns: %TRUE if the stream is closed.
1005  **/
1006 gboolean
1007 g_input_stream_is_closed (GInputStream *stream)
1008 {
1009   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
1010
1011   return stream->priv->closed;
1012 }
1013
1014 /**
1015  * g_input_stream_has_pending:
1016  * @stream: input stream.
1017  *
1018  * Checks if an input stream has pending actions.
1019  *
1020  * Returns: %TRUE if @stream has pending actions.
1021  **/
1022 gboolean
1023 g_input_stream_has_pending (GInputStream *stream)
1024 {
1025   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
1026
1027   return stream->priv->pending;
1028 }
1029
1030 /**
1031  * g_input_stream_set_pending:
1032  * @stream: input stream
1033  * @error: a #GError location to store the error occurring, or %NULL to
1034  * ignore.
1035  *
1036  * Sets @stream to have actions pending. If the pending flag is
1037  * already set or @stream is closed, it will return %FALSE and set
1038  * @error.
1039  *
1040  * Return value: %TRUE if pending was previously unset and is now set.
1041  **/
1042 gboolean
1043 g_input_stream_set_pending (GInputStream *stream, GError **error)
1044 {
1045   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
1046
1047   if (stream->priv->closed)
1048     {
1049       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
1050                            _("Stream is already closed"));
1051       return FALSE;
1052     }
1053
1054   if (stream->priv->pending)
1055     {
1056       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
1057                 /* Translators: This is an error you get if there is already an
1058                  * operation running against this stream when you try to start
1059                  * one */
1060                  _("Stream has outstanding operation"));
1061       return FALSE;
1062     }
1063
1064   stream->priv->pending = TRUE;
1065   return TRUE;
1066 }
1067
1068 /**
1069  * g_input_stream_clear_pending:
1070  * @stream: input stream
1071  *
1072  * Clears the pending flag on @stream.
1073  **/
1074 void
1075 g_input_stream_clear_pending (GInputStream *stream)
1076 {
1077   g_return_if_fail (G_IS_INPUT_STREAM (stream));
1078
1079   stream->priv->pending = FALSE;
1080 }
1081
1082 /********************************************
1083  *   Default implementation of async ops    *
1084  ********************************************/
1085
1086 typedef struct {
1087   void              *buffer;
1088   gsize              count_requested;
1089   gssize             count_read;
1090
1091   GCancellable      *cancellable;
1092   gint               io_priority;
1093   gboolean           need_idle;
1094 } ReadData;
1095
1096 static void
1097 free_read_data (ReadData *op)
1098 {
1099   if (op->cancellable)
1100     g_object_unref (op->cancellable);
1101   g_slice_free (ReadData, op);
1102 }
1103
1104 static void
1105 read_async_thread (GSimpleAsyncResult *res,
1106                    GObject            *object,
1107                    GCancellable       *cancellable)
1108 {
1109   ReadData *op;
1110   GInputStreamClass *class;
1111   GError *error = NULL;
1112
1113   op = g_simple_async_result_get_op_res_gpointer (res);
1114
1115   class = G_INPUT_STREAM_GET_CLASS (object);
1116
1117   op->count_read = class->read_fn (G_INPUT_STREAM (object),
1118                                    op->buffer, op->count_requested,
1119                                    cancellable, &error);
1120   if (op->count_read == -1)
1121     g_simple_async_result_take_error (res, error);
1122 }
1123
1124 static void read_async_pollable (GPollableInputStream *stream,
1125                                  GSimpleAsyncResult   *result);
1126
1127 static gboolean
1128 read_async_pollable_ready (GPollableInputStream *stream,
1129                            gpointer              user_data)
1130 {
1131   GSimpleAsyncResult *result = user_data;
1132
1133   read_async_pollable (stream, result);
1134   return FALSE;
1135 }
1136
1137 static void
1138 read_async_pollable (GPollableInputStream *stream,
1139                      GSimpleAsyncResult   *result)
1140 {
1141   GError *error = NULL;
1142   ReadData *op = g_simple_async_result_get_op_res_gpointer (result);
1143
1144   if (g_cancellable_set_error_if_cancelled (op->cancellable, &error))
1145     op->count_read = -1;
1146   else
1147     {
1148       op->count_read = G_POLLABLE_INPUT_STREAM_GET_INTERFACE (stream)->
1149         read_nonblocking (stream, op->buffer, op->count_requested, &error);
1150     }
1151
1152   if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
1153     {
1154       GSource *source;
1155
1156       g_error_free (error);
1157       op->need_idle = FALSE;
1158
1159       source = g_pollable_input_stream_create_source (stream, op->cancellable);
1160       g_source_set_callback (source,
1161                              (GSourceFunc) read_async_pollable_ready,
1162                              g_object_ref (result), g_object_unref);
1163       g_source_set_priority (source, op->io_priority);
1164       g_source_attach (source, g_main_context_get_thread_default ());
1165       g_source_unref (source);
1166       return;
1167     }
1168
1169   if (op->count_read == -1)
1170     g_simple_async_result_take_error (result, error);
1171
1172   if (op->need_idle)
1173     g_simple_async_result_complete_in_idle (result);
1174   else
1175     g_simple_async_result_complete (result);
1176 }
1177
1178 static void
1179 g_input_stream_real_read_async (GInputStream        *stream,
1180                                 void                *buffer,
1181                                 gsize                count,
1182                                 int                  io_priority,
1183                                 GCancellable        *cancellable,
1184                                 GAsyncReadyCallback  callback,
1185                                 gpointer             user_data)
1186 {
1187   GSimpleAsyncResult *res;
1188   ReadData *op;
1189
1190   op = g_slice_new0 (ReadData);
1191   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_input_stream_real_read_async);
1192   g_simple_async_result_set_op_res_gpointer (res, op, (GDestroyNotify) free_read_data);
1193   op->buffer = buffer;
1194   op->count_requested = count;
1195   op->cancellable = cancellable ? g_object_ref (cancellable) : NULL;
1196   op->io_priority = io_priority;
1197   op->need_idle = TRUE;
1198
1199   if (G_IS_POLLABLE_INPUT_STREAM (stream) &&
1200       g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (stream)))
1201     read_async_pollable (G_POLLABLE_INPUT_STREAM (stream), res);
1202   else
1203     g_simple_async_result_run_in_thread (res, read_async_thread, io_priority, cancellable);
1204   g_object_unref (res);
1205 }
1206
1207 static gssize
1208 g_input_stream_real_read_finish (GInputStream  *stream,
1209                                  GAsyncResult  *result,
1210                                  GError       **error)
1211 {
1212   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1213   ReadData *op;
1214
1215   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) ==
1216             g_input_stream_real_read_async);
1217
1218   if (g_simple_async_result_propagate_error (simple, error))
1219     return -1;
1220
1221   op = g_simple_async_result_get_op_res_gpointer (simple);
1222
1223   return op->count_read;
1224 }
1225
1226 typedef struct {
1227   gsize count_requested;
1228   gssize count_skipped;
1229 } SkipData;
1230
1231
1232 static void
1233 skip_async_thread (GSimpleAsyncResult *res,
1234                    GObject            *object,
1235                    GCancellable       *cancellable)
1236 {
1237   SkipData *op;
1238   GInputStreamClass *class;
1239   GError *error = NULL;
1240
1241   class = G_INPUT_STREAM_GET_CLASS (object);
1242   op = g_simple_async_result_get_op_res_gpointer (res);
1243   op->count_skipped = class->skip (G_INPUT_STREAM (object),
1244                                    op->count_requested,
1245                                    cancellable, &error);
1246   if (op->count_skipped == -1)
1247     g_simple_async_result_take_error (res, error);
1248 }
1249
1250 typedef struct {
1251   char buffer[8192];
1252   gsize count;
1253   gsize count_skipped;
1254   int io_prio;
1255   GCancellable *cancellable;
1256   gpointer user_data;
1257   GAsyncReadyCallback callback;
1258 } SkipFallbackAsyncData;
1259
1260 static void
1261 skip_callback_wrapper (GObject      *source_object,
1262                        GAsyncResult *res,
1263                        gpointer      user_data)
1264 {
1265   GInputStreamClass *class;
1266   SkipFallbackAsyncData *data = user_data;
1267   SkipData *op;
1268   GSimpleAsyncResult *simple;
1269   GError *error = NULL;
1270   gssize ret;
1271
1272   ret = g_input_stream_read_finish (G_INPUT_STREAM (source_object), res, &error);
1273
1274   if (ret > 0)
1275     {
1276       data->count -= ret;
1277       data->count_skipped += ret;
1278
1279       if (data->count > 0)
1280         {
1281           class = G_INPUT_STREAM_GET_CLASS (source_object);
1282           class->read_async (G_INPUT_STREAM (source_object), data->buffer, MIN (8192, data->count), data->io_prio, data->cancellable,
1283                              skip_callback_wrapper, data);
1284           return;
1285         }
1286     }
1287
1288   op = g_new0 (SkipData, 1);
1289   op->count_skipped = data->count_skipped;
1290   simple = g_simple_async_result_new (source_object,
1291                                       data->callback, data->user_data,
1292                                       g_input_stream_real_skip_async);
1293
1294   g_simple_async_result_set_op_res_gpointer (simple, op, g_free);
1295
1296   if (ret == -1)
1297     {
1298       if (data->count_skipped &&
1299           g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED))
1300         /* No error, return partial read */
1301         g_error_free (error);
1302       else
1303         g_simple_async_result_take_error (simple, error);
1304     }
1305
1306   /* Complete immediately, not in idle, since we're already in a mainloop callout */
1307   g_simple_async_result_complete (simple);
1308   g_object_unref (simple);
1309
1310   g_free (data);
1311  }
1312
1313 static void
1314 g_input_stream_real_skip_async (GInputStream        *stream,
1315                                 gsize                count,
1316                                 int                  io_priority,
1317                                 GCancellable        *cancellable,
1318                                 GAsyncReadyCallback  callback,
1319                                 gpointer             user_data)
1320 {
1321   GInputStreamClass *class;
1322   SkipData *op;
1323   SkipFallbackAsyncData *data;
1324   GSimpleAsyncResult *res;
1325
1326   class = G_INPUT_STREAM_GET_CLASS (stream);
1327
1328   if (class->read_async == g_input_stream_real_read_async)
1329     {
1330       /* Read is thread-using async fallback.
1331        * Make skip use threads too, so that we can use a possible sync skip
1332        * implementation. */
1333       op = g_new0 (SkipData, 1);
1334
1335       res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
1336                                        g_input_stream_real_skip_async);
1337
1338       g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1339
1340       op->count_requested = count;
1341
1342       g_simple_async_result_run_in_thread (res, skip_async_thread, io_priority, cancellable);
1343       g_object_unref (res);
1344     }
1345   else
1346     {
1347       /* TODO: Skip fallback uses too much memory, should do multiple read calls */
1348
1349       /* There is a custom async read function, lets use that. */
1350       data = g_new (SkipFallbackAsyncData, 1);
1351       data->count = count;
1352       data->count_skipped = 0;
1353       data->io_prio = io_priority;
1354       data->cancellable = cancellable;
1355       data->callback = callback;
1356       data->user_data = user_data;
1357       class->read_async (stream, data->buffer, MIN (8192, count), io_priority, cancellable,
1358                          skip_callback_wrapper, data);
1359     }
1360
1361 }
1362
1363 static gssize
1364 g_input_stream_real_skip_finish (GInputStream  *stream,
1365                                  GAsyncResult  *result,
1366                                  GError       **error)
1367 {
1368   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1369   SkipData *op;
1370
1371   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_skip_async);
1372
1373   if (g_simple_async_result_propagate_error (simple, error))
1374     return -1;
1375
1376   op = g_simple_async_result_get_op_res_gpointer (simple);
1377   return op->count_skipped;
1378 }
1379
1380 static void
1381 close_async_thread (GSimpleAsyncResult *res,
1382                     GObject            *object,
1383                     GCancellable       *cancellable)
1384 {
1385   GInputStreamClass *class;
1386   GError *error = NULL;
1387   gboolean result;
1388
1389   /* Auto handling of cancelation disabled, and ignore
1390      cancellation, since we want to close things anyway, although
1391      possibly in a quick-n-dirty way. At least we never want to leak
1392      open handles */
1393
1394   class = G_INPUT_STREAM_GET_CLASS (object);
1395   if (class->close_fn)
1396     {
1397       result = class->close_fn (G_INPUT_STREAM (object), cancellable, &error);
1398       if (!result)
1399         g_simple_async_result_take_error (res, error);
1400     }
1401 }
1402
1403 static void
1404 g_input_stream_real_close_async (GInputStream        *stream,
1405                                  int                  io_priority,
1406                                  GCancellable        *cancellable,
1407                                  GAsyncReadyCallback  callback,
1408                                  gpointer             user_data)
1409 {
1410   GSimpleAsyncResult *res;
1411
1412   res = g_simple_async_result_new (G_OBJECT (stream),
1413                                    callback,
1414                                    user_data,
1415                                    g_input_stream_real_close_async);
1416
1417   g_simple_async_result_set_handle_cancellation (res, FALSE);
1418
1419   g_simple_async_result_run_in_thread (res,
1420                                        close_async_thread,
1421                                        io_priority,
1422                                        cancellable);
1423   g_object_unref (res);
1424 }
1425
1426 static gboolean
1427 g_input_stream_real_close_finish (GInputStream  *stream,
1428                                   GAsyncResult  *result,
1429                                   GError       **error)
1430 {
1431   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1432
1433   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_close_async);
1434
1435   if (g_simple_async_result_propagate_error (simple, error))
1436     return FALSE;
1437
1438   return TRUE;
1439 }