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