Introduce generic RPC client objects

To facilitate creation of new clients using XDR RPC services,
pull alot of the remote driver code into a set of reusable
objects.

 - virNetClient: Encapsulates a socket connection to a
   remote RPC server. Handles all the network I/O for
   reading/writing RPC messages. Delegates RPC encoding
   and decoding to the registered programs

 - virNetClientProgram: Handles processing and dispatch
   of RPC messages for a single RPC (program,version).
   A program can register to receive async events
   from a client

 - virNetClientStream: Handles generic I/O stream
   integration to RPC layer

Each new client program now merely needs to define the list of
RPC procedures & events it wants and their handlers. It does
not need to deal with any of the network I/O functionality at
all.

cfg.mk

@@ -126,6 +126,9 @@ useless_free_options =				\
   --name=virJSONValueFree			\
   --name=virLastErrFreeData			\
   --name=virNetMessageFree                      \
+  --name=virNetClientFree                       \
+  --name=virNetClientProgramFree                \
+  --name=virNetClientStreamFree                 \
   --name=virNetServerFree                       \
   --name=virNetServerClientFree                 \
   --name=virNetServerMDNSFree                   \

po/POTFILES.in

@@ -69,6 +69,9 @@ src/qemu/qemu_monitor_text.c
 src/qemu/qemu_process.c
 src/remote/remote_client_bodies.h
 src/remote/remote_driver.c
+src/rpc/virnetclient.c
+src/rpc/virnetclientprogram.c
+src/rpc/virnetclientstream.c
 src/rpc/virnetmessage.c
 src/rpc/virnetsaslcontext.c
 src/rpc/virnetsocket.c

src/Makefile.am

@@ -1188,7 +1188,7 @@ else
 EXTRA_DIST += $(LOCK_DRIVER_SANLOCK_SOURCES)
 endif
 
-noinst_LTLIBRARIES += libvirt-net-rpc.la libvirt-net-rpc-server.la
+noinst_LTLIBRARIES += libvirt-net-rpc.la libvirt-net-rpc-server.la libvirt-net-rpc-client.la
 
 libvirt_net_rpc_la_SOURCES = \
 	rpc/virnetmessage.h rpc/virnetmessage.c \
@@ -1238,6 +1238,18 @@ libvirt_net_rpc_server_la_LDFLAGS = \
 libvirt_net_rpc_server_la_LIBADD = \
 			$(CYGWIN_EXTRA_LIBADD)
 
+libvirt_net_rpc_client_la_SOURCES = \
+	rpc/virnetclientprogram.h rpc/virnetclientprogram.c \
+	rpc/virnetclientstream.h rpc/virnetclientstream.c \
+	rpc/virnetclient.h rpc/virnetclient.c
+libvirt_net_rpc_client_la_CFLAGS = \
+			$(AM_CFLAGS)
+libvirt_net_rpc_client_la_LDFLAGS = \
+			$(AM_LDFLAGS) \
+			$(CYGWIN_EXTRA_LDFLAGS) \
+			$(MINGW_EXTRA_LDFLAGS)
+libvirt_net_rpc_client_la_LIBADD = \
+			$(CYGWIN_EXTRA_LIBADD)
 
 libexec_PROGRAMS =
 

src/rpc/virnetclient.h

@@ -0,0 +1,84 @@
+/*
+ * virnetclient.h: generic network RPC client
+ *
+ * Copyright (C) 2006-2011 Red Hat, Inc.
+ *
+ * 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
+ *
+ * Author: Daniel P. Berrange <berrange@redhat.com>
+ */
+
+#ifndef __VIR_NET_CLIENT_H__
+# define __VIR_NET_CLIENT_H__
+
+# include "virnettlscontext.h"
+# include "virnetmessage.h"
+# ifdef HAVE_SASL
+#  include "virnetsaslcontext.h"
+# endif
+# include "virnetclientprogram.h"
+# include "virnetclientstream.h"
+
+
+virNetClientPtr virNetClientNewUNIX(const char *path,
+                                    bool spawnDaemon,
+                                    const char *daemon);
+
+virNetClientPtr virNetClientNewTCP(const char *nodename,
+                                   const char *service);
+
+virNetClientPtr virNetClientNewSSH(const char *nodename,
+                                   const char *service,
+                                   const char *binary,
+                                   const char *username,
+                                   bool noTTY,
+                                   const char *netcat,
+                                   const char *path);
+
+virNetClientPtr virNetClientNewExternal(const char **cmdargv);
+
+void virNetClientRef(virNetClientPtr client);
+
+int virNetClientAddProgram(virNetClientPtr client,
+                           virNetClientProgramPtr prog);
+
+int virNetClientAddStream(virNetClientPtr client,
+                          virNetClientStreamPtr st);
+
+void virNetClientRemoveStream(virNetClientPtr client,
+                              virNetClientStreamPtr st);
+
+int virNetClientSend(virNetClientPtr client,
+                     virNetMessagePtr msg,
+                     bool expectReply);
+
+# ifdef HAVE_SASL
+void virNetClientSetSASLSession(virNetClientPtr client,
+                                virNetSASLSessionPtr sasl);
+# endif
+
+int virNetClientSetTLSSession(virNetClientPtr client,
+                              virNetTLSContextPtr tls);
+
+bool virNetClientIsEncrypted(virNetClientPtr client);
+
+const char *virNetClientLocalAddrString(virNetClientPtr client);
+const char *virNetClientRemoteAddrString(virNetClientPtr client);
+
+int virNetClientGetTLSKeySize(virNetClientPtr client);
+
+void virNetClientFree(virNetClientPtr client);
+
+#endif /* __VIR_NET_CLIENT_H__ */

src/rpc/virnetclientprogram.c

@@ -0,0 +1,339 @@
+/*
+ * virnetclientprogram.c: generic network RPC client program
+ *
+ * Copyright (C) 2006-2011 Red Hat, Inc.
+ *
+ * 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
+ *
+ * Author: Daniel P. Berrange <berrange@redhat.com>
+ */
+
+#include <config.h>
+
+#include "virnetclientprogram.h"
+#include "virnetclient.h"
+#include "virnetprotocol.h"
+
+#include "memory.h"
+#include "virterror_internal.h"
+#include "logging.h"
+
+#define VIR_FROM_THIS VIR_FROM_RPC
+#define virNetError(code, ...)                                    \
+    virReportErrorHelper(VIR_FROM_THIS, code, __FILE__,           \
+                         __FUNCTION__, __LINE__, __VA_ARGS__)
+
+struct _virNetClientProgram {
+    int refs;
+
+    unsigned program;
+    unsigned version;
+    virNetClientProgramEventPtr events;
+    size_t nevents;
+    void *eventOpaque;
+};
+
+virNetClientProgramPtr virNetClientProgramNew(unsigned program,
+                                              unsigned version,
+                                              virNetClientProgramEventPtr events,
+                                              size_t nevents,
+                                              void *eventOpaque)
+{
+    virNetClientProgramPtr prog;
+
+    if (VIR_ALLOC(prog) < 0) {
+        virReportOOMError();
+        return NULL;
+    }
+
+    prog->refs = 1;
+    prog->program = program;
+    prog->version = version;
+    prog->events = events;
+    prog->nevents = nevents;
+    prog->eventOpaque = eventOpaque;
+
+    return prog;
+}
+
+
+void virNetClientProgramRef(virNetClientProgramPtr prog)
+{
+    prog->refs++;
+}
+
+
+void virNetClientProgramFree(virNetClientProgramPtr prog)
+{
+    if (!prog)
+        return;
+
+    prog->refs--;
+    if (prog->refs > 0)
+        return;
+
+    VIR_FREE(prog);
+}
+
+
+unsigned virNetClientProgramGetProgram(virNetClientProgramPtr prog)
+{
+    return prog->program;
+}
+
+
+unsigned virNetClientProgramGetVersion(virNetClientProgramPtr prog)
+{
+    return prog->version;
+}
+
+
+int virNetClientProgramMatches(virNetClientProgramPtr prog,
+                               virNetMessagePtr msg)
+{
+    if (prog->program == msg->header.prog &&
+        prog->version == msg->header.vers)
+        return 1;
+    return 0;
+}
+
+
+static int
+virNetClientProgramDispatchError(virNetClientProgramPtr prog ATTRIBUTE_UNUSED,
+                                 virNetMessagePtr msg)
+{
+    virNetMessageError err;
+    int ret = -1;
+
+    memset(&err, 0, sizeof(err));
+
+    if (virNetMessageDecodePayload(msg, (xdrproc_t)xdr_virNetMessageError, &err) < 0)
+        goto cleanup;
+
+    /* Interop for virErrorNumber glitch in 0.8.0, if server is
+     * 0.7.1 through 0.7.7; see comments in virterror.h. */
+    switch (err.code) {
+    case VIR_WAR_NO_NWFILTER:
+        /* no way to tell old VIR_WAR_NO_SECRET apart from
+         * VIR_WAR_NO_NWFILTER, but both are very similar
+         * warnings, so ignore the difference */
+        break;
+    case VIR_ERR_INVALID_NWFILTER:
+    case VIR_ERR_NO_NWFILTER:
+    case VIR_ERR_BUILD_FIREWALL:
+        /* server was trying to pass VIR_ERR_INVALID_SECRET,
+         * VIR_ERR_NO_SECRET, or VIR_ERR_CONFIG_UNSUPPORTED */
+        if (err.domain != VIR_FROM_NWFILTER)
+            err.code += 4;
+        break;
+    case VIR_WAR_NO_SECRET:
+        if (err.domain == VIR_FROM_QEMU)
+            err.code = VIR_ERR_OPERATION_TIMEOUT;
+        break;
+    case VIR_ERR_INVALID_SECRET:
+        if (err.domain == VIR_FROM_XEN)
+            err.code = VIR_ERR_MIGRATE_PERSIST_FAILED;
+        break;
+    default:
+        /* Nothing to alter. */
+        break;
+    }
+
+    if (err.domain == VIR_FROM_REMOTE &&
+        err.code == VIR_ERR_RPC &&
+        err.level == VIR_ERR_ERROR &&
+        err.message &&
+        STRPREFIX(*err.message, "unknown procedure")) {
+        virRaiseErrorFull(__FILE__, __FUNCTION__, __LINE__,
+                          err.domain,
+                          VIR_ERR_NO_SUPPORT,
+                          err.level,
+                          err.str1 ? *err.str1 : NULL,
+                          err.str2 ? *err.str2 : NULL,
+                          err.str3 ? *err.str3 : NULL,
+                          err.int1,
+                          err.int2,
+                          "%s", *err.message);
+    } else {
+        virRaiseErrorFull(__FILE__, __FUNCTION__, __LINE__,
+                          err.domain,
+                          err.code,
+                          err.level,
+                          err.str1 ? *err.str1 : NULL,
+                          err.str2 ? *err.str2 : NULL,
+                          err.str3 ? *err.str3 : NULL,
+                          err.int1,
+                          err.int2,
+                          "%s", err.message ? *err.message : _("Unknown error"));
+    }
+
+    ret = 0;
+
+cleanup:
+    xdr_free((xdrproc_t)xdr_virNetMessageError, (void*)&err);
+    return ret;
+}
+
+
+static virNetClientProgramEventPtr virNetClientProgramGetEvent(virNetClientProgramPtr prog,
+                                                               int procedure)
+{
+    int i;
+
+    for (i = 0 ; i < prog->nevents ; i++) {
+        if (prog->events[i].proc == procedure)
+            return &prog->events[i];
+    }
+
+    return NULL;
+}
+
+
+int virNetClientProgramDispatch(virNetClientProgramPtr prog,
+                                virNetClientPtr client,
+                                virNetMessagePtr msg)
+{
+    virNetClientProgramEventPtr event;
+    char *evdata;
+
+    VIR_DEBUG("prog=%d ver=%d type=%d status=%d serial=%d proc=%d",
+              msg->header.prog, msg->header.vers, msg->header.type,
+              msg->header.status, msg->header.serial, msg->header.proc);
+
+    /* Check version, etc. */
+    if (msg->header.prog != prog->program) {
+        VIR_ERROR(_("program mismatch in event (actual %x, expected %x)"),
+                  msg->header.prog, prog->program);
+        return -1;
+    }
+
+    if (msg->header.vers != prog->version) {
+        VIR_ERROR(_("version mismatch in event (actual %x, expected %x)"),
+                  msg->header.vers, prog->version);
+        return -1;
+    }
+
+    if (msg->header.status != VIR_NET_OK) {
+        VIR_ERROR(_("status mismatch in event (actual %x, expected %x)"),
+                  msg->header.status, VIR_NET_OK);
+        return -1;
+    }
+
+    if (msg->header.type != VIR_NET_MESSAGE) {
+        VIR_ERROR(_("type mismatch in event (actual %x, expected %x)"),
+                  msg->header.type, VIR_NET_MESSAGE);
+        return -1;
+    }
+
+    event = virNetClientProgramGetEvent(prog, msg->header.proc);
+
+    if (!event) {
+        VIR_ERROR(_("No event expected with procedure %x"),
+                  msg->header.proc);
+        return -1;
+    }
+
+    if (VIR_ALLOC_N(evdata, event->msg_len) < 0) {
+        virReportOOMError();
+        return -1;
+    }
+
+    if (virNetMessageDecodePayload(msg, event->msg_filter, evdata) < 0)
+        goto cleanup;
+
+    event->func(prog, client, evdata, prog->eventOpaque);
+
+    xdr_free(event->msg_filter, evdata);
+
+cleanup:
+    VIR_FREE(evdata);
+    return 0;
+}
+
+
+int virNetClientProgramCall(virNetClientProgramPtr prog,
+                            virNetClientPtr client,
+                            unsigned serial,
+                            int proc,
+                            xdrproc_t args_filter, void *args,
+                            xdrproc_t ret_filter, void *ret)
+{
+    virNetMessagePtr msg;
+
+    if (!(msg = virNetMessageNew()))
+        return -1;
+
+    msg->header.prog = prog->program;
+    msg->header.vers = prog->version;
+    msg->header.status = VIR_NET_OK;
+    msg->header.type = VIR_NET_CALL;
+    msg->header.serial = serial;
+    msg->header.proc = proc;
+
+    if (virNetMessageEncodeHeader(msg) < 0)
+        goto error;
+
+    if (virNetMessageEncodePayload(msg, args_filter, args) < 0)
+        goto error;
+
+    if (virNetClientSend(client, msg, true) < 0)
+        goto error;
+
+    /* None of these 3 should ever happen here, because
+     * virNetClientSend should have validated the reply,
+     * but it doesn't hurt to check again.
+     */
+    if (msg->header.type != VIR_NET_REPLY) {
+        virNetError(VIR_ERR_INTERNAL_ERROR,
+                    _("Unexpected message type %d"), msg->header.type);
+        goto error;
+    }
+    if (msg->header.proc != proc) {
+        virNetError(VIR_ERR_INTERNAL_ERROR,
+                    _("Unexpected message proc %d != %d"),
+                    msg->header.proc, proc);
+        goto error;
+    }
+    if (msg->header.serial != serial) {
+        virNetError(VIR_ERR_INTERNAL_ERROR,
+                    _("Unexpected message serial %d != %d"),
+                    msg->header.serial, serial);
+        goto error;
+    }
+
+    switch (msg->header.status) {
+    case VIR_NET_OK:
+        if (virNetMessageDecodePayload(msg, ret_filter, ret) < 0)
+            goto error;
+        break;
+
+    case VIR_NET_ERROR:
+        virNetClientProgramDispatchError(prog, msg);
+        goto error;
+
+    default:
+        virNetError(VIR_ERR_RPC,
+                    _("Unexpected message status %d"), msg->header.status);
+        goto error;
+    }
+
+    VIR_FREE(msg);
+
+    return 0;
+
+error:
+    VIR_FREE(msg);
+    return -1;
+}

src/rpc/virnetclientprogram.h

@@ -0,0 +1,85 @@
+/*
+ * virnetclientprogram.h: generic network RPC client program
+ *
+ * Copyright (C) 2006-2011 Red Hat, Inc.
+ *
+ * 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
+ *
+ * Author: Daniel P. Berrange <berrange@redhat.com>
+ */
+
+#ifndef __VIR_NET_CLIENT_PROGRAM_H__
+# define __VIR_NET_CLIENT_PROGRAM_H__
+
+# include <rpc/types.h>
+# include <rpc/xdr.h>
+
+# include "virnetmessage.h"
+
+typedef struct _virNetClient virNetClient;
+typedef virNetClient *virNetClientPtr;
+
+typedef struct _virNetClientProgram virNetClientProgram;
+typedef virNetClientProgram *virNetClientProgramPtr;
+
+typedef struct _virNetClientProgramEvent virNetClientProgramEvent;
+typedef virNetClientProgramEvent *virNetClientProgramEventPtr;
+
+typedef struct _virNetClientProgramErrorHandler virNetClientProgramErrorHander;
+typedef virNetClientProgramErrorHander *virNetClientProgramErrorHanderPtr;
+
+
+typedef void (*virNetClientProgramDispatchFunc)(virNetClientProgramPtr prog,
+                                                virNetClientPtr client,
+                                                void *msg,
+                                                void *opaque);
+
+struct _virNetClientProgramEvent {
+    int proc;
+    virNetClientProgramDispatchFunc func;
+    size_t msg_len;
+    xdrproc_t msg_filter;
+};
+
+virNetClientProgramPtr virNetClientProgramNew(unsigned program,
+                                              unsigned version,
+                                              virNetClientProgramEventPtr events,
+                                              size_t nevents,
+                                              void *eventOpaque);
+
+unsigned virNetClientProgramGetProgram(virNetClientProgramPtr prog);
+unsigned virNetClientProgramGetVersion(virNetClientProgramPtr prog);
+
+void virNetClientProgramRef(virNetClientProgramPtr prog);
+
+void virNetClientProgramFree(virNetClientProgramPtr prog);
+
+int virNetClientProgramMatches(virNetClientProgramPtr prog,
+                               virNetMessagePtr msg);
+
+int virNetClientProgramDispatch(virNetClientProgramPtr prog,
+                                virNetClientPtr client,
+                                virNetMessagePtr msg);
+
+int virNetClientProgramCall(virNetClientProgramPtr prog,
+                            virNetClientPtr client,
+                            unsigned serial,
+                            int proc,
+                            xdrproc_t args_filter, void *args,
+                            xdrproc_t ret_filter, void *ret);
+
+
+
+#endif /* __VIR_NET_CLIENT_PROGRAM_H__ */

src/rpc/virnetclientstream.c

@@ -0,0 +1,442 @@
+/*
+ * virnetclientstream.c: generic network RPC client stream
+ *
+ * Copyright (C) 2006-2011 Red Hat, Inc.
+ *
+ * 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
+ *
+ * Author: Daniel P. Berrange <berrange@redhat.com>
+ */
+
+#include <config.h>
+
+#include "virnetclientstream.h"
+#include "virnetclient.h"
+#include "memory.h"
+#include "virterror_internal.h"
+#include "logging.h"
+#include "event.h"
+
+#define VIR_FROM_THIS VIR_FROM_RPC
+#define virNetError(code, ...)                                    \
+    virReportErrorHelper(VIR_FROM_THIS, code, __FILE__,           \
+                         __FUNCTION__, __LINE__, __VA_ARGS__)
+
+struct _virNetClientStream {
+    virNetClientProgramPtr prog;
+    int proc;
+    unsigned serial;
+    int refs;
+
+    virError err;
+
+    /* XXX this buffer is unbounded if the client
+     * app has domain events registered, since packets
+     * may be read off wire, while app isn't ready to
+     * recv them. Figure out how to address this some
+     * time by stopping consuming any incoming data
+     * off the socket....
+     */
+    char *incoming;
+    size_t incomingOffset;
+    size_t incomingLength;
+
+
+    virNetClientStreamEventCallback cb;
+    void *cbOpaque;
+    virFreeCallback cbFree;
+    int cbEvents;
+    int cbTimer;
+    int cbDispatch;
+};
+
+
+static void
+virNetClientStreamEventTimerUpdate(virNetClientStreamPtr st)
+{
+    if (!st->cb)
+        return;
+
+    VIR_DEBUG("Check timer offset=%zu %d", st->incomingOffset, st->cbEvents);
+
+    if ((st->incomingOffset &&
+         (st->cbEvents & VIR_STREAM_EVENT_READABLE)) ||
+        (st->cbEvents & VIR_STREAM_EVENT_WRITABLE)) {
+        VIR_DEBUG("Enabling event timer");
+        virEventUpdateTimeout(st->cbTimer, 0);
+    } else {
+        VIR_DEBUG("Disabling event timer");
+        virEventUpdateTimeout(st->cbTimer, -1);
+    }
+}
+
+
+static void
+virNetClientStreamEventTimer(int timer ATTRIBUTE_UNUSED, void *opaque)
+{
+    virNetClientStreamPtr st = opaque;
+    int events = 0;
+
+    /* XXX we need a mutex on 'st' to protect this callback */
+
+    if (st->cb &&
+        (st->cbEvents & VIR_STREAM_EVENT_READABLE) &&
+        st->incomingOffset)
+        events |= VIR_STREAM_EVENT_READABLE;
+    if (st->cb &&
+        (st->cbEvents & VIR_STREAM_EVENT_WRITABLE))
+        events |= VIR_STREAM_EVENT_WRITABLE;
+
+    VIR_DEBUG("Got Timer dispatch %d %d offset=%zu", events, st->cbEvents, st->incomingOffset);
+    if (events) {
+        virNetClientStreamEventCallback cb = st->cb;
+        void *cbOpaque = st->cbOpaque;
+        virFreeCallback cbFree = st->cbFree;
+
+        st->cbDispatch = 1;
+        (cb)(st, events, cbOpaque);
+        st->cbDispatch = 0;
+
+        if (!st->cb && cbFree)
+            (cbFree)(cbOpaque);
+    }
+}
+
+
+static void
+virNetClientStreamEventTimerFree(void *opaque)
+{
+    virNetClientStreamPtr st = opaque;
+    virNetClientStreamFree(st);
+}
+
+
+virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog,
+                                            int proc,
+                                            unsigned serial)
+{
+    virNetClientStreamPtr st;
+
+    if (VIR_ALLOC(st) < 0) {
+        virReportOOMError();
+        return NULL;
+    }
+
+    virNetClientProgramRef(prog);
+
+    st->refs = 1;
+    st->prog = prog;
+    st->proc = proc;
+    st->serial = serial;
+
+    return st;
+}
+
+
+void virNetClientStreamRef(virNetClientStreamPtr st)
+{
+    st->refs++;
+}
+
+void virNetClientStreamFree(virNetClientStreamPtr st)
+{
+    st->refs--;
+    if (st->refs > 0)
+        return;
+
+    virResetError(&st->err);
+    VIR_FREE(st->incoming);
+    virNetClientProgramFree(st->prog);
+    VIR_FREE(st);
+}
+
+bool virNetClientStreamMatches(virNetClientStreamPtr st,
+                               virNetMessagePtr msg)
+{
+    if (virNetClientProgramMatches(st->prog, msg) &&
+        st->proc == msg->header.proc &&
+        st->serial == msg->header.serial)
+        return 1;
+    return 0;
+}
+
+
+bool virNetClientStreamRaiseError(virNetClientStreamPtr st)
+{
+    if (st->err.code == VIR_ERR_OK)
+        return false;
+
+    virRaiseErrorFull(__FILE__, __FUNCTION__, __LINE__,
+                      st->err.domain,
+                      st->err.code,
+                      st->err.level,
+                      st->err.str1,
+                      st->err.str2,
+                      st->err.str3,
+                      st->err.int1,
+                      st->err.int2,
+                      "%s", st->err.message ? st->err.message : _("Unknown error"));
+
+    return true;
+}
+
+
+int virNetClientStreamSetError(virNetClientStreamPtr st,
+                               virNetMessagePtr msg)
+{
+    virNetMessageError err;
+    int ret = -1;
+
+    if (st->err.code != VIR_ERR_OK)
+        VIR_DEBUG("Overwriting existing stream error %s", NULLSTR(st->err.message));
+
+    virResetError(&st->err);
+    memset(&err, 0, sizeof(err));
+
+    if (virNetMessageDecodePayload(msg, (xdrproc_t)xdr_virNetMessageError, &err) < 0)
+        goto cleanup;
+
+    if (err.domain == VIR_FROM_REMOTE &&
+        err.code == VIR_ERR_RPC &&
+        err.level == VIR_ERR_ERROR &&
+        err.message &&
+        STRPREFIX(*err.message, "unknown procedure")) {
+        st->err.code = VIR_ERR_NO_SUPPORT;
+    } else {
+        st->err.code = err.code;
+    }
+    st->err.message = *err.message;
+    *err.message = NULL;
+    st->err.domain = err.domain;
+    st->err.level = err.level;
+    st->err.str1 = *err.str1;
+    st->err.str2 = *err.str2;
+    st->err.str3 = *err.str3;
+    st->err.int1 = err.int1;
+    st->err.int2 = err.int2;
+
+    ret = 0;
+
+cleanup:
+    xdr_free((xdrproc_t)xdr_virNetMessageError, (void*)&err);
+    return ret;
+}
+
+
+int virNetClientStreamQueuePacket(virNetClientStreamPtr st,
+                                  virNetMessagePtr msg)
+{
+    size_t avail = st->incomingLength - st->incomingOffset;
+    size_t need = msg->bufferLength - msg->bufferOffset;
+
+    if (need > avail) {
+        size_t extra = need - avail;
+        if (VIR_REALLOC_N(st->incoming,
+                          st->incomingLength + extra) < 0) {
+            VIR_DEBUG("Out of memory handling stream data");
+            return -1;
+        }
+        st->incomingLength += extra;
+    }
+
+    memcpy(st->incoming + st->incomingOffset,
+           msg->buffer + msg->bufferOffset,
+           msg->bufferLength - msg->bufferOffset);
+    st->incomingOffset += (msg->bufferLength - msg->bufferOffset);
+
+    VIR_DEBUG("Stream incoming data offset %zu length %zu",
+              st->incomingOffset, st->incomingLength);
+    return 0;
+}
+
+
+int virNetClientStreamSendPacket(virNetClientStreamPtr st,
+                                 virNetClientPtr client,
+                                 int status,
+                                 const char *data,
+                                 size_t nbytes)
+{
+    virNetMessagePtr msg;
+    bool wantReply;
+    VIR_DEBUG("st=%p status=%d data=%p nbytes=%zu", st, status, data, nbytes);
+
+    if (!(msg = virNetMessageNew()))
+        return -1;
+
+    msg->header.prog = virNetClientProgramGetProgram(st->prog);
+    msg->header.vers = virNetClientProgramGetVersion(st->prog);
+    msg->header.status = status;
+    msg->header.type = VIR_NET_STREAM;
+    msg->header.serial = st->serial;
+    msg->header.proc = st->proc;
+
+    if (virNetMessageEncodeHeader(msg) < 0)
+        goto error;
+
+    /* Data packets are async fire&forget, but OK/ERROR packets
+     * need a synchronous confirmation
+     */
+    if (status == VIR_NET_CONTINUE) {
+        if (virNetMessageEncodePayloadRaw(msg, data, nbytes) < 0)
+            goto error;
+        wantReply = false;
+    } else {
+        if (virNetMessageEncodePayloadRaw(msg, NULL, 0) < 0)
+            goto error;
+        wantReply = true;
+    }
+
+    if (virNetClientSend(client, msg, wantReply) < 0)
+        goto error;
+
+
+    return nbytes;
+
+error:
+    VIR_FREE(msg);
+    return -1;
+}
+
+int virNetClientStreamRecvPacket(virNetClientStreamPtr st,
+                                 virNetClientPtr client,
+                                 char *data,
+                                 size_t nbytes,
+                                 bool nonblock)
+{
+    int rv = -1;
+    VIR_DEBUG("st=%p client=%p data=%p nbytes=%zu nonblock=%d",
+              st, client, data, nbytes, nonblock);
+    if (!st->incomingOffset) {
+        virNetMessagePtr msg;
+        int ret;
+
+        if (nonblock) {
+            VIR_DEBUG("Non-blocking mode and no data available");
+            rv = -2;
+            goto cleanup;
+        }
+
+        if (!(msg = virNetMessageNew())) {
+            virReportOOMError();
+            goto cleanup;
+        }
+
+        msg->header.prog = virNetClientProgramGetProgram(st->prog);
+        msg->header.vers = virNetClientProgramGetVersion(st->prog);
+        msg->header.type = VIR_NET_STREAM;
+        msg->header.serial = st->serial;
+        msg->header.proc = st->proc;
+
+        VIR_DEBUG("Dummy packet to wait for stream data");
+        ret = virNetClientSend(client, msg, true);
+
+        virNetMessageFree(msg);
+
+        if (ret < 0)
+            goto cleanup;
+    }
+
+    VIR_DEBUG("After IO %zu", st->incomingOffset);
+    if (st->incomingOffset) {
+        int want = st->incomingOffset;
+        if (want > nbytes)
+            want = nbytes;
+        memcpy(data, st->incoming, want);
+        if (want < st->incomingOffset) {
+            memmove(st->incoming, st->incoming + want, st->incomingOffset - want);
+            st->incomingOffset -= want;
+        } else {
+            VIR_FREE(st->incoming);
+            st->incomingOffset = st->incomingLength = 0;
+        }
+        rv = want;
+    } else {
+        rv = 0;
+    }
+
+    virNetClientStreamEventTimerUpdate(st);
+
+cleanup:
+    return rv;
+}
+
+
+int virNetClientStreamEventAddCallback(virNetClientStreamPtr st,
+                                       int events,
+                                       virNetClientStreamEventCallback cb,
+                                       void *opaque,
+                                       virFreeCallback ff)
+{
+    if (st->cb) {
+        virNetError(VIR_ERR_INTERNAL_ERROR,
+                    "%s", _("multiple stream callbacks not supported"));
+        return 1;
+    }
+
+    virNetClientStreamRef(st);
+    if ((st->cbTimer =
+         virEventAddTimeout(-1,
+                            virNetClientStreamEventTimer,
+                            st,
+                            virNetClientStreamEventTimerFree)) < 0) {
+        virNetClientStreamFree(st);
+        return -1;
+    }
+
+    st->cb = cb;
+    st->cbOpaque = opaque;
+    st->cbFree = ff;
+    st->cbEvents = events;
+
+    virNetClientStreamEventTimerUpdate(st);
+
+    return 0;
+}
+
+int virNetClientStreamEventUpdateCallback(virNetClientStreamPtr st,
+                                          int events)
+{
+    if (!st->cb) {
+        virNetError(VIR_ERR_INTERNAL_ERROR,
+                    "%s", _("no stream callback registered"));
+        return -1;
+    }
+
+    st->cbEvents = events;
+
+    virNetClientStreamEventTimerUpdate(st);
+
+    return 0;
+}
+
+int virNetClientStreamEventRemoveCallback(virNetClientStreamPtr st)
+{
+    if (!st->cb) {
+        virNetError(VIR_ERR_INTERNAL_ERROR,
+                    "%s", _("no stream callback registered"));
+        return -1;
+    }
+
+    if (!st->cbDispatch &&
+        st->cbFree)
+        (st->cbFree)(st->cbOpaque);
+    st->cb = NULL;
+    st->cbOpaque = NULL;
+    st->cbFree = NULL;
+    st->cbEvents = 0;
+    virEventRemoveTimeout(st->cbTimer);
+
+    return 0;
+}

src/rpc/virnetclientstream.h

@@ -0,0 +1,76 @@
+/*
+ * virnetclientstream.h: generic network RPC client stream
+ *
+ * Copyright (C) 2006-2011 Red Hat, Inc.
+ *
+ * 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
+ *
+ * Author: Daniel P. Berrange <berrange@redhat.com>
+ */
+
+#ifndef __VIR_NET_CLIENT_STREAM_H__
+# define __VIR_NET_CLIENT_STREAM_H__
+
+# include "virnetclientprogram.h"
+
+typedef struct _virNetClientStream virNetClientStream;
+typedef virNetClientStream *virNetClientStreamPtr;
+
+typedef void (*virNetClientStreamEventCallback)(virNetClientStreamPtr stream,
+                                                int events, void *opaque);
+
+virNetClientStreamPtr virNetClientStreamNew(virNetClientProgramPtr prog,
+                                            int proc,
+                                            unsigned serial);
+
+void virNetClientStreamRef(virNetClientStreamPtr st);
+
+void virNetClientStreamFree(virNetClientStreamPtr st);
+
+bool virNetClientStreamRaiseError(virNetClientStreamPtr st);
+
+int virNetClientStreamSetError(virNetClientStreamPtr st,
+                               virNetMessagePtr msg);
+
+bool virNetClientStreamMatches(virNetClientStreamPtr st,
+                               virNetMessagePtr msg);
+
+int virNetClientStreamQueuePacket(virNetClientStreamPtr st,
+                                  virNetMessagePtr msg);
+
+int virNetClientStreamSendPacket(virNetClientStreamPtr st,
+                                 virNetClientPtr client,
+                                 int status,
+                                 const char *data,
+                                 size_t nbytes);
+
+int virNetClientStreamRecvPacket(virNetClientStreamPtr st,
+                                 virNetClientPtr client,
+                                 char *data,
+                                 size_t nbytes,
+                                 bool nonblock);
+
+int virNetClientStreamEventAddCallback(virNetClientStreamPtr st,
+                                       int events,
+                                       virNetClientStreamEventCallback cb,
+                                       void *opaque,
+                                       virFreeCallback ff);
+
+int virNetClientStreamEventUpdateCallback(virNetClientStreamPtr st,
+                                          int events);
+int virNetClientStreamEventRemoveCallback(virNetClientStreamPtr st);
+
+
+#endif /* __VIR_NET_CLIENT_STREAM_H__ */