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
  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_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: optional #GCancellable object, %NULL to ignore.
 143  * @error: location to store the error occuring, 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
 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: location to store the number of bytes that was read from the stream
 220  * @cancellable: optional #GCancellable object, %NULL to ignore.
 221  * @error: location to store the error occuring, 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_skip:
 278  * @stream: a #GInputStream.
 279  * @count: the number of bytes that will be skipped from the stream
 280  * @cancellable: optional #GCancellable object, %NULL to ignore.
 281  * @error: location to store the error occuring, or %NULL to ignore
 282  *
 283  * Tries to skip @count bytes from the stream. Will block during the operation.
 284  *
 285  * This is identical to g_input_stream_read(), from a behaviour standpoint,
 286  * but the bytes that are skipped are not returned to the user. Some
 287  * streams have an implementation that is more efficient than reading the data.
 288  *
 289  * This function is optional for inherited classes, as the default implementation
 290  * emulates it using read.
 291  *
 292  * If @cancellable is not %NULL, then the operation can be cancelled by
 293  * triggering the cancellable object from another thread. If the operation
 294  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
 295  * operation was partially finished when the operation was cancelled the
 296  * partial result will be returned, without an error.
 297  *
 298  * Return value: Number of bytes skipped, or -1 on error
 299  **/
 300 gssize
 301 g_input_stream_skip (GInputStream  *stream,
 302                      gsize          count,
 303                      GCancellable  *cancellable,
 304                      GError       **error)
 305 {
 306   GInputStreamClass *class;
 307   gssize res;
 308
 309   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 310
 311   if (count == 0)
 312     return 0;
 313
 314   if (((gssize) count) < 0)
 315     {
 316       g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 317                    _("Too large count value passed to %s"), G_STRFUNC);
 318       return -1;
 319     }
 320
 321   class = G_INPUT_STREAM_GET_CLASS (stream);
 322
 323   if (!g_input_stream_set_pending (stream, error))
 324     return -1;
 325
 326   if (cancellable)
 327     g_cancellable_push_current (cancellable);
 328
 329   res = class->skip (stream, count, cancellable, error);
 330
 331   if (cancellable)
 332     g_cancellable_pop_current (cancellable);
 333
 334   g_input_stream_clear_pending (stream);
 335
 336   return res;
 337 }
 338
 339 static gssize
 340 g_input_stream_real_skip (GInputStream  *stream,
 341                           gsize          count,
 342                           GCancellable  *cancellable,
 343                           GError       **error)
 344 {
 345   GInputStreamClass *class;
 346   gssize ret, read_bytes;
 347   char buffer[8192];
 348   GError *my_error;
 349
 350   if (G_IS_SEEKABLE (stream) && g_seekable_can_seek (G_SEEKABLE (stream)))
 351     {
 352       if (g_seekable_seek (G_SEEKABLE (stream),
 353                            count,
 354                            G_SEEK_CUR,
 355                            cancellable,
 356                            NULL))
 357         return count;
 358     }
 359
 360   /* If not seekable, or seek failed, fall back to reading data: */
 361
 362   class = G_INPUT_STREAM_GET_CLASS (stream);
 363
 364   read_bytes = 0;
 365   while (1)
 366     {
 367       my_error = NULL;
 368
 369       ret = class->read_fn (stream, buffer, MIN (sizeof (buffer), count),
 370                             cancellable, &my_error);
 371       if (ret == -1)
 372         {
 373           if (read_bytes > 0 &&
 374               my_error->domain == G_IO_ERROR &&
 375               my_error->code == G_IO_ERROR_CANCELLED)
 376             {
 377               g_error_free (my_error);
 378               return read_bytes;
 379             }
 380
 381           g_propagate_error (error, my_error);
 382           return -1;
 383         }
 384
 385       count -= ret;
 386       read_bytes += ret;
 387
 388       if (ret == 0 || count == 0)
 389         return read_bytes;
 390     }
 391 }
 392
 393 /**
 394  * g_input_stream_close:
 395  * @stream: A #GInputStream.
 396  * @cancellable: optional #GCancellable object, %NULL to ignore.
 397  * @error: location to store the error occuring, or %NULL to ignore
 398  *
 399  * Closes the stream, releasing resources related to it.
 400  *
 401  * Once the stream is closed, all other operations will return %G_IO_ERROR_CLOSED.
 402  * Closing a stream multiple times will not return an error.
 403  *
 404  * Streams will be automatically closed when the last reference
 405  * is dropped, but you might want to call this function to make sure
 406  * resources are released as early as possible.
 407  *
 408  * Some streams might keep the backing store of the stream (e.g. a file descriptor)
 409  * open after the stream is closed. See the documentation for the individual
 410  * stream for details.
 411  *
 412  * On failure the first error that happened will be reported, but the close
 413  * operation will finish as much as possible. A stream that failed to
 414  * close will still return %G_IO_ERROR_CLOSED for all operations. Still, it
 415  * is important to check and report the error to the user.
 416  *
 417  * If @cancellable is not NULL, then the operation can be cancelled by
 418  * triggering the cancellable object from another thread. If the operation
 419  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 420  * Cancelling a close will still leave the stream closed, but some streams
 421  * can use a faster close that doesn't block to e.g. check errors.
 422  *
 423  * Return value: %TRUE on success, %FALSE on failure
 424  **/
 425 gboolean
 426 g_input_stream_close (GInputStream  *stream,
 427                       GCancellable  *cancellable,
 428                       GError       **error)
 429 {
 430   GInputStreamClass *class;
 431   gboolean res;
 432
 433   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
 434
 435   class = G_INPUT_STREAM_GET_CLASS (stream);
 436
 437   if (stream->priv->closed)
 438     return TRUE;
 439
 440   res = TRUE;
 441
 442   if (!g_input_stream_set_pending (stream, error))
 443     return FALSE;
 444
 445   if (cancellable)
 446     g_cancellable_push_current (cancellable);
 447
 448   if (class->close_fn)
 449     res = class->close_fn (stream, cancellable, error);
 450
 451   if (cancellable)
 452     g_cancellable_pop_current (cancellable);
 453
 454   g_input_stream_clear_pending (stream);
 455
 456   stream->priv->closed = TRUE;
 457
 458   return res;
 459 }
 460
 461 static void
 462 async_ready_callback_wrapper (GObject      *source_object,
 463                               GAsyncResult *res,
 464                               gpointer      user_data)
 465 {
 466   GInputStream *stream = G_INPUT_STREAM (source_object);
 467
 468   g_input_stream_clear_pending (stream);
 469   if (stream->priv->outstanding_callback)
 470     (*stream->priv->outstanding_callback) (source_object, res, user_data);
 471   g_object_unref (stream);
 472 }
 473
 474 static void
 475 async_ready_close_callback_wrapper (GObject      *source_object,
 476                                     GAsyncResult *res,
 477                                     gpointer      user_data)
 478 {
 479   GInputStream *stream = G_INPUT_STREAM (source_object);
 480
 481   g_input_stream_clear_pending (stream);
 482   stream->priv->closed = TRUE;
 483   if (stream->priv->outstanding_callback)
 484     (*stream->priv->outstanding_callback) (source_object, res, user_data);
 485   g_object_unref (stream);
 486 }
 487
 488 /**
 489  * g_input_stream_read_async:
 490  * @stream: A #GInputStream.
 491  * @buffer: a buffer to read data into (which should be at least count bytes long).
 492  * @count: the number of bytes that will be read from the stream
 493  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 494  * of the request.
 495  * @cancellable: optional #GCancellable object, %NULL to ignore.
 496  * @callback: callback to call when the request is satisfied
 497  * @user_data: the data to pass to callback function
 498  *
 499  * Request an asynchronous read of @count bytes from the stream into the buffer
 500  * starting at @buffer. When the operation is finished @callback will be called.
 501  * You can then call g_input_stream_read_finish() to get the result of the
 502  * operation.
 503  *
 504  * During an async request no other sync and async calls are allowed on @stream, and will
 505  * result in %G_IO_ERROR_PENDING errors.
 506  *
 507  * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 508  *
 509  * On success, the number of bytes read into the buffer will be passed to the
 510  * callback. It is not an error if this is not the same as the requested size, as it
 511  * can happen e.g. near the end of a file, but generally we try to read
 512  * as many bytes as requested. Zero is returned on end of file
 513  * (or if @count is zero),  but never otherwise.
 514  *
 515  * Any outstanding i/o request with higher priority (lower numerical value) will
 516  * be executed before an outstanding request with lower priority. Default
 517  * priority is %G_PRIORITY_DEFAULT.
 518  *
 519  * The asyncronous methods have a default fallback that uses threads to implement
 520  * asynchronicity, so they are optional for inheriting classes. However, if you
 521  * override one you must override all.
 522  **/
 523 void
 524 g_input_stream_read_async (GInputStream        *stream,
 525                            void                *buffer,
 526                            gsize                count,
 527                            int                  io_priority,
 528                            GCancellable        *cancellable,
 529                            GAsyncReadyCallback  callback,
 530                            gpointer             user_data)
 531 {
 532   GInputStreamClass *class;
 533   GSimpleAsyncResult *simple;
 534   GError *error = NULL;
 535
 536   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 537   g_return_if_fail (buffer != NULL);
 538
 539   if (count == 0)
 540     {
 541       simple = g_simple_async_result_new (G_OBJECT (stream),
 542                                           callback,
 543                                           user_data,
 544                                           g_input_stream_read_async);
 545       g_simple_async_result_complete_in_idle (simple);
 546       g_object_unref (simple);
 547       return;
 548     }
 549
 550   if (((gssize) count) < 0)
 551     {
 552       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 553                                            callback,
 554                                            user_data,
 555                                            G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 556                                            _("Too large count value passed to %s"),
 557                                            G_STRFUNC);
 558       return;
 559     }
 560
 561   if (!g_input_stream_set_pending (stream, &error))
 562     {
 563       g_simple_async_report_gerror_in_idle (G_OBJECT (stream),
 564                                             callback,
 565                                             user_data,
 566                                             error);
 567       g_error_free (error);
 568       return;
 569     }
 570
 571   class = G_INPUT_STREAM_GET_CLASS (stream);
 572   stream->priv->outstanding_callback = callback;
 573   g_object_ref (stream);
 574   class->read_async (stream, buffer, count, io_priority, cancellable,
 575                      async_ready_callback_wrapper, user_data);
 576 }
 577
 578 /**
 579  * g_input_stream_read_finish:
 580  * @stream: a #GInputStream.
 581  * @result: a #GAsyncResult.
 582  * @error: a #GError location to store the error occuring, or %NULL to
 583  * ignore.
 584  *
 585  * Finishes an asynchronous stream read operation.
 586  *
 587  * Returns: number of bytes read in, or -1 on error.
 588  **/
 589 gssize
 590 g_input_stream_read_finish (GInputStream  *stream,
 591                             GAsyncResult  *result,
 592                             GError       **error)
 593 {
 594   GSimpleAsyncResult *simple;
 595   GInputStreamClass *class;
 596
 597   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 598   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 599
 600   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 601     {
 602       simple = G_SIMPLE_ASYNC_RESULT (result);
 603       if (g_simple_async_result_propagate_error (simple, error))
 604         return -1;
 605
 606       /* Special case read of 0 bytes */
 607       if (g_simple_async_result_get_source_tag (simple) == g_input_stream_read_async)
 608         return 0;
 609     }
 610
 611   class = G_INPUT_STREAM_GET_CLASS (stream);
 612   return class->read_finish (stream, result, error);
 613 }
 614
 615 /**
 616  * g_input_stream_skip_async:
 617  * @stream: A #GInputStream.
 618  * @count: the number of bytes that will be skipped from the stream
 619  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 620  * of the request.
 621  * @cancellable: optional #GCancellable object, %NULL to ignore.
 622  * @callback: callback to call when the request is satisfied
 623  * @user_data: the data to pass to callback function
 624  *
 625  * Request an asynchronous skip of @count bytes from the stream.
 626  * When the operation is finished @callback will be called.
 627  * You can then call g_input_stream_skip_finish() to get the result of the
 628  * operation.
 629  *
 630  * During an async request no other sync and async calls are allowed, and will
 631  * result in %G_IO_ERROR_PENDING errors.
 632  *
 633  * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 634  *
 635  * On success, the number of bytes skipped will be passed to the
 636  * callback. It is not an error if this is not the same as the requested size, as it
 637  * can happen e.g. near the end of a file, but generally we try to skip
 638  * as many bytes as requested. Zero is returned on end of file
 639  * (or if @count is zero), but never otherwise.
 640  *
 641  * Any outstanding i/o request with higher priority (lower numerical value) will
 642  * be executed before an outstanding request with lower priority. Default
 643  * priority is %G_PRIORITY_DEFAULT.
 644  *
 645  * The asyncronous methods have a default fallback that uses threads to implement
 646  * asynchronicity, so they are optional for inheriting classes. However, if you
 647  * override one you must override all.
 648  **/
 649 void
 650 g_input_stream_skip_async (GInputStream        *stream,
 651                            gsize                count,
 652                            int                  io_priority,
 653                            GCancellable        *cancellable,
 654                            GAsyncReadyCallback  callback,
 655                            gpointer             user_data)
 656 {
 657   GInputStreamClass *class;
 658   GSimpleAsyncResult *simple;
 659   GError *error = NULL;
 660
 661   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 662
 663   if (count == 0)
 664     {
 665       simple = g_simple_async_result_new (G_OBJECT (stream),
 666                                           callback,
 667                                           user_data,
 668                                           g_input_stream_skip_async);
 669
 670       g_simple_async_result_complete_in_idle (simple);
 671       g_object_unref (simple);
 672       return;
 673     }
 674
 675   if (((gssize) count) < 0)
 676     {
 677       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 678                                            callback,
 679                                            user_data,
 680                                            G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 681                                            _("Too large count value passed to %s"),
 682                                            G_STRFUNC);
 683       return;
 684     }
 685
 686   if (!g_input_stream_set_pending (stream, &error))
 687     {
 688       g_simple_async_report_gerror_in_idle (G_OBJECT (stream),
 689                                             callback,
 690                                             user_data,
 691                                             error);
 692       g_error_free (error);
 693       return;
 694     }
 695
 696   class = G_INPUT_STREAM_GET_CLASS (stream);
 697   stream->priv->outstanding_callback = callback;
 698   g_object_ref (stream);
 699   class->skip_async (stream, count, io_priority, cancellable,
 700                      async_ready_callback_wrapper, user_data);
 701 }
 702
 703 /**
 704  * g_input_stream_skip_finish:
 705  * @stream: a #GInputStream.
 706  * @result: a #GAsyncResult.
 707  * @error: a #GError location to store the error occuring, or %NULL to
 708  * ignore.
 709  *
 710  * Finishes a stream skip operation.
 711  *
 712  * Returns: the size of the bytes skipped, or %-1 on error.
 713  **/
 714 gssize
 715 g_input_stream_skip_finish (GInputStream  *stream,
 716                             GAsyncResult  *result,
 717                             GError       **error)
 718 {
 719   GSimpleAsyncResult *simple;
 720   GInputStreamClass *class;
 721
 722   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
 723   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 724
 725   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 726     {
 727       simple = G_SIMPLE_ASYNC_RESULT (result);
 728       if (g_simple_async_result_propagate_error (simple, error))
 729         return -1;
 730
 731       /* Special case skip of 0 bytes */
 732       if (g_simple_async_result_get_source_tag (simple) == g_input_stream_skip_async)
 733         return 0;
 734     }
 735
 736   class = G_INPUT_STREAM_GET_CLASS (stream);
 737   return class->skip_finish (stream, result, error);
 738 }
 739
 740 /**
 741  * g_input_stream_close_async:
 742  * @stream: A #GInputStream.
 743  * @io_priority: the <link linkend="io-priority">I/O priority</link>
 744  * of the request.
 745  * @cancellable: optional cancellable object
 746  * @callback: callback to call when the request is satisfied
 747  * @user_data: the data to pass to callback function
 748  *
 749  * Requests an asynchronous closes of the stream, releasing resources related to it.
 750  * When the operation is finished @callback will be called.
 751  * You can then call g_input_stream_close_finish() to get the result of the
 752  * operation.
 753  *
 754  * For behaviour details see g_input_stream_close().
 755  *
 756  * The asyncronous methods have a default fallback that uses threads to implement
 757  * asynchronicity, so they are optional for inheriting classes. However, if you
 758  * override one you must override all.
 759  **/
 760 void
 761 g_input_stream_close_async (GInputStream        *stream,
 762                             int                  io_priority,
 763                             GCancellable        *cancellable,
 764                             GAsyncReadyCallback  callback,
 765                             gpointer             user_data)
 766 {
 767   GInputStreamClass *class;
 768   GSimpleAsyncResult *simple;
 769   GError *error = NULL;
 770
 771   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 772
 773   if (stream->priv->closed)
 774     {
 775       simple = g_simple_async_result_new (G_OBJECT (stream),
 776                                           callback,
 777                                           user_data,
 778                                           g_input_stream_close_async);
 779
 780       g_simple_async_result_complete_in_idle (simple);
 781       g_object_unref (simple);
 782       return;
 783     }
 784
 785   if (!g_input_stream_set_pending (stream, &error))
 786     {
 787       g_simple_async_report_gerror_in_idle (G_OBJECT (stream),
 788                                             callback,
 789                                             user_data,
 790                                             error);
 791       g_error_free (error);
 792       return;
 793     }
 794
 795   class = G_INPUT_STREAM_GET_CLASS (stream);
 796   stream->priv->outstanding_callback = callback;
 797   g_object_ref (stream);
 798   class->close_async (stream, io_priority, cancellable,
 799                       async_ready_close_callback_wrapper, user_data);
 800 }
 801
 802 /**
 803  * g_input_stream_close_finish:
 804  * @stream: a #GInputStream.
 805  * @result: a #GAsyncResult.
 806  * @error: a #GError location to store the error occuring, or %NULL to
 807  * ignore.
 808  *
 809  * Finishes closing a stream asynchronously, started from g_input_stream_close_async().
 810  *
 811  * Returns: %TRUE if the stream was closed successfully.
 812  **/
 813 gboolean
 814 g_input_stream_close_finish (GInputStream  *stream,
 815                              GAsyncResult  *result,
 816                              GError       **error)
 817 {
 818   GSimpleAsyncResult *simple;
 819   GInputStreamClass *class;
 820
 821   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
 822   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
 823
 824   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 825     {
 826       simple = G_SIMPLE_ASYNC_RESULT (result);
 827       if (g_simple_async_result_propagate_error (simple, error))
 828         return FALSE;
 829
 830       /* Special case already closed */
 831       if (g_simple_async_result_get_source_tag (simple) == g_input_stream_close_async)
 832         return TRUE;
 833     }
 834
 835   class = G_INPUT_STREAM_GET_CLASS (stream);
 836   return class->close_finish (stream, result, error);
 837 }
 838
 839 /**
 840  * g_input_stream_is_closed:
 841  * @stream: input stream.
 842  *
 843  * Checks if an input stream is closed.
 844  *
 845  * Returns: %TRUE if the stream is closed.
 846  **/
 847 gboolean
 848 g_input_stream_is_closed (GInputStream *stream)
 849 {
 850   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
 851
 852   return stream->priv->closed;
 853 }
 854
 855 /**
 856  * g_input_stream_has_pending:
 857  * @stream: input stream.
 858  *
 859  * Checks if an input stream has pending actions.
 860  *
 861  * Returns: %TRUE if @stream has pending actions.
 862  **/
 863 gboolean
 864 g_input_stream_has_pending (GInputStream *stream)
 865 {
 866   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
 867
 868   return stream->priv->pending;
 869 }
 870
 871 /**
 872  * g_input_stream_set_pending:
 873  * @stream: input stream
 874  * @error: a #GError location to store the error occuring, or %NULL to
 875  * ignore.
 876  *
 877  * Sets @stream to have actions pending. If the pending flag is
 878  * already set or @stream is closed, it will return %FALSE and set
 879  * @error.
 880  *
 881  * Return value: %TRUE if pending was previously unset and is now set.
 882  **/
 883 gboolean
 884 g_input_stream_set_pending (GInputStream *stream, GError **error)
 885 {
 886   g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
 887
 888   if (stream->priv->closed)
 889     {
 890       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
 891                            _("Stream is already closed"));
 892       return FALSE;
 893     }
 894
 895   if (stream->priv->pending)
 896     {
 897       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
 898                 /* Translators: This is an error you get if there is already an
 899                  * operation running against this stream when you try to start
 900                  * one */
 901                  _("Stream has outstanding operation"));
 902       return FALSE;
 903     }
 904
 905   stream->priv->pending = TRUE;
 906   return TRUE;
 907 }
 908
 909 /**
 910  * g_input_stream_clear_pending:
 911  * @stream: input stream
 912  *
 913  * Clears the pending flag on @stream.
 914  **/
 915 void
 916 g_input_stream_clear_pending (GInputStream *stream)
 917 {
 918   g_return_if_fail (G_IS_INPUT_STREAM (stream));
 919
 920   stream->priv->pending = FALSE;
 921 }
 922
 923 /********************************************
 924  *   Default implementation of async ops    *
 925  ********************************************/
 926
 927 typedef struct {
 928   void              *buffer;
 929   gsize              count_requested;
 930   gssize             count_read;
 931 } ReadData;
 932
 933 static void
 934 read_async_thread (GSimpleAsyncResult *res,
 935                    GObject            *object,
 936                    GCancellable       *cancellable)
 937 {
 938   ReadData *op;
 939   GInputStreamClass *class;
 940   GError *error = NULL;
 941
 942   op = g_simple_async_result_get_op_res_gpointer (res);
 943
 944   class = G_INPUT_STREAM_GET_CLASS (object);
 945
 946   op->count_read = class->read_fn (G_INPUT_STREAM (object),
 947                                    op->buffer, op->count_requested,
 948                                    cancellable, &error);
 949   if (op->count_read == -1)
 950     {
 951       g_simple_async_result_set_from_error (res, error);
 952       g_error_free (error);
 953     }
 954 }
 955
 956 static void
 957 g_input_stream_real_read_async (GInputStream        *stream,
 958                                 void                *buffer,
 959                                 gsize                count,
 960                                 int                  io_priority,
 961                                 GCancellable        *cancellable,
 962                                 GAsyncReadyCallback  callback,
 963                                 gpointer             user_data)
 964 {
 965   GSimpleAsyncResult *res;
 966   ReadData *op;
 967
 968   op = g_new (ReadData, 1);
 969   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_input_stream_real_read_async);
 970   g_simple_async_result_set_op_res_gpointer (res, op, g_free);
 971   op->buffer = buffer;
 972   op->count_requested = count;
 973
 974   g_simple_async_result_run_in_thread (res, read_async_thread, io_priority, cancellable);
 975   g_object_unref (res);
 976 }
 977
 978 static gssize
 979 g_input_stream_real_read_finish (GInputStream  *stream,
 980                                  GAsyncResult  *result,
 981                                  GError       **error)
 982 {
 983   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
 984   ReadData *op;
 985
 986   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) ==
 987             g_input_stream_real_read_async);
 988
 989   op = g_simple_async_result_get_op_res_gpointer (simple);
 990
 991   return op->count_read;
 992 }
 993
 994 typedef struct {
 995   gsize count_requested;
 996   gssize count_skipped;
 997 } SkipData;
 998
 999
1000 static void
1001 skip_async_thread (GSimpleAsyncResult *res,
1002                    GObject            *object,
1003                    GCancellable       *cancellable)
1004 {
1005   SkipData *op;
1006   GInputStreamClass *class;
1007   GError *error = NULL;
1008
1009   class = G_INPUT_STREAM_GET_CLASS (object);
1010   op = g_simple_async_result_get_op_res_gpointer (res);
1011   op->count_skipped = class->skip (G_INPUT_STREAM (object),
1012                                    op->count_requested,
1013                                    cancellable, &error);
1014   if (op->count_skipped == -1)
1015     {
1016       g_simple_async_result_set_from_error (res, error);
1017       g_error_free (error);
1018     }
1019 }
1020
1021 typedef struct {
1022   char buffer[8192];
1023   gsize count;
1024   gsize count_skipped;
1025   int io_prio;
1026   GCancellable *cancellable;
1027   gpointer user_data;
1028   GAsyncReadyCallback callback;
1029 } SkipFallbackAsyncData;
1030
1031 static void
1032 skip_callback_wrapper (GObject      *source_object,
1033                        GAsyncResult *res,
1034                        gpointer      user_data)
1035 {
1036   GInputStreamClass *class;
1037   SkipFallbackAsyncData *data = user_data;
1038   SkipData *op;
1039   GSimpleAsyncResult *simple;
1040   GError *error = NULL;
1041   gssize ret;
1042
1043   ret = g_input_stream_read_finish (G_INPUT_STREAM (source_object), res, &error);
1044
1045   if (ret > 0)
1046     {
1047       data->count -= ret;
1048       data->count_skipped += ret;
1049
1050       if (data->count > 0)
1051         {
1052           class = G_INPUT_STREAM_GET_CLASS (source_object);
1053           class->read_async (G_INPUT_STREAM (source_object), data->buffer, MIN (8192, data->count), data->io_prio, data->cancellable,
1054                              skip_callback_wrapper, data);
1055           return;
1056         }
1057     }
1058
1059   op = g_new0 (SkipData, 1);
1060   op->count_skipped = data->count_skipped;
1061   simple = g_simple_async_result_new (source_object,
1062                                       data->callback, data->user_data,
1063                                       g_input_stream_real_skip_async);
1064
1065   g_simple_async_result_set_op_res_gpointer (simple, op, g_free);
1066
1067   if (ret == -1)
1068     {
1069       if (data->count_skipped &&
1070           error->domain == G_IO_ERROR &&
1071           error->code == G_IO_ERROR_CANCELLED)
1072         { /* No error, return partial read */ }
1073       else
1074         g_simple_async_result_set_from_error (simple, error);
1075       g_error_free (error);
1076     }
1077
1078   /* Complete immediately, not in idle, since we're already in a mainloop callout */
1079   g_simple_async_result_complete (simple);
1080   g_object_unref (simple);
1081
1082   g_free (data);
1083  }
1084
1085 static void
1086 g_input_stream_real_skip_async (GInputStream        *stream,
1087                                 gsize                count,
1088                                 int                  io_priority,
1089                                 GCancellable        *cancellable,
1090                                 GAsyncReadyCallback  callback,
1091                                 gpointer             user_data)
1092 {
1093   GInputStreamClass *class;
1094   SkipData *op;
1095   SkipFallbackAsyncData *data;
1096   GSimpleAsyncResult *res;
1097
1098   class = G_INPUT_STREAM_GET_CLASS (stream);
1099
1100   if (class->read_async == g_input_stream_real_read_async)
1101     {
1102       /* Read is thread-using async fallback.
1103        * Make skip use threads too, so that we can use a possible sync skip
1104        * implementation. */
1105       op = g_new0 (SkipData, 1);
1106
1107       res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
1108                                        g_input_stream_real_skip_async);
1109
1110       g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1111
1112       op->count_requested = count;
1113
1114       g_simple_async_result_run_in_thread (res, skip_async_thread, io_priority, cancellable);
1115       g_object_unref (res);
1116     }
1117   else
1118     {
1119       /* TODO: Skip fallback uses too much memory, should do multiple read calls */
1120
1121       /* There is a custom async read function, lets use that. */
1122       data = g_new (SkipFallbackAsyncData, 1);
1123       data->count = count;
1124       data->count_skipped = 0;
1125       data->io_prio = io_priority;
1126       data->cancellable = cancellable;
1127       data->callback = callback;
1128       data->user_data = user_data;
1129       class->read_async (stream, data->buffer, MIN (8192, count), io_priority, cancellable,
1130                          skip_callback_wrapper, data);
1131     }
1132
1133 }
1134
1135 static gssize
1136 g_input_stream_real_skip_finish (GInputStream  *stream,
1137                                  GAsyncResult  *result,
1138                                  GError       **error)
1139 {
1140   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1141   SkipData *op;
1142
1143   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_skip_async);
1144   op = g_simple_async_result_get_op_res_gpointer (simple);
1145   return op->count_skipped;
1146 }
1147
1148 static void
1149 close_async_thread (GSimpleAsyncResult *res,
1150                     GObject            *object,
1151                     GCancellable       *cancellable)
1152 {
1153   GInputStreamClass *class;
1154   GError *error = NULL;
1155   gboolean result;
1156
1157   /* Auto handling of cancelation disabled, and ignore
1158      cancellation, since we want to close things anyway, although
1159      possibly in a quick-n-dirty way. At least we never want to leak
1160      open handles */
1161
1162   class = G_INPUT_STREAM_GET_CLASS (object);
1163   if (class->close_fn)
1164     {
1165       result = class->close_fn (G_INPUT_STREAM (object), cancellable, &error);
1166       if (!result)
1167         {
1168           g_simple_async_result_set_from_error (res, error);
1169           g_error_free (error);
1170         }
1171     }
1172 }
1173
1174 static void
1175 g_input_stream_real_close_async (GInputStream        *stream,
1176                                  int                  io_priority,
1177                                  GCancellable        *cancellable,
1178                                  GAsyncReadyCallback  callback,
1179                                  gpointer             user_data)
1180 {
1181   GSimpleAsyncResult *res;
1182
1183   res = g_simple_async_result_new (G_OBJECT (stream),
1184                                    callback,
1185                                    user_data,
1186                                    g_input_stream_real_close_async);
1187
1188   g_simple_async_result_set_handle_cancellation (res, FALSE);
1189
1190   g_simple_async_result_run_in_thread (res,
1191                                        close_async_thread,
1192                                        io_priority,
1193                                        cancellable);
1194   g_object_unref (res);
1195 }
1196
1197 static gboolean
1198 g_input_stream_real_close_finish (GInputStream  *stream,
1199                                   GAsyncResult  *result,
1200                                   GError       **error)
1201 {
1202   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1203   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_close_async);
1204   return TRUE;
1205 }