ffa8fbd8f6f7ae6eeb1d4e6489c013eb6e6dfca2
[platform/upstream/glib.git] / 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  * @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_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_fn = g_buffered_output_stream_write;
138   ostream_class->flush = g_buffered_output_stream_flush;
139   ostream_class->close_fn = 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|
156                                                       G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
157
158   g_object_class_install_property (object_class,
159                                    PROP_AUTO_GROW,
160                                    g_param_spec_boolean ("auto-grow",
161                                                          P_("Auto-grow"),
162                                                          P_("Whether the buffer should automatically grow"),
163                                                          FALSE,
164                                                          G_PARAM_READWRITE|
165                                                          G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
166
167 }
168
169 /**
170  * g_buffered_output_stream_get_buffer_size:
171  * @stream: a #GBufferedOutputStream.
172  * 
173  * Gets the size of the buffer in the @stream.
174  * 
175  * Returns: the current size of the buffer.
176  **/
177 gsize
178 g_buffered_output_stream_get_buffer_size (GBufferedOutputStream *stream)
179 {
180   g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream), -1);
181
182   return stream->priv->len;
183 }
184
185 /**
186  * g_buffered_output_stream_set_buffer_size:
187  * @stream: a #GBufferedOutputStream.
188  * @size: a #gsize.
189  *
190  * Sets the size of the internal buffer to @size.
191  **/    
192 void
193 g_buffered_output_stream_set_buffer_size (GBufferedOutputStream *stream,
194                                           gsize                  size)
195 {
196   GBufferedOutputStreamPrivate *priv;
197   guint8 *buffer;
198   
199   g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream));
200
201   priv = stream->priv;
202   
203   if (size == priv->len)
204     return;
205
206   if (priv->buffer)
207     {
208       size = MAX (size, priv->pos);
209
210       buffer = g_malloc (size);
211       memcpy (buffer, priv->buffer, priv->pos);
212       g_free (priv->buffer);
213       priv->buffer = buffer;
214       priv->len = size;
215       /* Keep old pos */
216     }
217   else
218     {
219       priv->buffer = g_malloc (size);
220       priv->len = size;
221       priv->pos = 0;
222     }
223
224   g_object_notify (G_OBJECT (stream), "buffer-size");
225 }
226
227 /**
228  * g_buffered_output_stream_get_auto_grow:
229  * @stream: a #GBufferedOutputStream.
230  * 
231  * Checks if the buffer automatically grows as data is added.
232  * 
233  * Returns: %TRUE if the @stream's buffer automatically grows,
234  * %FALSE otherwise.
235  **/  
236 gboolean
237 g_buffered_output_stream_get_auto_grow (GBufferedOutputStream *stream)
238 {
239   g_return_val_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream), FALSE);
240
241   return stream->priv->auto_grow;
242 }
243
244 /**
245  * g_buffered_output_stream_set_auto_grow:
246  * @stream: a #GBufferedOutputStream.
247  * @auto_grow: a #gboolean.
248  *
249  * Sets whether or not the @stream's buffer should automatically grow.
250  * If @auto_grow is true, then each write will just make the buffer
251  * larger, and you must manually flush the buffer to actually write out
252  * the data to the underlying stream.
253  **/
254 void
255 g_buffered_output_stream_set_auto_grow (GBufferedOutputStream *stream,
256                                         gboolean               auto_grow)
257 {
258   GBufferedOutputStreamPrivate *priv;
259   g_return_if_fail (G_IS_BUFFERED_OUTPUT_STREAM (stream));
260   priv = stream->priv;
261   auto_grow = auto_grow != FALSE;
262   if (priv->auto_grow != auto_grow)
263     {
264       priv->auto_grow = auto_grow;
265       g_object_notify (G_OBJECT (stream), "auto-grow");
266     }
267 }
268
269 static void
270 g_buffered_output_stream_set_property (GObject      *object,
271                                        guint         prop_id,
272                                        const GValue *value,
273                                        GParamSpec   *pspec)
274 {
275   GBufferedOutputStream *stream;
276
277   stream = G_BUFFERED_OUTPUT_STREAM (object);
278
279   switch (prop_id) 
280     {
281     case PROP_BUFSIZE:
282       g_buffered_output_stream_set_buffer_size (stream, g_value_get_uint (value));
283       break;    
284
285     case PROP_AUTO_GROW:
286       g_buffered_output_stream_set_auto_grow (stream, g_value_get_boolean (value));
287       break;
288
289     default:
290       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
291       break;
292     }
293
294 }
295
296 static void
297 g_buffered_output_stream_get_property (GObject    *object,
298                                        guint       prop_id,
299                                        GValue     *value,
300                                        GParamSpec *pspec)
301 {
302   GBufferedOutputStream *buffered_stream;
303   GBufferedOutputStreamPrivate *priv;
304
305   buffered_stream = G_BUFFERED_OUTPUT_STREAM (object);
306   priv = buffered_stream->priv;
307
308   switch (prop_id)
309     {
310     case PROP_BUFSIZE:
311       g_value_set_uint (value, priv->len);
312       break;
313
314     case PROP_AUTO_GROW:
315       g_value_set_boolean (value, priv->auto_grow);
316       break;
317
318     default:
319       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
320       break;
321     }
322
323 }
324
325 static void
326 g_buffered_output_stream_finalize (GObject *object)
327 {
328   GBufferedOutputStream *stream;
329   GBufferedOutputStreamPrivate *priv;
330
331   stream = G_BUFFERED_OUTPUT_STREAM (object);
332   priv = stream->priv;
333
334   g_free (priv->buffer);
335
336   G_OBJECT_CLASS (g_buffered_output_stream_parent_class)->finalize (object);
337 }
338
339 static void
340 g_buffered_output_stream_init (GBufferedOutputStream *stream)
341 {
342   stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
343                                               G_TYPE_BUFFERED_OUTPUT_STREAM,
344                                               GBufferedOutputStreamPrivate);
345
346 }
347
348 /**
349  * g_buffered_output_stream_new:
350  * @base_stream: a #GOutputStream.
351  * 
352  * Creates a new buffered output stream for a base stream.
353  * 
354  * Returns: a #GOutputStream for the given @base_stream.
355  **/  
356 GOutputStream *
357 g_buffered_output_stream_new (GOutputStream *base_stream)
358 {
359   GOutputStream *stream;
360
361   g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), NULL);
362
363   stream = g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM,
364                          "base-stream", base_stream,
365                          NULL);
366
367   return stream;
368 }
369
370 /**
371  * g_buffered_output_stream_new_sized:
372  * @base_stream: a #GOutputStream.
373  * @size: a #gsize.
374  * 
375  * Creates a new buffered output stream with a given buffer size.
376  * 
377  * Returns: a #GOutputStream with an internal buffer set to @size.
378  **/  
379 GOutputStream *
380 g_buffered_output_stream_new_sized (GOutputStream *base_stream,
381                                     gsize          size)
382 {
383   GOutputStream *stream;
384
385   g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), NULL);
386
387   stream = g_object_new (G_TYPE_BUFFERED_OUTPUT_STREAM,
388                          "base-stream", base_stream,
389                          "buffer-size", size,
390                          NULL);
391
392   return stream;
393 }
394
395 static gboolean
396 flush_buffer (GBufferedOutputStream  *stream,
397               GCancellable           *cancellable,
398               GError                 **error)
399 {
400   GBufferedOutputStreamPrivate *priv;
401   GOutputStream                *base_stream;
402   gboolean                      res;
403   gsize                         bytes_written;
404   gsize                         count;
405
406   priv = stream->priv;
407   bytes_written = 0;
408   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
409
410   g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), FALSE);
411
412   res = g_output_stream_write_all (base_stream,
413                                    priv->buffer,
414                                    priv->pos,
415                                    &bytes_written,
416                                    cancellable,
417                                    error);
418
419   count = priv->pos - bytes_written;
420
421   if (count > 0)
422     g_memmove (priv->buffer, priv->buffer + bytes_written, count);
423   
424   priv->pos -= bytes_written;
425
426   return res;
427 }
428
429 static gssize
430 g_buffered_output_stream_write  (GOutputStream *stream,
431                                  const void    *buffer,
432                                  gsize          count,
433                                  GCancellable  *cancellable,
434                                  GError       **error)
435 {
436   GBufferedOutputStream        *bstream;
437   GBufferedOutputStreamPrivate *priv;
438   gboolean res;
439   gsize    n;
440   gsize new_size;
441
442   bstream = G_BUFFERED_OUTPUT_STREAM (stream);
443   priv = bstream->priv;
444
445   n = priv->len - priv->pos;
446
447   if (priv->auto_grow && n < count)
448     {
449       new_size = MAX (priv->len * 2, priv->len + count);
450       g_buffered_output_stream_set_buffer_size (bstream, new_size);
451     }
452   else if (n == 0)
453     {
454       res = flush_buffer (bstream, cancellable, error);
455       
456       if (res == FALSE)
457         return -1;
458     }
459
460   n = priv->len - priv->pos;
461   
462   count = MIN (count, n);
463   memcpy (priv->buffer + priv->pos, buffer, count);
464   priv->pos += count;
465
466   return count;
467 }
468
469 static gboolean
470 g_buffered_output_stream_flush (GOutputStream  *stream,
471                                 GCancellable   *cancellable,
472                                 GError        **error)
473 {
474   GBufferedOutputStream *bstream;
475   GOutputStream                *base_stream;
476   gboolean res;
477
478   bstream = G_BUFFERED_OUTPUT_STREAM (stream);
479   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
480
481   res = flush_buffer (bstream, cancellable, error);
482
483   if (res == FALSE)
484     return FALSE;
485
486   res = g_output_stream_flush (base_stream, cancellable, error);
487
488   return res;
489 }
490
491 static gboolean
492 g_buffered_output_stream_close (GOutputStream  *stream,
493                                 GCancellable   *cancellable,
494                                 GError        **error)
495 {
496   GBufferedOutputStream        *bstream;
497   GOutputStream                *base_stream;
498   gboolean                      res;
499
500   bstream = G_BUFFERED_OUTPUT_STREAM (stream);
501   base_stream = G_FILTER_OUTPUT_STREAM (bstream)->base_stream;
502   res = flush_buffer (bstream, cancellable, error);
503
504   if (g_filter_output_stream_get_close_base_stream (G_FILTER_OUTPUT_STREAM (stream)))
505     {
506       /* report the first error but still close the stream */
507       if (res)
508         res = g_output_stream_close (base_stream, cancellable, error);
509       else
510         g_output_stream_close (base_stream, cancellable, NULL);
511     }
512
513   return res;
514 }
515
516 /* ************************** */
517 /* Async stuff implementation */
518 /* ************************** */
519
520 /* TODO: This should be using the base class async ops, not threads */
521
522 typedef struct {
523
524   guint flush_stream : 1;
525   guint close_stream : 1;
526
527 } FlushData;
528
529 static void
530 free_flush_data (gpointer data)
531 {
532   g_slice_free (FlushData, data);
533 }
534
535 /* This function is used by all three (i.e. 
536  * _write, _flush, _close) functions since
537  * all of them will need to flush the buffer
538  * and so closing and writing is just a special
539  * case of flushing + some addition stuff */
540 static void
541 flush_buffer_thread (GSimpleAsyncResult *result,
542                      GObject            *object,
543                      GCancellable       *cancellable)
544 {
545   GBufferedOutputStream *stream;
546   GOutputStream *base_stream;
547   FlushData     *fdata;
548   gboolean       res;
549   GError        *error = NULL;
550
551   stream = G_BUFFERED_OUTPUT_STREAM (object);
552   fdata = g_simple_async_result_get_op_res_gpointer (result);
553   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
554
555   res = flush_buffer (stream, cancellable, &error);
556
557   /* if flushing the buffer didn't work don't even bother
558    * to flush the stream but just report that error */
559   if (res && fdata->flush_stream)
560     res = g_output_stream_flush (base_stream, cancellable, &error);
561
562   if (fdata->close_stream) 
563     {
564      
565       /* if flushing the buffer or the stream returned 
566        * an error report that first error but still try 
567        * close the stream */
568       if (g_filter_output_stream_get_close_base_stream (G_FILTER_OUTPUT_STREAM (stream)))
569         {
570           if (res == FALSE)
571             g_output_stream_close (base_stream, cancellable, NULL);
572           else 
573             res = g_output_stream_close (base_stream, cancellable, &error);
574         }
575     }
576
577   if (res == FALSE)
578     g_simple_async_result_take_error (result, error);
579 }
580
581 typedef struct {
582     
583   FlushData fdata;
584
585   gsize  count;
586   const void  *buffer;
587
588 } WriteData;
589
590 static void 
591 free_write_data (gpointer data)
592 {
593   g_slice_free (WriteData, data);
594 }
595
596 static void
597 g_buffered_output_stream_write_async (GOutputStream        *stream,
598                                       const void           *buffer,
599                                       gsize                 count,
600                                       int                   io_priority,
601                                       GCancellable         *cancellable,
602                                       GAsyncReadyCallback   callback,
603                                       gpointer              data)
604 {
605   GBufferedOutputStream *buffered_stream;
606   GBufferedOutputStreamPrivate *priv;
607   GSimpleAsyncResult *res;
608   WriteData *wdata;
609
610   buffered_stream = G_BUFFERED_OUTPUT_STREAM (stream);
611   priv = buffered_stream->priv;
612
613   wdata = g_slice_new (WriteData);
614   wdata->count  = count;
615   wdata->buffer = buffer;
616
617   res = g_simple_async_result_new (G_OBJECT (stream),
618                                    callback,
619                                    data,
620                                    g_buffered_output_stream_write_async);
621
622   g_simple_async_result_set_op_res_gpointer (res, wdata, free_write_data);
623
624   /* if we have space left directly call the
625    * callback (from idle) otherwise schedule a buffer 
626    * flush in the thread. In both cases the actual
627    * copying of the data to the buffer will be done in
628    * the write_finish () func since that should
629    * be fast enough */
630   if (priv->len - priv->pos > 0)
631     {
632       g_simple_async_result_complete_in_idle (res);
633     }
634   else
635     {
636       wdata->fdata.flush_stream = FALSE;
637       wdata->fdata.close_stream = FALSE;
638       g_simple_async_result_run_in_thread (res, 
639                                            flush_buffer_thread, 
640                                            io_priority,
641                                            cancellable);
642       g_object_unref (res);
643     }
644 }
645
646 static gssize
647 g_buffered_output_stream_write_finish (GOutputStream        *stream,
648                                        GAsyncResult         *result,
649                                        GError              **error)
650 {
651   GBufferedOutputStreamPrivate *priv;
652   GBufferedOutputStream        *buffered_stream;
653   GSimpleAsyncResult *simple;
654   WriteData          *wdata;
655   gssize              count;
656
657   simple = G_SIMPLE_ASYNC_RESULT (result);
658   buffered_stream = G_BUFFERED_OUTPUT_STREAM (stream);
659   priv = buffered_stream->priv;
660
661   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == 
662             g_buffered_output_stream_write_async);
663
664   wdata = g_simple_async_result_get_op_res_gpointer (simple);
665
666   /* Now do the real copying of data to the buffer */
667   count = priv->len - priv->pos; 
668   count = MIN (wdata->count, count);
669
670   memcpy (priv->buffer + priv->pos, wdata->buffer, count);
671   
672   priv->pos += count;
673
674   return count;
675 }
676
677 static void
678 g_buffered_output_stream_flush_async (GOutputStream        *stream,
679                                       int                   io_priority,
680                                       GCancellable         *cancellable,
681                                       GAsyncReadyCallback   callback,
682                                       gpointer              data)
683 {
684   GSimpleAsyncResult *res;
685   FlushData          *fdata;
686
687   fdata = g_slice_new (FlushData);
688   fdata->flush_stream = TRUE;
689   fdata->close_stream = FALSE;
690
691   res = g_simple_async_result_new (G_OBJECT (stream),
692                                    callback,
693                                    data,
694                                    g_buffered_output_stream_flush_async);
695
696   g_simple_async_result_set_op_res_gpointer (res, fdata, free_flush_data);
697
698   g_simple_async_result_run_in_thread (res, 
699                                        flush_buffer_thread, 
700                                        io_priority,
701                                        cancellable);
702   g_object_unref (res);
703 }
704
705 static gboolean
706 g_buffered_output_stream_flush_finish (GOutputStream        *stream,
707                                        GAsyncResult         *result,
708                                        GError              **error)
709 {
710   GSimpleAsyncResult *simple;
711
712   simple = G_SIMPLE_ASYNC_RESULT (result);
713
714   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == 
715             g_buffered_output_stream_flush_async);
716
717   return TRUE;
718 }
719
720 static void
721 g_buffered_output_stream_close_async (GOutputStream        *stream,
722                                       int                   io_priority,
723                                       GCancellable         *cancellable,
724                                       GAsyncReadyCallback   callback,
725                                       gpointer              data)
726 {
727   GSimpleAsyncResult *res;
728   FlushData          *fdata;
729
730   fdata = g_slice_new (FlushData);
731   fdata->close_stream = TRUE;
732
733   res = g_simple_async_result_new (G_OBJECT (stream),
734                                    callback,
735                                    data,
736                                    g_buffered_output_stream_close_async);
737
738   g_simple_async_result_set_op_res_gpointer (res, fdata, free_flush_data);
739
740   g_simple_async_result_run_in_thread (res, 
741                                        flush_buffer_thread, 
742                                        io_priority,
743                                        cancellable);
744   g_object_unref (res);
745 }
746
747 static gboolean
748 g_buffered_output_stream_close_finish (GOutputStream        *stream,
749                                        GAsyncResult         *result,
750                                        GError              **error)
751 {
752   GSimpleAsyncResult *simple;
753
754   simple = G_SIMPLE_ASYNC_RESULT (result);
755
756   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == 
757             g_buffered_output_stream_close_async);
758
759   return TRUE;
760 }