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"
32 #include "gpollableinputstream.h"
35 * SECTION:ginputstream
36 * @short_description: Base class for implementing streaming input
39 * #GInputStream has functions to read from a stream (g_input_stream_read()),
40 * to close a stream (g_input_stream_close()) and to skip some content
41 * (g_input_stream_skip()).
43 * To copy the content of an input stream to an output stream without
44 * manually handling the reads and writes, use g_output_stream_splice().
46 * All of these functions have async variants too.
49 struct _GInputStreamPrivate {
52 GAsyncReadyCallback outstanding_callback;
55 G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE (GInputStream, g_input_stream, G_TYPE_OBJECT)
57 static gssize g_input_stream_real_skip (GInputStream *stream,
59 GCancellable *cancellable,
61 static void g_input_stream_real_read_async (GInputStream *stream,
65 GCancellable *cancellable,
66 GAsyncReadyCallback callback,
68 static gssize g_input_stream_real_read_finish (GInputStream *stream,
71 static void g_input_stream_real_skip_async (GInputStream *stream,
74 GCancellable *cancellable,
75 GAsyncReadyCallback callback,
77 static gssize g_input_stream_real_skip_finish (GInputStream *stream,
80 static void g_input_stream_real_close_async (GInputStream *stream,
82 GCancellable *cancellable,
83 GAsyncReadyCallback callback,
85 static gboolean g_input_stream_real_close_finish (GInputStream *stream,
90 g_input_stream_dispose (GObject *object)
94 stream = G_INPUT_STREAM (object);
96 if (!stream->priv->closed)
97 g_input_stream_close (stream, NULL, NULL);
99 G_OBJECT_CLASS (g_input_stream_parent_class)->dispose (object);
104 g_input_stream_class_init (GInputStreamClass *klass)
106 GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
108 gobject_class->dispose = g_input_stream_dispose;
110 klass->skip = g_input_stream_real_skip;
111 klass->read_async = g_input_stream_real_read_async;
112 klass->read_finish = g_input_stream_real_read_finish;
113 klass->skip_async = g_input_stream_real_skip_async;
114 klass->skip_finish = g_input_stream_real_skip_finish;
115 klass->close_async = g_input_stream_real_close_async;
116 klass->close_finish = g_input_stream_real_close_finish;
120 g_input_stream_init (GInputStream *stream)
122 stream->priv = g_input_stream_get_instance_private (stream);
126 * g_input_stream_read:
127 * @stream: a #GInputStream.
128 * @buffer: (array length=count) (element-type guint8): a buffer to
129 * read data into (which should be at least count bytes long).
130 * @count: the number of bytes that will be read from the stream
131 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
132 * @error: location to store the error occurring, or %NULL to ignore
134 * Tries to read @count bytes from the stream into the buffer starting at
135 * @buffer. Will block during this read.
137 * If count is zero returns zero and does nothing. A value of @count
138 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
140 * On success, the number of bytes read into the buffer is returned.
141 * It is not an error if this is not the same as the requested size, as it
142 * can happen e.g. near the end of a file. Zero is returned on end of file
143 * (or if @count is zero), but never otherwise.
145 * If @cancellable is not %NULL, then the operation can be cancelled by
146 * triggering the cancellable object from another thread. If the operation
147 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
148 * operation was partially finished when the operation was cancelled the
149 * partial result will be returned, without an error.
151 * On error -1 is returned and @error is set accordingly.
153 * Return value: Number of bytes read, or -1 on error, or 0 on end of file.
156 g_input_stream_read (GInputStream *stream,
159 GCancellable *cancellable,
162 GInputStreamClass *class;
165 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
166 g_return_val_if_fail (buffer != NULL, 0);
171 if (((gssize) count) < 0)
173 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
174 _("Too large count value passed to %s"), G_STRFUNC);
178 class = G_INPUT_STREAM_GET_CLASS (stream);
180 if (class->read_fn == NULL)
182 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
183 _("Input stream doesn't implement read"));
187 if (!g_input_stream_set_pending (stream, error))
191 g_cancellable_push_current (cancellable);
193 res = class->read_fn (stream, buffer, count, cancellable, error);
196 g_cancellable_pop_current (cancellable);
198 g_input_stream_clear_pending (stream);
204 * g_input_stream_read_all:
205 * @stream: a #GInputStream.
206 * @buffer: (array length=count) (element-type guint8): a buffer to
207 * read data into (which should be at least count bytes long).
208 * @count: the number of bytes that will be read from the stream
209 * @bytes_read: (out): location to store the number of bytes that was read from the stream
210 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
211 * @error: location to store the error occurring, or %NULL to ignore
213 * Tries to read @count bytes from the stream into the buffer starting at
214 * @buffer. Will block during this read.
216 * This function is similar to g_input_stream_read(), except it tries to
217 * read as many bytes as requested, only stopping on an error or end of stream.
219 * On a successful read of @count bytes, or if we reached the end of the
220 * stream, %TRUE is returned, and @bytes_read is set to the number of bytes
223 * If there is an error during the operation %FALSE is returned and @error
224 * is set to indicate the error status, @bytes_read is updated to contain
225 * the number of bytes read into @buffer before the error occurred.
227 * Return value: %TRUE on success, %FALSE if there was an error
230 g_input_stream_read_all (GInputStream *stream,
234 GCancellable *cancellable,
240 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
241 g_return_val_if_fail (buffer != NULL, FALSE);
244 while (_bytes_read < count)
246 res = g_input_stream_read (stream, (char *)buffer + _bytes_read, count - _bytes_read,
251 *bytes_read = _bytes_read;
262 *bytes_read = _bytes_read;
267 * g_input_stream_read_bytes:
268 * @stream: a #GInputStream.
269 * @count: maximum number of bytes that will be read from the stream. Common
270 * values include 4096 and 8192.
271 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
272 * @error: location to store the error occurring, or %NULL to ignore
274 * Like g_input_stream_read(), this tries to read @count bytes from
275 * the stream in a blocking fashion. However, rather than reading into
276 * a user-supplied buffer, this will create a new #GBytes containing
277 * the data that was read. This may be easier to use from language
280 * If count is zero, returns a zero-length #GBytes and does nothing. A
281 * value of @count larger than %G_MAXSSIZE will cause a
282 * %G_IO_ERROR_INVALID_ARGUMENT error.
284 * On success, a new #GBytes is returned. It is not an error if the
285 * size of this object is not the same as the requested size, as it
286 * can happen e.g. near the end of a file. A zero-length #GBytes is
287 * returned on end of file (or if @count is zero), but never
290 * If @cancellable is not %NULL, then the operation can be cancelled by
291 * triggering the cancellable object from another thread. If the operation
292 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
293 * operation was partially finished when the operation was cancelled the
294 * partial result will be returned, without an error.
296 * On error %NULL is returned and @error is set accordingly.
298 * Return value: a new #GBytes, or %NULL on error
301 g_input_stream_read_bytes (GInputStream *stream,
303 GCancellable *cancellable,
309 buf = g_malloc (count);
310 nread = g_input_stream_read (stream, buf, count, cancellable, error);
319 return g_bytes_new_static ("", 0);
322 return g_bytes_new_take (buf, nread);
326 * g_input_stream_skip:
327 * @stream: a #GInputStream.
328 * @count: the number of bytes that will be skipped from the stream
329 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
330 * @error: location to store the error occurring, or %NULL to ignore
332 * Tries to skip @count bytes from the stream. Will block during the operation.
334 * This is identical to g_input_stream_read(), from a behaviour standpoint,
335 * but the bytes that are skipped are not returned to the user. Some
336 * streams have an implementation that is more efficient than reading the data.
338 * This function is optional for inherited classes, as the default implementation
339 * emulates it using read.
341 * If @cancellable is not %NULL, then the operation can be cancelled by
342 * triggering the cancellable object from another thread. If the operation
343 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
344 * operation was partially finished when the operation was cancelled the
345 * partial result will be returned, without an error.
347 * Return value: Number of bytes skipped, or -1 on error
350 g_input_stream_skip (GInputStream *stream,
352 GCancellable *cancellable,
355 GInputStreamClass *class;
358 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
363 if (((gssize) count) < 0)
365 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
366 _("Too large count value passed to %s"), G_STRFUNC);
370 class = G_INPUT_STREAM_GET_CLASS (stream);
372 if (!g_input_stream_set_pending (stream, error))
376 g_cancellable_push_current (cancellable);
378 res = class->skip (stream, count, cancellable, error);
381 g_cancellable_pop_current (cancellable);
383 g_input_stream_clear_pending (stream);
389 g_input_stream_real_skip (GInputStream *stream,
391 GCancellable *cancellable,
394 GInputStreamClass *class;
395 gssize ret, read_bytes;
399 if (G_IS_SEEKABLE (stream) && g_seekable_can_seek (G_SEEKABLE (stream)))
401 if (g_seekable_seek (G_SEEKABLE (stream),
409 /* If not seekable, or seek failed, fall back to reading data: */
411 class = G_INPUT_STREAM_GET_CLASS (stream);
418 ret = class->read_fn (stream, buffer, MIN (sizeof (buffer), count),
419 cancellable, &my_error);
422 if (read_bytes > 0 &&
423 my_error->domain == G_IO_ERROR &&
424 my_error->code == G_IO_ERROR_CANCELLED)
426 g_error_free (my_error);
430 g_propagate_error (error, my_error);
437 if (ret == 0 || count == 0)
443 * g_input_stream_close:
444 * @stream: A #GInputStream.
445 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
446 * @error: location to store the error occurring, or %NULL to ignore
448 * Closes the stream, releasing resources related to it.
450 * Once the stream is closed, all other operations will return %G_IO_ERROR_CLOSED.
451 * Closing a stream multiple times will not return an error.
453 * Streams will be automatically closed when the last reference
454 * is dropped, but you might want to call this function to make sure
455 * resources are released as early as possible.
457 * Some streams might keep the backing store of the stream (e.g. a file descriptor)
458 * open after the stream is closed. See the documentation for the individual
459 * stream for details.
461 * On failure the first error that happened will be reported, but the close
462 * operation will finish as much as possible. A stream that failed to
463 * close will still return %G_IO_ERROR_CLOSED for all operations. Still, it
464 * is important to check and report the error to the user.
466 * If @cancellable is not %NULL, then the operation can be cancelled by
467 * triggering the cancellable object from another thread. If the operation
468 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
469 * Cancelling a close will still leave the stream closed, but some streams
470 * can use a faster close that doesn't block to e.g. check errors.
472 * Return value: %TRUE on success, %FALSE on failure
475 g_input_stream_close (GInputStream *stream,
476 GCancellable *cancellable,
479 GInputStreamClass *class;
482 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
484 class = G_INPUT_STREAM_GET_CLASS (stream);
486 if (stream->priv->closed)
491 if (!g_input_stream_set_pending (stream, error))
495 g_cancellable_push_current (cancellable);
498 res = class->close_fn (stream, cancellable, error);
501 g_cancellable_pop_current (cancellable);
503 g_input_stream_clear_pending (stream);
505 stream->priv->closed = TRUE;
511 async_ready_callback_wrapper (GObject *source_object,
515 GInputStream *stream = G_INPUT_STREAM (source_object);
517 g_input_stream_clear_pending (stream);
518 if (stream->priv->outstanding_callback)
519 (*stream->priv->outstanding_callback) (source_object, res, user_data);
520 g_object_unref (stream);
524 async_ready_close_callback_wrapper (GObject *source_object,
528 GInputStream *stream = G_INPUT_STREAM (source_object);
530 g_input_stream_clear_pending (stream);
531 stream->priv->closed = TRUE;
532 if (stream->priv->outstanding_callback)
533 (*stream->priv->outstanding_callback) (source_object, res, user_data);
534 g_object_unref (stream);
538 * g_input_stream_read_async:
539 * @stream: A #GInputStream.
540 * @buffer: (array length=count) (element-type guint8): a buffer to
541 * read data into (which should be at least count bytes long).
542 * @count: the number of bytes that will be read from the stream
543 * @io_priority: the <link linkend="io-priority">I/O priority</link>
545 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
546 * @callback: (scope async): callback to call when the request is satisfied
547 * @user_data: (closure): the data to pass to callback function
549 * Request an asynchronous read of @count bytes from the stream into the buffer
550 * starting at @buffer. When the operation is finished @callback will be called.
551 * You can then call g_input_stream_read_finish() to get the result of the
554 * During an async request no other sync and async calls are allowed on @stream, and will
555 * result in %G_IO_ERROR_PENDING errors.
557 * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
559 * On success, the number of bytes read into the buffer will be passed to the
560 * callback. It is not an error if this is not the same as the requested size, as it
561 * can happen e.g. near the end of a file, but generally we try to read
562 * as many bytes as requested. Zero is returned on end of file
563 * (or if @count is zero), but never otherwise.
565 * Any outstanding i/o request with higher priority (lower numerical value) will
566 * be executed before an outstanding request with lower priority. Default
567 * priority is %G_PRIORITY_DEFAULT.
569 * The asyncronous methods have a default fallback that uses threads to implement
570 * asynchronicity, so they are optional for inheriting classes. However, if you
571 * override one you must override all.
574 g_input_stream_read_async (GInputStream *stream,
578 GCancellable *cancellable,
579 GAsyncReadyCallback callback,
582 GInputStreamClass *class;
583 GError *error = NULL;
585 g_return_if_fail (G_IS_INPUT_STREAM (stream));
586 g_return_if_fail (buffer != NULL);
592 task = g_task_new (stream, cancellable, callback, user_data);
593 g_task_set_source_tag (task, g_input_stream_read_async);
594 g_task_return_int (task, 0);
595 g_object_unref (task);
599 if (((gssize) count) < 0)
601 g_task_report_new_error (stream, callback, user_data,
602 g_input_stream_read_async,
603 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
604 _("Too large count value passed to %s"),
609 if (!g_input_stream_set_pending (stream, &error))
611 g_task_report_error (stream, callback, user_data,
612 g_input_stream_read_async,
617 class = G_INPUT_STREAM_GET_CLASS (stream);
618 stream->priv->outstanding_callback = callback;
619 g_object_ref (stream);
620 class->read_async (stream, buffer, count, io_priority, cancellable,
621 async_ready_callback_wrapper, user_data);
625 * g_input_stream_read_finish:
626 * @stream: a #GInputStream.
627 * @result: a #GAsyncResult.
628 * @error: a #GError location to store the error occurring, or %NULL to
631 * Finishes an asynchronous stream read operation.
633 * Returns: number of bytes read in, or -1 on error, or 0 on end of file.
636 g_input_stream_read_finish (GInputStream *stream,
637 GAsyncResult *result,
640 GInputStreamClass *class;
642 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
643 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
645 if (g_async_result_legacy_propagate_error (result, error))
647 else if (g_async_result_is_tagged (result, g_input_stream_read_async))
648 return g_task_propagate_int (G_TASK (result), error);
650 class = G_INPUT_STREAM_GET_CLASS (stream);
651 return class->read_finish (stream, result, error);
655 read_bytes_callback (GObject *stream,
656 GAsyncResult *result,
659 GTask *task = user_data;
660 guchar *buf = g_task_get_task_data (task);
661 GError *error = NULL;
663 GBytes *bytes = NULL;
665 nread = g_input_stream_read_finish (G_INPUT_STREAM (stream),
670 g_task_return_error (task, error);
675 bytes = g_bytes_new_static ("", 0);
678 bytes = g_bytes_new_take (buf, nread);
681 g_task_return_pointer (task, bytes, (GDestroyNotify)g_bytes_unref);
683 g_object_unref (task);
687 * g_input_stream_read_bytes_async:
688 * @stream: A #GInputStream.
689 * @count: the number of bytes that will be read from the stream
690 * @io_priority: the <link linkend="io-priority">I/O priority</link>
692 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
693 * @callback: (scope async): callback to call when the request is satisfied
694 * @user_data: (closure): the data to pass to callback function
696 * Request an asynchronous read of @count bytes from the stream into a
697 * new #GBytes. When the operation is finished @callback will be
698 * called. You can then call g_input_stream_read_bytes_finish() to get the
699 * result of the operation.
701 * During an async request no other sync and async calls are allowed
702 * on @stream, and will result in %G_IO_ERROR_PENDING errors.
704 * A value of @count larger than %G_MAXSSIZE will cause a
705 * %G_IO_ERROR_INVALID_ARGUMENT error.
707 * On success, the new #GBytes will be passed to the callback. It is
708 * not an error if this is smaller than the requested size, as it can
709 * happen e.g. near the end of a file, but generally we try to read as
710 * many bytes as requested. Zero is returned on end of file (or if
711 * @count is zero), but never otherwise.
713 * Any outstanding I/O request with higher priority (lower numerical
714 * value) will be executed before an outstanding request with lower
715 * priority. Default priority is %G_PRIORITY_DEFAULT.
718 g_input_stream_read_bytes_async (GInputStream *stream,
721 GCancellable *cancellable,
722 GAsyncReadyCallback callback,
728 task = g_task_new (stream, cancellable, callback, user_data);
729 buf = g_malloc (count);
730 g_task_set_task_data (task, buf, NULL);
732 g_input_stream_read_async (stream, buf, count,
733 io_priority, cancellable,
734 read_bytes_callback, task);
738 * g_input_stream_read_bytes_finish:
739 * @stream: a #GInputStream.
740 * @result: a #GAsyncResult.
741 * @error: a #GError location to store the error occurring, or %NULL to
744 * Finishes an asynchronous stream read-into-#GBytes operation.
746 * Returns: the newly-allocated #GBytes, or %NULL on error
749 g_input_stream_read_bytes_finish (GInputStream *stream,
750 GAsyncResult *result,
753 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), NULL);
754 g_return_val_if_fail (g_task_is_valid (result, stream), NULL);
756 return g_task_propagate_pointer (G_TASK (result), error);
760 * g_input_stream_skip_async:
761 * @stream: A #GInputStream.
762 * @count: the number of bytes that will be skipped from the stream
763 * @io_priority: the <link linkend="io-priority">I/O priority</link>
765 * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
766 * @callback: (scope async): callback to call when the request is satisfied
767 * @user_data: (closure): the data to pass to callback function
769 * Request an asynchronous skip of @count bytes from the stream.
770 * When the operation is finished @callback will be called.
771 * You can then call g_input_stream_skip_finish() to get the result
774 * During an async request no other sync and async calls are allowed,
775 * and will result in %G_IO_ERROR_PENDING errors.
777 * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
779 * On success, the number of bytes skipped will be passed to the callback.
780 * It is not an error if this is not the same as the requested size, as it
781 * can happen e.g. near the end of a file, but generally we try to skip
782 * as many bytes as requested. Zero is returned on end of file
783 * (or if @count is zero), but never otherwise.
785 * Any outstanding i/o request with higher priority (lower numerical value)
786 * will be executed before an outstanding request with lower priority.
787 * Default priority is %G_PRIORITY_DEFAULT.
789 * The asynchronous methods have a default fallback that uses threads to
790 * implement asynchronicity, so they are optional for inheriting classes.
791 * However, if you override one, you must override all.
794 g_input_stream_skip_async (GInputStream *stream,
797 GCancellable *cancellable,
798 GAsyncReadyCallback callback,
801 GInputStreamClass *class;
802 GError *error = NULL;
804 g_return_if_fail (G_IS_INPUT_STREAM (stream));
810 task = g_task_new (stream, cancellable, callback, user_data);
811 g_task_set_source_tag (task, g_input_stream_skip_async);
812 g_task_return_int (task, 0);
813 g_object_unref (task);
817 if (((gssize) count) < 0)
819 g_task_report_new_error (stream, callback, user_data,
820 g_input_stream_skip_async,
821 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
822 _("Too large count value passed to %s"),
827 if (!g_input_stream_set_pending (stream, &error))
829 g_task_report_error (stream, callback, user_data,
830 g_input_stream_skip_async,
835 class = G_INPUT_STREAM_GET_CLASS (stream);
836 stream->priv->outstanding_callback = callback;
837 g_object_ref (stream);
838 class->skip_async (stream, count, io_priority, cancellable,
839 async_ready_callback_wrapper, user_data);
843 * g_input_stream_skip_finish:
844 * @stream: a #GInputStream.
845 * @result: a #GAsyncResult.
846 * @error: a #GError location to store the error occurring, or %NULL to
849 * Finishes a stream skip operation.
851 * Returns: the size of the bytes skipped, or %-1 on error.
854 g_input_stream_skip_finish (GInputStream *stream,
855 GAsyncResult *result,
858 GInputStreamClass *class;
860 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
861 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
863 if (g_async_result_legacy_propagate_error (result, error))
865 else if (g_async_result_is_tagged (result, g_input_stream_skip_async))
866 return g_task_propagate_int (G_TASK (result), error);
868 class = G_INPUT_STREAM_GET_CLASS (stream);
869 return class->skip_finish (stream, result, error);
873 * g_input_stream_close_async:
874 * @stream: A #GInputStream.
875 * @io_priority: the <link linkend="io-priority">I/O priority</link>
877 * @cancellable: (allow-none): optional cancellable object
878 * @callback: (scope async): callback to call when the request is satisfied
879 * @user_data: (closure): the data to pass to callback function
881 * Requests an asynchronous closes of the stream, releasing resources related to it.
882 * When the operation is finished @callback will be called.
883 * You can then call g_input_stream_close_finish() to get the result of the
886 * For behaviour details see g_input_stream_close().
888 * The asyncronous methods have a default fallback that uses threads to implement
889 * asynchronicity, so they are optional for inheriting classes. However, if you
890 * override one you must override all.
893 g_input_stream_close_async (GInputStream *stream,
895 GCancellable *cancellable,
896 GAsyncReadyCallback callback,
899 GInputStreamClass *class;
900 GError *error = NULL;
902 g_return_if_fail (G_IS_INPUT_STREAM (stream));
904 if (stream->priv->closed)
908 task = g_task_new (stream, cancellable, callback, user_data);
909 g_task_set_source_tag (task, g_input_stream_close_async);
910 g_task_return_boolean (task, TRUE);
911 g_object_unref (task);
915 if (!g_input_stream_set_pending (stream, &error))
917 g_task_report_error (stream, callback, user_data,
918 g_input_stream_close_async,
923 class = G_INPUT_STREAM_GET_CLASS (stream);
924 stream->priv->outstanding_callback = callback;
925 g_object_ref (stream);
926 class->close_async (stream, io_priority, cancellable,
927 async_ready_close_callback_wrapper, user_data);
931 * g_input_stream_close_finish:
932 * @stream: a #GInputStream.
933 * @result: a #GAsyncResult.
934 * @error: a #GError location to store the error occurring, or %NULL to
937 * Finishes closing a stream asynchronously, started from g_input_stream_close_async().
939 * Returns: %TRUE if the stream was closed successfully.
942 g_input_stream_close_finish (GInputStream *stream,
943 GAsyncResult *result,
946 GInputStreamClass *class;
948 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
949 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
951 if (g_async_result_legacy_propagate_error (result, error))
953 else if (g_async_result_is_tagged (result, g_input_stream_close_async))
954 return g_task_propagate_boolean (G_TASK (result), error);
956 class = G_INPUT_STREAM_GET_CLASS (stream);
957 return class->close_finish (stream, result, error);
961 * g_input_stream_is_closed:
962 * @stream: input stream.
964 * Checks if an input stream is closed.
966 * Returns: %TRUE if the stream is closed.
969 g_input_stream_is_closed (GInputStream *stream)
971 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
973 return stream->priv->closed;
977 * g_input_stream_has_pending:
978 * @stream: input stream.
980 * Checks if an input stream has pending actions.
982 * Returns: %TRUE if @stream has pending actions.
985 g_input_stream_has_pending (GInputStream *stream)
987 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
989 return stream->priv->pending;
993 * g_input_stream_set_pending:
994 * @stream: input stream
995 * @error: a #GError location to store the error occurring, or %NULL to
998 * Sets @stream to have actions pending. If the pending flag is
999 * already set or @stream is closed, it will return %FALSE and set
1002 * Return value: %TRUE if pending was previously unset and is now set.
1005 g_input_stream_set_pending (GInputStream *stream, GError **error)
1007 g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
1009 if (stream->priv->closed)
1011 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
1012 _("Stream is already closed"));
1016 if (stream->priv->pending)
1018 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
1019 /* Translators: This is an error you get if there is already an
1020 * operation running against this stream when you try to start
1022 _("Stream has outstanding operation"));
1026 stream->priv->pending = TRUE;
1031 * g_input_stream_clear_pending:
1032 * @stream: input stream
1034 * Clears the pending flag on @stream.
1037 g_input_stream_clear_pending (GInputStream *stream)
1039 g_return_if_fail (G_IS_INPUT_STREAM (stream));
1041 stream->priv->pending = FALSE;
1044 /********************************************
1045 * Default implementation of async ops *
1046 ********************************************/
1054 free_read_data (ReadData *op)
1056 g_slice_free (ReadData, op);
1060 read_async_thread (GTask *task,
1061 gpointer source_object,
1063 GCancellable *cancellable)
1065 GInputStream *stream = source_object;
1066 ReadData *op = task_data;
1067 GInputStreamClass *class;
1068 GError *error = NULL;
1071 class = G_INPUT_STREAM_GET_CLASS (stream);
1073 nread = class->read_fn (stream,
1074 op->buffer, op->count,
1075 g_task_get_cancellable (task),
1078 g_task_return_error (task, error);
1080 g_task_return_int (task, nread);
1083 static void read_async_pollable (GPollableInputStream *stream,
1087 read_async_pollable_ready (GPollableInputStream *stream,
1090 GTask *task = user_data;
1092 read_async_pollable (stream, task);
1097 read_async_pollable (GPollableInputStream *stream,
1100 ReadData *op = g_task_get_task_data (task);
1101 GError *error = NULL;
1104 if (g_task_return_error_if_cancelled (task))
1107 nread = G_POLLABLE_INPUT_STREAM_GET_INTERFACE (stream)->
1108 read_nonblocking (stream, op->buffer, op->count, &error);
1110 if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
1114 g_error_free (error);
1116 source = g_pollable_input_stream_create_source (stream,
1117 g_task_get_cancellable (task));
1118 g_task_attach_source (task, source,
1119 (GSourceFunc) read_async_pollable_ready);
1120 g_source_unref (source);
1125 g_task_return_error (task, error);
1127 g_task_return_int (task, nread);
1128 /* g_input_stream_real_read_async() unrefs task */
1131 #define CAN_DO_NONBLOCKING_READS(stream) \
1132 (G_IS_POLLABLE_INPUT_STREAM (stream) && \
1133 g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (stream)))
1137 g_input_stream_real_read_async (GInputStream *stream,
1141 GCancellable *cancellable,
1142 GAsyncReadyCallback callback,
1148 op = g_slice_new0 (ReadData);
1149 task = g_task_new (stream, cancellable, callback, user_data);
1150 g_task_set_task_data (task, op, (GDestroyNotify) free_read_data);
1151 g_task_set_priority (task, io_priority);
1152 op->buffer = buffer;
1155 if (CAN_DO_NONBLOCKING_READS (stream))
1156 read_async_pollable (G_POLLABLE_INPUT_STREAM (stream), task);
1158 g_task_run_in_thread (task, read_async_thread);
1159 g_object_unref (task);
1163 g_input_stream_real_read_finish (GInputStream *stream,
1164 GAsyncResult *result,
1167 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1169 return g_task_propagate_int (G_TASK (result), error);
1174 skip_async_thread (GTask *task,
1175 gpointer source_object,
1177 GCancellable *cancellable)
1179 GInputStream *stream = source_object;
1180 gsize count = GPOINTER_TO_SIZE (task_data);
1181 GInputStreamClass *class;
1182 GError *error = NULL;
1185 class = G_INPUT_STREAM_GET_CLASS (stream);
1186 ret = class->skip (stream, count,
1187 g_task_get_cancellable (task),
1190 g_task_return_error (task, error);
1192 g_task_return_int (task, ret);
1198 gsize count_skipped;
1199 } SkipFallbackAsyncData;
1202 skip_callback_wrapper (GObject *source_object,
1206 GInputStreamClass *class;
1207 GTask *task = user_data;
1208 SkipFallbackAsyncData *data = g_task_get_task_data (task);
1209 GError *error = NULL;
1212 ret = g_input_stream_read_finish (G_INPUT_STREAM (source_object), res, &error);
1217 data->count_skipped += ret;
1219 if (data->count > 0)
1221 class = G_INPUT_STREAM_GET_CLASS (source_object);
1222 class->read_async (G_INPUT_STREAM (source_object),
1223 data->buffer, MIN (8192, data->count),
1224 g_task_get_priority (task),
1225 g_task_get_cancellable (task),
1226 skip_callback_wrapper, task);
1232 g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) &&
1233 data->count_skipped)
1235 /* No error, return partial read */
1236 g_clear_error (&error);
1240 g_task_return_error (task, error);
1242 g_task_return_int (task, data->count_skipped);
1243 g_object_unref (task);
1247 g_input_stream_real_skip_async (GInputStream *stream,
1250 GCancellable *cancellable,
1251 GAsyncReadyCallback callback,
1254 GInputStreamClass *class;
1255 SkipFallbackAsyncData *data;
1258 class = G_INPUT_STREAM_GET_CLASS (stream);
1260 task = g_task_new (stream, cancellable, callback, user_data);
1261 g_task_set_priority (task, io_priority);
1263 if (class->read_async == g_input_stream_real_read_async &&
1264 !CAN_DO_NONBLOCKING_READS (stream))
1266 /* Read is thread-using async fallback.
1267 * Make skip use threads too, so that we can use a possible sync skip
1268 * implementation. */
1269 g_task_set_task_data (task, GSIZE_TO_POINTER (count), NULL);
1271 g_task_run_in_thread (task, skip_async_thread);
1272 g_object_unref (task);
1276 /* TODO: Skip fallback uses too much memory, should do multiple read calls */
1278 /* There is a custom async read function, lets use that. */
1279 data = g_new (SkipFallbackAsyncData, 1);
1280 data->count = count;
1281 data->count_skipped = 0;
1282 g_task_set_task_data (task, data, g_free);
1283 g_task_set_check_cancellable (task, FALSE);
1284 class->read_async (stream, data->buffer, MIN (8192, count), io_priority, cancellable,
1285 skip_callback_wrapper, task);
1291 g_input_stream_real_skip_finish (GInputStream *stream,
1292 GAsyncResult *result,
1295 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1297 return g_task_propagate_int (G_TASK (result), error);
1301 close_async_thread (GTask *task,
1302 gpointer source_object,
1304 GCancellable *cancellable)
1306 GInputStream *stream = source_object;
1307 GInputStreamClass *class;
1308 GError *error = NULL;
1311 class = G_INPUT_STREAM_GET_CLASS (stream);
1312 if (class->close_fn)
1314 result = class->close_fn (stream,
1315 g_task_get_cancellable (task),
1319 g_task_return_error (task, error);
1324 g_task_return_boolean (task, TRUE);
1328 g_input_stream_real_close_async (GInputStream *stream,
1330 GCancellable *cancellable,
1331 GAsyncReadyCallback callback,
1336 task = g_task_new (stream, cancellable, callback, user_data);
1337 g_task_set_check_cancellable (task, FALSE);
1338 g_task_set_priority (task, io_priority);
1340 g_task_run_in_thread (task, close_async_thread);
1341 g_object_unref (task);
1345 g_input_stream_real_close_finish (GInputStream *stream,
1346 GAsyncResult *result,
1349 g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1351 return g_task_propagate_boolean (G_TASK (result), error);