1 /* GIO - GLib Input, Output and Streaming Library
3 * Copyright © 2008 codethink
4 * Copyright © 2009 Red Hat, Inc.
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2 of the License, or (at your option) any later version.
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
16 * You should have received a copy of the GNU Lesser General
17 * Public License along with this library; if not, write to the
18 * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
19 * Boston, MA 02111-1307, USA.
21 * Authors: Ryan Lortie <desrt@desrt.ca>
22 * Alexander Larsson <alexl@redhat.com>
29 #include "giostream.h"
30 #include "gasyncresult.h"
33 G_DEFINE_ABSTRACT_TYPE (GIOStream
, g_io_stream
, G_TYPE_OBJECT
);
37 * @short_description: Base class for implementing read/write streams
39 * @see_also: #GInputStream, #GOutputStream
41 * GIOStream represents an object that has both read and write streams.
42 * Generally the two streams acts as separate input and output streams,
43 * but they share some common resources and state. For instance, for
44 * seekable streams they may use the same position in both streams.
46 * Examples of #GIOStream objects are #GSocketConnection which represents
47 * a two-way network connection, and #GFileIOStream which represent a
48 * file handle opened in read-write mode.
50 * To do the actual reading and writing you need to get the substreams
51 * with g_io_stream_get_input_stream() and g_io_stream_get_output_stream().
53 * The #GIOStream object owns the input and the output streams, not the other
54 * way around, so keeping the substreams alive will not keep the #GIOStream
55 * object alive. If the #GIOStream object is freed it will be closed, thus
56 * closing the substream, so even if the substreams stay alive they will
57 * always just return a %G_IO_ERROR_CLOSED for all operations.
59 * To close a stream use g_io_stream_close() which will close the common
60 * stream object and also the individual substreams. You can also close
61 * the substreams themselves. In most cases this only marks the
62 * substream as closed, so further I/O on it fails. However, some streams
63 * may support "half-closed" states where one direction of the stream
64 * is actually shut down.
77 struct _GIOStreamPrivate
{
80 GAsyncReadyCallback outstanding_callback
;
83 static gboolean
g_io_stream_real_close (GIOStream
*stream
,
84 GCancellable
*cancellable
,
86 static void g_io_stream_real_close_async (GIOStream
*stream
,
88 GCancellable
*cancellable
,
89 GAsyncReadyCallback callback
,
91 static gboolean
g_io_stream_real_close_finish (GIOStream
*stream
,
96 g_io_stream_finalize (GObject
*object
)
98 G_OBJECT_CLASS (g_io_stream_parent_class
)->finalize (object
);
102 g_io_stream_dispose (GObject
*object
)
106 stream
= G_IO_STREAM (object
);
108 if (!stream
->priv
->closed
)
109 g_io_stream_close (stream
, NULL
, NULL
);
111 G_OBJECT_CLASS (g_io_stream_parent_class
)->dispose (object
);
115 g_io_stream_init (GIOStream
*stream
)
117 stream
->priv
= G_TYPE_INSTANCE_GET_PRIVATE (stream
,
123 g_io_stream_get_property (GObject
*object
,
128 GIOStream
*stream
= G_IO_STREAM (object
);
133 g_value_set_boolean (value
, stream
->priv
->closed
);
136 case PROP_INPUT_STREAM
:
137 g_value_set_object (value
, g_io_stream_get_input_stream (stream
));
140 case PROP_OUTPUT_STREAM
:
141 g_value_set_object (value
, g_io_stream_get_output_stream (stream
));
145 G_OBJECT_WARN_INVALID_PROPERTY_ID (object
, prop_id
, pspec
);
150 g_io_stream_class_init (GIOStreamClass
*klass
)
152 GObjectClass
*gobject_class
= G_OBJECT_CLASS (klass
);
154 g_type_class_add_private (klass
, sizeof (GIOStreamPrivate
));
156 gobject_class
->finalize
= g_io_stream_finalize
;
157 gobject_class
->dispose
= g_io_stream_dispose
;
158 gobject_class
->get_property
= g_io_stream_get_property
;
160 klass
->close_fn
= g_io_stream_real_close
;
161 klass
->close_async
= g_io_stream_real_close_async
;
162 klass
->close_finish
= g_io_stream_real_close_finish
;
164 g_object_class_install_property (gobject_class
, PROP_CLOSED
,
165 g_param_spec_boolean ("closed",
167 P_("Is the stream closed"),
169 G_PARAM_READABLE
| G_PARAM_STATIC_STRINGS
));
171 g_object_class_install_property (gobject_class
, PROP_INPUT_STREAM
,
172 g_param_spec_object ("input-stream",
174 P_("The GInputStream to read from"),
176 G_PARAM_READABLE
| G_PARAM_STATIC_STRINGS
));
177 g_object_class_install_property (gobject_class
, PROP_OUTPUT_STREAM
,
178 g_param_spec_object ("output-stream",
180 P_("The GOutputStream to write to"),
181 G_TYPE_OUTPUT_STREAM
,
182 G_PARAM_READABLE
| G_PARAM_STATIC_STRINGS
));
186 * g_io_stream_is_closed:
187 * @stream: a #GIOStream
189 * Checks if a stream is closed.
191 * Returns: %TRUE if the stream is closed.
196 g_io_stream_is_closed (GIOStream
*stream
)
198 g_return_val_if_fail (G_IS_IO_STREAM (stream
), TRUE
);
200 return stream
->priv
->closed
;
204 * g_io_stream_get_input_stream:
205 * @stream: a #GIOStream
207 * Gets the input stream for this object. This is used
210 * Returns: (transfer none): a #GInputStream, owned by the #GIOStream.
216 g_io_stream_get_input_stream (GIOStream
*stream
)
218 GIOStreamClass
*klass
;
220 klass
= G_IO_STREAM_GET_CLASS (stream
);
222 g_assert (klass
->get_input_stream
!= NULL
);
224 return klass
->get_input_stream (stream
);
228 * g_io_stream_get_output_stream:
229 * @stream: a #GIOStream
231 * Gets the output stream for this object. This is used for
234 * Returns: (transfer none): a #GOutputStream, owned by the #GIOStream.
240 g_io_stream_get_output_stream (GIOStream
*stream
)
242 GIOStreamClass
*klass
;
244 klass
= G_IO_STREAM_GET_CLASS (stream
);
246 g_assert (klass
->get_output_stream
!= NULL
);
247 return klass
->get_output_stream (stream
);
251 * g_io_stream_has_pending:
252 * @stream: a #GIOStream
254 * Checks if a stream has pending actions.
256 * Returns: %TRUE if @stream has pending actions.
261 g_io_stream_has_pending (GIOStream
*stream
)
263 g_return_val_if_fail (G_IS_IO_STREAM (stream
), FALSE
);
265 return stream
->priv
->pending
;
269 * g_io_stream_set_pending:
270 * @stream: a #GIOStream
271 * @error: a #GError location to store the error occurring, or %NULL to
274 * Sets @stream to have actions pending. If the pending flag is
275 * already set or @stream is closed, it will return %FALSE and set
278 * Return value: %TRUE if pending was previously unset and is now set.
283 g_io_stream_set_pending (GIOStream
*stream
,
286 g_return_val_if_fail (G_IS_IO_STREAM (stream
), FALSE
);
288 if (stream
->priv
->closed
)
290 g_set_error_literal (error
, G_IO_ERROR
, G_IO_ERROR_CLOSED
,
291 _("Stream is already closed"));
295 if (stream
->priv
->pending
)
297 g_set_error_literal (error
, G_IO_ERROR
, G_IO_ERROR_PENDING
,
298 /* Translators: This is an error you get if there is
299 * already an operation running against this stream when
300 * you try to start one */
301 _("Stream has outstanding operation"));
305 stream
->priv
->pending
= TRUE
;
310 * g_io_stream_clear_pending:
311 * @stream: a #GIOStream
313 * Clears the pending flag on @stream.
318 g_io_stream_clear_pending (GIOStream
*stream
)
320 g_return_if_fail (G_IS_IO_STREAM (stream
));
322 stream
->priv
->pending
= FALSE
;
326 g_io_stream_real_close (GIOStream
*stream
,
327 GCancellable
*cancellable
,
332 res
= g_output_stream_close (g_io_stream_get_output_stream (stream
),
335 /* If this errored out, unset error so that we don't report
336 further errors, but still do the following ops */
337 if (error
!= NULL
&& *error
!= NULL
)
340 res
&= g_input_stream_close (g_io_stream_get_input_stream (stream
),
348 * @stream: a #GIOStream
349 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
350 * @error: location to store the error occurring, or %NULL to ignore
352 * Closes the stream, releasing resources related to it. This will also
353 * closes the individual input and output streams, if they are not already
356 * Once the stream is closed, all other operations will return
357 * %G_IO_ERROR_CLOSED. Closing a stream multiple times will not
360 * Closing a stream will automatically flush any outstanding buffers
363 * Streams will be automatically closed when the last reference
364 * is dropped, but you might want to call this function to make sure
365 * resources are released as early as possible.
367 * Some streams might keep the backing store of the stream (e.g. a file
368 * descriptor) open after the stream is closed. See the documentation for
369 * the individual stream for details.
371 * On failure the first error that happened will be reported, but the
372 * close operation will finish as much as possible. A stream that failed
373 * to close will still return %G_IO_ERROR_CLOSED for all operations.
374 * Still, it is important to check and report the error to the user,
375 * otherwise there might be a loss of data as all data might not be written.
377 * If @cancellable is not NULL, then the operation can be cancelled by
378 * triggering the cancellable object from another thread. If the operation
379 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
380 * Cancelling a close will still leave the stream closed, but some streams
381 * can use a faster close that doesn't block to e.g. check errors.
383 * The default implementation of this method just calls close on the
384 * individual input/output streams.
386 * Return value: %TRUE on success, %FALSE on failure
391 g_io_stream_close (GIOStream
*stream
,
392 GCancellable
*cancellable
,
395 GIOStreamClass
*class;
398 g_return_val_if_fail (G_IS_IO_STREAM (stream
), FALSE
);
400 class = G_IO_STREAM_GET_CLASS (stream
);
402 if (stream
->priv
->closed
)
405 if (!g_io_stream_set_pending (stream
, error
))
409 g_cancellable_push_current (cancellable
);
413 res
= class->close_fn (stream
, cancellable
, error
);
416 g_cancellable_pop_current (cancellable
);
418 stream
->priv
->closed
= TRUE
;
419 g_io_stream_clear_pending (stream
);
425 async_ready_close_callback_wrapper (GObject
*source_object
,
429 GIOStream
*stream
= G_IO_STREAM (source_object
);
431 stream
->priv
->closed
= TRUE
;
432 g_io_stream_clear_pending (stream
);
433 if (stream
->priv
->outstanding_callback
)
434 (*stream
->priv
->outstanding_callback
) (source_object
, res
, user_data
);
435 g_object_unref (stream
);
439 * g_io_stream_close_async:
440 * @stream: a #GIOStream
441 * @io_priority: the io priority of the request
442 * @cancellable: (allow-none): optional cancellable object
443 * @callback: (scope async): callback to call when the request is satisfied
444 * @user_data: (closure): the data to pass to callback function
446 * Requests an asynchronous close of the stream, releasing resources
447 * related to it. When the operation is finished @callback will be
448 * called. You can then call g_io_stream_close_finish() to get
449 * the result of the operation.
451 * For behaviour details see g_io_stream_close().
453 * The asynchronous methods have a default fallback that uses threads
454 * to implement asynchronicity, so they are optional for inheriting
455 * classes. However, if you override one you must override all.
460 g_io_stream_close_async (GIOStream
*stream
,
462 GCancellable
*cancellable
,
463 GAsyncReadyCallback callback
,
466 GIOStreamClass
*class;
467 GError
*error
= NULL
;
469 g_return_if_fail (G_IS_IO_STREAM (stream
));
471 if (stream
->priv
->closed
)
475 task
= g_task_new (stream
, cancellable
, callback
, user_data
);
476 g_task_set_source_tag (task
, g_io_stream_close_async
);
477 g_task_return_boolean (task
, TRUE
);
478 g_object_unref (task
);
482 if (!g_io_stream_set_pending (stream
, &error
))
484 g_task_report_error (stream
, callback
, user_data
,
485 g_io_stream_close_async
,
490 class = G_IO_STREAM_GET_CLASS (stream
);
491 stream
->priv
->outstanding_callback
= callback
;
492 g_object_ref (stream
);
493 class->close_async (stream
, io_priority
, cancellable
,
494 async_ready_close_callback_wrapper
, user_data
);
498 * g_io_stream_close_finish:
499 * @stream: a #GIOStream
500 * @result: a #GAsyncResult
501 * @error: a #GError location to store the error occurring, or %NULL to
506 * Returns: %TRUE if stream was successfully closed, %FALSE otherwise.
511 g_io_stream_close_finish (GIOStream
*stream
,
512 GAsyncResult
*result
,
515 GIOStreamClass
*class;
517 g_return_val_if_fail (G_IS_IO_STREAM (stream
), FALSE
);
518 g_return_val_if_fail (G_IS_ASYNC_RESULT (result
), FALSE
);
520 if (g_async_result_legacy_propagate_error (result
, error
))
522 else if (g_async_result_is_tagged (result
, g_io_stream_close_async
))
523 return g_task_propagate_boolean (G_TASK (result
), error
);
525 class = G_IO_STREAM_GET_CLASS (stream
);
526 return class->close_finish (stream
, result
, error
);
531 close_async_thread (GTask
*task
,
532 gpointer source_object
,
534 GCancellable
*cancellable
)
536 GIOStream
*stream
= source_object
;
537 GIOStreamClass
*class;
538 GError
*error
= NULL
;
541 class = G_IO_STREAM_GET_CLASS (stream
);
544 result
= class->close_fn (stream
,
545 g_task_get_cancellable (task
),
549 g_task_return_error (task
, error
);
554 g_task_return_boolean (task
, TRUE
);
558 g_io_stream_real_close_async (GIOStream
*stream
,
560 GCancellable
*cancellable
,
561 GAsyncReadyCallback callback
,
566 task
= g_task_new (stream
, cancellable
, callback
, user_data
);
567 g_task_set_check_cancellable (task
, FALSE
);
568 g_task_set_priority (task
, io_priority
);
570 g_task_run_in_thread (task
, close_async_thread
);
571 g_object_unref (task
);
575 g_io_stream_real_close_finish (GIOStream
*stream
,
576 GAsyncResult
*result
,
579 g_return_val_if_fail (g_task_is_valid (result
, stream
), FALSE
);
581 return g_task_propagate_boolean (G_TASK (result
), error
);
588 GIOStreamSpliceFlags flags
;
590 GCancellable
*cancellable
;
592 GCancellable
*op1_cancellable
;
593 GCancellable
*op2_cancellable
;
599 splice_context_free (SpliceContext
*ctx
)
601 g_object_unref (ctx
->stream1
);
602 g_object_unref (ctx
->stream2
);
603 if (ctx
->cancellable
!= NULL
)
604 g_object_unref (ctx
->cancellable
);
605 g_object_unref (ctx
->op1_cancellable
);
606 g_object_unref (ctx
->op2_cancellable
);
607 g_clear_error (&ctx
->error
);
608 g_slice_free (SpliceContext
, ctx
);
612 splice_complete (GTask
*task
,
615 if (ctx
->cancelled_id
!= 0)
616 g_cancellable_disconnect (ctx
->cancellable
, ctx
->cancelled_id
);
617 ctx
->cancelled_id
= 0;
619 if (ctx
->error
!= NULL
)
621 g_task_return_error (task
, ctx
->error
);
625 g_task_return_boolean (task
, TRUE
);
629 splice_close_cb (GObject
*iostream
,
633 GTask
*task
= user_data
;
634 SpliceContext
*ctx
= g_task_get_task_data (task
);
635 GError
*error
= NULL
;
637 g_io_stream_close_finish (G_IO_STREAM (iostream
), res
, &error
);
641 /* Keep the first error that occurred */
642 if (error
!= NULL
&& ctx
->error
== NULL
)
645 g_clear_error (&error
);
647 /* If all operations are done, complete now */
648 if (ctx
->completed
== 4)
649 splice_complete (task
, ctx
);
651 g_object_unref (task
);
655 splice_cb (GObject
*ostream
,
659 GTask
*task
= user_data
;
660 SpliceContext
*ctx
= g_task_get_task_data (task
);
661 GError
*error
= NULL
;
663 g_output_stream_splice_finish (G_OUTPUT_STREAM (ostream
), res
, &error
);
667 /* ignore cancellation error if it was not requested by the user */
669 g_error_matches (error
, G_IO_ERROR
, G_IO_ERROR_CANCELLED
) &&
670 (ctx
->cancellable
== NULL
||
671 !g_cancellable_is_cancelled (ctx
->cancellable
)))
672 g_clear_error (&error
);
674 /* Keep the first error that occurred */
675 if (error
!= NULL
&& ctx
->error
== NULL
)
678 g_clear_error (&error
);
680 if (ctx
->completed
== 1 &&
681 (ctx
->flags
& G_IO_STREAM_SPLICE_WAIT_FOR_BOTH
) == 0)
683 /* We don't want to wait for the 2nd operation to finish, cancel it */
684 g_cancellable_cancel (ctx
->op1_cancellable
);
685 g_cancellable_cancel (ctx
->op2_cancellable
);
687 else if (ctx
->completed
== 2)
689 if (ctx
->cancellable
== NULL
||
690 !g_cancellable_is_cancelled (ctx
->cancellable
))
692 g_cancellable_reset (ctx
->op1_cancellable
);
693 g_cancellable_reset (ctx
->op2_cancellable
);
696 /* Close the IO streams if needed */
697 if ((ctx
->flags
& G_IO_STREAM_SPLICE_CLOSE_STREAM1
) != 0)
699 g_io_stream_close_async (ctx
->stream1
,
700 g_task_get_priority (task
),
701 ctx
->op1_cancellable
,
702 splice_close_cb
, g_object_ref (task
));
707 if ((ctx
->flags
& G_IO_STREAM_SPLICE_CLOSE_STREAM2
) != 0)
709 g_io_stream_close_async (ctx
->stream2
,
710 g_task_get_priority (task
),
711 ctx
->op2_cancellable
,
712 splice_close_cb
, g_object_ref (task
));
717 /* If all operations are done, complete now */
718 if (ctx
->completed
== 4)
719 splice_complete (task
, ctx
);
722 g_object_unref (task
);
726 splice_cancelled_cb (GCancellable
*cancellable
,
731 ctx
= g_task_get_task_data (task
);
732 g_cancellable_cancel (ctx
->op1_cancellable
);
733 g_cancellable_cancel (ctx
->op2_cancellable
);
737 * g_io_stream_splice_async:
738 * @stream1: a #GIOStream.
739 * @stream2: a #GIOStream.
740 * @flags: a set of #GIOStreamSpliceFlags.
741 * @io_priority: the io priority of the request.
742 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
743 * @callback: (scope async): a #GAsyncReadyCallback.
744 * @user_data: (closure): user data passed to @callback.
746 * Asyncronously splice the output stream of @stream1 to the input stream of
747 * @stream2, and splice the output stream of @stream2 to the input stream of
750 * When the operation is finished @callback will be called.
751 * You can then call g_io_stream_splice_finish() to get the
752 * result of the operation.
757 g_io_stream_splice_async (GIOStream
*stream1
,
759 GIOStreamSpliceFlags flags
,
761 GCancellable
*cancellable
,
762 GAsyncReadyCallback callback
,
767 GInputStream
*istream
;
768 GOutputStream
*ostream
;
770 if (cancellable
!= NULL
&& g_cancellable_is_cancelled (cancellable
))
772 g_task_report_new_error (NULL
, callback
, user_data
,
773 g_io_stream_splice_async
,
774 G_IO_ERROR
, G_IO_ERROR_CANCELLED
,
775 "Operation has been cancelled");
779 ctx
= g_slice_new0 (SpliceContext
);
780 ctx
->stream1
= g_object_ref (stream1
);
781 ctx
->stream2
= g_object_ref (stream2
);
783 ctx
->op1_cancellable
= g_cancellable_new ();
784 ctx
->op2_cancellable
= g_cancellable_new ();
787 task
= g_task_new (NULL
, cancellable
, callback
, user_data
);
788 g_task_set_task_data (task
, ctx
, (GDestroyNotify
) splice_context_free
);
790 if (cancellable
!= NULL
)
792 ctx
->cancellable
= g_object_ref (cancellable
);
793 ctx
->cancelled_id
= g_cancellable_connect (cancellable
,
794 G_CALLBACK (splice_cancelled_cb
), g_object_ref (task
),
798 istream
= g_io_stream_get_input_stream (stream1
);
799 ostream
= g_io_stream_get_output_stream (stream2
);
800 g_output_stream_splice_async (ostream
, istream
, G_OUTPUT_STREAM_SPLICE_NONE
,
801 io_priority
, ctx
->op1_cancellable
, splice_cb
,
802 g_object_ref (task
));
804 istream
= g_io_stream_get_input_stream (stream2
);
805 ostream
= g_io_stream_get_output_stream (stream1
);
806 g_output_stream_splice_async (ostream
, istream
, G_OUTPUT_STREAM_SPLICE_NONE
,
807 io_priority
, ctx
->op2_cancellable
, splice_cb
,
808 g_object_ref (task
));
810 g_object_unref (task
);
814 * g_io_stream_splice_finish:
815 * @result: a #GAsyncResult.
816 * @error: a #GError location to store the error occurring, or %NULL to
819 * Finishes an asynchronous io stream splice operation.
821 * Returns: %TRUE on success, %FALSE otherwise.
826 g_io_stream_splice_finish (GAsyncResult
*result
,
829 g_return_val_if_fail (g_task_is_valid (result
, NULL
), FALSE
);
831 return g_task_propagate_boolean (G_TASK (result
), error
);