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