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   GSimpleAsyncResult *simple;
 653   GInputStreamClass *class;
 654
 655   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 656   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 657
 658   if (g_async_result_legacy_propagate_error (result, error))
 659     return -1;
 660
 661   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 662     {
 663       simple = G_SIMPLE_ASYNC_RESULT (result);
 664
 665       /* Special case read of 0 bytes */
 666       if (g_simple_async_result_get_source_tag (simple) == g_input_stream_read_async)
 667         return 0;
 668     }
 669
 670   class = G_INPUT_STREAM_GET_CLASS (stream);
 671   return class->read_finish (stream, result, error);
 672 }
 673
 674 static void
 675 read_bytes_callback (GObject      *stream,
 676                      GAsyncResult *result,
 677                      gpointer      user_data)
 678 {
 679   GSimpleAsyncResult *simple = user_data;
 680   guchar *buf = g_simple_async_result_get_op_res_gpointer (simple);
 681   GError *error = NULL;
 682   gssize nread;
 683   GBytes *bytes = NULL;
 684
 685   nread = g_input_stream_read_finish (G_INPUT_STREAM (stream),
 686                                       result, &error);
 687   if (nread == -1)
 688     {
 689       g_free (buf);
 690       g_simple_async_result_take_error (simple, error);
 691     }
 692   else if (nread == 0)
 693     {
 694       g_free (buf);
 695       bytes = g_bytes_new_static ("", 0);
 696     }
 697   else
 698     bytes = g_bytes_new_take (buf, nread);
 699
 700   if (bytes)
 701     {
 702       g_simple_async_result_set_op_res_gpointer (simple, bytes,
 703                                                  (GDestroyNotify)g_bytes_unref);
 704     }
 705   g_simple_async_result_complete (simple);
 706   g_object_unref (simple);
 707 }
 708
 709 /**
 710  * g_input_stream_read_bytes_async:
 711  * @stream: A #GInputStream.
 712  * @count: the number of bytes that will be read from the stream
 713  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 714  *   of the request.
 715  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 716  * @callback: (scope async): callback to call when the request is satisfied
 717  * @user_data: (closure): the data to pass to callback function
 718  *
 719  * Request an asynchronous read of @count bytes from the stream into a
 720  * new #GBytes. When the operation is finished @callback will be
 721  * called. You can then call g_input_stream_read_bytes_finish() to get the
 722  * result of the operation.
 723  *
 724  * During an async request no other sync and async calls are allowed
 725  * on @stream, and will result in %G_IO_ERROR_PENDING errors.
 726  *
 727  * A value of @count larger than %G_MAXSSIZE will cause a
 728  * %G_IO_ERROR_INVALID_ARGUMENT error.
 729  *
 730  * On success, the new #GBytes will be passed to the callback. It is
 731  * not an error if this is smaller than the requested size, as it can
 732  * happen e.g. near the end of a file, but generally we try to read as
 733  * many bytes as requested. Zero is returned on end of file (or if
 734  * @count is zero), but never otherwise.
 735  *
 736  * Any outstanding I/O request with higher priority (lower numerical
 737  * value) will be executed before an outstanding request with lower
 738  * priority. Default priority is %G_PRIORITY_DEFAULT.
 739  **/
 740 void
 741 g_input_stream_read_bytes_async (GInputStream          *stream,
 742                                  gsize                  count,
 743                                  int                    io_priority,
 744                                  GCancellable          *cancellable,
 745                                  GAsyncReadyCallback    callback,
 746                                  gpointer               user_data)
 747 {
 748   GSimpleAsyncResult *simple;
 749   guchar *buf;
 750
 751   simple = g_simple_async_result_new (G_OBJECT (stream),
 752                                       callback, user_data,
 753                                       g_input_stream_read_bytes_async);
 754   buf = g_malloc (count);
 755   g_simple_async_result_set_op_res_gpointer (simple, buf, NULL);
 756
 757   g_input_stream_read_async (stream, buf, count,
 758                              io_priority, cancellable,
 759                              read_bytes_callback, simple);
 760 }
 761
 762 /**
 763  * g_input_stream_read_bytes_finish:
 764  * @stream: a #GInputStream.
 765  * @result: a #GAsyncResult.
 766  * @error: a #GError location to store the error occurring, or %NULL to
 767  *   ignore.
 768  *
 769  * Finishes an asynchronous stream read-into-#GBytes operation.
 770  *
 771  * Returns: the newly-allocated #GBytes, or %NULL on error
 772  **/
 773 GBytes *
 774 g_input_stream_read_bytes_finish (GInputStream  *stream,
 775                                   GAsyncResult  *result,
 776                                   GError       **error)
 777 {
 778   GSimpleAsyncResult *simple;
 779
 780   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), NULL);
 781   g_return_val_if_fail (g_simple_async_result_is_valid (result, G_OBJECT (stream), g_input_stream_read_bytes_async), NULL);
 782
 783   simple = G_SIMPLE_ASYNC_RESULT (result);
 784   if (g_simple_async_result_propagate_error (simple, error))
 785     return NULL;
 786   return g_bytes_ref (g_simple_async_result_get_op_res_gpointer (simple));
 787 }
 788
 789 /**
 790  * g_input_stream_skip_async:
 791  * @stream: A #GInputStream.
 792  * @count: the number of bytes that will be skipped from the stream
 793  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 794  * of the request.
 795  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 796  * @callback: (scope async): callback to call when the request is satisfied
 797  * @user_data: (closure): the data to pass to callback function
 798  *
 799  * Request an asynchronous skip of @count bytes from the stream.
 800  * When the operation is finished @callback will be called.
 801  * You can then call g_input_stream_skip_finish() to get the result
 802  * of the operation.
 803  *
 804  * During an async request no other sync and async calls are allowed,
 805  * and will result in %G_IO_ERROR_PENDING errors.
 806  *
 807  * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 808  *
 809  * On success, the number of bytes skipped will be passed to the callback.
 810  * It is not an error if this is not the same as the requested size, as it
 811  * can happen e.g. near the end of a file, but generally we try to skip
 812  * as many bytes as requested. Zero is returned on end of file
 813  * (or if @count is zero), but never otherwise.
 814  *
 815  * Any outstanding i/o request with higher priority (lower numerical value)
 816  * will be executed before an outstanding request with lower priority.
 817  * Default priority is %G_PRIORITY_DEFAULT.
 818  *
 819  * The asynchronous methods have a default fallback that uses threads to
 820  * implement asynchronicity, so they are optional for inheriting classes.
 821  * However, if you override one, you must override all.
 822  **/
 823 void
 824 g_input_stream_skip_async (GInputStream        *stream,
 825                            gsize                count,
 826                            int                  io_priority,
 827                            GCancellable        *cancellable,
 828                            GAsyncReadyCallback  callback,
 829                            gpointer             user_data)
 830 {
 831   GInputStreamClass *class;
 832   GSimpleAsyncResult *simple;
 833   GError *error = NULL;
 834
 835   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 836
 837   if (count == 0)
 838     {
 839       simple = g_simple_async_result_new (G_OBJECT (stream),
 840                                           callback,
 841                                           user_data,
 842                                           g_input_stream_skip_async);
 843
 844       g_simple_async_result_complete_in_idle (simple);
 845       g_object_unref (simple);
 846       return;
 847     }
 848
 849   if (((gssize) count) < 0)
 850     {
 851       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 852                                            callback,
 853                                            user_data,
 854                                            G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 855                                            _("Too large count value passed to %s"),
 856                                            G_STRFUNC);
 857       return;
 858     }
 859
 860   if (!g_input_stream_set_pending (stream, &error))
 861     {
 862       g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
 863                                             callback,
 864                                             user_data,
 865                                             error);
 866       return;
 867     }
 868
 869   class = G_INPUT_STREAM_GET_CLASS (stream);
 870   stream->priv->outstanding_callback = callback;
 871   g_object_ref (stream);
 872   class->skip_async (stream, count, io_priority, cancellable,
 873                      async_ready_callback_wrapper, user_data);
 874 }
 875
 876 /**
 877  * g_input_stream_skip_finish:
 878  * @stream: a #GInputStream.
 879  * @result: a #GAsyncResult.
 880  * @error: a #GError location to store the error occurring, or %NULL to
 881  * ignore.
 882  *
 883  * Finishes a stream skip operation.
 884  *
 885  * Returns: the size of the bytes skipped, or %-1 on error.
 886  **/
 887 gssize
 888 g_input_stream_skip_finish (GInputStream  *stream,
 889                             GAsyncResult  *result,
 890                             GError       **error)
 891 {
 892   GSimpleAsyncResult *simple;
 893   GInputStreamClass *class;
 894
 895   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 896   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 897
 898   if (g_async_result_legacy_propagate_error (result, error))
 899     return -1;
 900
 901   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 902     {
 903       simple = G_SIMPLE_ASYNC_RESULT (result);
 904
 905       /* Special case skip of 0 bytes */
 906       if (g_simple_async_result_get_source_tag (simple) == g_input_stream_skip_async)
 907         return 0;
 908     }
 909
 910   class = G_INPUT_STREAM_GET_CLASS (stream);
 911   return class->skip_finish (stream, result, error);
 912 }
 913
 914 /**
 915  * g_input_stream_close_async:
 916  * @stream: A #GInputStream.
 917  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 918  * of the request.
 919  * @cancellable: (allow-none): optional cancellable object
 920  * @callback: (scope async): callback to call when the request is satisfied
 921  * @user_data: (closure): the data to pass to callback function
 922  *
 923  * Requests an asynchronous closes of the stream, releasing resources related to it.
 924  * When the operation is finished @callback will be called.
 925  * You can then call g_input_stream_close_finish() to get the result of the
 926  * operation.
 927  *
 928  * For behaviour details see g_input_stream_close().
 929  *
 930  * The asyncronous methods have a default fallback that uses threads to implement
 931  * asynchronicity, so they are optional for inheriting classes. However, if you
 932  * override one you must override all.
 933  **/
 934 void
 935 g_input_stream_close_async (GInputStream        *stream,
 936                             int                  io_priority,
 937                             GCancellable        *cancellable,
 938                             GAsyncReadyCallback  callback,
 939                             gpointer             user_data)
 940 {
 941   GInputStreamClass *class;
 942   GSimpleAsyncResult *simple;
 943   GError *error = NULL;
 944
 945   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 946
 947   if (stream->priv->closed)
 948     {
 949       simple = g_simple_async_result_new (G_OBJECT (stream),
 950                                           callback,
 951                                           user_data,
 952                                           g_input_stream_close_async);
 953
 954       g_simple_async_result_complete_in_idle (simple);
 955       g_object_unref (simple);
 956       return;
 957     }
 958
 959   if (!g_input_stream_set_pending (stream, &error))
 960     {
 961       g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
 962                                             callback,
 963                                             user_data,
 964                                             error);
 965       return;
 966     }
 967
 968   class = G_INPUT_STREAM_GET_CLASS (stream);
 969   stream->priv->outstanding_callback = callback;
 970   g_object_ref (stream);
 971   class->close_async (stream, io_priority, cancellable,
 972                       async_ready_close_callback_wrapper, user_data);
 973 }
 974
 975 /**
 976  * g_input_stream_close_finish:
 977  * @stream: a #GInputStream.
 978  * @result: a #GAsyncResult.
 979  * @error: a #GError location to store the error occurring, or %NULL to
 980  * ignore.
 981  *
 982  * Finishes closing a stream asynchronously, started from g_input_stream_close_async().
 983  *
 984  * Returns: %TRUE if the stream was closed successfully.
 985  **/
 986 gboolean
 987 g_input_stream_close_finish (GInputStream  *stream,
 988                              GAsyncResult  *result,
 989                              GError       **error)
 990 {
 991   GSimpleAsyncResult *simple;
 992   GInputStreamClass *class;
 993
 994   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
 995   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
 996
 997   if (g_async_result_legacy_propagate_error (result, error))
 998     return FALSE;
 999
1000   if (G_IS_SIMPLE_ASYNC_RESULT (result))
1001     {
1002       simple = G_SIMPLE_ASYNC_RESULT (result);
1003
1004       /* Special case already closed */
1005       if (g_simple_async_result_get_source_tag (simple) == g_input_stream_close_async)
1006         return TRUE;
1007     }
1008
1009   class = G_INPUT_STREAM_GET_CLASS (stream);
1010   return class->close_finish (stream, result, error);
1011 }
1012
1013 /**
1014  * g_input_stream_is_closed:
1015  * @stream: input stream.
1016  *
1017  * Checks if an input stream is closed.
1018  *
1019  * Returns: %TRUE if the stream is closed.
1020  **/
1021 gboolean
1022 g_input_stream_is_closed (GInputStream *stream)
1023 {
1024   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
1025
1026   return stream->priv->closed;
1027 }
1028
1029 /**
1030  * g_input_stream_has_pending:
1031  * @stream: input stream.
1032  *
1033  * Checks if an input stream has pending actions.
1034  *
1035  * Returns: %TRUE if @stream has pending actions.
1036  **/
1037 gboolean
1038 g_input_stream_has_pending (GInputStream *stream)
1039 {
1040   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
1041
1042   return stream->priv->pending;
1043 }
1044
1045 /**
1046  * g_input_stream_set_pending:
1047  * @stream: input stream
1048  * @error: a #GError location to store the error occurring, or %NULL to
1049  * ignore.
1050  *
1051  * Sets @stream to have actions pending. If the pending flag is
1052  * already set or @stream is closed, it will return %FALSE and set
1053  * @error.
1054  *
1055  * Return value: %TRUE if pending was previously unset and is now set.
1056  **/
1057 gboolean
1058 g_input_stream_set_pending (GInputStream *stream, GError **error)
1059 {
1060   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
1061
1062   if (stream->priv->closed)
1063     {
1064       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
1065                            _("Stream is already closed"));
1066       return FALSE;
1067     }
1068
1069   if (stream->priv->pending)
1070     {
1071       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
1072                 /* Translators: This is an error you get if there is already an
1073                  * operation running against this stream when you try to start
1074                  * one */
1075                  _("Stream has outstanding operation"));
1076       return FALSE;
1077     }
1078
1079   stream->priv->pending = TRUE;
1080   return TRUE;
1081 }
1082
1083 /**
1084  * g_input_stream_clear_pending:
1085  * @stream: input stream
1086  *
1087  * Clears the pending flag on @stream.
1088  **/
1089 void
1090 g_input_stream_clear_pending (GInputStream *stream)
1091 {
1092   g_return_if_fail (G_IS_INPUT_STREAM (stream));
1093
1094   stream->priv->pending = FALSE;
1095 }
1096
1097 /********************************************
1098  *   Default implementation of async ops    *
1099  ********************************************/
1100
1101 typedef struct {
1102   void              *buffer;
1103   gsize              count_requested;
1104   gssize             count_read;
1105
1106   GCancellable      *cancellable;
1107   gint               io_priority;
1108   gboolean           need_idle;
1109 } ReadData;
1110
1111 static void
1112 free_read_data (ReadData *op)
1113 {
1114   if (op->cancellable)
1115     g_object_unref (op->cancellable);
1116   g_slice_free (ReadData, op);
1117 }
1118
1119 static void
1120 read_async_thread (GSimpleAsyncResult *res,
1121                    GObject            *object,
1122                    GCancellable       *cancellable)
1123 {
1124   ReadData *op;
1125   GInputStreamClass *class;
1126   GError *error = NULL;
1127
1128   op = g_simple_async_result_get_op_res_gpointer (res);
1129
1130   class = G_INPUT_STREAM_GET_CLASS (object);
1131
1132   op->count_read = class->read_fn (G_INPUT_STREAM (object),
1133                                    op->buffer, op->count_requested,
1134                                    cancellable, &error);
1135   if (op->count_read == -1)
1136     g_simple_async_result_take_error (res, error);
1137 }
1138
1139 static void read_async_pollable (GPollableInputStream *stream,
1140                                  GSimpleAsyncResult   *result);
1141
1142 static gboolean
1143 read_async_pollable_ready (GPollableInputStream *stream,
1144                            gpointer              user_data)
1145 {
1146   GSimpleAsyncResult *result = user_data;
1147
1148   read_async_pollable (stream, result);
1149   return FALSE;
1150 }
1151
1152 static void
1153 read_async_pollable (GPollableInputStream *stream,
1154                      GSimpleAsyncResult   *result)
1155 {
1156   GError *error = NULL;
1157   ReadData *op = g_simple_async_result_get_op_res_gpointer (result);
1158
1159   if (g_cancellable_set_error_if_cancelled (op->cancellable, &error))
1160     op->count_read = -1;
1161   else
1162     {
1163       op->count_read = G_POLLABLE_INPUT_STREAM_GET_INTERFACE (stream)->
1164         read_nonblocking (stream, op->buffer, op->count_requested, &error);
1165     }
1166
1167   if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
1168     {
1169       GSource *source;
1170
1171       g_error_free (error);
1172       op->need_idle = FALSE;
1173
1174       source = g_pollable_input_stream_create_source (stream, op->cancellable);
1175       g_source_set_callback (source,
1176                              (GSourceFunc) read_async_pollable_ready,
1177                              g_object_ref (result), g_object_unref);
1178       g_source_set_priority (source, op->io_priority);
1179       g_source_attach (source, g_main_context_get_thread_default ());
1180       g_source_unref (source);
1181       return;
1182     }
1183
1184   if (op->count_read == -1)
1185     g_simple_async_result_take_error (result, error);
1186
1187   if (op->need_idle)
1188     g_simple_async_result_complete_in_idle (result);
1189   else
1190     g_simple_async_result_complete (result);
1191 }
1192
1193 static void
1194 g_input_stream_real_read_async (GInputStream        *stream,
1195                                 void                *buffer,
1196                                 gsize                count,
1197                                 int                  io_priority,
1198                                 GCancellable        *cancellable,
1199                                 GAsyncReadyCallback  callback,
1200                                 gpointer             user_data)
1201 {
1202   GSimpleAsyncResult *res;
1203   ReadData *op;
1204
1205   op = g_slice_new0 (ReadData);
1206   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_input_stream_real_read_async);
1207   g_simple_async_result_set_op_res_gpointer (res, op, (GDestroyNotify) free_read_data);
1208   op->buffer = buffer;
1209   op->count_requested = count;
1210   op->cancellable = cancellable ? g_object_ref (cancellable) : NULL;
1211   op->io_priority = io_priority;
1212   op->need_idle = TRUE;
1213
1214   if (G_IS_POLLABLE_INPUT_STREAM (stream) &&
1215       g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (stream)))
1216     read_async_pollable (G_POLLABLE_INPUT_STREAM (stream), res);
1217   else
1218     g_simple_async_result_run_in_thread (res, read_async_thread, io_priority, cancellable);
1219   g_object_unref (res);
1220 }
1221
1222 static gssize
1223 g_input_stream_real_read_finish (GInputStream  *stream,
1224                                  GAsyncResult  *result,
1225                                  GError       **error)
1226 {
1227   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1228   ReadData *op;
1229
1230   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) ==
1231             g_input_stream_real_read_async);
1232
1233   if (g_simple_async_result_propagate_error (simple, error))
1234     return -1;
1235
1236   op = g_simple_async_result_get_op_res_gpointer (simple);
1237
1238   return op->count_read;
1239 }
1240
1241 typedef struct {
1242   gsize count_requested;
1243   gssize count_skipped;
1244 } SkipData;
1245
1246
1247 static void
1248 skip_async_thread (GSimpleAsyncResult *res,
1249                    GObject            *object,
1250                    GCancellable       *cancellable)
1251 {
1252   SkipData *op;
1253   GInputStreamClass *class;
1254   GError *error = NULL;
1255
1256   class = G_INPUT_STREAM_GET_CLASS (object);
1257   op = g_simple_async_result_get_op_res_gpointer (res);
1258   op->count_skipped = class->skip (G_INPUT_STREAM (object),
1259                                    op->count_requested,
1260                                    cancellable, &error);
1261   if (op->count_skipped == -1)
1262     g_simple_async_result_take_error (res, error);
1263 }
1264
1265 typedef struct {
1266   char buffer[8192];
1267   gsize count;
1268   gsize count_skipped;
1269   int io_prio;
1270   GCancellable *cancellable;
1271   gpointer user_data;
1272   GAsyncReadyCallback callback;
1273 } SkipFallbackAsyncData;
1274
1275 static void
1276 skip_callback_wrapper (GObject      *source_object,
1277                        GAsyncResult *res,
1278                        gpointer      user_data)
1279 {
1280   GInputStreamClass *class;
1281   SkipFallbackAsyncData *data = user_data;
1282   SkipData *op;
1283   GSimpleAsyncResult *simple;
1284   GError *error = NULL;
1285   gssize ret;
1286
1287   ret = g_input_stream_read_finish (G_INPUT_STREAM (source_object), res, &error);
1288
1289   if (ret > 0)
1290     {
1291       data->count -= ret;
1292       data->count_skipped += ret;
1293
1294       if (data->count > 0)
1295         {
1296           class = G_INPUT_STREAM_GET_CLASS (source_object);
1297           class->read_async (G_INPUT_STREAM (source_object), data->buffer, MIN (8192, data->count), data->io_prio, data->cancellable,
1298                              skip_callback_wrapper, data);
1299           return;
1300         }
1301     }
1302
1303   op = g_new0 (SkipData, 1);
1304   op->count_skipped = data->count_skipped;
1305   simple = g_simple_async_result_new (source_object,
1306                                       data->callback, data->user_data,
1307                                       g_input_stream_real_skip_async);
1308
1309   g_simple_async_result_set_op_res_gpointer (simple, op, g_free);
1310
1311   if (ret == -1)
1312     {
1313       if (data->count_skipped &&
1314           g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED))
1315         /* No error, return partial read */
1316         g_error_free (error);
1317       else
1318         g_simple_async_result_take_error (simple, error);
1319     }
1320
1321   /* Complete immediately, not in idle, since we're already in a mainloop callout */
1322   g_simple_async_result_complete (simple);
1323   g_object_unref (simple);
1324
1325   g_free (data);
1326  }
1327
1328 static void
1329 g_input_stream_real_skip_async (GInputStream        *stream,
1330                                 gsize                count,
1331                                 int                  io_priority,
1332                                 GCancellable        *cancellable,
1333                                 GAsyncReadyCallback  callback,
1334                                 gpointer             user_data)
1335 {
1336   GInputStreamClass *class;
1337   SkipData *op;
1338   SkipFallbackAsyncData *data;
1339   GSimpleAsyncResult *res;
1340
1341   class = G_INPUT_STREAM_GET_CLASS (stream);
1342
1343   if (class->read_async == g_input_stream_real_read_async)
1344     {
1345       /* Read is thread-using async fallback.
1346        * Make skip use threads too, so that we can use a possible sync skip
1347        * implementation. */
1348       op = g_new0 (SkipData, 1);
1349
1350       res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
1351                                        g_input_stream_real_skip_async);
1352
1353       g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1354
1355       op->count_requested = count;
1356
1357       g_simple_async_result_run_in_thread (res, skip_async_thread, io_priority, cancellable);
1358       g_object_unref (res);
1359     }
1360   else
1361     {
1362       /* TODO: Skip fallback uses too much memory, should do multiple read calls */
1363
1364       /* There is a custom async read function, lets use that. */
1365       data = g_new (SkipFallbackAsyncData, 1);
1366       data->count = count;
1367       data->count_skipped = 0;
1368       data->io_prio = io_priority;
1369       data->cancellable = cancellable;
1370       data->callback = callback;
1371       data->user_data = user_data;
1372       class->read_async (stream, data->buffer, MIN (8192, count), io_priority, cancellable,
1373                          skip_callback_wrapper, data);
1374     }
1375
1376 }
1377
1378 static gssize
1379 g_input_stream_real_skip_finish (GInputStream  *stream,
1380                                  GAsyncResult  *result,
1381                                  GError       **error)
1382 {
1383   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1384   SkipData *op;
1385
1386   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_skip_async);
1387
1388   if (g_simple_async_result_propagate_error (simple, error))
1389     return -1;
1390
1391   op = g_simple_async_result_get_op_res_gpointer (simple);
1392   return op->count_skipped;
1393 }
1394
1395 static void
1396 close_async_thread (GSimpleAsyncResult *res,
1397                     GObject            *object,
1398                     GCancellable       *cancellable)
1399 {
1400   GInputStreamClass *class;
1401   GError *error = NULL;
1402   gboolean result;
1403
1404   /* Auto handling of cancelation disabled, and ignore
1405      cancellation, since we want to close things anyway, although
1406      possibly in a quick-n-dirty way. At least we never want to leak
1407      open handles */
1408
1409   class = G_INPUT_STREAM_GET_CLASS (object);
1410   if (class->close_fn)
1411     {
1412       result = class->close_fn (G_INPUT_STREAM (object), cancellable, &error);
1413       if (!result)
1414         g_simple_async_result_take_error (res, error);
1415     }
1416 }
1417
1418 static void
1419 g_input_stream_real_close_async (GInputStream        *stream,
1420                                  int                  io_priority,
1421                                  GCancellable        *cancellable,
1422                                  GAsyncReadyCallback  callback,
1423                                  gpointer             user_data)
1424 {
1425   GSimpleAsyncResult *res;
1426
1427   res = g_simple_async_result_new (G_OBJECT (stream),
1428                                    callback,
1429                                    user_data,
1430                                    g_input_stream_real_close_async);
1431
1432   g_simple_async_result_set_handle_cancellation (res, FALSE);
1433
1434   g_simple_async_result_run_in_thread (res,
1435                                        close_async_thread,
1436                                        io_priority,
1437                                        cancellable);
1438   g_object_unref (res);
1439 }
1440
1441 static gboolean
1442 g_input_stream_real_close_finish (GInputStream  *stream,
1443                                   GAsyncResult  *result,
1444                                   GError       **error)
1445 {
1446   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1447
1448   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_close_async);
1449
1450   if (g_simple_async_result_propagate_error (simple, error))
1451     return FALSE;
1452
1453   return TRUE;
1454 }