[libvirt] [PATCH v2 24/38] daemon: Implement VIR_NET_STREAM_SKIP handling
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(a)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