X-Git-Url: http://review.tizen.org/git/?a=blobdiff_plain;f=gio%2Fginputstream.c;h=b0dce7655d150922da4cbd4c3e5a94203ac2685c;hb=e55a953642a9e402f4363f9fa347b6061dd78990;hp=8c61285c8118262d4142cb446143ca553b5069d1;hpb=800d6ff1115b27bf0e9cce818a0511048e2f1cc5;p=platform%2Fupstream%2Fglib.git diff --git a/gio/ginputstream.c b/gio/ginputstream.c index 8c61285..b0dce76 100644 --- a/gio/ginputstream.c +++ b/gio/ginputstream.c @@ -13,9 +13,7 @@ * Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser General - * Public License along with this library; if not, write to the - * Free Software Foundation, Inc., 59 Temple Place, Suite 330, - * Boston, MA 02111-1307, USA. + * Public License along with this library; if not, see . * * Author: Alexander Larsson */ @@ -25,10 +23,10 @@ #include "glibintl.h" #include "ginputstream.h" +#include "gioprivate.h" #include "gseekable.h" #include "gcancellable.h" #include "gasyncresult.h" -#include "gsimpleasyncresult.h" #include "gioerror.h" #include "gpollableinputstream.h" @@ -47,14 +45,14 @@ * All of these functions have async variants too. **/ -G_DEFINE_ABSTRACT_TYPE (GInputStream, g_input_stream, G_TYPE_OBJECT); - struct _GInputStreamPrivate { guint closed : 1; guint pending : 1; GAsyncReadyCallback outstanding_callback; }; +G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE (GInputStream, g_input_stream, G_TYPE_OBJECT) + static gssize g_input_stream_real_skip (GInputStream *stream, gsize count, GCancellable *cancellable, @@ -88,12 +86,6 @@ static gboolean g_input_stream_real_close_finish (GInputStream *stream, GError **error); static void -g_input_stream_finalize (GObject *object) -{ - G_OBJECT_CLASS (g_input_stream_parent_class)->finalize (object); -} - -static void g_input_stream_dispose (GObject *object) { GInputStream *stream; @@ -112,9 +104,6 @@ g_input_stream_class_init (GInputStreamClass *klass) { GObjectClass *gobject_class = G_OBJECT_CLASS (klass); - g_type_class_add_private (klass, sizeof (GInputStreamPrivate)); - - gobject_class->finalize = g_input_stream_finalize; gobject_class->dispose = g_input_stream_dispose; klass->skip = g_input_stream_real_skip; @@ -129,15 +118,14 @@ g_input_stream_class_init (GInputStreamClass *klass) static void g_input_stream_init (GInputStream *stream) { - stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream, - G_TYPE_INPUT_STREAM, - GInputStreamPrivate); + stream->priv = g_input_stream_get_instance_private (stream); } /** * g_input_stream_read: * @stream: a #GInputStream. - * @buffer: a buffer to read data into (which should be at least count bytes long). + * @buffer: (array length=count) (element-type guint8): a buffer to + * read data into (which should be at least count bytes long). * @count: the number of bytes that will be read from the stream * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore. * @error: location to store the error occurring, or %NULL to ignore @@ -153,6 +141,9 @@ g_input_stream_init (GInputStream *stream) * can happen e.g. near the end of a file. Zero is returned on end of file * (or if @count is zero), but never otherwise. * + * The returned @buffer is not a nul-terminated string, it can contain nul bytes + * at any position, and this function doesn't nul-terminate the @buffer. + * * If @cancellable is not %NULL, then the operation can be cancelled by * triggering the cancellable object from another thread. If the operation * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an @@ -161,7 +152,7 @@ g_input_stream_init (GInputStream *stream) * * On error -1 is returned and @error is set accordingly. * - * Return value: Number of bytes read, or -1 on error, or 0 on end of file. + * Returns: Number of bytes read, or -1 on error, or 0 on end of file. **/ gssize g_input_stream_read (GInputStream *stream, @@ -214,7 +205,8 @@ g_input_stream_read (GInputStream *stream, /** * g_input_stream_read_all: * @stream: a #GInputStream. - * @buffer: a buffer to read data into (which should be at least count bytes long). + * @buffer: (array length=count) (element-type guint8): a buffer to + * read data into (which should be at least count bytes long). * @count: the number of bytes that will be read from the stream * @bytes_read: (out): location to store the number of bytes that was read from the stream * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore. @@ -234,7 +226,7 @@ g_input_stream_read (GInputStream *stream, * is set to indicate the error status, @bytes_read is updated to contain * the number of bytes read into @buffer before the error occurred. * - * Return value: %TRUE on success, %FALSE if there was an error + * Returns: %TRUE on success, %FALSE if there was an error **/ gboolean g_input_stream_read_all (GInputStream *stream, @@ -305,7 +297,9 @@ g_input_stream_read_all (GInputStream *stream, * * On error %NULL is returned and @error is set accordingly. * - * Return value: a new #GBytes, or %NULL on error + * Returns: a new #GBytes, or %NULL on error + * + * Since: 2.34 **/ GBytes * g_input_stream_read_bytes (GInputStream *stream, @@ -354,7 +348,7 @@ g_input_stream_read_bytes (GInputStream *stream, * operation was partially finished when the operation was cancelled the * partial result will be returned, without an error. * - * Return value: Number of bytes skipped, or -1 on error + * Returns: Number of bytes skipped, or -1 on error **/ gssize g_input_stream_skip (GInputStream *stream, @@ -479,7 +473,7 @@ g_input_stream_real_skip (GInputStream *stream, * Cancelling a close will still leave the stream closed, but some streams * can use a faster close that doesn't block to e.g. check errors. * - * Return value: %TRUE on success, %FALSE on failure + * Returns: %TRUE on success, %FALSE on failure **/ gboolean g_input_stream_close (GInputStream *stream, @@ -547,9 +541,10 @@ async_ready_close_callback_wrapper (GObject *source_object, /** * g_input_stream_read_async: * @stream: A #GInputStream. - * @buffer: a buffer to read data into (which should be at least count bytes long). + * @buffer: (array length=count) (element-type guint8): a buffer to + * read data into (which should be at least count bytes long). * @count: the number of bytes that will be read from the stream - * @io_priority: the I/O priority + * @io_priority: the [I/O priority][io-priority] * of the request. * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore. * @callback: (scope async): callback to call when the request is satisfied @@ -589,7 +584,6 @@ g_input_stream_read_async (GInputStream *stream, gpointer user_data) { GInputStreamClass *class; - GSimpleAsyncResult *simple; GError *error = NULL; g_return_if_fail (G_IS_INPUT_STREAM (stream)); @@ -597,32 +591,30 @@ g_input_stream_read_async (GInputStream *stream, if (count == 0) { - simple = g_simple_async_result_new (G_OBJECT (stream), - callback, - user_data, - g_input_stream_read_async); - g_simple_async_result_complete_in_idle (simple); - g_object_unref (simple); + GTask *task; + + task = g_task_new (stream, cancellable, callback, user_data); + g_task_set_source_tag (task, g_input_stream_read_async); + g_task_return_int (task, 0); + g_object_unref (task); return; } if (((gssize) count) < 0) { - g_simple_async_report_error_in_idle (G_OBJECT (stream), - callback, - user_data, - G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT, - _("Too large count value passed to %s"), - G_STRFUNC); + g_task_report_new_error (stream, callback, user_data, + g_input_stream_read_async, + G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT, + _("Too large count value passed to %s"), + G_STRFUNC); return; } if (!g_input_stream_set_pending (stream, &error)) { - g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream), - callback, - user_data, - error); + g_task_report_error (stream, callback, user_data, + g_input_stream_read_async, + error); return; } @@ -649,22 +641,15 @@ g_input_stream_read_finish (GInputStream *stream, GAsyncResult *result, GError **error) { - GSimpleAsyncResult *simple; GInputStreamClass *class; g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1); g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1); - if (G_IS_SIMPLE_ASYNC_RESULT (result)) - { - simple = G_SIMPLE_ASYNC_RESULT (result); - if (g_simple_async_result_propagate_error (simple, error)) - return -1; - - /* Special case read of 0 bytes */ - if (g_simple_async_result_get_source_tag (simple) == g_input_stream_read_async) - return 0; - } + if (g_async_result_legacy_propagate_error (result, error)) + return -1; + else if (g_async_result_is_tagged (result, g_input_stream_read_async)) + return g_task_propagate_int (G_TASK (result), error); class = G_INPUT_STREAM_GET_CLASS (stream); return class->read_finish (stream, result, error); @@ -675,8 +660,8 @@ read_bytes_callback (GObject *stream, GAsyncResult *result, gpointer user_data) { - GSimpleAsyncResult *simple = user_data; - guchar *buf = g_simple_async_result_get_op_res_gpointer (simple); + GTask *task = user_data; + guchar *buf = g_task_get_task_data (task); GError *error = NULL; gssize nread; GBytes *bytes = NULL; @@ -686,7 +671,7 @@ read_bytes_callback (GObject *stream, if (nread == -1) { g_free (buf); - g_simple_async_result_take_error (simple, error); + g_task_return_error (task, error); } else if (nread == 0) { @@ -697,20 +682,16 @@ read_bytes_callback (GObject *stream, bytes = g_bytes_new_take (buf, nread); if (bytes) - { - g_simple_async_result_set_op_res_gpointer (simple, bytes, - (GDestroyNotify)g_bytes_unref); - } - g_simple_async_result_complete (simple); - g_object_unref (simple); + g_task_return_pointer (task, bytes, (GDestroyNotify)g_bytes_unref); + + g_object_unref (task); } /** * g_input_stream_read_bytes_async: * @stream: A #GInputStream. * @count: the number of bytes that will be read from the stream - * @io_priority: the I/O priority - * of the request. + * @io_priority: the [I/O priority][io-priority] of the request * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore. * @callback: (scope async): callback to call when the request is satisfied * @user_data: (closure): the data to pass to callback function @@ -735,6 +716,8 @@ read_bytes_callback (GObject *stream, * Any outstanding I/O request with higher priority (lower numerical * value) will be executed before an outstanding request with lower * priority. Default priority is %G_PRIORITY_DEFAULT. + * + * Since: 2.34 **/ void g_input_stream_read_bytes_async (GInputStream *stream, @@ -744,18 +727,16 @@ g_input_stream_read_bytes_async (GInputStream *stream, GAsyncReadyCallback callback, gpointer user_data) { - GSimpleAsyncResult *simple; + GTask *task; guchar *buf; - simple = g_simple_async_result_new (G_OBJECT (stream), - callback, user_data, - g_input_stream_read_bytes_async); + task = g_task_new (stream, cancellable, callback, user_data); buf = g_malloc (count); - g_simple_async_result_set_op_res_gpointer (simple, buf, NULL); + g_task_set_task_data (task, buf, NULL); g_input_stream_read_async (stream, buf, count, - io_priority, cancellable, - read_bytes_callback, simple); + io_priority, cancellable, + read_bytes_callback, task); } /** @@ -768,29 +749,25 @@ g_input_stream_read_bytes_async (GInputStream *stream, * Finishes an asynchronous stream read-into-#GBytes operation. * * Returns: the newly-allocated #GBytes, or %NULL on error + * + * Since: 2.34 **/ GBytes * g_input_stream_read_bytes_finish (GInputStream *stream, GAsyncResult *result, GError **error) { - GSimpleAsyncResult *simple; - g_return_val_if_fail (G_IS_INPUT_STREAM (stream), NULL); - g_return_val_if_fail (g_simple_async_result_is_valid (result, G_OBJECT (stream), g_input_stream_read_bytes_async), NULL); + g_return_val_if_fail (g_task_is_valid (result, stream), NULL); - simple = G_SIMPLE_ASYNC_RESULT (result); - if (g_simple_async_result_propagate_error (simple, error)) - return NULL; - return g_bytes_ref (g_simple_async_result_get_op_res_gpointer (simple)); + return g_task_propagate_pointer (G_TASK (result), error); } /** * g_input_stream_skip_async: * @stream: A #GInputStream. * @count: the number of bytes that will be skipped from the stream - * @io_priority: the I/O priority - * of the request. + * @io_priority: the [I/O priority][io-priority] of the request * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore. * @callback: (scope async): callback to call when the request is satisfied * @user_data: (closure): the data to pass to callback function @@ -828,40 +805,36 @@ g_input_stream_skip_async (GInputStream *stream, gpointer user_data) { GInputStreamClass *class; - GSimpleAsyncResult *simple; GError *error = NULL; g_return_if_fail (G_IS_INPUT_STREAM (stream)); if (count == 0) { - simple = g_simple_async_result_new (G_OBJECT (stream), - callback, - user_data, - g_input_stream_skip_async); + GTask *task; - g_simple_async_result_complete_in_idle (simple); - g_object_unref (simple); + task = g_task_new (stream, cancellable, callback, user_data); + g_task_set_source_tag (task, g_input_stream_skip_async); + g_task_return_int (task, 0); + g_object_unref (task); return; } if (((gssize) count) < 0) { - g_simple_async_report_error_in_idle (G_OBJECT (stream), - callback, - user_data, - G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT, - _("Too large count value passed to %s"), - G_STRFUNC); + g_task_report_new_error (stream, callback, user_data, + g_input_stream_skip_async, + G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT, + _("Too large count value passed to %s"), + G_STRFUNC); return; } if (!g_input_stream_set_pending (stream, &error)) { - g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream), - callback, - user_data, - error); + g_task_report_error (stream, callback, user_data, + g_input_stream_skip_async, + error); return; } @@ -888,22 +861,15 @@ g_input_stream_skip_finish (GInputStream *stream, GAsyncResult *result, GError **error) { - GSimpleAsyncResult *simple; GInputStreamClass *class; g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1); g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1); - if (G_IS_SIMPLE_ASYNC_RESULT (result)) - { - simple = G_SIMPLE_ASYNC_RESULT (result); - if (g_simple_async_result_propagate_error (simple, error)) - return -1; - - /* Special case skip of 0 bytes */ - if (g_simple_async_result_get_source_tag (simple) == g_input_stream_skip_async) - return 0; - } + if (g_async_result_legacy_propagate_error (result, error)) + return -1; + else if (g_async_result_is_tagged (result, g_input_stream_skip_async)) + return g_task_propagate_int (G_TASK (result), error); class = G_INPUT_STREAM_GET_CLASS (stream); return class->skip_finish (stream, result, error); @@ -912,8 +878,7 @@ g_input_stream_skip_finish (GInputStream *stream, /** * g_input_stream_close_async: * @stream: A #GInputStream. - * @io_priority: the I/O priority - * of the request. + * @io_priority: the [I/O priority][io-priority] of the request * @cancellable: (allow-none): optional cancellable object * @callback: (scope async): callback to call when the request is satisfied * @user_data: (closure): the data to pass to callback function @@ -937,29 +902,26 @@ g_input_stream_close_async (GInputStream *stream, gpointer user_data) { GInputStreamClass *class; - GSimpleAsyncResult *simple; GError *error = NULL; g_return_if_fail (G_IS_INPUT_STREAM (stream)); if (stream->priv->closed) { - simple = g_simple_async_result_new (G_OBJECT (stream), - callback, - user_data, - g_input_stream_close_async); + GTask *task; - g_simple_async_result_complete_in_idle (simple); - g_object_unref (simple); + task = g_task_new (stream, cancellable, callback, user_data); + g_task_set_source_tag (task, g_input_stream_close_async); + g_task_return_boolean (task, TRUE); + g_object_unref (task); return; } if (!g_input_stream_set_pending (stream, &error)) { - g_simple_async_report_take_gerror_in_idle (G_OBJECT (stream), - callback, - user_data, - error); + g_task_report_error (stream, callback, user_data, + g_input_stream_close_async, + error); return; } @@ -986,22 +948,15 @@ g_input_stream_close_finish (GInputStream *stream, GAsyncResult *result, GError **error) { - GSimpleAsyncResult *simple; GInputStreamClass *class; g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE); g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE); - if (G_IS_SIMPLE_ASYNC_RESULT (result)) - { - simple = G_SIMPLE_ASYNC_RESULT (result); - if (g_simple_async_result_propagate_error (simple, error)) - return FALSE; - - /* Special case already closed */ - if (g_simple_async_result_get_source_tag (simple) == g_input_stream_close_async) - return TRUE; - } + if (g_async_result_legacy_propagate_error (result, error)) + return FALSE; + else if (g_async_result_is_tagged (result, g_input_stream_close_async)) + return g_task_propagate_boolean (G_TASK (result), error); class = G_INPUT_STREAM_GET_CLASS (stream); return class->close_finish (stream, result, error); @@ -1049,7 +1004,7 @@ g_input_stream_has_pending (GInputStream *stream) * already set or @stream is closed, it will return %FALSE and set * @error. * - * Return value: %TRUE if pending was previously unset and is now set. + * Returns: %TRUE if pending was previously unset and is now set. **/ gboolean g_input_stream_set_pending (GInputStream *stream, GError **error) @@ -1091,102 +1046,116 @@ g_input_stream_clear_pending (GInputStream *stream) stream->priv->pending = FALSE; } +/** + * g_input_stream_async_read_is_via_threads: + * @stream: input stream + * + * Checks if an input stream's read_async function uses threads. + * + * Returns: %TRUE if @stream's read_async function uses threads. + **/ +gboolean +g_input_stream_async_read_is_via_threads (GInputStream *stream) +{ + GInputStreamClass *class; + + g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE); + + class = G_INPUT_STREAM_GET_CLASS (stream); + + return (class->read_async == g_input_stream_real_read_async && + !(G_IS_POLLABLE_INPUT_STREAM (stream) && + g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (stream)))); +} + /******************************************** * Default implementation of async ops * ********************************************/ typedef struct { - void *buffer; - gsize count_requested; - gssize count_read; - - GCancellable *cancellable; - gint io_priority; - gboolean need_idle; + void *buffer; + gsize count; } ReadData; static void free_read_data (ReadData *op) { - if (op->cancellable) - g_object_unref (op->cancellable); g_slice_free (ReadData, op); } static void -read_async_thread (GSimpleAsyncResult *res, - GObject *object, - GCancellable *cancellable) +read_async_thread (GTask *task, + gpointer source_object, + gpointer task_data, + GCancellable *cancellable) { - ReadData *op; + GInputStream *stream = source_object; + ReadData *op = task_data; GInputStreamClass *class; GError *error = NULL; + gssize nread; - op = g_simple_async_result_get_op_res_gpointer (res); - - class = G_INPUT_STREAM_GET_CLASS (object); + class = G_INPUT_STREAM_GET_CLASS (stream); - op->count_read = class->read_fn (G_INPUT_STREAM (object), - op->buffer, op->count_requested, - cancellable, &error); - if (op->count_read == -1) - g_simple_async_result_take_error (res, error); + nread = class->read_fn (stream, + op->buffer, op->count, + g_task_get_cancellable (task), + &error); + if (nread == -1) + g_task_return_error (task, error); + else + g_task_return_int (task, nread); } static void read_async_pollable (GPollableInputStream *stream, - GSimpleAsyncResult *result); + GTask *task); static gboolean read_async_pollable_ready (GPollableInputStream *stream, gpointer user_data) { - GSimpleAsyncResult *result = user_data; + GTask *task = user_data; - read_async_pollable (stream, result); + read_async_pollable (stream, task); return FALSE; } static void read_async_pollable (GPollableInputStream *stream, - GSimpleAsyncResult *result) + GTask *task) { + ReadData *op = g_task_get_task_data (task); GError *error = NULL; - ReadData *op = g_simple_async_result_get_op_res_gpointer (result); + gssize nread; - if (g_cancellable_set_error_if_cancelled (op->cancellable, &error)) - op->count_read = -1; - else - { - op->count_read = G_POLLABLE_INPUT_STREAM_GET_INTERFACE (stream)-> - read_nonblocking (stream, op->buffer, op->count_requested, &error); - } + if (g_task_return_error_if_cancelled (task)) + return; + + nread = G_POLLABLE_INPUT_STREAM_GET_INTERFACE (stream)-> + read_nonblocking (stream, op->buffer, op->count, &error); if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) { GSource *source; g_error_free (error); - op->need_idle = FALSE; - - source = g_pollable_input_stream_create_source (stream, op->cancellable); - g_source_set_callback (source, - (GSourceFunc) read_async_pollable_ready, - g_object_ref (result), g_object_unref); - g_source_set_priority (source, op->io_priority); - g_source_attach (source, g_main_context_get_thread_default ()); + + source = g_pollable_input_stream_create_source (stream, + g_task_get_cancellable (task)); + g_task_attach_source (task, source, + (GSourceFunc) read_async_pollable_ready); g_source_unref (source); return; } - if (op->count_read == -1) - g_simple_async_result_take_error (result, error); - - if (op->need_idle) - g_simple_async_result_complete_in_idle (result); + if (nread == -1) + g_task_return_error (task, error); else - g_simple_async_result_complete (result); + g_task_return_int (task, nread); + /* g_input_stream_real_read_async() unrefs task */ } + static void g_input_stream_real_read_async (GInputStream *stream, void *buffer, @@ -1196,24 +1165,21 @@ g_input_stream_real_read_async (GInputStream *stream, GAsyncReadyCallback callback, gpointer user_data) { - GSimpleAsyncResult *res; + GTask *task; ReadData *op; op = g_slice_new0 (ReadData); - res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, g_input_stream_real_read_async); - g_simple_async_result_set_op_res_gpointer (res, op, (GDestroyNotify) free_read_data); + task = g_task_new (stream, cancellable, callback, user_data); + g_task_set_task_data (task, op, (GDestroyNotify) free_read_data); + g_task_set_priority (task, io_priority); op->buffer = buffer; - op->count_requested = count; - op->cancellable = cancellable ? g_object_ref (cancellable) : NULL; - op->io_priority = io_priority; - op->need_idle = TRUE; - - if (G_IS_POLLABLE_INPUT_STREAM (stream) && - g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (stream))) - read_async_pollable (G_POLLABLE_INPUT_STREAM (stream), res); + op->count = count; + + if (!g_input_stream_async_read_is_via_threads (stream)) + read_async_pollable (G_POLLABLE_INPUT_STREAM (stream), task); else - g_simple_async_result_run_in_thread (res, read_async_thread, io_priority, cancellable); - g_object_unref (res); + g_task_run_in_thread (task, read_async_thread); + g_object_unref (task); } static gssize @@ -1221,49 +1187,38 @@ g_input_stream_real_read_finish (GInputStream *stream, GAsyncResult *result, GError **error) { - GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result); - ReadData *op; - - g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == - g_input_stream_real_read_async); - - op = g_simple_async_result_get_op_res_gpointer (simple); + g_return_val_if_fail (g_task_is_valid (result, stream), -1); - return op->count_read; + return g_task_propagate_int (G_TASK (result), error); } -typedef struct { - gsize count_requested; - gssize count_skipped; -} SkipData; - static void -skip_async_thread (GSimpleAsyncResult *res, - GObject *object, - GCancellable *cancellable) +skip_async_thread (GTask *task, + gpointer source_object, + gpointer task_data, + GCancellable *cancellable) { - SkipData *op; + GInputStream *stream = source_object; + gsize count = GPOINTER_TO_SIZE (task_data); GInputStreamClass *class; GError *error = NULL; - - class = G_INPUT_STREAM_GET_CLASS (object); - op = g_simple_async_result_get_op_res_gpointer (res); - op->count_skipped = class->skip (G_INPUT_STREAM (object), - op->count_requested, - cancellable, &error); - if (op->count_skipped == -1) - g_simple_async_result_take_error (res, error); + gssize ret; + + class = G_INPUT_STREAM_GET_CLASS (stream); + ret = class->skip (stream, count, + g_task_get_cancellable (task), + &error); + if (ret == -1) + g_task_return_error (task, error); + else + g_task_return_int (task, ret); } typedef struct { char buffer[8192]; gsize count; gsize count_skipped; - int io_prio; - GCancellable *cancellable; - gpointer user_data; - GAsyncReadyCallback callback; } SkipFallbackAsyncData; static void @@ -1272,9 +1227,8 @@ skip_callback_wrapper (GObject *source_object, gpointer user_data) { GInputStreamClass *class; - SkipFallbackAsyncData *data = user_data; - SkipData *op; - GSimpleAsyncResult *simple; + GTask *task = user_data; + SkipFallbackAsyncData *data = g_task_get_task_data (task); GError *error = NULL; gssize ret; @@ -1288,35 +1242,28 @@ skip_callback_wrapper (GObject *source_object, if (data->count > 0) { class = G_INPUT_STREAM_GET_CLASS (source_object); - class->read_async (G_INPUT_STREAM (source_object), data->buffer, MIN (8192, data->count), data->io_prio, data->cancellable, - skip_callback_wrapper, data); + class->read_async (G_INPUT_STREAM (source_object), + data->buffer, MIN (8192, data->count), + g_task_get_priority (task), + g_task_get_cancellable (task), + skip_callback_wrapper, task); return; } } - op = g_new0 (SkipData, 1); - op->count_skipped = data->count_skipped; - simple = g_simple_async_result_new (source_object, - data->callback, data->user_data, - g_input_stream_real_skip_async); - - g_simple_async_result_set_op_res_gpointer (simple, op, g_free); - - if (ret == -1) + if (ret == -1 && + g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) && + data->count_skipped) { - if (data->count_skipped && - g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) - /* No error, return partial read */ - g_error_free (error); - else - g_simple_async_result_take_error (simple, error); + /* No error, return partial read */ + g_clear_error (&error); } - /* Complete immediately, not in idle, since we're already in a mainloop callout */ - g_simple_async_result_complete (simple); - g_object_unref (simple); - - g_free (data); + if (error) + g_task_return_error (task, error); + else + g_task_return_int (task, data->count_skipped); + g_object_unref (task); } static void @@ -1328,28 +1275,23 @@ g_input_stream_real_skip_async (GInputStream *stream, gpointer user_data) { GInputStreamClass *class; - SkipData *op; SkipFallbackAsyncData *data; - GSimpleAsyncResult *res; + GTask *task; class = G_INPUT_STREAM_GET_CLASS (stream); - if (class->read_async == g_input_stream_real_read_async) + task = g_task_new (stream, cancellable, callback, user_data); + g_task_set_priority (task, io_priority); + + if (g_input_stream_async_read_is_via_threads (stream)) { /* Read is thread-using async fallback. * Make skip use threads too, so that we can use a possible sync skip * implementation. */ - op = g_new0 (SkipData, 1); - - res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data, - g_input_stream_real_skip_async); - - g_simple_async_result_set_op_res_gpointer (res, op, g_free); + g_task_set_task_data (task, GSIZE_TO_POINTER (count), NULL); - op->count_requested = count; - - g_simple_async_result_run_in_thread (res, skip_async_thread, io_priority, cancellable); - g_object_unref (res); + g_task_run_in_thread (task, skip_async_thread); + g_object_unref (task); } else { @@ -1359,12 +1301,10 @@ g_input_stream_real_skip_async (GInputStream *stream, data = g_new (SkipFallbackAsyncData, 1); data->count = count; data->count_skipped = 0; - data->io_prio = io_priority; - data->cancellable = cancellable; - data->callback = callback; - data->user_data = user_data; + g_task_set_task_data (task, data, g_free); + g_task_set_check_cancellable (task, FALSE); class->read_async (stream, data->buffer, MIN (8192, count), io_priority, cancellable, - skip_callback_wrapper, data); + skip_callback_wrapper, task); } } @@ -1374,35 +1314,36 @@ g_input_stream_real_skip_finish (GInputStream *stream, GAsyncResult *result, GError **error) { - GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result); - SkipData *op; + g_return_val_if_fail (g_task_is_valid (result, stream), -1); - g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_skip_async); - op = g_simple_async_result_get_op_res_gpointer (simple); - return op->count_skipped; + return g_task_propagate_int (G_TASK (result), error); } static void -close_async_thread (GSimpleAsyncResult *res, - GObject *object, - GCancellable *cancellable) +close_async_thread (GTask *task, + gpointer source_object, + gpointer task_data, + GCancellable *cancellable) { + GInputStream *stream = source_object; GInputStreamClass *class; GError *error = NULL; gboolean result; - /* Auto handling of cancelation disabled, and ignore - cancellation, since we want to close things anyway, although - possibly in a quick-n-dirty way. At least we never want to leak - open handles */ - - class = G_INPUT_STREAM_GET_CLASS (object); + class = G_INPUT_STREAM_GET_CLASS (stream); if (class->close_fn) { - result = class->close_fn (G_INPUT_STREAM (object), cancellable, &error); + result = class->close_fn (stream, + g_task_get_cancellable (task), + &error); if (!result) - g_simple_async_result_take_error (res, error); + { + g_task_return_error (task, error); + return; + } } + + g_task_return_boolean (task, TRUE); } static void @@ -1412,20 +1353,14 @@ g_input_stream_real_close_async (GInputStream *stream, GAsyncReadyCallback callback, gpointer user_data) { - GSimpleAsyncResult *res; - - res = g_simple_async_result_new (G_OBJECT (stream), - callback, - user_data, - g_input_stream_real_close_async); + GTask *task; - g_simple_async_result_set_handle_cancellation (res, FALSE); + task = g_task_new (stream, cancellable, callback, user_data); + g_task_set_check_cancellable (task, FALSE); + g_task_set_priority (task, io_priority); - g_simple_async_result_run_in_thread (res, - close_async_thread, - io_priority, - cancellable); - g_object_unref (res); + g_task_run_in_thread (task, close_async_thread); + g_object_unref (task); } static gboolean @@ -1433,7 +1368,7 @@ g_input_stream_real_close_finish (GInputStream *stream, GAsyncResult *result, GError **error) { - GSimpleAsyncResult *simple = G_SIMPLE_ASYNC_RESULT (result); - g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_input_stream_real_close_async); - return TRUE; + g_return_val_if_fail (g_task_is_valid (result, stream), FALSE); + + return g_task_propagate_boolean (G_TASK (result), error); }