Commit 41d08218 authored by Daniel P. Berrange's avatar Daniel P. Berrange

Add sparse stream API support

Signed-off-by: 's avatarDaniel P. Berrange <berrange@redhat.com>
parent 5ba8849e
......@@ -143,6 +143,9 @@ var (
/* Typedefs that don't need exposing as is */
"virStreamSinkFunc",
"virStreamSourceFunc",
"virStreamSinkHoleFunc",
"virStreamSourceHoleFunc",
"virStreamSourceSkipFunc",
/* Only needed at C level */
"virDomainGetConnect",
......
......@@ -32,6 +32,7 @@ package libvirt
#include <libvirt/virterror.h>
#include <stdlib.h>
#include "stream_cfuncs.h"
#include "stream_compat.h"
*/
import "C"
import (
......@@ -54,6 +55,12 @@ const (
STREAM_EVENT_HANGUP = StreamEventType(C.VIR_STREAM_EVENT_HANGUP)
)
type StreamRecvFlagsValues int
const (
STREAM_RECV_STOP_AT_HOLE = StreamRecvFlagsValues(C.VIR_STREAM_RECV_STOP_AT_HOLE)
)
type Stream struct {
ptr C.virStreamPtr
}
......@@ -104,6 +111,36 @@ func (v *Stream) Recv(p []byte) (int, error) {
return int(n), nil
}
func (v *Stream) RecvFlags(p []byte, flags StreamRecvFlagsValues) (int, error) {
if C.LIBVIR_VERSION_NUMBER < 3004000 {
return 0, GetNotImplementedError("virStreamRecvFlags")
}
n := C.virStreamRecvFlagsCompat(v.ptr, (*C.char)(unsafe.Pointer(&p[0])), C.size_t(len(p)), C.uint(flags))
if n < 0 {
return 0, GetLastError()
}
if n == 0 {
return 0, io.EOF
}
return int(n), nil
}
func (v *Stream) RecvHole(flags uint) (int64, error) {
if C.LIBVIR_VERSION_NUMBER < 3004000 {
return 0, GetNotImplementedError("virStreamSparseRecvHole")
}
var len C.longlong
ret := C.virStreamRecvHoleCompat(v.ptr, &len, C.uint(flags))
if ret < 0 {
return 0, GetLastError()
}
return int64(len), nil
}
func (v *Stream) Send(p []byte) (int, error) {
n := C.virStreamSend(v.ptr, (*C.char)(unsafe.Pointer(&p[0])), C.size_t(len(p)))
if n < 0 {
......@@ -116,7 +153,21 @@ func (v *Stream) Send(p []byte) (int, error) {
return int(n), nil
}
func (v *Stream) SendHole(len int64, flags uint32) error {
if C.LIBVIR_VERSION_NUMBER < 3004000 {
return GetNotImplementedError("virStreamSendHole")
}
ret := C.virStreamSendHoleCompat(v.ptr, C.longlong(len), C.uint(flags))
if ret < 0 {
return GetLastError()
}
return nil
}
type StreamSinkFunc func(*Stream, []byte) (int, error)
type StreamSinkHoleFunc func(*Stream, int64) error
//export streamSinkCallback
func streamSinkCallback(stream C.virStreamPtr, cdata *C.char, nbytes C.size_t, callbackID int) int {
......@@ -141,6 +192,23 @@ func streamSinkCallback(stream C.virStreamPtr, cdata *C.char, nbytes C.size_t, c
return retnbytes
}
//export streamSinkHoleCallback
func streamSinkHoleCallback(stream C.virStreamPtr, length C.longlong, callbackID int) int {
callbackFunc := getCallbackId(callbackID)
callback, ok := callbackFunc.(StreamSinkHoleFunc)
if !ok {
panic("Incorrect stream sink hole func callback")
}
err := callback(&Stream{ptr: stream}, int64(length))
if err != nil {
return -1
}
return 0
}
func (v *Stream) RecvAll(handler StreamSinkFunc) error {
callbackID := registerCallbackId(handler)
......@@ -154,7 +222,27 @@ func (v *Stream) RecvAll(handler StreamSinkFunc) error {
return nil
}
func (v *Stream) SparseRecvAll(handler StreamSinkFunc, holeHandler StreamSinkHoleFunc) error {
if C.LIBVIR_VERSION_NUMBER < 3004000 {
return GetNotImplementedError("virStreamSparseSendAll")
}
callbackID := registerCallbackId(handler)
holeCallbackID := registerCallbackId(holeHandler)
ret := C.virStreamSparseRecvAll_cgo(v.ptr, (C.int)(callbackID), (C.int)(holeCallbackID))
freeCallbackId(callbackID)
freeCallbackId(holeCallbackID)
if ret == -1 {
return GetLastError()
}
return nil
}
type StreamSourceFunc func(*Stream, int) ([]byte, error)
type StreamSourceHoleFunc func(*Stream) (bool, int64, error)
type StreamSourceSkipFunc func(*Stream, int64) error
//export streamSourceCallback
func streamSourceCallback(stream C.virStreamPtr, cdata *C.char, nbytes C.size_t, callbackID int) int {
......@@ -183,6 +271,47 @@ func streamSourceCallback(stream C.virStreamPtr, cdata *C.char, nbytes C.size_t,
return nretbytes
}
//export streamSourceHoleCallback
func streamSourceHoleCallback(stream C.virStreamPtr, cinData *C.int, clength *C.longlong, callbackID int) int {
callbackFunc := getCallbackId(callbackID)
callback, ok := callbackFunc.(StreamSourceHoleFunc)
if !ok {
panic("Incorrect stream sink hole func callback")
}
inData, length, err := callback(&Stream{ptr: stream})
if err != nil {
return -1
}
if inData {
*cinData = 1
} else {
*cinData = 0
}
*clength = C.longlong(length)
return 0
}
//export streamSourceSkipCallback
func streamSourceSkipCallback(stream C.virStreamPtr, length C.longlong, callbackID int) int {
callbackFunc := getCallbackId(callbackID)
callback, ok := callbackFunc.(StreamSourceSkipFunc)
if !ok {
panic("Incorrect stream sink skip func callback")
}
err := callback(&Stream{ptr: stream}, int64(length))
if err != nil {
return -1
}
return 0
}
func (v *Stream) SendAll(handler StreamSourceFunc) error {
callbackID := registerCallbackId(handler)
......@@ -196,6 +325,26 @@ func (v *Stream) SendAll(handler StreamSourceFunc) error {
return nil
}
func (v *Stream) SparseSendAll(handler StreamSourceFunc, holeHandler StreamSourceHoleFunc, skipHandler StreamSourceSkipFunc) error {
if C.LIBVIR_VERSION_NUMBER < 3004000 {
return GetNotImplementedError("virStreamSparseSendAll")
}
callbackID := registerCallbackId(handler)
holeCallbackID := registerCallbackId(holeHandler)
skipCallbackID := registerCallbackId(skipHandler)
ret := C.virStreamSparseSendAll_cgo(v.ptr, (C.int)(callbackID), (C.int)(holeCallbackID), (C.int)(skipCallbackID))
freeCallbackId(callbackID)
freeCallbackId(holeCallbackID)
freeCallbackId(skipCallbackID)
if ret == -1 {
return GetLastError()
}
return nil
}
type StreamEventCallback func(*Stream, StreamEventType)
func (v *Stream) EventAddCallback(events StreamEventType, callback StreamEventCallback) error {
......
......@@ -35,31 +35,84 @@ package libvirt
#include "stream_cfuncs.h"
int streamSourceCallback(virStreamPtr st, char *cdata, size_t nbytes, int callbackID);
int streamSourceHoleCallback(virStreamPtr st, int *inData, long long *length, int callbackID);
int streamSourceSkipCallback(virStreamPtr st, long long length, int callbackID);
int streamSinkCallback(virStreamPtr st, const char *cdata, size_t nbytes, int callbackID);
int streamSinkHoleCallback(virStreamPtr st, long long length, int callbackID);
struct CallbackData {
int callbackID;
int holeCallbackID;
int skipCallbackID;
};
static int streamSourceCallbackHelper(virStreamPtr st, char *data, size_t nbytes, void *opaque)
{
int *callbackID = opaque;
struct CallbackData *cbdata = opaque;
return streamSourceCallback(st, data, nbytes, cbdata->callbackID);
}
static int streamSourceHoleCallbackHelper(virStreamPtr st, int *inData, long long *length, void *opaque)
{
struct CallbackData *cbdata = opaque;
return streamSourceHoleCallback(st, inData, length, cbdata->holeCallbackID);
}
static int streamSourceSkipCallbackHelper(virStreamPtr st, long long length, void *opaque)
{
struct CallbackData *cbdata = opaque;
return streamSourceCallback(st, data, nbytes, *callbackID);
return streamSourceSkipCallback(st, length, cbdata->skipCallbackID);
}
static int streamSinkCallbackHelper(virStreamPtr st, const char *data, size_t nbytes, void *opaque)
{
int *callbackID = opaque;
struct CallbackData *cbdata = opaque;
return streamSinkCallback(st, data, nbytes, *callbackID);
return streamSinkCallback(st, data, nbytes, cbdata->callbackID);
}
static int streamSinkHoleCallbackHelper(virStreamPtr st, long long length, void *opaque)
{
struct CallbackData *cbdata = opaque;
return streamSinkHoleCallback(st, length, cbdata->holeCallbackID);
}
int virStreamSendAll_cgo(virStreamPtr st, int callbackID)
{
return virStreamSendAll(st, streamSourceCallbackHelper, &callbackID);
struct CallbackData cbdata = { .callbackID = callbackID };
return virStreamSendAll(st, streamSourceCallbackHelper, &cbdata);
}
int virStreamSparseSendAll_cgo(virStreamPtr st, int callbackID, int holeCallbackID, int skipCallbackID)
{
struct CallbackData cbdata = { .callbackID = callbackID, .holeCallbackID = holeCallbackID, .skipCallbackID = skipCallbackID };
#if LIBVIR_VERSION_NUMBER < 3004000
assert(0); // Caller should have checked version
#else
return virStreamSparseSendAll(st, streamSourceCallbackHelper, streamSourceHoleCallbackHelper, streamSourceSkipCallbackHelper, &cbdata);
#endif
}
int virStreamRecvAll_cgo(virStreamPtr st, int callbackID)
{
return virStreamRecvAll(st, streamSinkCallbackHelper, &callbackID);
struct CallbackData cbdata = { .callbackID = callbackID };
return virStreamRecvAll(st, streamSinkCallbackHelper, &cbdata);
}
int virStreamSparseRecvAll_cgo(virStreamPtr st, int callbackID, int holeCallbackID)
{
struct CallbackData cbdata = { .callbackID = callbackID, .holeCallbackID = holeCallbackID };
#if LIBVIR_VERSION_NUMBER < 3004000
assert(0); // Caller should have checked version
#else
return virStreamSparseRecvAll(st, streamSinkCallbackHelper, streamSinkHoleCallbackHelper, &cbdata);
#endif
}
void streamEventCallback(virStreamPtr st, int events, int callbackID);
......
......@@ -29,6 +29,8 @@
int virStreamSendAll_cgo(virStreamPtr st, int callbackID);
int virStreamRecvAll_cgo(virStreamPtr st, int callbackID);
int virStreamSparseSendAll_cgo(virStreamPtr st, int callbackID, int holeCallbackID, int skipCallbackID);
int virStreamSparseRecvAll_cgo(virStreamPtr st, int callbackID, int holeCallbackID);
int virStreamEventAddCallback_cgo(virStreamPtr st, int events, int callbackID);
......
/*
* This file is part of the libvirt-go project
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*
* Copyright (C) 2017 Red Hat, Inc.
*
*/
package libvirt
/*
#cgo pkg-config: libvirt
#include <libvirt/libvirt.h>
#include <assert.h>
#include "stream_compat.h"
int virStreamRecvFlagsCompat(virStreamPtr st,
char *data,
size_t nbytes,
unsigned int flags)
{
#if LIBVIR_VERSION_NUMBER < 3002000
assert(0); // Caller should have checked version
#else
return virStreamRecvFlags(st, data, nbytes, flags);
#endif
}
int virStreamSendHoleCompat(virStreamPtr st,
long long length,
unsigned int flags)
{
#if LIBVIR_VERSION_NUMBER < 3002000
assert(0); // Caller should have checked version
#else
return virStreamSendHole(st, length, flags);
#endif
}
int virStreamRecvHoleCompat(virStreamPtr st,
long long *length,
unsigned int flags)
{
#if LIBVIR_VERSION_NUMBER < 3002000
assert(0); // Caller should have checked version
#else
return virStreamRecvHole(st, length, flags);
#endif
}
*/
import "C"
/*
* This file is part of the libvirt-go project
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in
* all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
* THE SOFTWARE.
*
* Copyright (c) 2013 Alex Zorin
* Copyright (C) 2016 Red Hat, Inc.
*
*/
#ifndef LIBVIRT_GO_STREAM_COMPAT_H__
#define LIBVIRT_GO_STREAM_COMPAT_H__
/* 3.4.0 */
#ifndef VIR_STREAM_RECV_STOP_AT_HOLE
#define VIR_STREAM_RECV_STOP_AT_HOLE (1 << 0)
#endif
int virStreamRecvFlagsCompat(virStreamPtr st,
char *data,
size_t nbytes,
unsigned int flags);
int virStreamSendHoleCompat(virStreamPtr st,
long long length,
unsigned int flags);
int virStreamRecvHoleCompat(virStreamPtr,
long long *length,
unsigned int flags);
#endif /* LIBVIRT_GO_STREAM_COMPAT_H__ */
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment