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