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  * @callback: callback to call when the request is satisfied
 972  * @user_data: the data to pass to callback function
 973  * @cancellable: optional cancellable object
 974  *
 975  * Requests an asynchronous closes of the stream, releasing resources related to it.
 976  * When the operation is finished @callback will be called, giving the results.
 977  *
 978  * For behaviour details see g_output_stream_close().
 979  *
 980  * The asyncronous methods have a default fallback that uses threads to implement
 981  * asynchronicity, so they are optional for inheriting classes. However, if you
 982  * override one you must override all.
 983  **/
 984 void
 985 g_output_stream_close_async (GOutputStream      *stream,
 986                              int                 io_priority,
 987                              GCancellable       *cancellable,
 988                              GAsyncReadyCallback callback,
 989                              gpointer            user_data)
 990 {
 991   GOutputStreamClass *class;
 992   GSimpleAsyncResult *simple;
 993
 994   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
 995
 996   if (stream->priv->closed)
 997     {
 998       simple = g_simple_async_result_new (G_OBJECT (stream),
 999                                           callback,
1000                                           user_data,
1001                                           g_output_stream_close_async);
1002       g_simple_async_result_complete_in_idle (simple);
1003       g_object_unref (simple);
1004       return;
1005     }
1006
1007   if (stream->priv->pending)
1008     {
1009       g_simple_async_report_error_in_idle (G_OBJECT (stream),
1010                                            callback,
1011                                            user_data,
1012                                            G_IO_ERROR, G_IO_ERROR_PENDING,
1013                                            _("Stream has outstanding operation"));
1014       return;
1015     }
1016
1017   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1018   stream->priv->pending = TRUE;
1019   stream->priv->outstanding_callback = callback;
1020   g_object_ref (stream);
1021   class->close_async (stream, io_priority, cancellable,
1022                       async_ready_close_callback_wrapper, user_data);
1023 }
1024
1025 /**
1026  * g_output_stream_close_finish:
1027  * @stream: a #GOutputStream.
1028  * @result: a #GAsyncResult.
1029  * @error: a #GError location to store the error occuring, or %NULL to
1030  * ignore.
1031  *
1032  * Closes an output stream.
1033  *
1034  * Returns: %TRUE if stream was successfully closed, %FALSE otherwise.
1035  **/
1036 gboolean
1037 g_output_stream_close_finish (GOutputStream *stream,
1038                               GAsyncResult *result,
1039                               GError **error)
1040 {
1041   GSimpleAsyncResult *simple;
1042   GOutputStreamClass *class;
1043
1044   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1045   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
1046
1047   if (G_IS_SIMPLE_ASYNC_RESULT (result))
1048     {
1049       simple = G_SIMPLE_ASYNC_RESULT (result);
1050       if (g_simple_async_result_propagate_error (simple, error))
1051         return FALSE;
1052
1053       /* Special case already closed */
1054       if (g_simple_async_result_get_source_tag (simple) == g_output_stream_close_async)
1055         return TRUE;
1056     }
1057
1058   class = G_OUTPUT_STREAM_GET_CLASS (stream);
1059   return class->close_finish (stream, result, error);
1060 }
1061
1062 /**
1063  * g_output_stream_is_closed:
1064  * @stream: a #GOutputStream.
1065  *
1066  * Checks if an output stream has already been closed.
1067  *
1068  * Returns: %TRUE if @stream is closed. %FALSE otherwise.
1069  **/
1070 gboolean
1071 g_output_stream_is_closed (GOutputStream *stream)
1072 {
1073   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), TRUE);
1074
1075   return stream->priv->closed;
1076 }
1077
1078 /**
1079  * g_output_stream_has_pending:
1080  * @stream: a #GOutputStream.
1081  *
1082  * Checks if an ouput stream has pending actions.
1083  *
1084  * Returns: %TRUE if @stream has pending actions.
1085  **/
1086 gboolean
1087 g_output_stream_has_pending (GOutputStream *stream)
1088 {
1089   g_return_val_if_fail (G_IS_OUTPUT_STREAM (stream), FALSE);
1090
1091   return stream->priv->pending;
1092 }
1093
1094 /**
1095  * g_output_stream_set_pending:
1096  * @stream: a #GOutputStream.
1097  * @pending: a #gboolean.
1098  *
1099  * Sets the @stream as having pending actions if @pending is %TRUE.
1100  **/
1101 void
1102 g_output_stream_set_pending (GOutputStream              *stream,
1103                             gboolean                   pending)
1104 {
1105   g_return_if_fail (G_IS_OUTPUT_STREAM (stream));
1106
1107   stream->priv->pending = pending;
1108 }
1109
1110
1111 /********************************************
1112  *   Default implementation of async ops    *
1113  ********************************************/
1114
1115 typedef struct {
1116   const void         *buffer;
1117   gsize               count_requested;
1118   gssize              count_written;
1119 } WriteData;
1120
1121 static void
1122 write_async_thread (GSimpleAsyncResult *res,
1123                    GObject *object,
1124                    GCancellable *cancellable)
1125 {
1126   WriteData *op;
1127   GOutputStreamClass *class;
1128   GError *error = NULL;
1129
1130   class = G_OUTPUT_STREAM_GET_CLASS (object);
1131   op = g_simple_async_result_get_op_res_gpointer (res);
1132   op->count_written = class->write (G_OUTPUT_STREAM (object), op->buffer, op->count_requested,
1133                                     cancellable, &error);
1134   if (op->count_written == -1)
1135     {
1136       g_simple_async_result_set_from_error (res, error);
1137       g_error_free (error);
1138     }
1139 }
1140
1141 static void
1142 g_output_stream_real_write_async (GOutputStream       *stream,
1143                                   const void          *buffer,
1144                                   gsize                count,
1145                                   int                  io_priority,
1146                                   GCancellable        *cancellable,
1147                                   GAsyncReadyCallback  callback,
1148                                   gpointer             user_data)
1149 {
1150   GSimpleAsyncResult *res;
1151   WriteData *op;
1152
1153   op = g_new0 (WriteData, 1);
1154   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_write_async);
1155   g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1156   op->buffer = buffer;
1157   op->count_requested = count;
1158
1159   g_simple_async_result_run_in_thread (res, write_async_thread, io_priority, cancellable);
1160   g_object_unref (res);
1161 }
1162
1163 static gssize
1164 g_output_stream_real_write_finish (GOutputStream *stream,
1165                                    GAsyncResult *result,
1166                                    GError **error)
1167 {
1168   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1169   WriteData *op;
1170
1171   g_assert (g_simple_async_result_get_source_tag (simple) == g_output_stream_real_write_async);
1172   op = g_simple_async_result_get_op_res_gpointer (simple);
1173   return op->count_written;
1174 }
1175
1176 typedef struct {
1177   GInputStream *source;
1178   GOutputStreamSpliceFlags flags;
1179   gssize bytes_copied;
1180 } SpliceData;
1181
1182 static void
1183 splice_async_thread (GSimpleAsyncResult *result,
1184                      GObject *object,
1185                      GCancellable *cancellable)
1186 {
1187   SpliceData *op;
1188   GOutputStreamClass *class;
1189   GError *error = NULL;
1190   GOutputStream *stream;
1191
1192   stream = G_OUTPUT_STREAM (object);
1193   class = G_OUTPUT_STREAM_GET_CLASS (object);
1194   op = g_simple_async_result_get_op_res_gpointer (result);
1195
1196   stream->priv->pending = FALSE;
1197   op->bytes_copied =
1198     g_output_stream_splice (stream,
1199                             op->source,
1200                             op->flags,
1201                             cancellable,
1202                             &error);
1203   stream->priv->pending = TRUE;
1204
1205   if (op->bytes_copied == -1)
1206     {
1207       g_simple_async_result_set_from_error (result, error);
1208       g_error_free (error);
1209     }
1210 }
1211
1212 static void
1213 g_output_stream_real_splice_async  (GOutputStream             *stream,
1214                                     GInputStream              *source,
1215                                     GOutputStreamSpliceFlags   flags,
1216                                     int                        io_priority,
1217                                     GCancellable              *cancellable,
1218                                     GAsyncReadyCallback        callback,
1219                                     gpointer                   user_data)
1220 {
1221   GSimpleAsyncResult *res;
1222   SpliceData *op;
1223
1224   op = g_new0 (SpliceData, 1);
1225   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_splice_async);
1226   g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1227   op->flags = flags;
1228   op->source = source;
1229
1230   /* TODO: In the case where both source and destintion have
1231      non-threadbased async calls we can use a true async copy here */
1232
1233   g_simple_async_result_run_in_thread (res, splice_async_thread, io_priority, cancellable);
1234   g_object_unref (res);
1235 }
1236
1237 static gssize
1238 g_output_stream_real_splice_finish (GOutputStream             *stream,
1239                                     GAsyncResult              *result,
1240                                     GError                   **error)
1241 {
1242   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1243   SpliceData *op;
1244
1245   g_assert (g_simple_async_result_get_source_tag (simple) == g_output_stream_real_splice_async);
1246   op = g_simple_async_result_get_op_res_gpointer (simple);
1247   return op->bytes_copied;
1248 }
1249
1250
1251
1252 static void
1253 flush_async_thread (GSimpleAsyncResult *res,
1254                     GObject *object,
1255                     GCancellable *cancellable)
1256 {
1257   GOutputStreamClass *class;
1258   gboolean result;
1259   GError *error = NULL;
1260
1261   class = G_OUTPUT_STREAM_GET_CLASS (object);
1262   result = TRUE;
1263   if (class->flush)
1264     result = class->flush (G_OUTPUT_STREAM (object), cancellable, &error);
1265
1266   if (!result)
1267     {
1268       g_simple_async_result_set_from_error (res, error);
1269       g_error_free (error);
1270     }
1271 }
1272
1273 static void
1274 g_output_stream_real_flush_async (GOutputStream       *stream,
1275                                   int                  io_priority,
1276                                   GCancellable        *cancellable,
1277                                   GAsyncReadyCallback  callback,
1278                                   gpointer             user_data)
1279 {
1280   GSimpleAsyncResult *res;
1281
1282   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_write_async);
1283
1284   g_simple_async_result_run_in_thread (res, flush_async_thread, io_priority, cancellable);
1285   g_object_unref (res);
1286 }
1287
1288 static gboolean
1289 g_output_stream_real_flush_finish (GOutputStream *stream,
1290                                    GAsyncResult *result,
1291                                    GError **error)
1292 {
1293   return TRUE;
1294 }
1295
1296 static void
1297 close_async_thread (GSimpleAsyncResult *res,
1298                     GObject *object,
1299                     GCancellable *cancellable)
1300 {
1301   GOutputStreamClass *class;
1302   GError *error = NULL;
1303   gboolean result;
1304
1305   /* Auto handling of cancelation disabled, and ignore
1306      cancellation, since we want to close things anyway, although
1307      possibly in a quick-n-dirty way. At least we never want to leak
1308      open handles */
1309
1310   class = G_OUTPUT_STREAM_GET_CLASS (object);
1311   result = class->close (G_OUTPUT_STREAM (object), cancellable, &error);
1312   if (!result)
1313     {
1314       g_simple_async_result_set_from_error (res, error);
1315       g_error_free (error);
1316     }
1317 }
1318
1319 static void
1320 g_output_stream_real_close_async (GOutputStream      *stream,
1321                                   int                 io_priority,
1322                                   GCancellable       *cancellable,
1323                                   GAsyncReadyCallback callback,
1324                                   gpointer            user_data)
1325 {
1326   GSimpleAsyncResult *res;
1327
1328   res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_output_stream_real_close_async);
1329
1330   g_simple_async_result_set_handle_cancellation (res, FALSE);
1331
1332   g_simple_async_result_run_in_thread (res, close_async_thread, io_priority, cancellable);
1333   g_object_unref (res);
1334 }
1335
1336 static gboolean
1337 g_output_stream_real_close_finish (GOutputStream              *stream,
1338                                    GAsyncResult              *result,
1339                                    GError                   **error)
1340 {
1341   GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1342   g_assert (g_simple_async_result_get_source_tag (simple) == g_output_stream_real_close_async);
1343   return TRUE;
1344 }