[libvirt] [PATCH libvirt-glib] Allow cancelling gvir_stream_send_all & gvir_stream_receive_all

[Daniel P. Berrange]

From: "Daniel P. Berrange" <berrange at redhat.com>

Add GCancellable parameters to gvir_stream_send_all and
gvir_stream_receive_all to allow I/O to be interrupted
before completion
---
 libvirt-gobject/libvirt-gobject-stream.c |   23 +++++++++++++++++++++--
 libvirt-gobject/libvirt-gobject-stream.h |    2 ++
 2 files changed, 23 insertions(+), 2 deletions(-)

diff --git a/libvirt-gobject/libvirt-gobject-stream.c b/libvirt-gobject/libvirt-gobject-stream.c
index c1ae765..7204f98 100644
--- a/libvirt-gobject/libvirt-gobject-stream.c
+++ b/libvirt-gobject/libvirt-gobject-stream.c
@@ -26,6 +26,7 @@
 
 #include <libvirt/virterror.h>
 #include <string.h>
+#include <errno.h>
 
 #include "libvirt-glib/libvirt-glib.h"
 #include "libvirt-gobject/libvirt-gobject.h"
@@ -330,6 +331,7 @@ struct stream_sink_helper {
     GVirStream *self;
     GVirStreamSinkFunc func;
     gpointer user_data;
+    GCancellable *cancellable;
 };
 
 static int
@@ -340,12 +342,18 @@ stream_sink(virStreamPtr st G_GNUC_UNUSED,
 {
   struct stream_sink_helper *helper = opaque;
 
+  if (g_cancellable_is_cancelled(helper->cancellable)) {
+      errno = EIO;
+      return -1;
+  }
+
   return helper->func(helper->self, bytes, nbytes, helper->user_data);
 }
 
 /**
  * gvir_stream_receive_all:
  * @stream: the stream
+ * @cancellable: cancellation notifier
  * @func: (scope notified): the callback for writing data to application
  * @user_data: (closure): data to be passed to @callback
  * Returns: the number of bytes consumed or -1 upon error
@@ -356,6 +364,7 @@ stream_sink(virStreamPtr st G_GNUC_UNUSED,
  */
 gssize
 gvir_stream_receive_all(GVirStream *self,
+                        GCancellable *cancellable,
                         GVirStreamSinkFunc func,
                         gpointer user_data,
                         GError **err)
@@ -363,7 +372,8 @@ gvir_stream_receive_all(GVirStream *self,
     struct stream_sink_helper helper = {
         .self = self,
         .func = func,
-        .user_data = user_data
+        .user_data = user_data,
+        .cancellable = cancellable,
     };
     int r;
 
@@ -433,6 +443,7 @@ struct stream_source_helper {
     GVirStream *self;
     GVirStreamSourceFunc func;
     gpointer user_data;
+    GCancellable *cancellable;
 };
 
 static int
@@ -443,12 +454,18 @@ stream_source(virStreamPtr st G_GNUC_UNUSED,
 {
   struct stream_source_helper *helper = opaque;
 
+  if (g_cancellable_is_cancelled(helper->cancellable)) {
+      errno = EIO;
+      return -1;
+  }
+
   return helper->func(helper->self, bytes, nbytes, helper->user_data);
 }
 
 /**
  * gvir_stream_send_all:
  * @stream: the stream
+ * @cancellable: cancellation notifier
  * @func: (scope notified): the callback for writing data to application
  * @user_data: (closure): data to be passed to @callback
  * Returns: the number of bytes consumed or -1 upon error
@@ -459,6 +476,7 @@ stream_source(virStreamPtr st G_GNUC_UNUSED,
  */
 gssize
 gvir_stream_send_all(GVirStream *self,
+                     GCancellable *cancellable,
                      GVirStreamSourceFunc func,
                      gpointer user_data,
                      GError **err)
@@ -466,7 +484,8 @@ gvir_stream_send_all(GVirStream *self,
     struct stream_source_helper helper = {
         .self = self,
         .func = func,
-        .user_data = user_data
+        .user_data = user_data,
+        .cancellable = cancellable,
     };
     int r;
 
diff --git a/libvirt-gobject/libvirt-gobject-stream.h b/libvirt-gobject/libvirt-gobject-stream.h
index 3caadc3..9a02c7a 100644
--- a/libvirt-gobject/libvirt-gobject-stream.h
+++ b/libvirt-gobject/libvirt-gobject-stream.h
@@ -115,6 +115,7 @@ guint gvir_stream_add_watch_full(GVirStream *stream,
                                  GDestroyNotify notify);
 
 gssize gvir_stream_receive_all(GVirStream *stream,
+                               GCancellable *cancellable,
                                GVirStreamSinkFunc func,
                                gpointer user_data,
                                GError **error);
@@ -125,6 +126,7 @@ gssize gvir_stream_receive(GVirStream *stream,
                            GError **error);
 
 gssize gvir_stream_send_all(GVirStream *stream,
+                            GCancellable *cancellable,
                             GVirStreamSourceFunc func,
                             gpointer user_data,
                             GError **error);
-- 
1.7.7.3