gio/goutputstream.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 <string.h>
  25 #include "goutputstream.h"
  26 #include "gcancellable.h"
  27 #include "gasyncresult.h"
  28 #include "gtask.h"
  29 #include "ginputstream.h"
  30 #include "gioerror.h"
  31 #include "gioprivate.h"
  32 #include "glibintl.h"
  33 #include "gpollableoutputstream.h"
  34
  35 /**
  36  * SECTION:goutputstream
  37  * @short_description: Base class for implementing streaming output
  38  * @include: gio/gio.h
  39  *
  40  * #GOutputStream has functions to write to a stream (g_output_stream_write()),
  41  * to close a stream (g_output_stream_close()) and to flush pending writes
  42  * (g_output_stream_flush()).
  43  *
  44  * To copy the content of an input stream to an output stream without
  45  * manually handling the reads and writes, use g_output_stream_splice().
  46  *
  47  * All of these functions have async variants too.
  48  **/
  49
  50 struct _GOutputStreamPrivate {
  51   guint closed : 1;
  52   guint pending : 1;
  53   guint closing : 1;
  54   GAsyncReadyCallback outstanding_callback;
  55 };
  56
  57 G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE (GOutputStream, g_output_stream, G_TYPE_OBJECT)
  58
  59 static gssize   g_output_stream_real_splice        (GOutputStream             *stream,
  60                                                     GInputStream              *source,
  61                                                     GOutputStreamSpliceFlags   flags,
  62                                                     GCancellable              *cancellable,
  63                                                     GError                   **error);
  64 static void     g_output_stream_real_write_async   (GOutputStream             *stream,
  65                                                     const void                *buffer,
  66                                                     gsize                      count,
  67                                                     int                        io_priority,
  68                                                     GCancellable              *cancellable,
  69                                                     GAsyncReadyCallback        callback,
  70                                                     gpointer                   data);
  71 static gssize   g_output_stream_real_write_finish  (GOutputStream             *stream,
  72                                                     GAsyncResult              *result,
  73                                                     GError                   **error);
  74 static void     g_output_stream_real_splice_async  (GOutputStream             *stream,
  75                                                     GInputStream              *source,
  76                                                     GOutputStreamSpliceFlags   flags,
  77                                                     int                        io_priority,
  78                                                     GCancellable              *cancellable,
  79                                                     GAsyncReadyCallback        callback,
  80                                                     gpointer                   data);
  81 static gssize   g_output_stream_real_splice_finish (GOutputStream             *stream,
  82                                                     GAsyncResult              *result,
  83                                                     GError                   **error);
  84 static void     g_output_stream_real_flush_async   (GOutputStream             *stream,
  85                                                     int                        io_priority,
  86                                                     GCancellable              *cancellable,
  87                                                     GAsyncReadyCallback        callback,
  88                                                     gpointer                   data);
  89 static gboolean g_output_stream_real_flush_finish  (GOutputStream             *stream,
  90                                                     GAsyncResult              *result,
  91                                                     GError                   **error);
  92 static void     g_output_stream_real_close_async   (GOutputStream             *stream,
  93                                                     int                        io_priority,
  94                                                     GCancellable              *cancellable,
  95                                                     GAsyncReadyCallback        callback,
  96                                                     gpointer                   data);
  97 static gboolean g_output_stream_real_close_finish  (GOutputStream             *stream,
  98                                                     GAsyncResult              *result,
  99                                                     GError                   **error);
 100 static gboolean g_output_stream_internal_close     (GOutputStream             *stream,
 101                                                     GCancellable              *cancellable,
 102                                                     GError                   **error);
 103 static void     g_output_stream_internal_close_async (GOutputStream           *stream,
 104                                                       int                      io_priority,
 105                                                       GCancellable            *cancellable,
 106                                                       GAsyncReadyCallback      callback,
 107                                                       gpointer                 data);
 108 static gboolean g_output_stream_internal_close_finish (GOutputStream          *stream,
 109                                                        GAsyncResult           *result,
 110                                                        GError                **error);
 111
 112 static void
 113 g_output_stream_dispose (GObject *object)
 114 {
 115   GOutputStream *stream;
 116
 117   stream = G_OUTPUT_STREAM (object);
 118
 119   if (!stream->priv->closed)
 120     g_output_stream_close (stream, NULL, NULL);
 121
 122   G_OBJECT_CLASS (g_output_stream_parent_class)->dispose (object);
 123 }
 124
 125 static void
 126 g_output_stream_class_init (GOutputStreamClass *klass)
 127 {
 128   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 129
 130   gobject_class->dispose = g_output_stream_dispose;
 131
 132   klass->splice = g_output_stream_real_splice;
 133
 134   klass->write_async = g_output_stream_real_write_async;
 135   klass->write_finish = g_output_stream_real_write_finish;
 136   klass->splice_async = g_output_stream_real_splice_async;
 137   klass->splice_finish = g_output_stream_real_splice_finish;
 138   klass->flush_async = g_output_stream_real_flush_async;
 139   klass->flush_finish = g_output_stream_real_flush_finish;
 140   klass->close_async = g_output_stream_real_close_async;
 141   klass->close_finish = g_output_stream_real_close_finish;
 142 }
 143
 144 static void
 145 g_output_stream_init (GOutputStream *stream)
 146 {
 147   stream->priv = g_output_stream_get_instance_private (stream);
 148 }
 149
 150 /**
 151  * g_output_stream_write:
 152  * @stream: a #GOutputStream.
 153  * @buffer: (array length=count) (element-type guint8): the buffer containing the data to write.
 154  * @count: the number of bytes to write
 155  * @cancellable: (allow-none): optional cancellable object
 156  * @error: location to store the error occurring, or %NULL to ignore
 157  *
 158  * Tries to write @count bytes from @buffer into the stream. Will block
 159  * during the operation.
 160  *
 161  * If count is 0, returns 0 and does nothing. A value of @count
 162  * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 163  *
 164  * On success, the number of bytes written to the stream is returned.
 165  * It is not an error if this is not the same as the requested size, as it
 166  * can happen e.g. on a partial I/O error, or if there is not enough
 167  * storage in the stream. All writes block until at least one byte
 168  * is written or an error occurs; 0 is never returned (unless
 169  * @count is 0).
 170  *
 171  * If @cancellable is not %NULL, then the operation can be cancelled by
 172  * triggering the cancellable object from another thread. If the operation
 173  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
 174  * operation was partially finished when the operation was cancelled the
 175  * partial result will be returned, without an error.
 176  *
 177  * On error -1 is returned and @error is set accordingly.
 178  *
 179  * Virtual: write_fn
 180  *
 181  * Return value: Number of bytes written, or -1 on error
 182  **/
 183 gssize
 184 g_output_stream_write (GOutputStream  *stream,
 185                        const void     *buffer,
 186                        gsize           count,
 187                        GCancellable   *cancellable,
 188                        GError        **error)
 189 {
 190   GOutputStreamClass *class;
 191   gssize res;
 192
 193   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), -1);
 194   g_return_val_if_fail (buffer != NULL, 0);
 195
 196   if (count == 0)
 197     return 0;
 198
 199   if (((gssize) count) < 0)
 200     {
 201       g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 202                    _("Too large count value passed to %s"), G_STRFUNC);
 203       return -1;
 204     }
 205
 206   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 207
 208   if (class->write_fn == NULL)
 209     {
 210       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
 211                            _("Output stream doesn't implement write"));
 212       return -1;
 213     }
 214
 215   if (!g_output_stream_set_pending (stream, error))
 216     return -1;
 217
 218   if (cancellable)
 219     g_cancellable_push_current (cancellable);
 220
 221   res = class->write_fn (stream, buffer, count, cancellable, error);
 222
 223   if (cancellable)
 224     g_cancellable_pop_current (cancellable);
 225
 226   g_output_stream_clear_pending (stream);
 227
 228   return res;
 229 }
 230
 231 /**
 232  * g_output_stream_write_all:
 233  * @stream: a #GOutputStream.
 234  * @buffer: (array length=count) (element-type guint8): the buffer containing the data to write.
 235  * @count: the number of bytes to write
 236  * @bytes_written: (out): location to store the number of bytes that was
 237  *     written to the stream
 238  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 239  * @error: location to store the error occurring, or %NULL to ignore
 240  *
 241  * Tries to write @count bytes from @buffer into the stream. Will block
 242  * during the operation.
 243  *
 244  * This function is similar to g_output_stream_write(), except it tries to
 245  * write as many bytes as requested, only stopping on an error.
 246  *
 247  * On a successful write of @count bytes, %TRUE is returned, and @bytes_written
 248  * is set to @count.
 249  *
 250  * If there is an error during the operation %FALSE is returned and @error
 251  * is set to indicate the error status, @bytes_written is updated to contain
 252  * the number of bytes written into the stream before the error occurred.
 253  *
 254  * Return value: %TRUE on success, %FALSE if there was an error
 255  **/
 256 gboolean
 257 g_output_stream_write_all (GOutputStream  *stream,
 258                            const void     *buffer,
 259                            gsize           count,
 260                            gsize          *bytes_written,
 261                            GCancellable   *cancellable,
 262                            GError        **error)
 263 {
 264   gsize _bytes_written;
 265   gssize res;
 266
 267   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 268   g_return_val_if_fail (buffer != NULL, FALSE);
 269
 270   _bytes_written = 0;
 271   while (_bytes_written < count)
 272     {
 273       res = g_output_stream_write (stream, (char *)buffer + _bytes_written, count - _bytes_written,
 274                                    cancellable, error);
 275       if (res == -1)
 276         {
 277           if (bytes_written)
 278             *bytes_written = _bytes_written;
 279           return FALSE;
 280         }
 281
 282       if (res == 0)
 283         g_warning ("Write returned zero without error");
 284
 285       _bytes_written += res;
 286     }
 287
 288   if (bytes_written)
 289     *bytes_written = _bytes_written;
 290
 291   return TRUE;
 292 }
 293
 294 /**
 295  * g_output_stream_printf:
 296  * @stream: a #GOutputStream.
 297  * @bytes_written: (out): location to store the number of bytes that was
 298  *     written to the stream
 299  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 300  * @error: location to store the error occurring, or %NULL to ignore
 301  * @format: the format string. See the printf() documentation
 302  * @...: the parameters to insert into the format string
 303  *
 304  * This is a utility function around g_output_stream_write_all(). It
 305  * uses g_strdup_vprintf() to turn @format and @... into a string that
 306  * is then written to @stream.
 307  *
 308  * See the documentation of g_output_stream_write_all() about the
 309  * behavior of the actual write operation.
 310  *
 311  * Note that partial writes cannot be properly checked with this
 312  * function due to the variable length of the written string, if you
 313  * need precise control over partial write failures, you need to
 314  * create you own printf()-like wrapper around g_output_stream_write()
 315  * or g_output_stream_write_all().
 316  *
 317  * Since: 2.40
 318  *
 319  * Return value: %TRUE on success, %FALSE if there was an error
 320  **/
 321 gboolean
 322 g_output_stream_printf (GOutputStream  *stream,
 323                         gsize          *bytes_written,
 324                         GCancellable   *cancellable,
 325                         GError        **error,
 326                         const gchar    *format,
 327                         ...)
 328 {
 329   va_list  args;
 330   gboolean success;
 331
 332   va_start (args, format);
 333   success = g_output_stream_vprintf (stream, bytes_written, cancellable,
 334                                      error, format, args);
 335   va_end (args);
 336
 337   return success;
 338 }
 339
 340 /**
 341  * g_output_stream_vprintf:
 342  * @stream: a #GOutputStream.
 343  * @bytes_written: (out): location to store the number of bytes that was
 344  *     written to the stream
 345  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 346  * @error: location to store the error occurring, or %NULL to ignore
 347  * @format: the format string. See the printf() documentation
 348  * @args: the parameters to insert into the format string
 349  *
 350  * This is a utility function around g_output_stream_write_all(). It
 351  * uses g_strdup_vprintf() to turn @format and @args into a string that
 352  * is then written to @stream.
 353  *
 354  * See the documentation of g_output_stream_write_all() about the
 355  * behavior of the actual write operation.
 356  *
 357  * Note that partial writes cannot be properly checked with this
 358  * function due to the variable length of the written string, if you
 359  * need precise control over partial write failures, you need to
 360  * create you own printf()-like wrapper around g_output_stream_write()
 361  * or g_output_stream_write_all().
 362  *
 363  * Since: 2.40
 364  *
 365  * Return value: %TRUE on success, %FALSE if there was an error
 366  **/
 367 gboolean
 368 g_output_stream_vprintf (GOutputStream  *stream,
 369                          gsize          *bytes_written,
 370                          GCancellable   *cancellable,
 371                          GError        **error,
 372                          const gchar    *format,
 373                          va_list         args)
 374 {
 375   gchar    *text;
 376   gboolean  success;
 377
 378   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 379   g_return_val_if_fail (cancellable == NULL || G_IS_CANCELLABLE (stream), FALSE);
 380   g_return_val_if_fail (error == NULL || *error == NULL, FALSE);
 381   g_return_val_if_fail (format != NULL, FALSE);
 382
 383   text = g_strdup_vprintf (format, args);
 384   success = g_output_stream_write_all (stream,
 385                                        text, strlen (text),
 386                                        bytes_written, cancellable, error);
 387   g_free (text);
 388
 389   return success;
 390 }
 391
 392 /**
 393  * g_output_stream_write_bytes:
 394  * @stream: a #GOutputStream.
 395  * @bytes: the #GBytes to write
 396  * @cancellable: (allow-none): optional cancellable object
 397  * @error: location to store the error occurring, or %NULL to ignore
 398  *
 399  * A wrapper function for g_output_stream_write() which takes a
 400  * #GBytes as input.  This can be more convenient for use by language
 401  * bindings or in other cases where the refcounted nature of #GBytes
 402  * is helpful over a bare pointer interface.
 403  *
 404  * However, note that this function <emphasis>may</emphasis> still
 405  * perform partial writes, just like g_output_stream_write().  If that
 406  * occurs, to continue writing, you will need to create a new #GBytes
 407  * containing just the remaining bytes, using
 408  * g_bytes_new_from_bytes().  Passing the same #GBytes instance
 409  * multiple times potentially can result in duplicated data in the
 410  * output stream.
 411  *
 412  * Return value: Number of bytes written, or -1 on error
 413  **/
 414 gssize
 415 g_output_stream_write_bytes (GOutputStream  *stream,
 416                              GBytes         *bytes,
 417                              GCancellable   *cancellable,
 418                              GError        **error)
 419 {
 420   gsize size;
 421   gconstpointer data;
 422
 423   data = g_bytes_get_data (bytes, &size);
 424
 425   return g_output_stream_write (stream,
 426                                 data, size,
 427                                 cancellable,
 428                                 error);
 429 }
 430
 431 /**
 432  * g_output_stream_flush:
 433  * @stream: a #GOutputStream.
 434  * @cancellable: (allow-none): optional cancellable object
 435  * @error: location to store the error occurring, or %NULL to ignore
 436  *
 437  * Forces a write of all user-space buffered data for the given
 438  * @stream. Will block during the operation. Closing the stream will
 439  * implicitly cause a flush.
 440  *
 441  * This function is optional for inherited classes.
 442  *
 443  * If @cancellable is not %NULL, then the operation can be cancelled by
 444  * triggering the cancellable object from another thread. If the operation
 445  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 446  *
 447  * Return value: %TRUE on success, %FALSE on error
 448  **/
 449 gboolean
 450 g_output_stream_flush (GOutputStream  *stream,
 451                        GCancellable   *cancellable,
 452                        GError        **error)
 453 {
 454   GOutputStreamClass *class;
 455   gboolean res;
 456
 457   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 458
 459   if (!g_output_stream_set_pending (stream, error))
 460     return FALSE;
 461
 462   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 463
 464   res = TRUE;
 465   if (class->flush)
 466     {
 467       if (cancellable)
 468         g_cancellable_push_current (cancellable);
 469
 470       res = class->flush (stream, cancellable, error);
 471
 472       if (cancellable)
 473         g_cancellable_pop_current (cancellable);
 474     }
 475
 476   g_output_stream_clear_pending (stream);
 477
 478   return res;
 479 }
 480
 481 /**
 482  * g_output_stream_splice:
 483  * @stream: a #GOutputStream.
 484  * @source: a #GInputStream.
 485  * @flags: a set of #GOutputStreamSpliceFlags.
 486  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 487  * @error: a #GError location to store the error occurring, or %NULL to
 488  * ignore.
 489  *
 490  * Splices an input stream into an output stream.
 491  *
 492  * Returns: a #gssize containing the size of the data spliced, or
 493  *     -1 if an error occurred. Note that if the number of bytes
 494  *     spliced is greater than %G_MAXSSIZE, then that will be
 495  *     returned, and there is no way to determine the actual number
 496  *     of bytes spliced.
 497  **/
 498 gssize
 499 g_output_stream_splice (GOutputStream             *stream,
 500                         GInputStream              *source,
 501                         GOutputStreamSpliceFlags   flags,
 502                         GCancellable              *cancellable,
 503                         GError                   **error)
 504 {
 505   GOutputStreamClass *class;
 506   gssize bytes_copied;
 507
 508   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), -1);
 509   g_return_val_if_fail (G_IS_INPUT_STREAM (source), -1);
 510
 511   if (g_input_stream_is_closed (source))
 512     {
 513       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
 514                            _("Source stream is already closed"));
 515       return -1;
 516     }
 517
 518   if (!g_output_stream_set_pending (stream, error))
 519     return -1;
 520
 521   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 522
 523   if (cancellable)
 524     g_cancellable_push_current (cancellable);
 525
 526   bytes_copied = class->splice (stream, source, flags, cancellable, error);
 527
 528   if (cancellable)
 529     g_cancellable_pop_current (cancellable);
 530
 531   g_output_stream_clear_pending (stream);
 532
 533   return bytes_copied;
 534 }
 535
 536 static gssize
 537 g_output_stream_real_splice (GOutputStream             *stream,
 538                              GInputStream              *source,
 539                              GOutputStreamSpliceFlags   flags,
 540                              GCancellable              *cancellable,
 541                              GError                   **error)
 542 {
 543   GOutputStreamClass *class = G_OUTPUT_STREAM_GET_CLASS (stream);
 544   gssize n_read, n_written;
 545   gsize bytes_copied;
 546   char buffer[8192], *p;
 547   gboolean res;
 548
 549   bytes_copied = 0;
 550   if (class->write_fn == NULL)
 551     {
 552       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
 553                            _("Output stream doesn't implement write"));
 554       res = FALSE;
 555       goto notsupported;
 556     }
 557
 558   res = TRUE;
 559   do
 560     {
 561       n_read = g_input_stream_read (source, buffer, sizeof (buffer), cancellable, error);
 562       if (n_read == -1)
 563         {
 564           res = FALSE;
 565           break;
 566         }
 567
 568       if (n_read == 0)
 569         break;
 570
 571       p = buffer;
 572       while (n_read > 0)
 573         {
 574           n_written = class->write_fn (stream, p, n_read, cancellable, error);
 575           if (n_written == -1)
 576             {
 577               res = FALSE;
 578               break;
 579             }
 580
 581           p += n_written;
 582           n_read -= n_written;
 583           bytes_copied += n_written;
 584         }
 585
 586       if (bytes_copied > G_MAXSSIZE)
 587         bytes_copied = G_MAXSSIZE;
 588     }
 589   while (res);
 590
 591  notsupported:
 592   if (!res)
 593     error = NULL; /* Ignore further errors */
 594
 595   if (flags & G_OUTPUT_STREAM_SPLICE_CLOSE_SOURCE)
 596     {
 597       /* Don't care about errors in source here */
 598       g_input_stream_close (source, cancellable, NULL);
 599     }
 600
 601   if (flags & G_OUTPUT_STREAM_SPLICE_CLOSE_TARGET)
 602     {
 603       /* But write errors on close are bad! */
 604       res = g_output_stream_internal_close (stream, cancellable, error);
 605     }
 606
 607   if (res)
 608     return bytes_copied;
 609
 610   return -1;
 611 }
 612
 613 /* Must always be called inside
 614  * g_output_stream_set_pending()/g_output_stream_clear_pending(). */
 615 static gboolean
 616 g_output_stream_internal_close (GOutputStream  *stream,
 617                                 GCancellable   *cancellable,
 618                                 GError        **error)
 619 {
 620   GOutputStreamClass *class;
 621   gboolean res;
 622
 623   if (stream->priv->closed)
 624     return TRUE;
 625
 626   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 627
 628   stream->priv->closing = TRUE;
 629
 630   if (cancellable)
 631     g_cancellable_push_current (cancellable);
 632
 633   if (class->flush)
 634     res = class->flush (stream, cancellable, error);
 635   else
 636     res = TRUE;
 637
 638   if (!res)
 639     {
 640       /* flushing caused the error that we want to return,
 641        * but we still want to close the underlying stream if possible
 642        */
 643       if (class->close_fn)
 644         class->close_fn (stream, cancellable, NULL);
 645     }
 646   else
 647     {
 648       res = TRUE;
 649       if (class->close_fn)
 650         res = class->close_fn (stream, cancellable, error);
 651     }
 652
 653   if (cancellable)
 654     g_cancellable_pop_current (cancellable);
 655
 656   stream->priv->closing = FALSE;
 657   stream->priv->closed = TRUE;
 658
 659   return res;
 660 }
 661
 662 /**
 663  * g_output_stream_close:
 664  * @stream: A #GOutputStream.
 665  * @cancellable: (allow-none): optional cancellable object
 666  * @error: location to store the error occurring, or %NULL to ignore
 667  *
 668  * Closes the stream, releasing resources related to it.
 669  *
 670  * Once the stream is closed, all other operations will return %G_IO_ERROR_CLOSED.
 671  * Closing a stream multiple times will not return an error.
 672  *
 673  * Closing a stream will automatically flush any outstanding buffers in the
 674  * stream.
 675  *
 676  * Streams will be automatically closed when the last reference
 677  * is dropped, but you might want to call this function to make sure
 678  * resources are released as early as possible.
 679  *
 680  * Some streams might keep the backing store of the stream (e.g. a file descriptor)
 681  * open after the stream is closed. See the documentation for the individual
 682  * stream for details.
 683  *
 684  * On failure the first error that happened will be reported, but the close
 685  * operation will finish as much as possible. A stream that failed to
 686  * close will still return %G_IO_ERROR_CLOSED for all operations. Still, it
 687  * is important to check and report the error to the user, otherwise
 688  * there might be a loss of data as all data might not be written.
 689  *
 690  * If @cancellable is not %NULL, then the operation can be cancelled by
 691  * triggering the cancellable object from another thread. If the operation
 692  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 693  * Cancelling a close will still leave the stream closed, but there some streams
 694  * can use a faster close that doesn't block to e.g. check errors. On
 695  * cancellation (as with any error) there is no guarantee that all written
 696  * data will reach the target.
 697  *
 698  * Return value: %TRUE on success, %FALSE on failure
 699  **/
 700 gboolean
 701 g_output_stream_close (GOutputStream  *stream,
 702                        GCancellable   *cancellable,
 703                        GError        **error)
 704 {
 705   gboolean res;
 706
 707   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 708
 709   if (stream->priv->closed)
 710     return TRUE;
 711
 712   if (!g_output_stream_set_pending (stream, error))
 713     return FALSE;
 714
 715   res = g_output_stream_internal_close (stream, cancellable, error);
 716
 717   g_output_stream_clear_pending (stream);
 718
 719   return res;
 720 }
 721
 722 static void
 723 async_ready_write_callback_wrapper (GObject      *source_object,
 724                                     GAsyncResult *res,
 725                                     gpointer      user_data)
 726 {
 727   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
 728   GOutputStreamClass *class;
 729   GTask *task = user_data;
 730   gssize nwrote;
 731   GError *error = NULL;
 732
 733   g_output_stream_clear_pending (stream);
 734
 735   if (g_async_result_legacy_propagate_error (res, &error))
 736     nwrote = -1;
 737   else
 738     {
 739       class = G_OUTPUT_STREAM_GET_CLASS (stream);
 740       nwrote = class->write_finish (stream, res, &error);
 741     }
 742
 743   if (nwrote >= 0)
 744     g_task_return_int (task, nwrote);
 745   else
 746     g_task_return_error (task, error);
 747   g_object_unref (task);
 748 }
 749
 750 /**
 751  * g_output_stream_write_async:
 752  * @stream: A #GOutputStream.
 753  * @buffer: (array length=count) (element-type guint8): the buffer containing the data to write.
 754  * @count: the number of bytes to write
 755  * @io_priority: the io priority of the request.
 756  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 757  * @callback: (scope async): callback to call when the request is satisfied
 758  * @user_data: (closure): the data to pass to callback function
 759  *
 760  * Request an asynchronous write of @count bytes from @buffer into
 761  * the stream. When the operation is finished @callback will be called.
 762  * You can then call g_output_stream_write_finish() to get the result of the
 763  * operation.
 764  *
 765  * During an async request no other sync and async calls are allowed,
 766  * and will result in %G_IO_ERROR_PENDING errors.
 767  *
 768  * A value of @count larger than %G_MAXSSIZE will cause a
 769  * %G_IO_ERROR_INVALID_ARGUMENT error.
 770  *
 771  * On success, the number of bytes written will be passed to the
 772  * @callback. It is not an error if this is not the same as the
 773  * requested size, as it can happen e.g. on a partial I/O error,
 774  * but generally we try to write as many bytes as requested.
 775  *
 776  * You are guaranteed that this method will never fail with
 777  * %G_IO_ERROR_WOULD_BLOCK - if @stream can't accept more data, the
 778  * method will just wait until this changes.
 779  *
 780  * Any outstanding I/O request with higher priority (lower numerical
 781  * value) will be executed before an outstanding request with lower
 782  * priority. Default priority is %G_PRIORITY_DEFAULT.
 783  *
 784  * The asyncronous methods have a default fallback that uses threads
 785  * to implement asynchronicity, so they are optional for inheriting
 786  * classes. However, if you override one you must override all.
 787  *
 788  * For the synchronous, blocking version of this function, see
 789  * g_output_stream_write().
 790  **/
 791 void
 792 g_output_stream_write_async (GOutputStream       *stream,
 793                              const void          *buffer,
 794                              gsize                count,
 795                              int                  io_priority,
 796                              GCancellable        *cancellable,
 797                              GAsyncReadyCallback  callback,
 798                              gpointer             user_data)
 799 {
 800   GOutputStreamClass *class;
 801   GError *error = NULL;
 802   GTask *task;
 803
 804   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
 805   g_return_if_fail (buffer != NULL);
 806
 807   task = g_task_new (stream, cancellable, callback, user_data);
 808   g_task_set_source_tag (task, g_output_stream_write_async);
 809   g_task_set_priority (task, io_priority);
 810
 811   if (count == 0)
 812     {
 813       g_task_return_int (task, 0);
 814       g_object_unref (task);
 815       return;
 816     }
 817
 818   if (((gssize) count) < 0)
 819     {
 820       g_task_return_new_error (task, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 821                                _("Too large count value passed to %s"),
 822                                G_STRFUNC);
 823       g_object_unref (task);
 824       return;
 825     }
 826
 827   if (!g_output_stream_set_pending (stream, &error))
 828     {
 829       g_task_return_error (task, error);
 830       g_object_unref (task);
 831       return;
 832     }
 833
 834   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 835
 836   class->write_async (stream, buffer, count, io_priority, cancellable,
 837                       async_ready_write_callback_wrapper, task);
 838 }
 839
 840 /**
 841  * g_output_stream_write_finish:
 842  * @stream: a #GOutputStream.
 843  * @result: a #GAsyncResult.
 844  * @error: a #GError location to store the error occurring, or %NULL to
 845  * ignore.
 846  *
 847  * Finishes a stream write operation.
 848  *
 849  * Returns: a #gssize containing the number of bytes written to the stream.
 850  **/
 851 gssize
 852 g_output_stream_write_finish (GOutputStream  *stream,
 853                               GAsyncResult   *result,
 854                               GError        **error)
 855 {
 856   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 857   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
 858   g_return_val_if_fail (g_async_result_is_tagged (result, g_output_stream_write_async), FALSE);
 859
 860   /* @result is always the GTask created by g_output_stream_write_async();
 861    * we called class->write_finish() from async_ready_write_callback_wrapper.
 862    */
 863   return g_task_propagate_int (G_TASK (result), error);
 864 }
 865
 866 static void
 867 write_bytes_callback (GObject      *stream,
 868                       GAsyncResult *result,
 869                       gpointer      user_data)
 870 {
 871   GTask *task = user_data;
 872   GError *error = NULL;
 873   gssize nwrote;
 874
 875   nwrote = g_output_stream_write_finish (G_OUTPUT_STREAM (stream),
 876                                          result, &error);
 877   if (nwrote == -1)
 878     g_task_return_error (task, error);
 879   else
 880     g_task_return_int (task, nwrote);
 881   g_object_unref (task);
 882 }
 883
 884 /**
 885  * g_output_stream_write_bytes_async:
 886  * @stream: A #GOutputStream.
 887  * @bytes: The bytes to write
 888  * @io_priority: the io priority of the request.
 889  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 890  * @callback: (scope async): callback to call when the request is satisfied
 891  * @user_data: (closure): the data to pass to callback function
 892  *
 893  * This function is similar to g_output_stream_write_async(), but
 894  * takes a #GBytes as input.  Due to the refcounted nature of #GBytes,
 895  * this allows the stream to avoid taking a copy of the data.
 896  *
 897  * However, note that this function <emphasis>may</emphasis> still
 898  * perform partial writes, just like g_output_stream_write_async().
 899  * If that occurs, to continue writing, you will need to create a new
 900  * #GBytes containing just the remaining bytes, using
 901  * g_bytes_new_from_bytes().  Passing the same #GBytes instance
 902  * multiple times potentially can result in duplicated data in the
 903  * output stream.
 904  *
 905  * For the synchronous, blocking version of this function, see
 906  * g_output_stream_write_bytes().
 907  **/
 908 void
 909 g_output_stream_write_bytes_async (GOutputStream       *stream,
 910                                    GBytes              *bytes,
 911                                    int                  io_priority,
 912                                    GCancellable        *cancellable,
 913                                    GAsyncReadyCallback  callback,
 914                                    gpointer             user_data)
 915 {
 916   GTask *task;
 917   gsize size;
 918   gconstpointer data;
 919
 920   data = g_bytes_get_data (bytes, &size);
 921
 922   task = g_task_new (stream, cancellable, callback, user_data);
 923   g_task_set_task_data (task, g_bytes_ref (bytes),
 924                         (GDestroyNotify) g_bytes_unref);
 925
 926   g_output_stream_write_async (stream,
 927                                data, size,
 928                                io_priority,
 929                                cancellable,
 930                                write_bytes_callback,
 931                                task);
 932 }
 933
 934 /**
 935  * g_output_stream_write_bytes_finish:
 936  * @stream: a #GOutputStream.
 937  * @result: a #GAsyncResult.
 938  * @error: a #GError location to store the error occurring, or %NULL to
 939  * ignore.
 940  *
 941  * Finishes a stream write-from-#GBytes operation.
 942  *
 943  * Returns: a #gssize containing the number of bytes written to the stream.
 944  **/
 945 gssize
 946 g_output_stream_write_bytes_finish (GOutputStream  *stream,
 947                                     GAsyncResult   *result,
 948                                     GError        **error)
 949 {
 950   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), -1);
 951   g_return_val_if_fail (g_task_is_valid (result, stream), -1);
 952
 953   return g_task_propagate_int (G_TASK (result), error);
 954 }
 955
 956 static void
 957 async_ready_splice_callback_wrapper (GObject      *source_object,
 958                                      GAsyncResult *res,
 959                                      gpointer     _data)
 960 {
 961   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
 962   GOutputStreamClass *class;
 963   GTask *task = _data;
 964   gssize nspliced;
 965   GError *error = NULL;
 966
 967   g_output_stream_clear_pending (stream);
 968
 969   if (g_async_result_legacy_propagate_error (res, &error))
 970     nspliced = -1;
 971   else
 972     {
 973       class = G_OUTPUT_STREAM_GET_CLASS (stream);
 974       nspliced = class->splice_finish (stream, res, &error);
 975     }
 976
 977   if (nspliced >= 0)
 978     g_task_return_int (task, nspliced);
 979   else
 980     g_task_return_error (task, error);
 981   g_object_unref (task);
 982 }
 983
 984 /**
 985  * g_output_stream_splice_async:
 986  * @stream: a #GOutputStream.
 987  * @source: a #GInputStream.
 988  * @flags: a set of #GOutputStreamSpliceFlags.
 989  * @io_priority: the io priority of the request.
 990  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 991  * @callback: (scope async): a #GAsyncReadyCallback.
 992  * @user_data: (closure): user data passed to @callback.
 993  *
 994  * Splices a stream asynchronously.
 995  * When the operation is finished @callback will be called.
 996  * You can then call g_output_stream_splice_finish() to get the
 997  * result of the operation.
 998  *
 999  * For the synchronous, blocking version of this function, see
1000  * g_output_stream_splice().
1001  **/
1002 void
1003 g_output_stream_splice_async (GOutputStream            *stream,
1004                               GInputStream             *source,
1005                               GOutputStreamSpliceFlags  flags,
1006                               int                       io_priority,
1007                               GCancellable             *cancellable,
1008                               GAsyncReadyCallback       callback,
1009                               gpointer                  user_data)
1010 {
1011   GOutputStreamClass *class;
1012   GTask *task;
1013   GError *error = NULL;
1014
1015   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
1016   g_return_if_fail (G_IS_INPUT_STREAM (source));
1017
1018   task = g_task_new (stream, cancellable, callback, user_data);
1019   g_task_set_source_tag (task, g_output_stream_splice_async);
1020   g_task_set_priority (task, io_priority);
1021   g_task_set_task_data (task, g_object_ref (source), g_object_unref);
1022
1023   if (g_input_stream_is_closed (source))
1024     {
1025       g_task_return_new_error (task,
1026                                G_IO_ERROR, G_IO_ERROR_CLOSED,
1027                                _("Source stream is already closed"));
1028       g_object_unref (task);
1029       return;
1030     }
1031
1032   if (!g_output_stream_set_pending (stream, &error))
1033     {
1034       g_task_return_error (task, error);
1035       g_object_unref (task);
1036       return;
1037     }
1038
1039   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1040
1041   class->splice_async (stream, source, flags, io_priority, cancellable,
1042                        async_ready_splice_callback_wrapper, task);
1043 }
1044
1045 /**
1046  * g_output_stream_splice_finish:
1047  * @stream: a #GOutputStream.
1048  * @result: a #GAsyncResult.
1049  * @error: a #GError location to store the error occurring, or %NULL to
1050  * ignore.
1051  *
1052  * Finishes an asynchronous stream splice operation.
1053  *
1054  * Returns: a #gssize of the number of bytes spliced. Note that if the
1055  *     number of bytes spliced is greater than %G_MAXSSIZE, then that
1056  *     will be returned, and there is no way to determine the actual
1057  *     number of bytes spliced.
1058  **/
1059 gssize
1060 g_output_stream_splice_finish (GOutputStream  *stream,
1061                                GAsyncResult   *result,
1062                                GError        **error)
1063 {
1064   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1065   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1066   g_return_val_if_fail (g_async_result_is_tagged (result, g_output_stream_splice_async), FALSE);
1067
1068   /* @result is always the GTask created by g_output_stream_splice_async();
1069    * we called class->splice_finish() from async_ready_splice_callback_wrapper.
1070    */
1071   return g_task_propagate_int (G_TASK (result), error);
1072 }
1073
1074 static void
1075 async_ready_flush_callback_wrapper (GObject      *source_object,
1076                                     GAsyncResult *res,
1077                                     gpointer      user_data)
1078 {
1079   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
1080   GOutputStreamClass *class;
1081   GTask *task = user_data;
1082   gboolean flushed;
1083   GError *error = NULL;
1084
1085   g_output_stream_clear_pending (stream);
1086
1087   if (g_async_result_legacy_propagate_error (res, &error))
1088     flushed = FALSE;
1089   else
1090     {
1091       class = G_OUTPUT_STREAM_GET_CLASS (stream);
1092       flushed = class->flush_finish (stream, res, &error);
1093     }
1094
1095   if (flushed)
1096     g_task_return_boolean (task, TRUE);
1097   else
1098     g_task_return_error (task, error);
1099   g_object_unref (task);
1100 }
1101
1102 /**
1103  * g_output_stream_flush_async:
1104  * @stream: a #GOutputStream.
1105  * @io_priority: the io priority of the request.
1106  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
1107  * @callback: (scope async): a #GAsyncReadyCallback to call when the request is satisfied
1108  * @user_data: (closure): the data to pass to callback function
1109  *
1110  * Forces an asynchronous write of all user-space buffered data for
1111  * the given @stream.
1112  * For behaviour details see g_output_stream_flush().
1113  *
1114  * When the operation is finished @callback will be
1115  * called. You can then call g_output_stream_flush_finish() to get the
1116  * result of the operation.
1117  **/
1118 void
1119 g_output_stream_flush_async (GOutputStream       *stream,
1120                              int                  io_priority,
1121                              GCancellable        *cancellable,
1122                              GAsyncReadyCallback  callback,
1123                              gpointer             user_data)
1124 {
1125   GOutputStreamClass *class;
1126   GTask *task;
1127   GError *error = NULL;
1128
1129   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
1130
1131   task = g_task_new (stream, cancellable, callback, user_data);
1132   g_task_set_source_tag (task, g_output_stream_flush_async);
1133   g_task_set_priority (task, io_priority);
1134
1135   if (!g_output_stream_set_pending (stream, &error))
1136     {
1137       g_task_return_error (task, error);
1138       g_object_unref (task);
1139       return;
1140     }
1141
1142   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1143
1144   if (class->flush_async == NULL)
1145     {
1146       g_task_return_boolean (task, TRUE);
1147       g_object_unref (task);
1148       return;
1149     }
1150
1151   class->flush_async (stream, io_priority, cancellable,
1152                       async_ready_flush_callback_wrapper, task);
1153 }
1154
1155 /**
1156  * g_output_stream_flush_finish:
1157  * @stream: a #GOutputStream.
1158  * @result: a GAsyncResult.
1159  * @error: a #GError location to store the error occurring, or %NULL to
1160  * ignore.
1161  *
1162  * Finishes flushing an output stream.
1163  *
1164  * Returns: %TRUE if flush operation succeeded, %FALSE otherwise.
1165  **/
1166 gboolean
1167 g_output_stream_flush_finish (GOutputStream  *stream,
1168                               GAsyncResult   *result,
1169                               GError        **error)
1170 {
1171   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1172   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1173   g_return_val_if_fail (g_async_result_is_tagged (result, g_output_stream_flush_async), FALSE);
1174
1175   /* @result is always the GTask created by g_output_stream_flush_async();
1176    * we called class->flush_finish() from async_ready_flush_callback_wrapper.
1177    */
1178   return g_task_propagate_boolean (G_TASK (result), error);
1179 }
1180
1181
1182 static void
1183 async_ready_close_callback_wrapper (GObject      *source_object,
1184                                     GAsyncResult *res,
1185                                     gpointer      user_data)
1186 {
1187   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
1188   GOutputStreamClass *class;
1189   GTask *task = user_data;
1190   GError *error = g_task_get_task_data (task);
1191
1192   stream->priv->closing = FALSE;
1193   stream->priv->closed = TRUE;
1194
1195   if (!error && !g_async_result_legacy_propagate_error (res, &error))
1196     {
1197       class = G_OUTPUT_STREAM_GET_CLASS (stream);
1198
1199       class->close_finish (stream, res,
1200                            error ? NULL : &error);
1201     }
1202
1203   if (error != NULL)
1204     g_task_return_error (task, error);
1205   else
1206     g_task_return_boolean (task, TRUE);
1207   g_object_unref (task);
1208 }
1209
1210 static void
1211 async_ready_close_flushed_callback_wrapper (GObject      *source_object,
1212                                             GAsyncResult *res,
1213                                             gpointer      user_data)
1214 {
1215   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
1216   GOutputStreamClass *class;
1217   GTask *task = user_data;
1218   GError *error = NULL;
1219
1220   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1221
1222   if (!g_async_result_legacy_propagate_error (res, &error))
1223     {
1224       class->flush_finish (stream, res, &error);
1225     }
1226
1227   /* propagate the possible error */
1228   if (error)
1229     g_task_set_task_data (task, error, NULL);
1230
1231   /* we still close, even if there was a flush error */
1232   class->close_async (stream,
1233                       g_task_get_priority (task),
1234                       g_task_get_cancellable (task),
1235                       async_ready_close_callback_wrapper, task);
1236 }
1237
1238 static void
1239 real_close_async_cb (GObject      *source_object,
1240                      GAsyncResult *res,
1241                      gpointer      user_data)
1242 {
1243   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
1244   GTask *task = user_data;
1245   GError *error = NULL;
1246   gboolean ret;
1247
1248   g_output_stream_clear_pending (stream);
1249
1250   ret = g_output_stream_internal_close_finish (stream, res, &error);
1251
1252   if (error != NULL)
1253     g_task_return_error (task, error);
1254   else
1255     g_task_return_boolean (task, ret);
1256
1257   g_object_unref (task);
1258 }
1259
1260 /**
1261  * g_output_stream_close_async:
1262  * @stream: A #GOutputStream.
1263  * @io_priority: the io priority of the request.
1264  * @cancellable: (allow-none): optional cancellable object
1265  * @callback: (scope async): callback to call when the request is satisfied
1266  * @user_data: (closure): the data to pass to callback function
1267  *
1268  * Requests an asynchronous close of the stream, releasing resources
1269  * related to it. When the operation is finished @callback will be
1270  * called. You can then call g_output_stream_close_finish() to get
1271  * the result of the operation.
1272  *
1273  * For behaviour details see g_output_stream_close().
1274  *
1275  * The asyncronous methods have a default fallback that uses threads
1276  * to implement asynchronicity, so they are optional for inheriting
1277  * classes. However, if you override one you must override all.
1278  **/
1279 void
1280 g_output_stream_close_async (GOutputStream       *stream,
1281                              int                  io_priority,
1282                              GCancellable        *cancellable,
1283                              GAsyncReadyCallback  callback,
1284                              gpointer             user_data)
1285 {
1286   GTask *task;
1287   GError *error = NULL;
1288
1289   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
1290
1291   task = g_task_new (stream, cancellable, callback, user_data);
1292   g_task_set_source_tag (task, g_output_stream_close_async);
1293   g_task_set_priority (task, io_priority);
1294
1295   if (!g_output_stream_set_pending (stream, &error))
1296     {
1297       g_task_return_error (task, error);
1298       g_object_unref (task);
1299       return;
1300     }
1301
1302   g_output_stream_internal_close_async (stream, io_priority, cancellable,
1303                                         real_close_async_cb, task);
1304 }
1305
1306 /* Must always be called inside
1307  * g_output_stream_set_pending()/g_output_stream_clear_pending().
1308  */
1309 void
1310 g_output_stream_internal_close_async (GOutputStream       *stream,
1311                                       int                  io_priority,
1312                                       GCancellable        *cancellable,
1313                                       GAsyncReadyCallback  callback,
1314                                       gpointer             user_data)
1315 {
1316   GOutputStreamClass *class;
1317   GTask *task;
1318
1319   task = g_task_new (stream, cancellable, callback, user_data);
1320   g_task_set_source_tag (task, g_output_stream_internal_close_async);
1321   g_task_set_priority (task, io_priority);
1322
1323   if (stream->priv->closed)
1324     {
1325       g_task_return_boolean (task, TRUE);
1326       g_object_unref (task);
1327       return;
1328     }
1329
1330   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1331   stream->priv->closing = TRUE;
1332
1333   /* Call close_async directly if there is no need to flush, or if the flush
1334      can be done sync (in the output stream async close thread) */
1335   if (class->flush_async == NULL ||
1336       (class->flush_async == g_output_stream_real_flush_async &&
1337        (class->flush == NULL || class->close_async == g_output_stream_real_close_async)))
1338     {
1339       class->close_async (stream, io_priority, cancellable,
1340                           async_ready_close_callback_wrapper, task);
1341     }
1342   else
1343     {
1344       /* First do an async flush, then do the async close in the callback
1345          wrapper (see async_ready_close_flushed_callback_wrapper) */
1346       class->flush_async (stream, io_priority, cancellable,
1347                           async_ready_close_flushed_callback_wrapper, task);
1348     }
1349 }
1350
1351 static gboolean
1352 g_output_stream_internal_close_finish (GOutputStream  *stream,
1353                                        GAsyncResult   *result,
1354                                        GError        **error)
1355 {
1356   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1357   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1358   g_return_val_if_fail (g_async_result_is_tagged (result, g_output_stream_internal_close_async), FALSE);
1359
1360   return g_task_propagate_boolean (G_TASK (result), error);
1361 }
1362
1363 /**
1364  * g_output_stream_close_finish:
1365  * @stream: a #GOutputStream.
1366  * @result: a #GAsyncResult.
1367  * @error: a #GError location to store the error occurring, or %NULL to
1368  * ignore.
1369  *
1370  * Closes an output stream.
1371  *
1372  * Returns: %TRUE if stream was successfully closed, %FALSE otherwise.
1373  **/
1374 gboolean
1375 g_output_stream_close_finish (GOutputStream  *stream,
1376                               GAsyncResult   *result,
1377                               GError        **error)
1378 {
1379   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1380   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1381   g_return_val_if_fail (g_async_result_is_tagged (result, g_output_stream_close_async), FALSE);
1382
1383   /* @result is always the GTask created by g_output_stream_close_async();
1384    * we called class->close_finish() from async_ready_close_callback_wrapper.
1385    */
1386   return g_task_propagate_boolean (G_TASK (result), error);
1387 }
1388
1389 /**
1390  * g_output_stream_is_closed:
1391  * @stream: a #GOutputStream.
1392  *
1393  * Checks if an output stream has already been closed.
1394  *
1395  * Returns: %TRUE if @stream is closed. %FALSE otherwise.
1396  **/
1397 gboolean
1398 g_output_stream_is_closed (GOutputStream *stream)
1399 {
1400   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), TRUE);
1401
1402   return stream->priv->closed;
1403 }
1404
1405 /**
1406  * g_output_stream_is_closing:
1407  * @stream: a #GOutputStream.
1408  *
1409  * Checks if an output stream is being closed. This can be
1410  * used inside e.g. a flush implementation to see if the
1411  * flush (or other i/o operation) is called from within
1412  * the closing operation.
1413  *
1414  * Returns: %TRUE if @stream is being closed. %FALSE otherwise.
1415  *
1416  * Since: 2.24
1417  **/
1418 gboolean
1419 g_output_stream_is_closing (GOutputStream *stream)
1420 {
1421   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), TRUE);
1422
1423   return stream->priv->closing;
1424 }
1425
1426 /**
1427  * g_output_stream_has_pending:
1428  * @stream: a #GOutputStream.
1429  *
1430  * Checks if an ouput stream has pending actions.
1431  *
1432  * Returns: %TRUE if @stream has pending actions.
1433  **/
1434 gboolean
1435 g_output_stream_has_pending (GOutputStream *stream)
1436 {
1437   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1438
1439   return stream->priv->pending;
1440 }
1441
1442 /**
1443  * g_output_stream_set_pending:
1444  * @stream: a #GOutputStream.
1445  * @error: a #GError location to store the error occurring, or %NULL to
1446  * ignore.
1447  *
1448  * Sets @stream to have actions pending. If the pending flag is
1449  * already set or @stream is closed, it will return %FALSE and set
1450  * @error.
1451  *
1452  * Return value: %TRUE if pending was previously unset and is now set.
1453  **/
1454 gboolean
1455 g_output_stream_set_pending (GOutputStream *stream,
1456                              GError **error)
1457 {
1458   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1459
1460   if (stream->priv->closed)
1461     {
1462       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
1463                            _("Stream is already closed"));
1464       return FALSE;
1465     }
1466
1467   if (stream->priv->pending)
1468     {
1469       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
1470                            /* Translators: This is an error you get if there is
1471                             * already an operation running against this stream when
1472                             * you try to start one */
1473                            _("Stream has outstanding operation"));
1474       return FALSE;
1475     }
1476
1477   stream->priv->pending = TRUE;
1478   return TRUE;
1479 }
1480
1481 /**
1482  * g_output_stream_clear_pending:
1483  * @stream: output stream
1484  *
1485  * Clears the pending flag on @stream.
1486  **/
1487 void
1488 g_output_stream_clear_pending (GOutputStream *stream)
1489 {
1490   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
1491
1492   stream->priv->pending = FALSE;
1493 }
1494
1495 /**
1496  * g_output_stream_async_write_is_via_threads:
1497  * @stream: a #GOutputStream.
1498  *
1499  * Checks if an ouput stream's write_async function uses threads.
1500  *
1501  * Returns: %TRUE if @stream's write_async function uses threads.
1502  **/
1503 gboolean
1504 g_output_stream_async_write_is_via_threads (GOutputStream *stream)
1505 {
1506   GOutputStreamClass *class;
1507
1508   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1509
1510   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1511
1512   return (class->write_async == g_output_stream_real_write_async &&
1513       !(G_IS_POLLABLE_OUTPUT_STREAM (stream) &&
1514         g_pollable_output_stream_can_poll (G_POLLABLE_OUTPUT_STREAM (stream))));
1515 }
1516
1517
1518 /********************************************
1519  *   Default implementation of async ops    *
1520  ********************************************/
1521
1522 typedef struct {
1523   const void         *buffer;
1524   gsize               count_requested;
1525   gssize              count_written;
1526 } WriteData;
1527
1528 static void
1529 free_write_data (WriteData *op)
1530 {
1531   g_slice_free (WriteData, op);
1532 }
1533
1534 static void
1535 write_async_thread (GTask        *task,
1536                     gpointer      source_object,
1537                     gpointer      task_data,
1538                     GCancellable *cancellable)
1539 {
1540   GOutputStream *stream = source_object;
1541   WriteData *op = task_data;
1542   GOutputStreamClass *class;
1543   GError *error = NULL;
1544   gssize count_written;
1545
1546   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1547   count_written = class->write_fn (stream, op->buffer, op->count_requested,
1548                                    cancellable, &error);
1549   if (count_written == -1)
1550     g_task_return_error (task, error);
1551   else
1552     g_task_return_int (task, count_written);
1553 }
1554
1555 static void write_async_pollable (GPollableOutputStream *stream,
1556                                   GTask                 *task);
1557
1558 static gboolean
1559 write_async_pollable_ready (GPollableOutputStream *stream,
1560                             gpointer               user_data)
1561 {
1562   GTask *task = user_data;
1563
1564   write_async_pollable (stream, task);
1565   return FALSE;
1566 }
1567
1568 static void
1569 write_async_pollable (GPollableOutputStream *stream,
1570                       GTask                 *task)
1571 {
1572   GError *error = NULL;
1573   WriteData *op = g_task_get_task_data (task);
1574   gssize count_written;
1575
1576   if (g_task_return_error_if_cancelled (task))
1577     return;
1578
1579   count_written = G_POLLABLE_OUTPUT_STREAM_GET_INTERFACE (stream)->
1580     write_nonblocking (stream, op->buffer, op->count_requested, &error);
1581
1582   if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
1583     {
1584       GSource *source;
1585
1586       g_error_free (error);
1587
1588       source = g_pollable_output_stream_create_source (stream,
1589                                                        g_task_get_cancellable (task));
1590       g_task_attach_source (task, source,
1591                             (GSourceFunc) write_async_pollable_ready);
1592       g_source_unref (source);
1593       return;
1594     }
1595
1596   if (count_written == -1)
1597     g_task_return_error (task, error);
1598   else
1599     g_task_return_int (task, count_written);
1600 }
1601
1602 static void
1603 g_output_stream_real_write_async (GOutputStream       *stream,
1604                                   const void          *buffer,
1605                                   gsize                count,
1606                                   int                  io_priority,
1607                                   GCancellable        *cancellable,
1608                                   GAsyncReadyCallback  callback,
1609                                   gpointer             user_data)
1610 {
1611   GTask *task;
1612   WriteData *op;
1613
1614   op = g_slice_new0 (WriteData);
1615   task = g_task_new (stream, cancellable, callback, user_data);
1616   g_task_set_check_cancellable (task, FALSE);
1617   g_task_set_task_data (task, op, (GDestroyNotify) free_write_data);
1618   op->buffer = buffer;
1619   op->count_requested = count;
1620
1621   if (!g_output_stream_async_write_is_via_threads (stream))
1622     write_async_pollable (G_POLLABLE_OUTPUT_STREAM (stream), task);
1623   else
1624     g_task_run_in_thread (task, write_async_thread);
1625   g_object_unref (task);
1626 }
1627
1628 static gssize
1629 g_output_stream_real_write_finish (GOutputStream  *stream,
1630                                    GAsyncResult   *result,
1631                                    GError        **error)
1632 {
1633   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1634
1635   return g_task_propagate_int (G_TASK (result), error);
1636 }
1637
1638 typedef struct {
1639   GInputStream *source;
1640   GOutputStreamSpliceFlags flags;
1641   gssize n_read;
1642   gssize n_written;
1643   gsize bytes_copied;
1644   GError *error;
1645   guint8 *buffer;
1646 } SpliceData;
1647
1648 static void
1649 free_splice_data (SpliceData *op)
1650 {
1651   g_clear_pointer (&op->buffer, g_free);
1652   g_object_unref (op->source);
1653   g_clear_error (&op->error);
1654   g_free (op);
1655 }
1656
1657 static void
1658 real_splice_async_complete_cb (GTask *task)
1659 {
1660   SpliceData *op = g_task_get_task_data (task);
1661
1662   if (op->flags & G_OUTPUT_STREAM_SPLICE_CLOSE_SOURCE &&
1663       !g_input_stream_is_closed (op->source))
1664     return;
1665
1666   if (op->flags & G_OUTPUT_STREAM_SPLICE_CLOSE_TARGET &&
1667       !g_output_stream_is_closed (g_task_get_source_object (task)))
1668     return;
1669
1670   if (op->error != NULL)
1671     {
1672       g_task_return_error (task, op->error);
1673       op->error = NULL;
1674     }
1675   else
1676     {
1677       g_task_return_int (task, op->bytes_copied);
1678     }
1679
1680   g_object_unref (task);
1681 }
1682
1683 static void
1684 real_splice_async_close_input_cb (GObject      *source,
1685                                   GAsyncResult *res,
1686                                   gpointer      user_data)
1687 {
1688   GTask *task = user_data;
1689
1690   g_input_stream_close_finish (G_INPUT_STREAM (source), res, NULL);
1691
1692   real_splice_async_complete_cb (task);
1693 }
1694
1695 static void
1696 real_splice_async_close_output_cb (GObject      *source,
1697                                    GAsyncResult *res,
1698                                    gpointer      user_data)
1699 {
1700   GTask *task = G_TASK (user_data);
1701   SpliceData *op = g_task_get_task_data (task);
1702   GError **error = (op->error == NULL) ? &op->error : NULL;
1703
1704   g_output_stream_internal_close_finish (G_OUTPUT_STREAM (source), res, error);
1705
1706   real_splice_async_complete_cb (task);
1707 }
1708
1709 static void
1710 real_splice_async_complete (GTask *task)
1711 {
1712   SpliceData *op = g_task_get_task_data (task);
1713   gboolean done = TRUE;
1714
1715   if (op->flags & G_OUTPUT_STREAM_SPLICE_CLOSE_SOURCE)
1716     {
1717       done = FALSE;
1718       g_input_stream_close_async (op->source, g_task_get_priority (task),
1719                                   g_task_get_cancellable (task),
1720                                   real_splice_async_close_input_cb, task);
1721     }
1722
1723   if (op->flags & G_OUTPUT_STREAM_SPLICE_CLOSE_TARGET)
1724     {
1725       done = FALSE;
1726       g_output_stream_internal_close_async (g_task_get_source_object (task),
1727                                             g_task_get_priority (task),
1728                                             g_task_get_cancellable (task),
1729                                             real_splice_async_close_output_cb,
1730                                             task);
1731     }
1732
1733   if (done)
1734     real_splice_async_complete_cb (task);
1735 }
1736
1737 static void real_splice_async_read_cb (GObject      *source,
1738                                        GAsyncResult *res,
1739                                        gpointer      user_data);
1740
1741 static void
1742 real_splice_async_write_cb (GObject      *source,
1743                             GAsyncResult *res,
1744                             gpointer      user_data)
1745 {
1746   GOutputStreamClass *class;
1747   GTask *task = G_TASK (user_data);
1748   SpliceData *op = g_task_get_task_data (task);
1749   gssize ret;
1750
1751   class = G_OUTPUT_STREAM_GET_CLASS (g_task_get_source_object (task));
1752
1753   ret = class->write_finish (G_OUTPUT_STREAM (source), res, &op->error);
1754
1755   if (ret == -1)
1756     {
1757       real_splice_async_complete (task);
1758       return;
1759     }
1760
1761   op->n_written += ret;
1762   op->bytes_copied += ret;
1763   if (op->bytes_copied > G_MAXSSIZE)
1764     op->bytes_copied = G_MAXSSIZE;
1765
1766   if (op->n_written < op->n_read)
1767     {
1768       class->write_async (g_task_get_source_object (task),
1769                           op->buffer + op->n_written,
1770                           op->n_read - op->n_written,
1771                           g_task_get_priority (task),
1772                           g_task_get_cancellable (task),
1773                           real_splice_async_write_cb, task);
1774       return;
1775     }
1776
1777   g_input_stream_read_async (op->source, op->buffer, 8192,
1778                              g_task_get_priority (task),
1779                              g_task_get_cancellable (task),
1780                              real_splice_async_read_cb, task);
1781 }
1782
1783 static void
1784 real_splice_async_read_cb (GObject      *source,
1785                            GAsyncResult *res,
1786                            gpointer      user_data)
1787 {
1788   GOutputStreamClass *class;
1789   GTask *task = G_TASK (user_data);
1790   SpliceData *op = g_task_get_task_data (task);
1791   gssize ret;
1792
1793   class = G_OUTPUT_STREAM_GET_CLASS (g_task_get_source_object (task));
1794
1795   ret = g_input_stream_read_finish (op->source, res, &op->error);
1796   if (ret == -1 || ret == 0)
1797     {
1798       real_splice_async_complete (task);
1799       return;
1800     }
1801
1802   op->n_read = ret;
1803   op->n_written = 0;
1804
1805   class->write_async (g_task_get_source_object (task), op->buffer,
1806                       op->n_read, g_task_get_priority (task),
1807                       g_task_get_cancellable (task),
1808                       real_splice_async_write_cb, task);
1809 }
1810
1811 static void
1812 splice_async_thread (GTask        *task,
1813                      gpointer      source_object,
1814                      gpointer      task_data,
1815                      GCancellable *cancellable)
1816 {
1817   GOutputStream *stream = source_object;
1818   SpliceData *op = task_data;
1819   GOutputStreamClass *class;
1820   GError *error = NULL;
1821   gssize bytes_copied;
1822
1823   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1824
1825   bytes_copied = class->splice (stream,
1826                                 op->source,
1827                                 op->flags,
1828                                 cancellable,
1829                                 &error);
1830   if (bytes_copied == -1)
1831     g_task_return_error (task, error);
1832   else
1833     g_task_return_int (task, bytes_copied);
1834 }
1835
1836 static void
1837 g_output_stream_real_splice_async (GOutputStream             *stream,
1838                                    GInputStream              *source,
1839                                    GOutputStreamSpliceFlags   flags,
1840                                    int                        io_priority,
1841                                    GCancellable              *cancellable,
1842                                    GAsyncReadyCallback        callback,
1843                                    gpointer                   user_data)
1844 {
1845   GTask *task;
1846   SpliceData *op;
1847
1848   op = g_new0 (SpliceData, 1);
1849   task = g_task_new (stream, cancellable, callback, user_data);
1850   g_task_set_task_data (task, op, (GDestroyNotify)free_splice_data);
1851   op->flags = flags;
1852   op->source = g_object_ref (source);
1853
1854   if (g_input_stream_async_read_is_via_threads (source) &&
1855       g_output_stream_async_write_is_via_threads (stream))
1856     {
1857       g_task_run_in_thread (task, splice_async_thread);
1858       g_object_unref (task);
1859     }
1860   else
1861     {
1862       op->buffer = g_malloc (8192);
1863       g_input_stream_read_async (op->source, op->buffer, 8192,
1864                                  g_task_get_priority (task),
1865                                  g_task_get_cancellable (task),
1866                                  real_splice_async_read_cb, task);
1867     }
1868 }
1869
1870 static gssize
1871 g_output_stream_real_splice_finish (GOutputStream  *stream,
1872                                     GAsyncResult   *result,
1873                                     GError        **error)
1874 {
1875   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1876
1877   return g_task_propagate_int (G_TASK (result), error);
1878 }
1879
1880
1881 static void
1882 flush_async_thread (GTask        *task,
1883                     gpointer      source_object,
1884                     gpointer      task_data,
1885                     GCancellable *cancellable)
1886 {
1887   GOutputStream *stream = source_object;
1888   GOutputStreamClass *class;
1889   gboolean result;
1890   GError *error = NULL;
1891
1892   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1893   result = TRUE;
1894   if (class->flush)
1895     result = class->flush (stream, cancellable, &error);
1896
1897   if (result)
1898     g_task_return_boolean (task, TRUE);
1899   else
1900     g_task_return_error (task, error);
1901 }
1902
1903 static void
1904 g_output_stream_real_flush_async (GOutputStream       *stream,
1905                                   int                  io_priority,
1906                                   GCancellable        *cancellable,
1907                                   GAsyncReadyCallback  callback,
1908                                   gpointer             user_data)
1909 {
1910   GTask *task;
1911
1912   task = g_task_new (stream, cancellable, callback, user_data);
1913   g_task_set_priority (task, io_priority);
1914   g_task_run_in_thread (task, flush_async_thread);
1915   g_object_unref (task);
1916 }
1917
1918 static gboolean
1919 g_output_stream_real_flush_finish (GOutputStream  *stream,
1920                                    GAsyncResult   *result,
1921                                    GError        **error)
1922 {
1923   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1924
1925   return g_task_propagate_boolean (G_TASK (result), error);
1926 }
1927
1928 static void
1929 close_async_thread (GTask        *task,
1930                     gpointer      source_object,
1931                     gpointer      task_data,
1932                     GCancellable *cancellable)
1933 {
1934   GOutputStream *stream = source_object;
1935   GOutputStreamClass *class;
1936   GError *error = NULL;
1937   gboolean result = TRUE;
1938
1939   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1940
1941   /* Do a flush here if there is a flush function, and we did not have to do
1942    * an async flush before (see g_output_stream_close_async)
1943    */
1944   if (class->flush != NULL &&
1945       (class->flush_async == NULL ||
1946        class->flush_async == g_output_stream_real_flush_async))
1947     {
1948       result = class->flush (stream, cancellable, &error);
1949     }
1950
1951   /* Auto handling of cancelation disabled, and ignore
1952      cancellation, since we want to close things anyway, although
1953      possibly in a quick-n-dirty way. At least we never want to leak
1954      open handles */
1955
1956   if (class->close_fn)
1957     {
1958       /* Make sure to close, even if the flush failed (see sync close) */
1959       if (!result)
1960         class->close_fn (stream, cancellable, NULL);
1961       else
1962         result = class->close_fn (stream, cancellable, &error);
1963     }
1964
1965   if (result)
1966     g_task_return_boolean (task, TRUE);
1967   else
1968     g_task_return_error (task, error);
1969 }
1970
1971 static void
1972 g_output_stream_real_close_async (GOutputStream       *stream,
1973                                   int                  io_priority,
1974                                   GCancellable        *cancellable,
1975                                   GAsyncReadyCallback  callback,
1976                                   gpointer             user_data)
1977 {
1978   GTask *task;
1979
1980   task = g_task_new (stream, cancellable, callback, user_data);
1981   g_task_set_priority (task, io_priority);
1982   g_task_run_in_thread (task, close_async_thread);
1983   g_object_unref (task);
1984 }
1985
1986 static gboolean
1987 g_output_stream_real_close_finish (GOutputStream  *stream,
1988                                    GAsyncResult   *result,
1989                                    GError        **error)
1990 {
1991   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1992
1993   return g_task_propagate_boolean (G_TASK (result), error);
1994 }