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