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