[libvirt] [PATCH libvirt-glib 5/5] Convert GVirInputStream/GVirOutputStream over to use watches

From: "Daniel P. Berrange" <berrange(a)redhat.com&gt; Since virStreamEventAddCallback only allows a single callback to be registered, GVirInputStream/GVirOutputStream should not use that directly. Instead they should use the gvir_stream_add_watch function * libvirt-gobject/libvirt-gobject-input-stream.c, libvirt-gobject/libvirt-gobject-output-stream.c: Convert to use gvir_stream_add_watch --- libvirt-gobject/libvirt-gobject-input-stream.c | 62 +++++++++++----------- libvirt-gobject/libvirt-gobject-output-stream.c | 62 +++++++++++----------- 2 files changed, 62 insertions(+), 62 deletions(-) diff --git a/libvirt-gobject/libvirt-gobject-input-stream.c b/libvirt-gobject/libvirt-gobject-input-stream.c index 06ea716..04ecb18 100644 --- a/libvirt-gobject/libvirt-gobject-input-stream.c +++ b/libvirt-gobject/libvirt-gobject-input-stream.c @@ -99,30 +99,40 @@ static void gvir_input_stream_finalize(GObject *object) (*G_OBJECT_CLASS(gvir_input_stream_parent_class)->finalize)(object); } -static void -gvir_input_stream_read_ready(virStreamPtr st G_GNUC_UNUSED, - int events, +static gboolean +gvir_input_stream_read_ready(GVirStream *stream G_GNUC_UNUSED, + GVirStreamIOCondition cond, void *opaque) { - GVirInputStream *stream = GVIR_INPUT_STREAM(opaque); - GVirInputStreamPrivate *priv = stream->priv; - GSimpleAsyncResult *simple; + GVirInputStream *input_stream = GVIR_INPUT_STREAM(opaque); + GVirInputStreamPrivate *priv = input_stream->priv; + GSimpleAsyncResult *simple = priv->result; GError *error = NULL; gssize result; - g_return_if_fail(events & VIR_STREAM_EVENT_READABLE); + if (!(cond & GVIR_STREAM_IO_CONDITION_READABLE)) { + g_warn_if_reached(); + g_simple_async_result_set_error(simple, + G_IO_ERROR, + G_IO_ERROR_INVALID_ARGUMENT, + "%s", + "Expected stream to be readable"); + goto cleanup; + } - result = gvir_stream_receive(priv->stream, priv->buffer, priv->count, + result = gvir_stream_receive(stream, priv->buffer, priv->count, priv->cancellable, &error); if (g_error_matches(error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) { g_warn_if_reached(); - return; + g_simple_async_result_set_error(simple, + G_IO_ERROR, + G_IO_ERROR_INVALID_ARGUMENT, + "%s", + "Expected stream to be readable"); + goto cleanup; } - simple = stream->priv->result; - stream->priv->result = NULL; - if (result >= 0) g_simple_async_result_set_op_res_gssize(simple, result); @@ -130,14 +140,15 @@ gvir_input_stream_read_ready(virStreamPtr st G_GNUC_UNUSED, g_simple_async_result_take_error(simple, error); if (priv->cancellable) { - g_object_unref(stream->priv->cancellable); + g_object_unref(priv->cancellable); priv->cancellable = NULL; } +cleanup: + priv->result = NULL; g_simple_async_result_complete(simple); g_object_unref(simple); - - return; + return FALSE; } static void gvir_input_stream_read_async(GInputStream *stream, @@ -149,23 +160,15 @@ static void gvir_input_stream_read_async(GInputStream *stream, gpointer user_data) { GVirInputStream *input_stream = GVIR_INPUT_STREAM(stream); - virStreamPtr handle; g_return_if_fail(GVIR_IS_INPUT_STREAM(stream)); g_return_if_fail(input_stream->priv->result == NULL); - g_object_get(input_stream->priv->stream, "handle", &handle, NULL); - - if (virStreamEventAddCallback(handle, VIR_STREAM_EVENT_READABLE, - gvir_input_stream_read_ready, stream, NULL) < 0) { - g_simple_async_report_error_in_idle(G_OBJECT(stream), - callback, - user_data, - G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT, - "Couldn't add event callback %s", - G_STRFUNC); - goto end; - } + gvir_stream_add_watch(input_stream->priv->stream, + GVIR_STREAM_IO_CONDITION_READABLE, + gvir_input_stream_read_ready, + g_object_ref(stream), + (GDestroyNotify)g_object_unref); input_stream->priv->result = g_simple_async_result_new(G_OBJECT(stream), callback, user_data, @@ -175,9 +178,6 @@ static void gvir_input_stream_read_async(GInputStream *stream, input_stream->priv->cancellable = cancellable; input_stream->priv->buffer = buffer; input_stream->priv->count = count; - -end: - virStreamFree(handle); } diff --git a/libvirt-gobject/libvirt-gobject-output-stream.c b/libvirt-gobject/libvirt-gobject-output-stream.c index 74eddc5..39beb46 100644 --- a/libvirt-gobject/libvirt-gobject-output-stream.c +++ b/libvirt-gobject/libvirt-gobject-output-stream.c @@ -99,30 +99,40 @@ static void gvir_output_stream_finalize(GObject *object) (*G_OBJECT_CLASS(gvir_output_stream_parent_class)->finalize)(object); } -static void -gvir_output_stream_write_ready(virStreamPtr st G_GNUC_UNUSED, - int events, +static gboolean +gvir_output_stream_write_ready(GVirStream *stream, + GVirStreamIOCondition cond, void *opaque) { - GVirOutputStream *stream = GVIR_OUTPUT_STREAM(opaque); - GVirOutputStreamPrivate *priv = stream->priv; - GSimpleAsyncResult *simple; + GVirOutputStream *output_stream = GVIR_OUTPUT_STREAM(opaque); + GVirOutputStreamPrivate *priv = output_stream->priv; + GSimpleAsyncResult *simple = priv->result; GError *error = NULL; gssize result; - g_return_if_fail(events & VIR_STREAM_EVENT_WRITABLE); + if (!(cond & GVIR_STREAM_IO_CONDITION_WRITABLE)) { + g_warn_if_reached(); + g_simple_async_result_set_error(simple, + G_IO_ERROR, + G_IO_ERROR_INVALID_ARGUMENT, + "%s", + "Expected stream to be writable"); + goto cleanup; + } - result = gvir_stream_send(priv->stream, priv->buffer, priv->count, + result = gvir_stream_send(stream, priv->buffer, priv->count, priv->cancellable, &error); if (g_error_matches(error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) { g_warn_if_reached(); - return; + g_simple_async_result_set_error(simple, + G_IO_ERROR, + G_IO_ERROR_INVALID_ARGUMENT, + "%s", + "Expected stream to be writable"); + goto cleanup; } - simple = stream->priv->result; - stream->priv->result = NULL; - if (result >= 0) g_simple_async_result_set_op_res_gssize(simple, result); @@ -130,14 +140,15 @@ gvir_output_stream_write_ready(virStreamPtr st G_GNUC_UNUSED, g_simple_async_result_take_error(simple, error); if (priv->cancellable) { - g_object_unref(stream->priv->cancellable); + g_object_unref(priv->cancellable); priv->cancellable = NULL; } +cleanup: + priv->result = NULL; g_simple_async_result_complete(simple); g_object_unref(simple); - - return; + return FALSE; } static void gvir_output_stream_write_async(GOutputStream *stream, @@ -149,23 +160,15 @@ static void gvir_output_stream_write_async(GOutputStream *stream, gpointer user_data) { GVirOutputStream *output_stream = GVIR_OUTPUT_STREAM(stream); - virStreamPtr handle; g_return_if_fail(GVIR_IS_OUTPUT_STREAM(stream)); g_return_if_fail(output_stream->priv->result == NULL); - g_object_get(output_stream->priv->stream, "handle", &handle, NULL); - - if (virStreamEventAddCallback(handle, VIR_STREAM_EVENT_WRITABLE, - gvir_output_stream_write_ready, stream, NULL) < 0) { - g_simple_async_report_error_in_idle(G_OBJECT(stream), - callback, - user_data, - G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT, - "Couldn't add event callback %s", - G_STRFUNC); - goto end; - } + gvir_stream_add_watch(output_stream->priv->stream, + GVIR_STREAM_IO_CONDITION_WRITABLE, + gvir_output_stream_write_ready, + g_object_ref(stream), + (GDestroyNotify)g_object_unref); output_stream->priv->result = g_simple_async_result_new(G_OBJECT(stream), callback, user_data, @@ -175,9 +178,6 @@ static void gvir_output_stream_write_async(GOutputStream *stream, output_stream->priv->cancellable = cancellable; output_stream->priv->buffer = buffer; output_stream->priv->count = count; - -end: - virStreamFree(handle); } -- 1.7.6.4