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.
This commit is contained in:
Daniel P. Berrange 2010-12-01 16:35:50 +00:00
parent e23ec81db6
commit 434de30da5
9 changed files with 2211 additions and 1 deletions

3
cfg.mk
View File

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

View File

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

View File

@ -1188,7 +1188,7 @@ else
EXTRA_DIST += $(LOCK_DRIVER_SANLOCK_SOURCES) EXTRA_DIST += $(LOCK_DRIVER_SANLOCK_SOURCES)
endif 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 = \ libvirt_net_rpc_la_SOURCES = \
rpc/virnetmessage.h rpc/virnetmessage.c \ rpc/virnetmessage.h rpc/virnetmessage.c \
@ -1238,6 +1238,18 @@ libvirt_net_rpc_server_la_LDFLAGS = \
libvirt_net_rpc_server_la_LIBADD = \ libvirt_net_rpc_server_la_LIBADD = \
$(CYGWIN_EXTRA_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 = libexec_PROGRAMS =

1166
src/rpc/virnetclient.c Normal file

File diff suppressed because it is too large Load Diff

84
src/rpc/virnetclient.h Normal file
View File

@ -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__ */

View File

@ -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;
}

View File

@ -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__ */

View File

@ -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;
}

View File

@ -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__ */