gio/ginputstream.c

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