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