[libvirt] [PATCH v2 24/38] daemon: Implement VIR_NET_STREAM_SKIP handling

[Michal Privoznik]

Basically, whenever the new type of stream packet arrives to the
daemon call this function that decodes it and calls
virStreamSkipCallback().  Otherwise a regular data stream packet
has arrived and therefore continue its processing.

Signed-off-by: Michal Privoznik <mprivozn at redhat.com>
---
 daemon/stream.c | 81 +++++++++++++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 70 insertions(+), 11 deletions(-)

diff --git a/daemon/stream.c b/daemon/stream.c
index 6899c18..84709da 100644
--- a/daemon/stream.c
+++ b/daemon/stream.c
@@ -29,6 +29,7 @@
 #include "virlog.h"
 #include "virnetserverclient.h"
 #include "virerror.h"
+#include "libvirt_internal.h"
 
 #define VIR_FROM_THIS VIR_FROM_STREAMS
 
@@ -653,6 +654,52 @@ daemonStreamHandleAbort(virNetServerClientPtr client,
 }
 
 
+static int
+daemonStreamHandleSkip(virNetServerClientPtr client,
+                       daemonClientStream *stream,
+                       virNetMessagePtr msg)
+{
+    int ret;
+    virNetStreamSkip data;
+
+    VIR_DEBUG("client=%p, stream=%p, proc=%d, serial=%u",
+              client, stream, msg->header.proc, msg->header.serial);
+
+    /* Let's check if client plays nicely and advertised usage of
+     * sparse stream upfront. */
+    if (!stream->skippable) {
+        virReportError(VIR_ERR_RPC, "%s",
+                       _("Unexpected stream skip"));
+        return -1;
+    }
+
+    if (virNetMessageDecodePayload(msg,
+                                   (xdrproc_t) xdr_virNetStreamSkip,
+                                   &data) < 0)
+        return -1;
+
+    ret = virStreamSkip(stream->st, data.length);
+
+    if (ret < 0) {
+        virNetMessageError rerr;
+
+        memset(&rerr, 0, sizeof(rerr));
+
+        VIR_INFO("Stream skip failed");
+        stream->closed = true;
+        virStreamEventRemoveCallback(stream->st);
+        virStreamAbort(stream->st);
+
+        return virNetServerProgramSendReplyError(stream->prog,
+                                                 client,
+                                                 msg,
+                                                 &rerr,
+                                                 &msg->header);
+    }
+
+    return 0;
+}
+
 
 /*
  * Called when the stream is signalled has being able to accept
@@ -671,19 +718,31 @@ daemonStreamHandleWrite(virNetServerClientPtr client,
         virNetMessagePtr msg = stream->rx;
         int ret;
 
-        switch (msg->header.status) {
-        case VIR_NET_OK:
-            ret = daemonStreamHandleFinish(client, stream, msg);
-            break;
+        if (msg->header.type == VIR_NET_STREAM_SKIP) {
+            /* Handle special case when client sent us skip.
+             * Otherwise just carry on with processing stream
+             * data. */
+            ret = daemonStreamHandleSkip(client, stream, msg);
+        } else if (msg->header.type == VIR_NET_STREAM) {
+            switch (msg->header.status) {
+            case VIR_NET_OK:
+                ret = daemonStreamHandleFinish(client, stream, msg);
+                break;
 
-        case VIR_NET_CONTINUE:
-            ret = daemonStreamHandleWriteData(client, stream, msg);
-            break;
+            case VIR_NET_CONTINUE:
+                ret = daemonStreamHandleWriteData(client, stream, msg);
+                break;
 
-        case VIR_NET_ERROR:
-        default:
-            ret = daemonStreamHandleAbort(client, stream, msg);
-            break;
+            case VIR_NET_ERROR:
+            default:
+                ret = daemonStreamHandleAbort(client, stream, msg);
+                break;
+            }
+        } else {
+            virReportError(VIR_ERR_RPC,
+                           _("Unexpected message type: %d"),
+                           msg->header.type);
+            ret = -1;
         }
 
         if (ret > 0)
-- 
2.10.2