External/libvirt
1
0
Fork 0
mirror of https://gitlab.com/libvirt/libvirt.git synced 2024-07-11 04:15:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Introduce generic RPC client objects

Browse Source 
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
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
cfg.mk
View File

@ -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 \

3
po/POTFILES.in
View File

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

14
src/Makefile.am
View File

@ -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 =

1166
src/rpc/virnetclient.c Normal file
View 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__ */

339
src/rpc/virnetclientprogram.c Normal file
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;
}

85
src/rpc/virnetclientprogram.h Normal file
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__ */

442
src/rpc/virnetclientstream.c Normal file
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;
}

76
src/rpc/virnetclientstream.h Normal file
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__ */