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