1 /* GIO - GLib Input, Output and Streaming Library
3 * Copyright (C) 2006-2007 Red Hat, Inc.
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.
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.
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.
20 * Author: Alexander Larsson <alexl@redhat.com>
27 #include "ginputstream.h"
28 #include "gseekable.h"
29 #include "gcancellable.h"
30 #include "gasyncresult.h"
31 #include "gsimpleasyncresult.h"
36 * SECTION:ginputstream
37 * @short_description: Base class for implementing streaming input
40 * GInputStream has functions to read from a stream (g_input_stream_read()),
41 * to close a stream (g_input_stream_close()) and to skip some content
42 * (g_input_stream_skip()).
44 * To copy the content of an input stream to an output stream without
45 * manually handling the reads and writes, use g_output_stream_splice().
47 * All of these functions have async variants too.
50 G_DEFINE_ABSTRACT_TYPE (GInputStream, g_input_stream, G_TYPE_OBJECT);
52 struct _GInputStreamPrivate {
55 GAsyncReadyCallback outstanding_callback;
58 static gssize g_input_stream_real_skip (GInputStream *stream,
60 GCancellable *cancellable,
62 static void g_input_stream_real_read_async (GInputStream *stream,
66 GCancellable *cancellable,
67 GAsyncReadyCallback callback,
69 static gssize g_input_stream_real_read_finish (GInputStream *stream,
72 static void g_input_stream_real_skip_async (GInputStream *stream,
75 GCancellable *cancellable,
76 GAsyncReadyCallback callback,
78 static gssize g_input_stream_real_skip_finish (GInputStream *stream,
81 static void g_input_stream_real_close_async (GInputStream *stream,
83 GCancellable *cancellable,
84 GAsyncReadyCallback callback,
86 static gboolean g_input_stream_real_close_finish (GInputStream *stream,
91 g_input_stream_finalize (GObject *object)
93 G_OBJECT_CLASS (g_input_stream_parent_class)->finalize (object);
97 g_input_stream_dispose (GObject *object)
101 stream = G_INPUT_STREAM (object);
103 if (!stream->priv->closed)
104 g_input_stream_close (stream, NULL, NULL);
106 G_OBJECT_CLASS (g_input_stream_parent_class)->dispose (object);
111 g_input_stream_class_init (GInputStreamClass *klass)
113 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
115 g_type_class_add_private (klass, sizeof (GInputStreamPrivate));
117 gobject_class->finalize = g_input_stream_finalize;
118 gobject_class->dispose = g_input_stream_dispose;
120 klass->skip = g_input_stream_real_skip;
121 klass->read_async = g_input_stream_real_read_async;
122 klass->read_finish = g_input_stream_real_read_finish;
123 klass->skip_async = g_input_stream_real_skip_async;
124 klass->skip_finish = g_input_stream_real_skip_finish;
125 klass->close_async = g_input_stream_real_close_async;
126 klass->close_finish = g_input_stream_real_close_finish;
130 g_input_stream_init (GInputStream *stream)
132 stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
134 GInputStreamPrivate);
138 * g_input_stream_read:
139 * @stream: a #GInputStream.
140 * @buffer: a buffer to read data into (which should be at least count bytes long).
141 * @count: the number of bytes that will be read from the stream
142 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
143 * @error: location to store the error occurring, or %NULL to ignore
145 * Tries to read @count bytes from the stream into the buffer starting at
146 * @buffer. Will block during this read.
148 * If count is zero returns zero and does nothing. A value of @count
149 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
151 * On success, the number of bytes read into the buffer is returned.
152 * It is not an error if this is not the same as the requested size, as it
153 * can happen e.g. near the end of a file. Zero is returned on end of file
154 * (or if @count is zero), but never otherwise.
156 * If @cancellable is not NULL, then the operation can be cancelled by
157 * triggering the cancellable object from another thread. If the operation
158 * was cancelled, the error G_IO_ERROR_CANCELLED will be returned. If an
159 * operation was partially finished when the operation was cancelled the
160 * partial result will be returned, without an error.
162 * On error -1 is returned and @error is set accordingly.
164 * Return value: Number of bytes read, or -1 on error, or 0 on end of file.
167 g_input_stream_read (GInputStream *stream,
170 GCancellable *cancellable,
173 GInputStreamClass *class;
176 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
177 g_return_val_if_fail (buffer != NULL, 0);
182 if (((gssize) count) < 0)
184 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
185 _("Too large count value passed to %s"), G_STRFUNC);
189 class = G_INPUT_STREAM_GET_CLASS (stream);
191 if (class->read_fn == NULL)
193 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
194 _("Input stream doesn't implement read"));
198 if (!g_input_stream_set_pending (stream, error))
202 g_cancellable_push_current (cancellable);
204 res = class->read_fn (stream, buffer, count, cancellable, error);
207 g_cancellable_pop_current (cancellable);
209 g_input_stream_clear_pending (stream);
215 * g_input_stream_read_all:
216 * @stream: a #GInputStream.
217 * @buffer: a buffer to read data into (which should be at least count bytes long).
218 * @count: the number of bytes that will be read from the stream
219 * @bytes_read: (out): location to store the number of bytes that was read from the stream
220 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
221 * @error: location to store the error occurring, or %NULL to ignore
223 * Tries to read @count bytes from the stream into the buffer starting at
224 * @buffer. Will block during this read.
226 * This function is similar to g_input_stream_read(), except it tries to
227 * read as many bytes as requested, only stopping on an error or end of stream.
229 * On a successful read of @count bytes, or if we reached the end of the
230 * stream, %TRUE is returned, and @bytes_read is set to the number of bytes
233 * If there is an error during the operation %FALSE is returned and @error
234 * is set to indicate the error status, @bytes_read is updated to contain
235 * the number of bytes read into @buffer before the error occurred.
237 * Return value: %TRUE on success, %FALSE if there was an error
240 g_input_stream_read_all (GInputStream *stream,
244 GCancellable *cancellable,
250 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
251 g_return_val_if_fail (buffer != NULL, FALSE);
254 while (_bytes_read < count)
256 res = g_input_stream_read (stream, (char *)buffer + _bytes_read, count - _bytes_read,
261 *bytes_read = _bytes_read;
272 *bytes_read = _bytes_read;
277 * g_input_stream_skip:
278 * @stream: a #GInputStream.
279 * @count: the number of bytes that will be skipped from the stream
280 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
281 * @error: location to store the error occurring, or %NULL to ignore
283 * Tries to skip @count bytes from the stream. Will block during the operation.
285 * This is identical to g_input_stream_read(), from a behaviour standpoint,
286 * but the bytes that are skipped are not returned to the user. Some
287 * streams have an implementation that is more efficient than reading the data.
289 * This function is optional for inherited classes, as the default implementation
290 * emulates it using read.
292 * If @cancellable is not %NULL, then the operation can be cancelled by
293 * triggering the cancellable object from another thread. If the operation
294 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
295 * operation was partially finished when the operation was cancelled the
296 * partial result will be returned, without an error.
298 * Return value: Number of bytes skipped, or -1 on error
301 g_input_stream_skip (GInputStream *stream,
303 GCancellable *cancellable,
306 GInputStreamClass *class;
309 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
314 if (((gssize) count) < 0)
316 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
317 _("Too large count value passed to %s"), G_STRFUNC);
321 class = G_INPUT_STREAM_GET_CLASS (stream);
323 if (!g_input_stream_set_pending (stream, error))
327 g_cancellable_push_current (cancellable);
329 res = class->skip (stream, count, cancellable, error);
332 g_cancellable_pop_current (cancellable);
334 g_input_stream_clear_pending (stream);
340 g_input_stream_real_skip (GInputStream *stream,
342 GCancellable *cancellable,
345 GInputStreamClass *class;
346 gssize ret, read_bytes;
350 if (G_IS_SEEKABLE (stream) && g_seekable_can_seek (G_SEEKABLE (stream)))
352 if (g_seekable_seek (G_SEEKABLE (stream),
360 /* If not seekable, or seek failed, fall back to reading data: */
362 class = G_INPUT_STREAM_GET_CLASS (stream);
369 ret = class->read_fn (stream, buffer, MIN (sizeof (buffer), count),
370 cancellable, &my_error);
373 if (read_bytes > 0 &&
374 my_error->domain == G_IO_ERROR &&
375 my_error->code == G_IO_ERROR_CANCELLED)
377 g_error_free (my_error);
381 g_propagate_error (error, my_error);
388 if (ret == 0 || count == 0)
394 * g_input_stream_close:
395 * @stream: A #GInputStream.
396 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
397 * @error: location to store the error occurring, or %NULL to ignore
399 * Closes the stream, releasing resources related to it.
401 * Once the stream is closed, all other operations will return %G_IO_ERROR_CLOSED.
402 * Closing a stream multiple times will not return an error.
404 * Streams will be automatically closed when the last reference
405 * is dropped, but you might want to call this function to make sure
406 * resources are released as early as possible.
408 * Some streams might keep the backing store of the stream (e.g. a file descriptor)
409 * open after the stream is closed. See the documentation for the individual
410 * stream for details.
412 * On failure the first error that happened will be reported, but the close
413 * operation will finish as much as possible. A stream that failed to
414 * close will still return %G_IO_ERROR_CLOSED for all operations. Still, it
415 * is important to check and report the error to the user.
417 * If @cancellable is not NULL, then the operation can be cancelled by
418 * triggering the cancellable object from another thread. If the operation
419 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
420 * Cancelling a close will still leave the stream closed, but some streams
421 * can use a faster close that doesn't block to e.g. check errors.
423 * Return value: %TRUE on success, %FALSE on failure
426 g_input_stream_close (GInputStream *stream,
427 GCancellable *cancellable,
430 GInputStreamClass *class;
433 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
435 class = G_INPUT_STREAM_GET_CLASS (stream);
437 if (stream->priv->closed)
442 if (!g_input_stream_set_pending (stream, error))
446 g_cancellable_push_current (cancellable);
449 res = class->close_fn (stream, cancellable, error);
452 g_cancellable_pop_current (cancellable);
454 g_input_stream_clear_pending (stream);
456 stream->priv->closed = TRUE;
462 async_ready_callback_wrapper (GObject *source_object,
466 GInputStream *stream = G_INPUT_STREAM (source_object);
468 g_input_stream_clear_pending (stream);
469 if (stream->priv->outstanding_callback)
470 (*stream->priv->outstanding_callback) (source_object, res, user_data);
471 g_object_unref (stream);
475 async_ready_close_callback_wrapper (GObject *source_object,
479 GInputStream *stream = G_INPUT_STREAM (source_object);
481 g_input_stream_clear_pending (stream);
482 stream->priv->closed = TRUE;
483 if (stream->priv->outstanding_callback)
484 (*stream->priv->outstanding_callback) (source_object, res, user_data);
485 g_object_unref (stream);
489 * g_input_stream_read_async:
490 * @stream: A #GInputStream.
491 * @buffer: a buffer to read data into (which should be at least count bytes long).
492 * @count: the number of bytes that will be read from the stream
493 * @io_priority: the <link linkend="io-priority">I/O priority</link>
495 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
496 * @callback: (scope async): callback to call when the request is satisfied
497 * @user_data: (closure): the data to pass to callback function
499 * Request an asynchronous read of @count bytes from the stream into the buffer
500 * starting at @buffer. When the operation is finished @callback will be called.
501 * You can then call g_input_stream_read_finish() to get the result of the
504 * During an async request no other sync and async calls are allowed on @stream, and will
505 * result in %G_IO_ERROR_PENDING errors.
507 * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
509 * On success, the number of bytes read into the buffer will be passed to the
510 * callback. It is not an error if this is not the same as the requested size, as it
511 * can happen e.g. near the end of a file, but generally we try to read
512 * as many bytes as requested. Zero is returned on end of file
513 * (or if @count is zero), but never otherwise.
515 * Any outstanding i/o request with higher priority (lower numerical value) will
516 * be executed before an outstanding request with lower priority. Default
517 * priority is %G_PRIORITY_DEFAULT.
519 * The asyncronous methods have a default fallback that uses threads to implement
520 * asynchronicity, so they are optional for inheriting classes. However, if you
521 * override one you must override all.
524 g_input_stream_read_async (GInputStream *stream,
528 GCancellable *cancellable,
529 GAsyncReadyCallback callback,
532 GInputStreamClass *class;
533 GSimpleAsyncResult *simple;
534 GError *error = NULL;
536 g_return_if_fail (G_IS_INPUT_STREAM (stream));
537 g_return_if_fail (buffer != NULL);
541 simple = g_simple_async_result_new (G_OBJECT (stream),
544 g_input_stream_read_async);
545 g_simple_async_result_complete_in_idle (simple);
546 g_object_unref (simple);
550 if (((gssize) count) < 0)
552 g_simple_async_report_error_in_idle (G_OBJECT (stream),
555 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
556 _("Too large count value passed to %s"),
561 if (!g_input_stream_set_pending (stream, &error))
563 g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
570 class = G_INPUT_STREAM_GET_CLASS (stream);
571 stream->priv->outstanding_callback = callback;
572 g_object_ref (stream);
573 class->read_async (stream, buffer, count, io_priority, cancellable,
574 async_ready_callback_wrapper, user_data);
578 * g_input_stream_read_finish:
579 * @stream: a #GInputStream.
580 * @result: a #GAsyncResult.
581 * @error: a #GError location to store the error occurring, or %NULL to
584 * Finishes an asynchronous stream read operation.
586 * Returns: number of bytes read in, or -1 on error, or 0 on end of file.
589 g_input_stream_read_finish (GInputStream *stream,
590 GAsyncResult *result,
593 GSimpleAsyncResult *simple;
594 GInputStreamClass *class;
596 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
597 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
599 if (G_IS_SIMPLE_ASYNC_RESULT (result))
601 simple = G_SIMPLE_ASYNC_RESULT (result);
602 if (g_simple_async_result_propagate_error (simple, error))
605 /* Special case read of 0 bytes */
606 if (g_simple_async_result_get_source_tag (simple) == g_input_stream_read_async)
610 class = G_INPUT_STREAM_GET_CLASS (stream);
611 return class->read_finish (stream, result, error);
615 * g_input_stream_skip_async:
616 * @stream: A #GInputStream.
617 * @count: the number of bytes that will be skipped from the stream
618 * @io_priority: the <link linkend="io-priority">I/O priority</link>
620 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
621 * @callback: (scope async): callback to call when the request is satisfied
622 * @user_data: (closure): the data to pass to callback function
624 * Request an asynchronous skip of @count bytes from the stream.
625 * When the operation is finished @callback will be called.
626 * You can then call g_input_stream_skip_finish() to get the result
629 * During an async request no other sync and async calls are allowed,
630 * and will result in %G_IO_ERROR_PENDING errors.
632 * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
634 * On success, the number of bytes skipped will be passed to the callback.
635 * It is not an error if this is not the same as the requested size, as it
636 * can happen e.g. near the end of a file, but generally we try to skip
637 * as many bytes as requested. Zero is returned on end of file
638 * (or if @count is zero), but never otherwise.
640 * Any outstanding i/o request with higher priority (lower numerical value)
641 * will be executed before an outstanding request with lower priority.
642 * Default priority is %G_PRIORITY_DEFAULT.
644 * The asynchronous methods have a default fallback that uses threads to
645 * implement asynchronicity, so they are optional for inheriting classes.
646 * However, if you override one, you must override all.
649 g_input_stream_skip_async (GInputStream *stream,
652 GCancellable *cancellable,
653 GAsyncReadyCallback callback,
656 GInputStreamClass *class;
657 GSimpleAsyncResult *simple;
658 GError *error = NULL;
660 g_return_if_fail (G_IS_INPUT_STREAM (stream));
664 simple = g_simple_async_result_new (G_OBJECT (stream),
667 g_input_stream_skip_async);
669 g_simple_async_result_complete_in_idle (simple);
670 g_object_unref (simple);
674 if (((gssize) count) < 0)
676 g_simple_async_report_error_in_idle (G_OBJECT (stream),
679 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
680 _("Too large count value passed to %s"),
685 if (!g_input_stream_set_pending (stream, &error))
687 g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
694 class = G_INPUT_STREAM_GET_CLASS (stream);
695 stream->priv->outstanding_callback = callback;
696 g_object_ref (stream);
697 class->skip_async (stream, count, io_priority, cancellable,
698 async_ready_callback_wrapper, user_data);
702 * g_input_stream_skip_finish:
703 * @stream: a #GInputStream.
704 * @result: a #GAsyncResult.
705 * @error: a #GError location to store the error occurring, or %NULL to
708 * Finishes a stream skip operation.
710 * Returns: the size of the bytes skipped, or %-1 on error.
713 g_input_stream_skip_finish (GInputStream *stream,
714 GAsyncResult *result,
717 GSimpleAsyncResult *simple;
718 GInputStreamClass *class;
720 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
721 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
723 if (G_IS_SIMPLE_ASYNC_RESULT (result))
725 simple = G_SIMPLE_ASYNC_RESULT (result);
726 if (g_simple_async_result_propagate_error (simple, error))
729 /* Special case skip of 0 bytes */
730 if (g_simple_async_result_get_source_tag (simple) == g_input_stream_skip_async)
734 class = G_INPUT_STREAM_GET_CLASS (stream);
735 return class->skip_finish (stream, result, error);
739 * g_input_stream_close_async:
740 * @stream: A #GInputStream.
741 * @io_priority: the <link linkend="io-priority">I/O priority</link>
743 * @cancellable: (allow-none): optional cancellable object
744 * @callback: (scope async): callback to call when the request is satisfied
745 * @user_data: (closure): the data to pass to callback function
747 * Requests an asynchronous closes of the stream, releasing resources related to it.
748 * When the operation is finished @callback will be called.
749 * You can then call g_input_stream_close_finish() to get the result of the
752 * For behaviour details see g_input_stream_close().
754 * The asyncronous methods have a default fallback that uses threads to implement
755 * asynchronicity, so they are optional for inheriting classes. However, if you
756 * override one you must override all.
759 g_input_stream_close_async (GInputStream *stream,
761 GCancellable *cancellable,
762 GAsyncReadyCallback callback,
765 GInputStreamClass *class;
766 GSimpleAsyncResult *simple;
767 GError *error = NULL;
769 g_return_if_fail (G_IS_INPUT_STREAM (stream));
771 if (stream->priv->closed)
773 simple = g_simple_async_result_new (G_OBJECT (stream),
776 g_input_stream_close_async);
778 g_simple_async_result_complete_in_idle (simple);
779 g_object_unref (simple);
783 if (!g_input_stream_set_pending (stream, &error))
785 g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream),
792 class = G_INPUT_STREAM_GET_CLASS (stream);
793 stream->priv->outstanding_callback = callback;
794 g_object_ref (stream);
795 class->close_async (stream, io_priority, cancellable,
796 async_ready_close_callback_wrapper, user_data);
800 * g_input_stream_close_finish:
801 * @stream: a #GInputStream.
802 * @result: a #GAsyncResult.
803 * @error: a #GError location to store the error occurring, or %NULL to
806 * Finishes closing a stream asynchronously, started from g_input_stream_close_async().
808 * Returns: %TRUE if the stream was closed successfully.
811 g_input_stream_close_finish (GInputStream *stream,
812 GAsyncResult *result,
815 GSimpleAsyncResult *simple;
816 GInputStreamClass *class;
818 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
819 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
821 if (G_IS_SIMPLE_ASYNC_RESULT (result))
823 simple = G_SIMPLE_ASYNC_RESULT (result);
824 if (g_simple_async_result_propagate_error (simple, error))
827 /* Special case already closed */
828 if (g_simple_async_result_get_source_tag (simple) == g_input_stream_close_async)
832 class = G_INPUT_STREAM_GET_CLASS (stream);
833 return class->close_finish (stream, result, error);
837 * g_input_stream_is_closed:
838 * @stream: input stream.
840 * Checks if an input stream is closed.
842 * Returns: %TRUE if the stream is closed.
845 g_input_stream_is_closed (GInputStream *stream)
847 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
849 return stream->priv->closed;
853 * g_input_stream_has_pending:
854 * @stream: input stream.
856 * Checks if an input stream has pending actions.
858 * Returns: %TRUE if @stream has pending actions.
861 g_input_stream_has_pending (GInputStream *stream)
863 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
865 return stream->priv->pending;
869 * g_input_stream_set_pending:
870 * @stream: input stream
871 * @error: a #GError location to store the error occurring, or %NULL to
874 * Sets @stream to have actions pending. If the pending flag is
875 * already set or @stream is closed, it will return %FALSE and set
878 * Return value: %TRUE if pending was previously unset and is now set.
881 g_input_stream_set_pending (GInputStream *stream, GError **error)
883 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
885 if (stream->priv->closed)
887 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
888 _("Stream is already closed"));
892 if (stream->priv->pending)
894 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
895 /* Translators: This is an error you get if there is already an
896 * operation running against this stream when you try to start
898 _("Stream has outstanding operation"));
902 stream->priv->pending = TRUE;
907 * g_input_stream_clear_pending:
908 * @stream: input stream
910 * Clears the pending flag on @stream.
913 g_input_stream_clear_pending (GInputStream *stream)
915 g_return_if_fail (G_IS_INPUT_STREAM (stream));
917 stream->priv->pending = FALSE;
920 /********************************************
921 * Default implementation of async ops *
922 ********************************************/
926 gsize count_requested;
931 read_async_thread (GSimpleAsyncResult *res,
933 GCancellable *cancellable)
936 GInputStreamClass *class;
937 GError *error = NULL;
939 op = g_simple_async_result_get_op_res_gpointer (res);
941 class = G_INPUT_STREAM_GET_CLASS (object);
943 op->count_read = class->read_fn (G_INPUT_STREAM (object),
944 op->buffer, op->count_requested,
945 cancellable, &error);
946 if (op->count_read == -1)
947 g_simple_async_result_take_error (res, error);
951 g_input_stream_real_read_async (GInputStream *stream,
955 GCancellable *cancellable,
956 GAsyncReadyCallback callback,
959 GSimpleAsyncResult *res;
962 op = g_new (ReadData, 1);
963 res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_input_stream_real_read_async);
964 g_simple_async_result_set_op_res_gpointer (res, op, g_free);
966 op->count_requested = count;
968 g_simple_async_result_run_in_thread (res, read_async_thread, io_priority, cancellable);
969 g_object_unref (res);
973 g_input_stream_real_read_finish (GInputStream *stream,
974 GAsyncResult *result,
977 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
980 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) ==
981 g_input_stream_real_read_async);
983 op = g_simple_async_result_get_op_res_gpointer (simple);
985 return op->count_read;
989 gsize count_requested;
990 gssize count_skipped;
995 skip_async_thread (GSimpleAsyncResult *res,
997 GCancellable *cancellable)
1000 GInputStreamClass *class;
1001 GError *error = NULL;
1003 class = G_INPUT_STREAM_GET_CLASS (object);
1004 op = g_simple_async_result_get_op_res_gpointer (res);
1005 op->count_skipped = class->skip (G_INPUT_STREAM (object),
1006 op->count_requested,
1007 cancellable, &error);
1008 if (op->count_skipped == -1)
1009 g_simple_async_result_take_error (res, error);
1015 gsize count_skipped;
1017 GCancellable *cancellable;
1019 GAsyncReadyCallback callback;
1020 } SkipFallbackAsyncData;
1023 skip_callback_wrapper (GObject *source_object,
1027 GInputStreamClass *class;
1028 SkipFallbackAsyncData *data = user_data;
1030 GSimpleAsyncResult *simple;
1031 GError *error = NULL;
1034 ret = g_input_stream_read_finish (G_INPUT_STREAM (source_object), res, &error);
1039 data->count_skipped += ret;
1041 if (data->count > 0)
1043 class = G_INPUT_STREAM_GET_CLASS (source_object);
1044 class->read_async (G_INPUT_STREAM (source_object), data->buffer, MIN (8192, data->count), data->io_prio, data->cancellable,
1045 skip_callback_wrapper, data);
1050 op = g_new0 (SkipData, 1);
1051 op->count_skipped = data->count_skipped;
1052 simple = g_simple_async_result_new (source_object,
1053 data->callback, data->user_data,
1054 g_input_stream_real_skip_async);
1056 g_simple_async_result_set_op_res_gpointer (simple, op, g_free);
1060 if (data->count_skipped &&
1061 g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED))
1062 /* No error, return partial read */
1063 g_error_free (error);
1065 g_simple_async_result_take_error (simple, error);
1068 /* Complete immediately, not in idle, since we're already in a mainloop callout */
1069 g_simple_async_result_complete (simple);
1070 g_object_unref (simple);
1076 g_input_stream_real_skip_async (GInputStream *stream,
1079 GCancellable *cancellable,
1080 GAsyncReadyCallback callback,
1083 GInputStreamClass *class;
1085 SkipFallbackAsyncData *data;
1086 GSimpleAsyncResult *res;
1088 class = G_INPUT_STREAM_GET_CLASS (stream);
1090 if (class->read_async == g_input_stream_real_read_async)
1092 /* Read is thread-using async fallback.
1093 * Make skip use threads too, so that we can use a possible sync skip
1094 * implementation. */
1095 op = g_new0 (SkipData, 1);
1097 res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
1098 g_input_stream_real_skip_async);
1100 g_simple_async_result_set_op_res_gpointer (res, op, g_free);
1102 op->count_requested = count;
1104 g_simple_async_result_run_in_thread (res, skip_async_thread, io_priority, cancellable);
1105 g_object_unref (res);
1109 /* TODO: Skip fallback uses too much memory, should do multiple read calls */
1111 /* There is a custom async read function, lets use that. */
1112 data = g_new (SkipFallbackAsyncData, 1);
1113 data->count = count;
1114 data->count_skipped = 0;
1115 data->io_prio = io_priority;
1116 data->cancellable = cancellable;
1117 data->callback = callback;
1118 data->user_data = user_data;
1119 class->read_async (stream, data->buffer, MIN (8192, count), io_priority, cancellable,
1120 skip_callback_wrapper, data);
1126 g_input_stream_real_skip_finish (GInputStream *stream,
1127 GAsyncResult *result,
1130 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1133 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_skip_async);
1134 op = g_simple_async_result_get_op_res_gpointer (simple);
1135 return op->count_skipped;
1139 close_async_thread (GSimpleAsyncResult *res,
1141 GCancellable *cancellable)
1143 GInputStreamClass *class;
1144 GError *error = NULL;
1147 /* Auto handling of cancelation disabled, and ignore
1148 cancellation, since we want to close things anyway, although
1149 possibly in a quick-n-dirty way. At least we never want to leak
1152 class = G_INPUT_STREAM_GET_CLASS (object);
1153 if (class->close_fn)
1155 result = class->close_fn (G_INPUT_STREAM (object), cancellable, &error);
1157 g_simple_async_result_take_error (res, error);
1162 g_input_stream_real_close_async (GInputStream *stream,
1164 GCancellable *cancellable,
1165 GAsyncReadyCallback callback,
1168 GSimpleAsyncResult *res;
1170 res = g_simple_async_result_new (G_OBJECT (stream),
1173 g_input_stream_real_close_async);
1175 g_simple_async_result_set_handle_cancellation (res, FALSE);
1177 g_simple_async_result_run_in_thread (res,
1181 g_object_unref (res);
1185 g_input_stream_real_close_finish (GInputStream *stream,
1186 GAsyncResult *result,
1189 GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result);
1190 g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_close_async);