gio/gbufferedoutputstream.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.1 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Lesser General
  16  * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.
  17  *
  18  * Author: Christian Kellner <gicmo@gnome.org>
  19  */
  20
  21 #include "config.h"
  22 #include "gbufferedoutputstream.h"
  23 #include "goutputstream.h"
  24 #include "gseekable.h"
  25 #include "gtask.h"
  26 #include "string.h"
  27 #include "gioerror.h"
  28 #include "glibintl.h"
  29
  30 /**
  31  * SECTION:gbufferedoutputstream
  32  * @short_description: Buffered Output Stream
  33  * @include: gio/gio.h
  34  * @see_also: #GFilterOutputStream, #GOutputStream
  35  *
  36  * Buffered output stream implements #GFilterOutputStream and provides
  37  * for buffered writes.
  38  *
  39  * By default, #GBufferedOutputStream's buffer size is set at 4 kilobytes.
  40  *
  41  * To create a buffered output stream, use g_buffered_output_stream_new(),
  42  * or g_buffered_output_stream_new_sized() to specify the buffer's size
  43  * at construction.
  44  *
  45  * To get the size of a buffer within a buffered input stream, use
  46  * g_buffered_output_stream_get_buffer_size(). To change the size of a
  47  * buffered output stream's buffer, use
  48  * g_buffered_output_stream_set_buffer_size(). Note that the buffer's
  49  * size cannot be reduced below the size of the data within the buffer.
  50  **/
  51
  52 #define DEFAULT_BUFFER_SIZE 4096
  53
  54 struct _GBufferedOutputStreamPrivate {
  55   guint8 *buffer;
  56   gsize   len;
  57   goffset pos;
  58   gboolean auto_grow;
  59 };
  60
  61 enum {
  62   PROP_0,
  63   PROP_BUFSIZE,
  64   PROP_AUTO_GROW
  65 };
  66
  67 static void     g_buffered_output_stream_set_property (GObject      *object,
  68                                                        guint         prop_id,
  69                                                        const GValue *value,
  70                                                        GParamSpec   *pspec);
  71
  72 static void     g_buffered_output_stream_get_property (GObject    *object,
  73                                                        guint       prop_id,
  74                                                        GValue     *value,
  75                                                        GParamSpec *pspec);
  76 static void     g_buffered_output_stream_finalize     (GObject *object);
  77
  78
  79 static gssize   g_buffered_output_stream_write        (GOutputStream *stream,
  80                                                        const void    *buffer,
  81                                                        gsize          count,
  82                                                        GCancellable  *cancellable,
  83                                                        GError       **error);
  84 static gboolean g_buffered_output_stream_flush        (GOutputStream    *stream,
  85                                                        GCancellable  *cancellable,
  86                                                        GError          **error);
  87 static gboolean g_buffered_output_stream_close        (GOutputStream  *stream,
  88                                                        GCancellable   *cancellable,
  89                                                        GError        **error);
  90
  91 static void     g_buffered_output_stream_flush_async  (GOutputStream        *stream,
  92                                                        int                   io_priority,
  93                                                        GCancellable         *cancellable,
  94                                                        GAsyncReadyCallback   callback,
  95                                                        gpointer              data);
  96 static gboolean g_buffered_output_stream_flush_finish (GOutputStream        *stream,
  97                                                        GAsyncResult         *result,
  98                                                        GError              **error);
  99 static void     g_buffered_output_stream_close_async  (GOutputStream        *stream,
 100                                                        int                   io_priority,
 101                                                        GCancellable         *cancellable,
 102                                                        GAsyncReadyCallback   callback,
 103                                                        gpointer              data);
 104 static gboolean g_buffered_output_stream_close_finish (GOutputStream        *stream,
 105                                                        GAsyncResult         *result,
 106                                                        GError              **error);
 107
 108 static void     g_buffered_output_stream_seekable_iface_init (GSeekableIface  *iface);
 109 static goffset  g_buffered_output_stream_tell                (GSeekable       *seekable);
 110 static gboolean g_buffered_output_stream_can_seek            (GSeekable       *seekable);
 111 static gboolean g_buffered_output_stream_seek                (GSeekable       *seekable,
 112                                                               goffset          offset,
 113                                                               GSeekType        type,
 114                                                               GCancellable    *cancellable,
 115                                                               GError         **error);
 116 static gboolean g_buffered_output_stream_can_truncate        (GSeekable       *seekable);
 117 static gboolean g_buffered_output_stream_truncate            (GSeekable       *seekable,
 118                                                               goffset          offset,
 119                                                               GCancellable    *cancellable,
 120                                                               GError         **error);
 121
 122 G_DEFINE_TYPE_WITH_CODE (GBufferedOutputStream,
 123                          g_buffered_output_stream,
 124                          G_TYPE_FILTER_OUTPUT_STREAM,
 125                          G_ADD_PRIVATE (GBufferedOutputStream)
 126                          G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE,
 127                                                 g_buffered_output_stream_seekable_iface_init))
 128
 129
 130 static void
 131 g_buffered_output_stream_class_init (GBufferedOutputStreamClass *klass)
 132 {
 133   GObjectClass *object_class;
 134   GOutputStreamClass *ostream_class;
 135
 136   object_class = G_OBJECT_CLASS (klass);
 137   object_class->get_property = g_buffered_output_stream_get_property;
 138   object_class->set_property = g_buffered_output_stream_set_property;
 139   object_class->finalize     = g_buffered_output_stream_finalize;
 140
 141   ostream_class = G_OUTPUT_STREAM_CLASS (klass);
 142   ostream_class->write_fn = g_buffered_output_stream_write;
 143   ostream_class->flush = g_buffered_output_stream_flush;
 144   ostream_class->close_fn = g_buffered_output_stream_close;
 145   ostream_class->flush_async  = g_buffered_output_stream_flush_async;
 146   ostream_class->flush_finish = g_buffered_output_stream_flush_finish;
 147   ostream_class->close_async  = g_buffered_output_stream_close_async;
 148   ostream_class->close_finish = g_buffered_output_stream_close_finish;
 149
 150   g_object_class_install_property (object_class,
 151                                    PROP_BUFSIZE,
 152                                    g_param_spec_uint ("buffer-size",
 153                                                       P_("Buffer Size"),
 154                                                       P_("The size of the backend buffer"),
 155                                                       1,
 156                                                       G_MAXUINT,
 157                                                       DEFAULT_BUFFER_SIZE,
 158                                                       G_PARAM_READWRITE|G_PARAM_CONSTRUCT|
 159                                                       G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
 160
 161   g_object_class_install_property (object_class,
 162                                    PROP_AUTO_GROW,
 163                                    g_param_spec_boolean ("auto-grow",
 164                                                          P_("Auto-grow"),
 165                                                          P_("Whether the buffer should automatically grow"),
 166                                                          FALSE,
 167                                                          G_PARAM_READWRITE|
 168                                                          G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
 169
 170 }
 171
 172 /**
 173  * g_buffered_output_stream_get_buffer_size:
 174  * @stream: a #GBufferedOutputStream.
 175  *
 176  * Gets the size of the buffer in the @stream.
 177  *
 178  * Returns: the current size of the buffer.
 179  **/
 180 gsize
 181 g_buffered_output_stream_get_buffer_size (GBufferedOutputStream *stream)
 182 {
 183   g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream), -1);
 184
 185   return stream->priv->len;
 186 }
 187
 188 /**
 189  * g_buffered_output_stream_set_buffer_size:
 190  * @stream: a #GBufferedOutputStream.
 191  * @size: a #gsize.
 192  *
 193  * Sets the size of the internal buffer to @size.
 194  **/
 195 void
 196 g_buffered_output_stream_set_buffer_size (GBufferedOutputStream *stream,
 197                                           gsize                  size)
 198 {
 199   GBufferedOutputStreamPrivate *priv;
 200   guint8 *buffer;
 201
 202   g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream));
 203
 204   priv = stream->priv;
 205
 206   if (size == priv->len)
 207     return;
 208
 209   if (priv->buffer)
 210     {
 211       size = MAX (size, priv->pos);
 212
 213       buffer = g_malloc (size);
 214       memcpy (buffer, priv->buffer, priv->pos);
 215       g_free (priv->buffer);
 216       priv->buffer = buffer;
 217       priv->len = size;
 218       /* Keep old pos */
 219     }
 220   else
 221     {
 222       priv->buffer = g_malloc (size);
 223       priv->len = size;
 224       priv->pos = 0;
 225     }
 226
 227   g_object_notify (G_OBJECT (stream), "buffer-size");
 228 }
 229
 230 /**
 231  * g_buffered_output_stream_get_auto_grow:
 232  * @stream: a #GBufferedOutputStream.
 233  *
 234  * Checks if the buffer automatically grows as data is added.
 235  *
 236  * Returns: %TRUE if the @stream's buffer automatically grows,
 237  * %FALSE otherwise.
 238  **/
 239 gboolean
 240 g_buffered_output_stream_get_auto_grow (GBufferedOutputStream *stream)
 241 {
 242   g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream), FALSE);
 243
 244   return stream->priv->auto_grow;
 245 }
 246
 247 /**
 248  * g_buffered_output_stream_set_auto_grow:
 249  * @stream: a #GBufferedOutputStream.
 250  * @auto_grow: a #gboolean.
 251  *
 252  * Sets whether or not the @stream's buffer should automatically grow.
 253  * If @auto_grow is true, then each write will just make the buffer
 254  * larger, and you must manually flush the buffer to actually write out
 255  * the data to the underlying stream.
 256  **/
 257 void
 258 g_buffered_output_stream_set_auto_grow (GBufferedOutputStream *stream,
 259                                         gboolean               auto_grow)
 260 {
 261   GBufferedOutputStreamPrivate *priv;
 262   g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream));
 263   priv = stream->priv;
 264   auto_grow = auto_grow != FALSE;
 265   if (priv->auto_grow != auto_grow)
 266     {
 267       priv->auto_grow = auto_grow;
 268       g_object_notify (G_OBJECT (stream), "auto-grow");
 269     }
 270 }
 271
 272 static void
 273 g_buffered_output_stream_set_property (GObject      *object,
 274                                        guint         prop_id,
 275                                        const GValue *value,
 276                                        GParamSpec   *pspec)
 277 {
 278   GBufferedOutputStream *stream;
 279
 280   stream = G_BUFFERED_OUTPUT_STREAM (object);
 281
 282   switch (prop_id)
 283     {
 284     case PROP_BUFSIZE:
 285       g_buffered_output_stream_set_buffer_size (stream, g_value_get_uint (value));
 286       break;
 287
 288     case PROP_AUTO_GROW:
 289       g_buffered_output_stream_set_auto_grow (stream, g_value_get_boolean (value));
 290       break;
 291
 292     default:
 293       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 294       break;
 295     }
 296
 297 }
 298
 299 static void
 300 g_buffered_output_stream_get_property (GObject    *object,
 301                                        guint       prop_id,
 302                                        GValue     *value,
 303                                        GParamSpec *pspec)
 304 {
 305   GBufferedOutputStream *buffered_stream;
 306   GBufferedOutputStreamPrivate *priv;
 307
 308   buffered_stream = G_BUFFERED_OUTPUT_STREAM (object);
 309   priv = buffered_stream->priv;
 310
 311   switch (prop_id)
 312     {
 313     case PROP_BUFSIZE:
 314       g_value_set_uint (value, priv->len);
 315       break;
 316
 317     case PROP_AUTO_GROW:
 318       g_value_set_boolean (value, priv->auto_grow);
 319       break;
 320
 321     default:
 322       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 323       break;
 324     }
 325
 326 }
 327
 328 static void
 329 g_buffered_output_stream_finalize (GObject *object)
 330 {
 331   GBufferedOutputStream *stream;
 332   GBufferedOutputStreamPrivate *priv;
 333
 334   stream = G_BUFFERED_OUTPUT_STREAM (object);
 335   priv = stream->priv;
 336
 337   g_free (priv->buffer);
 338
 339   G_OBJECT_CLASS (g_buffered_output_stream_parent_class)->finalize (object);
 340 }
 341
 342 static void
 343 g_buffered_output_stream_init (GBufferedOutputStream *stream)
 344 {
 345   stream->priv = g_buffered_output_stream_get_instance_private (stream);
 346 }
 347
 348 static void
 349 g_buffered_output_stream_seekable_iface_init (GSeekableIface *iface)
 350 {
 351   iface->tell         = g_buffered_output_stream_tell;
 352   iface->can_seek     = g_buffered_output_stream_can_seek;
 353   iface->seek         = g_buffered_output_stream_seek;
 354   iface->can_truncate = g_buffered_output_stream_can_truncate;
 355   iface->truncate_fn  = g_buffered_output_stream_truncate;
 356 }
 357
 358 /**
 359  * g_buffered_output_stream_new:
 360  * @base_stream: a #GOutputStream.
 361  *
 362  * Creates a new buffered output stream for a base stream.
 363  *
 364  * Returns: a #GOutputStream for the given @base_stream.
 365  **/
 366 GOutputStream *
 367 g_buffered_output_stream_new (GOutputStream *base_stream)
 368 {
 369   GOutputStream *stream;
 370
 371   g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), NULL);
 372
 373   stream = g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM,
 374                          "base-stream", base_stream,
 375                          NULL);
 376
 377   return stream;
 378 }
 379
 380 /**
 381  * g_buffered_output_stream_new_sized:
 382  * @base_stream: a #GOutputStream.
 383  * @size: a #gsize.
 384  *
 385  * Creates a new buffered output stream with a given buffer size.
 386  *
 387  * Returns: a #GOutputStream with an internal buffer set to @size.
 388  **/
 389 GOutputStream *
 390 g_buffered_output_stream_new_sized (GOutputStream *base_stream,
 391                                     gsize          size)
 392 {
 393   GOutputStream *stream;
 394
 395   g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), NULL);
 396
 397   stream = g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM,
 398                          "base-stream", base_stream,
 399                          "buffer-size", size,
 400                          NULL);
 401
 402   return stream;
 403 }
 404
 405 static gboolean
 406 flush_buffer (GBufferedOutputStream  *stream,
 407               GCancellable           *cancellable,
 408               GError                 **error)
 409 {
 410   GBufferedOutputStreamPrivate *priv;
 411   GOutputStream                *base_stream;
 412   gboolean                      res;
 413   gsize                         bytes_written;
 414   gsize                         count;
 415
 416   priv = stream->priv;
 417   bytes_written = 0;
 418   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 419
 420   g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), FALSE);
 421
 422   res = g_output_stream_write_all (base_stream,
 423                                    priv->buffer,
 424                                    priv->pos,
 425                                    &bytes_written,
 426                                    cancellable,
 427                                    error);
 428
 429   count = priv->pos - bytes_written;
 430
 431   if (count > 0)
 432     memmove (priv->buffer, priv->buffer + bytes_written, count);
 433
 434   priv->pos -= bytes_written;
 435
 436   return res;
 437 }
 438
 439 static gssize
 440 g_buffered_output_stream_write  (GOutputStream *stream,
 441                                  const void    *buffer,
 442                                  gsize          count,
 443                                  GCancellable  *cancellable,
 444                                  GError       **error)
 445 {
 446   GBufferedOutputStream        *bstream;
 447   GBufferedOutputStreamPrivate *priv;
 448   gboolean res;
 449   gsize    n;
 450   gsize new_size;
 451
 452   bstream = G_BUFFERED_OUTPUT_STREAM (stream);
 453   priv = bstream->priv;
 454
 455   n = priv->len - priv->pos;
 456
 457   if (priv->auto_grow && n < count)
 458     {
 459       new_size = MAX (priv->len * 2, priv->len + count);
 460       g_buffered_output_stream_set_buffer_size (bstream, new_size);
 461     }
 462   else if (n == 0)
 463     {
 464       res = flush_buffer (bstream, cancellable, error);
 465
 466       if (res == FALSE)
 467         return -1;
 468     }
 469
 470   n = priv->len - priv->pos;
 471
 472   count = MIN (count, n);
 473   memcpy (priv->buffer + priv->pos, buffer, count);
 474   priv->pos += count;
 475
 476   return count;
 477 }
 478
 479 static gboolean
 480 g_buffered_output_stream_flush (GOutputStream  *stream,
 481                                 GCancellable   *cancellable,
 482                                 GError        **error)
 483 {
 484   GBufferedOutputStream *bstream;
 485   GOutputStream                *base_stream;
 486   gboolean res;
 487
 488   bstream = G_BUFFERED_OUTPUT_STREAM (stream);
 489   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 490
 491   res = flush_buffer (bstream, cancellable, error);
 492
 493   if (res == FALSE)
 494     return FALSE;
 495
 496   res = g_output_stream_flush (base_stream, cancellable, error);
 497
 498   return res;
 499 }
 500
 501 static gboolean
 502 g_buffered_output_stream_close (GOutputStream  *stream,
 503                                 GCancellable   *cancellable,
 504                                 GError        **error)
 505 {
 506   GBufferedOutputStream        *bstream;
 507   GOutputStream                *base_stream;
 508   gboolean                      res;
 509
 510   bstream = G_BUFFERED_OUTPUT_STREAM (stream);
 511   base_stream = G_FILTER_OUTPUT_STREAM (bstream)->base_stream;
 512   res = flush_buffer (bstream, cancellable, error);
 513
 514   if (g_filter_output_stream_get_close_base_stream (G_FILTER_OUTPUT_STREAM (stream)))
 515     {
 516       /* report the first error but still close the stream */
 517       if (res)
 518         res = g_output_stream_close (base_stream, cancellable, error);
 519       else
 520         g_output_stream_close (base_stream, cancellable, NULL);
 521     }
 522
 523   return res;
 524 }
 525
 526 static goffset
 527 g_buffered_output_stream_tell (GSeekable *seekable)
 528 {
 529   GBufferedOutputStream        *bstream;
 530   GBufferedOutputStreamPrivate *priv;
 531   GOutputStream *base_stream;
 532   GSeekable    *base_stream_seekable;
 533   goffset base_offset;
 534
 535   bstream = G_BUFFERED_OUTPUT_STREAM (seekable);
 536   priv = bstream->priv;
 537
 538   base_stream = G_FILTER_OUTPUT_STREAM (seekable)->base_stream;
 539   if (!G_IS_SEEKABLE (base_stream))
 540     return 0;
 541
 542   base_stream_seekable = G_SEEKABLE (base_stream);
 543
 544   base_offset = g_seekable_tell (base_stream_seekable);
 545   return base_offset + priv->pos;
 546 }
 547
 548 static gboolean
 549 g_buffered_output_stream_can_seek (GSeekable *seekable)
 550 {
 551   GOutputStream *base_stream;
 552
 553   base_stream = G_FILTER_OUTPUT_STREAM (seekable)->base_stream;
 554   return G_IS_SEEKABLE (base_stream) && g_seekable_can_seek (G_SEEKABLE (base_stream));
 555 }
 556
 557 static gboolean
 558 g_buffered_output_stream_seek (GSeekable     *seekable,
 559                                goffset        offset,
 560                                GSeekType      type,
 561                                GCancellable  *cancellable,
 562                                GError       **error)
 563 {
 564   GBufferedOutputStream *bstream;
 565   GOutputStream *base_stream;
 566   GSeekable *base_stream_seekable;
 567   gboolean flushed;
 568
 569   bstream = G_BUFFERED_OUTPUT_STREAM (seekable);
 570
 571   base_stream = G_FILTER_OUTPUT_STREAM (seekable)->base_stream;
 572   if (!G_IS_SEEKABLE (base_stream))
 573     {
 574       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
 575                            _("Seek not supported on base stream"));
 576       return FALSE;
 577     }
 578
 579   base_stream_seekable = G_SEEKABLE (base_stream);
 580   flushed = flush_buffer (bstream, cancellable, error);
 581   if (!flushed)
 582     return FALSE;
 583
 584   return g_seekable_seek (base_stream_seekable, offset, type, cancellable, error);
 585 }
 586
 587 static gboolean
 588 g_buffered_output_stream_can_truncate (GSeekable *seekable)
 589 {
 590   GOutputStream *base_stream;
 591
 592   base_stream = G_FILTER_OUTPUT_STREAM (seekable)->base_stream;
 593   return G_IS_SEEKABLE (base_stream) && g_seekable_can_truncate (G_SEEKABLE (base_stream));
 594 }
 595
 596 static gboolean
 597 g_buffered_output_stream_truncate (GSeekable     *seekable,
 598                                    goffset        offset,
 599                                    GCancellable  *cancellable,
 600                                    GError       **error)
 601 {
 602   GBufferedOutputStream        *bstream;
 603   GOutputStream *base_stream;
 604   GSeekable *base_stream_seekable;
 605   gboolean flushed;
 606
 607   bstream = G_BUFFERED_OUTPUT_STREAM (seekable);
 608   base_stream = G_FILTER_OUTPUT_STREAM (seekable)->base_stream;
 609   if (!G_IS_SEEKABLE (base_stream))
 610     {
 611       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
 612                            _("Truncate not supported on base stream"));
 613       return FALSE;
 614     }
 615
 616   base_stream_seekable = G_SEEKABLE (base_stream);
 617
 618   flushed = flush_buffer (bstream, cancellable, error);
 619   if (!flushed)
 620     return FALSE;
 621   return g_seekable_truncate (base_stream_seekable, offset, cancellable, error);
 622 }
 623
 624 /* ************************** */
 625 /* Async stuff implementation */
 626 /* ************************** */
 627
 628 /* TODO: This should be using the base class async ops, not threads */
 629
 630 typedef struct {
 631
 632   guint flush_stream : 1;
 633   guint close_stream : 1;
 634
 635 } FlushData;
 636
 637 static void
 638 free_flush_data (gpointer data)
 639 {
 640   g_slice_free (FlushData, data);
 641 }
 642
 643 /* This function is used by all three (i.e.
 644  * _write, _flush, _close) functions since
 645  * all of them will need to flush the buffer
 646  * and so closing and writing is just a special
 647  * case of flushing + some addition stuff */
 648 static void
 649 flush_buffer_thread (GTask        *task,
 650                      gpointer      object,
 651                      gpointer      task_data,
 652                      GCancellable *cancellable)
 653 {
 654   GBufferedOutputStream *stream;
 655   GOutputStream *base_stream;
 656   FlushData     *fdata;
 657   gboolean       res;
 658   GError        *error = NULL;
 659
 660   stream = G_BUFFERED_OUTPUT_STREAM (object);
 661   fdata = task_data;
 662   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 663
 664   res = flush_buffer (stream, cancellable, &error);
 665
 666   /* if flushing the buffer didn't work don't even bother
 667    * to flush the stream but just report that error */
 668   if (res && fdata->flush_stream)
 669     res = g_output_stream_flush (base_stream, cancellable, &error);
 670
 671   if (fdata->close_stream)
 672     {
 673
 674       /* if flushing the buffer or the stream returned
 675        * an error report that first error but still try
 676        * close the stream */
 677       if (g_filter_output_stream_get_close_base_stream (G_FILTER_OUTPUT_STREAM (stream)))
 678         {
 679           if (res == FALSE)
 680             g_output_stream_close (base_stream, cancellable, NULL);
 681           else
 682             res = g_output_stream_close (base_stream, cancellable, &error);
 683         }
 684     }
 685
 686   if (res == FALSE)
 687     g_task_return_error (task, error);
 688   else
 689     g_task_return_boolean (task, TRUE);
 690 }
 691
 692 static void
 693 g_buffered_output_stream_flush_async (GOutputStream        *stream,
 694                                       int                   io_priority,
 695                                       GCancellable         *cancellable,
 696                                       GAsyncReadyCallback   callback,
 697                                       gpointer              data)
 698 {
 699   GTask *task;
 700   FlushData *fdata;
 701
 702   fdata = g_slice_new0 (FlushData);
 703   fdata->flush_stream = TRUE;
 704   fdata->close_stream = FALSE;
 705
 706   task = g_task_new (stream, cancellable, callback, data);
 707   g_task_set_source_tag (task, g_buffered_output_stream_flush_async);
 708   g_task_set_task_data (task, fdata, free_flush_data);
 709   g_task_set_priority (task, io_priority);
 710
 711   g_task_run_in_thread (task, flush_buffer_thread);
 712   g_object_unref (task);
 713 }
 714
 715 static gboolean
 716 g_buffered_output_stream_flush_finish (GOutputStream        *stream,
 717                                        GAsyncResult         *result,
 718                                        GError              **error)
 719 {
 720   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
 721
 722   return g_task_propagate_boolean (G_TASK (result), error);
 723 }
 724
 725 static void
 726 g_buffered_output_stream_close_async (GOutputStream        *stream,
 727                                       int                   io_priority,
 728                                       GCancellable         *cancellable,
 729                                       GAsyncReadyCallback   callback,
 730                                       gpointer              data)
 731 {
 732   GTask *task;
 733   FlushData *fdata;
 734
 735   fdata = g_slice_new0 (FlushData);
 736   fdata->close_stream = TRUE;
 737
 738   task = g_task_new (stream, cancellable, callback, data);
 739   g_task_set_source_tag (task, g_buffered_output_stream_close_async);
 740   g_task_set_task_data (task, fdata, free_flush_data);
 741   g_task_set_priority (task, io_priority);
 742
 743   g_task_run_in_thread (task, flush_buffer_thread);
 744   g_object_unref (task);
 745 }
 746
 747 static gboolean
 748 g_buffered_output_stream_close_finish (GOutputStream        *stream,
 749                                        GAsyncResult         *result,
 750                                        GError              **error)
 751 {
 752   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
 753
 754   return g_task_propagate_boolean (G_TASK (result), error);
 755 }