Allows to read async from a stream with GVirInputStream. This is modelled after GSocket. --- libvirt-gobject/Makefile.am | 2 + libvirt-gobject/libvirt-gobject-connection.c | 2 +- libvirt-gobject/libvirt-gobject-input-stream.c | 239 ++++++++++++++++++++++++ libvirt-gobject/libvirt-gobject-input-stream.h | 68 +++++++ libvirt-gobject/libvirt-gobject-stream.c | 130 +++++++++++++- libvirt-gobject/libvirt-gobject-stream.h | 10 +- 6 files changed, 443 insertions(+), 8 deletions(-) create mode 100644 libvirt-gobject/libvirt-gobject-input-stream.c create mode 100644 libvirt-gobject/libvirt-gobject-input-stream.h diff --git a/libvirt-gobject/Makefile.am b/libvirt-gobject/Makefile.am index 8147db2..7013675 100644 --- a/libvirt-gobject/Makefile.am +++ b/libvirt-gobject/Makefile.am @@ -40,6 +40,8 @@ libvirt_gobject_1_0_la_HEADERS = \ libvirt_gobject_1_0_la_SOURCES = \ $(libvirt_gobject_1_0_la_HEADERS) \ libvirt-gobject-enums.c \ + libvirt-gobject-input-stream.c \ + libvirt-gobject-input-stream.h \ $(GOBJECT_SOURCE_FILES) libvirt_gobject_1_0_la_CFLAGS = \ -DDATADIR="\"$(datadir)\"" \ diff --git a/libvirt-gobject/libvirt-gobject-connection.c b/libvirt-gobject/libvirt-gobject-connection.c index 5fc0a9e..95cd878 100644 --- a/libvirt-gobject/libvirt-gobject-connection.c +++ b/libvirt-gobject/libvirt-gobject-connection.c @@ -1151,7 +1151,7 @@ GVirStream *gvir_connection_get_stream(GVirConnection *self, klass = GVIR_CONNECTION_GET_CLASS(self); g_return_val_if_fail(klass->stream_new, NULL); - virStreamPtr st = virStreamNew(self->priv->conn, flags); + virStreamPtr st = virStreamNew(self->priv->conn, flags | VIR_STREAM_NONBLOCK); return klass->stream_new(self, st); } diff --git a/libvirt-gobject/libvirt-gobject-input-stream.c b/libvirt-gobject/libvirt-gobject-input-stream.c new file mode 100644 index 0000000..a76d670 --- /dev/null +++ b/libvirt-gobject/libvirt-gobject-input-stream.c @@ -0,0 +1,239 @@ +/* + * libvirt-gobject-input-stream.h: libvirt gobject integration + * + * Copyright (C) 2011 Red Hat + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * 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 + * + * Authors: Daniel P. Berrange <berrange(a)redhat.com&gt; + * Marc-André Lureau <marcandre.lureau(a)redhat.com&gt; + */ + +#include <config.h> + +#include <libvirt/virterror.h> +#include <string.h> + +#include "libvirt-glib/libvirt-glib.h" +#include "libvirt-gobject/libvirt-gobject.h" +#include "libvirt-gobject-input-stream.h" + +extern gboolean debugFlag; + +#define DEBUG(fmt, ...) do { if (G_UNLIKELY(debugFlag)) g_debug(fmt, ## __VA_ARGS__); } while (0) + +#define gvir_input_stream_get_type _gvir_input_stream_get_type +G_DEFINE_TYPE(GVirInputStream, gvir_input_stream, G_TYPE_INPUT_STREAM); + +enum +{ + PROP_0, + PROP_STREAM +}; + +struct _GVirInputStreamPrivate +{ + GVirStream *stream; + + /* pending operation metadata */ + GSimpleAsyncResult *result; + GCancellable *cancellable; + gpointer buffer; + gsize count; +}; + +static void gvir_input_stream_get_property(GObject *object, + guint prop_id, + GValue *value, + GParamSpec *pspec) +{ + GVirInputStream *stream = GVIR_INPUT_STREAM(object); + + switch (prop_id) { + case PROP_STREAM: + g_value_set_object(value, stream->priv->stream); + break; + + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec); + } +} + +static void gvir_input_stream_set_property(GObject *object, + guint prop_id, + const GValue *value, + GParamSpec *pspec) +{ + GVirInputStream *stream = GVIR_INPUT_STREAM(object); + + switch (prop_id) { + case PROP_STREAM: + stream->priv->stream = g_value_dup_object(value); + break; + + default: + G_OBJECT_WARN_INVALID_PROPERTY_ID(object, prop_id, pspec); + } +} + +static void gvir_input_stream_finalize(GObject *object) +{ + GVirInputStream *stream = GVIR_INPUT_STREAM(object); + + if (stream->priv->stream) + g_object_unref(stream->priv->stream); + + if (G_OBJECT_CLASS(gvir_input_stream_parent_class)->finalize) + (*G_OBJECT_CLASS(gvir_input_stream_parent_class)->finalize)(object); +} + +static void +gvir_input_stream_read_ready (G_GNUC_UNUSED virStreamPtr st, + int events, void *opaque) +{ + GVirInputStream *stream = GVIR_INPUT_STREAM(opaque); + GVirInputStreamPrivate *priv = stream->priv; + GSimpleAsyncResult *simple; + GError *error = NULL; + gssize result; + + g_return_if_fail(events & VIR_STREAM_EVENT_READABLE); + + result = gvir_stream_receive(priv->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; + } + + simple = stream->priv->result; + stream->priv->result = NULL; + + if (result >= 0) + g_simple_async_result_set_op_res_gssize(simple, result); + + if (error) + g_simple_async_result_take_error(simple, error); + + if (priv->cancellable) { + g_object_unref(stream->priv->cancellable); + priv->cancellable = NULL; + } + + g_simple_async_result_complete(simple); + g_object_unref(simple); + + return; +} + +static void gvir_input_stream_read_async(GInputStream *stream, + void *buffer, + gsize count, + G_GNUC_UNUSED int io_priority, + GCancellable *cancellable, + GAsyncReadyCallback callback, + 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; + } + + input_stream->priv->result = + g_simple_async_result_new(G_OBJECT(stream), callback, user_data, + gvir_input_stream_read_async); + if (cancellable) + g_object_ref(cancellable); + input_stream->priv->cancellable = cancellable; + input_stream->priv->buffer = buffer; + input_stream->priv->count = count; + +end: + virStreamFree(handle); +} + + +static gssize gvir_input_stream_read_finish(GInputStream *stream, + GAsyncResult *result, + G_GNUC_UNUSED GError **error) +{ + GVirInputStream *input_stream = GVIR_INPUT_STREAM(stream); + GSimpleAsyncResult *simple; + virStreamPtr handle; + gssize count; + + g_return_val_if_fail(GVIR_IS_INPUT_STREAM(stream), -1); + g_object_get(input_stream->priv->stream, "handle", &handle, NULL); + + simple = G_SIMPLE_ASYNC_RESULT(result); + + g_warn_if_fail(g_simple_async_result_get_source_tag(simple) == gvir_input_stream_read_async); + + count = g_simple_async_result_get_op_res_gssize(simple); + + virStreamEventRemoveCallback(handle); + virStreamFree(handle); + + return count; +} + + +static void gvir_input_stream_class_init(GVirInputStreamClass *klass) +{ + GObjectClass *gobject_class = G_OBJECT_CLASS(klass); + GInputStreamClass *ginputstream_class = G_INPUT_STREAM_CLASS(klass); + + g_type_class_add_private(klass, sizeof(GVirInputStreamPrivate)); + + gobject_class->finalize = gvir_input_stream_finalize; + gobject_class->get_property = gvir_input_stream_get_property; + gobject_class->set_property = gvir_input_stream_set_property; + + ginputstream_class->read_fn = NULL; + ginputstream_class->read_async = gvir_input_stream_read_async; + ginputstream_class->read_finish = gvir_input_stream_read_finish; + + g_object_class_install_property(gobject_class, PROP_STREAM, + g_param_spec_object("stream", + "stream", + "GVirStream", + GVIR_TYPE_STREAM, G_PARAM_CONSTRUCT_ONLY | + G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); +} + +static void gvir_input_stream_init(GVirInputStream *stream) +{ + stream->priv = G_TYPE_INSTANCE_GET_PRIVATE(stream, GVIR_TYPE_INPUT_STREAM, GVirInputStreamPrivate); +} + +GVirInputStream* _gvir_input_stream_new(GVirStream *stream) +{ + return GVIR_INPUT_STREAM(g_object_new(GVIR_TYPE_INPUT_STREAM, "stream", stream, NULL)); +} diff --git a/libvirt-gobject/libvirt-gobject-input-stream.h b/libvirt-gobject/libvirt-gobject-input-stream.h new file mode 100644 index 0000000..e8002b9 --- /dev/null +++ b/libvirt-gobject/libvirt-gobject-input-stream.h @@ -0,0 +1,68 @@ +/* + * libvirt-gobject-input-stream.h: libvirt gobject integration + * + * Copyright (C) 2011 Red Hat + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * 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 + * + * Authors: Daniel P. Berrange <berrange(a)redhat.com&gt; + * Marc-André Lureau <marcandre.lureau(a)redhat.com&gt; + */ + +#if !defined(__LIBVIRT_GOBJECT_H__) && !defined(LIBVIRT_GOBJECT_BUILD) +#error "Only <libvirt-gobject/libvirt-gobject.h> can be included directly." +#endif + +#ifndef __LIBVIRT_GOBJECT_INPUT_STREAM_H__ +#define __LIBVIRT_GOBJECT_INPUT_STREAM_H__ + +#include <gio/gio.h> +#include "libvirt-gobject-stream.h" + +G_BEGIN_DECLS + +#define GVIR_TYPE_INPUT_STREAM (_gvir_input_stream_get_type ()) +#define GVIR_INPUT_STREAM(inst) (G_TYPE_CHECK_INSTANCE_CAST ((inst), \ + GVIR_TYPE_INPUT_STREAM, GVirInputStream)) +#define GVIR_INPUT_STREAM_CLASS(class) (G_TYPE_CHECK_CLASS_CAST ((class), \ + GVIR_TYPE_INPUT_STREAM, GVirInputStreamClass)) +#define GVIR_IS_INPUT_STREAM(inst) (G_TYPE_CHECK_INSTANCE_TYPE ((inst), \ + GVIR_TYPE_INPUT_STREAM)) +#define GVIR_IS_INPUT_STREAM_CLASS(class) (G_TYPE_CHECK_CLASS_TYPE ((class), \ + GVIR_TYPE_INPUT_STREAM)) +#define GVIR_INPUT_STREAM_GET_CLASS(inst) (G_TYPE_INSTANCE_GET_CLASS ((inst), \ + GVIR_TYPE_INPUT_STREAM, GVirInputStreamClass)) + +typedef struct _GVirInputStreamPrivate GVirInputStreamPrivate; +typedef struct _GVirInputStreamClass GVirInputStreamClass; +typedef struct _GVirInputStream GVirInputStream; + +struct _GVirInputStreamClass +{ + GInputStreamClass parent_class; +}; + +struct _GVirInputStream +{ + GInputStream parent_instance; + GVirInputStreamPrivate *priv; +}; + +GType _gvir_input_stream_get_type (void) G_GNUC_CONST; +GVirInputStream * _gvir_input_stream_new (GVirStream *stream); + +G_END_DECLS + +#endif /* __LIBVIRT_GOBJECT_INPUT_STREAM_H__ */ diff --git a/libvirt-gobject/libvirt-gobject-stream.c b/libvirt-gobject/libvirt-gobject-stream.c index 519d733..88e3a40 100644 --- a/libvirt-gobject/libvirt-gobject-stream.c +++ b/libvirt-gobject/libvirt-gobject-stream.c @@ -30,6 +30,8 @@ #include "libvirt-glib/libvirt-glib.h" #include "libvirt-gobject/libvirt-gobject.h" +#include "libvirt-gobject/libvirt-gobject-input-stream.h" + extern gboolean debugFlag; #define DEBUG(fmt, ...) do { if (G_UNLIKELY(debugFlag)) g_debug(fmt, ## __VA_ARGS__); } while (0) @@ -39,10 +41,12 @@ extern gboolean debugFlag; struct _GVirStreamPrivate { - virStreamPtr handle; + virStreamPtr handle; + GInputStream *input_stream; + gboolean in_dispose; }; -G_DEFINE_TYPE(GVirStream, gvir_stream, G_TYPE_OBJECT); +G_DEFINE_TYPE(GVirStream, gvir_stream, G_TYPE_IO_STREAM); enum { @@ -60,6 +64,71 @@ gvir_stream_error_quark(void) return g_quark_from_static_string("vir-g-stream"); } + +static GInputStream* gvir_stream_get_input_stream(GIOStream *io_stream) +{ + GVirStream *self = GVIR_STREAM(io_stream); + + if (self->priv->input_stream == NULL) + self->priv->input_stream = (GInputStream *)_gvir_input_stream_new(self); + + return self->priv->input_stream; +} + + +static gboolean gvir_stream_close(GIOStream *io_stream, + GCancellable *cancellable, G_GNUC_UNUSED GError **error) +{ + GVirStream *self = GVIR_STREAM(io_stream); + + if (self->priv->input_stream) + g_input_stream_close(self->priv->input_stream, cancellable, NULL); + + if (self->priv->in_dispose) + return TRUE; + + return TRUE; /* FIXME: really close the stream? */ +} + + +static void gvir_stream_close_async(GIOStream *stream, G_GNUC_UNUSED int io_priority, + GCancellable *cancellable, GAsyncReadyCallback callback, + gpointer user_data) +{ + GSimpleAsyncResult *res; + GIOStreamClass *class; + GError *error; + + class = G_IO_STREAM_GET_CLASS(stream); + + /* close is not blocked, just do it! */ + error = NULL; + if (class->close_fn && + !class->close_fn(stream, cancellable, &error)) { + g_simple_async_report_take_gerror_in_idle(G_OBJECT (stream), + callback, user_data, + error); + return; + } + + res = g_simple_async_result_new(G_OBJECT (stream), + callback, + user_data, + gvir_stream_close_async); + g_simple_async_result_complete_in_idle(res); + g_object_unref (res); +} + + +static gboolean +gvir_stream_close_finish(G_GNUC_UNUSED GIOStream *stream, + G_GNUC_UNUSED GAsyncResult *result, + G_GNUC_UNUSED GError **error) +{ + return TRUE; +} + + static void gvir_stream_get_property(GObject *object, guint prop_id, GValue *value, @@ -107,6 +176,9 @@ static void gvir_stream_finalize(GObject *object) DEBUG("Finalize GVirStream=%p", self); + if (self->priv->input_stream) + g_object_unref(self->priv->input_stream); + if (priv->handle) { if (virStreamFinish(priv->handle) < 0) g_critical("cannot finish stream"); @@ -120,12 +192,18 @@ static void gvir_stream_finalize(GObject *object) static void gvir_stream_class_init(GVirStreamClass *klass) { - GObjectClass *object_class = G_OBJECT_CLASS (klass); + GObjectClass *object_class = G_OBJECT_CLASS(klass); + GIOStreamClass *stream_class = G_IO_STREAM_CLASS(klass); object_class->finalize = gvir_stream_finalize; object_class->get_property = gvir_stream_get_property; object_class->set_property = gvir_stream_set_property; + stream_class->get_input_stream = gvir_stream_get_input_stream; + stream_class->close_fn = gvir_stream_close; + stream_class->close_async = gvir_stream_close_async; + stream_class->close_finish = gvir_stream_close_finish; + g_object_class_install_property(object_class, PROP_HANDLE, g_param_spec_boxed("handle", @@ -170,6 +248,50 @@ GType gvir_stream_handle_get_type(void) return handle_type; } +/** + * gvir_stream_receive: + * @stream: the stream + * @buffer: a buffer to read data into (which should be at least @size + * bytes long). + * @size: the number of bytes you want to read from the stream + * @cancellable: (allow-none): a %GCancellable or %NULL + * @error: #GError for error reporting, or %NULL to ignore. + * + * Receive data (up to @size bytes) from a stream. + * On error -1 is returned and @error is set accordingly. + * + * gvir_stream_receive() can return any number of bytes, up to + * @size. If more than @size bytes have been received, the additional + * data will be returned in future calls to gvir_stream_receive(). + * + * If there is no data available, a %G_IO_ERROR_WOULD_BLOCK error will be + * returned. + * + * Returns: Number of bytes read, or 0 if the end of stream reached, + * or -1 on error. + */ +gssize gvir_stream_receive(GVirStream *self, gchar *buffer, gsize size, + GCancellable *cancellable, GError **error) +{ + int got; + + g_return_val_if_fail(GVIR_IS_STREAM(self), -1); + g_return_val_if_fail(buffer != NULL, -1); + + if (g_cancellable_set_error_if_cancelled (cancellable, error)) + return -1; + + got = virStreamRecv(self->priv->handle, buffer, size); + + if (got == -2) { /* blocking */ + g_set_error(error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK, NULL); + } else if (got < 0) { + g_set_error(error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT, + "Got virStreamRecv error in %s", G_STRFUNC); + } + + return got; +} struct stream_sink_helper { GVirStream *self; @@ -197,7 +319,7 @@ stream_sink(virStreamPtr st G_GNUC_UNUSED, * requested data sink. This is simply a convenient alternative * to virStreamRecv, for apps that do blocking-I/o. */ -gint +gssize gvir_stream_receive_all(GVirStream *self, GVirStreamSinkFunc func, gpointer user_data, GError **err) { struct stream_sink_helper helper = { diff --git a/libvirt-gobject/libvirt-gobject-stream.h b/libvirt-gobject/libvirt-gobject-stream.h index 5181e24..35526db 100644 --- a/libvirt-gobject/libvirt-gobject-stream.h +++ b/libvirt-gobject/libvirt-gobject-stream.h @@ -28,6 +28,9 @@ #ifndef __LIBVIRT_GOBJECT_STREAM_H__ #define __LIBVIRT_GOBJECT_STREAM_H__ +#include <glib-object.h> +#include <gio/gio.h> + G_BEGIN_DECLS #define GVIR_TYPE_STREAM (gvir_stream_get_type ()) @@ -45,7 +48,7 @@ typedef struct _GVirStreamClass GVirStreamClass; struct _GVirStream { - GObject parent; + GIOStream parent_instance; GVirStreamPrivate *priv; @@ -54,7 +57,7 @@ struct _GVirStream struct _GVirStreamClass { - GObjectClass parent_class; + GIOStreamClass parent_class; gpointer padding[20]; }; @@ -76,7 +79,8 @@ typedef gint (* GVirStreamSinkFunc) (GVirStream *stream, GType gvir_stream_get_type(void); GType gvir_stream_handle_get_type(void); -gint gvir_stream_receive_all(GVirStream *stream, GVirStreamSinkFunc func, gpointer user_data, GError **err); +gssize gvir_stream_receive_all(GVirStream *stream, GVirStreamSinkFunc func, gpointer user_data, GError **err); +gssize gvir_stream_receive(GVirStream *stream, gchar *buffer, gsize size, GCancellable *cancellable, GError **error); G_END_DECLS -- 1.7.6.2