gio/ginputstream.c

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