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