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 "goutputstream.h"
  25 #include "gsimpleasyncresult.h"
  26 #include "glibintl.h"
  27
  28 /**
  29  * SECTION:goutputstream
  30  * @short_description: base class for implementing streaming output
  31  *
  32  *
  33  *
  34  **/
  35
  36 G_DEFINE_TYPE (GOutputStream, g_output_stream, G_TYPE_OBJECT);
  37
  38 struct _GOutputStreamPrivate {
  39   guint closed : 1;
  40   guint pending : 1;
  41   guint cancelled : 1;
  42   GAsyncReadyCallback outstanding_callback;
  43 };
  44
  45 static gssize   g_output_stream_real_splice        (GOutputStream             *stream,
  46                                                     GInputStream              *source,
  47                                                     GOutputStreamSpliceFlags   flags,
  48                                                     GCancellable              *cancellable,
  49                                                     GError                   **error);
  50 static void     g_output_stream_real_write_async   (GOutputStream             *stream,
  51                                                     const void                *buffer,
  52                                                     gsize                      count,
  53                                                     int                        io_priority,
  54                                                     GCancellable              *cancellable,
  55                                                     GAsyncReadyCallback        callback,
  56                                                     gpointer                   data);
  57 static gssize   g_output_stream_real_write_finish  (GOutputStream             *stream,
  58                                                     GAsyncResult              *result,
  59                                                     GError                   **error);
  60 static void     g_output_stream_real_splice_async  (GOutputStream             *stream,
  61                                                     GInputStream              *source,
  62                                                     GOutputStreamSpliceFlags   flags,
  63                                                     int                        io_priority,
  64                                                     GCancellable              *cancellable,
  65                                                     GAsyncReadyCallback        callback,
  66                                                     gpointer                   data);
  67 static gssize   g_output_stream_real_splice_finish (GOutputStream             *stream,
  68                                                     GAsyncResult              *result,
  69                                                     GError                   **error);
  70 static void     g_output_stream_real_flush_async   (GOutputStream             *stream,
  71                                                     int                        io_priority,
  72                                                     GCancellable              *cancellable,
  73                                                     GAsyncReadyCallback        callback,
  74                                                     gpointer                   data);
  75 static gboolean g_output_stream_real_flush_finish  (GOutputStream             *stream,
  76                                                     GAsyncResult              *result,
  77                                                     GError                   **error);
  78 static void     g_output_stream_real_close_async   (GOutputStream             *stream,
  79                                                     int                        io_priority,
  80                                                     GCancellable              *cancellable,
  81                                                     GAsyncReadyCallback        callback,
  82                                                     gpointer                   data);
  83 static gboolean g_output_stream_real_close_finish  (GOutputStream             *stream,
  84                                                     GAsyncResult              *result,
  85                                                     GError                   **error);
  86
  87 static void
  88 g_output_stream_finalize (GObject *object)
  89 {
  90   GOutputStream *stream;
  91
  92   stream = G_OUTPUT_STREAM (object);
  93
  94   if (G_OBJECT_CLASS (g_output_stream_parent_class)->finalize)
  95     (*G_OBJECT_CLASS (g_output_stream_parent_class)->finalize) (object);
  96 }
  97
  98 static void
  99 g_output_stream_dispose (GObject *object)
 100 {
 101   GOutputStream *stream;
 102
 103   stream = G_OUTPUT_STREAM (object);
 104
 105   if (!stream->priv->closed)
 106     g_output_stream_close (stream, NULL, NULL);
 107
 108   if (G_OBJECT_CLASS (g_output_stream_parent_class)->dispose)
 109     (*G_OBJECT_CLASS (g_output_stream_parent_class)->dispose) (object);
 110 }
 111
 112 static void
 113 g_output_stream_class_init (GOutputStreamClass *klass)
 114 {
 115   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 116
 117   g_type_class_add_private (klass, sizeof (GOutputStreamPrivate));
 118
 119   gobject_class->finalize = g_output_stream_finalize;
 120   gobject_class->dispose = g_output_stream_dispose;
 121
 122   klass->splice = g_output_stream_real_splice;
 123
 124   klass->write_async = g_output_stream_real_write_async;
 125   klass->write_finish = g_output_stream_real_write_finish;
 126   klass->splice_async = g_output_stream_real_splice_async;
 127   klass->splice_finish = g_output_stream_real_splice_finish;
 128   klass->flush_async = g_output_stream_real_flush_async;
 129   klass->flush_finish = g_output_stream_real_flush_finish;
 130   klass->close_async = g_output_stream_real_close_async;
 131   klass->close_finish = g_output_stream_real_close_finish;
 132 }
 133
 134 static void
 135 g_output_stream_init (GOutputStream *stream)
 136 {
 137   stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
 138                                               G_TYPE_OUTPUT_STREAM,
 139                                               GOutputStreamPrivate);
 140 }
 141
 142 /**
 143  * g_output_stream_write:
 144  * @stream: a #GOutputStream.
 145  * @buffer: the buffer containing the data to write.
 146  * @count: the number of bytes to write
 147  * @cancellable: optional cancellable object
 148  * @error: location to store the error occuring, or %NULL to ignore
 149  *
 150  * Tries to write @count bytes from @buffer into the stream. Will block
 151  * during the operation.
 152  *
 153  * If count is zero returns zero and does nothing. A value of @count
 154  * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 155  *
 156  * On success, the number of bytes written to the stream is returned.
 157  * It is not an error if this is not the same as the requested size, as it
 158  * can happen e.g. on a partial i/o error, or if the there is not enough
 159  * storage in the stream. All writes either block until at least one byte
 160  * is written, so zero is never returned (unless @count is zero).
 161  *
 162  * If @cancellable is not NULL, then the operation can be cancelled by
 163  * triggering the cancellable object from another thread. If the operation
 164  * was cancelled, the error G_IO_ERROR_CANCELLED will be returned. If an
 165  * operation was partially finished when the operation was cancelled the
 166  * partial result will be returned, without an error.
 167  *
 168  * On error -1 is returned and @error is set accordingly.
 169  *
 170  * Return value: Number of bytes written, or -1 on error
 171  **/
 172 gssize
 173 g_output_stream_write (GOutputStream *stream,
 174                        const void    *buffer,
 175                        gsize          count,
 176                        GCancellable  *cancellable,
 177                        GError       **error)
 178 {
 179   GOutputStreamClass *class;
 180   gssize res;
 181
 182   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), -1);
 183   g_return_val_if_fail (buffer != NULL, 0);
 184
 185   if (count == 0)
 186     return 0;
 187
 188   if (((gssize) count) < 0)
 189     {
 190       g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 191                    _("Too large count value passed to g_output_stream_write"));
 192       return -1;
 193     }
 194
 195   if (stream->priv->closed)
 196     {
 197       g_set_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
 198                    _("Stream is already closed"));
 199       return -1;
 200     }
 201
 202   if (stream->priv->pending)
 203     {
 204       g_set_error (error, G_IO_ERROR, G_IO_ERROR_PENDING,
 205                    _("Stream has outstanding operation"));
 206       return -1;
 207     }
 208
 209   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 210
 211   if (class->write == NULL)
 212     {
 213       g_set_error (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
 214                    _("Output stream doesn't implement write"));
 215       return -1;
 216     }
 217
 218   if (cancellable)
 219     g_push_current_cancellable (cancellable);
 220
 221   stream->priv->pending = TRUE;
 222   res = class->write (stream, buffer, count, cancellable, error);
 223   stream->priv->pending = FALSE;
 224
 225   if (cancellable)
 226     g_pop_current_cancellable (cancellable);
 227
 228   return res;
 229 }
 230
 231 /**
 232  * g_output_stream_write_all:
 233  * @stream: a #GOutputStream.
 234  * @buffer: the buffer containing the data to write.
 235  * @count: the number of bytes to write
 236  * @bytes_written: location to store the number of bytes that was written to the stream
 237  * @cancellable: optional #GCancellable object, %NULL to ignore.
 238  * @error: location to store the error occuring, or %NULL to ignore
 239  *
 240  * Tries to write @count bytes from @buffer into the stream. Will block
 241  * during the operation.
 242  *
 243  * This function is similar to g_output_stream_write(), except it tries to
 244  * write as many bytes as requested, only stopping on an error.
 245  *
 246  * On a successful write of @count bytes, %TRUE is returned, and @bytes_written
 247  * is set to @count.
 248  *
 249  * If there is an error during the operation FALSE is returned and @error
 250  * is set to indicate the error status, @bytes_written is updated to contain
 251  * the number of bytes written into the stream before the error occured.
 252  *
 253  * Return value: %TRUE on success, %FALSE if there was an error
 254  **/
 255 gboolean
 256 g_output_stream_write_all (GOutputStream *stream,
 257                            const void    *buffer,
 258                            gsize          count,
 259                            gsize         *bytes_written,
 260                            GCancellable  *cancellable,
 261                            GError       **error)
 262 {
 263   gsize _bytes_written;
 264   gssize res;
 265
 266   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 267   g_return_val_if_fail (buffer != NULL, FALSE);
 268
 269   _bytes_written = 0;
 270   while (_bytes_written < count)
 271     {
 272       res = g_output_stream_write (stream, (char *)buffer + _bytes_written, count - _bytes_written,
 273                                    cancellable, error);
 274       if (res == -1)
 275         {
 276           if (bytes_written)
 277             *bytes_written = _bytes_written;
 278           return FALSE;
 279         }
 280
 281       if (res == 0)
 282         g_warning ("Write returned zero without error");
 283
 284       _bytes_written += res;
 285     }
 286
 287   if (bytes_written)
 288     *bytes_written = _bytes_written;
 289   return TRUE;
 290 }
 291
 292 /**
 293  * g_output_stream_flush:
 294  * @stream: a #GOutputStream.
 295  * @cancellable: optional cancellable object
 296  * @error: location to store the error occuring, or %NULL to ignore
 297  *
 298  * Flushed any outstanding buffers in the stream. Will block during the operation.
 299  * Closing the stream will implicitly cause a flush.
 300  *
 301  * This function is optional for inherited classes.
 302  *
 303  * If @cancellable is not %NULL, then the operation can be cancelled by
 304  * triggering the cancellable object from another thread. If the operation
 305  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 306  *
 307  * Return value: %TRUE on success, %FALSE on error
 308  **/
 309 gboolean
 310 g_output_stream_flush (GOutputStream    *stream,
 311                        GCancellable  *cancellable,
 312                        GError          **error)
 313 {
 314   GOutputStreamClass *class;
 315   gboolean res;
 316
 317   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 318
 319   if (stream->priv->closed)
 320     {
 321       g_set_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
 322                    _("Stream is already closed"));
 323       return FALSE;
 324     }
 325
 326   if (stream->priv->pending)
 327     {
 328       g_set_error (error, G_IO_ERROR, G_IO_ERROR_PENDING,
 329                    _("Stream has outstanding operation"));
 330       return FALSE;
 331     }
 332
 333   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 334
 335   res = TRUE;
 336   if (class->flush)
 337     {
 338       if (cancellable)
 339         g_push_current_cancellable (cancellable);
 340
 341       stream->priv->pending = TRUE;
 342       res = class->flush (stream, cancellable, error);
 343       stream->priv->pending = FALSE;
 344
 345       if (cancellable)
 346         g_pop_current_cancellable (cancellable);
 347     }
 348
 349   return res;
 350 }
 351
 352 /**
 353  * g_output_stream_splice:
 354  * @stream: a #GOutputStream.
 355  * @source: a #GInputStream.
 356  * @flags: a set of #GOutputStreamSpliceFlags.
 357  * @cancellable: optional #GCancellable object, %NULL to ignore.
 358  * @error: a #GError location to store the error occuring, or %NULL to
 359  * ignore.
 360  *
 361  * Splices an input stream into an output stream.
 362  *
 363  * Returns: a #gssize containig the size of the data spliced.
 364  **/
 365 gssize
 366 g_output_stream_splice (GOutputStream *stream,
 367                         GInputStream *source,
 368                         GOutputStreamSpliceFlags flags,
 369                         GCancellable  *cancellable,
 370                         GError **error)
 371 {
 372   GOutputStreamClass *class;
 373   gboolean res;
 374
 375   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), -1);
 376   g_return_val_if_fail (G_IS_INPUT_STREAM (source), -1);
 377
 378   if (stream->priv->closed)
 379     {
 380       g_set_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
 381                    _("Target stream is already closed"));
 382       return -1;
 383     }
 384
 385   if (g_input_stream_is_closed (source))
 386     {
 387       g_set_error (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
 388                    _("Source stream is already closed"));
 389       return -1;
 390     }
 391
 392   if (stream->priv->pending)
 393     {
 394       g_set_error (error, G_IO_ERROR, G_IO_ERROR_PENDING,
 395                    _("Stream has outstanding operation"));
 396       return -1;
 397     }
 398
 399   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 400
 401   res = TRUE;
 402   if (cancellable)
 403     g_push_current_cancellable (cancellable);
 404
 405   stream->priv->pending = TRUE;
 406   res = class->splice (stream, source, flags, cancellable, error);
 407   stream->priv->pending = FALSE;
 408
 409   if (cancellable)
 410     g_pop_current_cancellable (cancellable);
 411
 412   return res;
 413 }
 414
 415 static gssize
 416 g_output_stream_real_splice (GOutputStream *stream,
 417                              GInputStream *source,
 418                              GOutputStreamSpliceFlags flags,
 419                              GCancellable  *cancellable,
 420                              GError **error)
 421 {
 422   gssize n_read, n_written;
 423   gssize bytes_copied;
 424   char buffer[8192], *p;
 425   gboolean res;
 426
 427   bytes_copied = 0;
 428   res = TRUE;
 429   do
 430     {
 431       n_read = g_input_stream_read (source, buffer, sizeof (buffer), cancellable, error);
 432       if (n_read == -1)
 433         {
 434           res = FALSE;
 435           break;
 436         }
 437
 438       if (n_read == 0)
 439         break;
 440
 441       p = buffer;
 442       while (n_read > 0)
 443         {
 444           stream->priv->pending = FALSE;
 445           n_written = g_output_stream_write (stream, p, n_read, cancellable, error);
 446           stream->priv->pending = TRUE;
 447           if (n_written == -1)
 448             {
 449               res = FALSE;
 450               break;
 451             }
 452
 453           p += n_written;
 454           n_read -= n_written;
 455           bytes_copied += n_written;
 456         }
 457     }
 458   while (res);
 459
 460   if (!res)
 461     error = NULL; /* Ignore further errors */
 462
 463   if (flags & G_OUTPUT_STREAM_SPLICE_FLAGS_CLOSE_SOURCE)
 464     {
 465       /* Don't care about errors in source here */
 466       g_input_stream_close (source, cancellable, NULL);
 467     }
 468
 469   if (flags & G_OUTPUT_STREAM_SPLICE_FLAGS_CLOSE_TARGET)
 470     {
 471       /* But write errors on close are bad! */
 472       stream->priv->pending = FALSE;
 473       if (!g_output_stream_close (stream, cancellable, error))
 474         res = FALSE;
 475       stream->priv->pending = TRUE;
 476     }
 477
 478   if (res)
 479     return bytes_copied;
 480
 481   return -1;
 482 }
 483
 484
 485 /**
 486  * g_output_stream_close:
 487  * @stream: A #GOutputStream.
 488  * @cancellable: optional cancellable object
 489  * @error: location to store the error occuring, or %NULL to ignore
 490  *
 491  * Closes the stream, releasing resources related to it.
 492  *
 493  * Once the stream is closed, all other operations will return %G_IO_ERROR_CLOSED.
 494  * Closing a stream multiple times will not return an error.
 495  *
 496  * Closing a stream will automatically flush any outstanding buffers in the
 497  * stream.
 498  *
 499  * Streams will be automatically closed when the last reference
 500  * is dropped, but you might want to call make sure resources
 501  * are released as early as possible.
 502  *
 503  * Some streams might keep the backing store of the stream (e.g. a file descriptor)
 504  * open after the stream is closed. See the documentation for the individual
 505  * stream for details.
 506  *
 507  * On failure the first error that happened will be reported, but the close
 508  * operation will finish as much as possible. A stream that failed to
 509  * close will still return %G_IO_ERROR_CLOSED all operations. Still, it
 510  * is important to check and report the error to the user, otherwise
 511  * there might be a loss of data as all data might not be written.
 512  *
 513  * If @cancellable is not NULL, then the operation can be cancelled by
 514  * triggering the cancellable object from another thread. If the operation
 515  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 516  * Cancelling a close will still leave the stream closed, but there some streams
 517  * can use a faster close that doesn't block to e.g. check errors. On
 518  * cancellation (as with any error) there is no guarantee that all written
 519  * data will reach the target.
 520  *
 521  * Return value: %TRUE on success, %FALSE on failure
 522  **/
 523 gboolean
 524 g_output_stream_close (GOutputStream  *stream,
 525                        GCancellable   *cancellable,
 526                        GError        **error)
 527 {
 528   GOutputStreamClass *class;
 529   gboolean res;
 530
 531   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 532
 533   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 534
 535   if (stream->priv->closed)
 536     return TRUE;
 537
 538   if (stream->priv->pending)
 539     {
 540       g_set_error (error, G_IO_ERROR, G_IO_ERROR_PENDING,
 541                    _("Stream has outstanding operation"));
 542       return FALSE;
 543     }
 544
 545   res = g_output_stream_flush (stream, cancellable, error);
 546
 547   stream->priv->pending = TRUE;
 548
 549   if (cancellable)
 550     g_push_current_cancellable (cancellable);
 551
 552   if (!res)
 553     {
 554       /* flushing caused the error that we want to return,
 555        * but we still want to close the underlying stream if possible
 556        */
 557       if (class->close)
 558         class->close (stream, cancellable, NULL);
 559     }
 560   else
 561     {
 562       res = TRUE;
 563       if (class->close)
 564         res = class->close (stream, cancellable, error);
 565     }
 566
 567   if (cancellable)
 568     g_pop_current_cancellable (cancellable);
 569
 570   stream->priv->closed = TRUE;
 571   stream->priv->pending = FALSE;
 572
 573   return res;
 574 }
 575
 576 static void
 577 async_ready_callback_wrapper (GObject *source_object,
 578                               GAsyncResult *res,
 579                               gpointer      user_data)
 580 {
 581   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
 582
 583   stream->priv->pending = FALSE;
 584   if (stream->priv->outstanding_callback)
 585     (*stream->priv->outstanding_callback) (source_object, res, user_data);
 586   g_object_unref (stream);
 587 }
 588
 589 static void
 590 async_ready_close_callback_wrapper (GObject *source_object,
 591                                     GAsyncResult *res,
 592                                     gpointer user_data)
 593 {
 594   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
 595
 596   stream->priv->pending = FALSE;
 597   stream->priv->closed = TRUE;
 598   if (stream->priv->outstanding_callback)
 599     (*stream->priv->outstanding_callback) (source_object, res, user_data);
 600   g_object_unref (stream);
 601 }
 602
 603 /**
 604  * g_output_stream_write_async:
 605  * @stream: A #GOutputStream.
 606  * @buffer: the buffer containing the data to write.
 607  * @count: the number of bytes to write
 608  * @io_priority: the io priority of the request. the io priority of the request
 609  * @cancellable: optional #GCancellable object, %NULL to ignore.
 610  * @callback: callback to call when the request is satisfied
 611  * @user_data: the data to pass to callback function
 612  *
 613  * Request an asynchronous write of @count bytes from @buffer into the stream.
 614  * When the operation is finished @callback will be called, giving the results.
 615  *
 616  * During an async request no other sync and async calls are allowed, and will
 617  * result in %G_IO_ERROR_PENDING errors.
 618  *
 619  * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 620  *
 621  * On success, the number of bytes written will be passed to the
 622  * @callback. It is not an error if this is not the same as the requested size, as it
 623  * can happen e.g. on a partial i/o error, but generally we try to write
 624  * as many bytes as requested.
 625  *
 626  * Any outstanding i/o request with higher priority (lower numerical value) will
 627  * be executed before an outstanding request with lower priority. Default
 628  * priority is %G_PRIORITY_DEFAULT.
 629  *
 630  * The asyncronous methods have a default fallback that uses threads to implement
 631  * asynchronicity, so they are optional for inheriting classes. However, if you
 632  * override one you must override all.
 633  *
 634  * For the synchronous, blocking version of this function, see g_output_stream_write().
 635  **/
 636 void
 637 g_output_stream_write_async (GOutputStream       *stream,
 638                              const void          *buffer,
 639                              gsize                count,
 640                              int                  io_priority,
 641                              GCancellable        *cancellable,
 642                              GAsyncReadyCallback  callback,
 643                              gpointer             user_data)
 644 {
 645   GOutputStreamClass *class;
 646   GSimpleAsyncResult *simple;
 647
 648   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
 649   g_return_if_fail (buffer != NULL);
 650
 651   if (count == 0)
 652     {
 653       simple = g_simple_async_result_new (G_OBJECT (stream),
 654                                           callback,
 655                                           user_data,
 656                                           g_output_stream_write_async);
 657       g_simple_async_result_complete_in_idle (simple);
 658       g_object_unref (simple);
 659       return;
 660     }
 661
 662   if (((gssize) count) < 0)
 663     {
 664       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 665                                            callback,
 666                                            user_data,
 667                                            G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
 668                                            _("Too large count value passed to g_output_stream_write_async"));
 669       return;
 670     }
 671
 672   if (stream->priv->closed)
 673     {
 674       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 675                                            callback,
 676                                            user_data,
 677                                            G_IO_ERROR, G_IO_ERROR_CLOSED,
 678                                            _("Stream is already closed"));
 679       return;
 680     }
 681
 682   if (stream->priv->pending)
 683     {
 684       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 685                                            callback,
 686                                            user_data,
 687                                            G_IO_ERROR, G_IO_ERROR_PENDING,
 688                                            _("Stream has outstanding operation"));
 689       return;
 690     }
 691
 692   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 693
 694   stream->priv->pending = TRUE;
 695   stream->priv->outstanding_callback = callback;
 696   g_object_ref (stream);
 697   class->write_async (stream, buffer, count, io_priority, cancellable,
 698                       async_ready_callback_wrapper, user_data);
 699 }
 700
 701 /**
 702  * g_output_stream_write_finish:
 703  * @stream: a #GOutputStream.
 704  * @result: a #GAsyncResult.
 705  * @error: a #GError location to store the error occuring, or %NULL to
 706  * ignore.
 707  *
 708  * Finishes a stream write operation.
 709  *
 710  * Returns: a #gssize containing the number of bytes written to the stream.
 711  **/
 712 gssize
 713 g_output_stream_write_finish (GOutputStream *stream,
 714                               GAsyncResult *result,
 715                               GError **error)
 716 {
 717   GSimpleAsyncResult *simple;
 718   GOutputStreamClass *class;
 719
 720   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), -1);
 721   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 722
 723   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 724     {
 725       simple = G_SIMPLE_ASYNC_RESULT (result);
 726       if (g_simple_async_result_propagate_error (simple, error))
 727         return -1;
 728
 729       /* Special case writes of 0 bytes */
 730       if (g_simple_async_result_get_source_tag (simple) == g_output_stream_write_async)
 731         return 0;
 732     }
 733
 734   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 735   return class->write_finish (stream, result, error);
 736 }
 737
 738 typedef struct {
 739   GInputStream *source;
 740   gpointer user_data;
 741   GAsyncReadyCallback callback;
 742 } SpliceUserData;
 743
 744 static void
 745 async_ready_splice_callback_wrapper (GObject *source_object,
 746                                      GAsyncResult *res,
 747                                      gpointer _data)
 748 {
 749   GOutputStream *stream = G_OUTPUT_STREAM (source_object);
 750   SpliceUserData *data = _data;
 751
 752   stream->priv->pending = FALSE;
 753
 754   if (data->callback)
 755     (*data->callback) (source_object, res, data->user_data);
 756
 757   g_object_unref (stream);
 758   g_object_unref (data->source);
 759   g_free (data);
 760 }
 761
 762 /**
 763  * g_output_stream_splice_async:
 764  * @stream: a #GOutputStream.
 765  * @source: a #GInputStream.
 766  * @flags: a set of #GOutputStreamSpliceFlags.
 767  * @io_priority: the io priority of the request.
 768  * @cancellable: optional #GCancellable object, %NULL to ignore.
 769  * @callback: a #GAsyncReadyCallback.
 770  * @user_data: user data passed to @callback.
 771  *
 772  * Splices a stream asynchronously.
 773  *
 774  **/
 775 void
 776 g_output_stream_splice_async (GOutputStream             *stream,
 777                               GInputStream              *source,
 778                               GOutputStreamSpliceFlags   flags,
 779                               int                        io_priority,
 780                               GCancellable              *cancellable,
 781                               GAsyncReadyCallback        callback,
 782                               gpointer                   user_data)
 783 {
 784   GOutputStreamClass *class;
 785   SpliceUserData *data;
 786
 787   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
 788   g_return_if_fail (G_IS_INPUT_STREAM (source));
 789
 790   if (stream->priv->closed)
 791     {
 792       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 793                                            callback,
 794                                            user_data,
 795                                            G_IO_ERROR, G_IO_ERROR_CLOSED,
 796                                            _("Target stream is already closed"));
 797       return;
 798     }
 799
 800   if (g_input_stream_is_closed (source))
 801     {
 802       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 803                                            callback,
 804                                            user_data,
 805                                            G_IO_ERROR, G_IO_ERROR_CLOSED,
 806                                            _("Source stream is already closed"));
 807       return;
 808     }
 809
 810   if (stream->priv->pending)
 811     {
 812       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 813                                            callback,
 814                                            user_data,
 815                                            G_IO_ERROR, G_IO_ERROR_PENDING,
 816                                            _("Stream has outstanding operation"));
 817       return;
 818     }
 819
 820   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 821
 822   stream->priv->pending = TRUE;
 823
 824   data = g_new0 (SpliceUserData, 1);
 825   data->callback = callback;
 826   data->user_data = user_data;
 827   data->source = g_object_ref (source);
 828
 829   g_object_ref (stream);
 830   class->splice_async (stream, source, flags, io_priority, cancellable,
 831                       async_ready_splice_callback_wrapper, data);
 832 }
 833
 834 /**
 835  * g_output_stream_splice_finish:
 836  * @stream: a #GOutputStream.
 837  * @result: a #GAsyncResult.
 838  * @error: a #GError location to store the error occuring, or %NULL to
 839  * ignore.
 840  *
 841  * Finishes an asynchronous stream splice operation.
 842  *
 843  * Returns: a #gssize of the number of bytes spliced.
 844  **/
 845 gssize
 846 g_output_stream_splice_finish (GOutputStream             *stream,
 847                                GAsyncResult              *result,
 848                                GError                   **error)
 849 {
 850   GSimpleAsyncResult *simple;
 851   GOutputStreamClass *class;
 852
 853   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), -1);
 854   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
 855
 856   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 857     {
 858       simple = G_SIMPLE_ASYNC_RESULT (result);
 859       if (g_simple_async_result_propagate_error (simple, error))
 860         return -1;
 861     }
 862
 863   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 864   return class->splice_finish (stream, result, error);
 865 }
 866
 867 /**
 868  * g_output_stream_flush_async:
 869  * @stream: a #GOutputStream.
 870  * @io_priority: the io priority of the request.
 871  * @cancellable: optional #GCancellable object, %NULL to ignore.
 872  * @callback: a #GAsyncReadyCallback to call when the request is satisfied
 873  * @user_data: the data to pass to callback function
 874  *
 875  * Flushes a stream asynchronously.
 876  *
 877  **/
 878 void
 879 g_output_stream_flush_async (GOutputStream       *stream,
 880                              int                  io_priority,
 881                              GCancellable        *cancellable,
 882                              GAsyncReadyCallback  callback,
 883                              gpointer             user_data)
 884 {
 885   GOutputStreamClass *class;
 886   GSimpleAsyncResult *simple;
 887
 888   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
 889
 890   if (stream->priv->closed)
 891     {
 892       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 893                                            callback,
 894                                            user_data,
 895                                            G_IO_ERROR, G_IO_ERROR_CLOSED,
 896                                            _("Stream is already closed"));
 897       return;
 898     }
 899
 900   if (stream->priv->pending)
 901     {
 902       g_simple_async_report_error_in_idle (G_OBJECT (stream),
 903                                            callback,
 904                                            user_data,
 905                                            G_IO_ERROR, G_IO_ERROR_PENDING,
 906                                            _("Stream has outstanding operation"));
 907       return;
 908     }
 909
 910   class = G_OUTPUT_STREAM_GET_CLASS (stream);
 911
 912   if (class->flush_async == NULL)
 913     {
 914       simple = g_simple_async_result_new (G_OBJECT (stream),
 915                                           callback,
 916                                           user_data,
 917                                           g_output_stream_flush_async);
 918       g_simple_async_result_complete_in_idle (simple);
 919       g_object_unref (simple);
 920       return;
 921     }
 922
 923   stream->priv->pending = TRUE;
 924   stream->priv->outstanding_callback = callback;
 925   g_object_ref (stream);
 926   class->flush_async (stream, io_priority, cancellable,
 927                       async_ready_callback_wrapper, user_data);
 928 }
 929
 930 /**
 931  * g_output_stream_flush_finish:
 932  * @stream: a #GOutputStream.
 933  * @result: a GAsyncResult.
 934  * @error: a #GError location to store the error occuring, or %NULL to
 935  * ignore.
 936  *
 937  * Finishes flushing an output stream.
 938  *
 939  * Returns: %TRUE if flush operation suceeded, %FALSE otherwise.
 940  **/
 941 gboolean
 942 g_output_stream_flush_finish (GOutputStream *stream,
 943                               GAsyncResult *result,
 944                               GError **error)
 945 {
 946   GSimpleAsyncResult *simple;
 947   GOutputStreamClass *klass;
 948
 949   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
 950   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
 951
 952   if (G_IS_SIMPLE_ASYNC_RESULT (result))
 953     {
 954       simple = G_SIMPLE_ASYNC_RESULT (result);
 955       if (g_simple_async_result_propagate_error (simple, error))
 956         return FALSE;
 957
 958       /* Special case default implementation */
 959       if (g_simple_async_result_get_source_tag (simple) == g_output_stream_flush_async)
 960         return TRUE;
 961     }
 962
 963   klass = G_OUTPUT_STREAM_GET_CLASS (stream);
 964   return klass->flush_finish (stream, result, error);
 965 }
 966
 967
 968 /**
 969  * g_output_stream_close_async:
 970  * @stream: A #GOutputStream.
 971  * @io_priority: the io priority of the request.
 972  * @callback: callback to call when the request is satisfied
 973  * @user_data: the data to pass to callback function
 974  * @cancellable: optional cancellable object
 975  *
 976  * Requests an asynchronous close of the stream, releasing resources related to it.
 977  * When the operation is finished @callback will be called, giving the results.
 978  *
 979  * For behaviour details see g_output_stream_close().
 980  *
 981  * The asyncronous methods have a default fallback that uses threads to implement
 982  * asynchronicity, so they are optional for inheriting classes. However, if you
 983  * override one you must override all.
 984  **/
 985 void
 986 g_output_stream_close_async (GOutputStream      *stream,
 987                              int                 io_priority,
 988                              GCancellable       *cancellable,
 989                              GAsyncReadyCallback callback,
 990                              gpointer            user_data)
 991 {
 992   GOutputStreamClass *class;
 993   GSimpleAsyncResult *simple;
 994
 995   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
 996
 997   if (stream->priv->closed)
 998     {
 999       simple = g_simple_async_result_new (G_OBJECT (stream),
1000                                           callback,
1001                                           user_data,
1002                                           g_output_stream_close_async);
1003       g_simple_async_result_complete_in_idle (simple);
1004       g_object_unref (simple);
1005       return;
1006     }
1007
1008   if (stream->priv->pending)
1009     {
1010       g_simple_async_report_error_in_idle (G_OBJECT (stream),
1011                                            callback,
1012                                            user_data,
1013                                            G_IO_ERROR, G_IO_ERROR_PENDING,
1014                                            _("Stream has outstanding operation"));
1015       return;
1016     }
1017
1018   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1019   stream->priv->pending = TRUE;
1020   stream->priv->outstanding_callback = callback;
1021   g_object_ref (stream);
1022   class->close_async (stream, io_priority, cancellable,
1023                       async_ready_close_callback_wrapper, user_data);
1024 }
1025
1026 /**
1027  * g_output_stream_close_finish:
1028  * @stream: a #GOutputStream.
1029  * @result: a #GAsyncResult.
1030  * @error: a #GError location to store the error occuring, or %NULL to
1031  * ignore.
1032  *
1033  * Closes an output stream.
1034  *
1035  * Returns: %TRUE if stream was successfully closed, %FALSE otherwise.
1036  **/
1037 gboolean
1038 g_output_stream_close_finish (GOutputStream *stream,
1039                               GAsyncResult *result,
1040                               GError **error)
1041 {
1042   GSimpleAsyncResult *simple;
1043   GOutputStreamClass *class;
1044
1045   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1046   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
1047
1048   if (G_IS_SIMPLE_ASYNC_RESULT (result))
1049     {
1050       simple = G_SIMPLE_ASYNC_RESULT (result);
1051       if (g_simple_async_result_propagate_error (simple, error))
1052         return FALSE;
1053
1054       /* Special case already closed */
1055       if (g_simple_async_result_get_source_tag (simple) == g_output_stream_close_async)
1056         return TRUE;
1057     }
1058
1059   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1060   return class->close_finish (stream, result, error);
1061 }
1062
1063 /**
1064  * g_output_stream_is_closed:
1065  * @stream: a #GOutputStream.
1066  *
1067  * Checks if an output stream has already been closed.
1068  *
1069  * Returns: %TRUE if @stream is closed. %FALSE otherwise.
1070  **/
1071 gboolean
1072 g_output_stream_is_closed (GOutputStream *stream)
1073 {
1074   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), TRUE);
1075
1076   return stream->priv->closed;
1077 }
1078
1079 /**
1080  * g_output_stream_has_pending:
1081  * @stream: a #GOutputStream.
1082  *
1083  * Checks if an ouput stream has pending actions.
1084  *
1085  * Returns: %TRUE if @stream has pending actions.
1086  **/
1087 gboolean
1088 g_output_stream_has_pending (GOutputStream *stream)
1089 {
1090   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1091
1092   return stream->priv->pending;
1093 }
1094
1095 /**
1096  * g_output_stream_set_pending:
1097  * @stream: a #GOutputStream.
1098  * @pending: a #gboolean.
1099  *
1100  * Sets the @stream as having pending actions if @pending is %TRUE.
1101  **/
1102 void
1103 g_output_stream_set_pending (GOutputStream              *stream,
1104                             gboolean                   pending)
1105 {
1106   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
1107
1108   stream->priv->pending = pending;
1109 }
1110
1111
1112 /********************************************
1113  *   Default implementation of async ops    *
1114  ********************************************/
1115
1116 typedef struct {
1117   const void         *buffer;
1118   gsize               count_requested;
1119   gssize              count_written;
1120 } WriteData;
1121
1122 static void
1123 write_async_thread (GSimpleAsyncResult *res,
1124                    GObject *object,
1125                    GCancellable *cancellable)
1126 {
1127   WriteData *op;
1128   GOutputStreamClass *class;
1129   GError *error = NULL;
1130
1131   class = G_OUTPUT_STREAM_GET_CLASS (object);
1132   op = g_simple_async_result_get_op_res_gpointer (res);
1133   op->count_written = class->write (G_OUTPUT_STREAM (object), op->buffer, op->count_requested,
1134                                     cancellable, &error);
1135   if (op->count_written == -1)
1136     {
1137       g_simple_async_result_set_from_error (res, error);
1138       g_error_free (error);
1139     }
1140 }
1141
1142 static void
1143 g_output_stream_real_write_async (GOutputStream       *stream,
1144                                   const void          *buffer,
1145                                   gsize                count,
1146                                   int                  io_priority,
1147                                   GCancellable        *cancellable,
1148                                   GAsyncReadyCallback  callback,
1149                                   gpointer             user_data)
1150 {
1151   GSimpleAsyncResult *res;
1152   WriteData *op;
1153
1154   op = g_new0 (WriteData, 1);
1155   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_write_async);
1156   g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1157   op->buffer = buffer;
1158   op->count_requested = count;
1159
1160   g_simple_async_result_run_in_thread (res, write_async_thread, io_priority, cancellable);
1161   g_object_unref (res);
1162 }
1163
1164 static gssize
1165 g_output_stream_real_write_finish (GOutputStream *stream,
1166                                    GAsyncResult *result,
1167                                    GError **error)
1168 {
1169   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1170   WriteData *op;
1171
1172   g_assert (g_simple_async_result_get_source_tag (simple) == g_output_stream_real_write_async);
1173   op = g_simple_async_result_get_op_res_gpointer (simple);
1174   return op->count_written;
1175 }
1176
1177 typedef struct {
1178   GInputStream *source;
1179   GOutputStreamSpliceFlags flags;
1180   gssize bytes_copied;
1181 } SpliceData;
1182
1183 static void
1184 splice_async_thread (GSimpleAsyncResult *result,
1185                      GObject *object,
1186                      GCancellable *cancellable)
1187 {
1188   SpliceData *op;
1189   GOutputStreamClass *class;
1190   GError *error = NULL;
1191   GOutputStream *stream;
1192
1193   stream = G_OUTPUT_STREAM (object);
1194   class = G_OUTPUT_STREAM_GET_CLASS (object);
1195   op = g_simple_async_result_get_op_res_gpointer (result);
1196
1197   stream->priv->pending = FALSE;
1198   op->bytes_copied =
1199     g_output_stream_splice (stream,
1200                             op->source,
1201                             op->flags,
1202                             cancellable,
1203                             &error);
1204   stream->priv->pending = TRUE;
1205
1206   if (op->bytes_copied == -1)
1207     {
1208       g_simple_async_result_set_from_error (result, error);
1209       g_error_free (error);
1210     }
1211 }
1212
1213 static void
1214 g_output_stream_real_splice_async  (GOutputStream             *stream,
1215                                     GInputStream              *source,
1216                                     GOutputStreamSpliceFlags   flags,
1217                                     int                        io_priority,
1218                                     GCancellable              *cancellable,
1219                                     GAsyncReadyCallback        callback,
1220                                     gpointer                   user_data)
1221 {
1222   GSimpleAsyncResult *res;
1223   SpliceData *op;
1224
1225   op = g_new0 (SpliceData, 1);
1226   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_splice_async);
1227   g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1228   op->flags = flags;
1229   op->source = source;
1230
1231   /* TODO: In the case where both source and destintion have
1232      non-threadbased async calls we can use a true async copy here */
1233
1234   g_simple_async_result_run_in_thread (res, splice_async_thread, io_priority, cancellable);
1235   g_object_unref (res);
1236 }
1237
1238 static gssize
1239 g_output_stream_real_splice_finish (GOutputStream             *stream,
1240                                     GAsyncResult              *result,
1241                                     GError                   **error)
1242 {
1243   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1244   SpliceData *op;
1245
1246   g_assert (g_simple_async_result_get_source_tag (simple) == g_output_stream_real_splice_async);
1247   op = g_simple_async_result_get_op_res_gpointer (simple);
1248   return op->bytes_copied;
1249 }
1250
1251
1252
1253 static void
1254 flush_async_thread (GSimpleAsyncResult *res,
1255                     GObject *object,
1256                     GCancellable *cancellable)
1257 {
1258   GOutputStreamClass *class;
1259   gboolean result;
1260   GError *error = NULL;
1261
1262   class = G_OUTPUT_STREAM_GET_CLASS (object);
1263   result = TRUE;
1264   if (class->flush)
1265     result = class->flush (G_OUTPUT_STREAM (object), cancellable, &error);
1266
1267   if (!result)
1268     {
1269       g_simple_async_result_set_from_error (res, error);
1270       g_error_free (error);
1271     }
1272 }
1273
1274 static void
1275 g_output_stream_real_flush_async (GOutputStream       *stream,
1276                                   int                  io_priority,
1277                                   GCancellable        *cancellable,
1278                                   GAsyncReadyCallback  callback,
1279                                   gpointer             user_data)
1280 {
1281   GSimpleAsyncResult *res;
1282
1283   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_write_async);
1284
1285   g_simple_async_result_run_in_thread (res, flush_async_thread, io_priority, cancellable);
1286   g_object_unref (res);
1287 }
1288
1289 static gboolean
1290 g_output_stream_real_flush_finish (GOutputStream *stream,
1291                                    GAsyncResult *result,
1292                                    GError **error)
1293 {
1294   return TRUE;
1295 }
1296
1297 static void
1298 close_async_thread (GSimpleAsyncResult *res,
1299                     GObject *object,
1300                     GCancellable *cancellable)
1301 {
1302   GOutputStreamClass *class;
1303   GError *error = NULL;
1304   gboolean result;
1305
1306   /* Auto handling of cancelation disabled, and ignore
1307      cancellation, since we want to close things anyway, although
1308      possibly in a quick-n-dirty way. At least we never want to leak
1309      open handles */
1310
1311   class = G_OUTPUT_STREAM_GET_CLASS (object);
1312   result = class->close (G_OUTPUT_STREAM (object), cancellable, &error);
1313   if (!result)
1314     {
1315       g_simple_async_result_set_from_error (res, error);
1316       g_error_free (error);
1317     }
1318 }
1319
1320 static void
1321 g_output_stream_real_close_async (GOutputStream      *stream,
1322                                   int                 io_priority,
1323                                   GCancellable       *cancellable,
1324                                   GAsyncReadyCallback callback,
1325                                   gpointer            user_data)
1326 {
1327   GSimpleAsyncResult *res;
1328
1329   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_close_async);
1330
1331   g_simple_async_result_set_handle_cancellation (res, FALSE);
1332
1333   g_simple_async_result_run_in_thread (res, close_async_thread, io_priority, cancellable);
1334   g_object_unref (res);
1335 }
1336
1337 static gboolean
1338 g_output_stream_real_close_finish (GOutputStream              *stream,
1339                                    GAsyncResult              *result,
1340                                    GError                   **error)
1341 {
1342   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1343   g_assert (g_simple_async_result_get_source_tag (simple) == g_output_stream_real_close_async);
1344   return TRUE;
1345 }